"""BSX root-cause remediation (run 1953a5992ac21d82): the segment "Cardiovascular" was mapped to Financial Svcs. (Non-bank & Insurance) because the keyword "card" had no closing word boundary; that one industry row then supplied both the EV/Sales value weight and the unlevered beta. Also: one template sentence named consensus next-fiscal-year growth as generic "forward revenue growth", and the portfolio manager weighed the Bear before the semantic gate removed every Bear argument. BSX-shaped inputs (its two filed segments and its company industry); nothing here names BSX in engine code. """ import copy import json import os import re 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 # noqa: E402 import damodaran_data as dd # noqa: E402 import pipeline_fixtures as F # noqa: E402 import segments as SG # noqa: E402 import test_arguments as TA # noqa: E402 (the debate harness with fake providers) PARENT = "Healthcare Products" FIN = "Financial Svcs. (Non-bank & Insurance)" OLD_OPEN_PATTERN = r"\b(payments?|card)" # the pattern as it was: no closing boundary def bsx_canon(): segs = [{"segment_id": "bsx:CardiovascularMember", "name": "Cardiovascular", "revenue": 7.126e9, "revenue_share": 0.669, "kind": "business_line", "product_lines": []}, {"segment_id": "bsx:MedsurgMember", "name": "Medsurg", "revenue": 3.519e9, "revenue_share": 0.331, "kind": "business_line", "product_lines": []}] return {"status": "RECONCILED", "segments": segs, "consolidated_revenue": 10.646e9, "coverage": 1.0, "excluded_revenue": 0.0, "business_mix_allowed": True, "reason": "reconciled"} def mapper(sid, text): return dd.map_segment(sid, text, (), PARENT, False, use_product_lines=False) def forged_mapping(status): """The mapping the old table produced, recorded with the given status.""" return {"segment_id": "bsx:CardiovascularMember", "input_text": "Cardiovascular", "selected_industry": FIN, "mapping_method": f"segment name matches {FIN}", "matched_pattern": OLD_OPEN_PATTERN, "matched_text": "card", "parent_industry": PARENT, "compatibility_status": status, "fallback_used": None, "eligible_for_value_weight": True, "eligible_for_beta": True} # ── A: token collision ── def test_A_keywords_match_whole_tokens_only(): assert re.search(OLD_OPEN_PATTERN, "cardiovascular") # the defect, as it was card = next(rx for rx, ind in dd.SEGMENT_INDUSTRY if ind == FIN) assert not re.search(card, "cardiovascular") for name in ("Cardiovascular", "Cardiology", "Interventional Cardiology", "Gastroenterology", "Powertrain", "Machine Vision", "Golden Nugget", "Medsurg"): assert dd.segment_industry(name, (), None)[0] is None, name # a real token still maps, and declared stems still match the words that begin with them for name, ind in (("Card Services", FIN), ("Payments", FIN), ("Oil and Gas", "Oil/Gas (Integrated)"), ("Power", "Power"), ("Pharmaceuticals", "Drugs (Pharmaceutical)"), ("Telecommunications", "Telecom. Services"), ("Advertising services", "Advertising"), ("Intelligent Cloud", "Software (System & Application)"), ("Online stores", "Retail (General)"), ("MedTech", "Healthcare Products")): assert dd.segment_industry(name, (), None)[0] == ind, name # every pattern in the table is bounded on both sides (stems are explicit: a stem ends in \w* inside the group) for rx, _ind in dd.SEGMENT_INDUSTRY: assert rx.startswith(r"\b(?:") and rx.endswith(r")\b"), rx # ── B: parent-sector conflict ── def test_B_a_nonfinancial_parents_segment_is_not_financial_without_evidence(monkeypatch): ok, why = dd.industry_compatible(PARENT, False, FIN, "Cardiovascular") assert not ok and "non-financial company" in why assert dd.industry_compatible(PARENT, False, FIN, "Payments")[0] # a financial business named in its own text assert dd.industry_compatible("Bank (Money Center)", True, FIN, "Cards")[0] # a financial parent is not restricted assert dd.industry_compatible(PARENT, False, "Healthcare Products", "Medsurg")[0] # the gate stands even if the keyword table regresses to the open pattern monkeypatch.setattr(dd, "SEGMENT_INDUSTRY", [(OLD_OPEN_PATTERN, FIN)]) mp = mapper("bsx:CardiovascularMember", "Cardiovascular") assert [r["compatibility_status"] for r in mp["rejected"]] == ["REJECTED"] and mp["rejected"][0]["industry"] == FIN assert mp["selected_industry"] == PARENT and mp["compatibility_status"] == "PASS" assert mp["fallback_used"] == "COMPANY_INDUSTRY" and "rejected as incompatible" in mp["mapping_method"] # ── C, D: a rejected mapping may supply neither the EV/Sales weight nor the beta ── @pytest.mark.parametrize("use", ["value_weight", "beta"]) @pytest.mark.parametrize("status", ["REJECTED", "PASS"]) # a record that merely says PASS is re-checked def test_CD_an_incompatible_mapping_cannot_supply_an_industry_row(use, status): with pytest.raises(dd.IncompatibleMapping): dd.industry_row(forged_mapping(status), "US", use) assert dd.industry_row(mapper("bsx:MedsurgMember", "Medsurg"), "US", use).get("unlevered_beta_cash") def test_CD_a_mix_over_an_incompatible_mapping_is_withheld_and_blocks_the_forecast_stage(): canon = bsx_canon() canon["segments"][0]["allocations"] = [{"industry": FIN, "fraction": 1.0, "basis": "segment name", "mapping": forged_mapping("PASS")}] canon["segments"][1]["allocations"] = [{"industry": PARENT, "fraction": 1.0, "basis": "company industry", "mapping": mapper("bsx:MedsurgMember", "Medsurg")}] assert dd.business_mix(canon["segments"], "US", PARENT, 12e9, 61e9) is None # no precision blend over it # the rows of the published run (Financial Svcs. 0.33 at 88.9%; Healthcare Products 0.86 at 11.1%) rows = [{"name": "Cardiovascular", "segment_id": "bsx:CardiovascularMember", "industry": FIN, "weight": 0.889, "unlevered_beta": 0.328}, {"name": "Medsurg", "segment_id": "bsx:MedsurgMember", "industry": PARENT, "weight": 0.111, "unlevered_beta": 0.857}] assert dd.business_mix_problems(rows, PARENT, False) bq = dd.beta_quality({"industry": PARENT, "industry_match": "ticker NYSE:BSX", "beta": 0.44, "business_mix": rows}, False) assert bq["status"] == "LOW" and bq["business_mix_beta_valid"] is False assert any("segment mapping incompatible" in r for r in bq["reasons"]) # S5: such a mix in the state fails the forecast stage d = F.all_data() d["segments_canonical"] = canon d["damodaran_inputs"] = dict(d["damodaran_inputs"], industry=PARENT, business_mix=rows) import pipeline as pl r = pl.Run("BSX", run_id="BSX-s5") assert r.data_stage(d, 160.0, price_as_of="2026-09-21", tech={"signals": {}}).passed fi = F.legacy_forecast_inputs(); fi["all_data"] = d v = r.forecast_stage(**fi) s5 = next(c for c in v.checks if c.id == "S5") assert not v.passed and s5.status == "FAIL" and "incompatible segment mapping" in s5.detail # ── E: fallback ── def test_E_the_bsx_segments_fall_back_to_the_company_industry(): canon = SG.allocate(bsx_canon(), PARENT, None, mapper=mapper) cardio, medsurg = canon["segments"] mp = cardio["industry_mapping"] assert mp["selected_industry"] == PARENT and mp["mapping_method"] == "no keyword match: company industry" assert mp["compatibility_status"] == "PASS" and mp["fallback_used"] == "COMPANY_INDUSTRY" and mp["rejected"] == [] assert mp["eligible_for_value_weight"] and mp["eligible_for_beta"] assert [a["industry"] for g in canon["segments"] for a in g["allocations"]] == [PARENT, PARENT] assert cardio["allocations"][0]["mapping"] is mp # one business in two segments: no business-mix beta, the company's single-industry beta applies assert dd.business_mix(canon["segments"], "US", PARENT, 12e9, 61e9) is None # with no company industry either, the mapping is unavailable and feeds nothing none = dd.map_segment("bsx:CardiovascularMember", "Cardiovascular", (), None, False) assert none["selected_industry"] is None and none["compatibility_status"] == "UNAVAILABLE" assert not none["eligible_for_value_weight"] and not none["eligible_for_beta"] with pytest.raises(dd.IncompatibleMapping): dd.industry_row(none, "US", "beta") # ── F: the two growth metrics keep distinct labels ── def test_F_engine_and_consensus_growth_never_share_a_generic_label(): run, payload, failed = F.run_pipeline() assert failed is None md = payload["markdown"] eng, con = A._READER["FORECAST_REVENUE_GROWTH"][0], A._READER["LIFECYCLE_FORWARD_GROWTH"][0] assert "engine" in eng and "next twelve months" in eng and "consensus" in con and "next fiscal year" in con assert not re.search(r"forward revenue growth of", md) # the generic label is gone if "LIFECYCLE_FORWARD_GROWTH" in payload["claims"]: v = payload["claims"]["LIFECYCLE_FORWARD_GROWTH"]["rendered"] assert f"consensus revenue growth for the next fiscal year of {v}" in md # ── G: the portfolio manager sees each seat's evidence state before it sets the probabilities ── def test_G_the_pm_is_told_a_seat_has_no_accepted_arguments(monkeypatch): prompts = [] def gate(prompt): # every Bear object fails semantic validation out = [] for ln in prompt.splitlines(): if ln.startswith('{"id"'): o = json.loads(ln) bad = o["id"].startswith(("bear_open:points", "bear:answers")) out.append({"id": o["id"], "verdict": "UNSUPPORTED" if bad else "OK", "reason": "not shown by its evidence", "quote": o["text"][-25:] if bad else None}) return {"verdicts": out} pmr = copy.deepcopy(TA.PMR) pmr["factors"] = [{"kind": "EARNINGS", "evidence": ["REL_FORECAST_EPS_CONSENSUS"], "view": "Earnings support holds the middle."}] r = TA.forecast_run() for k in ("GEMINI_API_KEY", "KIMI_API_KEY", "OPENAI_API_KEY", "ANTHROPIC_API_KEY"): monkeypatch.setattr(TA.PY, k, "x") seat, reb = TA.SEAT, TA.REB monkeypatch.setattr(TA.PY, "_call_gemini_3_pro", lambda p, system=None, verbose=False: TA._parsed(seat["neutral"])) monkeypatch.setattr(TA.PY, "_call_kimi", lambda p, system=None, verbose=False: TA._parsed( reb["bear"] if "answering the BULL" in p else seat["bear"])) monkeypatch.setattr(TA.PY, "_call_openai", lambda p, system=None, verbose=False: TA._parsed( reb["bull"] if "answering the BEAR" in p else seat["bull"])) monkeypatch.setattr(TA.PY, "_call_claude", lambda p, system=None, verbose=False: TA._parsed(seat["neutral"])) monkeypatch.setattr(TA.PY, "_call_claude_yearly", lambda p, system=None, verbose=False: TA._parsed(seat["neutral"])) def pm_call(p, verbose=False): prompts.append(p) return TA.PM._parse_pm_synthesis(json.dumps(pmr), 1, 1) monkeypatch.setattr(TA.PY, "_call_claude_synthesis", pm_call) res = TA.PY.analyze_llm_debate_yearly("BSX", 160.0, {}, {}, {}, {}, {}, data={}, state_mode=r.panel_inputs(gate_fn=gate)) # the PM's prompt: written AFTER the gate, it carries no Bear argument and says so explicitly pm_prompt = prompts[0] assert "Bear seat has no accepted supporting arguments" in pm_prompt assert "The price needs revenue or margins the story does not reach." not in pm_prompt assert res["pm_evidence_state"]["bear"]["accepted"] == 0 and res["pm_evidence_state"]["bear"]["rejected"] >= 1 assert res["pm_evidence_state"]["bull"]["accepted"] >= 1 # the probability is still the PM's (nothing zeroed, capped or renormalized) assert res["p_bear"] == pytest.approx(pmr["p_bear"]) prose = r.scenarios_from_panel(res) assert r.state.scenarios.get("pm_evidence_state")["bear"]["accepted"] == 0 assert r.state.scenarios.get("bear")["probability"] == pytest.approx(pmr["p_bear"]) v = r.finalize(F.blend_for(r.state.value("scenarios.pm_return"))) assert v.passed, [(c.id, c.detail) for c in v.blocking] # compose: the seats' verdicts are final (a gate that would now reject everything removes no seat argument) n_bull = A.supporting_count(prose["bull_open"]["arguments"]) payload = r.compose(prose, summary_fn=lambda _p: copy.deepcopy(TA.SUMMARY_V2), gate_fn=lambda p: {"verdicts": [{"id": json.loads(ln)["id"], "verdict": "OK"} for ln in p.splitlines() if ln.startswith('{"id"')]}) assert payload, [(c.id, c.detail) for c in r.report_audit.checks if c.status == "FAIL"] assert A.supporting_count(r.arguments_qa["bull_open"]["set"]) == n_bull md = payload["markdown"] assert ("The scenario return is the analyst's own; its probability was assigned by the portfolio manager, who was told " "this seat had no accepted supporting arguments.") in md assert "its weight above are the analyst's own" not in md