1
0

Compare commits

...

2 Commits

Author SHA1 Message Date
6a13f5129f
Added support for unknown assets and special casing of AssetAmounts.
All checks were successful
Test Python / Test (push) Successful in 23s
2024-10-20 18:00:57 +02:00
c36911d706
Integrating features from an old project 2024-10-12 19:42:34 +02:00

View File

@ -143,6 +143,8 @@ class Asset:
return f'index:{self.ticker}' return f'index:{self.ticker}'
if isinstance(self, Commodity): if isinstance(self, Commodity):
return f'commodity:{self.alpha_vantage_id}' return f'commodity:{self.alpha_vantage_id}'
if isinstance(self, UnknownAsset):
return 'unknown:XXX'
msg = f'Unsupported asset type: {repr(self)}' msg = f'Unsupported asset type: {repr(self)}'
raise ValueError(msg) raise ValueError(msg)
@ -179,6 +181,8 @@ class Asset:
return CryptoCurrency(ticker, **attrs) return CryptoCurrency(ticker, **attrs)
if prefix == 'commodity': if prefix == 'commodity':
return Commodity(ticker) return Commodity(ticker)
if prefix == 'unknown':
return UnknownAsset()
msg = f'Unsupported asset format: {asset_id}' msg = f'Unsupported asset format: {asset_id}'
raise ValueError(msg) raise ValueError(msg)
@ -187,6 +191,12 @@ class Asset:
return self.to_string_id() return self.to_string_id()
@enforce_typing.enforce_types
@dataclasses.dataclass(frozen=True)
class UnknownAsset(Asset):
pass
@enforce_typing.enforce_types @enforce_typing.enforce_types
@dataclasses.dataclass(frozen=True) @dataclasses.dataclass(frozen=True)
class Currency(Asset): class Currency(Asset):
@ -198,7 +208,7 @@ class Currency(Asset):
class FiatCurrency(Currency): class FiatCurrency(Currency):
iso_code: str iso_code: str
def __post_init__(self): def __post_init__(self) -> None:
if not re.match(RE_TICKER_FORMAT, self.iso_code): if not re.match(RE_TICKER_FORMAT, self.iso_code):
msg = f'iso_code was not in correct format: {self.iso_code}' msg = f'iso_code was not in correct format: {self.iso_code}'
raise ValueError(msg) raise ValueError(msg)
@ -206,6 +216,22 @@ class FiatCurrency(Currency):
def raw_short_name(self) -> str: def raw_short_name(self) -> str:
return self.iso_code return self.iso_code
@staticmethod
def from_currency_symbol(symbol: str) -> 'Currency | None':
for (currency, currency_symbol) in CURRENCY_SYMBOLS.items():
if currency_symbol == symbol:
return currency
return None
def to_currency_symbol(self) -> str:
return CURRENCY_SYMBOLS[self]
FiatCurrency.DKK = FiatCurrency('DKK')
FiatCurrency.NOK = FiatCurrency('NOK')
FiatCurrency.USD = FiatCurrency('USD')
FiatCurrency.EUR = FiatCurrency('EUR')
FiatCurrency.GBP = FiatCurrency('GBP')
FiatCurrency.JPY = FiatCurrency('JPY')
@enforce_typing.enforce_types @enforce_typing.enforce_types
@dataclasses.dataclass(frozen=True, eq=True) @dataclasses.dataclass(frozen=True, eq=True)
@ -213,7 +239,7 @@ class CryptoCurrency(Currency):
ccxt_symbol: str ccxt_symbol: str
coingecko_id: str | None = None coingecko_id: str | None = None
def __post_init__(self): def __post_init__(self) -> None:
"""TODO """TODO
if not re.match(RE_CRYPTO_TICKER_FORMAT, self.ccxt_symbol): if not re.match(RE_CRYPTO_TICKER_FORMAT, 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}'
@ -231,7 +257,7 @@ class Stock(Asset):
exchange: StockExchange exchange: StockExchange
nordnet_id: int | None = None nordnet_id: int | None = None
def __post_init__(self): def __post_init__(self) -> None:
if not re.match(RE_TICKER_FORMAT_FLEXIBLE, self.ticker): if not re.match(RE_TICKER_FORMAT_FLEXIBLE, self.ticker):
msg = f'ticker was not in correct format: {self.ticker}' msg = f'ticker was not in correct format: {self.ticker}'
raise ValueError(msg) raise ValueError(msg)
@ -249,7 +275,7 @@ class Stock(Asset):
class Index(Asset): class Index(Asset):
ticker: str ticker: str
def __post_init__(self): def __post_init__(self) -> None:
if not re.match(RE_TICKER_FORMAT_FLEXIBLE, self.ticker): if not re.match(RE_TICKER_FORMAT_FLEXIBLE, self.ticker):
msg = f'ticker was not in correct format: {self.ticker}' msg = f'ticker was not in correct format: {self.ticker}'
raise ValueError(msg) raise ValueError(msg)
@ -263,7 +289,7 @@ class Index(Asset):
class Commodity(Asset): class Commodity(Asset):
alpha_vantage_id: str alpha_vantage_id: str
def __post_init__(self): def __post_init__(self) -> None:
if not re.match(RE_TICKER_FORMAT, self.alpha_vantage_id): if not re.match(RE_TICKER_FORMAT, self.alpha_vantage_id):
msg = f'alpha_vantage_id was not in correct format: {self.alpha_vantage_id}' msg = f'alpha_vantage_id was not in correct format: {self.alpha_vantage_id}'
raise ValueError(msg) raise ValueError(msg)
@ -284,11 +310,10 @@ Commodity.COFFEE = Commodity('COFFEE')
# Well known # Well known
DKK = FiatCurrency('DKK') DKK = FiatCurrency.DKK
NOK = FiatCurrency('NOK') USD = FiatCurrency.USD
USD = FiatCurrency('USD') EUR = FiatCurrency.EUR
EUR = FiatCurrency('EUR')
GBP = FiatCurrency('GBP')
BTC = CryptoCurrency('BTC', coingecko_id='bitcoin') BTC = CryptoCurrency('BTC', coingecko_id='bitcoin')
MPC = CryptoCurrency('MPC', coingecko_id='partisia-blockchain') MPC = CryptoCurrency('MPC', coingecko_id='partisia-blockchain')
SPX = Index('SPX') SPX = Index('SPX')
@ -324,8 +349,7 @@ COMMON_NAMES: dict[Asset, list[str]] = {
USDT: ['Tether USDT'], USDT: ['Tether USDT'],
} }
WELL_KNOWN_SYMBOLS = ( CURRENCY_CODES = {
{
sym: FiatCurrency(sym) sym: FiatCurrency(sym)
for sym in [ for sym in [
'USD', 'USD',
@ -342,18 +366,25 @@ WELL_KNOWN_SYMBOLS = (
'CZK', 'CZK',
] ]
} }
WELL_KNOWN_SYMBOLS = (
CURRENCY_CODES
| {'XXBT': BTC} | {'XXBT': BTC}
| {k.raw_short_name(): k for k in COMMON_NAMES} | {k.raw_short_name(): k for k in COMMON_NAMES}
| {'SPX500': SPX, 'SP500': SPX, 'Nasdaq 100': NDX} | {'SPX500': SPX, 'SP500': SPX, 'Nasdaq 100': NDX}
) )
ASSET_PREFIX: dict[Asset, str] = { CURRENCY_SYMBOLS: dict[Currency, str] = {
USD: '$', USD: '$',
EUR: '', EUR: '',
GBP: '£', FiatCurrency.GBP: '£',
BTC: '', BTC: '',
FiatCurrency.JPY: '¥',
} }
ASSET_PREFIX: dict[Asset, str] = CURRENCY_SYMBOLS # TODO: Remove at some point.
NYSE = StockExchange( NYSE = StockExchange(
name='New York Stock Exchange', name='New York Stock Exchange',
mic='XNYS', mic='XNYS',
@ -467,26 +498,30 @@ class AssetAmount:
asset: Asset asset: Asset
amount: Decimal amount: Decimal
def __str__(self): def __str__(self) -> str:
return self.human_readable_str() return self.human_readable_str()
def human_readable_str(self): def human_readable_str(self) -> str:
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) -> 'AssetAmount':
if not isinstance(other, Decimal): if not isinstance(other, Decimal):
msg = f'other must be decimal, but was {type(other)}' 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): 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)}' msg = f'other must be AssetAmount, but was {type(other)}'
raise TypeError(msg) raise TypeError(msg)
if self.is_zero():
return other
if other.is_zero():
return self
if self.asset != other.asset: if self.asset != other.asset:
msg = ( msg = (
f'AssetAmount must have same asset, but: {self.asset} != {other.asset}' f'AssetAmount must have same asset, but: {self.asset} != {other.asset}'
@ -495,10 +530,14 @@ class AssetAmount:
return AssetAmount(self.asset, self.amount + other.amount) return AssetAmount(self.asset, self.amount + other.amount)
def __sub__(self, other): def __sub__(self, other: 'AssetAmount') -> 'AssetAmount':
if not isinstance(other, AssetAmount): if not isinstance(other, AssetAmount):
msg = f'other must be AssetAmount, but was {type(other)}' msg = f'other must be AssetAmount, but was {type(other)}'
raise TypeError(msg) raise TypeError(msg)
if self.is_zero():
return -other
if other.is_zero():
return self
if self.asset != other.asset: if self.asset != other.asset:
msg = ( msg = (
f'AssetAmount must have same asset, but: {self.asset} != {other.asset}' f'AssetAmount must have same asset, but: {self.asset} != {other.asset}'
@ -506,7 +545,7 @@ class AssetAmount:
raise ValueError(msg) raise ValueError(msg)
return AssetAmount(self.asset, self.amount - other.amount) return AssetAmount(self.asset, self.amount - other.amount)
def __truediv__(self, other): def __truediv__(self, other: 'Decimal | AssetAmount') -> 'Decimal | AssetAmount':
if isinstance(other, AssetAmount): if isinstance(other, AssetAmount):
if self.asset != other.asset: if self.asset != other.asset:
msg = f'AssetAmount must have same asset, but: {self.asset} != {other.asset}' msg = f'AssetAmount must have same asset, but: {self.asset} != {other.asset}'
@ -514,16 +553,38 @@ class AssetAmount:
return self.amount / other.amount return self.amount / other.amount
return AssetAmount(self.asset, self.amount / other) return AssetAmount(self.asset, self.amount / other)
def __lt__(self, other): def __neg__(self) -> 'AssetAmount':
return AssetAmount(self.asset, -self.amount)
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)}' msg = f'other must be AssetAmount, but was {type(other)}'
raise TypeError(msg) raise TypeError(msg)
if self.is_zero() or other.is_zero():
return self.amount - other.amount
if self.asset != other.asset: if self.asset != other.asset:
msg = ( msg = (
f'AssetAmount must have same asset, but: {self.asset} != {other.asset}' f'AssetAmount must have same asset, but: {self.asset} != {other.asset}'
) )
raise ValueError(msg) raise ValueError(msg)
return self.amount < other.amount return self.amount - other.amount
def is_zero(self) -> bool:
return self.amount == 0
def __lt__(self, other: 'AssetAmount') -> bool:
return self.cmp(other) < 0
def __le__(self, other: 'AssetAmount') -> bool:
return self.cmp(other) <= 0
def __gt__(self, other: 'AssetAmount') -> bool:
return self.cmp(other) > 0
def __ge__(self, other: 'AssetAmount') -> bool:
return self.cmp(other) >= 0
AssetAmount.ZERO = AssetAmount(UnknownAsset(), Decimal(0))
AssetPair = tuple[Asset, Asset] AssetPair = tuple[Asset, Asset]
@ -550,10 +611,10 @@ class ExchangeRateSample:
return self._average return self._average
return (self.high + self.low) / 2 return (self.high + self.low) / 2
def replace_timestamp(self, timestamp): def replace_timestamp(self, timestamp: datetime.date | datetime.datetime) -> 'ExchangeRateSample':
return dataclasses.replace(self, timestamp=timestamp) return dataclasses.replace(self, timestamp=timestamp)
def invert_exchange_rate(self): def invert_exchange_rate(self) -> 'ExchangeRateSample':
one = Decimal('1') one = Decimal('1')
return dataclasses.replace( return dataclasses.replace(
self, self,