Restructure by moving data stuff into data module, and added price parsing
This commit is contained in:
parent
aa2dd36051
commit
066b8cdac7
|
|
@ -16,15 +16,10 @@ Defined hierarchy:
|
||||||
- `Commodity`
|
- `Commodity`
|
||||||
"""
|
"""
|
||||||
|
|
||||||
import abc
|
|
||||||
import dataclasses
|
|
||||||
import datetime
|
|
||||||
import re
|
|
||||||
from collections.abc import Mapping
|
|
||||||
from decimal import Decimal
|
|
||||||
|
|
||||||
from ._version import __version__
|
from ._version import __version__
|
||||||
|
|
||||||
|
from .data import *
|
||||||
|
|
||||||
__all__ = [
|
__all__ = [
|
||||||
'Asset',
|
'Asset',
|
||||||
'AssetAmount',
|
'AssetAmount',
|
||||||
|
|
@ -33,618 +28,3 @@ __all__ = [
|
||||||
'__version__',
|
'__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
|
|
||||||
)
|
|
||||||
|
|
|
||||||
621
fin_defs/data.py
Normal file
621
fin_defs/data.py
Normal file
|
|
@ -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
|
||||||
|
)
|
||||||
66
fin_defs/parse_price.py
Normal file
66
fin_defs/parse_price.py
Normal file
|
|
@ -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)
|
||||||
36
test/test_parse_price.py
Normal file
36
test/test_parse_price.py
Normal file
|
|
@ -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
|
||||||
Loading…
Reference in New Issue
Block a user