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