1
0

Compare commits

...

2 Commits

Author SHA1 Message Date
bcae3d4e29
Ruff
Some checks failed
Test Python / Test (push) Failing after 26s
2024-07-27 03:14:12 +02:00
5f89c5d6df
Use nordnet client 2024-07-27 03:13:50 +02:00
6 changed files with 43 additions and 92 deletions

View File

@ -2,9 +2,9 @@
import datetime import datetime
import logging import logging
import time
from decimal import Decimal from decimal import Decimal
import time
import fin_defs import fin_defs
import kucoin.client import kucoin.client
@ -138,9 +138,9 @@ class KucoinDepoFetcher(DepoFetcher):
# Determine order details # Determine order details
return self._get_order_details(response['orderId'], input_asset, output_asset) return self._get_order_details(response['orderId'], input_asset, output_asset)
def _get_order_details(self, order_id: str, def _get_order_details(
input_asset: fin_defs.Asset, self, order_id: str, input_asset: fin_defs.Asset, output_asset: fin_defs.Asset,
output_asset: fin_defs.Asset) -> TradeOrderDetails: ) -> TradeOrderDetails:
"""Determine the order details for the order with the given id. """Determine the order details for the order with the given id.
Retries the order a few times, as KuCoin might not have propagated the Retries the order a few times, as KuCoin might not have propagated the
@ -164,13 +164,15 @@ class KucoinDepoFetcher(DepoFetcher):
fee_asset=fin_defs.WELL_KNOWN_SYMBOLS[order_details['feeCurrency']], fee_asset=fin_defs.WELL_KNOWN_SYMBOLS[order_details['feeCurrency']],
fee_amount=Decimal(order_details['fee']), fee_amount=Decimal(order_details['fee']),
executed_time=datetime.datetime.fromtimestamp( executed_time=datetime.datetime.fromtimestamp(
order_details['createdAt'] / 1000, tz=datetime.UTC order_details['createdAt'] / 1000, tz=datetime.UTC,
), ),
order_id=order_id, order_id=order_id,
raw_order_details=order_details, raw_order_details=order_details,
) )
def _get_order_with_retries(self, order_id: str, *, num_retries: int, sleep_between_tries:float = 1.0) -> dict: def _get_order_with_retries(
self, order_id: str, *, num_retries: int, sleep_between_tries: float = 1.0,
) -> dict:
"""Get the order details from KuCoin backend. """Get the order details from KuCoin backend.
Retries the order a few times, as KuCoin might not have propagated the Retries the order a few times, as KuCoin might not have propagated the
@ -179,6 +181,6 @@ class KucoinDepoFetcher(DepoFetcher):
for _ in range(num_retries): for _ in range(num_retries):
try: try:
return self.kucoin_client.get_order(order_id) return self.kucoin_client.get_order(order_id)
except kucoin.exceptions.KucoinAPIException as e: # noqa except kucoin.exceptions.KucoinAPIException as e: # noqa
time.sleep(sleep_between_tries) time.sleep(sleep_between_tries)
return self.kucoin_client.get_order(order_id) return self.kucoin_client.get_order(order_id)

View File

@ -13,33 +13,22 @@ import datetime
import logging import logging
from collections.abc import Mapping from collections.abc import Mapping
from decimal import Decimal from decimal import Decimal
from typing import Any
import fin_defs import fin_defs
import requests import requests
from frozendict import frozendict from frozendict import frozendict
from nordnet_api_client import NordnetClient
from .data import Depo, DepoFetcher, DepoGroup, DepoSingle 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 +52,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,
), ),

View File

@ -4,18 +4,18 @@ This module does not represent an API or other integration. This module
contains aggregators and static depositories. contains aggregators and static depositories.
""" """
import dataclasses
import datetime import datetime
import logging import logging
from decimal import Decimal from decimal import Decimal
import fin_defs import fin_defs
import krakenex
import dataclasses
from .data import Depo, DepoFetcher, DepoSingle, DepoGroup from .data import DepoFetcher, DepoGroup, DepoSingle
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@dataclasses.dataclass(frozen=True) @dataclasses.dataclass(frozen=True)
class StaticDepoFetcher(DepoFetcher): class StaticDepoFetcher(DepoFetcher):
"""Depository "fetcher" that doesn't do any fetching. """Depository "fetcher" that doesn't do any fetching.
@ -24,8 +24,10 @@ class StaticDepoFetcher(DepoFetcher):
""" """
name: str name: str
depo_assets: dict[fin_defs.Asset,Decimal] depo_assets: dict[fin_defs.Asset, Decimal]
last_updated: datetime.datetime = dataclasses.field(default_factory=lambda: datetime.datetime.now(tz=datetime.UTC)) last_updated: datetime.datetime = dataclasses.field(
default_factory=lambda: datetime.datetime.now(tz=datetime.UTC),
)
def get_depo(self) -> DepoSingle: def get_depo(self) -> DepoSingle:
return DepoSingle( return DepoSingle(
@ -34,6 +36,7 @@ class StaticDepoFetcher(DepoFetcher):
updated_time=self.last_updated, updated_time=self.last_updated,
) )
@dataclasses.dataclass(frozen=True) @dataclasses.dataclass(frozen=True)
class AggregateDepoFetcher(DepoFetcher): class AggregateDepoFetcher(DepoFetcher):
"""Depository "fetcher" that delegates to the aggregated fetchers.""" """Depository "fetcher" that delegates to the aggregated fetchers."""
@ -43,7 +46,7 @@ class AggregateDepoFetcher(DepoFetcher):
def get_depo(self) -> DepoGroup: def get_depo(self) -> DepoGroup:
return DepoGroup( return DepoGroup(
name=self.name, name=self.name,
nested=[fetcher.get_depo() for fetcher in self.aggregated], nested=[fetcher.get_depo() for fetcher in self.aggregated],
updated_time=datetime.datetime.now(tz=datetime.UTC), # TODO updated_time=datetime.datetime.now(tz=datetime.UTC), # TODO
) )

View File

@ -1,5 +1,4 @@
import pytest import pytest
import requests
import fin_depo import fin_depo

View File

@ -1,8 +1,8 @@
import datetime
from decimal import Decimal from decimal import Decimal
import fin_defs import fin_defs
import pytest import pytest
import datetime
import fin_depo import fin_depo
@ -19,6 +19,7 @@ fin_depo.defi_kucoin.logger.setLevel('INFO')
NOW = datetime.datetime.now(tz=datetime.UTC) NOW = datetime.datetime.now(tz=datetime.UTC)
@needs_secrets @needs_secrets
def test_get_depo(): def test_get_depo():
"""Can inspect kucoin depository.""" """Can inspect kucoin depository."""
@ -86,6 +87,7 @@ def test_place_buy_side_order():
assert NOW <= order_details.executed_time <= NOW + datetime.timedelta(minutes=10) assert NOW <= order_details.executed_time <= NOW + datetime.timedelta(minutes=10)
@needs_secrets @needs_secrets
def test_place_sell_side_order(): def test_place_sell_side_order():
"""Client can place sell side market orders.""" """Client can place sell side market orders."""

View File

@ -1,11 +1,13 @@
from decimal import Decimal
import fin_defs import fin_defs
import fin_depo import fin_depo
from decimal import Decimal
def test_get_depo(): def test_get_depo():
fetcher = fin_depo.static.StaticDepoFetcher( fetcher = fin_depo.static.StaticDepoFetcher(
'Test', {fin_defs.BTC: Decimal(1000), fin_defs.USD: Decimal(2000)} 'Test', {fin_defs.BTC: Decimal(1000), fin_defs.USD: Decimal(2000)},
) )
depo = fetcher.get_depo() depo = fetcher.get_depo()