"""Targeted tests for the semantic-type, label-propagation, final-object, reliability-language and causal-polarity work. Section numbers refer to the remediation brief (§27 A–L, §28 properties).""" import os, sys, importlib.util, pathlib sys.path.insert(0, os.path.dirname(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 sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import report_model as rm # noqa: E402 (the one audited renderer) import pipeline_fixtures as pf # noqa: E402 def _sa_ok(): return {"ok": True, "blocking": [], "checks": [{"id": "A1", "ok": True}], "warnings": []} def _kw(**over): d = dict(policy={"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=[], engine={"status": "OK", "name": "e", "display_name": "e"}) d.update(over); return d # ── §27 A — correct engine, wrong renderer label ─────────────────────────────────────── # ── §27 B/C — same concept, different currency or scale ──────────────────────────────── def test_B_same_concept_mismatched_currency_is_blocked_until_normalized(): twd = cc.metric(cc.NET_REVENUE, 4800.0, metric_id="fc", raw_currency="TWD", raw_scale=cc.BILLION, unit="USD", period_type="NTM") usd = cc.metric(cc.NET_REVENUE, 170.0, metric_id="cons", raw_currency="USD", raw_scale=cc.BILLION, unit="USD", period_type="NTM") assert twd["normalized_value"] is None and "no FX rate" in twd["normalization_steps"][-1] ok, why = cc.compatible(twd, usd, "level") assert not ok and "unnormalized" in why with pytest.raises(cc.IncompatibleMetricsError): cc.derive(cc.GROWTH_RATE, lambda a, b: a / b - 1, [twd, usd], kind="level") twd_ok = cc.metric(cc.NET_REVENUE, 4800.0, metric_id="fc", raw_currency="TWD", raw_scale=cc.BILLION, fx_to_usd=0.0316, unit="USD", period_type="NTM") assert abs(twd_ok["normalized_value"] - 151.68e9) < 1e6 and cc.compatible(twd_ok, usd, "level")[0] def test_C_same_currency_mismatched_scale_normalizes_or_blocks(): mm = cc.metric(cc.NET_REVENUE, 170_000.0, metric_id="a", raw_scale=cc.MILLION, unit="USD", period_type="NTM") bn = cc.metric(cc.NET_REVENUE, 170.0, metric_id="b", raw_scale=cc.BILLION, unit="USD", period_type="NTM") assert mm["normalized_value"] == bn["normalized_value"] == 170e9 assert mm["normalized_scale"] == bn["normalized_scale"] == cc.UNIT and cc.compatible(mm, bn, "level")[0] # ── §27 D — ADS price with ordinary-share EPS ────────────────────────────────────────── def test_D_ads_price_with_ordinary_share_eps_is_blocked(): price = cc.metric(cc.PRICE, 434.0, metric_id="price", unit="per share", period_type="spot", security_basis=cc.ADS, per_share_basis=cc.PER_SHARE) eps_ord = cc.metric(cc.GAAP_EPS, 1.41, metric_id="eps_ord", unit="per share", period_type="TTM", security_basis=cc.ORDINARY_SHARE, per_share_basis=cc.PER_SHARE) ok, why = cc.compatible(price, eps_ord, "multiple") assert not ok and "security basis" in why with pytest.raises(cc.IncompatibleMetricsError): cc.derive(cc.PE_RATIO, lambda p, e: p / e, [price, eps_ord], kind="multiple") eps_ads = cc.metric(cc.GAAP_EPS, 1.41, metric_id="eps_ads", unit="per share", period_type="TTM", security_basis=cc.ORDINARY_SHARE, target_basis=cc.ADS, adr_ads_ratio=5, per_share_basis=cc.PER_SHARE) pe = cc.derive(cc.PE_RATIO, lambda p, e: p / e, [price, eps_ads], kind="multiple", unit="x") assert abs(eps_ads["normalized_value"] - 7.05) < 0.01 and abs(pe["value"] - 61.6) < 0.5 # ── §27 E — peer multiple inconsistent with price ÷ EPS ──────────────────────────────── def test_E_peer_pe_inconsistent_with_price_over_eps_is_removed(): # the panel's P/E was computed from ORDINARY-share earnings against an ADS price: 434 ÷ 1.41 ≈ 308x, # while the report's own price ÷ per-ADS EPS is 434 ÷ 7.04 ≈ 62x d = {"peer_comps": {"subject": {"ticker": "T", "pe": 307.8}, "peers": [{"ticker": "P", "pe": 30.0, "ps": 9.0, "op_margin": 0.05}]}} rec = cn.reconcile_peer_table(d, {"price": {"value": 434.0}, "eps_ttm_profile": {"value": 7.04}}) assert rec["subject_ok"] is False and d["peer_comps"]["subject"]["pe"] is None assert d["peer_comps"]["peers"][0]["pe"] is None and "peer_table_subject_pe" in d["rejected_fields"] # ── §27 F — displayed scenarios always get arithmetic validation ─────────────────────── # ── §27 G — LIMITED reliability with "heavily backtested" prose ──────────────────────── # ── §27 H — an UNRELIABLE model's numbers never reach a persona ──────────────────────── # ── §27 I — intrinsic negative, pricing positive, momentum positive ──────────────────── # ── §27 J — causal polarity of a thesis trigger ──────────────────────────────────────── # ── §27 K — an executive from another organisation ───────────────────────────────────── # ── §27 L — unresolved MATERIAL text-audit claim blocks publication ──────────────────── # ── §28 property tests ───────────────────────────────────────────────────────────────── def test_P1_raw_currency_change_with_same_normalized_value_does_not_move_valuation(): a = cc.metric(cc.NET_REVENUE, 100.0, raw_currency="USD", raw_scale=cc.BILLION, unit="USD") b = cc.metric(cc.NET_REVENUE, 100_000.0 / 0.0316 * 0.0316, raw_currency="USD", raw_scale=cc.MILLION, unit="USD") c_ = cc.metric(cc.NET_REVENUE, 100.0 / 0.0316, raw_currency="TWD", raw_scale=cc.BILLION, fx_to_usd=0.0316, unit="USD") assert abs(a["normalized_value"] - c_["normalized_value"]) < 1e3 and abs(a["normalized_value"] - b["normalized_value"]) < 1e3 def test_P2_ads_ratio_change_leaves_market_cap_and_valuation_unchanged(): for k in (1, 2, 5, 10): px = 434.0 * (k / 5.0) # price per ADS scales with the ratio eps = cc.metric(cc.GAAP_EPS, 1.41, unit="per share", security_basis=cc.ORDINARY_SHARE, target_basis=cc.ADS, adr_ads_ratio=k, per_share_basis=cc.PER_SHARE, period_type="TTM") shares_ads = 25.9e9 / k assert abs(px * shares_ads - 434.0 * (25.9e9 / 5.0)) < 1e6 # market cap invariant assert abs((px / eps["normalized_value"]) - (434.0 / 7.05)) < 0.5 # P/E invariant def test_P3_reliability_downgrade_never_increases_weight_or_strength(): v = pol.quant_reliability({"dir_acc": 0.62, "rank_ic": 0.3, "n_oos_eff": 12}) u = pol.quant_reliability({"dir_acc": 0.40, "rank_ic": 0.0, "n_oos_eff": 5}) assert u["max_blend_weight"] <= v["max_blend_weight"] and not u["persona_visibility"] and v["persona_visibility"] assert set(v["forbidden_phrases"]) <= set(u["forbidden_phrases"]) def test_P4_engine_change_changes_the_methodology_label_automatically(): a = ve.select_engine("FcffValuation").metadata(); b = ve.select_engine("BankEquityValuation").metadata() assert a["display_name"] != b["display_name"] and a["methodology_description"] != b["methodology_description"] assert "FCFF" in a["allowed_renderer_terms"] and "FCFF" in b["forbidden_renderer_terms"] def test_P5_displayed_odds_force_validation(): s = cc.scenario_set_from(0.3, 0.3, 0.4, -0.08, 0.19, 0.30, 0.20) assert s.validate()[0] == "FAIL" # 0.133 ≠ 0.20 ok = cc.scenario_set_from(0.3, 0.3, 0.4, -0.08, 0.19, 0.30, 0.3 * -0.08 + 0.3 * 0.19 + 0.4 * 0.30) assert ok.validate()[0] == "PASS" def test_P7_trigger_polarity_change_changes_the_flip_language(): up = cc.build_triggers([{"metric": "operating margin", "direction": "up", "operator": ">=", "threshold": "6%", "scenario": "BULL"}])[0] dn = cc.build_triggers([{"metric": "operating margin", "direction": "down", "operator": "<=", "threshold": "4%", "scenario": "BEAR"}])[0] assert "bull case" in up.sentence() and "bear case" in dn.sentence() assert up.polarity == cc.POSITIVE_FOR_EQUITY and dn.polarity == cc.NEGATIVE_FOR_EQUITY # ── §29 invariants not covered above ─────────────────────────────────────────────────── def test_inv_models_never_consume_raw_when_normalized_exists(): m = cc.metric(cc.NET_REVENUE, 4800.0, raw_currency="TWD", raw_scale=cc.BILLION, fx_to_usd=0.0316, unit="USD") assert m["value"] == m["normalized_value"] != m["raw_value"] def test_inv_final_object_version_must_appear_in_the_document(): run, _, _ = pf.run_pipeline() import catalog as C rep = rm.build_user_report(run.state, C.build_catalog(run.state), {"summary": [], "thesis": [], "risks": [], "seats": {}}, []) assert rm.audit(rep, run.state, rm.render_markdown(rep), rm.render_html(rep)).passed for blk in rep.blocks: blk.content = [f for f in blk.content if getattr(f, "claim_id", None) != "VERSION"] if blk.kind == "para" else blk.content out = rm.audit(rep, run.state, rm.render_markdown(rep), rm.render_html(rep)) assert "R2" in {c.id for c in out.checks if c.status == "FAIL"} # ── false-positive regressions found in the acceptance runs ──────────────────────────── def test_reg_foreign_private_issuer_on_a_us_exchange_is_ads_quoted(): sub = {"cik": 1, "sub": {"filings": {"recent": {"form": ["20-F", "6-K"]}}}, } sub["sub"]["exchanges"] = ["NYSE", "OTC"] iss = cn.classify_issuer("TSM", {"Name": "Taiwan Semiconductor Manufacturing Company Limited", "Country": "USA", "Industry": "SEMICONDUCTORS"}, sub, {"currency": "TWD", "accounting_standard": "IFRS"}) assert iss["is_fpi"] and iss["is_adr"], "an FPI listed on a US exchange trades as ADS" # --------------------------------------------------------------------------- # §30 Consensus boundary normalization # --------------------------------------------------------------------------- def _consdat(**over): d = {"fx_usd": 0.031674, "overview": {"RevenueTTM": 1.41e11, "EPS": 13.39}, "income": {"quarterlyReports": []}, "earnings_estimates_consensus": {"current_year_revenue": 5.4422750681e12, "next_year_revenue": 6.85e12, "current_year_eps": 16.93, "next_year_eps": 21.93, "analyst_count": 13}} d["earnings_estimates_consensus"].update(over) return d def test_filing_currency_consensus_is_converted_before_any_comparison(): """A TWD consensus in a USD-labelled field must not be compared with a USD forecast.""" d = _consdat() notes = cn.normalize_consensus(d) ee = d["earnings_estimates_consensus"] assert notes and "consensus revenue" in notes[0] assert 1.5e11 < ee["current_year_revenue"] < 2.0e11, "consensus revenue must land near the USD anchor" assert ee["next_year_revenue"] > ee["current_year_revenue"], "the group converts together" assert ee["current_year_eps"] == 16.93, "an already-USD field is left untouched" def test_consensus_normalization_is_idempotent(): d = _consdat() first = cn.normalize_consensus(d) rev = d["earnings_estimates_consensus"]["current_year_revenue"] assert cn.normalize_consensus(d) == first assert d["earnings_estimates_consensus"]["current_year_revenue"] == rev, "no double conversion" def test_unresolvable_consensus_is_withheld_not_published(): d = _consdat(current_year_revenue=9.9e17, next_year_revenue=1.0e18) notes = cn.normalize_consensus(d) assert any("UNIT_UNRESOLVED" in n for n in notes) assert "current_year_revenue" not in d["earnings_estimates_consensus"] def test_single_currency_issuer_needs_no_consensus_conversion(): d = _consdat() d["fx_usd"] = 1.0 d["earnings_estimates_consensus"]["current_year_revenue"] = 3.9e11 assert cn.normalize_consensus(d) == [] assert d["earnings_estimates_consensus"]["current_year_revenue"] == 3.9e11