1
0

Compare commits

..

No commits in common. "a3227702054b1aa2021f22526d10d9ecc231f40e" and "a12fcacf77cb6739774b026a2bdc5c0aa7d1e7ef" have entirely different histories.

6 changed files with 51 additions and 258 deletions

View File

@ -27,7 +27,7 @@ import enforce_typing
from ._version import __version__ from ._version import __version__
__all__ = ['__version__', 'StockExchange', 'AssetAmount', 'Asset', 'ExchangeRateSample'] __all__ = ['__version__']
def parse_id_attr_key_value_pair(attr_datum: str) -> tuple[str, str | int] | None: def parse_id_attr_key_value_pair(attr_datum: str) -> tuple[str, str | int] | None:
@ -113,8 +113,8 @@ class Asset:
return f'X:{self.ccxt_symbol}USD' return f'X:{self.ccxt_symbol}USD'
# Else # Else
msg = f'Unknown type: {type(self).__name__}' msg = f'Unknown self type: {self}'
raise TypeError(msg) raise Exception(msg)
@staticmethod @staticmethod
def from_polygon_id(polygon_id: str) -> 'Asset': def from_polygon_id(polygon_id: str) -> 'Asset':
@ -160,8 +160,8 @@ class Asset:
if isinstance(self, UnknownAsset): if isinstance(self, UnknownAsset):
return 'unknown:XXX' return 'unknown:XXX'
msg = f'Unsupported asset type: {type(self).__name__}' msg = f'Unsupported asset type: {self:?}'
raise TypeError(msg) raise ValueError(msg)
@staticmethod @staticmethod
def from_string_id(asset_id: str) -> 'Asset': def from_string_id(asset_id: str) -> 'Asset':
@ -184,7 +184,9 @@ class Asset:
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, **attrs)
if prefix in {'currency', 'fiat'}: if prefix == 'currency':
return FiatCurrency(ticker, **attrs)
if prefix == 'fiat':
return FiatCurrency(ticker, **attrs) return FiatCurrency(ticker, **attrs)
if prefix == 'index': if prefix == 'index':
return Index(ticker, **attrs) return Index(ticker, **attrs)
@ -205,24 +207,18 @@ class Asset:
@enforce_typing.enforce_types @enforce_typing.enforce_types
@dataclasses.dataclass(frozen=True) @dataclasses.dataclass(frozen=True)
class UnknownAsset(Asset): class UnknownAsset(Asset):
"""An asset that does not exist.""" pass
@abc.abstractmethod
def raw_short_name(self) -> str:
return '???'
@enforce_typing.enforce_types @enforce_typing.enforce_types
@dataclasses.dataclass(frozen=True) @dataclasses.dataclass(frozen=True)
class Currency(Asset): class Currency(Asset):
"""Either a Fiat or a Crypto Currency.""" pass
@enforce_typing.enforce_types @enforce_typing.enforce_types
@dataclasses.dataclass(frozen=True, eq=True, order=True) @dataclasses.dataclass(frozen=True, eq=True, order=True)
class FiatCurrency(Currency): class FiatCurrency(Currency):
"""Fiat Currency."""
iso_code: str iso_code: str
def __post_init__(self) -> None: def __post_init__(self) -> None:
@ -255,13 +251,11 @@ FiatCurrency.JPY = FiatCurrency('JPY')
@enforce_typing.enforce_types @enforce_typing.enforce_types
@dataclasses.dataclass(frozen=True, eq=True) @dataclasses.dataclass(frozen=True, eq=True)
class CryptoCurrency(Currency): class CryptoCurrency(Currency):
"""Crypto Currency."""
ccxt_symbol: str ccxt_symbol: str
coingecko_id: str | None = None coingecko_id: str | None = None
def __post_init__(self) -> None: def __post_init__(self) -> None:
if not re.match(r'^\S.*\S$', self.ccxt_symbol): if not re.match('^.*$', self.ccxt_symbol):
msg = f'ccxt_symbol was not in correct format: {self.ccxt_symbol}' msg = f'ccxt_symbol was not in correct format: {self.ccxt_symbol}'
raise ValueError(msg) raise ValueError(msg)
@ -282,8 +276,8 @@ class Stock(Asset):
raise ValueError(msg) raise ValueError(msg)
@staticmethod @staticmethod
def from_polygon_id(polygon_id: str) -> 'Stock': def from_polygon_id(polygon_id: str, stock_exchange: StockExchange) -> 'Stock':
return Stock(polygon_id, NYSE) # TODO: Add support for non-NYSE exchanges. return Stock(polygon_id, stock_exchange)
def raw_short_name(self) -> str: def raw_short_name(self) -> str:
return self.ticker return self.ticker
@ -476,6 +470,9 @@ if True:
def add_by_id(exchange: StockExchange, key: str): def add_by_id(exchange: StockExchange, key: str):
exchange_id = getattr(exchange, key) exchange_id = getattr(exchange, key)
if exchange_id is not None: if exchange_id is not None:
if exchange_id in EXCHANGES_BY_IDS:
msg = f'Exchange {exchange_id} is already present'
raise ValueError(msg)
EXCHANGES_BY_IDS[exchange_id] = exchange EXCHANGES_BY_IDS[exchange_id] = exchange
for exchange in EXCHANGES: for exchange in EXCHANGES:
@ -521,25 +518,23 @@ class AssetAmount:
return self.human_readable_str() return self.human_readable_str()
def human_readable_str(self) -> str: def human_readable_str(self) -> str:
abs_amount = abs(self.amount) specificity = '2' if self.amount >= THREE_DECIMALS_UNDER_THIS_AMOUNT else '3'
specificity = '2' if abs_amount >= THREE_DECIMALS_UNDER_THIS_AMOUNT else '3'
prefix = ASSET_PREFIX.get(self.asset, '') prefix = ASSET_PREFIX.get(self.asset, '')
return ('{sign}{prefix}{amount:.' + specificity + 'f} {name}').format( return ('{}{:.' + specificity + 'f} {}').format(
sign='-' if self.amount < 0 else '', prefix,
prefix=prefix, self.amount,
amount=abs_amount, self.asset.raw_short_name(),
name=self.asset.raw_short_name(),
) )
def __mul__(self, other: Decimal) -> 'AssetAmount': def __mul__(self, other: Decimal) -> 'AssetAmount':
if not isinstance(other, Decimal): if not isinstance(other, Decimal):
msg = f'other must be Decimal, but was {type(other).__name__}' msg = f'other must be decimal, but was {type(other)}'
raise TypeError(msg) raise TypeError(msg)
return AssetAmount(self.asset, self.amount * other) return AssetAmount(self.asset, self.amount * other)
def __add__(self, other: 'AssetAmount') -> 'AssetAmount': def __add__(self, other: 'AssetAmount') -> 'AssetAmount':
if not isinstance(other, AssetAmount): if not isinstance(other, AssetAmount):
msg = f'other must be AssetAmount, but was {type(other).__name__}' msg = f'other must be AssetAmount, but was {type(other)}'
raise TypeError(msg) raise TypeError(msg)
if self.is_zero(): if self.is_zero():
return other return other
@ -554,9 +549,8 @@ class AssetAmount:
return AssetAmount(self.asset, self.amount + other.amount) return AssetAmount(self.asset, self.amount + other.amount)
def __sub__(self, other: 'AssetAmount') -> 'AssetAmount': def __sub__(self, other: 'AssetAmount') -> 'AssetAmount':
# TODO: Implement using add?
if not isinstance(other, AssetAmount): if not isinstance(other, AssetAmount):
msg = f'other must be AssetAmount, but was {type(other).__name__}' msg = f'other must be AssetAmount, but was {type(other)}'
raise TypeError(msg) raise TypeError(msg)
if self.is_zero(): if self.is_zero():
return -other return -other
@ -582,7 +576,7 @@ class AssetAmount:
def cmp(self, other) -> Decimal: def cmp(self, other) -> Decimal:
if not isinstance(other, AssetAmount): if not isinstance(other, AssetAmount):
msg = f'other must be AssetAmount, but was {type(other).__name__}' msg = f'other must be AssetAmount, but was {type(other)}'
raise TypeError(msg) raise TypeError(msg)
if self.is_zero() or other.is_zero(): if self.is_zero() or other.is_zero():
return self.amount - other.amount return self.amount - other.amount

View File

@ -1,122 +1,8 @@
from decimal import Decimal from decimal import Decimal
import pytest
import fin_defs import fin_defs
USD_0 = fin_defs.AssetAmount(fin_defs.USD, Decimal(0))
USD_10 = fin_defs.AssetAmount(fin_defs.USD, Decimal(10))
USD_11 = fin_defs.AssetAmount(fin_defs.USD, Decimal(11))
USD_20 = fin_defs.AssetAmount(fin_defs.USD, Decimal(20))
USD_21 = fin_defs.AssetAmount(fin_defs.USD, Decimal(21))
DKK_0 = fin_defs.AssetAmount(fin_defs.DKK, Decimal(0))
DKK_21 = fin_defs.AssetAmount(fin_defs.DKK, Decimal(21))
def test_add_asset():
assert USD_10 + USD_11 == USD_21
assert USD_10 + USD_0 == USD_10
assert DKK_0 + USD_10 == USD_10
def test_sub_asset():
assert USD_21 - USD_10 == USD_11
assert USD_10 - USD_0 == USD_10
assert DKK_0 - USD_10 == -USD_10
def test_div_asset():
assert USD_20 / USD_10 == 2
assert USD_10 / USD_10 == 1
assert USD_10 / USD_20 == 0.5
def test_div_amount():
assert USD_20 / Decimal(2) == USD_10
def test_mult():
assert USD_10 * Decimal(2) == USD_20
def test_negate():
assert USD_20 + (-USD_10) == USD_10
def test_compare():
assert USD_10 < USD_20
assert USD_10 <= USD_20
assert not (USD_10 > USD_20)
assert not (USD_10 >= USD_20)
def test_compare_zero():
assert DKK_0 < USD_20
assert DKK_0 <= USD_20
assert not (DKK_0 > USD_20)
assert not (DKK_0 >= USD_20)
def test_is_zero():
assert USD_0.is_zero()
assert not USD_10.is_zero()
def test_str(): def test_str():
assert str(USD_0) == '$0.000 USD' amount = fin_defs.AssetAmount(fin_defs.USD, Decimal(10))
assert str(USD_10) == '$10.00 USD' assert str(amount) == '$10.00 USD'
assert str(USD_11) == '$11.00 USD'
assert str(-USD_10) == '-$10.00 USD'
def test_mul_wrong():
with pytest.raises(TypeError, match='other must be Decimal, but was AssetAmount'):
assert USD_20 * USD_10
def test_add_wrong_type():
with pytest.raises(TypeError, match='other must be AssetAmount, but was Decimal'):
assert USD_20 + Decimal(1)
def test_add_wrong_asset():
with pytest.raises(
ValueError,
match='AssetAmount must have same asset, but: fiat:USD != fiat:DKK',
):
assert USD_20 + DKK_21
def test_sub_wrong_type():
with pytest.raises(TypeError, match='other must be AssetAmount, but was Decimal'):
assert USD_20 - Decimal(1)
def test_sub_wrong_asset():
with pytest.raises(
ValueError,
match='AssetAmount must have same asset, but: fiat:USD != fiat:DKK',
):
assert USD_20 - DKK_21
def test_cmp_wrong_type():
with pytest.raises(TypeError, match='other must be AssetAmount, but was Decimal'):
assert Decimal(1) > USD_20
def test_cmp_wrong_asset():
with pytest.raises(
ValueError,
match='AssetAmount must have same asset, but: fiat:USD != fiat:DKK',
):
assert USD_20 < DKK_21
def test_div_wrong_asset():
with pytest.raises(
ValueError,
match='AssetAmount must have same asset, but: fiat:USD != fiat:DKK',
):
assert USD_21 / DKK_21

View File

@ -1,10 +0,0 @@
import fin_defs
def test_from_currency_symbol():
assert fin_defs.FiatCurrency.from_currency_symbol('$') == fin_defs.USD
assert fin_defs.FiatCurrency.from_currency_symbol('TEST') is None
def test_currency_symbol():
assert fin_defs.USD.to_currency_symbol() == '$'

View File

@ -2,28 +2,24 @@ import pytest
import fin_defs import fin_defs
VALID_STOCK_TICKERS = ['TEST123', '123', 'TEST.EUR'] VALID_TICKERS = ['TEST123', '123', 'TEST.EUR']
BAD_STOCK_TICKERS = ['TEST:EUR', 'EUR:TEST'] BAD_TICKERS = ['TEST:EUR', 'EUR:TEST']
@pytest.mark.parametrize('ticker', VALID_STOCK_TICKERS) @pytest.mark.parametrize('ticker', VALID_TICKERS)
def test_valid_tickers(ticker: str): def test_valid_tickers(ticker: str):
asset = fin_defs.Stock(ticker, exchange=fin_defs.EXCHANGES_BY_IDS['NYSE']) fin_defs.Stock(ticker, exchange=fin_defs.EXCHANGES_BY_IDS['NYSE'])
assert asset.names() is not None
assert asset.to_polygon_id() is not None
@pytest.mark.parametrize('ticker', BAD_STOCK_TICKERS) @pytest.mark.parametrize('ticker', BAD_TICKERS)
def test_bad_tickers(ticker: str): def test_bad_tickers(ticker: str):
with pytest.raises(ValueError, match='ticker was not in correct format:.*'): with pytest.raises(ValueError, match='ticker was not in correct format:.*'):
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_STOCK_TICKERS) @pytest.mark.parametrize('ticker', BAD_TICKERS)
def test_crypto_tickers(ticker): def test_crypto_tickers(ticker):
asset = fin_defs.CryptoCurrency(ticker, 'not-known') fin_defs.CryptoCurrency(ticker, 'not-known')
assert asset.names() is not None
assert asset.to_polygon_id().startswith('X:')
def test_str(): def test_str():

View File

@ -1,37 +0,0 @@
import datetime
from decimal import Decimal
import fin_defs
NOW = datetime.datetime.now(tz=datetime.UTC)
THEN = NOW - datetime.timedelta(days=1)
SAMPLE_AVERAGE = fin_defs.ExchangeRateSample(
NOW,
(fin_defs.FiatCurrency.DKK, fin_defs.FiatCurrency.USD),
_average=Decimal(1),
)
SAMPLE_HIGH_LOW = fin_defs.ExchangeRateSample(
NOW,
(fin_defs.FiatCurrency.DKK, fin_defs.FiatCurrency.USD),
high=Decimal(1),
low=Decimal('0.5'),
)
def test_sample_average():
assert SAMPLE_AVERAGE.average == 1
assert SAMPLE_HIGH_LOW.average == Decimal('0.75')
def test_invert_sample():
inverted = SAMPLE_HIGH_LOW.invert_exchange_rate()
assert inverted.asset_pair == (fin_defs.FiatCurrency.USD, fin_defs.FiatCurrency.DKK)
assert inverted.high == 2
assert inverted.low == 1
assert inverted.average == 1.5
def test_replace_timestamp():
assert SAMPLE_HIGH_LOW.replace_timestamp(THEN).timestamp == THEN

View File

@ -2,38 +2,17 @@ import pytest
import fin_defs import fin_defs
NVO_ID = 'stock:NVO.NYSE{nordnet_id=16256554}'
NVO = fin_defs.Asset.from_string_id(NVO_ID)
VALID_IDS = [ VALID_IDS = [
'stock:NVO.NYSE', 'stock:NVO.NYSE',
NVO_ID, 'stock:NVO.NYSE{nordnet_id=16256554}',
'currency:USD', 'currency:USD',
'fiat:USD', 'fiat:USD',
'index:NDX', 'index:NDX',
'crypto:BTC', 'crypto:BTC',
'crypto:BTC{coingecko_id=bitcoin}', 'crypto:BTC{coingecko_id=bitcoin}',
'crypto:MPC',
'crypto:LOLCOIN',
'commodity:ALUMINUM', 'commodity:ALUMINUM',
'unknown:XXX',
] ]
ASSETS = list(fin_defs.WELL_KNOWN_SYMBOLS.values()) + [
fin_defs.Asset.from_string_id(vid) for vid in VALID_IDS
]
ASSETS_POLYGON_PRESERVES_FULL_ID = frozenset(
a
for a in ASSETS
if not isinstance(
a,
fin_defs.CryptoCurrency | fin_defs.Commodity | fin_defs.UnknownAsset,
)
) - frozenset([NVO])
INVALID_IDS = [ INVALID_IDS = [
'NVO', 'NVO',
'NVO.NYSE', 'NVO.NYSE',
@ -41,11 +20,6 @@ INVALID_IDS = [
'fiat:TEST:TEST', 'fiat:TEST:TEST',
'index:TEST:TEST', 'index:TEST:TEST',
'commodity:TEST:TEST', 'commodity:TEST:TEST',
'crypto: ',
'!!!!',
'::::',
'',
' ',
] ]
@ -62,15 +36,15 @@ def test_parse_attr():
def test_from_nordnet(): def test_from_nordnet():
asset = NVO derp = fin_defs.Asset.from_string_id('stock:NVO.NYSE{nordnet_id=16256554}')
assert isinstance(asset, fin_defs.Stock) assert isinstance(derp, fin_defs.Stock)
assert asset.ticker == 'NVO' assert derp.ticker == 'NVO'
assert asset.nordnet_id == 16256554 assert derp.nordnet_id == 16256554
@pytest.mark.parametrize('asset', ASSETS) @pytest.mark.parametrize('asset_id', VALID_IDS)
def test_raw_short_name(asset: fin_defs.Asset): def test_from_string_id_shortcut(asset_id: str):
assert asset.raw_short_name() is not None assert fin_defs.Asset.from_string_id(asset_id)
@pytest.mark.parametrize('asset_id', VALID_IDS) @pytest.mark.parametrize('asset_id', VALID_IDS)
@ -84,28 +58,18 @@ def test_from_string_id_invalid(asset_id: str):
fin_defs.Asset.from_string_id(asset_id) fin_defs.Asset.from_string_id(asset_id)
@pytest.mark.parametrize('asset', ASSETS) @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()) == asset
@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()) == asset
@pytest.mark.parametrize('asset', ASSETS) @pytest.mark.parametrize('asset', fin_defs.WELL_KNOWN_SYMBOLS.values())
def test_to_from_polygon_id_works(asset: fin_defs.Asset): def test_to_from_polygon_id(asset: fin_defs.Asset):
if isinstance(asset, fin_defs.Commodity | fin_defs.UnknownAsset): if isinstance(asset, fin_defs.CryptoCurrency):
return return
assert fin_defs.Asset.from_polygon_id(asset.to_polygon_id()) is not None
@pytest.mark.parametrize('asset', ASSETS_POLYGON_PRESERVES_FULL_ID)
def test_to_from_polygon_id_preserves_id(asset: fin_defs.Asset):
assert fin_defs.Asset.from_polygon_id(asset.to_polygon_id()) == asset assert fin_defs.Asset.from_polygon_id(asset.to_polygon_id()) == asset
def test_to_polygon_id_fail_for_commodity():
with pytest.raises(TypeError, match='Unknown type: Commodity'):
assert fin_defs.Commodity.CORN.to_polygon_id()
def test_to_string_id_wrong_type():
with pytest.raises(TypeError, match='Unsupported asset type: int'):
assert fin_defs.Asset.to_string_id(1)