""" Currency normalization rules (final master pass, §17). The rule that actually prevents damage is the negative one: **a ratio must never be converted.** A P/E is currency-free — price and EPS share a currency and the units cancel — so "normalise every numeric field to USD" silently multiplies every margin, return, growth rate and multiple by the FX rate. These tests make that impossible to do by accident. The second rule is that an unavailable rate must produce `None`, never the unconverted number wearing the target currency's label. A JPY figure presented as USD is wrong by roughly 150× and looks entirely normal. Pure, dependency-free — genuinely executed. Dual-mode: pytest, or `PYTHONPATH=. python3 tests/test_currency.py`. """ from __future__ import annotations from datetime import date from app.engines.currency import ( CURRENCY_INDEPENDENT_FIELDS, MONETARY_FIELDS, NON_MONETARY_FIELDS, PER_SHARE_MONETARY_FIELDS, SUPPORTED_CURRENCIES, CurrencyMismatchError, FxRate, UnsupportedConversionError, assert_consistent_currency, convert_amount, convert_line_items, convert_metric_value, identity_rate, ) from app.engines.types import LineItems _AS_OF = date(2025, 12, 31) _EURUSD = FxRate("EUR", "USD", 1.10, _AS_OF, source="TEST_FIXTURE_NOT_A_REAL_RATE") def _li(**kw) -> LineItems: base = dict( security_id="T", period_end=_AS_OF, period_type="FY", filing_date=_AS_OF, currency="EUR", revenue=1000.0, net_income=100.0, operating_cash_flow=200.0, capital_expenditure=50.0, total_debt=400.0, shareholders_equity=900.0, cash_and_equivalents=100.0, eps_diluted=1.0, diluted_shares=100.0, shares_outstanding=100.0, price=25.0, market_cap=2500.0, beta=1.2, ) base.update(kw) return LineItems(**base) # --- the negative rule: ratios are never converted --- def test_converting_a_ratio_is_refused_outright(): """Not skipped, not passed through — raised. A caller asking to convert a P/E has a bug that must surface immediately rather than producing a number 1.1x too large.""" for key in ("pe", "roic", "gross_margin", "debt_to_equity", "fcf_yield", "peg", "wacc"): try: convert_metric_value(key, 15.0, "EUR", _EURUSD) except UnsupportedConversionError as exc: assert "cancels" in str(exc) else: raise AssertionError(f"{key} was converted; it is currency-independent") def test_every_growth_and_margin_metric_is_marked_currency_independent(): for key in ("revenue_growth_yoy", "revenue_growth_cagr_3y", "eps_growth_cagr_3y", "fcf_growth_cagr_3y", "net_margin", "fcf_margin", "operating_margin", "margin_of_safety", "expected_return"): assert key in CURRENCY_INDEPENDENT_FIELDS, key def test_no_field_is_classified_as_both_monetary_and_currency_independent(): assert not (set(MONETARY_FIELDS) & CURRENCY_INDEPENDENT_FIELDS) assert not (set(PER_SHARE_MONETARY_FIELDS) & CURRENCY_INDEPENDENT_FIELDS) def test_share_counts_are_not_monetary(): """A share is a share in every currency.""" for key in ("shares_outstanding", "diluted_shares", "beta"): assert key in NON_MONETARY_FIELDS assert key not in MONETARY_FIELDS def test_a_monetary_metric_is_convertible(): """The complement: `fcf` is an absolute amount and MUST convert.""" out = convert_metric_value("fcf", 150.0, "EUR", _EURUSD) assert abs(out.value - 165.0) < 1e-9 # --- an unavailable rate yields None, not the unconverted number --- def test_a_missing_rate_produces_none_not_the_original_amount(): out = convert_amount(1000.0, "JPY", rate=None) assert out.value is None assert out.original_value == 1000.0 assert out.normalized_currency == "JPY" # nothing was claimed assert out.was_converted is False def test_a_rate_for_the_wrong_currency_is_refused(): try: convert_amount(1000.0, "GBP", _EURUSD) except CurrencyMismatchError as exc: assert "EUR" in str(exc) and "GBP" in str(exc) return raise AssertionError("a EUR rate was applied to a GBP amount") def test_a_none_amount_stays_none_and_still_records_the_rate(): out = convert_amount(None, "EUR", _EURUSD) assert out.value is None assert out.fx_rate == 1.10 and out.fx_source == "TEST_FIXTURE_NOT_A_REAL_RATE" # --- provenance is mandatory, not optional --- def test_a_converted_value_carries_rate_date_and_source(): out = convert_amount(1000.0, "EUR", _EURUSD) assert abs(out.value - 1100.0) < 1e-9 assert out.original_value == 1000.0 assert out.original_currency == "EUR" and out.normalized_currency == "USD" assert out.fx_rate == 1.10 assert out.fx_as_of == _AS_OF assert out.fx_source assert out.was_converted is True assert set(out.as_dict()) == { "value", "original_value", "original_currency", "normalized_currency", "fx_rate", "fx_as_of", "fx_source", "was_converted", } def test_a_rate_without_a_source_is_rejected_at_construction(): for bad in ({"source": ""}, {"rate": 0.0}, {"rate": -1.0}): kwargs = {"from_currency": "EUR", "to_currency": "USD", "rate": 1.1, "as_of": _AS_OF, "source": "x"} kwargs.update(bad) try: FxRate(**kwargs) except ValueError: continue raise AssertionError(f"FxRate accepted {bad}") def test_an_identity_rate_must_actually_be_one(): identity_rate("USD", _AS_OF) # fine try: FxRate("USD", "USD", 1.5, _AS_OF, source="x") except ValueError: return raise AssertionError("a same-currency rate of 1.5 was accepted") # --- converting a whole period keeps the arithmetic self-consistent --- def test_converting_a_period_scales_monetary_fields_and_leaves_counts_alone(): out = convert_line_items(_li(), _EURUSD) assert out.currency == "USD" assert abs(out.revenue - 1100.0) < 1e-9 assert abs(out.total_debt - 440.0) < 1e-9 assert abs(out.price - 27.5) < 1e-9 assert abs(out.market_cap - 2750.0) < 1e-9 assert out.diluted_shares == 100.0 # unchanged assert out.shares_outstanding == 100.0 # unchanged assert out.beta == 1.2 # unchanged def test_per_share_identities_survive_a_conversion(): """`fcf_per_share × shares == fcf` and `market_cap == price × shares` must still hold after converting, or the aggregate and per-share fields were scaled with different rates.""" original = _li() converted = convert_line_items(original, _EURUSD) fcf_before = original.operating_cash_flow - original.capital_expenditure fcf_after = converted.operating_cash_flow - converted.capital_expenditure assert abs(fcf_after - fcf_before * 1.10) < 1e-9 assert abs(converted.market_cap - converted.price * converted.diluted_shares) < 1e-9 def test_ratios_computed_before_and_after_conversion_are_identical(): """The whole point of the currency-independence rule, demonstrated end to end.""" original = _li() converted = convert_line_items(original, _EURUSD) for name in ("net_income", "revenue"): assert getattr(converted, name) != getattr(original, name) # they really did convert margin_before = original.net_income / original.revenue margin_after = converted.net_income / converted.revenue assert abs(margin_before - margin_after) < 1e-12 pe_before = original.price / original.eps_diluted pe_after = converted.price / converted.eps_diluted assert abs(pe_before - pe_after) < 1e-12 def test_converting_a_period_with_the_wrong_rate_currency_is_refused(): try: convert_line_items(_li(currency="GBP"), _EURUSD) except CurrencyMismatchError: return raise AssertionError("a EUR rate was applied to a GBP period") # --- snapshot-level consistency (§17's "check consistency in DCF/multiples/Fair Value/...") --- def test_a_snapshot_of_one_currency_passes(): assert assert_consistent_currency([_li(), _li(), _li()]) == "EUR" def test_a_snapshot_mixing_currencies_is_refused(): try: assert_consistent_currency([_li(), _li(currency="USD")]) except CurrencyMismatchError as exc: assert "multiple currencies" in str(exc) return raise AssertionError("a mixed-currency snapshot was accepted") def test_an_empty_snapshot_is_refused_rather_than_defaulting_to_usd(): try: assert_consistent_currency([]) except CurrencyMismatchError: return raise AssertionError("an empty period list returned a currency") def test_the_supported_currency_list_covers_every_currency_the_spec_names(): for code in ("USD", "EUR", "GBP", "CAD", "JPY", "CNY", "HKD", "INR", "AUD", "CHF"): assert code in SUPPORTED_CURRENCIES ALL_TESTS = [v for k, v in sorted(globals().items()) if k.startswith("test_")] if __name__ == "__main__": passed = failed = 0 for t in ALL_TESTS: try: t() print(f"PASS {t.__name__}") passed += 1 except Exception as exc: # noqa: BLE001 print(f"FAIL {t.__name__}: {exc}") failed += 1 print(f"\n{passed}/{passed + failed} passed") raise SystemExit(1 if failed else 0)