""" Known-case tests for the TTM aggregation engine (StockLab final engineering pass, Part A9). Pure, dependency-free — genuinely executed here, not a claim. Dual-mode: pytest, or `PYTHONPATH=. python3 tests/test_ttm.py`. The hand-computed cases below are deliberately arithmetic that can be checked by eye: four quarters of revenue 100/110/120/130 must produce exactly 460, and a balance-sheet field must produce the LATEST quarter's value, never 4x it and never the mean. """ from __future__ import annotations from datetime import date from app.engines.ttm import ( FLOW_FIELDS, STOCK_FIELDS, build_ttm_line_items, validate_field_classification, ) from app.engines.types import LineItems def _q(period_end: str, filing: str | None = None, ptype: str = "Q1", **kw) -> LineItems: return LineItems( security_id="SEC1", period_end=date.fromisoformat(period_end), period_type=ptype, filing_date=date.fromisoformat(filing) if filing else None, currency="USD", **kw, ) def _four_quarters(**per_quarter) -> list[LineItems]: """Four consecutive quarters ending 2025-12-31, oldest first in the returned list order is NOT assumed anywhere — build_ttm_line_items sorts for itself.""" ends = ["2025-03-31", "2025-06-30", "2025-09-30", "2025-12-31"] filings = ["2025-05-01", "2025-08-01", "2025-11-01", "2026-02-01"] types = ["Q1", "Q2", "Q3", "Q4"] out = [] for i, (e, f, t) in enumerate(zip(ends, filings, types)): kw = {k: (v[i] if isinstance(v, (list, tuple)) else v) for k, v in per_quarter.items()} out.append(_q(e, f, t, **kw)) return out # --- the core arithmetic --- def test_flow_fields_are_summed_across_four_quarters(): qs = _four_quarters(revenue=[100, 110, 120, 130], net_income=[10, 11, 12, 13]) r = build_ttm_line_items(qs) assert r.ok, r.reason assert r.line_items.revenue == 460 assert r.line_items.net_income == 46 assert r.quarters_used == 4 def test_stock_fields_take_the_most_recent_quarter_not_a_sum(): """The failure this guards: summing four quarters of total_assets gives a number ~4x too large that still looks like a plausible balance sheet.""" qs = _four_quarters(total_assets=[900, 950, 1000, 1050], shareholders_equity=[400, 420, 440, 460]) r = build_ttm_line_items(qs) assert r.ok, r.reason assert r.line_items.total_assets == 1050 assert r.line_items.shareholders_equity == 460 def test_stock_fields_are_not_averaged_either(): qs = _four_quarters(total_debt=[100, 200, 300, 400]) r = build_ttm_line_items(qs) assert r.line_items.total_debt == 400 # not 250 (mean), not 1000 (sum) def test_eps_is_treated_as_a_flow_and_summed(): """EPS is earnings per share for the period — a flow. Taking only the latest quarter's EPS would understate trailing EPS roughly fourfold and quadruple every P/E on the platform.""" qs = _four_quarters(eps_diluted=[0.5, 0.6, 0.7, 0.8]) r = build_ttm_line_items(qs) assert abs(r.line_items.eps_diluted - 2.6) < 1e-9 def test_share_count_is_treated_as_a_stock_not_a_flow(): qs = _four_quarters(diluted_shares=[100, 101, 102, 103]) r = build_ttm_line_items(qs) assert r.line_items.diluted_shares == 103 def test_price_and_market_cap_come_from_the_latest_quarter(): qs = _four_quarters(price=[10.0, 11.0, 12.0, 13.0], market_cap=[1000, 1100, 1200, 1300]) r = build_ttm_line_items(qs) assert r.line_items.price == 13.0 assert r.line_items.market_cap == 1300 # --- rule 2: a missing quarter poisons the sum --- def test_missing_flow_value_in_any_quarter_yields_none_not_a_partial_sum(): """The single most dangerous TTM bug: three quarters summed and presented as twelve months.""" qs = _four_quarters(revenue=[100, None, 120, 130]) r = build_ttm_line_items(qs) assert r.ok, r.reason assert r.line_items.revenue is None, "a 3-of-4 partial sum was returned as TTM revenue" def test_one_poisoned_field_does_not_poison_the_others(): qs = _four_quarters(revenue=[100, None, 120, 130], net_income=[10, 11, 12, 13]) r = build_ttm_line_items(qs) assert r.line_items.revenue is None assert r.line_items.net_income == 46 def test_missing_stock_value_in_an_older_quarter_is_irrelevant(): qs = _four_quarters(total_assets=[None, None, None, 1050]) r = build_ttm_line_items(qs) assert r.line_items.total_assets == 1050 def test_missing_stock_value_in_the_latest_quarter_is_none(): qs = _four_quarters(total_assets=[900, 950, 1000, None]) r = build_ttm_line_items(qs) assert r.line_items.total_assets is None # --- rule 3: four CONSECUTIVE quarters --- def test_three_quarters_is_refused_with_a_reason(): qs = _four_quarters(revenue=[100, 110, 120, 130])[:3] r = build_ttm_line_items(qs) assert not r.ok assert r.reason == "insufficient_quarters:3_of_4" assert r.quarters_used == 3 def test_no_quarters_is_refused_with_a_reason(): r = build_ttm_line_items([]) assert not r.ok and r.reason == "no_quarters_supplied" def test_a_skipped_quarter_is_refused_not_summed_over_fifteen_months(): qs = [ _q("2024-12-31", "2025-02-01", "Q4", revenue=100), _q("2025-03-31", "2025-05-01", "Q1", revenue=110), _q("2025-06-30", "2025-08-01", "Q2", revenue=120), _q("2025-12-31", "2026-02-01", "Q4", revenue=130), # Q3 missing ] r = build_ttm_line_items(qs) assert not r.ok assert r.reason.startswith("period_gap_too_large") def test_duplicate_period_end_is_refused(): qs = _four_quarters(revenue=[100, 110, 120, 130]) qs.append(_q("2025-12-31", "2026-02-01", "Q4", revenue=999)) r = build_ttm_line_items(qs) assert not r.ok and r.reason == "duplicate_period_end" def test_monthly_periods_are_refused_as_too_close_together(): qs = [ _q("2025-09-30", "2025-10-15", "Q3", revenue=100), _q("2025-10-31", "2025-11-15", "Q3", revenue=100), _q("2025-11-30", "2025-12-15", "Q4", revenue=100), _q("2025-12-31", "2026-01-15", "Q4", revenue=100), ] r = build_ttm_line_items(qs) assert not r.ok and r.reason.startswith("period_gap_too_small") def test_fiscal_calendar_drift_within_tolerance_is_accepted(): """A 52/53-week fiscal calendar produces 91- and 98-day quarters. These must NOT be rejected.""" qs = [ _q("2025-03-29", "2025-05-01", "Q1", revenue=100), _q("2025-06-28", "2025-08-01", "Q2", revenue=110), _q("2025-09-27", "2025-11-01", "Q3", revenue=120), _q("2026-01-03", "2026-02-05", "Q4", revenue=130), # 98-day 14-week quarter ] r = build_ttm_line_items(qs) assert r.ok, r.reason assert r.line_items.revenue == 460 def test_more_than_four_quarters_uses_the_four_most_recent(): qs = _four_quarters(revenue=[100, 110, 120, 130]) qs.append(_q("2024-12-31", "2025-02-01", "Q4", revenue=999)) r = build_ttm_line_items(qs) assert r.ok, r.reason assert r.line_items.revenue == 460 # the 999 quarter is outside the window def test_input_order_does_not_matter(): qs = _four_quarters(revenue=[100, 110, 120, 130]) reversed_result = build_ttm_line_items(list(reversed(qs))) assert reversed_result.ok and reversed_result.line_items.revenue == 460 # --- identity, provenance and refusals --- def test_period_type_is_ttm_and_period_end_is_the_latest_quarter(): r = build_ttm_line_items(_four_quarters(revenue=[100, 110, 120, 130])) assert r.line_items.period_type == "TTM" assert r.line_items.period_end == date(2025, 12, 31) def test_filing_date_is_the_latest_quarters_filing_date(): """Rule 4 — point-in-time integrity. A TTM figure is not knowable until its final quarter is filed; dating it any earlier reintroduces look-ahead bias.""" r = build_ttm_line_items(_four_quarters(revenue=[100, 110, 120, 130])) assert r.line_items.filing_date == date(2026, 2, 1) def test_annual_rows_are_refused_rather_than_silently_filtered(): qs = _four_quarters(revenue=[100, 110, 120, 130]) qs[0] = LineItems( security_id="SEC1", period_end=qs[0].period_end, period_type="FY", filing_date=qs[0].filing_date, currency="USD", revenue=400, ) r = build_ttm_line_items(qs) assert not r.ok and r.reason.startswith("non_quarterly_period_type") def test_mixed_currencies_are_refused(): qs = _four_quarters(revenue=[100, 110, 120, 130]) qs[2] = LineItems( security_id="SEC1", period_end=qs[2].period_end, period_type="Q3", filing_date=qs[2].filing_date, currency="EUR", revenue=120, ) r = build_ttm_line_items(qs) assert not r.ok and r.reason.startswith("mixed_currencies") def test_mixed_securities_are_refused(): qs = _four_quarters(revenue=[100, 110, 120, 130]) qs[1] = LineItems( security_id="SEC2", period_end=qs[1].period_end, period_type="Q2", filing_date=qs[1].filing_date, currency="USD", revenue=110, ) r = build_ttm_line_items(qs) assert not r.ok and r.reason.startswith("mixed_securities") # --- the classification itself --- def test_every_line_items_field_is_classified_exactly_once(): """If someone adds a field to LineItems and forgets to classify it, it would silently vanish from every TTM row. This test is the tripwire.""" assert validate_field_classification() == [] def test_flow_and_stock_buckets_do_not_overlap(): assert set(FLOW_FIELDS).isdisjoint(set(STOCK_FIELDS)) 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)