diff --git a/fin_defs/__init__.py b/fin_defs/__init__.py index db93f0e..f31883c 100644 --- a/fin_defs/__init__.py +++ b/fin_defs/__init__.py @@ -16,15 +16,10 @@ Defined hierarchy: - `Commodity` """ -import abc -import dataclasses -import datetime -import re -from collections.abc import Mapping -from decimal import Decimal - from ._version import __version__ +from .data import * + __all__ = [ 'Asset', 'AssetAmount', @@ -33,618 +28,3 @@ __all__ = [ '__version__', ] - -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+$' - - -@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', -) - - -@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() - - -@dataclasses.dataclass(frozen=True) -class UnknownAsset(Asset): - """An asset that does not exist.""" - - @abc.abstractmethod - def raw_short_name(self) -> str: - return '???' - - -@dataclasses.dataclass(frozen=True) -class Currency(Asset): - """Either a Fiat or a Crypto Currency.""" - - -@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') - - -@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 - - -@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 - - -@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 - - -@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 - - -@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 - - -def assert_same_asset(do: str, a_asset: 'AssetAmount', b_asset: 'AssetAmount'): - if a_asset.asset != b_asset.asset: - msg = f'Attempting to {do} {a_asset} and {b_asset}, but these must have the same asset' - raise ValueError(msg) - - -@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 __post_init__(self): - assert isinstance(self.asset, Asset), 'Incorrect value for self.asset' - assert isinstance(self.amount, Decimal), 'Incorrect value for self.amount' - - def __str__(self) -> str: - return self.human_readable_str() - - def human_readable_str(self) -> str: - abs_amount = abs(self.amount) - - # Determine specificity - specificity = '3' - if abs_amount != abs_amount: - specificity = '0' - elif abs_amount >= THREE_DECIMALS_UNDER_THIS_AMOUNT: - specificity = '2' - - prefix = ASSET_PREFIX.get(self.asset, '') - return ('{sign}{prefix}{amount:.' + specificity + 'f} {name}').format( - sign='-' if (self.amount == self.amount and 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 - assert_same_asset('add', self, other) - - 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 - assert_same_asset('subtract', self, other) - return AssetAmount(self.asset, self.amount - other.amount) - - def __truediv__(self, other: 'Decimal | AssetAmount') -> 'Decimal | AssetAmount': - if isinstance(other, AssetAmount): - assert_same_asset('divide', self, other) - 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 - assert_same_asset('compare', self, other) - 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 - ) diff --git a/fin_defs/data.py b/fin_defs/data.py new file mode 100644 index 0000000..ac77ce7 --- /dev/null +++ b/fin_defs/data.py @@ -0,0 +1,621 @@ +import abc +import dataclasses +import datetime +import re +from collections.abc import Mapping +from decimal import Decimal + +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+$' + + +@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', +) + + +@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() + + +@dataclasses.dataclass(frozen=True) +class UnknownAsset(Asset): + """An asset that does not exist.""" + + @abc.abstractmethod + def raw_short_name(self) -> str: + return '???' + + +@dataclasses.dataclass(frozen=True) +class Currency(Asset): + """Either a Fiat or a Crypto Currency.""" + + +@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') + + +@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 + + +@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 + + +@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 + + +@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 + + +@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 + + +def assert_same_asset(do: str, a_asset: 'AssetAmount', b_asset: 'AssetAmount'): + if a_asset.asset != b_asset.asset: + msg = f'Attempting to {do} {a_asset} and {b_asset}, but these must have the same asset' + raise ValueError(msg) + + +@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 __post_init__(self): + assert isinstance(self.asset, Asset), 'Incorrect value for self.asset' + assert isinstance(self.amount, Decimal), 'Incorrect value for self.amount' + + def __str__(self) -> str: + return self.human_readable_str() + + def human_readable_str(self) -> str: + abs_amount = abs(self.amount) + + # Determine specificity + specificity = '3' + if abs_amount != abs_amount: + specificity = '0' + elif abs_amount >= THREE_DECIMALS_UNDER_THIS_AMOUNT: + specificity = '2' + + prefix = ASSET_PREFIX.get(self.asset, '') + return ('{sign}{prefix}{amount:.' + specificity + 'f} {name}').format( + sign='-' if (self.amount == self.amount and 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 + assert_same_asset('add', self, other) + + 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 + assert_same_asset('subtract', self, other) + return AssetAmount(self.asset, self.amount - other.amount) + + def __truediv__(self, other: 'Decimal | AssetAmount') -> 'Decimal | AssetAmount': + if isinstance(other, AssetAmount): + assert_same_asset('divide', self, other) + 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 + assert_same_asset('compare', self, other) + 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 + ) diff --git a/fin_defs/parse_price.py b/fin_defs/parse_price.py new file mode 100644 index 0000000..d34cb4e --- /dev/null +++ b/fin_defs/parse_price.py @@ -0,0 +1,66 @@ +import logging +import re +from decimal import Decimal + +from .data import DKK, USD, Asset, AssetAmount, FiatCurrency + +logger = logging.getLogger(__name__) + +RE_PRICE_RAW = r'\b(?:dkk|sek|usd|nok|eur)?\s*([1-9][\d.]*[\d](?:,\d+)?)\s*(?:,-|:-|.-|;-)?\s*(?:(?:kr|kroner|krone|dkk|sek|usd|eur|nok)\b)?\.?' + +RE_PRICE = re.compile(RE_PRICE_RAW, flags=re.IGNORECASE) + +def parse_price(text: str, default_currency: Asset) -> AssetAmount | None: + """ + Attempts to parse price from the given text. + + Text does not need to be stripped beforehand. + """ + if isinstance(text, AssetAmount): + return text + text = str(text) + + if m := re.match(r'^Kr\s*([\d.]+(?:,\d+))?$', text): + return AssetAmount( + DKK, + Decimal(m.group(1).replace('.', '').replace(',', '.')), + ) + if m := re.match(r'^(\d+)\s*DKK$', text): + return AssetAmount(DKK, Decimal(m.group(1))) + + if m := re.match(r'^\$\s*([0-9.]+)(\s+USD)?$', text): + return AssetAmount(USD, Decimal(m.group(1))) + + if text.lower().strip() == 'free': + return AssetAmount(default_currency, Decimal(0.0)) + + text = str(text).strip().lower().removesuffix('.') + if m := RE_PRICE.fullmatch(text): + currency = default_currency + price_tag = m.group(1).replace('.', '').replace(',', '.') # TODO + if text.endswith('dkk') or text.startswith('dkk'): + currency = FiatCurrency('DKK') + elif text.endswith('sek') or text.startswith('sek'): + currency = FiatCurrency('SEK') + elif text.endswith('nok') or text.startswith('nok'): + currency = FiatCurrency('NOK') + elif text.endswith('usd') or text.startswith('usd'): + currency = FiatCurrency('USD') + + return AssetAmount(currency, Decimal(price_tag)) + + logger.warning('Unknown price format: %s', text) + return None + + +def parse_usd_price(s: str | int) -> AssetAmount: + assert s is not None + if isinstance(s, str) or isinstance(s, int): + text = str(s) + else: + text = s.text_content() + text = text.strip().replace(',', '').removeprefix('$') + if text in {'-', ''}: + return AssetAmount(USD, Decimal(0)) # TODO + dollar_amount = Decimal(text) + return AssetAmount(USD, dollar_amount) diff --git a/test/test_parse_price.py b/test/test_parse_price.py new file mode 100644 index 0000000..807e45c --- /dev/null +++ b/test/test_parse_price.py @@ -0,0 +1,36 @@ +import pytest +from decimal import Decimal +from fin_defs import DKK, USD, AssetAmount, FiatCurrency +from fin_defs.parse_price import parse_price + +def dkk(amount): + return AssetAmount(DKK, Decimal(amount)) + +PRICES_PARSABLE = [ + ('DKK100', dkk(100)), + ('100;-', dkk(100)), + ('100 kr', dkk(100)), + (' 100 kr ', dkk(100)), + ('349.-', dkk(349)), + ('3.000 kr.', dkk(3000)), + ('25,00 kr.', dkk(25)), + ('300,00 kr.', dkk(300)), + ('300kr.', dkk(300)), + ('300kr', dkk(300)), + ('5,00 dkk', dkk(5)), + ('9,99 dkk', dkk('9.99')), + ('17900 kr', dkk(17900)), +] + +PRICES_UNPARSABLE = [ + '007', +] + +@pytest.mark.parametrize(('price_string', 'parsed_amount'), PRICES_PARSABLE) +def test_parse_price(price_string: str, parsed_amount: AssetAmount): + result = parse_price(price_string, parsed_amount.asset) + assert result == parsed_amount + +@pytest.mark.parametrize('price_string', PRICES_UNPARSABLE) +def test_parse_unparsable(price_string: str): + assert parse_price(price_string, USD) is None