"""# 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 decimal import Decimal import enforce_typing from ._version import __version__ # noqa: F401 def parse_attr_data(attr_data: str) -> dict[str, str | int]: d = {} if attr_data is None: return d for s in attr_data.split(','): s = s.strip() if s == '': continue (attr_key, attr_value) = s.split('=') if re.match(r'^\d+$', attr_value): attr_value = int(attr_value) d[attr_key] = attr_value 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 @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 self type: {self}' raise Exception(msg) @staticmethod def from_polygon_id(polygon_id: str) -> '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) @abc.abstractmethod def raw_short_name(self) -> str: """A short name or ticker that 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}' msg = f'Unsupported asset type: {repr(self)}' raise NotImplementedError(msg) @staticmethod def from_string_id(asset_id: str, shortcut_well_known=True) -> '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 NotImplementedError(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_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 == 'currency': return FiatCurrency(ticker, **attrs) if prefix == 'fiat': return FiatCurrency(ticker, **attrs) if prefix == 'index': return Index(ticker, **attrs) if prefix == 'crypto': return CryptoCurrency(ticker, **attrs) msg = f'Unsupported asset format: {asset_id}' raise NotImplementedError(msg) def __str__(self): return self.to_string_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 raw_short_name(self) -> str: return self.iso_code @enforce_typing.enforce_types @dataclasses.dataclass(frozen=True, eq=True) class CryptoCurrency(Currency): ccxt_symbol: str coingecko_id: str | None = 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 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): 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, stock_exchange: StockExchange) -> 'Stock': return Stock(polygon_id, stock_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): 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): 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 # TODO 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') GBP = FiatCurrency('GBP') 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'], } WELL_KNOWN_SYMBOLS = ( { sym: FiatCurrency(sym) for sym in [ 'USD', 'DKK', 'EUR', 'JPY', 'CAD', 'GBP', 'CNY', 'PLN', 'ILS', 'CHF', 'ZAR', 'CZK', ] } | {'XXBT': BTC} | {k.raw_short_name(): k for k in COMMON_NAMES} | {'SPX500': SPX, 'SP500': SPX, 'Nasdaq 100': NDX} ) ASSET_PREFIX: dict[Asset, str] = { USD: '$', EUR: '€', GBP: '£', BTC: '₿', } 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. Fields: - `asset`: Asset unit of amount. - `amount`: Amount of given asset. """ asset: Asset amount: Decimal def __str__(self): return self.human_readable_str() def human_readable_str(self): specificity = '2' if self.amount >= 0.10 else '3' prefix = ASSET_PREFIX.get(self.asset, '') return ('{}{:.' + specificity + 'f} {}').format( prefix, self.amount, self.asset.raw_short_name() ) 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 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): return dataclasses.replace(self, timestamp=timestamp) def invert_exchange_rate(self): 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 )