fin-defs/fin_defs/__init__.py

"""# 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 __post_init__(self) -> None:
        if not re.match(r'^\S.*\S$', 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) -> 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
        )