Compare commits
No commits in common. "6a13f5129fb7411ad72ca202d1f7527b76af43e1" and "84ab1e4218afeeabef647cf203b35696c8634545" have entirely different histories.
6a13f5129f
...
84ab1e4218
|
@ -143,8 +143,6 @@ 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)
|
||||||
|
@ -181,8 +179,6 @@ 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)
|
||||||
|
@ -191,12 +187,6 @@ 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):
|
||||||
|
@ -208,7 +198,7 @@ class Currency(Asset):
|
||||||
class FiatCurrency(Currency):
|
class FiatCurrency(Currency):
|
||||||
iso_code: str
|
iso_code: str
|
||||||
|
|
||||||
def __post_init__(self) -> None:
|
def __post_init__(self):
|
||||||
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)
|
||||||
|
@ -216,22 +206,6 @@ 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)
|
||||||
|
@ -239,7 +213,7 @@ class CryptoCurrency(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):
|
||||||
"""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}'
|
||||||
|
@ -257,7 +231,7 @@ class Stock(Asset):
|
||||||
exchange: StockExchange
|
exchange: StockExchange
|
||||||
nordnet_id: int | None = None
|
nordnet_id: int | None = None
|
||||||
|
|
||||||
def __post_init__(self) -> None:
|
def __post_init__(self):
|
||||||
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)
|
||||||
|
@ -275,7 +249,7 @@ class Stock(Asset):
|
||||||
class Index(Asset):
|
class Index(Asset):
|
||||||
ticker: str
|
ticker: str
|
||||||
|
|
||||||
def __post_init__(self) -> None:
|
def __post_init__(self):
|
||||||
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)
|
||||||
|
@ -289,7 +263,7 @@ class Index(Asset):
|
||||||
class Commodity(Asset):
|
class Commodity(Asset):
|
||||||
alpha_vantage_id: str
|
alpha_vantage_id: str
|
||||||
|
|
||||||
def __post_init__(self) -> None:
|
def __post_init__(self):
|
||||||
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)
|
||||||
|
@ -310,10 +284,11 @@ Commodity.COFFEE = Commodity('COFFEE')
|
||||||
|
|
||||||
# Well known
|
# Well known
|
||||||
|
|
||||||
DKK = FiatCurrency.DKK
|
DKK = FiatCurrency('DKK')
|
||||||
USD = FiatCurrency.USD
|
NOK = FiatCurrency('NOK')
|
||||||
EUR = FiatCurrency.EUR
|
USD = FiatCurrency('USD')
|
||||||
|
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')
|
||||||
|
@ -349,42 +324,36 @@ COMMON_NAMES: dict[Asset, list[str]] = {
|
||||||
USDT: ['Tether USDT'],
|
USDT: ['Tether USDT'],
|
||||||
}
|
}
|
||||||
|
|
||||||
CURRENCY_CODES = {
|
|
||||||
sym: FiatCurrency(sym)
|
|
||||||
for sym in [
|
|
||||||
'USD',
|
|
||||||
'DKK',
|
|
||||||
'EUR',
|
|
||||||
'JPY',
|
|
||||||
'CAD',
|
|
||||||
'GBP',
|
|
||||||
'CNY',
|
|
||||||
'PLN',
|
|
||||||
'ILS',
|
|
||||||
'CHF',
|
|
||||||
'ZAR',
|
|
||||||
'CZK',
|
|
||||||
]
|
|
||||||
}
|
|
||||||
|
|
||||||
|
|
||||||
WELL_KNOWN_SYMBOLS = (
|
WELL_KNOWN_SYMBOLS = (
|
||||||
CURRENCY_CODES
|
{
|
||||||
|
sym: FiatCurrency(sym)
|
||||||
|
for sym in [
|
||||||
|
'USD',
|
||||||
|
'DKK',
|
||||||
|
'EUR',
|
||||||
|
'JPY',
|
||||||
|
'CAD',
|
||||||
|
'GBP',
|
||||||
|
'CNY',
|
||||||
|
'PLN',
|
||||||
|
'ILS',
|
||||||
|
'CHF',
|
||||||
|
'ZAR',
|
||||||
|
'CZK',
|
||||||
|
]
|
||||||
|
}
|
||||||
| {'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}
|
||||||
)
|
)
|
||||||
|
|
||||||
CURRENCY_SYMBOLS: dict[Currency, str] = {
|
ASSET_PREFIX: dict[Asset, str] = {
|
||||||
USD: '$',
|
USD: '$',
|
||||||
EUR: '€',
|
EUR: '€',
|
||||||
FiatCurrency.GBP: '£',
|
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',
|
||||||
|
@ -498,30 +467,26 @@ class AssetAmount:
|
||||||
asset: Asset
|
asset: Asset
|
||||||
amount: Decimal
|
amount: Decimal
|
||||||
|
|
||||||
def __str__(self) -> str:
|
def __str__(self):
|
||||||
return self.human_readable_str()
|
return self.human_readable_str()
|
||||||
|
|
||||||
def human_readable_str(self) -> str:
|
def human_readable_str(self):
|
||||||
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) -> 'AssetAmount':
|
def __mul__(self, other: Decimal):
|
||||||
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: 'AssetAmount') -> 'AssetAmount':
|
def __add__(self, other):
|
||||||
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}'
|
||||||
|
@ -530,14 +495,10 @@ 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):
|
||||||
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}'
|
||||||
|
@ -545,7 +506,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: 'Decimal | AssetAmount') -> 'Decimal | AssetAmount':
|
def __truediv__(self, other):
|
||||||
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}'
|
||||||
|
@ -553,38 +514,16 @@ 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 __neg__(self) -> 'AssetAmount':
|
def __lt__(self, other):
|
||||||
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]
|
||||||
|
@ -611,10 +550,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: datetime.date | datetime.datetime) -> 'ExchangeRateSample':
|
def replace_timestamp(self, timestamp):
|
||||||
return dataclasses.replace(self, timestamp=timestamp)
|
return dataclasses.replace(self, timestamp=timestamp)
|
||||||
|
|
||||||
def invert_exchange_rate(self) -> 'ExchangeRateSample':
|
def invert_exchange_rate(self):
|
||||||
one = Decimal('1')
|
one = Decimal('1')
|
||||||
return dataclasses.replace(
|
return dataclasses.replace(
|
||||||
self,
|
self,
|
||||||
|
|
Loading…
Reference in New Issue
Block a user