"""# Finance Definitions. Python library defining base types for financial processing. Defines a base `Asset` type, and various subtypes, for universal representation of these assets. Defined hierarchy: * `Asset` - `Currency` + `FiatCurrency` + `CryptoCurrency` - `Stock` - `Index` - `Commodity` """ import abc import dataclasses import datetime import re from collections.abc import Mapping from decimal import Decimal import enforce_typing from ._version import __version__ __all__ = ['__version__', 'StockExchange', 'AssetAmount', 'Asset', 'ExchangeRateSample'] def parse_id_attr_key_value_pair(attr_datum: str) -> tuple[str, str | int] | None: attr_datum = attr_datum.strip() if attr_datum == '': return None (attr_key, attr_value) = attr_datum.split('=') if re.match(r'^\d+$', attr_value): attr_value = int(attr_value) return attr_key, attr_value def parse_id_attr_data(attr_data: str) -> Mapping[str, str | int]: if attr_data is None: return {} key_value_pairs_str = attr_data.split(',') key_value_pairs = [parse_id_attr_key_value_pair(kvp) for kvp in key_value_pairs_str] d = {} for kvp in key_value_pairs: if kvp is None: continue d[kvp[0]] = kvp[1] return d ## 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. Fields: - `name`: Human-readable name. - `mic`: Official MIC identifier. - `bloomberg_id`: Identifier for lookup on Bloomberg. - `crunchbase_id`: Identifier for lookup on Crunchbase. - `yahoo_id`: Identifier for lookup on Yahoo! Finance. - `eod_id`: Identifier for lookup on EOD. """ name: str mic: str bloomberg_id: str | None = None crunchbase_id: str | None = None yahoo_id: str | None = None eod_id: str | None = None NYSE = StockExchange( name='New York Stock Exchange', mic='XNYS', bloomberg_id='UN', crunchbase_id='NYSE', ) @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]: """Returns a list of human readable names for this `Asset`.""" return COMMON_NAMES.get(self, []) def to_polygon_id(self) -> str: """Formats this asset to the [`polygon.io`](https://polygon.io) identifier format. Inverse of `from_polygon_id`. """ # 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 type: {type(self).__name__}' raise TypeError(msg) @staticmethod def from_polygon_id(polygon_id: str, exchange: StockExchange = NYSE) -> 'Asset': """Parses an asset from the [`polygon.io`](https://polygon.io) identifier format. Inverse of `to_polygon_id`. """ 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, exchange) @abc.abstractmethod def raw_short_name(self) -> str: """Short name or ticker. Does not nessisarily uniquely identify the asset, but may be commonly used within the industry. """ def to_string_id(self) -> str: """Formats the asset id using the fin_defs asset format.""" 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}{attrs_str}' if isinstance(self, FiatCurrency): return f'fiat:{self.iso_code}' if isinstance(self, CryptoCurrency): attrs_str = ( f'{{coingecko_id={self.coingecko_id}}}' if self.coingecko_id else '' ) return f'crypto:{self.ccxt_symbol}{attrs_str}' if isinstance(self, Index): return f'index:{self.ticker}' if isinstance(self, Commodity): return f'commodity:{self.alpha_vantage_id}' if isinstance(self, UnknownAsset): return 'unknown:XXX' msg = f'Unsupported asset type: {type(self).__name__}' raise TypeError(msg) @staticmethod def from_string_id(asset_id: str) -> 'Asset': """Parses an `Asset` using the fin_defs asset format.""" m = re.match(r'^(?:(\w+):)?([^{]+)(?:\{(.*)\})?$', asset_id) if m is None: msg = f'Unsupported asset format: {asset_id}' raise ValueError(msg) prefix = m.group(1) ticker = m.group(2) attr_data = m.group(3) if known_symbol := WELL_KNOWN_SYMBOLS.get(ticker, None): return known_symbol attrs: dict[str, str | int] = parse_id_attr_data(attr_data) if prefix == 'stock': if m := re.match(r'^(\w+)(?:\.(\w+))?$', ticker): exchange = EXCHANGES_BY_IDS[m.group(2) or 'NYSE'] # TODO? return Stock(m.group(1), exchange, **attrs) if prefix in {'currency', 'fiat'}: return FiatCurrency(ticker, **attrs) if prefix == 'index': return Index(ticker, **attrs) if prefix == 'crypto': return CryptoCurrency(ticker, **attrs) if prefix == 'commodity': return Commodity(ticker) if prefix == 'unknown': return UnknownAsset() msg = f'Unsupported asset format: {asset_id}' raise ValueError(msg) def __str__(self): return self.to_string_id() @enforce_typing.enforce_types @dataclasses.dataclass(frozen=True) class UnknownAsset(Asset): """An asset that does not exist.""" @abc.abstractmethod def raw_short_name(self) -> str: return '???' @enforce_typing.enforce_types @dataclasses.dataclass(frozen=True) class Currency(Asset): """Either a Fiat or a Crypto Currency.""" @enforce_typing.enforce_types @dataclasses.dataclass(frozen=True, eq=True, order=True) class FiatCurrency(Currency): """Fiat Currency.""" iso_code: str def __post_init__(self) -> None: 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 raw_short_name(self) -> str: 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 @dataclasses.dataclass(frozen=True, eq=True) class CryptoCurrency(Currency): """Crypto Currency.""" ccxt_symbol: str coingecko_id: str | None = None def raw_short_name(self) -> str: return self.ccxt_symbol @enforce_typing.enforce_types @dataclasses.dataclass(frozen=True, eq=True) class Stock(Asset): ticker: str exchange: StockExchange nordnet_id: int | None = None def __post_init__(self) -> None: if not re.match(RE_TICKER_FORMAT_FLEXIBLE, self.ticker): msg = f'ticker was not in correct format: {self.ticker}' raise ValueError(msg) @staticmethod def from_polygon_id(polygon_id: str, exchange: StockExchange = NYSE) -> 'Stock': return Stock(polygon_id, exchange) def raw_short_name(self) -> str: return self.ticker @enforce_typing.enforce_types @dataclasses.dataclass(frozen=True, eq=True) class Index(Asset): ticker: str def __post_init__(self) -> None: if not re.match(RE_TICKER_FORMAT_FLEXIBLE, self.ticker): msg = f'ticker was not in correct format: {self.ticker}' raise ValueError(msg) def raw_short_name(self) -> str: return self.ticker @enforce_typing.enforce_types @dataclasses.dataclass(frozen=True, eq=True) class Commodity(Asset): alpha_vantage_id: str def __post_init__(self) -> None: 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 raw_short_name(self) -> str: 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', coingecko_id='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', coingecko_id='ethereum'): ['Ethereum Ether'], CryptoCurrency('DOGE', coingecko_id='dogecoin'): ['Dogecoin'], CryptoCurrency('SOL', coingecko_id='solana'): ['Solana SOL'], CryptoCurrency('ADA', coingecko_id='cardano'): ['Cardano ADA'], CryptoCurrency('BNB', coingecko_id='bnb'): ['Binance BNB'], CryptoCurrency('MATIC', coingecko_id='polygon'): ['Polygon MATIC'], # Stable-coins CryptoCurrency('DAI', coingecko_id='dai'): ['DAI'], CryptoCurrency('USDC', coingecko_id='usdc'): ['USD Coin'], 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 = ( CURRENCY_CODES | {'XXBT': BTC} | {k.raw_short_name(): k for k in COMMON_NAMES} | {'SPX500': SPX, 'SP500': SPX, 'Nasdaq 100': NDX} ) CURRENCY_SYMBOLS: dict[Currency, str] = { USD: '$', EUR: '€', FiatCurrency.GBP: '£', BTC: '₿', FiatCurrency.JPY: '¥', } ASSET_PREFIX: dict[Asset, str] = CURRENCY_SYMBOLS # TODO: Remove at some point. 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: 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/) THREE_DECIMALS_UNDER_THIS_AMOUNT = 0.10 @enforce_typing.enforce_types @dataclasses.dataclass(frozen=True, eq=True, slots=True) class AssetAmount: """Decimal with associated asset unit. Fields: - `asset`: Asset unit of amount. - `amount`: Amount of given asset. """ asset: Asset amount: Decimal def __str__(self) -> str: return self.human_readable_str() def human_readable_str(self) -> str: abs_amount = abs(self.amount) specificity = '2' if abs_amount >= THREE_DECIMALS_UNDER_THIS_AMOUNT else '3' prefix = ASSET_PREFIX.get(self.asset, '') return ('{sign}{prefix}{amount:.' + specificity + 'f} {name}').format( sign='-' if self.amount < 0 else '', prefix=prefix, amount=abs_amount, name=self.asset.raw_short_name(), ) def __mul__(self, other: Decimal) -> 'AssetAmount': if not isinstance(other, Decimal): msg = f'other must be Decimal, but was {type(other).__name__}' raise TypeError(msg) return AssetAmount(self.asset, self.amount * other) def __add__(self, other: 'AssetAmount') -> 'AssetAmount': if not isinstance(other, AssetAmount): msg = f'other must be AssetAmount, but was {type(other).__name__}' raise TypeError(msg) if self.is_zero(): return other if other.is_zero(): return self 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: 'AssetAmount') -> 'AssetAmount': # TODO: Implement using add? if not isinstance(other, AssetAmount): msg = f'other must be AssetAmount, but was {type(other).__name__}' raise TypeError(msg) if self.is_zero(): return -other if other.is_zero(): return self 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: 'Decimal | AssetAmount') -> 'Decimal | AssetAmount': 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 __neg__(self) -> 'AssetAmount': return AssetAmount(self.asset, -self.amount) def cmp(self, other) -> Decimal: if not isinstance(other, AssetAmount): msg = f'other must be AssetAmount, but was {type(other).__name__}' raise TypeError(msg) if self.is_zero() or other.is_zero(): return self.amount - other.amount 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 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] @dataclasses.dataclass(frozen=True, eq=True) class ExchangeRateSample: """Single exchange rate sample with a timestamp and various stats.""" timestamp: datetime.date | datetime.datetime asset_pair: AssetPair _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: datetime.date | datetime.datetime, ) -> 'ExchangeRateSample': return dataclasses.replace(self, timestamp=timestamp) def invert_exchange_rate(self) -> 'ExchangeRateSample': one = Decimal('1') return dataclasses.replace( self, asset_pair=(self.asset_pair[1], self.asset_pair[0]), _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 )