| """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 |
| 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 |
|
|
| |
| 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"} |
|
|
| |
| 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 |
|
|
|
|
| |
| 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, |
| "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, |
| } |
|
|
|
|
| |
| 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) |
|
|
|
|
| |
| _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 |
|
|
|
|
| |
|
|
|
|
|
|
|
|
|
|
|
|
| |
| POSITIVE, NEGATIVE, NEUTRAL = "POSITIVE", "NEGATIVE", "NEUTRAL" |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| |
| @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 |
|
|
|
|
| |
| BULL, BEAR = "BULL", "BEAR" |
| POSITIVE_FOR_EQUITY, NEGATIVE_FOR_EQUITY = "POSITIVE_FOR_EQUITY", "NEGATIVE_FOR_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", |
| |
| |
| "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 |
| |
| |
| 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 |
|
|
| 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": |
| |
| |
| |
| 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 |
|
|
|
|
| |
| def object_version(obj) -> str: |
| try: |
| blob = json.dumps(obj, sort_keys=True, default=str) |
| except Exception: |
| blob = repr(obj) |
| return hashlib.sha256(blob.encode("utf-8")).hexdigest()[:16] |
|
|
|
|
| |
|
|
|
|
| |
| @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 |
|
|
|
|
| |
| @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()}} |
|
|
