fin-depo/fin_depo/investbank_nordnet.py

"""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
from typing import Any

import fin_defs
import requests
from frozendict import frozendict

from .data import Depo, DepoFetcher, DepoGroup, 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'


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])


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.session = self.assert_param('session', requests.Session, session)
        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:
        self.login()
        now = datetime.datetime.now(
            tz=datetime.UTC,
        )  # TODO: Use info from json requests

        json_accounts = self._get_json(API_ACCOUNTS)

        nested: list[Depo] = []
        for json_account in json_accounts:
            account_id = json_account['accid']

            assets: dict[fin_defs.Asset, Decimal] = {}

            # Determine amount of usable currency stored on Nordnet
            if True:
                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
            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'])
                if asset is None:
                    continue
                amount = Decimal(pos['qty'])
                assets[asset] = amount
                del asset, amount

            nested.append(
                DepoSingle(
                    name='{} {}'.format(json_account['type'], json_account['alias']),
                    _assets=assets,
                    updated_time=now,
                ),
            )

        return DepoGroup(
            name='Nordnet',
            updated_time=now,
            nested=nested,
        )