fin_depo/defi_kucoin.py

@@ -2,9 +2,9 @@
 
 import datetime
 import logging
-import time
 from decimal import Decimal
 
+import time
 import fin_defs
 import kucoin.client
 
@@ -138,9 +138,9 @@ class KucoinDepoFetcher(DepoFetcher):
         # Determine order details
         return self._get_order_details(response['orderId'], input_asset, output_asset)
 
-    def _get_order_details(
-        self, order_id: str, input_asset: fin_defs.Asset, output_asset: fin_defs.Asset,
-    ) -> TradeOrderDetails:
+    def _get_order_details(self, order_id: str,
+                           input_asset: fin_defs.Asset,
+                           output_asset: fin_defs.Asset) -> TradeOrderDetails:
         """Determine the order details for the order with the given id.
 
         Retries the order a few times, as KuCoin might not have propagated the
@@ -164,15 +164,13 @@ class KucoinDepoFetcher(DepoFetcher):
             fee_asset=fin_defs.WELL_KNOWN_SYMBOLS[order_details['feeCurrency']],
             fee_amount=Decimal(order_details['fee']),
             executed_time=datetime.datetime.fromtimestamp(
-                order_details['createdAt'] / 1000, tz=datetime.UTC,
+                order_details['createdAt'] / 1000, tz=datetime.UTC
             ),
             order_id=order_id,
             raw_order_details=order_details,
         )
 
-    def _get_order_with_retries(
-        self, order_id: str, *, num_retries: int, sleep_between_tries: float = 1.0,
-    ) -> dict:
+    def _get_order_with_retries(self, order_id: str, *, num_retries: int, sleep_between_tries:float = 1.0) -> dict:
         """Get the order details from KuCoin backend.
 
         Retries the order a few times, as KuCoin might not have propagated the

fin_depo/investbank_nordnet.py

@@ -13,22 +13,33 @@ import datetime
 import logging
 from collections.abc import Mapping
 from decimal import Decimal
+from typing import Any
 
 import fin_defs
 import requests
 from frozendict import frozendict
-from nordnet_api_client import NordnetClient
 
 from .data import Depo, DepoFetcher, DepoGroup, DepoSingle
 
 logger = logging.getLogger(__name__)
 
+API_HOSTNAME = 'public.nordnet.dk'
+API_ROOT = f'https://{API_HOSTNAME}/api/2'
 
-def asset_from_instrument(intrument) -> fin_defs.Asset | None:
-    if intrument.instrument_group_type == 'FND':
+API_LOGIN_STEP_1 = 'https://www.nordnet.dk/logind'
+API_LOGIN_STEP_2 = API_ROOT + '/authentication/basic/login'
+
+API_ACCOUNTS = API_ROOT + '/accounts'
+API_ACCOUNT_INFO = API_ROOT + '/accounts/{accid}/info'
+API_ACCOUNT_LEDGERS = API_ROOT + '/accounts/{accid}/ledgers'
+API_ACCOUNT_POSITIONS = API_ROOT + '/accounts/{accid}/positions'
+
+
+def asset_from_instrument_json(json) -> fin_defs.Asset | None:
+    if json['instrument_group_type'] == 'FND':
         return None
-    symbol = intrument.symbol
-    exchange_id = intrument.tradables[0].mic
+    symbol = json['symbol']
+    exchange_id = json['tradables'][0]['mic']
     return fin_defs.Stock(symbol, fin_defs.EXCHANGES_BY_IDS[exchange_id])
 
 
@@ -52,42 +63,88 @@ class NordnetDepoFetcher(DepoFetcher):
     """
 
     def __init__(self, session: requests.Session, username: str, password: str):
-        self.client = NordnetClient(session, username, password)
+        self.session = self.assert_param('session', requests.Session, session)
+        self.username: str = self.assert_param('username', str, username)
+        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:
+        if not url.startswith(API_ROOT):
+            msg = f'Given url must be located below API ROOT: {url}'
+            raise ValueError(msg)
+
+        headers = {
+            'Accepts': 'application/json',
+        }
+        response = self.session.get(url, params=params, headers=headers)
+
+        json = response.json()
+        response.raise_for_status()
+        return json
+
+    def login(self) -> None:
+        """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.
+
+        This method is based on Morten Helmstedt's version:
+        <https://github.com/helmstedt/nordnet-utilities/blob/main/nordnet_login.py>
+        """
+        if self.is_logged_in:
+            return
+
+        self.session.get(API_LOGIN_STEP_1)
+        self.session.headers['client-id'] = 'NEXT'
+        self.session.headers['sub-client-id'] = 'NEXT'
+        response = self.session.post(
+            API_LOGIN_STEP_2,
+            data={'username': self.username, 'password': self.password},
+        )
+        response.raise_for_status()
+
+        self.is_logged_in = True
 
     def get_depo(self) -> Depo:
+        self.login()
         now = datetime.datetime.now(
             tz=datetime.UTC,
         )  # TODO: Use info from json requests
 
-        accounts = self.client.get_accounts()
+        json_accounts = self._get_json(API_ACCOUNTS)
 
         nested: list[Depo] = []
-        for account in accounts:
-            account_id = account.accid
+        for json_account in json_accounts:
+            account_id = json_account['accid']
 
             assets: dict[fin_defs.Asset, Decimal] = {}
 
             # Determine amount of usable currency stored on Nordnet
-            for account_info in self.client.get_account_info(account_id):
-                asset = fin_defs.FiatCurrency(
-                    account_info.account_sum.currency,
+            if True:
+                json_account_info = self._get_json(
+                    API_ACCOUNT_INFO.format(accid=account_id),
                 )
-                amount = Decimal(account_info.account_sum.value)
+                asset = fin_defs.FiatCurrency(
+                    json_account_info[0]['account_sum']['currency'],
+                )
+                amount = Decimal(json_account_info[0]['account_sum']['value'])
                 assets[asset] = amount
                 del asset, amount
 
             # Determine positions on Nordnet
-            for pos in self.client.get_account_positions(account_id):
-                asset = asset_from_instrument(pos.instrument)
+            json_account_positions = self._get_json(
+                API_ACCOUNT_POSITIONS.format(accid=account_id),
+            )
+            for pos in json_account_positions:
+                asset = asset_from_instrument_json(pos['instrument'])
                 if asset is None:
                     continue
-                amount = Decimal(pos.qty)
+                amount = Decimal(pos['qty'])
                 assets[asset] = amount
                 del asset, amount
 
             nested.append(
                 DepoSingle(
-                    name=f'{account.type} {account.alias}',
+                    name='{} {}'.format(json_account['type'], json_account['alias']),
                     _assets=assets,
                     updated_time=now,
                 ),

fin_depo/static.py

@@ -4,18 +4,18 @@ This module does not represent an API or other integration. This module
 contains aggregators and static depositories.
 """
 
-import dataclasses
 import datetime
 import logging
 from decimal import Decimal
 
 import fin_defs
+import krakenex
+import dataclasses
 
-from .data import DepoFetcher, DepoGroup, DepoSingle
+from .data import Depo, DepoFetcher, DepoSingle, DepoGroup
 
 logger = logging.getLogger(__name__)
 
-
 @dataclasses.dataclass(frozen=True)
 class StaticDepoFetcher(DepoFetcher):
     """Depository "fetcher" that doesn't do any fetching.
@@ -25,9 +25,7 @@ class StaticDepoFetcher(DepoFetcher):
 
     name: str
     depo_assets: dict[fin_defs.Asset,Decimal]
-    last_updated: datetime.datetime = dataclasses.field(
-        default_factory=lambda: datetime.datetime.now(tz=datetime.UTC),
-    )
+    last_updated: datetime.datetime = dataclasses.field(default_factory=lambda: datetime.datetime.now(tz=datetime.UTC))
 
     def get_depo(self) -> DepoSingle:
         return DepoSingle(
@@ -36,7 +34,6 @@ class StaticDepoFetcher(DepoFetcher):
             updated_time=self.last_updated,
         )
 
-
 @dataclasses.dataclass(frozen=True)
 class AggregateDepoFetcher(DepoFetcher):
     """Depository "fetcher" that delegates to the aggregated fetchers."""

test/test_kraken.py

@@ -1,4 +1,5 @@
 import pytest
+import requests
 
 import fin_depo
 

test/test_kucoin.py

@@ -1,8 +1,8 @@
-import datetime
 from decimal import Decimal
 
 import fin_defs
 import pytest
+import datetime
 
 import fin_depo
 
@@ -19,7 +19,6 @@ fin_depo.defi_kucoin.logger.setLevel('INFO')
 
 NOW = datetime.datetime.now(tz=datetime.UTC)
 
-
 @needs_secrets
 def test_get_depo():
     """Can inspect kucoin depository."""
@@ -87,7 +86,6 @@ def test_place_buy_side_order():
 
     assert NOW <= order_details.executed_time <= NOW + datetime.timedelta(minutes=10)
 
-
 @needs_secrets
 def test_place_sell_side_order():
     """Client can place sell side market orders."""

test/test_static.py

@@ -1,13 +1,11 @@
-from decimal import Decimal
-
 import fin_defs
 
 import fin_depo
-
+from decimal import Decimal
 
 def test_get_depo():
     fetcher = fin_depo.static.StaticDepoFetcher(
-        'Test', {fin_defs.BTC: Decimal(1000), fin_defs.USD: Decimal(2000)},
+            'Test', {fin_defs.BTC: Decimal(1000), fin_defs.USD: Decimal(2000)}
     )
 
     depo = fetcher.get_depo()