""" Currency normalization (final master pass, §17). ## What this module is, and what it is very deliberately not §17 requires that StockLab be able to work across currencies: keep the original currency, the normalized currency, and the FX source / date / rate; **never convert currency-independent ratios**; and check currency consistency across the DCF, the multiples, Fair Value, market cap, FCF, revenue, EBITDA, net debt and share price. This module implements the *rules* — which fields may be converted, which must never be, what a converted value must carry with it, and what must happen when the rate is unknown. It does **not** fetch FX rates, because **no adapter in this repository provides them**: neither `FMPAdapter` nor `EODHDAdapter` exposes an FX endpoint, and there is no network access here to call one. Inventing a rate table would be exactly the fabrication the governing rules forbid, and a wrong FX rate is worse than no conversion at all — it produces a plausible number that is silently off by tens of percent. So: `FxRate` is supplied by the caller. `convert()` refuses without one. Wiring a real FX source is listed as **NOT IMPLEMENTED** in `docs/AUDIT_CURRENCY_IDENTITY.md`. ## The rule that actually prevents damage **A ratio must never be converted.** A P/E is a currency-free number: price and EPS are in the same currency, and the units cancel. Converting the price and forgetting the EPS — or converting the ratio itself — changes the answer by the FX rate for no reason. So do margins, returns, growth rates, yields, and leverage multiples where both sides share a currency. The failure mode is not hypothetical: it is what happens when someone "normalises everything to USD" with a loop over every numeric field. `CURRENCY_INDEPENDENT_FIELDS` below is the guard, and `convert_line_items()` raises rather than silently skipping, because a caller that asks to convert a ratio has a bug that should surface immediately. Pure and dependency-free (stdlib only) so it is genuinely testable here. """ from __future__ import annotations from dataclasses import dataclass, replace from datetime import date from typing import Iterable, Optional from app.engines.types import LineItems #: Currencies §17 names as the minimum set. Membership here says nothing about whether a rate is #: available — it is a validation list, not a claim of coverage. SUPPORTED_CURRENCIES: tuple[str, ...] = ( "USD", "EUR", "GBP", "CAD", "JPY", "CNY", "HKD", "INR", "AUD", "CHF", ) #: Monetary `LineItems` fields: absolute amounts in the reporting currency. These MAY be converted. MONETARY_FIELDS: tuple[str, ...] = ( "revenue", "cogs", "gross_profit", "operating_income", "ebit", "ebitda", "net_income", "tax_expense", "pretax_income", "interest_expense", "adjusted_net_income", "cash_and_equivalents", "short_term_investments", "total_debt", "short_term_debt", "long_term_debt", "lease_liabilities", "total_assets", "current_assets", "current_liabilities", "shareholders_equity", "minority_interest", "preferred_equity", "goodwill", "intangible_assets", "receivables", "inventory", "operating_cash_flow", "capital_expenditure", "dividends_paid", "buybacks", "stock_issuance", "depreciation_and_amortization", "stock_based_compensation", ) #: Per-share monetary fields. Also convertible — a per-share amount is still money — but listed #: separately because they must be converted with the SAME rate as the aggregate fields, or #: `fcf_per_share × shares == fcf` stops holding. PER_SHARE_MONETARY_FIELDS: tuple[str, ...] = ("eps_diluted", "price") #: Counts and unitless quantities. Never converted: a share is a share in every currency. NON_MONETARY_FIELDS: tuple[str, ...] = ("shares_outstanding", "diluted_shares", "beta") #: Metric keys that are ratios or rates. Converting any of these is always a bug — the currency #: cancels between numerator and denominator. `convert_metric_value()` raises on them. CURRENCY_INDEPENDENT_FIELDS: frozenset[str] = frozenset({ "gross_margin", "operating_margin", "net_margin", "fcf_margin", "roic", "roic_minus_wacc", "roe", "earnings_yield", "fcf_yield", "dividend_yield", "buyback_yield", "shareholder_yield", "debt_to_ebitda", "net_debt_to_ebitda", "debt_to_equity", "interest_coverage", "current_ratio", "pe", "forward_pe", "peg", "ev_to_ebitda", "p_fcf", "ev_to_fcf", "capex_to_revenue", "fcf_payout_ratio", "revenue_growth_yoy", "revenue_growth_cagr_3y", "revenue_growth_cagr_5y", "revenue_growth_cagr_10y", "revenue_cagr_5y", "eps_growth_cagr_3y", "fcf_growth_cagr_3y", "share_count_growth_cagr_3y", "wacc", "margin_of_safety", "expected_return", }) class CurrencyMismatchError(ValueError): """Raised when two amounts in different currencies would be combined.""" class UnsupportedConversionError(ValueError): """Raised when a currency-independent value is offered for conversion.""" @dataclass(frozen=True) class FxRate: """One rate, with everything needed to audit it later. `rate` converts ONE unit of `from_currency` into `to_currency`: an amount in `from_currency` is multiplied by it. `as_of` and `source` are not optional — a converted number whose rate date and origin are unknown cannot be reproduced or checked, which makes it unusable as evidence. """ from_currency: str to_currency: str rate: float as_of: date source: str def __post_init__(self) -> None: if self.rate <= 0: raise ValueError(f"FX rate must be positive, got {self.rate}") if not self.source: raise ValueError("FX rate must name its source") if self.from_currency == self.to_currency and self.rate != 1.0: raise ValueError("an identity conversion must have rate 1.0") @dataclass(frozen=True) class ConvertedValue: """A converted amount that carries its own provenance.""" value: Optional[float] original_value: Optional[float] original_currency: str normalized_currency: str fx_rate: Optional[float] fx_as_of: Optional[date] fx_source: Optional[str] @property def was_converted(self) -> bool: return self.original_currency != self.normalized_currency def as_dict(self) -> dict: return { "value": self.value, "original_value": self.original_value, "original_currency": self.original_currency, "normalized_currency": self.normalized_currency, "fx_rate": self.fx_rate, "fx_as_of": self.fx_as_of.isoformat() if self.fx_as_of else None, "fx_source": self.fx_source, "was_converted": self.was_converted, } def identity_rate(currency: str, as_of: date) -> FxRate: """The no-op rate. Explicit rather than implicit, so a same-currency value still records that a conversion step ran and found nothing to do.""" return FxRate(currency, currency, 1.0, as_of, source="identity") def convert_amount( amount: Optional[float], from_currency: str, rate: Optional[FxRate], ) -> ConvertedValue: """Convert one monetary amount. `rate` of None means the rate is unavailable. An unavailable rate produces `value=None` — **not** the unconverted number. Returning the original amount labelled as the target currency is the silent-fallback failure this project forbids: a JPY figure presented as USD is wrong by a factor of ~150 and looks entirely normal. """ if rate is None: return ConvertedValue(None, amount, from_currency, from_currency, None, None, None) if rate.from_currency != from_currency: raise CurrencyMismatchError( f"rate converts {rate.from_currency}, but the amount is in {from_currency}" ) converted = None if amount is None else amount * rate.rate return ConvertedValue( value=converted, original_value=amount, original_currency=from_currency, normalized_currency=rate.to_currency, fx_rate=rate.rate, fx_as_of=rate.as_of, fx_source=rate.source, ) def convert_metric_value(metric_key: str, value: Optional[float], from_currency: str, rate: Optional[FxRate]) -> ConvertedValue: """Convert a metric's value, refusing outright for currency-independent metrics.""" if metric_key in CURRENCY_INDEPENDENT_FIELDS: raise UnsupportedConversionError( f"{metric_key!r} is a ratio or rate: its currency cancels between numerator and " f"denominator. Converting it would change the answer by the FX rate for no reason." ) return convert_amount(value, from_currency, rate) def convert_line_items(li: LineItems, rate: FxRate) -> LineItems: """Convert every monetary field of one period, leaving counts and ratios untouched. The SAME rate is applied to aggregate and per-share monetary fields, so identities such as `fcf_per_share × diluted_shares == fcf` survive the conversion — a test asserts exactly that. `market_cap` is converted too: it is `price × shares`, and converting the price without the market cap would leave the two disagreeing. """ if li.currency != rate.from_currency: raise CurrencyMismatchError( f"period is in {li.currency}, rate converts {rate.from_currency}" ) updates: dict = {"currency": rate.to_currency} for name in MONETARY_FIELDS + PER_SHARE_MONETARY_FIELDS + ("market_cap",): current = getattr(li, name, None) if current is not None: updates[name] = current * rate.rate return replace(li, **updates) def assert_consistent_currency(periods: Iterable[LineItems]) -> str: """Return the single currency shared by every period, or raise. §17 requires currency consistency across the DCF, multiples, Fair Value, market cap, FCF, revenue, EBITDA, net debt and share price. All of those are computed from a `FinancialSnapshot`, so checking the snapshot's periods share one currency is where that requirement is actually enforceable — one check covering every downstream consumer. """ currencies = {p.currency for p in periods if p is not None} if not currencies: raise CurrencyMismatchError("no periods supplied") if len(currencies) > 1: raise CurrencyMismatchError( f"periods span multiple currencies {sorted(currencies)}; every valuation output would " f"be a meaningless mix. Normalize before building the snapshot." ) return currencies.pop()