1
0

Compare commits

..

No commits in common. "bcae3d4e29743ddb4daf3bcbc0139f7846c1d233" and "4340d3ece2c64017f5e2f55335b0e7ae01470705" have entirely different histories.

6 changed files with 93 additions and 44 deletions

View File

@ -2,9 +2,9 @@
import datetime
import logging
import time
from decimal import Decimal
import time
import fin_defs
import kucoin.client
@ -138,9 +138,9 @@ class KucoinDepoFetcher(DepoFetcher):
# Determine order details
return self._get_order_details(response['orderId'], input_asset, output_asset)
def _get_order_details(
self, order_id: str, input_asset: fin_defs.Asset, output_asset: fin_defs.Asset,
) -> TradeOrderDetails:
def _get_order_details(self, order_id: str,
input_asset: fin_defs.Asset,
output_asset: fin_defs.Asset) -> TradeOrderDetails:
"""Determine the order details for the order with the given id.
Retries the order a few times, as KuCoin might not have propagated the
@ -164,15 +164,13 @@ class KucoinDepoFetcher(DepoFetcher):
fee_asset=fin_defs.WELL_KNOWN_SYMBOLS[order_details['feeCurrency']],
fee_amount=Decimal(order_details['fee']),
executed_time=datetime.datetime.fromtimestamp(
order_details['createdAt'] / 1000, tz=datetime.UTC,
order_details['createdAt'] / 1000, tz=datetime.UTC
),
order_id=order_id,
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.
Retries the order a few times, as KuCoin might not have propagated the

View File

@ -13,22 +13,33 @@ 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 nordnet_api_client import NordnetClient
from .data import Depo, DepoFetcher, DepoGroup, DepoSingle
logger = logging.getLogger(__name__)
API_HOSTNAME = 'public.nordnet.dk'
API_ROOT = f'https://{API_HOSTNAME}/api/2'
def asset_from_instrument(intrument) -> fin_defs.Asset | None:
if intrument.instrument_group_type == 'FND':
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 = intrument.symbol
exchange_id = intrument.tradables[0].mic
symbol = json['symbol']
exchange_id = json['tradables'][0]['mic']
return fin_defs.Stock(symbol, fin_defs.EXCHANGES_BY_IDS[exchange_id])
@ -52,42 +63,88 @@ class NordnetDepoFetcher(DepoFetcher):
"""
def __init__(self, session: requests.Session, username: str, password: str):
self.client = NordnetClient(session, username, password)
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
accounts = self.client.get_accounts()
json_accounts = self._get_json(API_ACCOUNTS)
nested: list[Depo] = []
for account in accounts:
account_id = account.accid
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
for account_info in self.client.get_account_info(account_id):
asset = fin_defs.FiatCurrency(
account_info.account_sum.currency,
if True:
json_account_info = self._get_json(
API_ACCOUNT_INFO.format(accid=account_id),
)
amount = Decimal(account_info.account_sum.value)
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
for pos in self.client.get_account_positions(account_id):
asset = asset_from_instrument(pos.instrument)
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)
amount = Decimal(pos['qty'])
assets[asset] = amount
del asset, amount
nested.append(
DepoSingle(
name=f'{account.type} {account.alias}',
name='{} {}'.format(json_account['type'], json_account['alias']),
_assets=assets,
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.
"""
import dataclasses
import datetime
import logging
from decimal import Decimal
import fin_defs
import krakenex
import dataclasses
from .data import DepoFetcher, DepoGroup, DepoSingle
from .data import Depo, DepoFetcher, DepoSingle, DepoGroup
logger = logging.getLogger(__name__)
@dataclasses.dataclass(frozen=True)
class StaticDepoFetcher(DepoFetcher):
"""Depository "fetcher" that doesn't do any fetching.
@ -24,10 +24,8 @@ class StaticDepoFetcher(DepoFetcher):
"""
name: str
depo_assets: dict[fin_defs.Asset, Decimal]
last_updated: datetime.datetime = dataclasses.field(
default_factory=lambda: datetime.datetime.now(tz=datetime.UTC),
)
depo_assets: dict[fin_defs.Asset,Decimal]
last_updated: datetime.datetime = dataclasses.field(default_factory=lambda: datetime.datetime.now(tz=datetime.UTC))
def get_depo(self) -> DepoSingle:
return DepoSingle(
@ -36,7 +34,6 @@ class StaticDepoFetcher(DepoFetcher):
updated_time=self.last_updated,
)
@dataclasses.dataclass(frozen=True)
class AggregateDepoFetcher(DepoFetcher):
"""Depository "fetcher" that delegates to the aggregated fetchers."""

View File

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

View File

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

View File

@ -1,13 +1,11 @@
from decimal import Decimal
import fin_defs
import fin_depo
from decimal import Decimal
def test_get_depo():
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()