| """Applicability policy: the issuer type decides which metrics, models and language may run. |
| |
| One policy object is consumed by canonicalisation, the scorecard, the pricing composite, the |
| valuation engine selector, the blend, the personas and the renderer. Modules that do not apply |
| are DISABLED here (their inputs are removed or their outputs filtered), not down-weighted. |
| |
| Also here: |
| canonicalize_statements() — rewrites the provider's statement revenue to the canonical concept |
| for balance-sheet financials ONCE, upstream, and quarantines the rest |
| apply_policy_to_scorecard()— removes non-applicable generic rows, adds issuer-specific ones |
| quant_reliability() — VALIDATED / LIMITED / UNRELIABLE / UNAVAILABLE with language + weight caps |
| """ |
| from __future__ import annotations |
|
|
| from dataclasses import dataclass |
|
|
| from concepts import GROSS_REVENUE, NET_REVENUE, FINANCIAL_ISSUER_CLASSES |
|
|
|
|
| def _f(x, default=None): |
| try: |
| v = float(x) |
| return v if v == v else default |
| except (TypeError, ValueError): |
| return default |
|
|
|
|
| |
| _INDUSTRIAL_LEVERAGE_ROWS = {"Debt/Equity", "Interest Coverage", "Current Ratio"} |
| _INDUSTRIAL_CASH_ROWS = {"FCF Margin", "FCF Conversion", "OCF > Net Income", "P/FCF"} |
| _INDUSTRIAL_MARGIN_ROWS = {"Operating Margin", "Gross Margin", "ROIC", "EV/EBITDA", "EV/Revenue", "P/S Ratio"} |
|
|
|
|
| @dataclass |
| class ApplicabilityPolicy: |
| issuer_class: str |
| mixed_financial: bool = False |
| revenue_concept: str = NET_REVENUE |
| valuation_engine: str = "FcffValuation" |
| mechanical_basis: str = "eps" |
| story_eligible: bool = True |
| ev_meaningful: bool = True |
| sotp_required: bool = False |
| allowed_pricing_methods: tuple = ("P/E", "PEG", "P/FCF", "P/S", "Analyst Target") |
| disallowed_scorecard_rows: frozenset = frozenset() |
| disallowed_scorecard_categories: frozenset = frozenset() |
| issuer_metrics: tuple = () |
| persona_excluded_topics: tuple = () |
| allowed_metric_concepts: tuple = () |
| disallowed_metric_concepts: tuple = () |
| reason: str = "" |
|
|
| def as_dict(self): |
| d = self.__dict__.copy() |
| for k in ("disallowed_scorecard_rows", "disallowed_scorecard_categories"): |
| d[k] = sorted(d[k]) |
| return d |
|
|
| @classmethod |
| def for_issuer(cls, issuer: dict | None) -> "ApplicabilityPolicy": |
| c = (issuer or {}).get("issuer_class") or "industrial" |
| mixed = bool((issuer or {}).get("mixed_financial")) |
| if c in ("bank", "lender", "bdc"): |
| p = cls(c, mixed, revenue_concept=NET_REVENUE, valuation_engine="BankEquityValuation", mechanical_basis="book_roe", |
| story_eligible=False, ev_meaningful=False, |
| allowed_pricing_methods=("P/E", "Analyst Target"), |
| disallowed_scorecard_rows=frozenset(_INDUSTRIAL_LEVERAGE_ROWS | _INDUSTRIAL_CASH_ROWS | _INDUSTRIAL_MARGIN_ROWS), |
| disallowed_scorecard_categories=frozenset({"Cash Flow", "Health"}), |
| issuer_metrics=("ROE", "ROTCE", "Tangible book value / share", "P/B", "P/TBV", "Efficiency ratio", "Net interest income growth", |
| "Net revenue growth", "Payout ratio", "CET1 ratio", "Net interest margin", "Loan growth", "Deposit growth", "Net charge-off ratio", "Reserve coverage"), |
| persona_excluded_topics=("operating margin", "op margin", "gross margin", "free cash flow", "fcf", "current ratio", "debt/equity", "debt-to-equity", "interest coverage", "ev/ebitda", "ev/revenue", "working capital", "p/s", "price-to-sales"), |
| allowed_metric_concepts=("ROE", "ROTCE", "CET1", "TANGIBLE_BOOK", "P_TBV", "P_B", "EFFICIENCY_RATIO", "NII", "NIM", "DEPOSIT_GROWTH", "LOAN_GROWTH", "CREDIT_COSTS", "NCO", "RESERVE_COVERAGE", "CAPITAL_RETURN", NET_REVENUE), |
| disallowed_metric_concepts=("EV_EBITDA", "EV_REVENUE", "DEBT_EQUITY", "CURRENT_RATIO", "INTEREST_COVERAGE", "FCF_CONVERSION", "OCF_GT_NI", "GENERIC_OPERATING_MARGIN", "FCFF_WACC", GROSS_REVENUE), |
| reason="deposits and borrowings are the raw material of a bank: enterprise value, EBIT margins, leverage and cash-conversion ratios built for industrial firms do not describe its economics") |
| elif c == "insurer": |
| p = cls(c, mixed, revenue_concept=NET_REVENUE, valuation_engine="InsurerEquityValuation", mechanical_basis="book_roe", |
| story_eligible=False, ev_meaningful=False, allowed_pricing_methods=("P/E", "Analyst Target"), |
| disallowed_scorecard_rows=frozenset(_INDUSTRIAL_LEVERAGE_ROWS | _INDUSTRIAL_CASH_ROWS | _INDUSTRIAL_MARGIN_ROWS), |
| disallowed_scorecard_categories=frozenset({"Cash Flow", "Health"}), |
| issuer_metrics=("ROE", "Book value / share", "P/B", "Combined ratio", "Reserve adequacy", "Capital adequacy", "Payout ratio"), |
| persona_excluded_topics=("operating margin", "op margin", "free cash flow", "fcf", "current ratio", "debt/equity", "interest coverage", "ev/ebitda", "ev/revenue", "p/s"), |
| disallowed_metric_concepts=("EV_EBITDA", "DEBT_EQUITY", "CURRENT_RATIO", "INTEREST_COVERAGE", "FCF_CONVERSION", "GENERIC_OPERATING_MARGIN", "FCFF_WACC"), |
| reason="reserves and float make operating margin and free cash flow meaningless for an insurer; book value, ROE and underwriting ratios carry the case") |
| elif c == "reit": |
| p = cls(c, mixed, revenue_concept=NET_REVENUE, valuation_engine="ReitValuation", mechanical_basis="affo", |
| story_eligible=False, ev_meaningful=True, allowed_pricing_methods=("P/S", "Analyst Target"), |
| disallowed_scorecard_rows=frozenset({"P/E Ratio", "PEG (provider, trailing basis)", "PEG (forward P/E ÷ forecast EPS growth)", "PEG Ratio", "Earnings Growth YoY", "Profit Margin", "Current Ratio", "Interest Coverage"}), |
| issuer_metrics=("FFO / share", "AFFO / share", "P/AFFO", "Dividend yield", "Payout of AFFO", "Net debt / EBITDA", "Occupancy", "Same-store NOI growth"), |
| persona_excluded_topics=("p/e", "peg", "eps growth", "current ratio"), |
| disallowed_metric_concepts=("GAAP_EPS_MULTIPLE", "PEG", "CURRENT_RATIO"), |
| reason="GAAP earnings are depressed by non-cash depreciation on appreciating property; cash funds from operations and NAV drive value") |
| elif c in ("asset_manager", "exchange", "broker_dealer", "insurance_broker"): |
| p = cls(c, mixed, valuation_engine="FcffValuation", mechanical_basis="eps", allowed_pricing_methods=("P/E", "PEG", "P/FCF", "Analyst Target"), |
| disallowed_scorecard_rows=frozenset({"Current Ratio", "Interest Coverage", "EV/Revenue"}), |
| persona_excluded_topics=("current ratio",), |
| reason="fee-based financial: earns on revenue and margins like an operating company") |
| elif c in ("holding", "spac"): |
| p = cls(c, mixed, valuation_engine="SumOfPartsValuation", sotp_required=True, reason="holding structure: a consolidated margin blends unrelated businesses") |
| else: |
| p = cls(c, mixed, reason="ordinary operating company: debt is financing, enterprise value and FCFF are meaningful") |
| if mixed and c not in FINANCIAL_ISSUER_CLASSES: |
| |
| |
| |
| p.sotp_required = True |
| p.valuation_engine = "MixedEquityValuation" |
| p.story_eligible = False |
| p.reason += "; a material financial-services segment sits inside an operating company — the consolidated FCFF/WACC story is not applicable; the equity model (earnings at the cost of equity) is, and sum-of-parts is used when the filing discloses business-line profits" |
| return p |
|
|
|
|
| |
| def canonicalize_statements(data: dict, issuer: dict | None) -> dict: |
| """For balance-sheet financials, the provider's statement `totalRevenue` is GROSS revenue |
| (interest income + noninterest revenue, before interest expense) while consensus, the profile |
| and the SEC filing carry NET (managed) revenue. Rewrite the statement series to NET revenue |
| ONCE so every downstream reader (TTM, growth, forecast model, gates, prompts) sees one concept, |
| and quarantine the gross series. Returns the rejected-field registry entry (or {}).""" |
| c = (issuer or {}).get("issuer_class") |
| if c not in FINANCIAL_ISSUER_CLASSES: |
| data.setdefault("canonical_concepts", {})["revenue"] = NET_REVENUE |
| return {} |
| inc = data.get("income") or {} |
| rewritten = 0; unverifiable = 0 |
| gross_kept = {"quarterlyReports": [], "annualReports": []} |
| for sec in ("quarterlyReports", "annualReports"): |
| for rep in inc.get(sec) or []: |
| gross = _f(rep.get("totalRevenue")) |
| nii = _f(rep.get("netInterestIncome")); nonii = _f(rep.get("nonInterestIncome")) |
| iexp = _f(rep.get("interestExpense")) |
| net = None; how = None |
| if nii is not None and nonii is not None and (nii + nonii) > 0: |
| net, how = nii + nonii, "netInterestIncome + nonInterestIncome" |
| elif gross is not None and iexp is not None and gross - abs(iexp) > 0: |
| net, how = gross - abs(iexp), "totalRevenue − interestExpense" |
| gross_kept[sec].append({"fiscalDateEnding": rep.get("fiscalDateEnding"), "grossRevenue": gross}) |
| if net is not None: |
| rep["grossRevenue_rejected"] = gross |
| rep["totalRevenue"] = net |
| rep["revenue_concept"] = NET_REVENUE |
| rep["revenue_derivation"] = how |
| rewritten += 1 |
| else: |
| rep["revenue_concept"] = "UNVERIFIED" |
| unverifiable += 1 |
| reg = {} |
| if rewritten: |
| reg = {"field": "provider_statement_revenue", "concept": GROSS_REVENUE, |
| "reason": f"gross {c} revenue (interest income + noninterest revenue before interest expense); the canonical model uses managed NET revenue", |
| "replaced_by": NET_REVENUE, "reports_rewritten": rewritten, "reports_unverified": unverifiable, |
| "derivation": "netInterestIncome + nonInterestIncome (fallback: totalRevenue − interestExpense)"} |
| data.setdefault("rejected_fields", {})["provider_statement_revenue"] = reg |
| data.setdefault("canonical_concepts", {})["revenue"] = NET_REVENUE |
| data["_gross_revenue_quarantined"] = gross_kept |
| |
| data.setdefault("lineage", {}).setdefault("revenue", []).append( |
| f"provider statement totalRevenue rewritten to NET revenue ({reg['derivation']}) for {rewritten} reports; gross series quarantined") |
| else: |
| data.setdefault("canonical_concepts", {})["revenue"] = "UNVERIFIED" |
| data.setdefault("rejected_fields", {})["provider_statement_revenue"] = { |
| "field": "provider_statement_revenue", "concept": GROSS_REVENUE, "replaced_by": None, |
| "reason": f"gross {c} revenue and no net-revenue components in the statements — revenue concept UNVERIFIED; revenue-based forecasts and consensus comparisons are blocked"} |
| return reg |
|
|
|
|
| |
| def _ttm(reports, key, n=4, start=0): |
| vals = [_f((r or {}).get(key)) for r in (reports or [])[start:start + n]] |
| vals = [v for v in vals if v is not None] |
| return sum(vals) if len(vals) == n else None |
|
|
|
|
| def bank_rows_from_facts(bf: dict, price: float | None, shares: float | None): |
| """Bank capital/credit rows from SEC company facts (10-K/10-Q XBRL). Missing tags stay UNAVAILABLE.""" |
| rows = [] |
| if not bf or bf.get("error"): |
| return rows |
| def g(k): return (bf.get(k) or {}) if isinstance(bf.get(k), dict) else {} |
| def add(name, val_str, rating, note): rows.append({"cat": "Issuer metrics", "name": name, "value": val_str, "rating": rating, "note": note}) |
| def growth(k, label): |
| x = g(k) |
| if x.get("value") is not None and x.get("prior_year"): |
| gr = x["value"] / x["prior_year"] - 1 |
| add(label, f"{gr*100:+.1f}%", "STRONG" if gr > 0.08 else "GOOD" if gr > 0.03 else "NEUTRAL" if gr > -0.03 else "WEAK", f"SEC {x.get('form')} {x.get('tag')} {x.get('end')} vs prior year") |
| else: |
| add(label, "UNAVAILABLE", "UNVERIFIED", (x.get("note") or "prior-year value not tagged")) |
| growth("deposits", "Deposit growth"); growth("loans_gross", "Loan growth") |
| lo, al, nco = g("loans_gross"), g("allowance"), g("net_charge_offs") |
| if lo.get("value") and al.get("value") is not None: |
| cov = al["value"] / lo["value"]; add("Reserve coverage (allowance ÷ loans)", f"{cov*100:.2f}%", "NEUTRAL", f"SEC {al.get('tag')} ÷ {lo.get('tag')} at {lo.get('end')}") |
| else: |
| add("Reserve coverage (allowance ÷ loans)", "UNAVAILABLE", "UNVERIFIED", al.get("note") or lo.get("note") or "not tagged") |
| if lo.get("value") and nco.get("value") is not None: |
| days = nco.get("period_days") or 365 |
| ann = nco["value"] * (365.0 / max(days, 1)) |
| r_ = ann / lo["value"]; add("Net charge-off ratio (annualised)", f"{r_*100:.2f}%", "STRONG" if r_ < 0.003 else "GOOD" if r_ < 0.006 else "NEUTRAL" if r_ < 0.012 else "WEAK", f"SEC {nco.get('tag')} ({days}-day period, annualised) ÷ loans") |
| else: |
| add("Net charge-off ratio (annualised)", "UNAVAILABLE", "UNVERIFIED", nco.get("note") or "not tagged") |
| nii, ta = g("nii"), g("total_assets") |
| if nii.get("value") and ta.get("value"): |
| days = nii.get("period_days") or 365 |
| nim = nii["value"] * (365.0 / max(days, 1)) / ta["value"] |
| add("Net interest margin (approx., NII ÷ total assets)", f"{nim*100:.2f}%", "NEUTRAL", f"SEC {nii.get('tag')} annualised ÷ {ta.get('tag')} — approximation (earning assets not tagged)") |
| else: |
| add("Net interest margin (approx., NII ÷ total assets)", "UNAVAILABLE", "UNVERIFIED", nii.get("note") or "not tagged") |
| for k, label in (("cet1_ratio", "CET1 ratio"), ("tier1_ratio", "Tier 1 capital ratio")): |
| x = g(k) |
| if x.get("value") is not None: |
| add(label, f"{x['value']*100:.1f}%", "STRONG" if x["value"] >= 0.13 else "GOOD" if x["value"] >= 0.11 else "NEUTRAL" if x["value"] >= 0.09 else "WEAK", f"SEC {x.get('form')} {x.get('tag')} {x.get('end')}") |
| else: |
| add(label, "UNAVAILABLE", "UNVERIFIED", "not tagged in the filer's XBRL — the regulatory capital table (FR Y-9C / 10-K) is not machine-readable here") |
| return rows |
|
|
|
|
| def issuer_metric_rows(policy: ApplicabilityPolicy, data: dict, price: float | None): |
| """Issuer-specific scorecard rows computed from statements; rows the provider cannot supply |
| are emitted as UNAVAILABLE rather than silently skipped.""" |
| ov = data.get("overview") or {} |
| inc = (data.get("income") or {}).get("quarterlyReports") or [] |
| bal = (data.get("balance") or {}).get("quarterlyReports") or [] |
| rows = [] |
| def add(name, val_str, rating, note=""): |
| rows.append({"cat": "Issuer metrics", "name": name, "value": val_str, "rating": rating, "note": note}) |
| if policy.issuer_class in ("bank", "lender", "bdc", "insurer"): |
| b0 = bal[0] if bal else {} |
| eq = _f(b0.get("totalShareholderEquity")); gw = _f(b0.get("goodwill")) or 0.0; ia = _f(b0.get("intangibleAssets")) or 0.0 |
| sh = _f(ov.get("SharesOutstanding")) |
| ni = _ttm(inc, "netIncome") |
| roe = _f(ov.get("ReturnOnEquityTTM")) |
| if roe is not None: add("ROE", f"{roe*100:.1f}%", "STRONG" if roe >= 0.15 else "GOOD" if roe >= 0.11 else "NEUTRAL" if roe >= 0.08 else "WEAK", "provider TTM") |
| if eq and sh: |
| bvps = eq / sh; add("Book value / share", f"${bvps:,.2f}", "NEUTRAL", "latest balance sheet") |
| tb = (eq - gw - ia) / sh |
| add("Tangible book value / share", f"${tb:,.2f}", "NEUTRAL", "equity less goodwill and intangibles") |
| if price: |
| pb = price / bvps; add("P/B", f"{pb:.2f}x", "STRONG" if pb < 1.0 else "GOOD" if pb < 1.5 else "NEUTRAL" if pb < 2.5 else "WEAK", "price ÷ book value per share") |
| if tb > 0: |
| ptb = price / tb; add("P/TBV", f"{ptb:.2f}x", "STRONG" if ptb < 1.2 else "GOOD" if ptb < 1.8 else "NEUTRAL" if ptb < 3.0 else "WEAK", "price ÷ tangible book per share") |
| if ni and (eq - gw - ia) > 0: |
| rotce = ni / (eq - gw - ia); add("ROTCE", f"{rotce*100:.1f}%", "STRONG" if rotce >= 0.18 else "GOOD" if rotce >= 0.13 else "NEUTRAL" if rotce >= 0.09 else "WEAK", "TTM net income ÷ tangible common equity (approx.)") |
| if policy.issuer_class != "insurer": |
| nii0 = _ttm(inc, "netInterestIncome"); nii1 = _ttm(inc, "netInterestIncome", start=4) |
| if nii0 is not None and nii1: |
| g = nii0 / nii1 - 1; add("Net interest income growth", f"{g*100:+.1f}%", "STRONG" if g > 0.08 else "GOOD" if g > 0.03 else "NEUTRAL" if g > -0.03 else "WEAK", "TTM vs prior TTM") |
| rev0 = _ttm(inc, "totalRevenue"); rev1 = _ttm(inc, "totalRevenue", start=4) |
| if rev0 and rev1: |
| g = rev0 / rev1 - 1; add("Net revenue growth", f"{g*100:+.1f}%", "STRONG" if g > 0.08 else "GOOD" if g > 0.03 else "NEUTRAL" if g > -0.03 else "WEAK", "canonical net revenue, TTM vs prior TTM") |
| opex = _ttm(inc, "operatingExpenses") |
| if opex and rev0: |
| eff = abs(opex) / rev0; add("Efficiency ratio", f"{eff*100:.0f}%", "STRONG" if eff < 0.55 else "GOOD" if eff < 0.62 else "NEUTRAL" if eff < 0.70 else "WEAK", "operating expenses ÷ net revenue (provider line items)") |
| _bf_rows = bank_rows_from_facts(data.get("bank_facts") or {}, price, sh) |
| if _bf_rows: |
| rows.extend(_bf_rows) |
| else: |
| for name in ("CET1 ratio", "Net interest margin", "Loan growth", "Deposit growth", "Net charge-off ratio", "Reserve coverage"): |
| add(name, "UNAVAILABLE", "UNVERIFIED", "not supplied by the market-data provider and SEC company facts unavailable") |
| else: |
| for name in ("Combined ratio", "Reserve adequacy", "Capital adequacy"): |
| add(name, "UNAVAILABLE", "UNVERIFIED", "not supplied by the market-data provider; from the statutory filings") |
| pr = _f(ov.get("PayoutRatio")) |
| if pr is not None: add("Payout ratio", f"{pr*100:.0f}%", "NEUTRAL", "dividends ÷ earnings (provider)") |
| elif policy.issuer_class == "reit": |
| cf = (data.get("cashflow") or {}).get("quarterlyReports") or [] |
| ni = _ttm(inc, "netIncome"); dep = _ttm(cf, "depreciationDepletionAndAmortization") |
| sh = _f(ov.get("SharesOutstanding")) |
| if ni is not None and dep and sh: |
| ffo = (ni + dep) / sh; add("FFO / share (approx.)", f"${ffo:,.2f}", "NEUTRAL", "net income + depreciation (no gain adjustments available)") |
| if price and ffo > 0: add("P/FFO", f"{price/ffo:.1f}x", "STRONG" if price / ffo < 14 else "GOOD" if price / ffo < 18 else "NEUTRAL" if price / ffo < 24 else "WEAK", "price ÷ FFO per share") |
| for name in ("Occupancy", "Same-store NOI growth", "Cap rate", "NAV / share"): |
| add(name, "UNAVAILABLE", "UNVERIFIED", "not supplied by the market-data provider; from the supplemental package") |
| return rows |
|
|
|
|
| def apply_policy_to_scorecard(fund: dict | None, policy: ApplicabilityPolicy, data: dict, price: float | None): |
| """Remove non-applicable generic rows and categories from the scorecard object every |
| downstream consumer (prompt builder, report, UI) reads; add issuer-specific rows.""" |
| if not fund: |
| return fund, {"removed": [], "added": []} |
| removed = [] |
| kept = [] |
| for row in fund.get("assessments") or []: |
| cat, name = row[0], row[1] |
| if name in policy.disallowed_scorecard_rows or cat in policy.disallowed_scorecard_categories: |
| removed.append(f"{cat}: {name}") |
| continue |
| kept.append(row) |
| added = issuer_metric_rows(policy, data, price) |
| for r in added: |
| kept.append((r["cat"], r["name"], r["value"], r["rating"], r["note"])) |
| fund["assessments"] = kept |
| fund["policy_removed"] = removed |
| fund["policy_added"] = [r["name"] for r in added] |
| sc = fund.get("scorecard") or {} |
| for cat in list(sc): |
| if cat in policy.disallowed_scorecard_categories: |
| pts, mx = (sc[cat][0], sc[cat][1]) if isinstance(sc[cat], tuple) else (0, 0) |
| fund["total_points"] = float(fund.get("total_points", 0)) - float(pts or 0) |
| fund["max_points"] = float(fund.get("max_points", 0)) - float(mx or 0) |
| sc.pop(cat) |
| fund["scorecard"] = sc |
| fund["applicability"] = {"issuer_class": policy.issuer_class, "removed": removed, "added": [r["name"] for r in added], "reason": policy.reason} |
| return fund, {"removed": removed, "added": [r["name"] for r in added]} |
|
|
|
|
| |
| def quant_reliability(ml_detail: dict | None) -> dict: |
| """Deterministic reliability state that controls both blend weight and narrative strength.""" |
| m = ml_detail or {} |
| n = int(m.get("n_oos_eff") or 0); acc = _f(m.get("dir_acc")); ic = _f(m.get("rank_ic")) |
| if not m or acc is None: |
| state = "UNAVAILABLE" |
| elif n >= 8 and acc >= 0.55 and (ic is None or ic > 0.05): |
| state = "VALIDATED" |
| elif n >= 5 and acc >= 0.50: |
| state = "LIMITED" |
| else: |
| state = "UNRELIABLE" |
| return reliability_record(state, basis=f"{n} independent out-of-sample points, direction accuracy {'' if acc is None else f'{acc*100:.0f}%'}, rank correlation {'' if ic is None else f'{ic:.2f}'}", |
| extra={"n_oos_eff": n, "dir_acc": acc, "rank_ic": ic}) |
|
|
|
|
| |
| |
| |
| |
| |
| _RELIABILITY = { |
| "VALIDATED": {"strength": "strong quantitative evidence", "cap": 1.0, "visible": True, "usable": True, |
| "allowed": ["strong quantitative evidence", "a validated signal", "supports", "adds independent evidence"], "forbidden": []}, |
| "LIMITED": {"strength": "limited signal", "cap": 0.5, "visible": True, "usable": True, |
| "allowed": ["a limited signal", "weak directional evidence", "points the same way but carries little weight"], |
| "forbidden": ["confirms", "confirmation", "heavily backtested", "rigorously backtested", "strong independent evidence", "decisive", "validated", "proves"]}, |
| "UNRELIABLE": {"strength": "exploratory only", "cap": 0.0, "visible": False, "usable": False, |
| "allowed": ["an exploratory signal with no demonstrated out-of-sample skill", "low-confidence model output"], |
| "forbidden": ["confirms", "confirmation", "supports the thesis", "heavily backtested", "rigorously backtested", "backtested", |
| "strong independent evidence", "independent evidence", "validated", "rigorously", "decisive", "highest-signal", "weigh this heavily"]}, |
| "UNAVAILABLE": {"strength": "none", "cap": 0.0, "visible": False, "usable": False, |
| "allowed": ["unavailable"], "forbidden": ["confirms", "backtested", "independent evidence", "quant model supports"]}, |
| } |
|
|
|
|
| |
| EVIDENCE_POLICY = { |
| "VALIDATED": {"recommendation_evidence_allowed": "full", "scenario_weight_allowed": True, |
| "scenario_adjustment_cap": 0.10, "headline_blend_weight_cap": 1.0, |
| "persona_visibility": True, "diagnostics_visibility": True}, |
| "LIMITED": {"recommendation_evidence_allowed": "weak_only", "scenario_weight_allowed": True, |
| "scenario_adjustment_cap": 0.03, "headline_blend_weight_cap": 0.5, |
| "persona_visibility": True, "diagnostics_visibility": True}, |
| "UNRELIABLE": {"recommendation_evidence_allowed": "none", "scenario_weight_allowed": False, |
| "scenario_adjustment_cap": 0.0, "headline_blend_weight_cap": 0.0, |
| "persona_visibility": False, "diagnostics_visibility": True}, |
| "UNAVAILABLE": {"recommendation_evidence_allowed": "none", "scenario_weight_allowed": False, |
| "scenario_adjustment_cap": 0.0, "headline_blend_weight_cap": 0.0, |
| "persona_visibility": False, "diagnostics_visibility": False}, |
| } |
|
|
|
|
|
|
|
|
| _ORDER = ("UNAVAILABLE", "UNRELIABLE", "LIMITED", "VALIDATED") |
|
|
|
|
| def reliability_rank(state): |
| return _ORDER.index(state) if state in _ORDER else 0 |
|
|
|
|
| def reliability_is_monotone(): |
| """Lowering reliability can never raise blend weight, scenario influence or claim strength.""" |
| caps = [EVIDENCE_POLICY[s]["headline_blend_weight_cap"] for s in _ORDER] |
| adjs = [EVIDENCE_POLICY[s]["scenario_adjustment_cap"] for s in _ORDER] |
| return caps == sorted(caps) and adjs == sorted(adjs) |
|
|
|
|
| def reliability_record(state, basis="", extra=None): |
| """One record per quantitative signal: it fixes BOTH the maximum blend weight AND the strongest |
| claim the prose may make from it, and whether a persona may see the numbers at all.""" |
| r = _RELIABILITY[state] |
| ev = EVIDENCE_POLICY[state] |
| out = {"state": state, "weight_cap": r["cap"], "max_blend_weight": ev["headline_blend_weight_cap"], |
| "recommendation_evidence_allowed": ev["recommendation_evidence_allowed"], |
| "scenario_weight_allowed": ev["scenario_weight_allowed"], |
| "scenario_adjustment_cap": ev["scenario_adjustment_cap"], |
| "diagnostics_visibility": ev["diagnostics_visibility"], |
| "allowed_language_strength": r["strength"], "allowed_claim_strength": r["strength"], |
| "allowed_phrases": list(r["allowed"]), "forbidden_phrases": list(r["forbidden"]), |
| "persona_visibility": r["visible"], "usable_for_argument": r["usable"], "basis": basis} |
| out.update(extra or {}) |
| return out |
|
|
|
|
|
|
|
|
| def panel_reliability(meta=None): |
| """Cross-sectional panel-ML rank: a different model from the single-stock price model. Its state |
| comes ONLY from the snapshot's own walk-forward validation metadata; with no metadata it is |
| UNAVAILABLE. There are no default statistics: an unvalidated rank cannot borrow a record.""" |
| m = meta or {} |
| ic, sharpe, n = _f(m.get("rank_ic")), _f(m.get("decile_sharpe")), int(_f(m.get("n_periods")) or 0) |
| if ic is None or sharpe is None: |
| return reliability_record("UNAVAILABLE", basis="the panel snapshot carries no walk-forward validation metadata", |
| extra={"model": "cross-sectional panel-ML rank"}) |
| |
| |
| |
| t, n_ind = _f(m.get("ic_t")), _f(m.get("n_independent")) or n / 12.0 |
| folds = m.get("folds") or [] |
| recent = _f((folds[-1] if folds else {}).get("rank_ic")) |
| recent_ok = recent is not None and recent >= 0 |
| if ic >= 0.10 and sharpe >= 0.8 and (t or 0) >= 2.0 and n_ind >= 8 and recent is not None and recent > 0: |
| state = "VALIDATED" |
| elif ic >= 0.02 and (t or 0) >= 1.0 and n_ind >= 5 and recent_ok: |
| state = "LIMITED" |
| else: |
| state = "UNRELIABLE" |
| why = ("" if state != "UNRELIABLE" else |
| "; the latest test block was negative" if (recent is not None and recent < 0) else |
| "; too weak or too few independent periods") |
| basis = (f"walk-forward rank correlation {ic:+.3f} (t {t if t is not None else float('nan'):.1f} over {n_ind:.1f} independent years), " |
| f"positive in {(_f(m.get('ic_positive_share')) or 0) * 100:.0f}% of months, decile Sharpe {sharpe:.2f}" |
| + (f", latest test block {(folds[-1] or {}).get('test')} {recent:+.3f}" if recent is not None else "") + why) |
| return reliability_record(state, basis=basis, |
| extra={"model": "cross-sectional panel-ML rank", "rank_ic": ic, "decile_sharpe": sharpe, "n_periods": n, |
| "ic_t": t, "n_independent": n_ind, "recent_block_ic": recent, |
| "recent_block": (folds[-1] if folds else {}).get("test"), |
| "ic_positive_share": _f(m.get("ic_positive_share")), |
| "failed_recently": bool(recent is not None and recent < 0)}) |
