fin_depo/__init__.py

@@ -25,8 +25,6 @@ __all__ = [
     'defi_kucoin',
     'defi_partisia_blockchain',
     'investbank_nordnet',
-    '__version__',
 ]
 
 from . import defi_kraken, defi_kucoin, defi_partisia_blockchain, investbank_nordnet
-from ._version import __version__

fin_depo/data.py

@@ -10,7 +10,7 @@ from fin_defs import Asset
 
 
 @enforce_typing.enforce_types
-@dataclasses.dataclass(frozen=True)
+@dataclasses.dataclass
 class Depo(abc.ABC):
     """A depository tracking some amount of assets.
 
@@ -38,7 +38,7 @@ class Depo(abc.ABC):
 
 
 @enforce_typing.enforce_types
-@dataclasses.dataclass(frozen=True)
+@dataclasses.dataclass
 class DepoSingle(Depo):
     """Base level of depository."""
 
@@ -52,7 +52,7 @@ class DepoSingle(Depo):
 
 
 @enforce_typing.enforce_types
-@dataclasses.dataclass(frozen=True)
+@dataclasses.dataclass
 class DepoGroup(Depo):
     """Nested depository."""
 
@@ -92,25 +92,3 @@ class DepoFetcher(abc.ABC):
             msg = f'{self} expected {param_name} parameter of type {param_type}, but got: {param_value}'
             raise TypeError(msg)
         return param_value
-
-
-@enforce_typing.enforce_types
-@dataclasses.dataclass(frozen=True)
-class TradeOrderDetails:
-    """Information about an executed trade.
-
-    Includes both well-structured data about the order, and unstructured data
-    from the backend. The unstructured data might be needed for tax purposes.
-    """
-
-    input_asset: Asset
-    input_amount: Decimal
-
-    output_asset: Asset
-    output_amount: Decimal
-
-    fee_asset: Asset
-    fee_amount: Decimal
-
-    order_id: object
-    raw_order_details: object

fin_depo/defi_kucoin.py

@@ -7,7 +7,7 @@ from decimal import Decimal
 import fin_defs
 import kucoin.client
 
-from .data import DepoFetcher, DepoGroup, DepoSingle, TradeOrderDetails
+from .data import DepoFetcher, DepoGroup, DepoSingle
 
 logger = logging.getLogger(__name__)
 
@@ -19,8 +19,7 @@ class KucoinDepoFetcher(DepoFetcher):
     - Account on [Kucoin](https://www.kucoin.com).
     - Have performed Know Your Customer (KYC) for your account.
     - Created API key from [settings menu](https://www.kucoin.com/account/api).
-      API key must have the **General Permission**. Unless you are using the
+      API key must have the **General Permission**, and **should not have
-      `place_market_order` functionality, the API key **should not have
       any additional permissions**. Employ principle of least priviledge.
     - Install [`python-kucoin`](https://python-kucoin.readthedocs.io/en/latest/) library.
 
@@ -28,18 +27,11 @@ class KucoinDepoFetcher(DepoFetcher):
     sub-accounts (Funding, Trading, Margin, Futures...)
     """
 
-    def __init__(
+    def __init__(self, kucoin_key: str, kucoin_secret: str, kucoin_pass: str):
-        self,
-        kucoin_key: str,
-        kucoin_secret: str,
-        kucoin_pass: str,
-        allow_trades: bool = False,
-    ):
         self.assert_param('kucoin_key', str, kucoin_key)
         self.assert_param('kucoin_secret', str, kucoin_secret)
         self.assert_param('kucoin_pass', str, kucoin_pass)
-        self.allow_trades = allow_trades
+        self.client = kucoin.client.Client(
-        self.kucoin_client = kucoin.client.Client(
             kucoin_key,
             kucoin_secret,
             kucoin_pass,
@@ -54,7 +46,7 @@ class KucoinDepoFetcher(DepoFetcher):
         # clustered into different depos.
         assets_by_account_type: dict[str, dict[fin_defs.Asset, Decimal]] = {}
 
-        for account_data in self.kucoin_client.get_accounts():
+        for account_data in self.client.get_accounts():
             asset = fin_defs.WELL_KNOWN_SYMBOLS[account_data['currency']]
             balance = Decimal(account_data['balance'])
             assets_for_account_type = assets_by_account_type.setdefault(
@@ -74,81 +66,3 @@ class KucoinDepoFetcher(DepoFetcher):
                 for account_type, assets in assets_by_account_type.items()
             ],
         )
-
-    def place_market_order(
-        self,
-        input_asset: fin_defs.Asset,
-        input_amount: Decimal,
-        output_asset: fin_defs.Asset,
-    ) -> TradeOrderDetails:
-        """Executes a market order through the Kucoin market.
-
-        Will automatically determine the market based on the input and output
-        assets.
-
-        Requirements:
-
-        - Fetcher must have been created with `allow_trades=True`.
-        - API key used with fetcher must have **Spot Trading** permissions.
-        - Assets must be on trading account. Assets on funding accounts or
-          other accounts cannot be used.
-
-        Note:
-        ----
-        - A fee will be paid to Kucoin, with the rate determined by your WIP
-          level and the asset being traded.
-        - The full `input_amount` may not be used. Inspect the resulting
-          `TradeOrderDetails` to see how much of the `input_amount` have been
-          used.
-
-        """
-        # Check requirements
-        if not self.allow_trades:
-            msg = 'KucoinDepoFetcher.allow_trades is not enabled: Cannot make trades'
-            raise PermissionError(msg)
-        assert fin_defs.USDT in [input_asset, output_asset], 'USDT markets only for now'
-
-        # Convert arguments to kucoin client arguments
-        if input_asset == fin_defs.USDT:
-            symbol: str = f'{output_asset}-{input_asset}'
-            side: str = 'buy'
-            size = None
-            funds = str(input_amount)
-        else:
-            symbol = f'{input_asset}-{output_asset}'
-            side = 'sell'
-            size = str(input_amount)
-            funds = None
-
-        # Place order
-        logger.info('Placing order: %s', str([symbol, side, size, funds]))
-        response = self.kucoin_client.create_market_order(
-            symbol=symbol,
-            side=side,
-            size=size,
-            funds=funds,
-        )
-        del symbol, side, size, funds, input_amount
-
-        # Determine order details
-        order_id = response['orderId']
-        order_details = self.kucoin_client.get_order(order_id)
-
-        # Convert from kucoin results
-        if input_asset == fin_defs.USDT:
-            input_amount_final = Decimal(order_details['dealFunds'])
-            output_amount_final = Decimal(order_details['dealSize'])
-        else:
-            output_amount_final = Decimal(order_details['dealFunds'])
-            input_amount_final = Decimal(order_details['dealSize'])
-
-        return TradeOrderDetails(
-            input_asset=input_asset,
-            input_amount=input_amount_final,
-            output_asset=output_asset,
-            output_amount=output_amount_final,
-            fee_asset=fin_defs.WELL_KNOWN_SYMBOLS[order_details['feeCurrency']],
-            fee_amount=Decimal(order_details['fee']),
-            order_id=order_id,
-            raw_order_details=order_details,
-        )

fin_depo/defi_partisia_blockchain.py

@@ -7,7 +7,6 @@ import json
 import logging
 from collections.abc import Mapping
 from decimal import Decimal
-from typing import Any
 
 import fin_defs
 import requests
@@ -56,9 +55,9 @@ class PbcClient:
     def get_json(
         self,
         url: str,
-        data: Mapping[str, Any] = frozendict(),
+        data: Mapping[str, str] = frozendict(),
         method='POST',
-    ) -> tuple[Any, datetime.datetime]:
+    ) -> tuple[dict, datetime.datetime]:
         headers = {
             'Content-Type': 'application/json',
             'Accept': 'application/json',
@@ -81,10 +80,10 @@ class PbcClient:
             raise Exception(msg)
         return (json_data, date)
 
-    def determine_coins(self) -> list[dict[str, Any]]:
+    def determine_coins(self) -> list[dict[str, str]]:
         data: dict = {'path': []}
 
-        url: str = URL_ACCOUNT_PLUGIN_GLOBAL.format(
+        url = URL_ACCOUNT_PLUGIN_GLOBAL.format(
             hostname=HOSTNAME,
             shard='',
         )

fin_depo/investbank_nordnet.py

@@ -11,9 +11,7 @@ I am grateful for his pioneering work.
 
 import datetime
 import logging
-from collections.abc import Mapping
 from decimal import Decimal
-from typing import Any
 
 import fin_defs
 import requests
@@ -43,7 +41,7 @@ def asset_from_instrument_json(json) -> fin_defs.Asset | None:
     return fin_defs.Stock(symbol, fin_defs.EXCHANGES_BY_IDS[exchange_id])
 
 
-EMPTY_DICT: Mapping[str, str | int] = frozendict()
+EMPTY_DICT: dict[str, str | int] = frozendict()
 
 
 class NordnetDepoFetcher(DepoFetcher):
@@ -68,7 +66,7 @@ class NordnetDepoFetcher(DepoFetcher):
         self.password: str = self.assert_param('password', str, password)
         self.is_logged_in = False
 
-    def _get_json(self, url: str, params: Mapping[str, str | int] = EMPTY_DICT) -> Any:
+    def _get_json(self, url: str, params: dict[str, str | int] = EMPTY_DICT) -> dict:
         if not url.startswith(API_ROOT):
             msg = f'Given url must be located below API ROOT: {url}'
             raise ValueError(msg)
@@ -82,7 +80,7 @@ class NordnetDepoFetcher(DepoFetcher):
         response.raise_for_status()
         return json
 
-    def login(self) -> None:
+    def login(self):
         """Performs authentication with the login server if not already logged
         in. Does not need to be manually called; most methods that require this
         information will do it themselves.

test/secrets.py

@@ -1,20 +1,6 @@
-"""Secret loading for tests.
-
-Add these secrets to an [appropriate secret loader
-location](https://gitfub.space/Jmaa/secret_loader). Missing some secrets will
-result in skipped tests.
-"""
-
 from secret_loader import SecretLoader
 
 load_secret = SecretLoader().load
 
 NORDNET_USERNAME = load_secret('NORDNET_USERNAME')
 NORDNET_PASSWORD = load_secret('NORDNET_PASSWORD')
-
-KRAKEN_KEY = load_secret('KRAKEN_KEY')
-KRAKEN_SECRET = load_secret('KRAKEN_SECRET')
-
-KUCOIN_KEY = load_secret('KUCOIN_KEY')
-KUCOIN_SECRET = load_secret('KUCOIN_SECRET')
-KUCOIN_PASS = load_secret('KUCOIN_PASS')

test/test_kraken.py

@@ -1,23 +0,0 @@
-import pytest
-import requests
-
-from fin_depo import defi_kraken
-
-from . import secrets
-
-needs_secrets = pytest.mark.skipif(
-    not secrets.KRAKEN_KEY,
-    reason='Secret kraken_USERNAME required',
-)
-
-
-@needs_secrets
-def test_get_depo():
-    session = requests.Session()
-    fetcher = defi_kraken.KrakenDepoFetcher(
-        secrets.KRAKEN_KEY,
-        secrets.KRAKEN_SECRET,
-    )
-
-    depo = fetcher.get_depo()
-    assert depo is not None

test/test_kucoin.py

@@ -1,110 +0,0 @@
-from decimal import Decimal
-
-import fin_defs
-import pytest
-
-from fin_depo import defi_kucoin
-
-from . import secrets
-
-TEST_MARKET_ORDERS = False
-
-needs_secrets = pytest.mark.skipif(
-    not secrets.KUCOIN_KEY,
-    reason='Secret kucoin_USERNAME required',
-)
-
-defi_kucoin.logger.setLevel('INFO')
-
-
-@needs_secrets
-def test_get_depo():
-    """Can inspect kucoin depository."""
-    fetcher = defi_kucoin.KucoinDepoFetcher(
-        secrets.KUCOIN_KEY,
-        secrets.KUCOIN_SECRET,
-        secrets.KUCOIN_PASS,
-    )
-
-    depo = fetcher.get_depo()
-    assert depo is not None
-
-
-@needs_secrets
-def test_place_market_order_requires_allow_trades():
-    """Client fails if allow_trades is not enabled when doing market orders."""
-    fetcher = defi_kucoin.KucoinDepoFetcher(
-        secrets.KUCOIN_KEY,
-        secrets.KUCOIN_SECRET,
-        secrets.KUCOIN_PASS,
-        allow_trades=False,
-    )
-
-    with pytest.raises(PermissionError) as m:
-        fetcher.place_market_order(fin_defs.MPC, Decimal(1), fin_defs.USDT)
-
-    assert 'KucoinDepoFetcher.allow_trades is not enabled: Cannot make trades' in str(m)
-
-
-@needs_secrets
-def test_place_buy_side_order():
-    """Client can place buy side market orders."""
-    fetcher = defi_kucoin.KucoinDepoFetcher(
-        secrets.KUCOIN_KEY,
-        secrets.KUCOIN_SECRET,
-        secrets.KUCOIN_PASS,
-        allow_trades=TEST_MARKET_ORDERS,
-    )
-
-    input_amount = Decimal('0.1')
-
-    order_details = fetcher.place_market_order(
-        fin_defs.USDT, input_amount, fin_defs.MPC,
-    )
-
-    assert order_details is not None
-    assert order_details.order_id is not None
-    assert order_details.raw_order_details is not None
-
-    assert order_details.input_asset == fin_defs.USDT
-    assert order_details.output_asset == fin_defs.MPC
-
-    assert order_details.input_amount <= input_amount
-    assert order_details.output_amount >= 0
-
-    assert order_details.input_amount != order_details.output_amount
-
-    assert order_details.fee_asset == fin_defs.USDT
-    assert order_details.fee_amount <= Decimal('0.0002')
-
-
-@needs_secrets
-def test_place_sell_side_order():
-    """Client can place sell side market orders."""
-    fetcher = defi_kucoin.KucoinDepoFetcher(
-        secrets.KUCOIN_KEY,
-        secrets.KUCOIN_SECRET,
-        secrets.KUCOIN_PASS,
-        allow_trades=TEST_MARKET_ORDERS,
-    )
-
-    input_amount = Decimal('1')
-
-    order_details = fetcher.place_market_order(
-        fin_defs.MPC, input_amount, fin_defs.USDT,
-    )
-
-    assert order_details is not None
-    assert order_details.order_id is not None
-    assert order_details.raw_order_details is not None
-
-    assert order_details.input_asset == fin_defs.MPC
-    assert order_details.output_asset == fin_defs.USDT
-
-    assert order_details.input_amount <= input_amount
-    assert order_details.output_amount >= 0
-
-    assert order_details.input_amount != order_details.output_amount
-
-    assert order_details.fee_asset == fin_defs.USDT
-    assert order_details.fee_amount is not None