| """Structured arguments (crucible.args/2): the contract that ends the prose-remediation loop. |
| |
| Measured on 14 live SPGI runs (1,047 labelled sentences): free prose with spliced number tokens was 52% defective as |
| written; a word-list validator published 30% defects and discarded 35% of good prose; 62% of published defects were |
| semantic. These tests pin the structural replacement: typed evidence, figure-free judgement, templated figures, |
| validated forwarding between rounds, one semantic gate, and a debate that can never crash a run. |
| """ |
| import copy |
| import json |
| import os |
| import sys |
|
|
| import pytest |
|
|
| 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 arguments as A |
| import catalog as C |
| import pipeline_fixtures as F |
| import predict_monthly as PM |
| import predict_yearly as PY |
| import test_spgi_relations as TR |
|
|
|
|
| def forecast_run(): |
| import pipeline as pl |
| d, st = TR._story_with_breakeven() |
| pf_ = F.pred_fund(); pf_.update(revenue_annual=11.8e9, eps_annual=4.3) |
| r = pl.Run("SPGI", run_id="SPGI-args-test") |
| v = r.data_stage(d, 160.0, price_as_of="2026-09-21", tech={"signals": {"momentum_3m": 0.04, "momentum_12m": -0.118}}) |
| assert v.passed, [c.id for c in v.blocking] |
| fi = F.legacy_forecast_inputs(pred_fund=pf_, story=F.story_invariants(st)); fi["all_data"] = d |
| v = r.forecast_stage(**fi) |
| assert v.passed, [(c.id, c.detail) for c in v.blocking] |
| return r |
|
|
|
|
| @pytest.fixture(scope="module") |
| def ctx_run(): |
| r = forecast_run() |
| cat = C.build_catalog(r.state, include_deferred=False) |
| return r, r.args_context(cat) |
|
|
|
|
| |
| def test_offer_is_the_kind_map_minus_withheld(ctx_run): |
| r, ctx = ctx_run |
| for cid in ctx.offer: |
| assert cid not in A.WITHHELD and "__" not in cid |
| assert A.kinds_for(cid, ctx.cat) or cid.startswith(("OWN_13F_", "INSIDER_")), cid |
| for cid in ("ADJ_GAAP_BRIDGE", "CEO", "CFO", "DMD_MARGIN", "WACC_BENCHMARK", "SEGMENT_COUNT_FILED"): |
| assert cid not in ctx.offer |
| assert not any(c.startswith(("OWN_13F_", "INSIDER_")) for c in ctx.offer) |
|
|
|
|
| def test_compatible_is_not_permission_and_relations_expand_into_constituents(ctx_run): |
| _r, ctx = ctx_run |
| rs = A.evidence_reasons(ctx, ["FORECAST_REVENUE", "REL_BREAK_EVEN_REVENUE_STORY"], None, |
| allowed=set(ctx.offer)) if "REL_BREAK_EVEN_REVENUE_STORY" in ctx.offer else ["PAIR_skip"] |
| assert rs and rs[0].startswith("PAIR"), rs |
| assert A.evidence_reasons(ctx, ["REL_PRICE_INTRINSIC", "REL_PRICE_MARKET_COMPOSITE"], "VALUATION") == [] |
| assert A.evidence_reasons(ctx, [], "VALUATION")[0].startswith("NO_EVIDENCE") |
| assert A.evidence_reasons(ctx, ["DMD_MARGIN"], "COMPETITION")[0].startswith(("WITHHELD", "UNKNOWN")) |
| |
| assert A.evidence_reasons(ctx, ["PE_FORWARD"], "EXPECTATIONS") == [] |
|
|
|
|
| |
| @pytest.mark.parametrize("text,code", [ |
| ("At [[PE_FORWARD]] the multiple is rich.", "TOKEN_IN_VIEW"), |
| ("The multiple sits near 20 times earnings.", "NUMERAL"), |
| ("Growth stays in the high-teens for years.", "NUMERAL"), |
| ("Two adds and four new positions show interest.", "NUMERAL"), |
| ("We keep our SELL view on the shares.", "RATING_WORD"), |
| ("Management said “monetization is hard” this quarter.", "QUOTE_IN_VIEW"), |
| ("This leaves little room for error.", "ANAPHORIC_OPENING"), |
| ("The PE_FORWARD multiple looks stretched.", "ID_TEXT_IN_VIEW"), |
| ("A premium that leaves little room for error", "INCOMPLETE"), |
| ]) |
| def test_text_rules(ctx_run, text, code): |
| _r, ctx = ctx_run |
| assert any(x.startswith(code) for x in A.text_reasons(ctx, text, max_words=30)), (text, A.text_reasons(ctx, text, max_words=30)) |
|
|
|
|
| def test_direction_rules_read_typed_state(ctx_run): |
| _r, ctx = ctx_run |
| vr = ctx.rules.valuation_relations |
| rid = "REL_PRICE_INTRINSIC" |
| wrong = "cheap" if vr.get(rid) == "ABOVE" else "expensive" |
| assert any(x.startswith("VALUATION") for x in A.direction_reasons(ctx, f"The stock looks {wrong} here.", [rid], A._own(None, None))) |
| assert any(x.startswith("VALUATION_UNCITED") for x in A.direction_reasons(ctx, "The stock looks cheap here.", [], A._own(None, None))) |
| assert any(x.startswith("RETURN_SIGN") for x in A.direction_reasons(ctx, "The expected return is bearish.", [], A._own(0.07, "HOLD"))) |
| assert any(x.startswith("ATTRIBUTION") for x in A.direction_reasons(ctx, "Management conceded that monetization is hard.", [], |
| A._own(None, None))) |
|
|
|
|
| |
| SEAT = { |
| "neutral": {"expected_return_pct": 5.0, "predicted_price": 168, "recommendation": "HOLD", "confidence": 0.6, |
| "thesis": "The franchise is sound but the valuation already asks a lot.", |
| "points": [{"kind": "VALUATION", "evidence": ["PE_FORWARD"], "view": "At [[PE_FORWARD]] the multiple is 20x."}, |
| {"kind": "COMPETITION", "evidence": ["DMD_MARGIN"], "view": "Margins far exceed the industry."}]}, |
| "bear": {"expected_return_pct": -12.0, "predicted_price": 141, "recommendation": "SELL", "confidence": 0.55, |
| "thesis": "Today's price asks for more than the story delivers.", "thesis_rests_on": [1], |
| "points": [{"kind": "EXPECTATIONS", "evidence": ["REL_EXPECTATIONS_STORY"], |
| "view": "The price needs revenue or margins the story does not reach."}, |
| {"kind": "RISK", "evidence": ["STORY_FAILURE"], "view": "Management conceded that monetization remains hard."}]}, |
| "bull": {"expected_return_pct": 22.0, "predicted_price": 195, "recommendation": "BUY", "confidence": 0.55, |
| "thesis": "Earnings power supports the shares.", "thesis_rests_on": [1], |
| "points": [{"kind": "EARNINGS", "evidence": ["REL_FORECAST_EPS_CONSENSUS"], |
| "view": "Estimates have room to move toward the engine's view."}]}, |
| } |
| REB = {"bear": {"expected_return_pct": -10.0, "predicted_price": 144, "recommendation": "SELL", "confidence": 0.5, |
| "thesis": "Upgrades help, yet the valuation still asks too much.", "thesis_rests_on": [1], |
| "answers": [{"to": "P1", "verdict": "PARTLY", "evidence": ["REL_FORECAST_EPS_CONSENSUS"], |
| "view": "Upgrades are plausible yet do not close the valuation gap."}, |
| {"to": "P9", "verdict": "FAILS", "evidence": [], "view": "No such point exists."}]}, |
| "bull": {"stance": "HOLD", "thesis": "No numbers here.", "answers": []}} |
| PMR = {"p_bear": 0.35, "p_bull": 0.3, "p_neutral": 0.35, "conviction": 0.6, |
| "thesis": "The valuation bar outweighs the earnings upside, so the odds lean slightly to the bear case.", |
| "thesis_rests_on": [1], |
| "factors": [{"kind": "EXPECTATIONS", "evidence": ["REL_EXPECTATIONS_STORY"], "view": "What the price requires sets a high bar."}, |
| {"kind": "VALUATION", "evidence": ["REL_PRICE_MECHANICAL"], "view": "The cross-check adds little."}], |
| "triggers": [], "risk_flags": [], "evidence_used": [], "quant_tilt": {"bear": 0, "neutral": 0, "bull": 0}} |
| SUMMARY_V2 = {"summary": "The shares already price a demanding path, so the view is balanced.", |
| "thesis": [{"evidence": ["REL_FORECAST_EPS_CONSENSUS"], "view": "Earnings estimates have room to rise."}], |
| "risks": [{"evidence": ["REL_EXPECTATIONS_STORY"], "view": "The price requires more than the story delivers."}, |
| {"evidence": ["PE_FORWARD"], "view": "The multiple could compress."}]} |
|
|
|
|
| def _parsed(obj): |
| return PM._parse_llm_response(json.dumps(obj), 1, 1, "fake") |
|
|
|
|
| def _all_ok(prompt): |
| ids = [ln.split('"id": "')[1].split('"')[0] for ln in prompt.splitlines() if ln.startswith('{"id"')] |
| return {"verdicts": [{"id": i, "verdict": "OK", "reason": ""} for i in ids]} |
|
|
|
|
| def run_debate(monkeypatch, gate=_all_ok, seat=SEAT, reb=REB, pmr=PMR, raise_in=None, repair=None): |
| r = forecast_run() |
| for k in ("GEMINI_API_KEY", "KIMI_API_KEY", "OPENAI_API_KEY", "ANTHROPIC_API_KEY"): |
| monkeypatch.setattr(PY, k, "x") |
| monkeypatch.setattr(PY, "_call_gemini_3_pro", lambda p, system=None, verbose=False: _parsed(seat["neutral"])) |
| monkeypatch.setattr(PY, "_call_kimi", lambda p, system=None, verbose=False: _parsed( |
| (repair or {}).get("bear") if (repair and "=== REPAIR" in p and "answering the BULL" not in p) else |
| reb["bear"] if "answering the BULL" in p else seat["bear"])) |
| monkeypatch.setattr(PY, "_call_openai", lambda p, system=None, verbose=False: _parsed(reb["bull"] if "answering the BEAR" in p |
| else seat["bull"])) |
| monkeypatch.setattr(PY, "_call_claude", lambda p, system=None, verbose=False: _parsed(seat["neutral"])) |
| monkeypatch.setattr(PY, "_call_claude_yearly", lambda p, system=None, verbose=False: _parsed(seat["neutral"])) |
| monkeypatch.setattr(PY, "_call_claude_synthesis", lambda p, verbose=False: PM._parse_pm_synthesis(json.dumps(pmr), 1, 1)) |
| if raise_in: |
| monkeypatch.setattr(A, raise_in, lambda *a, **k: (_ for _ in ()).throw(RuntimeError("boom"))) |
| res = PY.analyze_llm_debate_yearly("SPGI", 160.0, {}, {}, {}, {}, {}, data={}, state_mode=r.panel_inputs(gate_fn=gate)) |
| prose = r.scenarios_from_panel(res) |
| v = r.finalize(F.blend_for(r.state.value("scenarios.pm_return"))) |
| assert v.passed, [(c.id, c.detail) for c in v.blocking] |
| payload = r.compose(prose, summary_fn=lambda _p: copy.deepcopy(SUMMARY_V2), gate_fn=gate) |
| return r, res, payload |
|
|
|
|
| def test_structured_debate_publishes_only_validated_objects(monkeypatch): |
| r, res, payload = run_debate(monkeypatch) |
| assert json.dumps(res, default=str) |
| assert payload, [(c.id, c.detail) for c in r.report_audit.checks if c.status == "FAIL"] |
| ids = {c.id: c.status for c in r.report_audit.checks} |
| assert ids["STRUCTURED_ARGUMENTS"] == "PASS" |
| md = payload["markdown"] |
| for dropped in ("20x", "Margins far exceed", "Management conceded", "No such point", "No numbers here", |
| "The multiple could compress"): |
| assert dropped not in md, dropped |
| for kept in ("Estimates have room to move toward the engine's view.", "The price needs revenue or margins", |
| "Upgrades are plausible yet do not close the valuation gap.", "Partly conceded", |
| "The bear revises its return after the exchange"): |
| assert kept in md, kept |
| reasons = " ".join(x.reason for x in r.rejections) |
| for code in ("TOKEN_IN_VIEW", "WITHHELD_EVIDENCE", "ATTRIBUTION", "BAD_TARGET", "REBUTTAL_NO_NUMBERS", "ID_NOT_ALLOWED"): |
| assert code in reasons, code |
| assert r.arguments_qa and "neutral" in r.arguments_qa |
|
|
|
|
| def test_the_semantic_gate_removes_what_its_evidence_does_not_support(monkeypatch): |
| def gate(prompt): |
| v = _all_ok(prompt) |
| for x in v["verdicts"]: |
| if x["id"] == "bull_open:points:1": |
| x.update(verdict="CONTRADICTED", reason="the evidence shows the forecast below consensus", |
| quote="room to move toward the engine's view") |
| return v |
| r, _res, payload = run_debate(monkeypatch, gate=gate) |
| md = payload["markdown"] |
| assert "Estimates have room to move toward the engine's view." not in md |
| assert "Upgrades are plausible" not in md |
| assert any("SEMANTIC_CONTRADICTED" in x.reason for x in r.rejections) |
|
|
|
|
| def test_gate_unavailable_publishes_on_typed_validation_with_a_note(monkeypatch): |
| r, _res, payload = run_debate(monkeypatch, gate=lambda p: None) |
| assert payload and any("semantic gate unavailable" in n for n in r.notes) |
|
|
|
|
| def test_a_validation_crash_never_kills_the_debate(monkeypatch): |
| r, res, payload = run_debate(monkeypatch, raise_in="validate_rebuttal") |
| assert any("INTERNAL_ERROR" in d["reason"] for d in res.get("argument_drops") or []) |
| assert payload is not None |
|
|
|
|
| def test_malformed_replies_are_dropped_not_crashing(monkeypatch): |
| bad = copy.deepcopy(SEAT) |
| bad["bull"]["points"] = [None, 7, {"kind": {"x": 1}, "evidence": [{"a": 1}, 3, None], "view": ["list"]}, |
| {"kind": "EARNINGS", "evidence": "REL_FORECAST_EPS_CONSENSUS, [[BEAT_RATE]]", |
| "view": "Estimates have room to move toward the engine's view."}] |
| r, res, payload = run_debate(monkeypatch, seat=bad) |
| assert payload |
| assert not any("INTERNAL_ERROR" in str(d.get("reason")) for d in res.get("argument_drops") or []) |
| assert sum("SCHEMA" in x.reason or "KIND_UNKNOWN" in x.reason for x in r.rejections) >= 3 |
|
|
|
|
| def test_compose_safe_turns_a_crash_into_a_withheld_record(monkeypatch): |
| r = forecast_run() |
| monkeypatch.setattr(r, "compose", lambda *a, **k: (_ for _ in ()).throw(RuntimeError("render boom"))) |
| assert r.compose_safe({}) is None and any("compose failed" in n for n in r.notes) |
|
|
|
|
| |
| def test_pm_probabilities_as_strings_parse_and_prompts_show_numbers_unquoted(): |
| """Live: the PM returned "p_bear": "0.37" (the skeleton quoted "<float>"), np.clip raised, the synthesis failed.""" |
| import panel_prompts as PP |
| r = PM._parse_pm_synthesis(json.dumps({"p_bear": "0.37", "p_bull": "0.30", "p_neutral": "0.33", "conviction": "0.6", |
| "thesis": "x", "factors": []}), 1, 1) |
| assert r and r["p_bear"] == pytest.approx(0.37) and r["conviction"] == pytest.approx(0.6) |
| txt = PP.pm("SPGI", "", "", {}, "", "a", "none") |
| assert '"<float>"' not in txt and '"p_bear": <float>' in txt |
| assert '"expected_return_pct": <number' in PP.seat_r1("bear", "SPGI", "", "") |
|
|
|
|
| @pytest.mark.parametrize("view", [ |
| "The twelve-month underperformance leaves sentiment fragile into the next results.", |
| "The margin gap confirms pricing power across the core franchises.", |
| ]) |
| def test_live_false_positives_stay_publishable(ctx_run, view): |
| _r, ctx = ctx_run |
| assert A.text_reasons(ctx, view, max_words=30) == [], A.text_reasons(ctx, view, max_words=30) |
|
|
|
|
| def test_a_collective_judgement_of_the_assumptions_is_checked_against_their_typed_assessments(ctx_run): |
| """Live: "The price embeds heroic long-run assumptions …" was dropped. Two independent rubric reviewers judged the drop |
| correct (the typed assessments rate the maturity margin and terminal cost of capital NEUTRAL); a post-run relaxation |
| that scoped T7 to cited assumptions was measured wrong and reverted. Keep the collective check.""" |
| _r, ctx = ctx_run |
| t = "The price embeds heroic long-run assumptions that leave little room for a slowdown." |
| assert any(x.startswith("ASSESSMENT") for x in A.direction_reasons(ctx, t, [], A._own(None, None))) |
|
|
|
|
| |
| def test_a_thesis_rests_on_named_points_and_falls_with_them(ctx_run): |
| _r, ctx = ctx_run |
| own = A._own(-0.12, "SELL") |
| good = {"kind": "EXPECTATIONS", "evidence": ["REL_EXPECTATIONS_STORY"], "view": "The price needs more than the story delivers."} |
| bad = {"kind": "RISK", "evidence": ["DMD_MARGIN"], "view": "Margins could compress toward the industry."} |
| th = "Today's price asks for more than the story delivers." |
| a, d = A.validate_seat(ctx, "bear", {"structured": {"thesis": th, "thesis_rests_on": [2], "points": [bad, good]}}, own=own) |
| assert a["thesis"] == th and a["thesis_rests_on"] == [1] |
| a, d = A.validate_seat(ctx, "bear", {"structured": {"thesis": th, "thesis_rests_on": [1], "points": [bad, good]}}, own=own) |
| assert a["thesis"] is None and any("THESIS_ORPHANED" in x["reason"] for x in d) |
| a, d = A.validate_seat(ctx, "bear", {"structured": {"thesis": th, "points": [good]}}, own=own) |
| assert a["thesis"] is None and any("name the points" in x["reason"] for x in d) |
| |
| sets, drops = A.orphan_theses({"bear_open": {"thesis": th, "thesis_rests_on": [1], "points": []}}) |
| assert sets["bear_open"]["thesis"] is None and drops[0]["reason"].startswith("THESIS_ORPHANED") |
|
|
|
|
| def test_a_rebuttal_thesis_may_not_restate_the_opening(ctx_run): |
| _r, ctx = ctx_run |
| opp = {"points": [{"n": 1, "kind": "EARNINGS", "evidence": ["REL_FORECAST_EPS_CONSENSUS"], "view": "Estimates can rise."}]} |
| th = "Valuation demands too much while earnings quality is overstated." |
| reply = {"structured": {"thesis": th, "thesis_rests_on": [1], |
| "answers": [{"to": "P1", "verdict": "PARTLY", "evidence": [], "view": "Upgrades may already be priced."}]}} |
| a, d = A.validate_rebuttal(ctx, "bear", reply, own=A._own(-0.15, "SELL"), opponent=opp, opponent_role="bull_open", |
| r1_value=-0.15, r1_label="SELL", opening_thesis=th) |
| assert a["thesis"] is None and any("REPEATS_OPENING" in x["reason"] for x in d) |
| assert a["answers"] and a["answers"][0]["view"] == "Upgrades may already be priced." |
|
|
|
|
| def test_the_joint_case_rule_reads_the_evidence_whatever_the_kind(ctx_run): |
| _r, ctx = ctx_run |
| if not ctx.cases: |
| pytest.skip("no alternative-case frontier in the fixture") |
| a, b = (next(c for c in ids if c in ctx.offer) for ids in list(ctx.cases.values())[:2]) |
| for kind in ("RISK", "VALUATION", None): |
| assert any(x.startswith("EXPECTATIONS_JOINT_CASES") for x in A.evidence_reasons(ctx, [a, b], kind)), kind |
| |
| assert A.evidence_reasons(ctx, ["REL_EXPECTATIONS_STORY", a, b, "PE_FORWARD"], "VALUATION") == [] |
|
|
|
|
| def test_evidence_is_the_whole_offer_and_past_returns_are_not_offered(ctx_run): |
| _r, ctx = ctx_run |
| assert "RETURN_12M_PHRASE" not in ctx.offer and "RETURN_3M_PHRASE" not in ctx.offer |
| any_id = sorted(ctx.offer)[0] |
| for kind in A.KINDS: |
| assert not any(x.startswith("ID_NOT_ALLOWED") for x in A.evidence_reasons(ctx, [any_id], kind)), kind |
|
|
|
|
| def test_an_answer_is_checked_against_its_targets_evidence(ctx_run): |
| _r, ctx = ctx_run |
| ids = sorted(c for c in ctx.offer if c not in ctx.relations) |
| bad = next(((x, y) for x in ids for y in ids if x != y and (ctx.rules.pair_fn(x, y) or ("",))[0] == "INCOMPATIBLE" |
| and not A.benign_cocitation(ctx, x, y)), None) |
| if bad is None: |
| pytest.skip("no incompatible pair in the fixture") |
| assert A.cross_reasons(ctx, [bad[0]], [bad[1]])[0].startswith("PAIR_INCOMPATIBLE_WITH_TARGET") |
| assert A.cross_reasons(ctx, ["REL_PRICE_INTRINSIC"], [bad[1]]) == [] or True |
|
|
|
|
| def test_the_gate_judges_theses_after_their_points(monkeypatch): |
| seen = [] |
|
|
| def gate(prompt): |
| seen.append(prompt) |
| v = _all_ok(prompt) |
| for x in v["verdicts"]: |
| if x["id"] == "bear_open:points:1": |
| x.update(verdict="UNSUPPORTED", reason="not shown by the evidence", quote="the story does not reach") |
| return v |
| r, _res, payload = run_debate(monkeypatch, gate=gate) |
| |
| assert len(seen) == 4 and "THESIS" in seen[1] and "Judge ONLY the thesis text" in seen[1] |
| assert "THE REST OF THE REPORT" in seen[0] and "## Intrinsic value" in seen[0] and "## The analyst debate" not in seen[0] |
| assert "## What drives the headline" not in seen[0] |
| assert '"bear_open:' in seen[0] and '"pm:' not in seen[0] |
| assert '"summary:' in seen[2] and '"bear_open:' not in seen[2] and '"bull_open:' not in seen[2] |
| assert '"bear_open:thesis"' not in seen[1] |
| assert any(x.reason.startswith("THESIS_ORPHANED") and x.source == "bear_open.thesis" for x in r.rejections) |
| assert "Today's price asks for more than the story delivers." not in payload["markdown"] |
|
|
|
|
| def test_one_repair_turn_recovers_a_point_with_the_typed_reason(monkeypatch): |
| fixed = copy.deepcopy(SEAT["bear"]) |
| fixed["points"][1] = {"kind": "RISK", "evidence": ["STORY_FAILURE"], |
| "view": "A failure haircut is small, so a stumble would land fully on the shares."} |
| fixed["expected_return_pct"] = 40.0 |
| r, res, payload = run_debate(monkeypatch, repair={"bear": fixed}) |
| md = payload["markdown"] |
| assert "A failure haircut is small, so a stumble would land fully on the shares." in md |
| drops = res.get("argument_drops") or [] |
| assert any(d["source"].startswith("attempt1:bear.points") and d["reason"].startswith("REPAIR_REQUESTED: ") |
| and "ATTRIBUTION" in d["reason"] for d in drops) |
| assert abs(r.state.scenarios.get("bear")["return"] - (-0.10)) < 1e-9 |
|
|
|
|
| def test_the_pm_decision_is_always_stated_and_evidence_prints_in_full_once(monkeypatch): |
| r, _res, payload = run_debate(monkeypatch) |
| md = payload["markdown"] |
| assert "The portfolio manager weights the bear case at" in md |
| assert "The valuation bar outweighs the earnings upside" in md |
| full = str(C.build_catalog(r.state, include_deferred=True)["REL_FORECAST_EPS_CONSENSUS"].rendered) |
| full = full[0].upper() + full[1:] |
| argued = md[md.find("## In brief"):md.find("## Intrinsic value")] |
| assert argued.count(full) == 1, argued.count(full) |
| assert "The engine's EPS forecast for the next twelve months is " in argued |
| ids = {c.id: c.status for c in r.report_audit.checks} |
| assert ids["AUTHORED_ONCE"] == "PASS" and ids["NO_INTERNAL_IDENTIFIERS"] == "PASS" |
|
|
|
|
| def test_the_audit_blocks_identifiers_and_repeated_authored_text(monkeypatch): |
| import report_model as RM |
| r, _res, payload = run_debate(monkeypatch) |
| cat = C.build_catalog(r.state, include_deferred=True) |
| rep = RM.build_user_report(r.state, cat, {"summary": [], "thesis": [], "risks": [], "seats": {}, "stories": [], |
| "arguments": {}}, []) |
| rep.blocks.append(RM.Block("para", "debate", [RM.Fragment("If revenue_growth_next_fy falls, the view moves.")])) |
| rep.blocks.append(RM.Block("para", "debate", [RM._authored("Same words.")])) |
| rep.blocks.append(RM.Block("para", "debate", [RM._authored("Same words.")])) |
| md, html = RM.render_markdown(rep), RM.render_html(rep) |
| ids = {c.id: c.status for c in RM.audit(rep, r.state, md, html).checks} |
| assert ids["NO_INTERNAL_IDENTIFIERS"] == "FAIL" and ids["AUTHORED_ONCE"] == "FAIL" |
|
|
|
|
| |
| def test_a_legacy_reply_is_never_repaired(monkeypatch): |
| legacy = {"expected_return_pct": -12.0, "predicted_price": 141, "recommendation": "SELL", "confidence": 0.55, |
| "thesis": "The price asks too much.", "points": ["valuation is stretched", "growth may slow"]} |
| seat = dict(SEAT, bear=legacy) |
| r, res, payload = run_debate(monkeypatch, seat=seat, repair={"bear": SEAT["bear"]}) |
| drops = res.get("argument_drops") or [] |
| assert not any(str(d.get("source", "")).startswith("attempt1:bear") for d in drops) |
| assert "The price needs revenue or margins the story does not reach." not in payload["markdown"] |
|
|
|
|
| def test_a_repair_is_kept_only_when_no_class_shrinks(): |
| first = {"thesis": "T", "points": [{"n": 1}], "answers": []} |
| assert not A.repair_keeps(first, {"thesis": None, "points": [{"n": 1}, {"n": 2}]}) |
| assert not A.repair_keeps({"thesis": None, "points": [{"n": 1}, {"n": 2}, {"n": 3}]}, |
| {"thesis": "T", "points": [{"n": 1}, {"n": 2}]}) |
| assert A.repair_keeps(first, {"thesis": "T", "points": [{"n": 1}, {"n": 2}]}) |
| assert not A.repair_keeps(first, dict(first)) |
|
|
|
|
| def test_thesis_labels_resolve_in_their_own_namespace(): |
| answers = [{"n": 1, "src": 1, "to": "P2"}, {"n": 2, "src": 2, "to": "P3"}] |
| assert A.rests_on(["P2"], answers, own="A", by_target=True) == [1] |
| assert A.rests_on(["A2", 1], answers, own="A", by_target=True) == [2, 1] |
| factors = [{"n": 1, "src": 1}, {"n": 2, "src": 2}] |
| assert A.rests_on(["P2"], factors, own="F") == [] and A.rests_on(["F2", "1"], factors, own="F") == [2, 1] |
|
|
|
|
| def test_a_repeat_is_whole_tokens_not_a_substring(): |
| assert A.same_text("Growth is slowing.", "Growth is slowing.") |
| assert A.same_text("growth is slowing", "Valuation is rich and growth is slowing.") |
| assert not A.same_text("The bull shows margins hold, but growth is slowing faster than the price allows.", |
| "Growth is slowing.") |
| assert not A.same_text("The shares look richly valued only if margins stall.", "The shares look rich.") |
|
|
|
|
| def test_a_kind_needs_permitted_evidence_and_unfixable_drops_are_not_repaired(ctx_run): |
| _r, ctx = ctx_run |
| pts, d = A.validate_points(ctx, [{"kind": "OWNERSHIP", "evidence": ["REL_PRICE_INTRINSIC"], |
| "view": "Long-term holders keep adding despite the premium."}], |
| own=A._own(None, None), limit=4, source="bull.points") |
| assert not pts and "KIND_UNKNOWN" in d[0]["reason"] |
| assert "OWNERSHIP" not in A.evidence_menu(ctx).split("KINDS")[1].split(".")[0] |
| a, d = A.validate_rebuttal(ctx, "bull", {"structured": {"thesis": "x.", "answers": [{"to": "P1", "verdict": "FAILS"}]}}, |
| own=A._own(0.2, "BUY"), opponent={"points": []}, opponent_role="bear_open", |
| r1_value=0.2, r1_label="BUY") |
| assert d and d[0]["reason"].startswith("NO_TARGETS") and not A.repairable(d) |
| _p, d = A.validate_pm(ctx, {"structured": {"factors": [{"kind": "VALUATION", "evidence": ["PE_FORWARD"], "view": "x."}]}}, |
| own=A._own(0.05, "HOLD"), debate_ids=set()) |
| assert d and d[0]["reason"].startswith("NO_ALLOWED_EVIDENCE") and not A.repairable(d) |
|
|
|
|
| def test_orphan_pass_drops_answers_to_removed_points_and_the_theses_on_them(): |
| sets = {"bull_open": {"thesis": None, "points": [{"n": 2, "view": "v"}]}, |
| "bear": {"thesis": "Reply.", "thesis_rests_on": [1], |
| "answers": [{"n": 1, "to": "P1", "to_role": "bull_open", "view": "a"}]}} |
| out, drops = A.orphan_theses(sets) |
| assert out["bear"]["answers"] == [] and out["bear"]["thesis"] is None |
| assert [d["reason"].split(":")[0] for d in drops] == ["TARGET_REMOVED", "THESIS_ORPHANED"] |
|
|
|
|
| def test_identifiers_are_dropped_at_authoring_and_summary_items_are_its_own_words(ctx_run): |
| _r, ctx = ctx_run |
| assert any(x.startswith("ID_TEXT_IN_VIEW") for x in A.text_reasons(ctx, "The quant_price_model adds nothing here.", |
| max_words=30)) |
| s, d = A.validate_summary(ctx, {"summary": "The shares price a demanding path.", |
| "thesis": [{"evidence": ["PE_FORWARD"], "view": "Earnings estimates have room to rise."}]}, |
| own=A._own(0.05, "HOLD"), debate_ids={"PE_FORWARD"}, |
| debate_texts=["Earnings estimates have room to rise."]) |
| assert not s["items_thesis"] and any(x["reason"].startswith("COPIES_DEBATE") for x in d) |
| assert s["thesis"] is None |
|
|
|
|
| def test_guidance_rows_without_a_label_get_a_reader_label(): |
| import guidance as G |
| assert G.concept_label("REVENUE_GROWTH") == "Revenue growth" and G.concept_label("EPS_GROWTH") == "EPS growth" |
|
|
|
|
| def test_compact_relations_keep_their_condition(ctx_run): |
| r, ctx = ctx_run |
| import report_model as RM |
| R = RM._R(r.state, C.build_catalog(r.state, include_deferred=True)) |
| rels = RM._rels(r.state) |
| for rid in ("REL_BREAK_EVEN_REVENUE_STORY",): |
| if rid in R.cat: |
| full = "".join(f.text for f in A._relation(R, rid, R.cat, rels)) |
| assert "".join(f.text for f in A._relation(R, rid, R.cat, rels, compact=True)) == full |
| if "REL_BREAK_EVEN_MARGIN_STORY__CLAUSE" in R.cat: |
| txt = "".join(f.text for f in A._relation(R, "REL_BREAK_EVEN_MARGIN_STORY", R.cat, rels, compact=True)) |
| assert txt.startswith("The operating margin today's price needs at the story's revenue is "), txt |
|
|
|
|
| def test_a_gate_failure_in_the_second_pass_keeps_the_first_pass(monkeypatch): |
| calls = [] |
|
|
| def gate(prompt): |
| calls.append(prompt) |
| if "THESIS" in prompt.split("\n", 1)[0] or len(calls) > 1: |
| raise RuntimeError("gate down") |
| v = _all_ok(prompt) |
| for x in v["verdicts"]: |
| if x["id"] == "bull_open:points:1": |
| x.update(verdict="UNSUPPORTED", reason="no", quote="Estimates have room") |
| return v |
| r, _res, payload = run_debate(monkeypatch, gate=gate) |
| assert "Estimates have room to move toward the engine's view." not in payload["markdown"] |
| assert any("the last completed pass stands" in n for n in r.notes) |
|
|
|
|
| |
| def test_a_gate_verdict_must_quote_the_objects_own_words(): |
| sets = {"bull": {"thesis": "Valuation tension is real, but conservative expectations leave room for surprise.", |
| "thesis_rests_on": [1], "answers": [{"n": 1, "to": "P1", "to_role": "bear_open", "evidence": [], |
| "view": "A strong beat cadence supports rerating."}]}, |
| "bear_open": {"points": [{"n": 1, "view": "x", "evidence": []}]}} |
| un = [] |
| out, drops = A.apply_gate(sets, [{"id": "bull:thesis", "verdict": "UNSUPPORTED", "quote": "strong beat cadence", |
| "reason": "no beat data"}], phase="theses", unattributed=un) |
| assert out["bull"]["thesis"] and not drops and un == ["bull:thesis"] |
| out, drops = A.apply_gate(sets, [{"id": "bull:thesis", "verdict": "UNSUPPORTED", "quote": "conservative expectations", |
| "reason": "no support"}], phase="theses", unattributed=[]) |
| assert out["bull"]["thesis"] is None and drops[0]["reason"].startswith("SEMANTIC_UNSUPPORTED") |
|
|
|
|
| def test_the_thesis_pass_sees_its_points_evidence_not_their_words(ctx_run): |
| _r, ctx = ctx_run |
| sets = {"bull": {"thesis": "Earnings power supports the shares.", "thesis_rests_on": [1], |
| "answers": [{"n": 1, "to": "P1", "to_role": "bear_open", "evidence": ["REL_FORECAST_EPS_CONSENSUS"], |
| "view": "A strong beat cadence supports rerating."}]}} |
| objs = A.gate_objects(sets, ctx.cat, ctx.relations, phase="theses") |
| assert len(objs) == 1 and "strong beat cadence" not in json.dumps(objs) and objs[0]["evidence"] != "(no evidence)" |
|
|
|
|
| def test_the_frontier_pointer_is_omitted_when_the_unit_prints_its_cases(ctx_run): |
| r, ctx = ctx_run |
| import report_model as RM |
| R = RM._R(r.state, C.build_catalog(r.state, include_deferred=True)) |
| if "REL_EXPECTATIONS_STORY__COMPACT" not in R.cat or "REL_BREAK_EVEN_MARGIN_STORY" not in R.cat: |
| pytest.skip("no frontier in the fixture") |
| rels = RM._rels(r.state) |
| alone = "".join(f.text for f in A.render_evidence(R, ["REL_EXPECTATIONS_STORY"], r.state, R.cat, rels)) |
| both = "".join(f.text for f in A.render_evidence(R, ["REL_EXPECTATIONS_STORY", "REL_BREAK_EVEN_MARGIN_STORY"], |
| r.state, R.cat, rels)) |
| assert "valuation section" in alone and "valuation section" not in both |
|
|
|
|
| |
| def test_a_trigger_on_a_trailing_figure_that_includes_a_separated_business_is_not_published(): |
| import state_builder as SB |
| import test_spgi_relations as TRr |
| run, _payload = TRr.spgi_run()[:2] |
| s = run.state |
| if not ((s.historical.get("perimeter") or {}).get("event") or {}).get("business"): |
| pytest.skip("no perimeter event in the fixture") |
| assert SB.trigger_perimeter_mismatch(s, "OPERATING_MARGIN_TTM") |
| assert not SB.trigger_perimeter_mismatch(s, "SMA200") and not SB.trigger_perimeter_mismatch(s, "LIFECYCLE_FORWARD_GROWTH") |
| assert SB._typed_trigger({"metric_id": "trailing operating margin", "condition_operator": "<=", "threshold": "44%", |
| "source_claim": "OPERATING_MARGIN_TTM", "scenario_destination": "BEAR"}, s) is None |
|
|
|
|
| |
| def test_the_gate_call_reasons_before_answering(monkeypatch): |
| import types |
| import brief as B |
| import llm_compat |
| seen = {} |
|
|
| class _Msg: |
| content = [types.SimpleNamespace(type="thinking", thinking="..."), |
| types.SimpleNamespace(type="text", text='{"verdicts": [{"id": "a", "verdict": "OK"}]}')] |
| fake = types.ModuleType("anthropic") |
| fake.Anthropic = lambda **kw: seen.setdefault("client", kw) |
| monkeypatch.setitem(sys.modules, "anthropic", fake) |
| monkeypatch.setattr(llm_compat, "claude_create", lambda client, **kw: seen.setdefault("call", kw) and _Msg()) |
| monkeypatch.setenv("ANTHROPIC_API_KEY", "x") |
| assert B.call_gate("prompt") == {"verdicts": [{"id": "a", "verdict": "OK"}]} |
| |
| assert seen["call"]["thinking"] == {"type": "adaptive"} and B.LAST_GATE_MODE == "claude-adaptive-thinking" |
| assert seen["client"]["max_retries"] <= 1 |
| assert "thinking" in llm_compat._DROPPABLE |
|
|
|
|
| def test_a_rejected_thinking_configuration_retries_claude_without_it(monkeypatch): |
| import types |
| import brief as B |
| import llm_compat |
| calls = [] |
|
|
| class _Msg: |
| content = [types.SimpleNamespace(type="text", text='{"verdicts": []}')] |
|
|
| def create(client, **kw): |
| calls.append(kw) |
| if "thinking" in kw: |
| raise RuntimeError('Error code: 400 - "thinking.type.enabled" is not supported for this model') |
| return _Msg() |
| fake = types.ModuleType("anthropic") |
| fake.Anthropic = lambda **kw: object() |
| monkeypatch.setitem(sys.modules, "anthropic", fake) |
| monkeypatch.setattr(llm_compat, "claude_create", create) |
| monkeypatch.setenv("ANTHROPIC_API_KEY", "x") |
| monkeypatch.delenv("OPENAI_API_KEY", raising=False) |
| assert B.call_gate("p") == {"verdicts": []} and len(calls) == 2 and "thinking" not in calls[1] |
| assert B.LAST_GATE_MODE == "claude-no-thinking" |
|
|
|
|
| def test_the_calibrated_rubric_names_the_measured_defect_classes(): |
| for phrase in ("every table in the report", "price series", "model's own assumptions", "hedged"): |
| assert phrase in A.GATE_RUBRIC, phrase |
| assert "Judge ONLY the thesis text" in A.GATE_THESIS_RUBRIC and "quote" in A.GATE_THESIS_RUBRIC |
|
|
|
|
| |
| def _spgi_like_ctx(op_margin=0.49417): |
| return dict(ticker="X", sector="FINANCIAL SERVICES", industry="FINANCIAL DATA & STOCK EXCHANGES", revenue=14.31e9, |
| op_margin=op_margin, price=403.3, shares=3.0e8, debt=12e9, cash=1.5e9, rf=0.042, erp=0.041, beta=0.36, |
| beta_stable=0.9, stage="mature-growth", damodaran={"industry_sales_to_capital": 0.07, "erp": 0.041}) |
|
|
|
|
| def test_a_margin_floor_never_cuts_a_target_and_the_break_even_keeps_the_adopted_margin(): |
| """Live SPGI: the break-even was solved at 45% (the sector cap) while the adopted story used 49.4%: the rounded |
| adopted margin (0.494) sat a hair under today's 0.4942 and the 'floor' cut it to the cap.""" |
| import story_dcf as sd |
| A0 = {"tam_at_maturity_usd_b": 180.0, "plausible_peak_share": 0.131, "years_to_maturity": 10, |
| "target_operating_margin": 0.494, "sales_to_capital": 1.16, "failure_probability": 0.01} |
| s = sd.story_fcff(_spgi_like_ctx(), A0, strict=True) |
| assert s["assumptions"]["margin_target"] >= 0.494 |
| be = sd.story_breakeven(_spgi_like_ctx(), A0) |
| assert be["story_margin"] == 0.494 and abs(be["story_value"] / s["value"] - 1) < 0.02 |
|
|
|
|
| def test_market_size_times_share_reproduces_the_revenue_at_the_printed_precision(): |
| import story_dcf as sd |
| A0 = {"segments": [{"name": "a", "tam_at_maturity_usd_b": 30, "plausible_peak_share": 0.228, "target_operating_margin": 0.30}, |
| {"name": "b", "tam_at_maturity_usd_b": 52, "plausible_peak_share": 0.2927, "target_operating_margin": 0.62}], |
| "years_to_maturity": 10, "sales_to_capital": 1.5, "failure_probability": 0.01} |
| import business as BZ |
| ctx = dict(_spgi_like_ctx(), business_decomposition=BZ.declared("X", ["a", "b"], "test")) |
| a = sd.story_fcff(ctx, A0, strict=True)["assumptions"] |
| assert abs(round(a["tam_usd_b"], 2) * round(a["peak_share"], 4) - a["revenue_N_usd_b"]) < 0.01 |
| assert abs(sum(r["rev_usd_b"] for r in a["segments"]) - a["revenue_N_usd_b"]) < 0.01 |
|
|
|
|
| def test_insider_counts_are_counts_of_transactions(): |
| import insider as I |
| txt = I.characterization({"by_class": {I.TAX_WITHHOLDING: {"n": 2}, I.OPTION_EXERCISE: {"n": 1}}}) |
| assert "2 tax-withholding transactions" in txt and "1 option exercise (" in txt and "shares withheld" not in txt |
|
|
|
|
| def test_filer_names_keep_their_legal_form(): |
| import state_builder as SB |
| assert SB.filer_display_name("Egerton Capital Uk Llp") == "Egerton Capital UK LLP" |
| assert SB.filer_display_name("Co Capital Partners") == "Co Capital Partners" |
|
|
|
|
| def test_a_pricing_multiple_names_whose_figure_it_is_applied_to(): |
| import report_model as RM |
| b = RM._method_basis({"raw_input_basis": "NTM adjusted EPS (ReconciledForecastState.eps)", "multiple_basis": "TRAILING_ADJUSTED"}) |
| assert b == " on the engine's next-12-month adjusted EPS (multiple set on a trailing basis)" |
| assert RM._method_basis({"raw_input_basis": "mean sell-side 12-month target"}) == "" |
|
|
|
|
| def test_price_derived_multiples_are_not_trigger_sources(): |
| import state_builder as SB |
| import test_spgi_relations as TRr |
| run = TRr.spgi_run()[0] |
| assert "price-derived multiple" in (SB.trigger_source_refusal(run.state, "PE_FORWARD") or "") |
| assert SB.trigger_source_refusal(run.state, "SMA200") is None |
|
|
|
|
| def test_co_citing_incompatible_facts_is_allowed_only_when_it_cannot_read_as_a_comparison(): |
| import test_spgi_relations as TRr |
| run = TRr.spgi_run()[0] |
| cat = C.build_catalog(run.state, include_deferred=False) |
| ctx = run.args_context(cat) |
| checked = 0 |
| for a, b, ok in (("OPERATING_MARGIN_TTM", "NET_MARGIN_TTM", True), ("EPS_ADJUSTED_TTM", "EPS_GAAP_TTM", False), |
| ("CONSENSUS_EPS_FY1", "CONSENSUS_EPS_FY2", True), ("FORECAST_REVENUE", "CONSENSUS_REVENUE_FY1", False)): |
| if a in ctx.offer and b in ctx.offer and (ctx.rules.pair_fn(a, b) or ("",))[0] == "INCOMPATIBLE": |
| assert A.benign_cocitation(ctx, a, b) is ok, (a, b) |
| checked += 1 |
| assert checked >= 1 |
|
|
