Spaces:

s-ttp
/

crucible

Running

crucible / engine /tests /test_withheld_fixes.py

Withheld-run fixes: recorded margin floor; re-blend bounded by the recorded cap; precise break-even revenue and sign-aware growth clause; pre-revenue message; no retry invitation for deterministic failures

7e513ed verified about 1 month ago

Raw History Blame Contribute Delete

9.51 kB

"""The four withheld-run causes of 2026-09-28 … 10-01 (NVDA, PGY, NNE, and the retry wording), each with the generic fix:
NVDA  a floor raised the segments' blended margin (53.5% → 60%) but the segment check knew only a cap;
PGY   a GAAP re-blend after the sanity cap left the composite above its recorded bound (S6), and the break-even revenue
      was stored more coarsely than the growth rate computed from it (RELATION_MAGNITUDE); a negative rate also rendered
      as "requires … growth";
NNE   a pre-revenue company fails the forecast checks on every run and was told to "try again later";
all   deterministic failures invited a retry.
"""
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 business as BZ                    # noqa: E402
import pipeline_fixtures as F            # noqa: E402
import relations as RL                   # noqa: E402
import report_model as RM                # noqa: E402
import story_dcf as SD                   # noqa: E402
import test_pipeline as TP               # noqa: E402  (_canon_two: two reconciled filed segments)


# ── NVDA: the margin floor is recorded, accepted by the segment check and disclosed ──
def _nvda_ctx():
    return {"ticker": "NVDA", "revenue": 187e9, "shares": 24.3e9, "price": 180.0, "op_margin": 0.652, "rf": 0.0524, "erp": 0.0409,
            "beta": 1.6, "debt": 10e9, "cash": 60e9, "market_cap": 4.4e12, "stage": "high-growth", "interest_coverage": 300,
            "sector": "TECHNOLOGY", "industry": "SEMICONDUCTORS",
            "business_decomposition": BZ.declared("NVDA", ["Compute And Networking", "Graphics"], "test")}


NVDA_A = {"tam_at_maturity_usd_b": 2670.0, "plausible_peak_share": 0.43, "years_to_maturity": 10, "target_operating_margin": 0.535,
          "sales_to_capital": 2.25, "failure_probability": 0.0,
          "segments": [{"name": "Compute And Networking", "tam_at_maturity_usd_b": 2450.0, "plausible_peak_share": 0.44,
                        "target_operating_margin": 0.55, "sales_to_capital": 2.3},
                       {"name": "Graphics", "tam_at_maturity_usd_b": 220.0, "plausible_peak_share": 0.35,
                        "target_operating_margin": 0.32, "sales_to_capital": 1.8}]}


def test_nvda_the_margin_floor_is_recorded():
    a = SD.story_fcff(_nvda_ctx(), copy.deepcopy(NVDA_A))["assumptions"]
    assert a["margin_proposed"] == pytest.approx(0.535, abs=0.001) and a["margin_target"] == 0.6 and a["margin_floor"] == 0.6
    # a storyteller margin above the floor records none
    hi = copy.deepcopy(NVDA_A)
    for g in hi["segments"]:
        g["target_operating_margin"] = 0.62
    assert SD.story_fcff(_nvda_ctx(), hi)["assumptions"]["margin_floor"] is None


def _floored_story(floor):
    st = F.story()
    st["assumptions"].update(segments=[{"name": "Generation", "rev_usd_b": 70.0, "margin": 0.55, "sales_to_capital": 2.3},
                                       {"name": "Retail", "rev_usd_b": 30.0, "margin": 0.32, "sales_to_capital": 1.8}],
                             revenue_N_usd_b=100.0, margin_proposed=0.481, margin_target=0.6, margin_cap=0.652, margin_floor=floor)
    return st


def test_nvda_a_recorded_floor_passes_the_segment_check_and_is_disclosed():
    d = F.all_data(); d["segments_canonical"] = TP._canon_two()
    run, payload, failed = F.run_pipeline(data=d, forecast_over={"story": _floored_story(0.6)})
    chk = next(c for v in run.validations for c in v.checks if c.id == "SEGMENT_BUILD_RECONCILES")
    assert failed is None and chk.status == "PASS", (failed, chk.detail)
    md = payload["markdown"]
    assert "is raised to 60.0%" in md and "best sustained margin in its sector" in md
    # the same difference with no recorded floor is unexplained and still blocks
    run, payload, failed = F.run_pipeline(data=d, forecast_over={"story": _floored_story(None)})
    chk = next(c for v in run.validations for c in v.checks if c.id == "SEGMENT_BUILD_RECONCILES")
    assert failed == "FORECAST" and chk.status == "FAIL" and "nor a recorded floor" in chk.detail


# ── PGY: a later composite step stays inside the recorded cap ──
def test_pgy_the_gaap_reblend_is_bounded_by_the_recorded_cap():
    import predict_yearly as PY
    price = 17.62
    fv = {"methods": {"P/E": {"price": 54.2, "conf": 0.54}, "PEG": {"price": 94.0, "conf": 0.43}, "P/FCF": {"price": 42.0, "conf": 0.38},
                      "P/S": {"price": 45.4, "conf": 0.36}, "Analyst Target": {"price": 24.0, "conf": 0.40}, "DCF": {"price": 75.4, "conf": 0.49}},
          "fair_price": 44.05, "composite_audit": {"cap_upper": 2.5, "cap_lower": 0.78, "methods": [], "final": 44.05,
                                                   "steps": [{"step": "sanity cap", "set_to": 44.05, "note": "FV_CAPPED_HIGH"}]}}
    PY._gaap_reblend(fv, {"gaap_distortion": True, "distortion_reason": "reported EPS far below adjusted"}, price)
    steps = fv["composite_audit"]["steps"]
    assert [s["step"] for s in steps] == ["sanity cap", "GAAP-distortion re-blend", "sanity cap"]
    assert steps[1]["set_to"] > 2.5 * price                       # the re-blend alone would have left the bound
    assert fv["fair_price"] == pytest.approx(2.5 * price) == pytest.approx(fv["composite_audit"]["final"]) == pytest.approx(steps[-1]["set_to"])
    # a re-blend inside the bound adds no cap step
    fv2 = copy.deepcopy(fv); fv2["composite_audit"]["steps"] = fv2["composite_audit"]["steps"][:1]
    for m in fv2["methods"].values():
        m["price"] = 30.0
    PY._gaap_reblend(fv2, {"gaap_distortion": True, "distortion_reason": "x"}, price)
    assert [s["step"] for s in fv2["composite_audit"]["steps"]] == ["sanity cap", "GAAP-distortion re-blend"]


# ── PGY: the break-even revenue keeps the precision of its growth rate; a shrinking path is not "growth" ──
def test_pgy_break_even_revenue_and_its_growth_reproduce_each_other():
    ctx = {"revenue": 1.325553e9, "shares": 77e6, "price": 17.62, "op_margin": 0.212, "rf": 0.0524, "erp": 0.0409, "beta": 1.5,
           "debt": 0.7e9, "cash": 0.3e9, "market_cap": 1.36e9, "stage": "mature-growth", "interest_coverage": 5,
           "sector": "FINANCIAL SERVICES"}
    A = {"tam_at_maturity_usd_b": 51.0, "plausible_peak_share": 0.08, "years_to_maturity": 8, "target_operating_margin": 0.256,
         "sales_to_capital": 2.0, "failure_probability": 0.0}
    base = SD.story_fcff(ctx, A, strict=False)["value"]
    ctx["price"] = 0.75 * base                                   # a price the solve brackets
    be = SD.story_breakeven(ctx, A)
    rev, g = be["revenue_needed_N_usd_b"], be["rev_cagr_needed"]
    assert rev != round(rev, 2)                                  # not rounded to the printed two decimals
    assert abs((rev / 1.325553) ** (1 / 8) - 1 - g) < 1e-4        # the relation recomputes the stored rate


def test_a_negative_growth_rate_is_rendered_as_shrinkage():
    import catalog as C
    spec = RL.Spec("REL_BREAK_EVEN_REVENUE_GROWTH", "BE_REVENUE", "STORY_START_REVENUE", "path", RL.CAGR, RL.GROWTH_FROM,
                   RL.GROWTH_FROM, "BE_REVENUE_CAGR", lhs_words="the revenue today's price needs at maturity",
                   rhs_words="the story's starting revenue")
    d = C.BY_ID["BE_REVENUE"]
    down = RL._render(spec, RL.GROWTH_FROM, -0.0612, 0.8e9, 1.3256e9, d, d, 8)
    up = RL._render(spec, RL.GROWTH_FROM, 0.092, 50.4e9, 21.0e9, d, d, 10)
    flat = RL._render(spec, RL.GROWTH_FROM, 0.0001, 1.3e9, 1.3e9, d, d, 8)
    assert "allows revenue to shrink 6.1% a year" in down and "requires" not in down
    assert "requires 9.2% annual revenue growth" in up and "requires no revenue growth" in flat


# ── NNE and the retry wording ──
def _patched_run(revenue, operating_income, financial=False):
    run, payload, failed = F.run_pipeline()
    s = run.state
    h = s.historical._data
    h["revenue_ttm"] = h["revenue_ttm"].with_value(revenue)
    h["operating_income_ttm"] = h["operating_income_ttm"].with_value(operating_income)
    s.issuer._data["is_financial"] = financial
    return run


def test_nne_a_pre_revenue_company_is_told_so(tmp_path):
    run = _patched_run(214042.0, -51.87e6)
    assert run.withheld_reason() == "PRE_REVENUE"
    out = run.write(tmp_path / "x.json")
    assert out["user"]["status"] == "WITHHELD" and out["user"]["reason"] == "PRE_REVENUE"
    assert "little or no revenue" in out["user"]["message"] and "try again" not in out["user"]["message"]
    assert _patched_run(0.0, -5e6).withheld_reason() == "PRE_REVENUE"
    assert _patched_run(19e9, 3.4e9).withheld_reason() is None              # an established business
    assert _patched_run(214042.0, -51.87e6, financial=True).withheld_reason() is None
    assert _patched_run(None, None).withheld_reason() is None              # no figures is a data failure, not this


def test_deterministic_failures_do_not_invite_a_retry():
    for stage in ("INITIAL", "DATA_VALIDATED"):
        m = RM.withheld_payload("X", stage_reached=stage)["message"]
        assert "try again later" not in m.lower() and "same result" in m
    for stage in ("FORECAST_VALIDATED", "ANALYSIS_FINALIZED"):              # these depend on the analyst models' answers
        assert "try again later" in RM.withheld_payload("X", stage_reached=stage)["message"].lower()
    assert "reason" not in RM.withheld_payload("X", stage_reached="DATA_VALIDATED")

	"""The four withheld-run causes of 2026-09-28 … 10-01 (NVDA, PGY, NNE, and the retry wording), each with the generic fix:

	NVDA a floor raised the segments' blended margin (53.5% → 60%) but the segment check knew only a cap;
	PGY a GAAP re-blend after the sanity cap left the composite above its recorded bound (S6), and the break-even revenue
	was stored more coarsely than the growth rate computed from it (RELATION_MAGNITUDE); a negative rate also rendered
	as "requires … growth";
	NNE a pre-revenue company fails the forecast checks on every run and was told to "try again later";
	all deterministic failures invited a retry.
	"""
	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 business as BZ # noqa: E402
	import pipeline_fixtures as F # noqa: E402
	import relations as RL # noqa: E402
	import report_model as RM # noqa: E402
	import story_dcf as SD # noqa: E402
	import test_pipeline as TP # noqa: E402 (_canon_two: two reconciled filed segments)


	# ── NVDA: the margin floor is recorded, accepted by the segment check and disclosed ──
	def _nvda_ctx():
	return {"ticker": "NVDA", "revenue": 187e9, "shares": 24.3e9, "price": 180.0, "op_margin": 0.652, "rf": 0.0524, "erp": 0.0409,
	"beta": 1.6, "debt": 10e9, "cash": 60e9, "market_cap": 4.4e12, "stage": "high-growth", "interest_coverage": 300,
	"sector": "TECHNOLOGY", "industry": "SEMICONDUCTORS",
	"business_decomposition": BZ.declared("NVDA", ["Compute And Networking", "Graphics"], "test")}


	NVDA_A = {"tam_at_maturity_usd_b": 2670.0, "plausible_peak_share": 0.43, "years_to_maturity": 10, "target_operating_margin": 0.535,
	"sales_to_capital": 2.25, "failure_probability": 0.0,
	"segments": [{"name": "Compute And Networking", "tam_at_maturity_usd_b": 2450.0, "plausible_peak_share": 0.44,
	"target_operating_margin": 0.55, "sales_to_capital": 2.3},
	{"name": "Graphics", "tam_at_maturity_usd_b": 220.0, "plausible_peak_share": 0.35,
	"target_operating_margin": 0.32, "sales_to_capital": 1.8}]}


	def test_nvda_the_margin_floor_is_recorded():
	a = SD.story_fcff(_nvda_ctx(), copy.deepcopy(NVDA_A))["assumptions"]
	assert a["margin_proposed"] == pytest.approx(0.535, abs=0.001) and a["margin_target"] == 0.6 and a["margin_floor"] == 0.6
	# a storyteller margin above the floor records none
	hi = copy.deepcopy(NVDA_A)
	for g in hi["segments"]:
	g["target_operating_margin"] = 0.62
	assert SD.story_fcff(_nvda_ctx(), hi)["assumptions"]["margin_floor"] is None


	def _floored_story(floor):
	st = F.story()
	st["assumptions"].update(segments=[{"name": "Generation", "rev_usd_b": 70.0, "margin": 0.55, "sales_to_capital": 2.3},
	{"name": "Retail", "rev_usd_b": 30.0, "margin": 0.32, "sales_to_capital": 1.8}],
	revenue_N_usd_b=100.0, margin_proposed=0.481, margin_target=0.6, margin_cap=0.652, margin_floor=floor)
	return st


	def test_nvda_a_recorded_floor_passes_the_segment_check_and_is_disclosed():
	d = F.all_data(); d["segments_canonical"] = TP._canon_two()
	run, payload, failed = F.run_pipeline(data=d, forecast_over={"story": _floored_story(0.6)})
	chk = next(c for v in run.validations for c in v.checks if c.id == "SEGMENT_BUILD_RECONCILES")
	assert failed is None and chk.status == "PASS", (failed, chk.detail)
	md = payload["markdown"]
	assert "is raised to 60.0%" in md and "best sustained margin in its sector" in md
	# the same difference with no recorded floor is unexplained and still blocks
	run, payload, failed = F.run_pipeline(data=d, forecast_over={"story": _floored_story(None)})
	chk = next(c for v in run.validations for c in v.checks if c.id == "SEGMENT_BUILD_RECONCILES")
	assert failed == "FORECAST" and chk.status == "FAIL" and "nor a recorded floor" in chk.detail


	# ── PGY: a later composite step stays inside the recorded cap ──
	def test_pgy_the_gaap_reblend_is_bounded_by_the_recorded_cap():
	import predict_yearly as PY
	price = 17.62
	fv = {"methods": {"P/E": {"price": 54.2, "conf": 0.54}, "PEG": {"price": 94.0, "conf": 0.43}, "P/FCF": {"price": 42.0, "conf": 0.38},
	"P/S": {"price": 45.4, "conf": 0.36}, "Analyst Target": {"price": 24.0, "conf": 0.40}, "DCF": {"price": 75.4, "conf": 0.49}},
	"fair_price": 44.05, "composite_audit": {"cap_upper": 2.5, "cap_lower": 0.78, "methods": [], "final": 44.05,
	"steps": [{"step": "sanity cap", "set_to": 44.05, "note": "FV_CAPPED_HIGH"}]}}
	PY._gaap_reblend(fv, {"gaap_distortion": True, "distortion_reason": "reported EPS far below adjusted"}, price)
	steps = fv["composite_audit"]["steps"]
	assert [s["step"] for s in steps] == ["sanity cap", "GAAP-distortion re-blend", "sanity cap"]
	assert steps[1]["set_to"] > 2.5 * price # the re-blend alone would have left the bound
	assert fv["fair_price"] == pytest.approx(2.5 * price) == pytest.approx(fv["composite_audit"]["final"]) == pytest.approx(steps[-1]["set_to"])
	# a re-blend inside the bound adds no cap step
	fv2 = copy.deepcopy(fv); fv2["composite_audit"]["steps"] = fv2["composite_audit"]["steps"][:1]
	for m in fv2["methods"].values():
	m["price"] = 30.0
	PY._gaap_reblend(fv2, {"gaap_distortion": True, "distortion_reason": "x"}, price)
	assert [s["step"] for s in fv2["composite_audit"]["steps"]] == ["sanity cap", "GAAP-distortion re-blend"]


	# ── PGY: the break-even revenue keeps the precision of its growth rate; a shrinking path is not "growth" ──
	def test_pgy_break_even_revenue_and_its_growth_reproduce_each_other():
	ctx = {"revenue": 1.325553e9, "shares": 77e6, "price": 17.62, "op_margin": 0.212, "rf": 0.0524, "erp": 0.0409, "beta": 1.5,
	"debt": 0.7e9, "cash": 0.3e9, "market_cap": 1.36e9, "stage": "mature-growth", "interest_coverage": 5,
	"sector": "FINANCIAL SERVICES"}
	A = {"tam_at_maturity_usd_b": 51.0, "plausible_peak_share": 0.08, "years_to_maturity": 8, "target_operating_margin": 0.256,
	"sales_to_capital": 2.0, "failure_probability": 0.0}
	base = SD.story_fcff(ctx, A, strict=False)["value"]
	ctx["price"] = 0.75 * base # a price the solve brackets
	be = SD.story_breakeven(ctx, A)
	rev, g = be["revenue_needed_N_usd_b"], be["rev_cagr_needed"]
	assert rev != round(rev, 2) # not rounded to the printed two decimals
	assert abs((rev / 1.325553) ** (1 / 8) - 1 - g) < 1e-4 # the relation recomputes the stored rate


	def test_a_negative_growth_rate_is_rendered_as_shrinkage():
	import catalog as C
	spec = RL.Spec("REL_BREAK_EVEN_REVENUE_GROWTH", "BE_REVENUE", "STORY_START_REVENUE", "path", RL.CAGR, RL.GROWTH_FROM,
	RL.GROWTH_FROM, "BE_REVENUE_CAGR", lhs_words="the revenue today's price needs at maturity",
	rhs_words="the story's starting revenue")
	d = C.BY_ID["BE_REVENUE"]
	down = RL._render(spec, RL.GROWTH_FROM, -0.0612, 0.8e9, 1.3256e9, d, d, 8)
	up = RL._render(spec, RL.GROWTH_FROM, 0.092, 50.4e9, 21.0e9, d, d, 10)
	flat = RL._render(spec, RL.GROWTH_FROM, 0.0001, 1.3e9, 1.3e9, d, d, 8)
	assert "allows revenue to shrink 6.1% a year" in down and "requires" not in down
	assert "requires 9.2% annual revenue growth" in up and "requires no revenue growth" in flat


	# ── NNE and the retry wording ──
	def _patched_run(revenue, operating_income, financial=False):
	run, payload, failed = F.run_pipeline()
	s = run.state
	h = s.historical._data
	h["revenue_ttm"] = h["revenue_ttm"].with_value(revenue)
	h["operating_income_ttm"] = h["operating_income_ttm"].with_value(operating_income)
	s.issuer._data["is_financial"] = financial
	return run


	def test_nne_a_pre_revenue_company_is_told_so(tmp_path):
	run = _patched_run(214042.0, -51.87e6)
	assert run.withheld_reason() == "PRE_REVENUE"
	out = run.write(tmp_path / "x.json")
	assert out["user"]["status"] == "WITHHELD" and out["user"]["reason"] == "PRE_REVENUE"
	assert "little or no revenue" in out["user"]["message"] and "try again" not in out["user"]["message"]
	assert _patched_run(0.0, -5e6).withheld_reason() == "PRE_REVENUE"
	assert _patched_run(19e9, 3.4e9).withheld_reason() is None # an established business
	assert _patched_run(214042.0, -51.87e6, financial=True).withheld_reason() is None
	assert _patched_run(None, None).withheld_reason() is None # no figures is a data failure, not this


	def test_deterministic_failures_do_not_invite_a_retry():
	for stage in ("INITIAL", "DATA_VALIDATED"):
	m = RM.withheld_payload("X", stage_reached=stage)["message"]
	assert "try again later" not in m.lower() and "same result" in m
	for stage in ("FORECAST_VALIDATED", "ANALYSIS_FINALIZED"): # these depend on the analyst models' answers
	assert "try again later" in RM.withheld_payload("X", stage_reached=stage)["message"].lower()
	assert "reason" not in RM.withheld_payload("X", stage_reached="DATA_VALIDATED")