Compare commits
No commits in common. "1870de8109b31087ea9ee14e1b9f87d4c57c1ed8" and "075db4aefced136c9c0c7b1b0958c175b496a3cc" have entirely different histories.
1870de8109
...
075db4aefc
|
@ -129,13 +129,12 @@ 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'
|
||||
|
@ -147,9 +146,7 @@ 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,
|
||||
|
|
|
@ -1,125 +0,0 @@
|
|||
"""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, DepoGroup
|
||||
|
||||
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])
|
||||
|
||||
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] = {}) -> dict:
|
||||
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()
|
||||
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']
|
||||
|
||||
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=f'Nordnet',
|
||||
updated_time=now,
|
||||
nested=nested,
|
||||
)
|
Loading…
Reference in New Issue
Block a user