Compare commits
No commits in common. "eeffb23c9838b4292291f5ea5080ff0af89576da" and "3a96b50b68e4365f42933cef857c0e80f1f59cca" have entirely different histories.
eeffb23c98
...
3a96b50b68
|
@ -26,22 +26,6 @@ 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_]+$'
|
||||||
|
@ -130,15 +114,11 @@ 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):
|
||||||
attrs_str = f'{{nordnet_id={self.nordnet_id}}}' if self.nordnet_id else ''
|
return f'stock:{self.ticker}.{self.exchange.mic}'
|
||||||
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):
|
||||||
attrs_str = (
|
return f'crypto:{self.ccxt_symbol}'
|
||||||
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):
|
||||||
|
@ -148,35 +128,28 @@ class Asset:
|
||||||
raise NotImplementedError(msg)
|
raise NotImplementedError(msg)
|
||||||
|
|
||||||
@staticmethod
|
@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."""
|
"""Parses an `Asset` using the fin_defs asset format."""
|
||||||
|
if ':' in asset_id:
|
||||||
m = re.match(r'^(?:(\w+):)?([^{]+)(?:\{(.*)\})?$', asset_id)
|
prefix, ticker = asset_id.split(':')
|
||||||
if m is None:
|
else:
|
||||||
msg = f'Unsupported asset format: {asset_id}'
|
prefix, ticker = None, 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, **attrs)
|
return Stock(m.group(1), exchange)
|
||||||
if prefix == 'currency':
|
if prefix == 'currency':
|
||||||
return FiatCurrency(ticker, **attrs)
|
return FiatCurrency(ticker)
|
||||||
if prefix == 'fiat':
|
if prefix == 'fiat':
|
||||||
return FiatCurrency(ticker, **attrs)
|
return FiatCurrency(ticker)
|
||||||
if prefix == 'index':
|
if prefix == 'index':
|
||||||
return Index(ticker, **attrs)
|
return Index(ticker)
|
||||||
if prefix == 'crypto':
|
if prefix == 'crypto':
|
||||||
return CryptoCurrency(ticker, **attrs)
|
return CryptoCurrency(ticker, None) # TODO
|
||||||
|
|
||||||
msg = f'Unsupported asset format: {asset_id}'
|
msg = f'Unsupported asset format: {asset_id}'
|
||||||
raise NotImplementedError(msg)
|
raise NotImplementedError(msg)
|
||||||
|
@ -209,7 +182,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 = None
|
coingecko_id: str | None
|
||||||
|
|
||||||
def __post_init__(self):
|
def __post_init__(self):
|
||||||
"""TODO
|
"""TODO
|
||||||
|
@ -471,7 +444,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):
|
||||||
|
|
|
@ -8,13 +8,12 @@ BAD_TICKERS = ['TEST:EUR', 'EUR:TEST']
|
||||||
|
|
||||||
@pytest.mark.parametrize('ticker', VALID_TICKERS)
|
@pytest.mark.parametrize('ticker', VALID_TICKERS)
|
||||||
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):
|
||||||
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)
|
||||||
|
|
|
@ -3,43 +3,11 @@ 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 (
|
assert fin_defs.Asset.from_string_id(asset.to_string_id()) == asset
|
||||||
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
|
||||||
|
|
Loading…
Reference in New Issue
Block a user