Jmaa/fin-depo
1
0
Fork 0
Code Actions Packages

Ruff

Browse Source
This commit is contained in:
Jon Michael Aanes 2024-07-19 00:22:54 +02:00
parent 88a57bbc6c
commit ef417c6175
Signed by: Jmaa
SSH Key Fingerprint: SHA256:Ab0GfHGCblESJx7JRE4fj4bFy/KRpeLhi41y4pF3sNA
5 changed files with 14 additions and 14 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
fin_depo/data.py
View File

@ -71,8 +71,10 @@ class DepoGroup(Depo):
del nested_depo del nested_depo
return summed return summed
T = TypeVar('T') T = TypeVar('T')
class DepoFetcher(abc.ABC): class DepoFetcher(abc.ABC):
"""Base `Depo` fetcher interface. """Base `Depo` fetcher interface.

10
fin_depo/defi_kraken.py
View File

@ -7,7 +7,7 @@ from decimal import Decimal
import fin_defs import fin_defs
import krakenex import krakenex
from .data import Depo, DepoSingle, DepoFetcher from .data import Depo, DepoFetcher, DepoSingle
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@ -29,8 +29,6 @@ class KrakenDepoFetcher(DepoFetcher):
def __init__(self, kraken_key: str, kraken_secret: str): def __init__(self, kraken_key: str, kraken_secret: str):
self.assert_param('kraken_key', str, kraken_key) self.assert_param('kraken_key', str, kraken_key)
self.assert_param('kraken_secret', str, kraken_secret) self.assert_param('kraken_secret', str, kraken_secret)
assert kraken_key is not None, 'Missing kraken_key'
assert kraken_secret is not None, 'Missing kraken_secret'
self.client = krakenex.API( self.client = krakenex.API(
kraken_key, kraken_key,
kraken_secret, kraken_secret,
@ -42,13 +40,13 @@ class KrakenDepoFetcher(DepoFetcher):
result = self.client.query_private('Balance') result = self.client.query_private('Balance')
assets: dict[fin_defs.Asset, Decimal] = {} assets: dict[fin_defs.Asset, Decimal] = {}
for account, balance_str in result['result'].items(): for account_raw, balance_str in result['result'].items():
account = account.removesuffix('.HOLD') account = account_raw.removesuffix('.HOLD')
asset = fin_defs.WELL_KNOWN_SYMBOLS[account] asset = fin_defs.WELL_KNOWN_SYMBOLS[account]
balance = Decimal(balance_str) balance = Decimal(balance_str)
if balance != 0: if balance != 0:
assets[asset] = assets.get(asset, Decimal(0)) + balance assets[asset] = assets.get(asset, Decimal(0)) + balance
del account, balance_str, asset, balance del account, account_raw, balance_str, asset, balance
return DepoSingle( return DepoSingle(
name='Kraken', name='Kraken',

2
fin_depo/defi_kucoin.py
View File

@ -7,7 +7,7 @@ from decimal import Decimal
import fin_defs import fin_defs
import kucoin.client import kucoin.client
from .data import DepoGroup, DepoSingle, DepoFetcher from .data import DepoFetcher, DepoGroup, DepoSingle
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)

3
fin_depo/defi_partisia_blockchain.py
View File

@ -2,7 +2,6 @@
import dataclasses import dataclasses
import datetime import datetime
import requests
import email.utils import email.utils
import json import json
import logging import logging
@ -13,7 +12,7 @@ import fin_defs
import requests import requests
from frozendict import frozendict from frozendict import frozendict
from .data import DepoGroup, DepoSingle, DepoFetcher from .data import DepoFetcher, DepoGroup, DepoSingle
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)

11
fin_depo/investbank_nordnet.py
View File

@ -11,13 +11,13 @@ I am grateful for his pioneering work.
import datetime import datetime
import logging import logging
import requests
from decimal import Decimal from decimal import Decimal
import fin_defs import fin_defs
import requests
from frozendict import frozendict from frozendict import frozendict
from .data import Depo, DepoGroup, DepoSingle, DepoFetcher from .data import Depo, DepoFetcher, DepoGroup, DepoSingle
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@ -40,7 +40,9 @@ def asset_from_instrument_json(json) -> fin_defs.Asset | None:
exchange_id = json['tradables'][0]['mic'] exchange_id = json['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])
EMPTY_DICT: dict[str,str | int] = frozendict()
EMPTY_DICT: dict[str, str | int] = frozendict()
class NordnetDepoFetcher(DepoFetcher): class NordnetDepoFetcher(DepoFetcher):
"""Depository fetcher for [Nordnet](https://nordnet.dk), the online investment bank. """Depository fetcher for [Nordnet](https://nordnet.dk), the online investment bank.
@ -59,13 +61,12 @@ 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.session = self.assert_param('session', requests.Session, session)
self.username: str = self.assert_param('username', str, username) self.username: str = self.assert_param('username', str, username)
self.password: str = self.assert_param('password', str, password) self.password: str = self.assert_param('password', str, password)
self.is_logged_in = False self.is_logged_in = False
def _get_json(self, url: str, params: dict[str, str | int] = EMPTY_DICT)-> dict: def _get_json(self, url: str, params: dict[str, str | int] = EMPTY_DICT) -> dict:
if not url.startswith(API_ROOT): if not url.startswith(API_ROOT):
msg = f'Given url must be located below API ROOT: {url}' msg = f'Given url must be located below API ROOT: {url}'
raise ValueError(msg) raise ValueError(msg)