"""Semantic types: normalized financial metrics, typed comparisons, signals, scenarios, thesis triggers, the rendered-claim registry and the immutable analysis context. The rule this module exists to enforce: a number may not cross a currency, scale, share-basis, accounting-basis, period or perimeter boundary silently. Every metric therefore carries BOTH its raw representation (for audit/display) and a NORMALIZED representation (USD, unit scale, and the security basis the stock is quoted in). Models read `normalized_value` only. Two metrics may be compared or divided only when their full semantic signature matches; otherwise the comparison is blocked, not warned about. """ from __future__ import annotations import hashlib from units import DECIMAL_FRACTION, PERCENT_NUMBER, RATIO, Pct, RangeClaim # noqa: F401 from units import format_percent as _fmt_pct, format_pp as _fmt_pp, format_ratio as _fmt_ratio import json import re from dataclasses import dataclass, field, asdict from types import MappingProxyType from typing import Any # ── economic concepts ───────────────────────────────────────────────────────────────── PRICE = "PRICE"; MARKET_CAP = "MARKET_CAP"; SHARES = "SHARES_OUTSTANDING" GROSS_REVENUE = "GROSS_REVENUE"; NET_REVENUE = "NET_REVENUE" INTEREST_INCOME = "INTEREST_INCOME"; INTEREST_EXPENSE = "INTEREST_EXPENSE"; NET_INTEREST_INCOME = "NET_INTEREST_INCOME" NONINTEREST_REVENUE = "NONINTEREST_REVENUE" CONSENSUS_REVENUE = "CONSENSUS_REVENUE"; MODEL_REVENUE = "MODEL_REVENUE" NET_INCOME = "NET_INCOME"; OPERATING_INCOME = "OPERATING_INCOME" GAAP_EPS = "GAAP_EPS"; ADJUSTED_EPS = "ADJUSTED_EPS"; PROVIDER_EPS = "PROVIDER_EPS"; CONSENSUS_EPS = "CONSENSUS_EPS"; MODEL_EPS = "MODEL_EPS" BOOK_VALUE = "BOOK_VALUE_PER_SHARE"; TANGIBLE_BOOK = "TANGIBLE_BOOK_PER_SHARE" GENERIC_OPERATING_MARGIN = "GENERIC_OPERATING_MARGIN"; NET_MARGIN = "NET_MARGIN" BANK_EFFICIENCY_RATIO = "BANK_EFFICIENCY_RATIO"; NET_INTEREST_MARGIN = "NET_INTEREST_MARGIN" INDUSTRIAL_DEBT = "INDUSTRIAL_DEBT"; BANK_DEPOSITS = "BANK_DEPOSITS"; BANK_BORROWINGS = "BANK_BORROWINGS"; REGULATORY_CAPITAL = "REGULATORY_CAPITAL" PE_RATIO = "PE_RATIO"; PEG_RATIO = "PEG_RATIO"; GROWTH_RATE = "GROWTH_RATE"; DIVIDEND_YIELD = "DIVIDEND_YIELD"; FX_RATE = "FX_RATE" INTRINSIC_VALUE = "INTRINSIC_VALUE"; STORY_VALUE = "STORY_VALUE"; MECHANICAL_VALUE = "MECHANICAL_VALUE" MARKET_COMPOSITE = "MARKET_COMPOSITE"; ANALYST_TARGET = "ANALYST_TARGET" HEADLINE_RETURN = "HEADLINE_RETURN"; PM_RETURN = "PM_RETURN" FINANCIAL_ISSUER_CLASSES = {"bank", "lender", "insurer", "bdc"} # ── scales and security bases ───────────────────────────────────────────────────────── UNIT, THOUSAND, MILLION, BILLION = "UNIT", "THOUSAND", "MILLION", "BILLION" _SCALE_FACTOR = {UNIT: 1.0, THOUSAND: 1e3, MILLION: 1e6, BILLION: 1e9} ORDINARY_SHARE, ADS, NA_BASIS = "ORDINARY_SHARE", "ADS", "NA" PER_SHARE, NOT_PER_SHARE = "PER_SHARE", "NONE" class IncompatibleMetricsError(ValueError): """Two metrics whose semantic signatures differ were compared or divided.""" def _f(x, default=None): try: v = float(x) return v if v == v else default except (TypeError, ValueError): return default # ── the normalization layer ─────────────────────────────────────────────────────────── def normalize(raw_value, raw_currency=None, raw_scale=UNIT, fx_to_usd=None, security_basis=NA_BASIS, target_basis=None, adr_ads_ratio=None): """raw → (normalized_value, normalized_currency, normalized_scale, security_basis, steps). One path for every metric: scale to units, convert to USD at an explicit rate, and express a per-share figure in the security basis the stock is quoted in. A non-USD value with no rate returns None: an unconvertible number must not reach a model.""" steps = [] v = _f(raw_value) if v is None: return None, (raw_currency or None), UNIT, security_basis, ["value unavailable"] sf = _SCALE_FACTOR.get(str(raw_scale or UNIT).upper(), 1.0) if sf != 1.0: v *= sf steps.append(f"scale {raw_scale} → UNIT (×{sf:,.0f})") cur = (raw_currency or "").upper() or None if cur and cur != "USD": r = _f(fx_to_usd) if not r: return None, cur, UNIT, security_basis, steps + [f"BLOCKED: {cur} value with no FX rate — cannot normalize to USD"] v *= r steps.append(f"{cur} → USD (×{r:.6f})") cur = "USD" basis = security_basis if target_basis and security_basis in (ORDINARY_SHARE, ADS) and target_basis != security_basis: k = _f(adr_ads_ratio) if not k: return None, cur, UNIT, security_basis, steps + [f"BLOCKED: {security_basis} → {target_basis} needs an ADS ratio"] v = v * k if (security_basis == ORDINARY_SHARE and target_basis == ADS) else v / k basis = target_basis steps.append(f"{security_basis} → {target_basis} (ADS ratio {k:g}:1)") return v, cur, UNIT, basis, steps def metric(concept, raw_value, *, metric_id=None, raw_currency="USD", raw_scale=UNIT, fx_to_usd=None, unit=None, accounting_basis=None, period_start=None, period_end=None, period_type=None, entity_scope="firm", perimeter="consolidated", security_basis=NA_BASIS, target_basis=None, adr_ads_ratio=None, per_share_basis=NOT_PER_SHARE, source=None, source_date=None, source_priority=None, reported_or_derived="reported", forecast_type="historical", sample_size=None, derivation=None, inputs=None): """Build one canonical metric. Returns a dict (JSON-serializable, one schema everywhere). `value` is an alias of `normalized_value`: downstream code that reads `value` therefore always reads the normalized figure.""" ratio_like = unit in ("ratio", "x", "pct", None) and concept in ( GENERIC_OPERATING_MARGIN, NET_MARGIN, PE_RATIO, PEG_RATIO, GROWTH_RATE, DIVIDEND_YIELD, BANK_EFFICIENCY_RATIO, NET_INTEREST_MARGIN, FX_RATE) or unit in ("ratio", "x") if ratio_like: nv, ncur, nscale, nbasis, steps = _f(raw_value), None, UNIT, NA_BASIS, ["dimensionless"] raw_currency = None else: nv, ncur, nscale, nbasis, steps = normalize(raw_value, raw_currency, raw_scale, fx_to_usd, security_basis, target_basis, adr_ads_ratio) return { "metric_id": metric_id or concept.lower(), "economic_concept": concept, "raw_value": _f(raw_value), "raw_currency": raw_currency, "raw_scale": str(raw_scale or UNIT).upper(), "normalized_value": nv, "normalized_currency": ncur, "normalized_scale": nscale, "value": nv, # alias: models read the normalized figure "currency": ncur, "units": unit, "accounting_basis": accounting_basis, "period_start": period_start, "period_end": period_end, "period_type": period_type, "period": period_end, "entity_scope": entity_scope, "perimeter": perimeter, "security_basis": nbasis, "adr_ads_ratio": _f(adr_ads_ratio), "per_share_basis": per_share_basis, "source": source, "source_date": source_date, "source_priority": source_priority, "reported_or_derived": reported_or_derived, "forecast_type": forecast_type, "is_consensus": forecast_type == "consensus", "is_model_estimate": forecast_type == "model_forecast", "is_company_guidance": forecast_type == "guidance", "sample_size": sample_size, "derivation": derivation, "inputs": list(inputs or []), "normalization_steps": steps, "blocked": nv is None and _f(raw_value) is not None, } # back-compat alias used by canonical.build_metrics def M(name, value, *, currency="USD", units=None, accounting_basis=None, period_type=None, period_end=None, perimeter="consolidated", source=None, source_date=None, source_priority=None, reported_or_derived="reported", forecast_or_historical="historical", derivation=None, sample_size=None, is_consensus=False, is_model_estimate=False, is_company_guidance=False, concept=None, raw_scale=UNIT, fx_to_usd=None, raw_currency=None, security_basis=NA_BASIS, target_basis=None, adr_ads_ratio=None, per_share_basis=NOT_PER_SHARE, entity_scope="firm", inputs=None, period_start=None): ft = ("consensus" if is_consensus else "model_forecast" if is_model_estimate else "guidance" if is_company_guidance else ("forecast" if forecast_or_historical == "forecast" else "historical")) return metric(concept or name.upper(), value, metric_id=name, raw_currency=(raw_currency if raw_currency is not None else currency), raw_scale=raw_scale, fx_to_usd=fx_to_usd, unit=units, accounting_basis=accounting_basis, period_start=period_start, period_end=period_end, period_type=period_type, entity_scope=entity_scope, perimeter=perimeter, security_basis=security_basis, target_basis=target_basis, adr_ads_ratio=adr_ads_ratio, per_share_basis=per_share_basis, source=source, source_date=source_date, source_priority=source_priority, reported_or_derived=reported_or_derived, forecast_type=ft, sample_size=sample_size, derivation=derivation, inputs=inputs) # ── full-signature compatibility ────────────────────────────────────────────────────── _FWD = ("NTM", "FORWARD", "FY1", "FY2", "NEXT") _ADJ = ("ADJUST", "NON-GAAP", "NONGAAP", "PRO FORMA", "CORE ", "UNDERLYING") def _is_forward(p): return any(k in str(p or "").upper() for k in _FWD) def _is_adjusted(b): return any(k in str(b or "").upper() for k in _ADJ) def compatible(a, b, kind="ratio"): """(ok, reason). kind: 'ratio' (a ÷ b, same period expected), 'growth' (periods differ by design, everything else must match), 'multiple' (price ÷ per-share flow), 'level' (same thing measured twice, e.g. two sources of revenue).""" if not a or not b: return False, "missing metric" if a.get("normalized_value") is None or b.get("normalized_value") is None: return False, f"unnormalized value ({(a.get('normalization_steps') or b.get('normalization_steps') or ['no value'])[-1]})" if kind == "level" and a.get("economic_concept") != b.get("economic_concept"): return False, f"different economic concepts: {a.get('economic_concept')} vs {b.get('economic_concept')}" if kind == "growth" and a.get("economic_concept") != b.get("economic_concept"): return False, f"growth needs one concept: {a.get('economic_concept')} vs {b.get('economic_concept')}" ca, cb = a.get("normalized_currency"), b.get("normalized_currency") if ca and cb and ca != cb: return False, f"currency {ca} vs {cb}" sa, sb = a.get("normalized_scale", UNIT), b.get("normalized_scale", UNIT) if sa != sb: return False, f"scale {sa} vs {sb}" if _is_adjusted(a.get("accounting_basis")) != _is_adjusted(b.get("accounting_basis")) and kind in ("ratio", "growth", "level"): return False, f"accounting basis '{a.get('accounting_basis')}' vs '{b.get('accounting_basis')}'" _spot = lambda p: str(p or "").upper() in ("SPOT", "POINT", "") if (kind in ("ratio", "multiple", "level") and not (_spot(a.get("period_type")) or _spot(b.get("period_type"))) and _is_forward(a.get("period_type")) != _is_forward(b.get("period_type"))): return False, f"forward vs trailing ({a.get('period_type')} vs {b.get('period_type')})" if a.get("entity_scope") != b.get("entity_scope"): return False, f"entity scope {a.get('entity_scope')} vs {b.get('entity_scope')}" pa = str(a.get("perimeter") or "consolidated").split(" (")[0].split(" ·")[0] pb = str(b.get("perimeter") or "consolidated").split(" (")[0].split(" ·")[0] if pa != pb and kind in ("growth", "level"): return False, f"perimeter {a.get('perimeter')} vs {b.get('perimeter')}" ba, bb = a.get("security_basis", NA_BASIS), b.get("security_basis", NA_BASIS) if ba != NA_BASIS and bb != NA_BASIS and ba != bb: return False, f"security basis {ba} vs {bb}" qa, qb = a.get("per_share_basis", NOT_PER_SHARE), b.get("per_share_basis", NOT_PER_SHARE) if kind in ("ratio", "growth", "level") and qa != qb: return False, f"per-share basis {qa} vs {qb}" if kind == "multiple" and not (qa == NOT_PER_SHARE or qb == PER_SHARE or qa == PER_SHARE): return False, "a multiple needs a per-share denominator" return True, "compatible" def require_compatible(a, b, kind="ratio", what=""): ok, why = compatible(a, b, kind) if not ok: raise IncompatibleMetricsError(f"{what or 'comparison'} blocked: {why}") return True def derive(concept, fn, inputs, *, kind="ratio", unit=None, period_type=None, source="derived", derivation=None, metric_id=None, accounting_basis=None, forecast_type="historical", **kw): """Compute a derived metric from NORMALIZED inputs after a compatibility gate. `inputs` is a list of metric dicts; their ids are recorded so every derived number is traceable.""" for i in range(len(inputs) - 1): require_compatible(inputs[i], inputs[i + 1], kind, what=f"{concept} from {inputs[i].get('metric_id')} and {inputs[i+1].get('metric_id')}") vals = [m["normalized_value"] for m in inputs] try: v = fn(*vals) except ZeroDivisionError: v = None out = metric(concept, v, metric_id=metric_id or concept.lower(), unit=unit or "ratio", period_type=period_type, source=source, reported_or_derived="derived", forecast_type=forecast_type, derivation=derivation or f"{concept} = f({', '.join(m.get('metric_id', '?') for m in inputs)})", inputs=[m.get("metric_id") for m in inputs], accounting_basis=accounting_basis, **kw) return out # ── typed comparisons ───────────────────────────────────────────────────────────────── # ── three separate signals ──────────────────────────────────────────────────────────── POSITIVE, NEGATIVE, NEUTRAL = "POSITIVE", "NEGATIVE", "NEUTRAL" # ── scenarios: ONE object for the renderer and the validator ────────────────────────── @dataclass class ScenarioSet: bear_return: float | None = None neutral_return: float | None = None bull_return: float | None = None bear_probability: float | None = None neutral_probability: float | None = None bull_probability: float | None = None pm_expected_return: float | None = None object_id: str = "" @property def displayed(self): return all(x is not None for x in (self.bear_probability, self.neutral_probability, self.bull_probability)) @property def complete(self): return self.displayed and all(x is not None for x in (self.bear_return, self.neutral_return, self.bull_return)) def expected_value(self): if not self.complete: return None pb, pn, pu = float(self.bear_probability or 0), float(self.neutral_probability or 0), float(self.bull_probability or 0) rb, rn, ru = float(self.bear_return or 0), float(self.neutral_return or 0), float(self.bull_return or 0) return pb * rb + pn * rn + pu * ru def formula(self): if not self.complete: return None pb, pn, pu = float(self.bear_probability or 0), float(self.neutral_probability or 0), float(self.bull_probability or 0) rb, rn, ru = float(self.bear_return or 0), float(self.neutral_return or 0), float(self.bull_return or 0) return (f"bear {pb*100:.0f}% × {rb*100:+.1f}% + neutral {pn*100:.0f}% × {rn*100:+.1f}% + " f"bull {pu*100:.0f}% × {ru*100:+.1f}% = {(self.expected_value() or 0)*100:+.2f}%") def validate(self): """(status, detail). SKIPPED only when no scenario odds exist at all.""" if not self.displayed: return "SKIPPED_NOT_APPLICABLE", "no scenario odds were produced or displayed", "NO_SCENARIO_OBJECT" ps = float(self.bear_probability or 0) + float(self.neutral_probability or 0) + float(self.bull_probability or 0) if abs(ps - 1.0) > 0.01: return "FAIL", f"probabilities sum to {ps:.3f}, not 1.0", None if not self.complete: return "FAIL", "scenario odds are displayed but one or more scenario returns are missing — the arithmetic cannot be verified", None ev = float(self.expected_value() or 0) if self.pm_expected_return is None: return "FAIL", "scenario odds displayed with no PM expected return to reconcile", None if abs(ev - float(self.pm_expected_return)) * 100 > 0.1: return "FAIL", f"{self.formula()} vs PM {self.pm_expected_return*100:+.2f}%", None return "PASS", self.formula(), None def as_dict(self): d = asdict(self) d.update(displayed=self.displayed, complete=self.complete, expected_value=self.expected_value(), formula=self.formula()) return d def scenario_set_from(p_bear, p_neutral, p_bull, r_bear, r_neutral, r_bull, pm_return): s = ScenarioSet(_f(r_bear), _f(r_neutral), _f(r_bull), _f(p_bear), _f(p_neutral), _f(p_bull), _f(pm_return)) s.object_id = object_version(s.as_dict()) return s # ── structured thesis triggers with causal polarity ─────────────────────────────────── BULL, BEAR = "BULL", "BEAR" POSITIVE_FOR_EQUITY, NEGATIVE_FOR_EQUITY = "POSITIVE_FOR_EQUITY", "NEGATIVE_FOR_EQUITY" # concept → (direction that is good for equity). "up" means a rise helps the equity. _POLARITY = { "margin": "up", "operating margin": "up", "net margin": "up", "gross margin": "up", "efficiency ratio": "down", "revenue": "up", "revenue growth": "up", "eps": "up", "earnings": "up", "eps growth": "up", "free cash flow": "up", "bookings": "up", "backlog": "up", "orders": "up", "volume": "up", "occupancy": "up", "same-store": "up", "deposit": "up", "deposits": "up", "loan growth": "up", "net interest income": "up", "net interest margin": "up", "roe": "up", "rotce": "up", "roic": "up", "book value": "up", "capital return": "up", "buyback": "up", "dividend": "up", "market share": "up", "pricing": "up", "utilisation": "up", "utilization": "up", "guidance": "up", "cet1": "up", "charge-off": "down", "charge offs": "down", "credit costs": "down", "provision": "down", "delinquenc": "down", "churn": "down", "attrition": "down", "leverage": "down", "net debt": "down", "interest expense": "down", "cost": "down", "costs": "down", "opex": "down", "capex intensity": "down", "vacancy": "down", "cap rate": "down", # intensity ratios: the numerator governs the economics, so they are listed explicitly rather # than left to a substring match that would find the denominator ("capex-to-revenue" → revenue) "capex-to-revenue": "down", "capex to revenue": "down", "capex/revenue": "down", "capex ratio": "down", "capex-to-sales": "down", "cost-to-income": "down", "cost to income": "down", "opex-to-revenue": "down", "debt-to-equity": "down", "debt-to-ebitda": "down", "net debt to ebitda": "down", "leverage ratio": "down", "days sales outstanding": "down", "inventory days": "down", "churn": "down", "dilution": "down", "competition": "down", "discount": "down", "dilution": "down", "litigation": "down", "regulatory": "down", "insider selling": "down", "short interest": "down", "inventory": "down", "days sales": "down", } _UP_WORDS = re.compile(r"\b(above|exceed|beats?|rise|rises|rising|increase|increases|improv\w+|accelerat\w+|expand\w+|grow\w+|higher|>|≥|at least|recover\w+|return to)\b", re.I) _DOWN_WORDS = re.compile(r"\b(below|miss|misses|fall|falls|falling|declin\w+|decelerat\w+|contract\w+|compress\w+|deteriorat\w+|lower|<|≤|at most|shrink\w+|rollover|roll over|weaken\w+)\b", re.I) def trigger_polarity(metric_name, direction_of_change): """POSITIVE_FOR_EQUITY / NEGATIVE_FOR_EQUITY from the metric's own economics and the direction.""" key = None low = str(metric_name or "").lower() for k in sorted(_POLARITY, key=len, reverse=True): i = low.find(k) if i < 0: continue # a term in the denominator does not govern: "capex-to-revenue" is a capex-intensity # metric, and reading it as "revenue" inverts the economics before = low[max(0, i - 8):i] if re.search(r"(?:-to-|\sto\s|/|\sper\s)$", before): continue key = k break if key is None: return None, f"no polarity known for '{metric_name}'" good = _POLARITY[key] d = str(direction_of_change or "").lower() up = "up" if d in ("up", "increase", "rise", "above", "higher", ">") else "down" if d in ("down", "decrease", "fall", "below", "lower", "<") else None if up is None: return None, f"ambiguous direction '{direction_of_change}'" pol = POSITIVE_FOR_EQUITY if up == good else NEGATIVE_FOR_EQUITY return pol, f"a {'rise' if up == 'up' else 'fall'} in {key} is {'good' if up == good else 'bad'} for the equity" @dataclass class ThesisTrigger: metric_id: str current_state: str | None condition_operator: str threshold: str direction_of_change: str scenario_destination: str rationale: str = "" polarity: str | None = None polarity_basis: str = "" status: str = "OK" source_claim: str | None = None # the claim the threshold is stated against (its unit is inherited) def validate(self): pol, basis = trigger_polarity(self.metric_id, self.direction_of_change) self.polarity, self.polarity_basis = pol, basis if pol is None: self.status = "UNVERIFIED_POLARITY" return self.status expect = BULL if pol == POSITIVE_FOR_EQUITY else BEAR self.status = "OK" if self.scenario_destination == expect else "POLARITY_REVERSED" return self.status def sentence(self): move = "rises" if self.direction_of_change in ("up", "increase", "rise", "above", "higher", ">") else "falls" dest = "the bull case" if self.scenario_destination == BULL else "the bear case" return (f"if {self.metric_id} {move} {self.condition_operator} {self.threshold}" + (f" (currently {self.current_state})" if self.current_state else "") + f", the view moves toward {dest}" + (f" — {self.rationale}" if self.rationale else "")) def as_dict(self): return asdict(self) def build_triggers(raw_list): """Validate a list of raw trigger dicts; reversed-polarity triggers are corrected to the destination their economics imply and marked, so prose cannot state a reversed causal claim.""" out = [] for t in raw_list or []: if not isinstance(t, dict) or not t.get("metric"): continue tr = ThesisTrigger(metric_id=str(t.get("metric"))[:60], current_state=(str(t.get("current")) [:40] if t.get("current") else None), condition_operator=str(t.get("operator") or "beyond")[:12], threshold=str(t.get("threshold") or "")[:40], direction_of_change=str(t.get("direction") or "")[:12].lower(), scenario_destination=(BULL if str(t.get("scenario") or "").upper().startswith("BULL") else BEAR), rationale=str(t.get("rationale") or "")[:160], source_claim=(str(t.get("source_claim")).strip()[:40] if t.get("source_claim") else None)) st = tr.validate() if st == "POLARITY_REVERSED": # the destination is corrected here, so the trigger that reaches the report is sound; # the status records that a correction happened. Check AA fails on a reversal that was # NOT corrected, which is why a corrected one must not keep the failing status. tr.scenario_destination = BULL if tr.polarity == POSITIVE_FOR_EQUITY else BEAR tr.rationale = (tr.rationale + " ").strip() + f"[direction corrected: {tr.polarity_basis}]" tr.status = "POLARITY_CORRECTED" out.append(tr) return out # ── object identity ─────────────────────────────────────────────────────────────────── def object_version(obj) -> str: try: blob = json.dumps(obj, sort_keys=True, default=str) except Exception: # noqa: BLE001 blob = repr(obj) return hashlib.sha256(blob.encode("utf-8")).hexdigest()[:16] # ── the immutable analysis context ──────────────────────────────────────────────────── # ── §15 scenario probabilities: an auditable decomposition, not one opaque number ────── @dataclass class ScenarioProbabilities: """Where each scenario's odds came from, and what was allowed to move them. The panel manager states the odds, so the base is the manager's own judgement. Any tilt attributed to a quantitative model is recorded separately and bounded by that model's reliability: an UNRELIABLE model's adjustment is exactly zero, so it cannot move the odds through the back door of an LLM's prose. """ base: dict quant_adjustment: dict = field(default_factory=dict) fundamental_adjustment: dict = field(default_factory=dict) macro_adjustment: dict = field(default_factory=dict) quant_cap: float = 0.0 quant_state: str | None = None notes: list = field(default_factory=list) KEYS = ("bear", "neutral", "bull") def _clamped_quant(self): out = {} for k in self.KEYS: v = float(self.quant_adjustment.get(k, 0.0) or 0.0) c = float(self.quant_cap or 0.0) out[k] = max(-c, min(c, v)) return out def final(self): q = self._clamped_quant() raw = {} for k in self.KEYS: raw[k] = (float(self.base.get(k, 0.0) or 0.0) + q[k] + float(self.fundamental_adjustment.get(k, 0.0) or 0.0) + float(self.macro_adjustment.get(k, 0.0) or 0.0)) raw[k] = max(0.01, raw[k]) tot = sum(raw.values()) or 1.0 return {k: raw[k] / tot for k in self.KEYS} def audit(self): """The record the report shows and the validator checks.""" q = self._clamped_quant() clipped = [k for k in self.KEYS if abs(float(self.quant_adjustment.get(k, 0.0) or 0.0)) > abs(q[k]) + 1e-12] return {"base_probability": {k: self.base.get(k) for k in self.KEYS}, "quant_adjustment": q, "quant_adjustment_requested": dict(self.quant_adjustment), "quant_adjustment_cap": self.quant_cap, "quant_reliability_state": self.quant_state, "fundamental_adjustment": dict(self.fundamental_adjustment), "macro_adjustment": dict(self.macro_adjustment), "final_probability": self.final(), "quant_influence_blocked": self.quant_cap == 0.0, "clipped_by_reliability": clipped, "notes": list(self.notes)} def scenario_probabilities(base, quant_reliability=None, quant_adjustment=None, fundamental_adjustment=None, macro_adjustment=None): rel = quant_reliability or {} cap = float(rel.get("scenario_adjustment_cap") or 0.0) sp = ScenarioProbabilities(base=dict(base or {}), quant_adjustment=dict(quant_adjustment or {}), fundamental_adjustment=dict(fundamental_adjustment or {}), macro_adjustment=dict(macro_adjustment or {}), quant_cap=cap, quant_state=rel.get("state")) if cap == 0.0 and any(abs(float(v or 0)) > 0 for v in (quant_adjustment or {}).values()): sp.notes.append(f"quant reliability {rel.get('state')} — the requested scenario tilt was set to zero") return sp # ── §16 evidence lineage for a panel conclusion ─────────────────────────────────────── @dataclass class EvidenceItem: evidence_id: str label: str eligible: bool reason: str = "" value: object = None class EvidenceLedger: """What a conclusion was allowed to rest on, and what it actually rested on.""" def __init__(self): self.available: dict[str, EvidenceItem] = {} self.used: dict[str, list] = {} def offer(self, evidence_id, label, eligible=True, reason="", value=None): self.available[evidence_id] = EvidenceItem(evidence_id, label, bool(eligible), reason, value) return self.available[evidence_id] def cite(self, conclusion, *evidence_ids): self.used.setdefault(conclusion, []).extend(evidence_ids) def audit(self): problems = [] for conclusion, ids in self.used.items(): for eid in ids: item = self.available.get(eid) if item is None: problems.append(f"{conclusion} cites unknown evidence '{eid}'") elif not item.eligible: problems.append(f"{conclusion} cites '{item.label}', which is not eligible as evidence: {item.reason}") return {"ok": not problems, "problems": problems, "available": {k: {"label": v.label, "eligible": v.eligible, "reason": v.reason} for k, v in self.available.items()}, "used": {k: list(v) for k, v in self.used.items()}}