crucible / engine /tests /test_architecture.py
s-ttp's picture
Root-cause remediation: reconciled forecast basis, canonical segments, PricingState, company guidance, ownership/insider permissions, S1-S12 validators
931437f verified
Raw History Blame Contribute Delete
17.7 kB
"""Architecture tests: invariants, property/metamorphic behaviour, golden cross-sector fixtures and
intentional corruption. Pure fixtures — no network, no LLM. These test the mechanisms that make
invalid states impossible, not a particular ticker's prose."""
import os, sys, copy
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import pytest
import concepts as cc
import policy as pol
import valuation_engines as ve
import canonical as cn
# ── fixtures ───────────────────────────────────────────────────────────────────────────
def bank_data(gross=True):
q = []
for i in range(8):
rep = {"fiscalDateEnding": f"2026-0{(6 - i) % 12 + 1:d}-30" if i < 6 else f"2025-0{i}-30", "netInterestIncome": 24e9, "nonInterestIncome": 22e9, "interestExpense": 28e9,
"netIncome": 14e9, "operatingExpenses": 24e9}
rep["totalRevenue"] = 74e9 if gross else 46e9
q.append(rep)
ann = [{"fiscalDateEnding": "2025-12-31", "totalRevenue": (296e9 if gross else 184e9), "netInterestIncome": 96e9, "nonInterestIncome": 88e9, "interestExpense": 112e9}]
return {"overview": {"Name": "Big Bank", "Industry": "NATIONAL COMMERCIAL BANKS", "Sector": "FINANCIAL SERVICES", "MarketCapitalization": 800e9, "SharesOutstanding": 2.7e9,
"RevenueTTM": 186e9, "EPS": 20.0, "PERatio": 15.0, "ReturnOnEquityTTM": 0.17, "PayoutRatio": 0.26, "DividendYield": 0.017},
"income": {"quarterlyReports": q, "annualReports": ann},
"balance": {"quarterlyReports": [{"totalShareholderEquity": 350e9, "goodwill": 50e9, "intangibleAssets": 10e9}]},
"cashflow": {"quarterlyReports": [{"dividendPayoutCommonStock": -3.5e9}] * 4}}
def issuer(cls, **kw):
d = {"issuer_class": cls, "is_financial": cls in cc.FINANCIAL_ISSUER_CLASSES, "mixed_financial": False}
d.update(kw); return d
def generic_fund():
return {"assessments": [("Valuation", "P/E Ratio", "15.0", "GOOD", ""), ("Health", "Debt/Equity", "3.30", "BAD", ""), ("Health", "Current Ratio", "0.84", "WEAK", ""),
("Health", "Interest Coverage", "0.9x", "BAD", ""), ("Cash Flow", "OCF > Net Income", "No", "WEAK", ""), ("Profitability", "Operating Margin", "50%", "STRONG", ""),
("Profitability", "ROE", "17.8%", "GOOD", ""), ("Valuation", "EV/Revenue", "5.8", "WEAK", "")],
"scorecard": {"Valuation": (5, 8, {}), "Health": (1, 4, {}), "Cash Flow": (1, 4, {}), "Profitability": (4, 8, {})}, "total_points": 11.0, "max_points": 24.0}
def _sa_ok():
return {"ok": True, "blocking": [], "checks": [{"id": "A1", "ok": True}], "warnings": []}
def _gate_kw(p=None, **over):
kw = dict(policy=(p.as_dict() if p is not None else {"issuer_class": "industrial", "disallowed_scorecard_rows": []}),
scorecard={"metrics": []}, concepts={"revenue": cc.NET_REVENUE}, render_audit={"ok": True, "detail": ""},
metrics={"price": {"normalized_value": 1.0}}, text_audit_unresolved=[])
kw.update(over); return kw
# ── INVARIANTS ─────────────────────────────────────────────────────────────────────────
def test_inv_bank_has_no_generic_leverage_rows():
p = pol.ApplicabilityPolicy.for_issuer(issuer("bank"))
fund, info = pol.apply_policy_to_scorecard(generic_fund(), p, bank_data(), 300.0)
names = {r[1] for r in fund["assessments"]}
assert not (names & {"Debt/Equity", "Current Ratio", "Interest Coverage", "OCF > Net Income", "Operating Margin", "EV/Revenue"})
assert {"ROE", "ROTCE", "P/TBV", "Efficiency ratio", "CET1 ratio"} <= names
cet1 = next(r for r in fund["assessments"] if r[1] == "CET1 ratio")
assert cet1[2] == "UNAVAILABLE" and cet1[3] == "UNVERIFIED" # missing stays missing, visibly
def test_inv_value_label_equals_engine_type():
eng = ve.select_engine("BankEquityValuation")
out = eng.run({"methods": {"DCF": {"price": 300.0}}, "dcf_meta": {"basis": "eps"}})
assert out["status"] == "FAILED_BASIS_MISMATCH" and out["value"] is None
ok = eng.run({"methods": {"DCF": {"price": 300.0}}, "dcf_meta": {"basis": "book_roe", "bvps": 136.0, "justified_pb": 2.2}})
assert ok["status"] == "OK" and "justified P/B" in ok["display_name"] and ok["components"][0]["value"] == 136.0
assert "justified price-to-book" in ok["methodology_description"] and "FCFF" in ok["forbidden_renderer_terms"]
def test_inv_render_audit_catches_mismatch():
"""The delivered document is audited against the frozen state (engine/report_model.audit)."""
import pipeline_fixtures as pf
run, payload, _ = pf.run_pipeline()
assert payload is not None # consistent → passes
fcff_forbidden = run.state.valuation.get("engine")["forbidden_renderer_terms"][0]
res = pf.audit_with_text(run, f"The {fcff_forbidden} model implies more.")
assert res["R4"].status == "FAIL" # engine terminology leak
res = pf.audit_with_text(run, "The composite is $432.80.")
assert res["R1"].status == "FAIL" # a figure that is not a claim
# ── PROPERTY / METAMORPHIC ─────────────────────────────────────────────────────────────
def test_prop_gross_to_net_switch_keeps_canonical_revenue_unchanged():
dg, dn = bank_data(gross=True), bank_data(gross=False)
pol.canonicalize_statements(dg, issuer("bank")); pol.canonicalize_statements(dn, issuer("bank"))
tg_ = sum(r["totalRevenue"] for r in dg["income"]["quarterlyReports"][:4]); tn_ = sum(r["totalRevenue"] for r in dn["income"]["quarterlyReports"][:4])
assert abs(tg_ - tn_) < 1 and abs(tg_ - 184e9) < 1
assert dg["rejected_fields"]["provider_statement_revenue"]["replaced_by"] == cc.NET_REVENUE and dg["canonical_concepts"]["revenue"] == cc.NET_REVENUE
def test_prop_issuer_switch_industrial_to_bank_removes_leverage_metrics():
fund_i, _ = pol.apply_policy_to_scorecard(generic_fund(), pol.ApplicabilityPolicy.for_issuer(issuer("industrial")), bank_data(), 300.0)
fund_b, _ = pol.apply_policy_to_scorecard(generic_fund(), pol.ApplicabilityPolicy.for_issuer(issuer("bank")), bank_data(), 300.0)
ni = {r[1] for r in fund_i["assessments"]}; nb = {r[1] for r in fund_b["assessments"]}
assert "Debt/Equity" in ni and "Debt/Equity" not in nb and "Current Ratio" in ni and "Current Ratio" not in nb
def test_prop_quant_reliability_ordering():
states = [pol.quant_reliability(m)["state"] for m in ({"dir_acc": 0.62, "rank_ic": 0.3, "n_oos_eff": 12}, {"dir_acc": 0.52, "rank_ic": 0.1, "n_oos_eff": 6}, {"dir_acc": 0.40, "rank_ic": 0.0, "n_oos_eff": 5}, {})]
assert states == ["VALIDATED", "LIMITED", "UNRELIABLE", "UNAVAILABLE"]
caps = [pol.quant_reliability(m)["weight_cap"] for m in ({"dir_acc": 0.62, "rank_ic": 0.3, "n_oos_eff": 12}, {"dir_acc": 0.40, "rank_ic": 0.0, "n_oos_eff": 5})]
assert caps[0] > caps[1]
# ── GOLDEN CROSS-SECTOR FIXTURES ───────────────────────────────────────────────────────
GOLDEN = [
("large US bank", "bank", "BankEquityValuation", "book_roe", False, ("P/E", "Analyst Target"), cc.NET_REVENUE),
("regional bank", "bank", "BankEquityValuation", "book_roe", False, ("P/E", "Analyst Target"), cc.NET_REVENUE),
("insurer", "insurer", "InsurerEquityValuation", "book_roe", False, ("P/E", "Analyst Target"), cc.NET_REVENUE),
("REIT", "reit", "ReitValuation", "affo", False, ("P/S", "Analyst Target"), cc.NET_REVENUE),
("software", "software", "FcffValuation", "eps", True, ("P/E", "PEG", "P/FCF", "P/S", "Analyst Target"), cc.NET_REVENUE),
("industrial", "industrial", "FcffValuation", "eps", True, ("P/E", "PEG", "P/FCF", "P/S", "Analyst Target"), cc.NET_REVENUE),
("foreign ADR (semis)", "industrial", "FcffValuation", "eps", True, ("P/E", "PEG", "P/FCF", "P/S", "Analyst Target"), cc.NET_REVENUE),
("services (spin-off parent)", "services", "FcffValuation", "eps", True, ("P/E", "PEG", "P/FCF", "P/S", "Analyst Target"), cc.NET_REVENUE),
("conglomerate / holding", "holding", "SumOfPartsValuation", "eps", True, ("P/E", "PEG", "P/FCF", "P/S", "Analyst Target"), cc.NET_REVENUE),
]
@pytest.mark.parametrize("label,cls,engine,basis,story,methods,revc", GOLDEN)
def test_golden_policy_per_sector(label, cls, engine, basis, story, methods, revc):
p = pol.ApplicabilityPolicy.for_issuer(issuer(cls))
assert p.valuation_engine == engine and p.mechanical_basis == basis and p.story_eligible == story
assert tuple(p.allowed_pricing_methods) == methods and p.revenue_concept == revc
e = ve.select_engine(p.valuation_engine)
assert e.basis == basis and e.name
# ── INTENTIONAL CORRUPTION ─────────────────────────────────────────────────────────────
def test_corrupt_wrong_ceo_business_unit_title_is_not_firm_level():
import predict_yearly as py
f4 = [{"filed": "2026-05-15", "owner": "Marianne Lake", "title": "CEO CCB"}, {"filed": "2026-05-10", "owner": "Jeremy Barnum", "title": "Chief Financial Officer"},
{"filed": "2026-05-01", "owner": "Jamie Dimon", "title": "Chairman & CEO"}, {"filed": "2026-04-01", "owner": "Someone Else", "title": "CEO of Asset & Wealth Management"}]
lc = py._leadership_canon([], None, f4)
assert lc["CEO"]["name"] == "Jamie Dimon" and lc["CFO"]["name"] == "Jeremy Barnum" and lc["CEO"]["verified"] and lc["CEO"]["scope"] == "firm"
def test_corrupt_unreliable_quant_cannot_hold_large_weight():
import predict_yearly as py
fund = {"expected_return": 0.20, "confidence": 0.5}; macro = {"expected_return": 0.0, "confidence": 0.5}
ml_bad = {"expected_return": -0.30, "confidence": 0.6, "backtest_dir_accuracy": 0.40, "_reliability": pol.quant_reliability({"dir_acc": 0.40, "rank_ic": 0.0, "n_oos_eff": 5})}
ml_good = {"expected_return": -0.30, "confidence": 0.6, "backtest_dir_accuracy": 0.62, "_reliability": pol.quant_reliability({"dir_acc": 0.62, "rank_ic": 0.3, "n_oos_eff": 12})}
wb = next(r for r in py.blend_predictions_yearly(fund, macro, ml_bad)["blend_audit"]["rows"] if r["name"].startswith("Price Model"))
wg = next(r for r in py.blend_predictions_yearly(fund, macro, ml_good)["blend_audit"]["rows"] if r["name"].startswith("Price Model"))
assert wb["share"] < wg["share"] and "reliability UNRELIABLE" in (wb["cap_note"] or "")
# ── P1 additions: primary-source segment facts, bank facts, ADS ratio, 20-F officers, engines ──
def test_segment_facts_parser_geographic_vs_business_line():
inst = '''<xbrl><context id="c1"><entity><segment><xbrldi:explicitMember dimension="us-gaap:StatementBusinessSegmentsAxis">x:BrazilSegmentMember</xbrldi:explicitMember><xbrldi:explicitMember dimension="srt:ConsolidationItemsAxis">us-gaap:OperatingSegmentsMember</xbrldi:explicitMember></segment></entity><period><startDate>2025-01-01</startDate><endDate>2025-12-31</endDate></period></context>
<context id="c0"><entity></entity><period><startDate>2025-01-01</startDate><endDate>2025-12-31</endDate></period></context>
<context id="c2"><entity><segment><xbrldi:explicitMember dimension="us-gaap:StatementBusinessSegmentsAxis">x:PaymentsSegmentMember</xbrldi:explicitMember><xbrldi:explicitMember dimension="srt:ConsolidationItemsAxis">us-gaap:OperatingSegmentsMember</xbrldi:explicitMember></segment></entity><period><startDate>2025-01-01</startDate><endDate>2025-12-31</endDate></period></context>
<us-gaap:Revenues contextRef="c0" unitRef="usd">100</us-gaap:Revenues><us-gaap:Revenues contextRef="c1" unitRef="usd">60</us-gaap:Revenues><x:DirectContribution contextRef="c1" unitRef="usd">9</x:DirectContribution>
<us-gaap:Revenues contextRef="c2" unitRef="usd">40</us-gaap:Revenues><us-gaap:OperatingIncomeLoss contextRef="c2" unitRef="usd">7</us-gaap:OperatingIncomeLoss></xbrl>'''
cn._INST_CACHE[999] = {"text": inst, "url": "u", "filed": "2026-02-01", "form": "10-K", "accession": "a"}
sf = cn.edgar_segment_facts(999, {"filings": {"recent": {}}})
kinds = {s["name"]: (s["kind"], s["profit"]) for s in sf["segments"]}
assert kinds["Brazil"] == ("geographic", 9.0) and kinds["Payments"] == ("business_line", 7.0) and sf["business_line_profit"]
def test_sotp_engine_values_business_lines_and_mixed_engine_when_not_disclosed():
sf_bl = {"segments": [{"name": "Marketplace", "kind": "business_line", "revenue": 20e9, "profit": 2e9}, {"name": "Fintech", "kind": "business_line", "revenue": 10e9, "profit": 1e9}], "business_line_profit": True, "source": "SEC 10-K"}
out = ve.select_engine("SumOfPartsValuation").run({"methods": {"DCF": {"price": 100.0}}, "dcf_meta": {"basis": "eps", "g1": 0.08, "g_terminal": 0.04, "high_growth_years": 5, "beta": 1.1, "beta_stable": 1.0, "roe": 0.2}},
{"segment_facts": sf_bl, "segments_canonical": {"status": "RECONCILED"}, "shares": 100e6, "tax_rate": 0.25, "net_debt": 0.0, "rf": 0.045, "erp": 0.0428})
assert out["status"] == "OK" and out["value"] > 0 and {c["kind"] for c in out["components"] if "kind" in c} == {"operating", "financial"}
sf_geo = {"segments": [{"name": "Brazil", "kind": "geographic", "revenue": 15e9, "profit": 2e9}], "revenue_by_line": [{"name": "Fintech", "revenue": 12.6e9}], "business_line_profit": False, "source": "SEC 10-K"}
out2 = ve.select_engine("SumOfPartsValuation").run({"methods": {"DCF": {"price": 100.0}}, "dcf_meta": {"basis": "eps"}}, {"segment_facts": sf_geo, "segments_canonical": {"status": "RECONCILED"}, "shares": 1e6, "rf": 0.045, "erp": 0.0428})
assert out2["status"] == "UNAVAILABLE_COMPONENTS" and out2["value"] is None
# segments that do not reconcile to the company never reach a sum-of-parts
out3 = ve.select_engine("SumOfPartsValuation").run({"methods": {"DCF": {"price": 100.0}}, "dcf_meta": {"basis": "eps"}},
{"segment_facts": sf_bl, "segments_canonical": {"status": "NOT_RECONCILED", "reason": "Σ 160%"},
"shares": 100e6, "rf": 0.045, "erp": 0.0428})
assert out3["status"] == "UNAVAILABLE" and out3["value"] is None
mixed = ve.select_engine("MixedEquityValuation").run({"methods": {"DCF": {"price": 680.0}}, "dcf_meta": {"basis": "eps"}}, {"segment_facts": sf_geo})
assert mixed["status"] == "OK" and mixed["value"] == 680.0 and any("geographic" in n for n in mixed["notes"])
p = pol.ApplicabilityPolicy.for_issuer(issuer("software", mixed_financial=True))
assert p.valuation_engine == "MixedEquityValuation" and not p.story_eligible
def test_bank_rows_from_sec_facts_and_unavailable_cet1():
bf = {"deposits": {"value": 2713.7e9, "prior_year": 2562.4e9, "end": "2026-06-30", "tag": "Deposits", "form": "10-Q"},
"loans_gross": {"value": 1463.8e9, "prior_year": 1345.5e9, "end": "2026-06-30", "tag": "FinancingReceivable…", "form": "10-Q"},
"allowance": {"value": 26.2e9, "end": "2026-06-30", "tag": "Allowance"}, "net_charge_offs": {"value": 4.68e9, "period_days": 181, "tag": "Writeoff"},
"nii": {"value": 50.9e9, "period_days": 181, "tag": "InterestIncomeExpenseNet"}, "total_assets": {"value": 5015e9, "tag": "Assets"},
"cet1_ratio": {"value": None, "status": "UNAVAILABLE"}, "tier1_ratio": {"value": None, "status": "UNAVAILABLE"}, "source": "SEC"}
rows = {r["name"]: r for r in pol.bank_rows_from_facts(bf, 350.0, 2.7e9)}
assert rows["Deposit growth"]["value"].startswith("+5.") and rows["Loan growth"]["value"].startswith("+8.")
assert rows["Reserve coverage (allowance ÷ loans)"]["value"] == "1.79%" and rows["Net charge-off ratio (annualised)"]["value"].endswith("%")
assert rows["CET1 ratio"]["value"] == "UNAVAILABLE" and rows["CET1 ratio"]["rating"] == "UNVERIFIED"
def test_ads_ratio_and_officer_parsers():
cn._DOC_CACHE[777] = {"text": "… shares held in the form of ADSs (each ADS represents five (5) common shares) held by such individuals. Executive Officers Name Position with our company Years C.C. Wei Chairman; Chief Executive Officer 28 Wendell Huang Senior Vice President and Chief Financial Officer 25 Y.P. Chyn Executive Vice President 40", "url": "u", "filed": "2026-04-16"}
assert cn.edgar_ads_ratio(777, {"x": 1})["ratio"] == 5
o = cn.edgar_20f_officers(777, {"x": 1})
assert o["CEO"]["name"] == "C.C. Wei" and o["CFO"]["name"] == "Wendell Huang"
def test_no_risk_free_no_value():
import story_dcf as sd
assert sd.story_fcff({"revenue": 1e9, "shares": 1e6, "price": 10.0, "rf": None, "sector": "TECHNOLOGY"}, {"tam_at_maturity_usd_b": 10, "plausible_peak_share": 0.1, "years_to_maturity": 5, "target_operating_margin": 0.2, "sales_to_capital": 2.0, "failure_probability": 0.0}) is None
assert "risk-free" in sd.LAST_REASON
import predict_yearly as py
assert py._median_roe({"income": {"annualReports": []}, "balance": {"annualReports": []}}, {}) is None