430 lines
14 KiB
Python
430 lines
14 KiB
Python
"""# Finance Definitions.
|
|
|
|
Python library defining base types for financial processing.
|
|
"""
|
|
|
|
import dataclasses
|
|
import datetime
|
|
import re
|
|
from decimal import Decimal
|
|
|
|
import enforce_typing
|
|
|
|
from ._version import __version__ # noqa: F401
|
|
|
|
## Ids
|
|
|
|
RE_TICKER_FORMAT = r'^[A-Z0-9_]+$'
|
|
RE_TICKER_FORMAT_FLEXIBLE = r'^[^:\s](?:[^:]*[^:\s])?$'
|
|
RE_CRYPTO_TICKER_FORMAT = r'^\S+$'
|
|
|
|
|
|
@enforce_typing.enforce_types
|
|
@dataclasses.dataclass(frozen=True)
|
|
class StockExchange:
|
|
"""Unified Stock Exchange identifiers."""
|
|
|
|
name: str
|
|
mic: str
|
|
bloomberg_id: str | None = None
|
|
crunchbase_id: str | None = None
|
|
yahoo_id: str | None = None
|
|
eod_id: str | None = None
|
|
|
|
|
|
@enforce_typing.enforce_types
|
|
@dataclasses.dataclass(frozen=True)
|
|
class Asset:
|
|
"""An identifier representing some abstract financial asset.
|
|
|
|
Subtypes represent the various asset types in existence, for example fiat
|
|
currencies, stocks or even crypto currencies.
|
|
"""
|
|
|
|
def names(self) -> list[str]:
|
|
return COMMON_NAMES.get(self, [])
|
|
|
|
def to_polygon_id(self) -> str:
|
|
# TODO: Support currency pairs not involving USD.
|
|
if isinstance(self, Stock):
|
|
return f'{self.ticker}'
|
|
if isinstance(self, Index):
|
|
return f'I:{self.ticker}'
|
|
if isinstance(self, FiatCurrency):
|
|
return f'C:{self.iso_code}USD'
|
|
if isinstance(self, CryptoCurrency):
|
|
return f'X:{self.ccxt_symbol}USD'
|
|
|
|
# Else
|
|
msg = f'Unknown self type: {self}'
|
|
raise Exception(msg)
|
|
|
|
@staticmethod
|
|
def from_polygon_id(polygon_id: str) -> 'Asset':
|
|
if polygon_id.startswith('I:'):
|
|
return Index(polygon_id.removeprefix('I:'))
|
|
if polygon_id.startswith('X:'):
|
|
return CryptoCurrency(
|
|
polygon_id.removeprefix('X:').removesuffix('USD'),
|
|
None,
|
|
)
|
|
if polygon_id.startswith('C:'):
|
|
return FiatCurrency(polygon_id.removeprefix('C:').removesuffix('USD'))
|
|
return Stock.from_polygon_id(polygon_id)
|
|
|
|
|
|
@enforce_typing.enforce_types
|
|
@dataclasses.dataclass(frozen=True)
|
|
class Currency(Asset):
|
|
pass
|
|
|
|
|
|
@enforce_typing.enforce_types
|
|
@dataclasses.dataclass(frozen=True, eq=True, order=True)
|
|
class FiatCurrency(Currency):
|
|
iso_code: str
|
|
|
|
def __post_init__(self):
|
|
if not re.match(RE_TICKER_FORMAT, self.iso_code):
|
|
msg = f'iso_code was not in correct format: {self.iso_code}'
|
|
raise ValueError(msg)
|
|
|
|
def __str__(self):
|
|
return self.iso_code
|
|
|
|
|
|
@enforce_typing.enforce_types
|
|
@dataclasses.dataclass(frozen=True, eq=True)
|
|
class CryptoCurrency(Currency):
|
|
ccxt_symbol: str
|
|
coingecko_id: str | None
|
|
|
|
def __post_init__(self):
|
|
"""TODO
|
|
if not re.match(RE_CRYPTO_TICKER_FORMAT, self.ccxt_symbol):
|
|
msg = f'ccxt_symbol was not in correct format: {self.ccxt_symbol}'
|
|
raise ValueError(msg)
|
|
"""
|
|
|
|
def __str__(self):
|
|
return self.ccxt_symbol
|
|
|
|
|
|
@enforce_typing.enforce_types
|
|
@dataclasses.dataclass(frozen=True, eq=True)
|
|
class Stock(Asset):
|
|
ticker: str
|
|
exchange: StockExchange
|
|
|
|
def __post_init__(self):
|
|
if not re.match(RE_TICKER_FORMAT_FLEXIBLE, self.ticker):
|
|
msg = f'ticker was not in correct format: {self.ticker}'
|
|
raise ValueError(msg)
|
|
|
|
def __str__(self):
|
|
return f'{self.ticker}.{self.exchange.mic}'
|
|
|
|
@staticmethod
|
|
def from_polygon_id(polygon_id: str, stock_exchange: StockExchange) -> 'Stock':
|
|
return Stock(polygon_id, stock_exchange)
|
|
|
|
|
|
@enforce_typing.enforce_types
|
|
@dataclasses.dataclass(frozen=True, eq=True)
|
|
class Index(Asset):
|
|
ticker: str
|
|
|
|
def __post_init__(self):
|
|
if not re.match(RE_TICKER_FORMAT_FLEXIBLE, self.ticker):
|
|
msg = f'ticker was not in correct format: {self.ticker}'
|
|
raise ValueError(msg)
|
|
|
|
def __str__(self):
|
|
return self.ticker
|
|
|
|
|
|
@enforce_typing.enforce_types
|
|
@dataclasses.dataclass(frozen=True, eq=True)
|
|
class Commodity(Asset):
|
|
alpha_vantage_id: str
|
|
|
|
def __post_init__(self):
|
|
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}'
|
|
raise ValueError(msg)
|
|
|
|
def __str__(self):
|
|
return self.alpha_vantage_id
|
|
|
|
|
|
Commodity.CRUDE_OIL_WTI = Commodity('WTI')
|
|
Commodity.CRUDE_OIL_BRENT = Commodity('BRENT')
|
|
Commodity.NATURAL_GAS = Commodity('NATURAL_GAS')
|
|
Commodity.COPPER = Commodity('COPPER')
|
|
Commodity.WHEAT = Commodity('WHEAT')
|
|
Commodity.CORN = Commodity('CORN')
|
|
Commodity.COTTON = Commodity('COTTON')
|
|
Commodity.SUGAR = Commodity('SUGAR')
|
|
Commodity.COFFEE = Commodity('COFFEE')
|
|
|
|
# Well known
|
|
|
|
DKK = FiatCurrency('DKK')
|
|
USD = FiatCurrency('USD')
|
|
EUR = FiatCurrency('EUR')
|
|
BTC = CryptoCurrency('BTC', coingecko_id='bitcoin')
|
|
MPC = CryptoCurrency('MPC', coingecko_id='partisia-blockchain')
|
|
SPX = Index('SPX')
|
|
NDX = Index('NDX')
|
|
USDT = CryptoCurrency('USDT', 'tether')
|
|
|
|
COMMON_NAMES: dict[Asset, list[str]] = {
|
|
FiatCurrency('USD'): ['U.S. Dollar'],
|
|
FiatCurrency('DKK'): ['Danske Kroner', 'Danish Crowns'],
|
|
FiatCurrency('EUR'): ['Euro'],
|
|
FiatCurrency('JPY'): ['Japanese Yen'],
|
|
FiatCurrency('CAD'): ['Canadian Dollar'],
|
|
FiatCurrency('GBP'): ['British Pound'],
|
|
FiatCurrency('CNY'): ['Renminbi'],
|
|
FiatCurrency('PLN'): ['Polish Zloty'],
|
|
FiatCurrency('ILS'): ['Israeli new Shekel'],
|
|
FiatCurrency('CHF'): ['Swiss Franc'],
|
|
FiatCurrency('ZAR'): ['South African Rand'],
|
|
FiatCurrency('CZK'): ['Czech Koruna'],
|
|
SPX: ['Standard and Poor 500', 'S&P 500'],
|
|
NDX: ['Nasdaq 100'],
|
|
BTC: ['Bitcoin BTC'],
|
|
MPC: ['Partisia Blockchain MPC Token'],
|
|
CryptoCurrency('ETH', 'ethereum'): ['Ethereum Ether'],
|
|
CryptoCurrency('DOGE', 'dogecoin'): ['Dogecoin'],
|
|
CryptoCurrency('SOL', 'solana'): ['Solana SOL'],
|
|
CryptoCurrency('ADA', 'cardano'): ['Cardano ADA'],
|
|
CryptoCurrency('BNB', 'bnb'): ['Binance BNB'],
|
|
CryptoCurrency('MATIC', 'polygon'): ['Polygon MATIC'],
|
|
# Stable-coins
|
|
CryptoCurrency('DAI', 'dai'): ['DAI'],
|
|
CryptoCurrency('USDC', 'usdc'): ['USD Coin'],
|
|
USDT: ['Tether USDT'],
|
|
}
|
|
|
|
WELL_KNOWN_SYMBOLS = (
|
|
{
|
|
sym: FiatCurrency(sym)
|
|
for sym in [
|
|
'USD',
|
|
'DKK',
|
|
'EUR',
|
|
'JPY',
|
|
'CAD',
|
|
'GBP',
|
|
'CNY',
|
|
'PLN',
|
|
'ILS',
|
|
'CHF',
|
|
'ZAR',
|
|
'CZK',
|
|
]
|
|
}
|
|
| {'XXBT': BTC}
|
|
| {str(k): k for k in COMMON_NAMES}
|
|
| {'SPX500': SPX, 'SP500': SPX, 'Nasdaq 100': NDX}
|
|
)
|
|
|
|
NYSE = StockExchange(
|
|
name='New York Stock Exchange',
|
|
mic='XNYS',
|
|
bloomberg_id='UN',
|
|
crunchbase_id='NYSE',
|
|
)
|
|
|
|
EXCHANGES = [
|
|
NYSE,
|
|
StockExchange(name='Nasdaq', mic='XNAS', bloomberg_id='US', crunchbase_id='NASDAQ'),
|
|
StockExchange(
|
|
name='Hong Kong Exchanges And Clearing',
|
|
mic='XHKG',
|
|
bloomberg_id='HK',
|
|
crunchbase_id='SEHK',
|
|
),
|
|
StockExchange(
|
|
name='Moscow Exchange',
|
|
mic='MISC',
|
|
bloomberg_id='RX',
|
|
yahoo_id='ME',
|
|
eod_id='MCX',
|
|
crunchbase_id='MOEX',
|
|
),
|
|
StockExchange(name='Shanghai Stock Exchange', mic='XSHG'),
|
|
StockExchange(name='Euronext Amsterdam', mic='XAMS'),
|
|
StockExchange(name='Euronext Brussels', mic='XBRU'),
|
|
StockExchange(name='Euronext Dublin', mic='XMSM'),
|
|
StockExchange(name='Euronext Lisbon', mic='XLIS'),
|
|
StockExchange(name='Euronext Milan', mic='XMIL'),
|
|
StockExchange(name='Euronext Oslo', mic='XOSL'),
|
|
StockExchange(name='Euronext Paris', mic='XPAR'),
|
|
StockExchange(name='Japan Exchange Group', mic='XJPX'),
|
|
StockExchange(name='Shenzhen Stock Exchange', mic='XSHE'),
|
|
StockExchange(name='Indian National Stock Exchange', mic='XNSE'),
|
|
StockExchange(
|
|
name='Bombay Stock Exchange',
|
|
mic='XBOM',
|
|
bloomberg_id='IB',
|
|
crunchbase_id='BOM',
|
|
yahoo_id='BO',
|
|
),
|
|
StockExchange(name='Toronto Stock Exchange', mic='XTSE'),
|
|
StockExchange(name='London Stock Exchange', mic='XLON'),
|
|
StockExchange(name='Saudi Stock Exchange', mic='XSAU'),
|
|
StockExchange(name='German Stock Exchange', mic='XFRA'),
|
|
StockExchange(name='SIX Swiss Exchange', mic='XSWX'),
|
|
StockExchange(name='Nasdaq Copenhagen Stock Exchange', mic='XCSE'),
|
|
StockExchange(name='Nasdaq Stockholm Stock Exchange', mic='XSTO'),
|
|
StockExchange(name='Nasdaq Helsinki Stock Exchange', mic='XHEL'),
|
|
StockExchange(name='Nasdaq Tallinn Stock Exchange', mic='XTAL'),
|
|
StockExchange(name='Nasdaq Riga Stock Exchange', mic='XRIS'),
|
|
StockExchange(name='Nasdaq Vilnius Stock Exchange', mic='XLIT'),
|
|
StockExchange(name='Nasdaq Iceland Stock Exchange', mic='XICE'),
|
|
StockExchange(name='Korea Exchange', mic='XKOS'),
|
|
StockExchange(name='Taiwan Stock Exchange', mic='XTAI'),
|
|
StockExchange(name='Australian Securities Exchange', mic='XASX'),
|
|
StockExchange(name='Johannesburg Stock Exchange', mic='XJSE'),
|
|
StockExchange(name='Tehran Stock Exchange', mic='XTEH'),
|
|
StockExchange(name='Over The Counter (Unlisted)', mic='OTC'), # TODO: Unofficial
|
|
StockExchange(name='NYSE Arca', mic='ARCX'),
|
|
StockExchange(name='BATS Global Markets', mic='BATS'),
|
|
StockExchange(name='NYSE American', mic='XASE'),
|
|
StockExchange(name='Xetra (Frankfurt Stock Exchange)', mic='XETR'),
|
|
]
|
|
|
|
EXCHANGES_BY_IDS = {}
|
|
if True:
|
|
|
|
def add_by_id(exchange: StockExchange, key: str):
|
|
exchange_id = getattr(exchange, key)
|
|
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
|
|
|
|
for exchange in EXCHANGES:
|
|
add_by_id(exchange, 'mic')
|
|
add_by_id(exchange, 'bloomberg_id')
|
|
add_by_id(exchange, 'crunchbase_id')
|
|
add_by_id(exchange, 'yahoo_id')
|
|
add_by_id(exchange, 'eod_id')
|
|
del exchange
|
|
del add_by_id
|
|
|
|
|
|
@enforce_typing.enforce_types
|
|
@dataclasses.dataclass(frozen=True, slots=True)
|
|
class AssetInformation:
|
|
"""Extensive information about a given asset."""
|
|
|
|
asset: Asset
|
|
full_name: str
|
|
description: str | None = None
|
|
asset_base: Asset | None = None
|
|
sec_cik: str | None = None
|
|
# TODO: FIGI types? (https://www.openfigi.com/)
|
|
|
|
|
|
@enforce_typing.enforce_types
|
|
@dataclasses.dataclass(frozen=True, eq=True, slots=True)
|
|
class AssetAmount:
|
|
"""Decimal with associated asset unit."""
|
|
|
|
asset: Asset
|
|
amount: Decimal
|
|
|
|
def __str__(self):
|
|
specificity = '2' if self.amount >= 0.10 else '3'
|
|
return ('{:.' + specificity + 'f} {}').format(self.amount, self.asset)
|
|
|
|
def __mul__(self, other: Decimal):
|
|
if not isinstance(other, Decimal):
|
|
msg = f'other must be decimal, but was {type(other)}'
|
|
raise TypeError(msg)
|
|
return AssetAmount(self.asset, self.amount * other)
|
|
|
|
def __add__(self, other):
|
|
if not isinstance(other, AssetAmount):
|
|
msg = f'other must be AssetAmount, but was {type(other)}'
|
|
raise TypeError(msg)
|
|
if self.asset != other.asset:
|
|
msg = (
|
|
f'AssetAmount must have same asset, but: {self.asset} != {other.asset}'
|
|
)
|
|
raise ValueError(msg)
|
|
|
|
return AssetAmount(self.asset, self.amount + other.amount)
|
|
|
|
def __sub__(self, other):
|
|
if not isinstance(other, AssetAmount):
|
|
msg = f'other must be AssetAmount, but was {type(other)}'
|
|
raise TypeError(msg)
|
|
if self.asset != other.asset:
|
|
msg = (
|
|
f'AssetAmount must have same asset, but: {self.asset} != {other.asset}'
|
|
)
|
|
raise ValueError(msg)
|
|
return AssetAmount(self.asset, self.amount - other.amount)
|
|
|
|
def __truediv__(self, other):
|
|
if isinstance(other, AssetAmount):
|
|
if self.asset != other.asset:
|
|
msg = f'AssetAmount must have same asset, but: {self.asset} != {other.asset}'
|
|
raise ValueError(msg)
|
|
return self.amount / other.amount
|
|
return AssetAmount(self.asset, self.amount / other)
|
|
|
|
def __lt__(self, other):
|
|
if not isinstance(other, AssetAmount):
|
|
msg = f'other must be AssetAmount, but was {type(other)}'
|
|
raise TypeError(msg)
|
|
if self.asset != other.asset:
|
|
msg = (
|
|
f'AssetAmount must have same asset, but: {self.asset} != {other.asset}'
|
|
)
|
|
raise ValueError(msg)
|
|
return self.amount < other.amount
|
|
|
|
|
|
@dataclasses.dataclass(frozen=True, eq=True)
|
|
class ExchangeRateSample(Asset):
|
|
"""Single exchange rate sample with a timestamp and various stats."""
|
|
|
|
timestamp: datetime.date | datetime.datetime
|
|
_average: Decimal | None = None
|
|
last: Decimal | None = None
|
|
open: Decimal | None = None
|
|
close: Decimal | None = None
|
|
high: Decimal | None = None
|
|
low: Decimal | None = None
|
|
volume: Decimal | None = None
|
|
market_cap: Decimal | None = None
|
|
|
|
@property
|
|
def average(self) -> Decimal:
|
|
if self._average:
|
|
return self._average
|
|
return (self.high + self.low) / 2
|
|
|
|
def replace_timestamp(self, timestamp):
|
|
return dataclasses.replace(self, timestamp=timestamp)
|
|
|
|
def invert_exchange_rate(self):
|
|
one = Decimal('1')
|
|
return dataclasses.replace(
|
|
self,
|
|
_average=one / self._average if self._average else None,
|
|
last=one / self.last if self.last else None,
|
|
open=one / self.open if self.open else None,
|
|
close=one / self.close if self.close else None,
|
|
high=one / self.low if self.low else None,
|
|
low=one / self.high if self.high else None,
|
|
# TODO: Support volume
|
|
# TODO: Support market_cap
|
|
)
|