1
0

Code quality
All checks were successful
Python Ruff Code Quality / ruff (push) Successful in 22s
Run Python tests (through Pytest) / Test (push) Successful in 25s
Verify Python project can be installed, loaded and have version checked / Test (push) Successful in 22s

This commit is contained in:
Jon Michael Aanes 2024-10-27 16:18:21 +01:00
parent d501c4da9f
commit a12fcacf77
Signed by: Jmaa
SSH Key Fingerprint: SHA256:Ab0GfHGCblESJx7JRE4fj4bFy/KRpeLhi41y4pF3sNA
3 changed files with 79 additions and 66 deletions

View File

@ -20,25 +20,37 @@ import abc
import dataclasses
import datetime
import re
from collections.abc import Mapping
from decimal import Decimal
import enforce_typing
from ._version import __version__ # noqa: F401
from ._version import __version__
__all__ = ['__version__']
def parse_attr_data(attr_data: str) -> dict[str, str | int]:
d = {}
def parse_id_attr_key_value_pair(attr_datum: str) -> tuple[str, str | int] | None:
attr_datum = attr_datum.strip()
if attr_datum == '':
return None
(attr_key, attr_value) = attr_datum.split('=')
if re.match(r'^\d+$', attr_value):
attr_value = int(attr_value)
return attr_key, attr_value
def parse_id_attr_data(attr_data: str) -> Mapping[str, str | int]:
if attr_data is None:
return d
for s in attr_data.split(','):
s = s.strip()
if s == '':
return {}
key_value_pairs_str = attr_data.split(',')
key_value_pairs = [parse_id_attr_key_value_pair(kvp) for kvp in key_value_pairs_str]
d = {}
for kvp in key_value_pairs:
if kvp is None:
continue
(attr_key, attr_value) = s.split('=')
if re.match(r'^\d+$', attr_value):
attr_value = int(attr_value)
d[attr_key] = attr_value
d[kvp[0]] = kvp[1]
return d
@ -124,8 +136,10 @@ class Asset:
@abc.abstractmethod
def raw_short_name(self) -> str:
"""A short name or ticker that does not nessisarily uniquely identify
the asset, but may be commonly used within the industry."""
"""Short name or ticker.
Does not nessisarily uniquely identify the asset, but may be commonly used within the industry.
"""
def to_string_id(self) -> str:
"""Formats the asset id using the fin_defs asset format."""
@ -146,13 +160,12 @@ class Asset:
if isinstance(self, UnknownAsset):
return 'unknown:XXX'
msg = f'Unsupported asset type: {repr(self)}'
msg = f'Unsupported asset type: {self:?}'
raise ValueError(msg)
@staticmethod
def from_string_id(asset_id: str, shortcut_well_known=True) -> 'Asset':
def from_string_id(asset_id: str) -> 'Asset':
"""Parses an `Asset` using the fin_defs asset format."""
m = re.match(r'^(?:(\w+):)?([^{]+)(?:\{(.*)\})?$', asset_id)
if m is None:
msg = f'Unsupported asset format: {asset_id}'
@ -165,7 +178,7 @@ class Asset:
if known_symbol := WELL_KNOWN_SYMBOLS.get(ticker, None):
return known_symbol
attrs: dict[str, str | int] = parse_attr_data(attr_data)
attrs: dict[str, str | int] = parse_id_attr_data(attr_data)
if prefix == 'stock':
if m := re.match(r'^(\w+)(?:\.(\w+))?$', ticker):
@ -218,7 +231,7 @@ class FiatCurrency(Currency):
@staticmethod
def from_currency_symbol(symbol: str) -> 'Currency | None':
for (currency, currency_symbol) in CURRENCY_SYMBOLS.items():
for currency, currency_symbol in CURRENCY_SYMBOLS.items():
if currency_symbol == symbol:
return currency
return None
@ -226,6 +239,7 @@ class FiatCurrency(Currency):
def to_currency_symbol(self) -> str:
return CURRENCY_SYMBOLS[self]
FiatCurrency.DKK = FiatCurrency('DKK')
FiatCurrency.NOK = FiatCurrency('NOK')
FiatCurrency.USD = FiatCurrency('USD')
@ -233,6 +247,7 @@ FiatCurrency.EUR = FiatCurrency('EUR')
FiatCurrency.GBP = FiatCurrency('GBP')
FiatCurrency.JPY = FiatCurrency('JPY')
@enforce_typing.enforce_types
@dataclasses.dataclass(frozen=True, eq=True)
class CryptoCurrency(Currency):
@ -240,11 +255,9 @@ class CryptoCurrency(Currency):
coingecko_id: str | None = None
def __post_init__(self) -> None:
"""TODO
if not re.match(RE_CRYPTO_TICKER_FORMAT, self.ccxt_symbol):
if not re.match('^.*$', self.ccxt_symbol):
msg = f'ccxt_symbol was not in correct format: {self.ccxt_symbol}'
raise ValueError(msg)
"""
def raw_short_name(self) -> str:
return self.ccxt_symbol
@ -295,7 +308,7 @@ class Commodity(Asset):
raise ValueError(msg)
def raw_short_name(self) -> str:
return self.alpha_vantage_id # TODO
return self.alpha_vantage_id
Commodity.CRUDE_OIL_WTI = Commodity('WTI')
@ -369,7 +382,7 @@ CURRENCY_CODES = {
WELL_KNOWN_SYMBOLS = (
CURRENCY_CODES
CURRENCY_CODES
| {'XXBT': BTC}
| {k.raw_short_name(): k for k in COMMON_NAMES}
| {'SPX500': SPX, 'SP500': SPX, 'Nasdaq 100': NDX}
@ -383,7 +396,7 @@ CURRENCY_SYMBOLS: dict[Currency, str] = {
FiatCurrency.JPY: '¥',
}
ASSET_PREFIX: dict[Asset, str] = CURRENCY_SYMBOLS # TODO: Remove at some point.
ASSET_PREFIX: dict[Asset, str] = CURRENCY_SYMBOLS # TODO: Remove at some point.
NYSE = StockExchange(
name='New York Stock Exchange',
@ -485,6 +498,9 @@ class AssetInformation:
# TODO: FIGI types? (https://www.openfigi.com/)
THREE_DECIMALS_UNDER_THIS_AMOUNT = 0.10
@enforce_typing.enforce_types
@dataclasses.dataclass(frozen=True, eq=True, slots=True)
class AssetAmount:
@ -502,10 +518,12 @@ class AssetAmount:
return self.human_readable_str()
def human_readable_str(self) -> str:
specificity = '2' if self.amount >= 0.10 else '3'
specificity = '2' if self.amount >= THREE_DECIMALS_UNDER_THIS_AMOUNT else '3'
prefix = ASSET_PREFIX.get(self.asset, '')
return ('{}{:.' + specificity + 'f} {}').format(
prefix, self.amount, self.asset.raw_short_name(),
prefix,
self.amount,
self.asset.raw_short_name(),
)
def __mul__(self, other: Decimal) -> 'AssetAmount':
@ -584,6 +602,7 @@ class AssetAmount:
def __ge__(self, other: 'AssetAmount') -> bool:
return self.cmp(other) >= 0
AssetAmount.ZERO = AssetAmount(UnknownAsset(), Decimal(0))
@ -611,7 +630,10 @@ class ExchangeRateSample:
return self._average
return (self.high + self.low) / 2
def replace_timestamp(self, timestamp: datetime.date | datetime.datetime) -> 'ExchangeRateSample':
def replace_timestamp(
self,
timestamp: datetime.date | datetime.datetime,
) -> 'ExchangeRateSample':
return dataclasses.replace(self, timestamp=timestamp)
def invert_exchange_rate(self) -> 'ExchangeRateSample':

View File

@ -13,7 +13,7 @@ def test_valid_tickers(ticker: str):
@pytest.mark.parametrize('ticker', BAD_TICKERS)
def test_bad_tickers(ticker: str):
with pytest.raises(ValueError):
with pytest.raises(ValueError, match='ticker was not in correct format:.*'):
fin_defs.Stock(ticker, exchange=fin_defs.EXCHANGES_BY_IDS['NYSE'])
@ -23,5 +23,5 @@ def test_crypto_tickers(ticker):
def test_str():
NVO = fin_defs.Stock('NVO', fin_defs.EXCHANGES_BY_IDS['NYSE'])
assert str(NVO) == 'stock:NVO.XNYS'
asset = fin_defs.Stock('NVO', fin_defs.EXCHANGES_BY_IDS['NYSE'])
assert str(asset) == 'stock:NVO.XNYS'

View File

@ -3,32 +3,33 @@ import pytest
import fin_defs
VALID_IDS = [
'stock:NVO.NYSE',
'stock:NVO.NYSE{nordnet_id=16256554}',
'currency:USD',
'fiat:USD',
'index:NDX',
'crypto:BTC',
'crypto:BTC{coingecko_id=bitcoin}',
'commodity:ALUMINUM',
'stock:NVO.NYSE',
'stock:NVO.NYSE{nordnet_id=16256554}',
'currency:USD',
'fiat:USD',
'index:NDX',
'crypto:BTC',
'crypto:BTC{coingecko_id=bitcoin}',
'commodity:ALUMINUM',
]
INVALID_IDS = [
'NVO',
'NVO.NYSE',
'test:test',
'fiat:TEST:TEST',
'index:TEST:TEST',
'commodity:TEST:TEST',
'NVO',
'NVO.NYSE',
'test:test',
'fiat:TEST:TEST',
'index:TEST:TEST',
'commodity:TEST:TEST',
]
def test_parse_attr():
assert fin_defs.parse_attr_data('') == {}
assert fin_defs.parse_attr_data(' ') == {}
assert fin_defs.parse_attr_data('abc=abc') == {'abc': 'abc'}
assert fin_defs.parse_attr_data('abc=123') == {'abc': 123}
assert fin_defs.parse_attr_data('abc=123,xyz=abc') == {'abc': 123, 'xyz': 'abc'}
assert fin_defs.parse_attr_data(' abc=123 , xyz=abc ') == {
assert fin_defs.parse_id_attr_data('') == {}
assert fin_defs.parse_id_attr_data(' ') == {}
assert fin_defs.parse_id_attr_data('abc=abc') == {'abc': 'abc'}
assert fin_defs.parse_id_attr_data('abc=123') == {'abc': 123}
assert fin_defs.parse_id_attr_data('abc=123,xyz=abc') == {'abc': 123, 'xyz': 'abc'}
assert fin_defs.parse_id_attr_data(' abc=123 , xyz=abc ') == {
'abc': 123,
'xyz': 'abc',
}
@ -43,38 +44,28 @@ def test_from_nordnet():
@pytest.mark.parametrize('asset_id', VALID_IDS)
def test_from_string_id_shortcut(asset_id: str):
assert (
fin_defs.Asset.from_string_id(asset_id, shortcut_well_known=True)
)
assert fin_defs.Asset.from_string_id(asset_id)
@pytest.mark.parametrize('asset_id', VALID_IDS)
def test_from_string_id(asset_id: str):
assert (
fin_defs.Asset.from_string_id(asset_id, shortcut_well_known=False)
)
assert fin_defs.Asset.from_string_id(asset_id)
@pytest.mark.parametrize('asset_id', INVALID_IDS)
def test_from_string_id_invalid(asset_id: str):
with pytest.raises(ValueError) as excinfo:
fin_defs.Asset.from_string_id(asset_id, shortcut_well_known=False)
with pytest.raises(ValueError, match='.*format.*'):
fin_defs.Asset.from_string_id(asset_id)
@pytest.mark.parametrize('asset', fin_defs.WELL_KNOWN_SYMBOLS.values())
def test_to_from_string_id_shortcut(asset: fin_defs.Asset):
assert (
fin_defs.Asset.from_string_id(asset.to_string_id(), shortcut_well_known=True)
== asset
)
assert fin_defs.Asset.from_string_id(asset.to_string_id()) == asset
@pytest.mark.parametrize('asset', fin_defs.WELL_KNOWN_SYMBOLS.values())
def test_to_from_string_id(asset: fin_defs.Asset):
assert (
fin_defs.Asset.from_string_id(asset.to_string_id(), shortcut_well_known=False)
== asset
)
assert fin_defs.Asset.from_string_id(asset.to_string_id()) == asset
@pytest.mark.parametrize('asset', fin_defs.WELL_KNOWN_SYMBOLS.values())