Jmaa/fin-defs
1
0
Fork 0
Code Actions Packages

Initialize fin-defs

Browse Source
This commit is contained in:
Jon Michael Aanes 2024-05-27 23:03:16 +02:00
parent 95d6db21ec
commit 144f9ba8ed
Signed by: Jmaa
SSH Key Fingerprint: SHA256:Ab0GfHGCblESJx7JRE4fj4bFy/KRpeLhi41y4pF3sNA
9 changed files with 538 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
.gitea/workflows/python-package.yml Normal file
View File

@ -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 }}

18
.gitignore vendored Normal file
View File

@ -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/

412
fin_defs/__init__.py Normal file
View File

@ -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
)

1
fin_defs/_version.py Normal file
View File

@ -0,0 +1 @@
__version__ = '0.1.10'

1
requirements.txt Normal file
View File

@ -0,0 +1 @@
enforce-typing

1
requirements_test.txt Normal file
View File

@ -0,0 +1 @@
pytest

68
setup.py Normal file
View File

@ -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',
)

0
test/__init__.py Normal file
View File

23
test/test_data.py Normal file
View File

@ -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'