1
0

Use nordnet client

This commit is contained in:
Jon Michael Aanes 2024-07-27 03:13:50 +02:00
parent 4340d3ece2
commit 5f89c5d6df
Signed by: Jmaa
SSH Key Fingerprint: SHA256:Ab0GfHGCblESJx7JRE4fj4bFy/KRpeLhi41y4pF3sNA

View File

@ -14,6 +14,7 @@ import logging
from collections.abc import Mapping from collections.abc import Mapping
from decimal import Decimal from decimal import Decimal
from typing import Any from typing import Any
from nordnet_api_client import NordnetClient
import fin_defs import fin_defs
import requests import requests
@ -23,23 +24,12 @@ from .data import Depo, DepoFetcher, DepoGroup, DepoSingle
logger = logging.getLogger(__name__) 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' def asset_from_instrument(intrument) -> fin_defs.Asset | None:
API_LOGIN_STEP_2 = API_ROOT + '/authentication/basic/login' if intrument.instrument_group_type == 'FND':
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 return None
symbol = json['symbol'] symbol = intrument.symbol
exchange_id = json['tradables'][0]['mic'] exchange_id = intrument.tradables[0].mic
return fin_defs.Stock(symbol, fin_defs.EXCHANGES_BY_IDS[exchange_id]) return fin_defs.Stock(symbol, fin_defs.EXCHANGES_BY_IDS[exchange_id])
@ -63,88 +53,42 @@ class NordnetDepoFetcher(DepoFetcher):
""" """
def __init__(self, session: requests.Session, username: str, password: str): def __init__(self, session: requests.Session, username: str, password: str):
self.session = self.assert_param('session', requests.Session, session) self.client = NordnetClient(session, username, password)
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: def get_depo(self) -> Depo:
self.login()
now = datetime.datetime.now( now = datetime.datetime.now(
tz=datetime.UTC, tz=datetime.UTC,
) # TODO: Use info from json requests ) # TODO: Use info from json requests
json_accounts = self._get_json(API_ACCOUNTS) accounts = self.client.get_accounts()
nested: list[Depo] = [] nested: list[Depo] = []
for json_account in json_accounts: for account in accounts:
account_id = json_account['accid'] account_id = account.accid
assets: dict[fin_defs.Asset, Decimal] = {} assets: dict[fin_defs.Asset, Decimal] = {}
# Determine amount of usable currency stored on Nordnet # Determine amount of usable currency stored on Nordnet
if True: for account_info in self.client.get_account_info(account_id):
json_account_info = self._get_json(
API_ACCOUNT_INFO.format(accid=account_id),
)
asset = fin_defs.FiatCurrency( asset = fin_defs.FiatCurrency(
json_account_info[0]['account_sum']['currency'], account_info.account_sum.currency,
) )
amount = Decimal(json_account_info[0]['account_sum']['value']) amount = Decimal(account_info.account_sum.value)
assets[asset] = amount assets[asset] = amount
del asset, amount del asset, amount
# Determine positions on Nordnet # Determine positions on Nordnet
json_account_positions = self._get_json( for pos in self.client.get_account_positions(account_id):
API_ACCOUNT_POSITIONS.format(accid=account_id), asset = asset_from_instrument(pos.instrument)
)
for pos in json_account_positions:
asset = asset_from_instrument_json(pos['instrument'])
if asset is None: if asset is None:
continue continue
amount = Decimal(pos['qty']) amount = Decimal(pos.qty)
assets[asset] = amount assets[asset] = amount
del asset, amount del asset, amount
nested.append( nested.append(
DepoSingle( DepoSingle(
name='{} {}'.format(json_account['type'], json_account['alias']), name=f'{account.type} {account.alias}',
_assets=assets, _assets=assets,
updated_time=now, updated_time=now,
), ),