Blame for engine/tests/test_semantics.py

"""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