From 066b8cdac7aac4166cf83896586fbb4f2ade100c Mon Sep 17 00:00:00 2001
From: Jon Michael Aanes <jonjmaa@gmail.com>
Date: Wed, 14 May 2025 20:33:15 +0200
Subject: [PATCH] Restructure by moving data stuff into data module, and added
price parsing
---
fin_defs/__init__.py | 624 +--------------------------------------
fin_defs/data.py | 621 ++++++++++++++++++++++++++++++++++++++
fin_defs/parse_price.py | 66 +++++
test/test_parse_price.py | 36 +++
4 files changed, 725 insertions(+), 622 deletions(-)
create mode 100644 fin_defs/data.py
create mode 100644 fin_defs/parse_price.py
create mode 100644 test/test_parse_price.py
diff --git a/fin_defs/__init__.py b/fin_defs/__init__.py
index db93f0e..f31883c 100644
--- a/fin_defs/__init__.py
+++ b/fin_defs/__init__.py
@@ -16,15 +16,10 @@ Defined hierarchy:
- `Commodity`
"""
-import abc
-import dataclasses
-import datetime
-import re
-from collections.abc import Mapping
-from decimal import Decimal
-
from ._version import __version__
+from .data import *
+
__all__ = [
'Asset',
'AssetAmount',
@@ -33,618 +28,3 @@ __all__ = [
'__version__',
]
-
-def parse_id_attr_key_value_pair(attr_datum: str) -> tuple[str, str | int] | None:
- attr_datum = attr_datum.strip()
- if attr_datum == '':
- return None
- (attr_key, attr_value) = attr_datum.split('=')
- if re.match(r'^\d+$', attr_value):
- attr_value = int(attr_value)
- return attr_key, attr_value
-
-
-def parse_id_attr_data(attr_data: str) -> Mapping[str, str | int]:
- if attr_data is None:
- return {}
- key_value_pairs_str = attr_data.split(',')
- key_value_pairs = [parse_id_attr_key_value_pair(kvp) for kvp in key_value_pairs_str]
-
- d = {}
- for kvp in key_value_pairs:
- if kvp is None:
- continue
- d[kvp[0]] = kvp[1]
- return d
-
-
-## Ids
-
-RE_TICKER_FORMAT = r'^[A-Z0-9_]+$'
-RE_TICKER_FORMAT_FLEXIBLE = r'^[^:\s](?:[^:]*[^:\s])?$'
-RE_CRYPTO_TICKER_FORMAT = r'^\S+$'
-
-
-@dataclasses.dataclass(frozen=True)
-class StockExchange:
- """Unified Stock Exchange identifiers.
-
- Fields:
- - `name`: Human-readable name.
- - `mic`: Official MIC identifier.
- - `bloomberg_id`: Identifier for lookup on Bloomberg.
- - `crunchbase_id`: Identifier for lookup on Crunchbase.
- - `yahoo_id`: Identifier for lookup on Yahoo! Finance.
- - `eod_id`: Identifier for lookup on EOD.
- """
-
- name: str
- mic: str
- bloomberg_id: str | None = None
- crunchbase_id: str | None = None
- yahoo_id: str | None = None
- eod_id: str | None = None
-
-
-NYSE = StockExchange(
- name='New York Stock Exchange',
- mic='XNYS',
- bloomberg_id='UN',
- crunchbase_id='NYSE',
-)
-
-
-@dataclasses.dataclass(frozen=True)
-class Asset:
- """An identifier representing some abstract financial asset.
-
- Subtypes represent the various asset types in existence, for example fiat
- currencies, stocks or even crypto currencies.
- """
-
- def names(self) -> list[str]:
- """Returns a list of human readable names for this `Asset`."""
- return COMMON_NAMES.get(self, [])
-
- def to_polygon_id(self) -> str:
- """Formats this asset to the [`polygon.io`](https://polygon.io)
- identifier format.
-
- Inverse of `from_polygon_id`.
- """
- # TODO: Support currency pairs not involving USD.
- if isinstance(self, Stock):
- return f'{self.ticker}'
- if isinstance(self, Index):
- return f'I:{self.ticker}'
- if isinstance(self, FiatCurrency):
- return f'C:{self.iso_code}USD'
- if isinstance(self, CryptoCurrency):
- return f'X:{self.ccxt_symbol}USD'
-
- # Else
- msg = f'Unknown type: {type(self).__name__}'
- raise TypeError(msg)
-
- @staticmethod
- def from_polygon_id(polygon_id: str, exchange: StockExchange = NYSE) -> 'Asset':
- """Parses an asset from the [`polygon.io`](https://polygon.io)
- identifier format.
-
- Inverse of `to_polygon_id`.
- """
- if polygon_id.startswith('I:'):
- return Index(polygon_id.removeprefix('I:'))
- if polygon_id.startswith('X:'):
- return CryptoCurrency(
- polygon_id.removeprefix('X:').removesuffix('USD'),
- None,
- )
- if polygon_id.startswith('C:'):
- return FiatCurrency(polygon_id.removeprefix('C:').removesuffix('USD'))
- return Stock.from_polygon_id(polygon_id, exchange)
-
- @abc.abstractmethod
- def raw_short_name(self) -> str:
- """Short name or ticker.
-
- Does not nessisarily uniquely identify the asset, but may be commonly used within the industry.
- """
-
- def to_string_id(self) -> str:
- """Formats the asset id using the fin_defs asset format."""
- if isinstance(self, Stock):
- attrs_str = f'{{nordnet_id={self.nordnet_id}}}' if self.nordnet_id else ''
- return f'stock:{self.ticker}.{self.exchange.mic}{attrs_str}'
- if isinstance(self, FiatCurrency):
- return f'fiat:{self.iso_code}'
- if isinstance(self, CryptoCurrency):
- attrs_str = (
- f'{{coingecko_id={self.coingecko_id}}}' if self.coingecko_id else ''
- )
- return f'crypto:{self.ccxt_symbol}{attrs_str}'
- if isinstance(self, Index):
- return f'index:{self.ticker}'
- if isinstance(self, Commodity):
- return f'commodity:{self.alpha_vantage_id}'
- if isinstance(self, UnknownAsset):
- return 'unknown:XXX'
-
- msg = f'Unsupported asset type: {type(self).__name__}'
- raise TypeError(msg)
-
- @staticmethod
- def from_string_id(asset_id: str) -> 'Asset':
- """Parses an `Asset` using the fin_defs asset format."""
- m = re.match(r'^(?:(\w+):)?([^{]+)(?:\{(.*)\})?$', asset_id)
- if m is None:
- msg = f'Unsupported asset format: {asset_id}'
- raise ValueError(msg)
-
- prefix = m.group(1)
- ticker = m.group(2)
- attr_data = m.group(3)
-
- if known_symbol := WELL_KNOWN_SYMBOLS.get(ticker, None):
- return known_symbol
-
- attrs: dict[str, str | int] = parse_id_attr_data(attr_data)
-
- if prefix == 'stock':
- if m := re.match(r'^(\w+)(?:\.(\w+))?$', ticker):
- exchange = EXCHANGES_BY_IDS[m.group(2) or 'NYSE'] # TODO?
- return Stock(m.group(1), exchange, **attrs)
- if prefix in {'currency', 'fiat'}:
- return FiatCurrency(ticker, **attrs)
- if prefix == 'index':
- return Index(ticker, **attrs)
- if prefix == 'crypto':
- return CryptoCurrency(ticker, **attrs)
- if prefix == 'commodity':
- return Commodity(ticker)
- if prefix == 'unknown':
- return UnknownAsset()
-
- msg = f'Unsupported asset format: {asset_id}'
- raise ValueError(msg)
-
- def __str__(self):
- return self.to_string_id()
-
-
-@dataclasses.dataclass(frozen=True)
-class UnknownAsset(Asset):
- """An asset that does not exist."""
-
- @abc.abstractmethod
- def raw_short_name(self) -> str:
- return '???'
-
-
-@dataclasses.dataclass(frozen=True)
-class Currency(Asset):
- """Either a Fiat or a Crypto Currency."""
-
-
-@dataclasses.dataclass(frozen=True, eq=True, order=True)
-class FiatCurrency(Currency):
- """Fiat Currency."""
-
- iso_code: str
-
- def __post_init__(self) -> None:
- if not re.match(RE_TICKER_FORMAT, self.iso_code):
- msg = f'iso_code was not in correct format: {self.iso_code}'
- raise ValueError(msg)
-
- def raw_short_name(self) -> str:
- return self.iso_code
-
- @staticmethod
- def from_currency_symbol(symbol: str) -> 'Currency | None':
- for currency, currency_symbol in CURRENCY_SYMBOLS.items():
- if currency_symbol == symbol:
- return currency
- return None
-
- def to_currency_symbol(self) -> str:
- return CURRENCY_SYMBOLS[self]
-
-
-FiatCurrency.DKK = FiatCurrency('DKK')
-FiatCurrency.NOK = FiatCurrency('NOK')
-FiatCurrency.USD = FiatCurrency('USD')
-FiatCurrency.EUR = FiatCurrency('EUR')
-FiatCurrency.GBP = FiatCurrency('GBP')
-FiatCurrency.JPY = FiatCurrency('JPY')
-
-
-@dataclasses.dataclass(frozen=True, eq=True)
-class CryptoCurrency(Currency):
- """Crypto Currency."""
-
- ccxt_symbol: str
- coingecko_id: str | None = None
-
- def raw_short_name(self) -> str:
- return self.ccxt_symbol
-
-
-@dataclasses.dataclass(frozen=True, eq=True)
-class Stock(Asset):
- ticker: str
- exchange: StockExchange
- nordnet_id: int | None = None
-
- def __post_init__(self) -> None:
- if not re.match(RE_TICKER_FORMAT_FLEXIBLE, self.ticker):
- msg = f'ticker was not in correct format: {self.ticker}'
- raise ValueError(msg)
-
- @staticmethod
- def from_polygon_id(polygon_id: str, exchange: StockExchange = NYSE) -> 'Stock':
- return Stock(polygon_id, exchange)
-
- def raw_short_name(self) -> str:
- return self.ticker
-
-
-@dataclasses.dataclass(frozen=True, eq=True)
-class Index(Asset):
- ticker: str
-
- def __post_init__(self) -> None:
- if not re.match(RE_TICKER_FORMAT_FLEXIBLE, self.ticker):
- msg = f'ticker was not in correct format: {self.ticker}'
- raise ValueError(msg)
-
- def raw_short_name(self) -> str:
- return self.ticker
-
-
-@dataclasses.dataclass(frozen=True, eq=True)
-class Commodity(Asset):
- alpha_vantage_id: str
-
- def __post_init__(self) -> None:
- if not re.match(RE_TICKER_FORMAT, self.alpha_vantage_id):
- msg = f'alpha_vantage_id was not in correct format: {self.alpha_vantage_id}'
- raise ValueError(msg)
-
- def raw_short_name(self) -> str:
- return self.alpha_vantage_id
-
-
-Commodity.CRUDE_OIL_WTI = Commodity('WTI')
-Commodity.CRUDE_OIL_BRENT = Commodity('BRENT')
-Commodity.NATURAL_GAS = Commodity('NATURAL_GAS')
-Commodity.COPPER = Commodity('COPPER')
-Commodity.WHEAT = Commodity('WHEAT')
-Commodity.CORN = Commodity('CORN')
-Commodity.COTTON = Commodity('COTTON')
-Commodity.SUGAR = Commodity('SUGAR')
-Commodity.COFFEE = Commodity('COFFEE')
-
-# Well known
-
-DKK = FiatCurrency.DKK
-USD = FiatCurrency.USD
-EUR = FiatCurrency.EUR
-
-BTC = CryptoCurrency('BTC', coingecko_id='bitcoin')
-MPC = CryptoCurrency('MPC', coingecko_id='partisia-blockchain')
-SPX = Index('SPX')
-NDX = Index('NDX')
-USDT = CryptoCurrency('USDT', coingecko_id='tether')
-
-COMMON_NAMES: dict[Asset, list[str]] = {
- FiatCurrency('USD'): ['U.S. Dollar'],
- FiatCurrency('DKK'): ['Danske Kroner', 'Danish Crowns'],
- FiatCurrency('EUR'): ['Euro'],
- FiatCurrency('JPY'): ['Japanese Yen'],
- FiatCurrency('CAD'): ['Canadian Dollar'],
- FiatCurrency('GBP'): ['British Pound'],
- FiatCurrency('CNY'): ['Renminbi'],
- FiatCurrency('PLN'): ['Polish Zloty'],
- FiatCurrency('ILS'): ['Israeli new Shekel'],
- FiatCurrency('CHF'): ['Swiss Franc'],
- FiatCurrency('ZAR'): ['South African Rand'],
- FiatCurrency('CZK'): ['Czech Koruna'],
- SPX: ['Standard and Poor 500', 'S&P 500'],
- NDX: ['Nasdaq 100'],
- BTC: ['Bitcoin BTC'],
- MPC: ['Partisia Blockchain MPC Token'],
- CryptoCurrency('ETH', coingecko_id='ethereum'): ['Ethereum Ether'],
- CryptoCurrency('DOGE', coingecko_id='dogecoin'): ['Dogecoin'],
- CryptoCurrency('SOL', coingecko_id='solana'): ['Solana SOL'],
- CryptoCurrency('ADA', coingecko_id='cardano'): ['Cardano ADA'],
- CryptoCurrency('BNB', coingecko_id='bnb'): ['Binance BNB'],
- CryptoCurrency('MATIC', coingecko_id='polygon'): ['Polygon MATIC'],
- # Stable-coins
- CryptoCurrency('DAI', coingecko_id='dai'): ['DAI'],
- CryptoCurrency('USDC', coingecko_id='usdc'): ['USD Coin'],
- USDT: ['Tether USDT'],
-}
-
-CURRENCY_CODES = {
- sym: FiatCurrency(sym)
- for sym in [
- 'USD',
- 'DKK',
- 'EUR',
- 'JPY',
- 'CAD',
- 'GBP',
- 'CNY',
- 'PLN',
- 'ILS',
- 'CHF',
- 'ZAR',
- 'CZK',
- ]
-}
-
-
-WELL_KNOWN_SYMBOLS = (
- CURRENCY_CODES
- | {'XXBT': BTC}
- | {k.raw_short_name(): k for k in COMMON_NAMES}
- | {'SPX500': SPX, 'SP500': SPX, 'Nasdaq 100': NDX}
-)
-
-CURRENCY_SYMBOLS: dict[Currency, str] = {
- USD: '$',
- EUR: '€',
- FiatCurrency.GBP: '£',
- BTC: '₿',
- FiatCurrency.JPY: '¥',
-}
-
-ASSET_PREFIX: dict[Asset, str] = CURRENCY_SYMBOLS # TODO: Remove at some point.
-
-EXCHANGES = [
- NYSE,
- StockExchange(name='Nasdaq', mic='XNAS', bloomberg_id='US', crunchbase_id='NASDAQ'),
- StockExchange(
- name='Hong Kong Exchanges And Clearing',
- mic='XHKG',
- bloomberg_id='HK',
- crunchbase_id='SEHK',
- ),
- StockExchange(
- name='Moscow Exchange',
- mic='MISC',
- bloomberg_id='RX',
- yahoo_id='ME',
- eod_id='MCX',
- crunchbase_id='MOEX',
- ),
- StockExchange(name='Shanghai Stock Exchange', mic='XSHG'),
- StockExchange(name='Euronext Amsterdam', mic='XAMS'),
- StockExchange(name='Euronext Brussels', mic='XBRU'),
- StockExchange(name='Euronext Dublin', mic='XMSM'),
- StockExchange(name='Euronext Lisbon', mic='XLIS'),
- StockExchange(name='Euronext Milan', mic='XMIL'),
- StockExchange(name='Euronext Oslo', mic='XOSL'),
- StockExchange(name='Euronext Paris', mic='XPAR'),
- StockExchange(name='Japan Exchange Group', mic='XJPX'),
- StockExchange(name='Shenzhen Stock Exchange', mic='XSHE'),
- StockExchange(name='Indian National Stock Exchange', mic='XNSE'),
- StockExchange(
- name='Bombay Stock Exchange',
- mic='XBOM',
- bloomberg_id='IB',
- crunchbase_id='BOM',
- yahoo_id='BO',
- ),
- StockExchange(name='Toronto Stock Exchange', mic='XTSE'),
- StockExchange(name='London Stock Exchange', mic='XLON'),
- StockExchange(name='Saudi Stock Exchange', mic='XSAU'),
- StockExchange(name='German Stock Exchange', mic='XFRA'),
- StockExchange(name='SIX Swiss Exchange', mic='XSWX'),
- StockExchange(name='Nasdaq Copenhagen Stock Exchange', mic='XCSE'),
- StockExchange(name='Nasdaq Stockholm Stock Exchange', mic='XSTO'),
- StockExchange(name='Nasdaq Helsinki Stock Exchange', mic='XHEL'),
- StockExchange(name='Nasdaq Tallinn Stock Exchange', mic='XTAL'),
- StockExchange(name='Nasdaq Riga Stock Exchange', mic='XRIS'),
- StockExchange(name='Nasdaq Vilnius Stock Exchange', mic='XLIT'),
- StockExchange(name='Nasdaq Iceland Stock Exchange', mic='XICE'),
- StockExchange(name='Korea Exchange', mic='XKOS'),
- StockExchange(name='Taiwan Stock Exchange', mic='XTAI'),
- StockExchange(name='Australian Securities Exchange', mic='XASX'),
- StockExchange(name='Johannesburg Stock Exchange', mic='XJSE'),
- StockExchange(name='Tehran Stock Exchange', mic='XTEH'),
- StockExchange(name='Over The Counter (Unlisted)', mic='OTC'), # TODO: Unofficial
- StockExchange(name='NYSE Arca', mic='ARCX'),
- StockExchange(name='BATS Global Markets', mic='BATS'),
- StockExchange(name='NYSE American', mic='XASE'),
- StockExchange(name='Xetra (Frankfurt Stock Exchange)', mic='XETR'),
-]
-
-EXCHANGES_BY_IDS = {}
-if True:
-
- def add_by_id(exchange: StockExchange, key: str):
- exchange_id = getattr(exchange, key)
- if exchange_id is not None:
- EXCHANGES_BY_IDS[exchange_id] = exchange
-
- for exchange in EXCHANGES:
- add_by_id(exchange, 'mic')
- add_by_id(exchange, 'bloomberg_id')
- add_by_id(exchange, 'crunchbase_id')
- add_by_id(exchange, 'yahoo_id')
- add_by_id(exchange, 'eod_id')
- del exchange
- del add_by_id
-
-
-@dataclasses.dataclass(frozen=True, slots=True)
-class AssetInformation:
- """Extensive information about a given asset."""
-
- asset: Asset
- full_name: str
- description: str | None = None
- asset_base: Asset | None = None
- sec_cik: str | None = None
- # TODO: FIGI types? (https://www.openfigi.com/)
-
-
-THREE_DECIMALS_UNDER_THIS_AMOUNT = 0.10
-
-
-def assert_same_asset(do: str, a_asset: 'AssetAmount', b_asset: 'AssetAmount'):
- if a_asset.asset != b_asset.asset:
- msg = f'Attempting to {do} {a_asset} and {b_asset}, but these must have the same asset'
- raise ValueError(msg)
-
-
-@dataclasses.dataclass(frozen=True, eq=True, slots=True)
-class AssetAmount:
- """Decimal with associated asset unit.
-
- Fields:
- - `asset`: Asset unit of amount.
- - `amount`: Amount of given asset.
- """
-
- asset: Asset
- amount: Decimal
-
- def __post_init__(self):
- assert isinstance(self.asset, Asset), 'Incorrect value for self.asset'
- assert isinstance(self.amount, Decimal), 'Incorrect value for self.amount'
-
- def __str__(self) -> str:
- return self.human_readable_str()
-
- def human_readable_str(self) -> str:
- abs_amount = abs(self.amount)
-
- # Determine specificity
- specificity = '3'
- if abs_amount != abs_amount:
- specificity = '0'
- elif abs_amount >= THREE_DECIMALS_UNDER_THIS_AMOUNT:
- specificity = '2'
-
- prefix = ASSET_PREFIX.get(self.asset, '')
- return ('{sign}{prefix}{amount:.' + specificity + 'f} {name}').format(
- sign='-' if (self.amount == self.amount and self.amount < 0) else '',
- prefix=prefix,
- amount=abs_amount,
- name=self.asset.raw_short_name(),
- )
-
- def __mul__(self, other: Decimal) -> 'AssetAmount':
- if not isinstance(other, Decimal):
- msg = f'other must be Decimal, but was {type(other).__name__}'
- raise TypeError(msg)
- return AssetAmount(self.asset, self.amount * other)
-
- def __add__(self, other: 'AssetAmount') -> 'AssetAmount':
- if not isinstance(other, AssetAmount):
- msg = f'other must be AssetAmount, but was {type(other).__name__}'
- raise TypeError(msg)
- if self.is_zero():
- return other
- if other.is_zero():
- return self
- assert_same_asset('add', self, other)
-
- return AssetAmount(self.asset, self.amount + other.amount)
-
- def __sub__(self, other: 'AssetAmount') -> 'AssetAmount':
- # TODO: Implement using add?
- if not isinstance(other, AssetAmount):
- msg = f'other must be AssetAmount, but was {type(other).__name__}'
- raise TypeError(msg)
- if self.is_zero():
- return -other
- if other.is_zero():
- return self
- assert_same_asset('subtract', self, other)
- return AssetAmount(self.asset, self.amount - other.amount)
-
- def __truediv__(self, other: 'Decimal | AssetAmount') -> 'Decimal | AssetAmount':
- if isinstance(other, AssetAmount):
- assert_same_asset('divide', self, other)
- return self.amount / other.amount
- return AssetAmount(self.asset, self.amount / other)
-
- def __neg__(self) -> 'AssetAmount':
- return AssetAmount(self.asset, -self.amount)
-
- def cmp(self, other) -> Decimal:
- if not isinstance(other, AssetAmount):
- msg = f'other must be AssetAmount, but was {type(other).__name__}'
- raise TypeError(msg)
- if self.is_zero() or other.is_zero():
- return self.amount - other.amount
- assert_same_asset('compare', self, other)
- return self.amount - other.amount
-
- def is_zero(self) -> bool:
- return self.amount == 0
-
- def __lt__(self, other: 'AssetAmount') -> bool:
- return self.cmp(other) < 0
-
- def __le__(self, other: 'AssetAmount') -> bool:
- return self.cmp(other) <= 0
-
- def __gt__(self, other: 'AssetAmount') -> bool:
- return self.cmp(other) > 0
-
- def __ge__(self, other: 'AssetAmount') -> bool:
- return self.cmp(other) >= 0
-
-
-AssetAmount.ZERO = AssetAmount(UnknownAsset(), Decimal(0))
-
-
-AssetPair = tuple[Asset, Asset]
-
-
-@dataclasses.dataclass(frozen=True, eq=True)
-class ExchangeRateSample:
- """Single exchange rate sample with a timestamp and various stats."""
-
- timestamp: datetime.date | datetime.datetime
- asset_pair: AssetPair
- _average: Decimal | None = None
- last: Decimal | None = None
- open: Decimal | None = None
- close: Decimal | None = None
- high: Decimal | None = None
- low: Decimal | None = None
- volume: Decimal | None = None
- market_cap: Decimal | None = None
-
- @property
- def average(self) -> Decimal:
- if self._average:
- return self._average
- return (self.high + self.low) / 2
-
- def replace_timestamp(
- self,
- timestamp: datetime.date | datetime.datetime,
- ) -> 'ExchangeRateSample':
- return dataclasses.replace(self, timestamp=timestamp)
-
- def invert_exchange_rate(self) -> 'ExchangeRateSample':
- one = Decimal('1')
- return dataclasses.replace(
- self,
- asset_pair=(self.asset_pair[1], self.asset_pair[0]),
- _average=one / self._average if self._average else None,
- last=one / self.last if self.last else None,
- open=one / self.open if self.open else None,
- close=one / self.close if self.close else None,
- high=one / self.low if self.low else None,
- low=one / self.high if self.high else None,
- # TODO: Support volume
- # TODO: Support market_cap
- )
diff --git a/fin_defs/data.py b/fin_defs/data.py
new file mode 100644
index 0000000..ac77ce7
--- /dev/null
+++ b/fin_defs/data.py
@@ -0,0 +1,621 @@
+import abc
+import dataclasses
+import datetime
+import re
+from collections.abc import Mapping
+from decimal import Decimal
+
+def parse_id_attr_key_value_pair(attr_datum: str) -> tuple[str, str | int] | None:
+ attr_datum = attr_datum.strip()
+ if attr_datum == '':
+ return None
+ (attr_key, attr_value) = attr_datum.split('=')
+ if re.match(r'^\d+$', attr_value):
+ attr_value = int(attr_value)
+ return attr_key, attr_value
+
+
+def parse_id_attr_data(attr_data: str) -> Mapping[str, str | int]:
+ if attr_data is None:
+ return {}
+ key_value_pairs_str = attr_data.split(',')
+ key_value_pairs = [parse_id_attr_key_value_pair(kvp) for kvp in key_value_pairs_str]
+
+ d = {}
+ for kvp in key_value_pairs:
+ if kvp is None:
+ continue
+ d[kvp[0]] = kvp[1]
+ return d
+
+
+## Ids
+
+RE_TICKER_FORMAT = r'^[A-Z0-9_]+$'
+RE_TICKER_FORMAT_FLEXIBLE = r'^[^:\s](?:[^:]*[^:\s])?$'
+RE_CRYPTO_TICKER_FORMAT = r'^\S+$'
+
+
+@dataclasses.dataclass(frozen=True)
+class StockExchange:
+ """Unified Stock Exchange identifiers.
+
+ Fields:
+ - `name`: Human-readable name.
+ - `mic`: Official MIC identifier.
+ - `bloomberg_id`: Identifier for lookup on Bloomberg.
+ - `crunchbase_id`: Identifier for lookup on Crunchbase.
+ - `yahoo_id`: Identifier for lookup on Yahoo! Finance.
+ - `eod_id`: Identifier for lookup on EOD.
+ """
+
+ name: str
+ mic: str
+ bloomberg_id: str | None = None
+ crunchbase_id: str | None = None
+ yahoo_id: str | None = None
+ eod_id: str | None = None
+
+
+NYSE = StockExchange(
+ name='New York Stock Exchange',
+ mic='XNYS',
+ bloomberg_id='UN',
+ crunchbase_id='NYSE',
+)
+
+
+@dataclasses.dataclass(frozen=True)
+class Asset:
+ """An identifier representing some abstract financial asset.
+
+ Subtypes represent the various asset types in existence, for example fiat
+ currencies, stocks or even crypto currencies.
+ """
+
+ def names(self) -> list[str]:
+ """Returns a list of human readable names for this `Asset`."""
+ return COMMON_NAMES.get(self, [])
+
+ def to_polygon_id(self) -> str:
+ """Formats this asset to the [`polygon.io`](https://polygon.io)
+ identifier format.
+
+ Inverse of `from_polygon_id`.
+ """
+ # TODO: Support currency pairs not involving USD.
+ if isinstance(self, Stock):
+ return f'{self.ticker}'
+ if isinstance(self, Index):
+ return f'I:{self.ticker}'
+ if isinstance(self, FiatCurrency):
+ return f'C:{self.iso_code}USD'
+ if isinstance(self, CryptoCurrency):
+ return f'X:{self.ccxt_symbol}USD'
+
+ # Else
+ msg = f'Unknown type: {type(self).__name__}'
+ raise TypeError(msg)
+
+ @staticmethod
+ def from_polygon_id(polygon_id: str, exchange: StockExchange = NYSE) -> 'Asset':
+ """Parses an asset from the [`polygon.io`](https://polygon.io)
+ identifier format.
+
+ Inverse of `to_polygon_id`.
+ """
+ if polygon_id.startswith('I:'):
+ return Index(polygon_id.removeprefix('I:'))
+ if polygon_id.startswith('X:'):
+ return CryptoCurrency(
+ polygon_id.removeprefix('X:').removesuffix('USD'),
+ None,
+ )
+ if polygon_id.startswith('C:'):
+ return FiatCurrency(polygon_id.removeprefix('C:').removesuffix('USD'))
+ return Stock.from_polygon_id(polygon_id, exchange)
+
+ @abc.abstractmethod
+ def raw_short_name(self) -> str:
+ """Short name or ticker.
+
+ Does not nessisarily uniquely identify the asset, but may be commonly used within the industry.
+ """
+
+ def to_string_id(self) -> str:
+ """Formats the asset id using the fin_defs asset format."""
+ if isinstance(self, Stock):
+ attrs_str = f'{{nordnet_id={self.nordnet_id}}}' if self.nordnet_id else ''
+ return f'stock:{self.ticker}.{self.exchange.mic}{attrs_str}'
+ if isinstance(self, FiatCurrency):
+ return f'fiat:{self.iso_code}'
+ if isinstance(self, CryptoCurrency):
+ attrs_str = (
+ f'{{coingecko_id={self.coingecko_id}}}' if self.coingecko_id else ''
+ )
+ return f'crypto:{self.ccxt_symbol}{attrs_str}'
+ if isinstance(self, Index):
+ return f'index:{self.ticker}'
+ if isinstance(self, Commodity):
+ return f'commodity:{self.alpha_vantage_id}'
+ if isinstance(self, UnknownAsset):
+ return 'unknown:XXX'
+
+ msg = f'Unsupported asset type: {type(self).__name__}'
+ raise TypeError(msg)
+
+ @staticmethod
+ def from_string_id(asset_id: str) -> 'Asset':
+ """Parses an `Asset` using the fin_defs asset format."""
+ m = re.match(r'^(?:(\w+):)?([^{]+)(?:\{(.*)\})?$', asset_id)
+ if m is None:
+ msg = f'Unsupported asset format: {asset_id}'
+ raise ValueError(msg)
+
+ prefix = m.group(1)
+ ticker = m.group(2)
+ attr_data = m.group(3)
+
+ if known_symbol := WELL_KNOWN_SYMBOLS.get(ticker, None):
+ return known_symbol
+
+ attrs: dict[str, str | int] = parse_id_attr_data(attr_data)
+
+ if prefix == 'stock':
+ if m := re.match(r'^(\w+)(?:\.(\w+))?$', ticker):
+ exchange = EXCHANGES_BY_IDS[m.group(2) or 'NYSE'] # TODO?
+ return Stock(m.group(1), exchange, **attrs)
+ if prefix in {'currency', 'fiat'}:
+ return FiatCurrency(ticker, **attrs)
+ if prefix == 'index':
+ return Index(ticker, **attrs)
+ if prefix == 'crypto':
+ return CryptoCurrency(ticker, **attrs)
+ if prefix == 'commodity':
+ return Commodity(ticker)
+ if prefix == 'unknown':
+ return UnknownAsset()
+
+ msg = f'Unsupported asset format: {asset_id}'
+ raise ValueError(msg)
+
+ def __str__(self):
+ return self.to_string_id()
+
+
+@dataclasses.dataclass(frozen=True)
+class UnknownAsset(Asset):
+ """An asset that does not exist."""
+
+ @abc.abstractmethod
+ def raw_short_name(self) -> str:
+ return '???'
+
+
+@dataclasses.dataclass(frozen=True)
+class Currency(Asset):
+ """Either a Fiat or a Crypto Currency."""
+
+
+@dataclasses.dataclass(frozen=True, eq=True, order=True)
+class FiatCurrency(Currency):
+ """Fiat Currency."""
+
+ iso_code: str
+
+ def __post_init__(self) -> None:
+ if not re.match(RE_TICKER_FORMAT, self.iso_code):
+ msg = f'iso_code was not in correct format: {self.iso_code}'
+ raise ValueError(msg)
+
+ def raw_short_name(self) -> str:
+ return self.iso_code
+
+ @staticmethod
+ def from_currency_symbol(symbol: str) -> 'Currency | None':
+ for currency, currency_symbol in CURRENCY_SYMBOLS.items():
+ if currency_symbol == symbol:
+ return currency
+ return None
+
+ def to_currency_symbol(self) -> str:
+ return CURRENCY_SYMBOLS[self]
+
+
+FiatCurrency.DKK = FiatCurrency('DKK')
+FiatCurrency.NOK = FiatCurrency('NOK')
+FiatCurrency.USD = FiatCurrency('USD')
+FiatCurrency.EUR = FiatCurrency('EUR')
+FiatCurrency.GBP = FiatCurrency('GBP')
+FiatCurrency.JPY = FiatCurrency('JPY')
+
+
+@dataclasses.dataclass(frozen=True, eq=True)
+class CryptoCurrency(Currency):
+ """Crypto Currency."""
+
+ ccxt_symbol: str
+ coingecko_id: str | None = None
+
+ def raw_short_name(self) -> str:
+ return self.ccxt_symbol
+
+
+@dataclasses.dataclass(frozen=True, eq=True)
+class Stock(Asset):
+ ticker: str
+ exchange: StockExchange
+ nordnet_id: int | None = None
+
+ def __post_init__(self) -> None:
+ if not re.match(RE_TICKER_FORMAT_FLEXIBLE, self.ticker):
+ msg = f'ticker was not in correct format: {self.ticker}'
+ raise ValueError(msg)
+
+ @staticmethod
+ def from_polygon_id(polygon_id: str, exchange: StockExchange = NYSE) -> 'Stock':
+ return Stock(polygon_id, exchange)
+
+ def raw_short_name(self) -> str:
+ return self.ticker
+
+
+@dataclasses.dataclass(frozen=True, eq=True)
+class Index(Asset):
+ ticker: str
+
+ def __post_init__(self) -> None:
+ if not re.match(RE_TICKER_FORMAT_FLEXIBLE, self.ticker):
+ msg = f'ticker was not in correct format: {self.ticker}'
+ raise ValueError(msg)
+
+ def raw_short_name(self) -> str:
+ return self.ticker
+
+
+@dataclasses.dataclass(frozen=True, eq=True)
+class Commodity(Asset):
+ alpha_vantage_id: str
+
+ def __post_init__(self) -> None:
+ if not re.match(RE_TICKER_FORMAT, self.alpha_vantage_id):
+ msg = f'alpha_vantage_id was not in correct format: {self.alpha_vantage_id}'
+ raise ValueError(msg)
+
+ def raw_short_name(self) -> str:
+ return self.alpha_vantage_id
+
+
+Commodity.CRUDE_OIL_WTI = Commodity('WTI')
+Commodity.CRUDE_OIL_BRENT = Commodity('BRENT')
+Commodity.NATURAL_GAS = Commodity('NATURAL_GAS')
+Commodity.COPPER = Commodity('COPPER')
+Commodity.WHEAT = Commodity('WHEAT')
+Commodity.CORN = Commodity('CORN')
+Commodity.COTTON = Commodity('COTTON')
+Commodity.SUGAR = Commodity('SUGAR')
+Commodity.COFFEE = Commodity('COFFEE')
+
+# Well known
+
+DKK = FiatCurrency.DKK
+USD = FiatCurrency.USD
+EUR = FiatCurrency.EUR
+
+BTC = CryptoCurrency('BTC', coingecko_id='bitcoin')
+MPC = CryptoCurrency('MPC', coingecko_id='partisia-blockchain')
+SPX = Index('SPX')
+NDX = Index('NDX')
+USDT = CryptoCurrency('USDT', coingecko_id='tether')
+
+COMMON_NAMES: dict[Asset, list[str]] = {
+ FiatCurrency('USD'): ['U.S. Dollar'],
+ FiatCurrency('DKK'): ['Danske Kroner', 'Danish Crowns'],
+ FiatCurrency('EUR'): ['Euro'],
+ FiatCurrency('JPY'): ['Japanese Yen'],
+ FiatCurrency('CAD'): ['Canadian Dollar'],
+ FiatCurrency('GBP'): ['British Pound'],
+ FiatCurrency('CNY'): ['Renminbi'],
+ FiatCurrency('PLN'): ['Polish Zloty'],
+ FiatCurrency('ILS'): ['Israeli new Shekel'],
+ FiatCurrency('CHF'): ['Swiss Franc'],
+ FiatCurrency('ZAR'): ['South African Rand'],
+ FiatCurrency('CZK'): ['Czech Koruna'],
+ SPX: ['Standard and Poor 500', 'S&P 500'],
+ NDX: ['Nasdaq 100'],
+ BTC: ['Bitcoin BTC'],
+ MPC: ['Partisia Blockchain MPC Token'],
+ CryptoCurrency('ETH', coingecko_id='ethereum'): ['Ethereum Ether'],
+ CryptoCurrency('DOGE', coingecko_id='dogecoin'): ['Dogecoin'],
+ CryptoCurrency('SOL', coingecko_id='solana'): ['Solana SOL'],
+ CryptoCurrency('ADA', coingecko_id='cardano'): ['Cardano ADA'],
+ CryptoCurrency('BNB', coingecko_id='bnb'): ['Binance BNB'],
+ CryptoCurrency('MATIC', coingecko_id='polygon'): ['Polygon MATIC'],
+ # Stable-coins
+ CryptoCurrency('DAI', coingecko_id='dai'): ['DAI'],
+ CryptoCurrency('USDC', coingecko_id='usdc'): ['USD Coin'],
+ USDT: ['Tether USDT'],
+}
+
+CURRENCY_CODES = {
+ sym: FiatCurrency(sym)
+ for sym in [
+ 'USD',
+ 'DKK',
+ 'EUR',
+ 'JPY',
+ 'CAD',
+ 'GBP',
+ 'CNY',
+ 'PLN',
+ 'ILS',
+ 'CHF',
+ 'ZAR',
+ 'CZK',
+ ]
+}
+
+
+WELL_KNOWN_SYMBOLS = (
+ CURRENCY_CODES
+ | {'XXBT': BTC}
+ | {k.raw_short_name(): k for k in COMMON_NAMES}
+ | {'SPX500': SPX, 'SP500': SPX, 'Nasdaq 100': NDX}
+)
+
+CURRENCY_SYMBOLS: dict[Currency, str] = {
+ USD: '$',
+ EUR: '€',
+ FiatCurrency.GBP: '£',
+ BTC: '₿',
+ FiatCurrency.JPY: '¥',
+}
+
+ASSET_PREFIX: dict[Asset, str] = CURRENCY_SYMBOLS # TODO: Remove at some point.
+
+EXCHANGES = [
+ NYSE,
+ StockExchange(name='Nasdaq', mic='XNAS', bloomberg_id='US', crunchbase_id='NASDAQ'),
+ StockExchange(
+ name='Hong Kong Exchanges And Clearing',
+ mic='XHKG',
+ bloomberg_id='HK',
+ crunchbase_id='SEHK',
+ ),
+ StockExchange(
+ name='Moscow Exchange',
+ mic='MISC',
+ bloomberg_id='RX',
+ yahoo_id='ME',
+ eod_id='MCX',
+ crunchbase_id='MOEX',
+ ),
+ StockExchange(name='Shanghai Stock Exchange', mic='XSHG'),
+ StockExchange(name='Euronext Amsterdam', mic='XAMS'),
+ StockExchange(name='Euronext Brussels', mic='XBRU'),
+ StockExchange(name='Euronext Dublin', mic='XMSM'),
+ StockExchange(name='Euronext Lisbon', mic='XLIS'),
+ StockExchange(name='Euronext Milan', mic='XMIL'),
+ StockExchange(name='Euronext Oslo', mic='XOSL'),
+ StockExchange(name='Euronext Paris', mic='XPAR'),
+ StockExchange(name='Japan Exchange Group', mic='XJPX'),
+ StockExchange(name='Shenzhen Stock Exchange', mic='XSHE'),
+ StockExchange(name='Indian National Stock Exchange', mic='XNSE'),
+ StockExchange(
+ name='Bombay Stock Exchange',
+ mic='XBOM',
+ bloomberg_id='IB',
+ crunchbase_id='BOM',
+ yahoo_id='BO',
+ ),
+ StockExchange(name='Toronto Stock Exchange', mic='XTSE'),
+ StockExchange(name='London Stock Exchange', mic='XLON'),
+ StockExchange(name='Saudi Stock Exchange', mic='XSAU'),
+ StockExchange(name='German Stock Exchange', mic='XFRA'),
+ StockExchange(name='SIX Swiss Exchange', mic='XSWX'),
+ StockExchange(name='Nasdaq Copenhagen Stock Exchange', mic='XCSE'),
+ StockExchange(name='Nasdaq Stockholm Stock Exchange', mic='XSTO'),
+ StockExchange(name='Nasdaq Helsinki Stock Exchange', mic='XHEL'),
+ StockExchange(name='Nasdaq Tallinn Stock Exchange', mic='XTAL'),
+ StockExchange(name='Nasdaq Riga Stock Exchange', mic='XRIS'),
+ StockExchange(name='Nasdaq Vilnius Stock Exchange', mic='XLIT'),
+ StockExchange(name='Nasdaq Iceland Stock Exchange', mic='XICE'),
+ StockExchange(name='Korea Exchange', mic='XKOS'),
+ StockExchange(name='Taiwan Stock Exchange', mic='XTAI'),
+ StockExchange(name='Australian Securities Exchange', mic='XASX'),
+ StockExchange(name='Johannesburg Stock Exchange', mic='XJSE'),
+ StockExchange(name='Tehran Stock Exchange', mic='XTEH'),
+ StockExchange(name='Over The Counter (Unlisted)', mic='OTC'), # TODO: Unofficial
+ StockExchange(name='NYSE Arca', mic='ARCX'),
+ StockExchange(name='BATS Global Markets', mic='BATS'),
+ StockExchange(name='NYSE American', mic='XASE'),
+ StockExchange(name='Xetra (Frankfurt Stock Exchange)', mic='XETR'),
+]
+
+EXCHANGES_BY_IDS = {}
+if True:
+
+ def add_by_id(exchange: StockExchange, key: str):
+ exchange_id = getattr(exchange, key)
+ if exchange_id is not None:
+ EXCHANGES_BY_IDS[exchange_id] = exchange
+
+ for exchange in EXCHANGES:
+ add_by_id(exchange, 'mic')
+ add_by_id(exchange, 'bloomberg_id')
+ add_by_id(exchange, 'crunchbase_id')
+ add_by_id(exchange, 'yahoo_id')
+ add_by_id(exchange, 'eod_id')
+ del exchange
+ del add_by_id
+
+
+@dataclasses.dataclass(frozen=True, slots=True)
+class AssetInformation:
+ """Extensive information about a given asset."""
+
+ asset: Asset
+ full_name: str
+ description: str | None = None
+ asset_base: Asset | None = None
+ sec_cik: str | None = None
+ # TODO: FIGI types? (https://www.openfigi.com/)
+
+
+THREE_DECIMALS_UNDER_THIS_AMOUNT = 0.10
+
+
+def assert_same_asset(do: str, a_asset: 'AssetAmount', b_asset: 'AssetAmount'):
+ if a_asset.asset != b_asset.asset:
+ msg = f'Attempting to {do} {a_asset} and {b_asset}, but these must have the same asset'
+ raise ValueError(msg)
+
+
+@dataclasses.dataclass(frozen=True, eq=True, slots=True)
+class AssetAmount:
+ """Decimal with associated asset unit.
+
+ Fields:
+ - `asset`: Asset unit of amount.
+ - `amount`: Amount of given asset.
+ """
+
+ asset: Asset
+ amount: Decimal
+
+ def __post_init__(self):
+ assert isinstance(self.asset, Asset), 'Incorrect value for self.asset'
+ assert isinstance(self.amount, Decimal), 'Incorrect value for self.amount'
+
+ def __str__(self) -> str:
+ return self.human_readable_str()
+
+ def human_readable_str(self) -> str:
+ abs_amount = abs(self.amount)
+
+ # Determine specificity
+ specificity = '3'
+ if abs_amount != abs_amount:
+ specificity = '0'
+ elif abs_amount >= THREE_DECIMALS_UNDER_THIS_AMOUNT:
+ specificity = '2'
+
+ prefix = ASSET_PREFIX.get(self.asset, '')
+ return ('{sign}{prefix}{amount:.' + specificity + 'f} {name}').format(
+ sign='-' if (self.amount == self.amount and self.amount < 0) else '',
+ prefix=prefix,
+ amount=abs_amount,
+ name=self.asset.raw_short_name(),
+ )
+
+ def __mul__(self, other: Decimal) -> 'AssetAmount':
+ if not isinstance(other, Decimal):
+ msg = f'other must be Decimal, but was {type(other).__name__}'
+ raise TypeError(msg)
+ return AssetAmount(self.asset, self.amount * other)
+
+ def __add__(self, other: 'AssetAmount') -> 'AssetAmount':
+ if not isinstance(other, AssetAmount):
+ msg = f'other must be AssetAmount, but was {type(other).__name__}'
+ raise TypeError(msg)
+ if self.is_zero():
+ return other
+ if other.is_zero():
+ return self
+ assert_same_asset('add', self, other)
+
+ return AssetAmount(self.asset, self.amount + other.amount)
+
+ def __sub__(self, other: 'AssetAmount') -> 'AssetAmount':
+ # TODO: Implement using add?
+ if not isinstance(other, AssetAmount):
+ msg = f'other must be AssetAmount, but was {type(other).__name__}'
+ raise TypeError(msg)
+ if self.is_zero():
+ return -other
+ if other.is_zero():
+ return self
+ assert_same_asset('subtract', self, other)
+ return AssetAmount(self.asset, self.amount - other.amount)
+
+ def __truediv__(self, other: 'Decimal | AssetAmount') -> 'Decimal | AssetAmount':
+ if isinstance(other, AssetAmount):
+ assert_same_asset('divide', self, other)
+ return self.amount / other.amount
+ return AssetAmount(self.asset, self.amount / other)
+
+ def __neg__(self) -> 'AssetAmount':
+ return AssetAmount(self.asset, -self.amount)
+
+ def cmp(self, other) -> Decimal:
+ if not isinstance(other, AssetAmount):
+ msg = f'other must be AssetAmount, but was {type(other).__name__}'
+ raise TypeError(msg)
+ if self.is_zero() or other.is_zero():
+ return self.amount - other.amount
+ assert_same_asset('compare', self, other)
+ return self.amount - other.amount
+
+ def is_zero(self) -> bool:
+ return self.amount == 0
+
+ def __lt__(self, other: 'AssetAmount') -> bool:
+ return self.cmp(other) < 0
+
+ def __le__(self, other: 'AssetAmount') -> bool:
+ return self.cmp(other) <= 0
+
+ def __gt__(self, other: 'AssetAmount') -> bool:
+ return self.cmp(other) > 0
+
+ def __ge__(self, other: 'AssetAmount') -> bool:
+ return self.cmp(other) >= 0
+
+
+AssetAmount.ZERO = AssetAmount(UnknownAsset(), Decimal(0))
+
+
+AssetPair = tuple[Asset, Asset]
+
+
+@dataclasses.dataclass(frozen=True, eq=True)
+class ExchangeRateSample:
+ """Single exchange rate sample with a timestamp and various stats."""
+
+ timestamp: datetime.date | datetime.datetime
+ asset_pair: AssetPair
+ _average: Decimal | None = None
+ last: Decimal | None = None
+ open: Decimal | None = None
+ close: Decimal | None = None
+ high: Decimal | None = None
+ low: Decimal | None = None
+ volume: Decimal | None = None
+ market_cap: Decimal | None = None
+
+ @property
+ def average(self) -> Decimal:
+ if self._average:
+ return self._average
+ return (self.high + self.low) / 2
+
+ def replace_timestamp(
+ self,
+ timestamp: datetime.date | datetime.datetime,
+ ) -> 'ExchangeRateSample':
+ return dataclasses.replace(self, timestamp=timestamp)
+
+ def invert_exchange_rate(self) -> 'ExchangeRateSample':
+ one = Decimal('1')
+ return dataclasses.replace(
+ self,
+ asset_pair=(self.asset_pair[1], self.asset_pair[0]),
+ _average=one / self._average if self._average else None,
+ last=one / self.last if self.last else None,
+ open=one / self.open if self.open else None,
+ close=one / self.close if self.close else None,
+ high=one / self.low if self.low else None,
+ low=one / self.high if self.high else None,
+ # TODO: Support volume
+ # TODO: Support market_cap
+ )
diff --git a/fin_defs/parse_price.py b/fin_defs/parse_price.py
new file mode 100644
index 0000000..d34cb4e
--- /dev/null
+++ b/fin_defs/parse_price.py
@@ -0,0 +1,66 @@
+import logging
+import re
+from decimal import Decimal
+
+from .data import DKK, USD, Asset, AssetAmount, FiatCurrency
+
+logger = logging.getLogger(__name__)
+
+RE_PRICE_RAW = r'\b(?:dkk|sek|usd|nok|eur)?\s*([1-9][\d.]*[\d](?:,\d+)?)\s*(?:,-|:-|.-|;-)?\s*(?:(?:kr|kroner|krone|dkk|sek|usd|eur|nok)\b)?\.?'
+
+RE_PRICE = re.compile(RE_PRICE_RAW, flags=re.IGNORECASE)
+
+def parse_price(text: str, default_currency: Asset) -> AssetAmount | None:
+ """
+ Attempts to parse price from the given text.
+
+ Text does not need to be stripped beforehand.
+ """
+ if isinstance(text, AssetAmount):
+ return text
+ text = str(text)
+
+ if m := re.match(r'^Kr\s*([\d.]+(?:,\d+))?$', text):
+ return AssetAmount(
+ DKK,
+ Decimal(m.group(1).replace('.', '').replace(',', '.')),
+ )
+ if m := re.match(r'^(\d+)\s*DKK$', text):
+ return AssetAmount(DKK, Decimal(m.group(1)))
+
+ if m := re.match(r'^\$\s*([0-9.]+)(\s+USD)?$', text):
+ return AssetAmount(USD, Decimal(m.group(1)))
+
+ if text.lower().strip() == 'free':
+ return AssetAmount(default_currency, Decimal(0.0))
+
+ text = str(text).strip().lower().removesuffix('.')
+ if m := RE_PRICE.fullmatch(text):
+ currency = default_currency
+ price_tag = m.group(1).replace('.', '').replace(',', '.') # TODO
+ if text.endswith('dkk') or text.startswith('dkk'):
+ currency = FiatCurrency('DKK')
+ elif text.endswith('sek') or text.startswith('sek'):
+ currency = FiatCurrency('SEK')
+ elif text.endswith('nok') or text.startswith('nok'):
+ currency = FiatCurrency('NOK')
+ elif text.endswith('usd') or text.startswith('usd'):
+ currency = FiatCurrency('USD')
+
+ return AssetAmount(currency, Decimal(price_tag))
+
+ logger.warning('Unknown price format: %s', text)
+ return None
+
+
+def parse_usd_price(s: str | int) -> AssetAmount:
+ assert s is not None
+ if isinstance(s, str) or isinstance(s, int):
+ text = str(s)
+ else:
+ text = s.text_content()
+ text = text.strip().replace(',', '').removeprefix('$')
+ if text in {'-', ''}:
+ return AssetAmount(USD, Decimal(0)) # TODO
+ dollar_amount = Decimal(text)
+ return AssetAmount(USD, dollar_amount)
diff --git a/test/test_parse_price.py b/test/test_parse_price.py
new file mode 100644
index 0000000..807e45c
--- /dev/null
+++ b/test/test_parse_price.py
@@ -0,0 +1,36 @@
+import pytest
+from decimal import Decimal
+from fin_defs import DKK, USD, AssetAmount, FiatCurrency
+from fin_defs.parse_price import parse_price
+
+def dkk(amount):
+ return AssetAmount(DKK, Decimal(amount))
+
+PRICES_PARSABLE = [
+ ('DKK100', dkk(100)),
+ ('100;-', dkk(100)),
+ ('100 kr', dkk(100)),
+ (' 100 kr ', dkk(100)),
+ ('349.-', dkk(349)),
+ ('3.000 kr.', dkk(3000)),
+ ('25,00 kr.', dkk(25)),
+ ('300,00 kr.', dkk(300)),
+ ('300kr.', dkk(300)),
+ ('300kr', dkk(300)),
+ ('5,00 dkk', dkk(5)),
+ ('9,99 dkk', dkk('9.99')),
+ ('17900 kr', dkk(17900)),
+]
+
+PRICES_UNPARSABLE = [
+ '007',
+]
+
+@pytest.mark.parametrize(('price_string', 'parsed_amount'), PRICES_PARSABLE)
+def test_parse_price(price_string: str, parsed_amount: AssetAmount):
+ result = parse_price(price_string, parsed_amount.asset)
+ assert result == parsed_amount
+
+@pytest.mark.parametrize('price_string', PRICES_UNPARSABLE)
+def test_parse_unparsable(price_string: str):
+ assert parse_price(price_string, USD) is None