1
0
fin-depo/fin_depo/investbank_nordnet.py

156 lines
5.3 KiB
Python
Raw Normal View History

2024-06-20 21:43:45 +00:00
"""Nordnet Depository fetching for [Nordnet](https://nordnet.dk), the online scandinavian investment bank.
2024-06-04 19:30:42 +00:00
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
2024-06-04 20:15:21 +00:00
from .data import Depo, DepoGroup, DepoSingle
2024-06-04 19:30:42 +00:00
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'
def asset_from_instrument_json(json) -> fin_defs.Asset | None:
if json['instrument_group_type'] == 'FND':
return None
symbol = json['symbol']
exchange_id = json['tradables'][0]['mic']
return fin_defs.Stock(symbol, fin_defs.EXCHANGES_BY_IDS[exchange_id])
2024-06-04 20:15:21 +00:00
2024-06-04 19:30:42 +00:00
class NordnetDepoFetcher:
2024-06-20 21:43:45 +00:00
"""Depository fetcher for [Nordnet](https://nordnet.dk), the online investment bank.
Requirements for use:
- Account on [Nordnet](https://nordnet.dk). This requires a MitID.
- Password login enabled from [settings](https://www.nordnet.dk/indstillinger/min-profil)
**Warning**: This system uses an unofficial API which uses your normal
Nordnet username and password for login access, with full read/write
access! **This is dangerous**, and any potential leak would give an
attacker full access to your account.
Depository structure: Each account you have access to will be given its own
`Depo`, with all of them nested under a "Nordnet" nested depository.
"""
2024-06-04 19:30:42 +00:00
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
2024-07-07 22:48:27 +00:00
assert self.username is not None, 'Username is missing'
assert self.password is not None, 'Password is missing'
def get_json(self, url: str, params: dict[str, str | int] = {}) -> dict:
2024-06-04 19:30:42 +00:00
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()
2024-06-04 20:15:21 +00:00
now = datetime.datetime.now(
tz=datetime.UTC,
) # TODO: Use info from json requests
json_accounts = self.get_json(API_ACCOUNTS)
nested = []
for json_account in json_accounts:
account_id = json_account['accid']
2024-06-04 20:15:21 +00:00
assets: dict[fin_defs.Asset, Decimal] = {}
# Determine amount of usable currency stored on Nordnet
if True:
2024-06-04 20:15:21 +00:00
json_account_info = self.get_json(
API_ACCOUNT_INFO.format(accid=account_id),
)
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
2024-06-04 20:15:21 +00:00
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'])
2024-06-04 20:15:21 +00:00
if asset is None:
continue
amount = Decimal(pos['qty'])
assets[asset] = amount
del asset, amount
2024-06-04 20:15:21 +00:00
nested.append(
DepoSingle(
name='{} {}'.format(json_account['type'], json_account['alias']),
_assets=assets,
updated_time=now,
),
)
return DepoGroup(
name=f'Nordnet',
2024-06-04 19:30:42 +00:00
updated_time=now,
nested=nested,
2024-06-04 19:30:42 +00:00
)