"""PricingState: the market composite as a reproducible, audited object. The composite is a confidence × lifecycle-weighted average of the pricing methods, followed by a fixed sequence of adjustments. Every adjustment the composite (or its confidence) can receive is registered here with the single risk it addresses and the quantity it acts on. Two penalties for the same risk on the same quantity are a double count: ``double_penalties`` finds them and validation fails the run. Stacked conservatism, documented (each acts on a different risk or quantity): * ANALYST_TARGET_OPTIMISM sell-side target optimism → the analyst-target method's input * SECTOR_RERATE_SHRINK fundamental deterioration → the peer-anchor weight (composite magnitude) * OUTLIER_TRIM a method on a broken basis → which methods enter the composite * SANITY_CAP an implausible composite size → composite magnitude (bounded) * VALUE_TRAP_CONFIDENCE fundamental deterioration → confidence * METHOD_DISPERSION methods disagree with each other → confidence * SANITY_CAP_CONFIDENCE an implausible composite size → confidence * FORECAST_QUALITY weak forecast inputs → confidence * BLEND_DISAGREEMENT the price model disagrees → confidence (only when that model is reliable) The former value-trap magnitude haircut (×0.85/0.72/0.60 on the composite) addressed fundamental deterioration on the composite magnitude, which SECTOR_RERATE_SHRINK already does; it was removed so the risk is priced once. Premiums follow the same discipline (``double_adjustments``). Each premium names the characteristics it prices and the multiple component it acts on. The company's own historical multiple already embeds its quality, growth and risk, so an absolute quality/growth premium on it is a DOUBLE_PREMIUM. The only quality premium is INDUSTRY_QUALITY_PREMIUM: relative to the industry, on the industry anchor. Change against the company's own history is a separate HISTORICAL_REGIME_ADJUSTMENT on the own-history component. The retired GROWTH_QUALITY_PREMIUM (absolute thresholds, applied to the whole blend) stays registered only so that it is detected if it ever reappears. """ from __future__ import annotations MAGNITUDE, CONFIDENCE, METHOD_INPUT, INCLUSION = "composite magnitude", "confidence", "method input", "method inclusion" ADJUSTMENTS = { "ANALYST_TARGET_OPTIMISM": { "kind": "penalty", "risk": "sell-side target optimism", "quantity": METHOD_INPUT, "rule": "keep 55% of the implied upside of the mean target (a below-price target is used as published)", "source": "Brav & Lehavy (2003); Bradshaw, Brown & Huang (2013): consensus 12-month targets imply roughly " "twice the return later realized"}, "SECTOR_RERATE_SHRINK": { "kind": "penalty", "risk": "fundamental deterioration", "quantity": MAGNITUDE, "rule": "peer-anchor weight × (1 − 0.30 × deterioration signals), floor 0.10", "source": "engine rule: a deteriorating company is not credited a re-rating toward its peers"}, "OUTLIER_TRIM": { "kind": "exclusion", "risk": "method on a broken basis", "quantity": INCLUSION, "rule": "drop a method above 2.5× or below 0.4× the median of all pricing methods (needs ≥3 methods, keeps ≥2)", "source": "engine rule"}, "SANITY_CAP": { "kind": "bound", "risk": "implausible composite magnitude", "quantity": MAGNITUDE, "rule": "composite bounded to a multiple of the price set by market-cap tier and realized volatility", "source": "engine rule"}, "COMPOUNDER_FLOOR_RELIEF": { "kind": "relief", "risk": "implausible composite magnitude", "quantity": MAGNITUDE, "rule": "lower bound raised to 0.78× for durable high-ROE, non-cyclical, growing companies", "source": "engine rule"}, "GROWTH_QUALITY_PREMIUM": { "kind": "premium", "risk": None, "quantity": MAGNITUDE, "characteristics": ("growth", "quality"), "component": "blended multiple (own history and industry anchor)", "relative_to_component": False, "rule": "RETIRED 2026-09-24: absolute premium (ROE > 20%, margin > 20%, growth > 15%) on the whole blend", "source": "engine rule (retired)"}, "INDUSTRY_QUALITY_PREMIUM": { "kind": "premium", "risk": None, "quantity": MAGNITUDE, "characteristics": ("growth", "quality", "risk"), "component": "industry_anchor", "relative_to_component": True, "rule": "industry anchor × justified multiple(company growth, ROE, cost of equity) ÷ justified multiple(industry " "inputs), same stable phase; × relative net margin for P/S; bounded 0.5–2.0", "source": "Damodaran, relative valuation controlling for fundamentals (pricing_premium)"}, "HISTORICAL_REGIME_ADJUSTMENT": { "kind": "premium", "risk": None, "quantity": MAGNITUDE, "characteristics": ("change vs own history",), "component": "own_history", "relative_to_component": True, "rule": "own historical multiple × measured change vs the company's own history (default 1.0)", "source": "engine rule"}, "VALUE_TRAP_CONFIDENCE": { "kind": "penalty", "risk": "fundamental deterioration", "quantity": CONFIDENCE, "rule": "confidence × 0.80 with two or more deterioration signals", "source": "engine rule"}, "METHOD_DISPERSION": { "kind": "penalty", "risk": "method dispersion", "quantity": CONFIDENCE, "rule": "confidence × 0.80 when the coefficient of variation of method prices exceeds 0.30", "source": "engine rule"}, "SANITY_CAP_CONFIDENCE": { "kind": "penalty", "risk": "implausible composite magnitude", "quantity": CONFIDENCE, "rule": "confidence × 0.45 (max 0.35) when the composite was bounded", "source": "engine rule"}, "FORECAST_QUALITY": { "kind": "penalty", "risk": "forecast input quality", "quantity": CONFIDENCE, "rule": "heavy anchoring / heavy cyclical dampening / FCF-EPS inconsistency / leverage multipliers, bounded 0.40–1.25; " "gross-margin compression is not counted here when deterioration is already priced", "source": "engine rule"}, "BLEND_DISAGREEMENT": { "kind": "penalty", "risk": "price-model disagreement", "quantity": CONFIDENCE, "rule": "fundamental confidence × 0.40–0.70 when the price model disagrees, only if that model's out-of-sample " "direction accuracy is at least 55%", "source": "engine rule"}, } METHOD_LABELS = {"Analyst Target": "Analyst target (risk-adjusted contribution)"} def applied(aid, **detail): """One applied-adjustment record.""" d = ADJUSTMENTS[aid] return dict({"id": aid, "kind": d["kind"], "risk": d["risk"], "quantity": d["quantity"]}, **detail) # characteristics each multiple component already carries (an absolute premium on them counts them twice) EMBEDDED = {"own_history": ("growth", "quality", "risk"), "blended multiple (own history and industry anchor)": ("growth", "quality", "risk"), "peg_multiple": ("growth",)} def double_penalties(adjustments): """[(risk, quantity, [ids])] where one risk is penalized more than once on one quantity.""" seen = {} for a in adjustments or []: if a.get("kind") != "penalty" or not a.get("risk"): continue seen.setdefault((a["risk"], a["quantity"]), []).append(a["id"]) return [(r, q, ids) for (r, q), ids in seen.items() if len(set(ids)) > 1] def double_adjustments(adjustments): """Every economic characteristic counted twice, penalties and premiums alike: * DOUBLE_PENALTY — one risk penalized twice on one quantity; * DOUBLE_PREMIUM — one characteristic premium'd twice on one component of one method, or an absolute (not component-relative) premium on a component that already embeds that characteristic (the company's own history embeds its quality; a PEG multiple embeds growth).""" out = [{"type": "DOUBLE_PENALTY", "characteristic": r, "component": q, "ids": ids} for r, q, ids in double_penalties(adjustments)] seen = {} for a in adjustments or []: if a.get("kind") != "premium": continue d = ADJUSTMENTS.get(a.get("id"), {}) comp = a.get("component") or d.get("component") chars = tuple(a.get("characteristics") or d.get("characteristics") or ()) for c in chars: seen.setdefault((c, comp, a.get("method")), []).append(a["id"]) emb = set(EMBEDDED.get(comp, ())) rel = a.get("relative_to_component", d.get("relative_to_component", False)) if emb & set(chars) and not rel: out.append({"type": "DOUBLE_PREMIUM", "characteristic": ", ".join(sorted(emb & set(chars))), "component": comp, "ids": [a["id"]], "method": a.get("method"), "reason": f"{comp} already embeds {', '.join(sorted(emb & set(chars)))}; an absolute premium counts it again"}) for (c, comp, m), ids in seen.items(): if len(ids) > 1: out.append({"type": "DOUBLE_PREMIUM", "characteristic": c, "component": comp, "method": m, "ids": ids}) return out def multiple_architecture(ps, tol=1e-6): """[problems]: each multiple must rebuild from its lineage as own_history × historical_regime_adjustment × (1 − w) + industry_anchor × industry_quality_premium × w, P/FCF (no industry anchor) must carry no industry premium, and PEG (growth in the metric) no premium at all.""" bad = [] for m in ps.get("methods") or []: L, name = m.get("multiple_lineage") or {}, m.get("method") if name in ("P/E", "P/S") and m.get("target_multiple") is not None: own, w = L.get("own_history"), float(L.get("sector_weight") or 0.0) reg = float(((L.get("historical_regime_adjustment") or {}).get("value")) or 1.0) prem = float(((L.get("industry_quality_premium") or {}).get("value")) or 1.0) anc = L.get("industry_anchor") if anc is None: bad.append(f"{name}: no industry anchor recorded") continue own_ok = own is not None and w < 1.0 want = ((float(own) * reg) * (1 - w) if own_ok else 0.0) + float(anc) * prem * w if abs(want - float(m["target_multiple"])) > tol * max(1.0, want): bad.append(f"{name}: target {m['target_multiple']:.4f} ≠ own × regime × (1 − w) + industry × premium × w = {want:.4f}") if name == "P/FCF" and (L.get("industry_quality_premium") or float(L.get("sector_weight") or 0.0) > 0): bad.append("P/FCF: an industry premium or weight without a sourced industry anchor") if name == "PEG" and (L.get("industry_quality_premium") or L.get("growth_quality_in_metric") is not True): bad.append("PEG: growth is in the metric; no premium may be added") return bad def cap_diagnostics(au, price): """How the sanity bound interacted with the composite: a cap is a safeguard, not a target.""" raw, up, lo = au.get("raw"), au.get("cap_upper"), au.get("cap_lower") if not raw or not price: return {} r = raw / price steps = au.get("steps") or [] applied = any(s.get("step") == "sanity cap" for s in steps) above = [{"method": m.get("method"), "price_vs_price": m["price"] / price} for m in au.get("methods") or [] if m.get("in_composite") and up and m.get("price") and m["price"] / price > up] return {"raw_vs_price": r, "upper": up, "lower": lo, "cap_applied": applied, "direction": ("upper" if (up and r > up) else "lower" if (lo and r < lo) else None) if applied else None, "distance_above_cap": (r - up) if (up and r > up) else 0.0, "distance_below_floor": (lo - r) if (lo and r < lo) else 0.0, "methods_above_cap": above} SATURATION_SHARE = 0.25 # monitoring threshold: a quarter of valid companies at a bound means the multiples need a look def saturation(records, threshold=SATURATION_SHARE): """Across a regression universe: how often the upper bound binds, by issuer type and by method, and how far. ``records``: [{ticker, issuer_class, pricing_state}]. PRICING_MODEL_SATURATION when the binding share reaches ``threshold`` on five or more companies: investigate the upstream multiples, do not raise the cap.""" rows = [r for r in records if (r.get("pricing_state") or {}).get("methods")] bound = [r for r in rows if ((r["pricing_state"].get("cap_diagnostics") or {}).get("direction") == "upper")] by_type, by_method = {}, {} for r in rows: t = r.get("issuer_class") or "unknown" b = by_type.setdefault(t, {"n": 0, "upper_bound": 0}) b["n"] += 1 if r in bound: b["upper_bound"] += 1 for m in (r["pricing_state"].get("cap_diagnostics") or {}).get("methods_above_cap") or []: by_method[m["method"]] = by_method.get(m["method"], 0) + 1 ex = sorted((r["pricing_state"]["cap_diagnostics"]["distance_above_cap"] for r in bound)) share = len(bound) / len(rows) if rows else 0.0 return {"n": len(rows), "upper_bound": len(bound), "share": share, "by_issuer_type": by_type, "methods_above_cap": by_method, "tickers": [r.get("ticker") for r in bound], "median_excess": ex[len(ex) // 2] if ex else None, "max_excess": ex[-1] if ex else None, "flag": "PRICING_MODEL_SATURATION" if (len(rows) >= 5 and share >= threshold) else None} def build(fv, current_price): """PricingState dict from the fair-value result's composite audit (engine output → canonical record).""" au = (fv or {}).get("composite_audit") or {} lin = au.get("method_lineage") or {} rows = [] for m in au.get("methods") or []: name = m.get("method") L = lin.get(name) or {} row = {"method": name, "label": METHOD_LABELS.get(name, name), "raw_input": L.get("flow"), "raw_input_basis": L.get("flow_basis"), "target_multiple": L.get("target_multiple"), "multiple_basis": L.get("multiple_basis"), "multiple_lineage": {k: L.get(k) for k in ("own_history", "own_history_basis", "sector_anchor", "sector_weight", "industry_anchor", "industry_quality_premium", "industry_adjusted", "historical_regime_adjustment", "growth_quality_in_metric", "application") if k in L}, "raw_price": m.get("price"), "confidence": m.get("confidence"), "stage_weight": m.get("stage_weight"), "effective_weight": m.get("eff_weight"), "share_raw": m.get("share_raw"), "share_final": m.get("share_final"), "in_raw": m.get("in_raw"), "in_composite": m.get("in_composite"), "dropped_reason": m.get("dropped_reason"), "adjustments": [a["id"] for a in (au.get("adjustments") or []) if a.get("method") == name], "final_price": m.get("price") if m.get("in_composite") else None, "source": L.get("source") or ("consensus.target_price" if name == "Analyst Target" else "pricing composite")} if name == "Analyst Target": at = au.get("analyst_target") or {} row["raw_analyst_target"] = at.get("raw") row["risk_adjusted_analyst_target_contribution"] = at.get("used") row["raw_input"], row["raw_input_basis"] = at.get("raw"), "mean sell-side 12-month target" rows.append(row) steps = [dict(x) for x in (au.get("steps") or [])] post_trim = au.get("raw") for st in steps: if st.get("step") == "outlier trimming": post_trim = st.get("set_to") cap = next((st for st in steps if st.get("step") == "sanity cap"), None) return { "formula": au.get("formula"), "stage": au.get("stage"), "methods": rows, "raw_composite": au.get("raw"), "post_trim_composite": post_trim, "cap": ({"applied": True, "upper": au.get("cap_upper"), "lower": au.get("cap_lower"), "set_to": cap.get("set_to"), "note": cap.get("note")} if cap else {"applied": False, "upper": au.get("cap_upper"), "lower": au.get("cap_lower")}), "final_composite": au.get("final"), "steps": steps, "adjustments": list(au.get("adjustments") or []), "anchor_lineage": au.get("anchor_lineage") or {}, "input_lineage": au.get("input_lineage") or {}, "double_penalties": double_penalties(au.get("adjustments")), "double_adjustments": double_adjustments(au.get("adjustments")), "cap_diagnostics": cap_diagnostics(au, current_price), "price": current_price, } def reproduce(ps, tol=0.005): """(ok, detail): recompute raw → post-trim → final from the persisted PricingState alone.""" rows = [m for m in ps.get("methods") or [] if m.get("in_raw")] if not rows: return False, "no methods in the raw composite" w = sum(float(m.get("effective_weight") or 0) for m in rows) if w <= 0: return False, "zero total weight" raw = sum(float(m["raw_price"]) * float(m["effective_weight"]) for m in rows) / w if ps.get("raw_composite") is None or abs(raw / ps["raw_composite"] - 1) > tol: return False, f"raw composite {ps.get('raw_composite')} ≠ Σ price × weight ÷ Σ weight = {raw:.4f}" kept = [m for m in rows if m.get("in_composite")] wk = sum(float(m["effective_weight"]) for m in kept) post = sum(float(m["raw_price"]) * float(m["effective_weight"]) for m in kept) / wk if wk else None if post is None or ps.get("post_trim_composite") is None or abs(post / ps["post_trim_composite"] - 1) > tol: return False, f"post-trim composite {ps.get('post_trim_composite')} ≠ recomputed {post}" fin = post cap = ps.get("cap") or {} for st in ps.get("steps") or []: if st.get("factor") is not None: fin *= float(st["factor"]) elif st.get("set_to") is not None and st.get("step") != "outlier trimming": fin = float(st["set_to"]) if cap.get("applied") and ps.get("price") and cap.get("upper") and cap.get("lower"): p = float(ps["price"]) if not (cap["lower"] * p * (1 - tol) <= fin <= cap["upper"] * p * (1 + tol)): return False, f"final {fin:.4f} outside the recorded cap [{cap['lower']}, {cap['upper']}] × price" if ps.get("final_composite") is None or abs(fin / ps["final_composite"] - 1) > tol: return False, f"final composite {ps.get('final_composite')} ≠ recomputed {fin:.4f}" return True, f"raw {raw:.4f} → post-trim {post:.4f} → final {fin:.4f}"