diff --git a/fin_depo/_version.py b/fin_depo/_version.py index e2888cc..66a87bb 100644 --- a/fin_depo/_version.py +++ b/fin_depo/_version.py @@ -1 +1 @@ -__version__ = '0.1.5' \ No newline at end of file +__version__ = '0.1.5' diff --git a/fin_depo/defi_partisia_blockchain.py b/fin_depo/defi_partisia_blockchain.py index 1d40368..069a4f9 100644 --- a/fin_depo/defi_partisia_blockchain.py +++ b/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, diff --git a/fin_depo/investbank_nordnet.py b/fin_depo/investbank_nordnet.py new file mode 100644 index 0000000..d73eec9 --- /dev/null +++ b/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: + + """ + 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, + )