1
0

Compare commits

...

2 Commits

Author SHA1 Message Date
eeffb23c98
Ruff
All checks were successful
Test Python / Test (push) Successful in 25s
2024-09-01 19:37:28 +02:00
fb07c495fa
Added attributes system for asset ids 2024-09-01 19:37:05 +02:00
3 changed files with 77 additions and 17 deletions

View File

@ -26,6 +26,22 @@ import enforce_typing
from ._version import __version__ # noqa: F401 from ._version import __version__ # noqa: F401
def parse_attr_data(attr_data: str) -> dict[str, str | int]:
d = {}
if attr_data is None:
return d
for s in attr_data.split(','):
s = s.strip()
if s == '':
continue
(attr_key, attr_value) = s.split('=')
if re.match(r'^\d+$', attr_value):
attr_value = int(attr_value)
d[attr_key] = attr_value
return d
## Ids ## Ids
RE_TICKER_FORMAT = r'^[A-Z0-9_]+$' RE_TICKER_FORMAT = r'^[A-Z0-9_]+$'
@ -114,11 +130,15 @@ class Asset:
def to_string_id(self) -> str: def to_string_id(self) -> str:
"""Formats the asset id using the fin_defs asset format.""" """Formats the asset id using the fin_defs asset format."""
if isinstance(self, Stock): if isinstance(self, Stock):
return f'stock:{self.ticker}.{self.exchange.mic}' attrs_str = f'{{nordnet_id={self.nordnet_id}}}' if self.nordnet_id else ''
return f'stock:{self.ticker}.{self.exchange.mic}{attrs_str}'
if isinstance(self, FiatCurrency): if isinstance(self, FiatCurrency):
return f'fiat:{self.iso_code}' return f'fiat:{self.iso_code}'
if isinstance(self, CryptoCurrency): if isinstance(self, CryptoCurrency):
return f'crypto:{self.ccxt_symbol}' attrs_str = (
f'{{coingecko_id={self.coingecko_id}}}' if self.coingecko_id else ''
)
return f'crypto:{self.ccxt_symbol}{attrs_str}'
if isinstance(self, Index): if isinstance(self, Index):
return f'index:{self.ticker}' return f'index:{self.ticker}'
if isinstance(self, Commodity): if isinstance(self, Commodity):
@ -128,28 +148,35 @@ class Asset:
raise NotImplementedError(msg) raise NotImplementedError(msg)
@staticmethod @staticmethod
def from_string_id(asset_id: str) -> 'Asset': def from_string_id(asset_id: str, shortcut_well_known=True) -> 'Asset':
"""Parses an `Asset` using the fin_defs asset format.""" """Parses an `Asset` using the fin_defs asset format."""
if ':' in asset_id:
prefix, ticker = asset_id.split(':') m = re.match(r'^(?:(\w+):)?([^{]+)(?:\{(.*)\})?$', asset_id)
else: if m is None:
prefix, ticker = None, asset_id msg = f'Unsupported asset format: {asset_id}'
raise NotImplementedError(msg)
prefix = m.group(1)
ticker = m.group(2)
attr_data = m.group(3)
if known_symbol := WELL_KNOWN_SYMBOLS.get(ticker, None): if known_symbol := WELL_KNOWN_SYMBOLS.get(ticker, None):
return known_symbol return known_symbol
attrs: dict[str, str | int] = parse_attr_data(attr_data)
if prefix == 'stock': if prefix == 'stock':
if m := re.match(r'^(\w+)(?:\.(\w+))?$', ticker): if m := re.match(r'^(\w+)(?:\.(\w+))?$', ticker):
exchange = EXCHANGES_BY_IDS[m.group(2) or 'NYSE'] # TODO? exchange = EXCHANGES_BY_IDS[m.group(2) or 'NYSE'] # TODO?
return Stock(m.group(1), exchange) return Stock(m.group(1), exchange, **attrs)
if prefix == 'currency': if prefix == 'currency':
return FiatCurrency(ticker) return FiatCurrency(ticker, **attrs)
if prefix == 'fiat': if prefix == 'fiat':
return FiatCurrency(ticker) return FiatCurrency(ticker, **attrs)
if prefix == 'index': if prefix == 'index':
return Index(ticker) return Index(ticker, **attrs)
if prefix == 'crypto': if prefix == 'crypto':
return CryptoCurrency(ticker, None) # TODO return CryptoCurrency(ticker, **attrs)
msg = f'Unsupported asset format: {asset_id}' msg = f'Unsupported asset format: {asset_id}'
raise NotImplementedError(msg) raise NotImplementedError(msg)
@ -182,7 +209,7 @@ class FiatCurrency(Currency):
@dataclasses.dataclass(frozen=True, eq=True) @dataclasses.dataclass(frozen=True, eq=True)
class CryptoCurrency(Currency): class CryptoCurrency(Currency):
ccxt_symbol: str ccxt_symbol: str
coingecko_id: str | None coingecko_id: str | None = None
def __post_init__(self): def __post_init__(self):
"""TODO """TODO
@ -444,7 +471,7 @@ class AssetAmount:
specificity = '2' if self.amount >= 0.10 else '3' specificity = '2' if self.amount >= 0.10 else '3'
prefix = ASSET_PREFIX.get(self.asset, '') prefix = ASSET_PREFIX.get(self.asset, '')
return ('{}{:.' + specificity + 'f} {}').format( 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): def __mul__(self, other: Decimal):

View File

@ -10,6 +10,7 @@ BAD_TICKERS = ['TEST:EUR', 'EUR:TEST']
def test_valid_tickers(ticker: str): def test_valid_tickers(ticker: str):
fin_defs.Stock(ticker, exchange=fin_defs.EXCHANGES_BY_IDS['NYSE']) fin_defs.Stock(ticker, exchange=fin_defs.EXCHANGES_BY_IDS['NYSE'])
@pytest.mark.parametrize('ticker', BAD_TICKERS) @pytest.mark.parametrize('ticker', BAD_TICKERS)
def test_bad_tickers(ticker: str): def test_bad_tickers(ticker: str):
with pytest.raises(ValueError): with pytest.raises(ValueError):

View File

@ -3,11 +3,43 @@ import pytest
import fin_defs import fin_defs
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 ') == {
'abc': 123,
'xyz': 'abc',
}
def test_from_nordnet():
derp = fin_defs.Asset.from_string_id('stock:NVO.NYSE{nordnet_id=123}')
assert isinstance(derp, fin_defs.Stock)
assert derp.ticker == 'NVO'
assert derp.nordnet_id == 123
@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
)
@pytest.mark.parametrize('asset', fin_defs.WELL_KNOWN_SYMBOLS.values()) @pytest.mark.parametrize('asset', fin_defs.WELL_KNOWN_SYMBOLS.values())
def test_to_from_string_id(asset: fin_defs.Asset): def test_to_from_string_id(asset: fin_defs.Asset):
assert fin_defs.Asset.from_string_id(asset.to_string_id()) == asset assert (
fin_defs.Asset.from_string_id(asset.to_string_id(), shortcut_well_known=False)
== asset
)
@pytest.mark.parametrize('asset', fin_defs.WELL_KNOWN_SYMBOLS.values()) @pytest.mark.parametrize('asset', fin_defs.WELL_KNOWN_SYMBOLS.values())
def test_to_from_polygon_id(asset: fin_defs.Asset): def test_to_from_polygon_id(asset: fin_defs.Asset):
if isinstance(asset, fin_defs.CryptoCurrency):
return
assert fin_defs.Asset.from_polygon_id(asset.to_polygon_id()) == asset assert fin_defs.Asset.from_polygon_id(asset.to_polygon_id()) == asset