Working on Nordnet

fin_depo/_version.py

@@ -1 +1 @@
-__version__ = '0.1.5'
+__version__ = '0.1.5'

fin_depo/defi_partisia_blockchain.py

@@ -129,12 +129,13 @@ class PbcClient:
 
 
 BYOC_ASSETS = {
-        'ETH': fin_defs.WELL_KNOWN_SYMBOLS['ETH'],
-        'POLYGON_USDC': fin_defs.WELL_KNOWN_SYMBOLS['USDC'],
-        'BNB': fin_defs.WELL_KNOWN_SYMBOLS['BNB'],
-        'MATIC': fin_defs.WELL_KNOWN_SYMBOLS['MATIC'],
+    'ETH': fin_defs.WELL_KNOWN_SYMBOLS['ETH'],
+    'POLYGON_USDC': fin_defs.WELL_KNOWN_SYMBOLS['USDC'],
+    'BNB': fin_defs.WELL_KNOWN_SYMBOLS['BNB'],
+    'MATIC': fin_defs.WELL_KNOWN_SYMBOLS['MATIC'],
 }
 
+
 class PartisiaBlockchainAccountDepoFetcher:
     def __init__(self, session, blockchain_address: str):
         assert session is not None, 'missing session'
@@ -146,7 +147,9 @@ class PartisiaBlockchainAccountDepoFetcher:
     def get_depo(self) -> DepoGroup:
         balances = self.client.get_account_balances(self.blockchain_address)
 
-        depo_mpc = DepoSingle('Native', balances.update_time, {fin_defs.MPC: balances.mpc})
+        depo_mpc = DepoSingle(
+            'Native', balances.update_time, {fin_defs.MPC: balances.mpc}
+        )
         depo_byoc = DepoSingle(
             'BYOC',
             balances.update_time,

fin_depo/investbank_nordnet.py

@@ -0,0 +1,102 @@
+"""Nordnet Depository fetching for Nordnet, the scandinavian investment bank.
+
+This fetcher uses the [Nordnet
+API](https://www.nordnet.dk/externalapi/docs/api), which requires a Nordnet
+account.
+
+Some of the code here is based on [Morten Helmstedt's Nordnet
+utilities](https://github.com/helmstedt/nordnet-utilities/blob/main/nordnet_login.py).
+I am grateful for his pioneering work.
+"""
+
+import datetime
+import logging
+from decimal import Decimal
+
+import fin_defs
+
+from .data import Depo, DepoSingle
+
+logger = logging.getLogger(__name__)
+
+API_HOSTNAME = 'public.nordnet.dk'
+API_ROOT = f'https://{API_HOSTNAME}/api/2'
+
+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'
+
+
+class NordnetDepoFetcher:
+    def __init__(self, session, username: str, password: str):
+        assert session is not None, 'missing session'
+        self.session = session
+        self.username = username
+        self.password = password
+        self.is_logged_in = False
+
+    def get_json(self, url: str, params: dict[str, str | int] = {}) -> object:
+        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):
+        """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()
+
+        accounts = self.get_json(API_ACCOUNTS)
+        print(accounts)
+
+
+
+        # Old
+        now = datetime.datetime.now(tz=datetime.UTC)
+        result = self.session.query_private('Balance')
+        assets: dict[fin_defs.Asset, Decimal] = {}
+        for account, balance_str in result['result'].items():
+            account = account.removesuffix('.HOLD')
+            asset = fin_defs.WELL_KNOWN_SYMBOLS[account]
+            balance = Decimal(balance_str)
+            if balance != 0:
+                assets[asset] = assets.get(asset, Decimal(0)) + balance
+            del account, balance_str, asset, balance
+
+        return DepoSingle(
+            name='nordnet',
+            _assets=assets,
+            updated_time=now,
+        )