From 144f9ba8ed903d715c366484c512e8b715f21b77 Mon Sep 17 00:00:00 2001 From: Jon Michael Aanes Date: Mon, 27 May 2024 23:03:16 +0200 Subject: [PATCH] Initialize fin-defs --- .gitea/workflows/python-package.yml | 14 + .gitignore | 18 ++ fin_defs/__init__.py | 412 ++++++++++++++++++++++++++++ fin_defs/_version.py | 1 + requirements.txt | 1 + requirements_test.txt | 1 + setup.py | 68 +++++ test/__init__.py | 0 test/test_data.py | 23 ++ 9 files changed, 538 insertions(+) create mode 100644 .gitea/workflows/python-package.yml create mode 100644 .gitignore create mode 100644 fin_defs/__init__.py create mode 100644 fin_defs/_version.py create mode 100644 requirements.txt create mode 100644 requirements_test.txt create mode 100644 setup.py create mode 100644 test/__init__.py create mode 100644 test/test_data.py diff --git a/.gitea/workflows/python-package.yml b/.gitea/workflows/python-package.yml new file mode 100644 index 0000000..f0512f8 --- /dev/null +++ b/.gitea/workflows/python-package.yml @@ -0,0 +1,14 @@ +name: Test and Package Python +on: [push] + +jobs: + Test: + uses: jmaa/workflows/.gitea/workflows/python-test.yaml@v6.21 + Package: + uses: jmaa/workflows/.gitea/workflows/python-package.yaml@v6.21 + with: + REGISTRY_DOMAIN: gitfub.space + REGISTRY_ORGANIZATION: usagi-keiretsu + secrets: + PIPY_REPO_USER: ${{ secrets.PIPY_REPO_USER }} + PIPY_REPO_PASS: ${{ secrets.PIPY_REPO_PASS }} diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..e6e989c --- /dev/null +++ b/.gitignore @@ -0,0 +1,18 @@ +# Program specific +/output/ +/deps/ +/secrets/ +/private_deps/ +/data/ + +# Python +__pycache__/ +/build/ +/dist/ +*.egg-info/ + +# Python, Testing +/test/secrets.py +/.coverage +/.hypothesis/ +/htmlcov/ diff --git a/fin_defs/__init__.py b/fin_defs/__init__.py new file mode 100644 index 0000000..bf65ad7 --- /dev/null +++ b/fin_defs/__init__.py @@ -0,0 +1,412 @@ +import abc +import dataclasses +import datetime +import re +from collections.abc import Iterator +from decimal import Decimal + +import enforce_typing + +## Ids + +RE_TICKER_FORMAT = r'^[A-Z0-9_]+$' +RE_TICKER_FORMAT_FLEXIBLE = r'^[^:\s]+$' +RE_CRYPTO_TICKER_FORMAT = r'^\S+$' + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True) +class StockExchange: + """Unified Stock Exchange identifiers.""" + + name: str + mic: str + bloomberg_id: str | None = None + crunchbase_id: str | None = None + yahoo_id: str | None = None + eod_id: str | None = None + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True) +class Asset: + + def names(self) -> list[str]: + return COMMON_NAMES.get(self, []) + + def to_polygon_id(self) -> str: + # TODO: Support currency pairs not involving USD. + if isinstance(self, Stock): + return f'{self.ticker}' + if isinstance(self, Index): + return f'I:{self.ticker}' + if isinstance(self, FiatCurrency): + return f'C:{self.iso_code}USD' + if isinstance(self, CryptoCurrency): + return f'X:{self.ccxt_symbol}USD' + + # Else + msg = f'Unknown self type: {self}' + raise Exception(msg) + + @staticmethod + def from_polygon_id(polygon_id: str) -> 'Asset': + 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) + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True) +class Currency(Asset): + pass + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, eq=True, order=True) +class FiatCurrency(Currency): + iso_code: str + + def __post_init__(self): + if not re.match(RE_TICKER_FORMAT, self.iso_code): + msg = f'iso_code was not in correct format: {self.iso_code}' + raise ValueError(msg) + + def __str__(self): + return self.iso_code + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, eq=True) +class CryptoCurrency(Currency): + ccxt_symbol: str + coingecko_id: str | None + + def __post_init__(self): + '''TODO + if not re.match(RE_CRYPTO_TICKER_FORMAT, self.ccxt_symbol): + msg = f'ccxt_symbol was not in correct format: {self.ccxt_symbol}' + raise ValueError(msg) + ''' + + def __str__(self): + return self.ccxt_symbol + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, eq=True) +class Stock(Asset): + ticker: str + exchange: StockExchange + + def __post_init__(self): + if not re.match(RE_TICKER_FORMAT_FLEXIBLE, self.ticker): + msg = f'ticker was not in correct format: {self.ticker}' + raise ValueError(msg) + + def __str__(self): + return f'{self.ticker}.{self.exchange.mic}' + + @staticmethod + def from_polygon_id(polygon_id: str, stock_exchange: StockExchange) -> 'Stock': + return Stock(polygon_id, stock_exchange) + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, eq=True) +class Index(Asset): + ticker: str + + def __post_init__(self): + if not re.match(RE_TICKER_FORMAT_FLEXIBLE, self.ticker): + msg = f'ticker was not in correct format: {self.ticker}' + raise ValueError(msg) + + def __str__(self): + return self.ticker + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, eq=True) +class Commodity(Asset): + alpha_vantage_id: str + + def __post_init__(self): + if not re.match(RE_TICKER_FORMAT, self.alpha_vantage_id): + msg = f'alpha_vantage_id was not in correct format: {self.alpha_vantage_id}' + raise ValueError(msg) + + def __str__(self): + 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') + +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', 'ethereum'): ['Ethereum Ether'], + CryptoCurrency('DOGE', 'dogecoin'): ['Dogecoin'], + CryptoCurrency('SOL', 'solana'): ['Solana SOL'], + CryptoCurrency('ADA', 'cardano'): ['Cardano ADA'], + CryptoCurrency('BNB', 'bnb'): ['Binance BNB'], + CryptoCurrency('MATIC', 'polygon'): ['Polygon MATIC'], + # Stable-coins + CryptoCurrency('DAI', 'dai'): ['DAI'], + CryptoCurrency('USDC', 'usdc'): ['USD Coin'], + CryptoCurrency('USDT', 'tether'): ['Tether USDT'], +} + +WELL_KNOWN_SYMBOLS = ( + { + sym: FiatCurrency(sym) + for sym in [ + 'USD', + 'DKK', + 'EUR', + 'JPY', + 'CAD', + 'GBP', + 'CNY', + 'PLN', + 'ILS', + 'CHF', + 'ZAR', + 'CZK', + ] + } + | {str(k): k for k in COMMON_NAMES} + | {'SPX500': SPX, 'SP500': SPX, 'Nasdaq 100': NDX} +) + +NYSE = StockExchange( + name='New York Stock Exchange', + mic='XNYS', + bloomberg_id='UN', + crunchbase_id='NYSE', +) + +EXCHANGES = [ + NYSE, + StockExchange(name='Nasdaq', mic='XNAS', bloomberg_id='US', crunchbase_id='NASDAQ'), + StockExchange( + name='Hong Kong Exchanges And Clearing', + mic='XHKG', + bloomberg_id='HK', + crunchbase_id='SEHK', + ), + StockExchange( + name='Moscow Exchange', + mic='MISC', + bloomberg_id='RX', + yahoo_id='ME', + eod_id='MCX', + crunchbase_id='MOEX', + ), + StockExchange(name='Shanghai Stock Exchange', mic='XSHG'), + StockExchange(name='Euronext Amsterdam', mic='XAMS'), + StockExchange(name='Euronext Brussels', mic='XBRU'), + StockExchange(name='Euronext Dublin', mic='XMSM'), + StockExchange(name='Euronext Lisbon', mic='XLIS'), + StockExchange(name='Euronext Milan', mic='XMIL'), + StockExchange(name='Euronext Oslo', mic='XOSL'), + StockExchange(name='Euronext Paris', mic='XPAR'), + StockExchange(name='Japan Exchange Group', mic='XJPX'), + StockExchange(name='Shenzhen Stock Exchange', mic='XSHE'), + StockExchange(name='Indian National Stock Exchange', mic='XNSE'), + StockExchange( + name='Bombay Stock Exchange', + mic='XBOM', + bloomberg_id='IB', + crunchbase_id='BOM', + yahoo_id='BO', + ), + StockExchange(name='Toronto Stock Exchange', mic='XTSE'), + StockExchange(name='London Stock Exchange', mic='XLON'), + StockExchange(name='Saudi Stock Exchange', mic='XSAU'), + StockExchange(name='German Stock Exchange', mic='XFRA'), + StockExchange(name='SIX Swiss Exchange', mic='XSWX'), + StockExchange(name='Nasdaq Copenhagen Stock Exchange', mic='XCSE'), + StockExchange(name='Nasdaq Stockholm Stock Exchange', mic='XSTO'), + StockExchange(name='Nasdaq Helsinki Stock Exchange', mic='XHEL'), + StockExchange(name='Nasdaq Tallinn Stock Exchange', mic='XTAL'), + StockExchange(name='Nasdaq Riga Stock Exchange', mic='XRIS'), + StockExchange(name='Nasdaq Vilnius Stock Exchange', mic='XLIT'), + StockExchange(name='Nasdaq Iceland Stock Exchange', mic='XICE'), + StockExchange(name='Korea Exchange', mic='XKOS'), + StockExchange(name='Taiwan Stock Exchange', mic='XTAI'), + StockExchange(name='Australian Securities Exchange', mic='XASX'), + StockExchange(name='Johannesburg Stock Exchange', mic='XJSE'), + StockExchange(name='Tehran Stock Exchange', mic='XTEH'), +] + +EXCHANGES_BY_IDS = {} +if True: + + def add_by_id(exchange: StockExchange, key: str): + exchange_id = getattr(exchange, key) + if exchange_id is not None: + if exchange_id in EXCHANGES_BY_IDS: + msg = f'Exchange {exchange_id} is already present' + raise ValueError(msg) + EXCHANGES_BY_IDS[exchange_id] = exchange + + for exchange in EXCHANGES: + add_by_id(exchange, 'mic') + add_by_id(exchange, 'bloomberg_id') + add_by_id(exchange, 'crunchbase_id') + add_by_id(exchange, 'yahoo_id') + add_by_id(exchange, 'eod_id') + del exchange + del add_by_id + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, slots=True) +class AssetInformation: + """Extensive information about a given asset.""" + + asset: Asset + full_name: str + description: str | None = None + asset_base: Asset | None = None + sec_cik: str | None = None + # TODO: FIGI types? (https://www.openfigi.com/) + + +@enforce_typing.enforce_types +@dataclasses.dataclass(frozen=True, eq=True, slots=True) +class AssetAmount: + """Decimal with associated asset unit.""" + + asset: Asset + amount: Decimal + + def __str__(self): + specificity = '2' if self.amount >= 0.10 else '3' + return ('{:.' + specificity + 'f} {}').format(self.amount, self.asset) + + def __mul__(self, other: Decimal): + if not isinstance(other, Decimal): + msg = f'other must be decimal, but was {type(other)}' + raise TypeError(msg) + return AssetAmount(self.asset, self.amount * other) + + def __add__(self, other): + if not isinstance(other, AssetAmount): + msg = f'other must be AssetAmount, but was {type(other)}' + raise TypeError(msg) + if self.asset != other.asset: + msg = ( + f'AssetAmount must have same asset, but: {self.asset} != {other.asset}' + ) + raise ValueError(msg) + + return AssetAmount(self.asset, self.amount + other.amount) + + def __sub__(self, other): + if not isinstance(other, AssetAmount): + msg = f'other must be AssetAmount, but was {type(other)}' + raise TypeError(msg) + if self.asset != other.asset: + msg = ( + f'AssetAmount must have same asset, but: {self.asset} != {other.asset}' + ) + raise ValueError(msg) + return AssetAmount(self.asset, self.amount - other.amount) + + def __truediv__(self, other): + if isinstance(other, AssetAmount): + if self.asset != other.asset: + msg = f'AssetAmount must have same asset, but: {self.asset} != {other.asset}' + raise ValueError(msg) + return self.amount / other.amount + return AssetAmount(self.asset, self.amount / other) + + def __lt__(self, other): + if not isinstance(other, AssetAmount): + msg = f'other must be AssetAmount, but was {type(other)}' + raise TypeError(msg) + if self.asset != other.asset: + msg = ( + f'AssetAmount must have same asset, but: {self.asset} != {other.asset}' + ) + raise ValueError(msg) + return self.amount < other.amount + + +@dataclasses.dataclass(frozen=True, eq=True) +class ExchangeRateSample(Asset): + """Single exchange rate sample with a timestamp and various stats.""" + + timestamp: datetime.date | datetime.datetime + _average: Decimal | None = None + last: Decimal | None = None + open: Decimal | None = None + close: Decimal | None = None + high: Decimal | None = None + low: Decimal | None = None + volume: Decimal | None = None + market_cap: Decimal | None = None + + @property + def average(self) -> Decimal: + if self._average: + return self._average + return (self.high + self.low) / 2 + + def replace_timestamp(self, timestamp): + return dataclasses.replace(self, timestamp=timestamp) + + def invert_exchange_rate(self): + one = Decimal('1') + return dataclasses.replace( + self, + _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/_version.py b/fin_defs/_version.py new file mode 100644 index 0000000..850505a --- /dev/null +++ b/fin_defs/_version.py @@ -0,0 +1 @@ +__version__ = '0.1.10' diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..63a601e --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +enforce-typing diff --git a/requirements_test.txt b/requirements_test.txt new file mode 100644 index 0000000..e079f8a --- /dev/null +++ b/requirements_test.txt @@ -0,0 +1 @@ +pytest diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..091c8a2 --- /dev/null +++ b/setup.py @@ -0,0 +1,68 @@ +#!/usr/bin/env python +# +# WARNING +# +# THIS IS AN AUTOGENERATED FILE. +# +# MANUAL CHANGES CAN AND WILL BE OVERWRITTEN. + +import re + +from setuptools import setup + +PACKAGE_NAME = 'fin_defs' + +with open('README.md') as f: + readme = f.read() + +with open(PACKAGE_NAME + '/_version.py') as f: + text = f.read() + match = re.match(r'^__version__\s*=\s*(["\'])([\d\.]+)\1$', text) + version = match.group(2) + del match, text + + +def parse_requirements(text: str) -> list[str]: + return text.strip().split('\n') + + +def read_requirements(path: str): + with open(path) as f: + return parse_requirements(f.read()) + + +def get_short_description(readme: str): + readme = re.sub(r'#+[^\n]*\n+', '', readme) + m = re.search(r'^\s*(\w+[\w\s,`-]+\.)', readme) + try: + return m.group(1) + except AttributeError as err: + msg = 'Could not determine short description' + raise Exception(msg) from err + + +REQUIREMENTS_MAIN = """ +enforce-typing +""" + +REQUIREMENTS_TEST = """ +pytest +""" + + +setup( + name=PACKAGE_NAME, + version=version, + description=get_short_description(readme), + long_description=readme, + long_description_content_type='text/markdown', + author='Jmaa', + author_email='jonjmaa@gmail.com', + url='https://gitfub.space/Jmaa/' + PACKAGE_NAME, + packages=[PACKAGE_NAME], + install_requires=parse_requirements(REQUIREMENTS_MAIN), + extras_require={ + 'test': parse_requirements(REQUIREMENTS_TEST), + }, + python_requires='>=3.9', +) diff --git a/test/__init__.py b/test/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/test/test_data.py b/test/test_data.py new file mode 100644 index 0000000..055c933 --- /dev/null +++ b/test/test_data.py @@ -0,0 +1,23 @@ +import pytest + +import fin_defs + +BAD_TICKERS = ['TEST123', '123', 'TEST.EUR', 'TEST:EUR', 'EUR:TEST'] + + +@pytest.mark.parametrize('ticker', BAD_TICKERS) +def test_bad_tickers(ticker): + try: + fin_defs.Stock(ticker) + except Exception as e: + assert e + + +@pytest.mark.parametrize('ticker', BAD_TICKERS) +def test_crypto_tickers(ticker): + fin_defs.CryptoCurrency(ticker, 'not-known') + + +def test_str(): + NVO = fin_defs.Stock('NVO', fin_defs.EXCHANGES_BY_IDS['NYSE']) + assert str(NVO) == 'NVO.XNYS'