106 lines
3.5 KiB
Python
106 lines
3.5 KiB
Python
"""Nordnet Depository fetching for [Nordnet](https://nordnet.dk), the online 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 collections.abc import Mapping
|
|
from decimal import Decimal
|
|
|
|
import fin_defs
|
|
import requests
|
|
from frozendict import frozendict
|
|
from nordnet_api_client import NordnetClient
|
|
from nordnet_api_client.data import Instrument
|
|
|
|
from .data import Depo, DepoFetcher, DepoGroup, DepoSingle
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def asset_from_instrument(instrument: Instrument) -> fin_defs.Asset | None:
|
|
if instrument.instrument_group_type == 'FND':
|
|
return None
|
|
symbol = instrument.symbol
|
|
exchange_id = instrument.tradables[0].mic
|
|
return fin_defs.Stock(
|
|
symbol,
|
|
fin_defs.EXCHANGES_BY_IDS[exchange_id],
|
|
nordnet_id=instrument.instrument_id,
|
|
)
|
|
|
|
|
|
EMPTY_DICT: Mapping[str, str | int] = frozendict()
|
|
|
|
|
|
class NordnetDepoFetcher(DepoFetcher):
|
|
"""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.
|
|
"""
|
|
|
|
def __init__(self, session: requests.Session, username: str, password: str):
|
|
self.client = NordnetClient(session, username, password)
|
|
|
|
def get_depo(self) -> Depo:
|
|
now = datetime.datetime.now(
|
|
tz=datetime.UTC,
|
|
) # TODO: Use info from json requests
|
|
|
|
accounts = self.client.get_accounts()
|
|
|
|
nested: list[Depo] = []
|
|
for account in accounts:
|
|
account_id = 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,
|
|
)
|
|
amount = Decimal(account_info.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)
|
|
if asset is None:
|
|
continue
|
|
amount = Decimal(pos.qty)
|
|
assets[asset] = amount
|
|
del asset, amount
|
|
|
|
nested.append(
|
|
DepoSingle(
|
|
name=f'{account.type} {account.alias}',
|
|
_assets=assets,
|
|
updated_time=now,
|
|
),
|
|
)
|
|
|
|
return DepoGroup(
|
|
name='Nordnet',
|
|
updated_time=now,
|
|
nested=nested,
|
|
)
|