| """AnalysisState: the one authoritative object for a run. |
| |
| Every published concept has exactly one location here (``state.forecast.eps``, |
| ``state.pricing.market_composite``, ``state.blend.headline_return`` …). Legacy dicts produced by |
| the older modules are raw inputs to adapters that write this object; no persona, validator, |
| summary generator, renderer, claim audit or recommendation mapper reads them. |
| |
| Three guarantees are enforced by the object itself rather than by convention: |
| |
| * **Typed metrics.** Every number is a ``Metric`` carrying concept, period class, accounting basis, |
| scope, currency, security basis, source, date, derivation and quality. Two metrics are only |
| compared through ``compatible()``; an incompatible pair raises instead of producing a number. |
| * **Stage order.** ``advance()`` moves through DATA_VALIDATED → FORECAST_VALIDATED → |
| ANALYSIS_FINALIZED → REPORT_VALIDATED → RENDERED, only on a passing validation result, and only |
| in order. |
| * **No mutation after validation.** Passing a stage seals the namespaces that stage validated. |
| Writing to a sealed namespace raises ``SealedStateError``; after ANALYSIS_FINALIZED the whole |
| state is frozen and carries a content version. |
| """ |
| from __future__ import annotations |
|
|
| import copy |
| import hashlib |
| import json |
| from dataclasses import asdict, dataclass, replace |
| from types import MappingProxyType |
|
|
| |
| TTM, FY, NTM, FY1, FY2, SPOT, INTRINSIC, POINT = "TTM", "FY", "NTM", "FY1", "FY2", "SPOT", "INTRINSIC", "POINT" |
| TRAILING = frozenset({TTM, FY}) |
| FORWARD = frozenset({NTM, FY1, FY2}) |
|
|
| GAAP, IFRS, ADJUSTED, MODEL, MARKET, NA = "GAAP", "IFRS", "ADJUSTED", "MODEL", "MARKET", "NA" |
| REPORTED_BASES = frozenset({GAAP, IFRS}) |
|
|
| ORDINARY_SHARE, ADS = "ORDINARY_SHARE", "ADS" |
|
|
| |
| VALIDATED, RECONCILED, UNVERIFIED, DERIVED, MODEL_ESTIMATE, WITHHELD, UNAVAILABLE = ( |
| "VALIDATED", "RECONCILED", "UNVERIFIED", "DERIVED", "MODEL_ESTIMATE", "WITHHELD", "UNAVAILABLE") |
| PARTIALLY_RECONCILED = "PARTIALLY_RECONCILED" |
|
|
|
|
| class IncompatibleMetrics(Exception): |
| """Two metrics were compared that do not describe comparable quantities.""" |
|
|
|
|
| class SealedStateError(Exception): |
| """A namespace was written after the stage that validated it had passed.""" |
|
|
|
|
| class StageOrderError(Exception): |
| """A stage was entered before the stage it depends on had passed.""" |
|
|
|
|
| @dataclass(frozen=True) |
| class Metric: |
| metric_id: str |
| economic_concept: str |
| value: float | None |
| period_type: str |
| accounting_basis: str = NA |
| currency: str | None = "USD" |
| scale: str = "UNIT" |
| period: str | None = None |
| scope: str = "CONSOLIDATED" |
| perimeter: str = "consolidated" |
| security_basis: str = "NA" |
| per_share: bool = False |
| source: str = "" |
| source_date: str | None = None |
| derivation: str | None = None |
| quality_status: str = VALIDATED |
| unit: str = "USD" |
| |
| |
| adjustment_definition_id: str | None = None |
| adjustment_provider: str | None = None |
| adjustment_version: str | None = None |
| excluded_items: tuple = () |
| definition_effective_date: str | None = None |
| |
| period_start: str | None = None |
| period_end: str | None = None |
| forecast_horizon_days: int | None = None |
| |
| quantity_type: str | None = None |
| display_unit: str | None = None |
| |
| corporate_action_id: str | None = None |
|
|
| @property |
| def period_class(self): |
| if self.period_type in FORWARD: |
| return "FORWARD" |
| if self.period_type in TRAILING: |
| return "TRAILING" |
| return self.period_type |
|
|
| @property |
| def basis_class(self): |
| if self.accounting_basis in REPORTED_BASES: |
| return "REPORTED" |
| return self.accounting_basis |
|
|
| def to_dict(self): |
| d = asdict(self) |
| d["period_class"] = self.period_class |
| return d |
|
|
| def with_value(self, value, **kw): |
| return replace(self, value=value, **kw) |
|
|
|
|
| def metric(metric_id, concept, value, period_type, **kw): |
| try: |
| v = None if value is None else float(value) |
| except (TypeError, ValueError): |
| v = None |
| if v is not None and v != v: |
| v = None |
| if "quantity_type" not in kw: |
| import quantities as _q |
| kw["quantity_type"] = _q.CONCEPT_QUANTITY.get(concept) or _EXTRA_QUANTITY.get(concept) |
| return Metric(metric_id=metric_id, economic_concept=concept, value=v, period_type=period_type, **kw) |
|
|
|
|
| _EXTRA_QUANTITY = {"PRICE": "LEVEL", "MARKET_CAP": "LEVEL", "SHARES": "LEVEL", "DILUTED_SHARES": "LEVEL", |
| "DIVIDEND_YIELD": "YIELD", "ROE": "MARGIN_LEVEL", "ROA": "MARGIN_LEVEL", "PE": "LEVEL", "PEG": "LEVEL", |
| "INTRINSIC_VALUE": "LEVEL", "MARKET_PRICING": "LEVEL", "TARGET_PRICE": "LEVEL", |
| "PRETAX_INCOME": "LEVEL", "NET_INCOME": "LEVEL", "BOOK_EQUITY": "LEVEL", "EPS_GAP": "CHANGE", |
| "COST_OF_CAPITAL": "YIELD", "TERMINAL_GROWTH": "GROWTH_RATE", "TARGET_GAP": "CHANGE", |
| "BREAK_EVEN_REVENUE": "LEVEL", "BREAK_EVEN_OPERATING_MARGIN": "MARGIN_LEVEL", |
| "BREAK_EVEN_REVENUE_CAGR": "GROWTH_RATE", "MARKET_SIZE": "LEVEL", "YEARS": "LEVEL", |
| "MARKET_SHARE": "MARGIN_LEVEL"} |
|
|
|
|
| def stamp_perimeter(state, action_id, namespaces): |
| """Attach the perimeter's corporate_action_id to every Metric on a separated perimeter (pre-separation or |
| continuing company), in the given namespaces, including Metrics nested one level in dicts.""" |
| if not action_id: |
| return 0 |
| n = 0 |
| split = ("pre-separation", "continuing company") |
|
|
| def fix(v): |
| nonlocal n |
| if isinstance(v, Metric) and v.perimeter in split and v.corporate_action_id != action_id: |
| n += 1 |
| return replace(v, corporate_action_id=action_id) |
| return v |
| for ns in namespaces: |
| space = getattr(state, ns, None) |
| if space is None: |
| continue |
| for k in list(space.keys()): |
| v = space.get(k) |
| if isinstance(v, Metric): |
| nv = fix(v) |
| if nv is not v: |
| space.set(k, nv) |
| elif isinstance(v, dict) and any(isinstance(x, Metric) for x in v.values()): |
| nd = {kk: fix(x) for kk, x in v.items()} |
| space.set(k, nd) |
| return n |
|
|
|
|
| |
| |
| CONSOLIDATED, PRE_SEPARATION, CONTINUING = "consolidated", "pre-separation", "continuing company" |
|
|
|
|
| def _same_perimeter(a, b): |
| pa, pb = a.perimeter or CONSOLIDATED, b.perimeter or CONSOLIDATED |
| |
| norm = lambda p: CONTINUING if p == CONSOLIDATED else p |
| return norm(pa) == norm(pb) |
|
|
|
|
| |
| COMPARISON_TYPES = ("level", "margin_relationship", "growth", "ratio", "benchmark", "valuation", "trend", "scenario", "path") |
| |
| |
| SCENARIO_BASE = {"BREAK_EVEN_REVENUE": "REVENUE", "BREAK_EVEN_OPERATING_MARGIN": "OPERATING_MARGIN", |
| "BREAK_EVEN_REVENUE_CAGR": "REVENUE_GROWTH"} |
|
|
|
|
| def base_concept(c): |
| return SCENARIO_BASE.get(c, c) |
|
|
|
|
| def compatible(a: Metric, b: Metric, comparison_type: str): |
| """(ok, reasons). Mandatory before any comparison or derivation. |
| |
| level same concept, same period class, same basis class, same scope, currency, |
| security basis — e.g. model EPS vs consensus EPS |
| margin_relationship different concepts of one income statement: same period class, same basis |
| class, same scope — e.g. forward net margin vs forward operating margin |
| growth same concept and basis; one trailing, one forward (a forecast vs its base) |
| ratio numerator and denominator of a multiple: same currency and security basis; |
| a SPOT price may divide a forward or trailing per-share figure |
| benchmark a company figure against an industry or peer aggregate of the SAME concept, period class |
| and basis class (scope and perimeter may differ: the benchmark is another universe) |
| valuation two per-share values (price, intrinsic value, composite, target): same currency and |
| security basis; concepts may differ |
| """ |
| if comparison_type not in COMPARISON_TYPES: |
| raise ValueError(f"unknown comparison type {comparison_type!r}") |
| r = [] |
| if a is None or b is None or a.value is None or b.value is None: |
| return False, ["a value is missing"] |
| if comparison_type in ("scenario", "path"): |
| |
| |
| if base_concept(a.economic_concept) != base_concept(b.economic_concept): |
| r.append(f"concept {a.economic_concept} ≠ {b.economic_concept}") |
| if a.quantity_type and b.quantity_type and a.quantity_type != b.quantity_type: |
| r.append(f"quantity {a.quantity_type} ≠ {b.quantity_type}") |
| if comparison_type == "scenario" and a.period_class != b.period_class: |
| r.append(f"period {a.period_class} ≠ {b.period_class}") |
| if (a.currency or "") != (b.currency or "") and a.unit not in ("ratio", "x") and b.unit not in ("ratio", "x"): |
| r.append(f"currency {a.currency} ≠ {b.currency}") |
| if not _same_perimeter(a, b): |
| r.append(f"perimeter {a.perimeter} ≠ {b.perimeter}") |
| return (not r), r |
| if comparison_type in ("level", "growth", "benchmark") and a.economic_concept != b.economic_concept: |
| r.append(f"concept {a.economic_concept} ≠ {b.economic_concept}") |
| if comparison_type in ("level", "growth", "benchmark") and a.quantity_type and b.quantity_type \ |
| and a.quantity_type != b.quantity_type: |
| r.append(f"quantity {a.quantity_type} ≠ {b.quantity_type}") |
| if comparison_type == "benchmark": |
| if a.period_class != b.period_class: |
| r.append(f"period {a.period_class} ≠ {b.period_class}") |
| if MODEL not in (a.basis_class, b.basis_class) and a.basis_class != b.basis_class: |
| r.append(f"accounting basis {a.accounting_basis} ≠ {b.accounting_basis}") |
| return (not r), r |
| if comparison_type == "trend": |
| |
| if a.economic_concept != b.economic_concept: |
| r.append(f"concept {a.economic_concept} ≠ {b.economic_concept}") |
| if a.quantity_type and b.quantity_type and a.quantity_type != b.quantity_type: |
| r.append(f"quantity {a.quantity_type} ≠ {b.quantity_type}") |
| if MODEL not in (a.basis_class, b.basis_class) and a.basis_class != b.basis_class: |
| r.append(f"accounting basis {a.accounting_basis} ≠ {b.accounting_basis}") |
| if a.scope != b.scope: |
| r.append(f"scope {a.scope} ≠ {b.scope}") |
| if not _same_perimeter(a, b): |
| r.append(f"perimeter {a.perimeter} ≠ {b.perimeter}") |
| return (not r), r |
| if comparison_type == "valuation": |
| if (a.currency or "") != (b.currency or ""): |
| r.append(f"currency {a.currency} ≠ {b.currency}") |
| if not (a.per_share and b.per_share): |
| r.append("a valuation comparison needs two per-share values") |
| elif a.security_basis != b.security_basis and "NA" not in (a.security_basis, b.security_basis): |
| r.append(f"security basis {a.security_basis} ≠ {b.security_basis}") |
| return (not r), r |
| if comparison_type in ("level", "margin_relationship") and a.period_class != b.period_class: |
| r.append(f"period {a.period_class} ≠ {b.period_class}") |
| if comparison_type == "growth" and not ({a.period_class, b.period_class} == {"FORWARD", "TRAILING"}): |
| r.append("growth needs one trailing base and one forward figure") |
| if comparison_type in ("level", "margin_relationship", "growth"): |
| ba, bb = a.basis_class, b.basis_class |
| if MODEL not in (ba, bb) and ba != bb: |
| r.append(f"accounting basis {a.accounting_basis} ≠ {b.accounting_basis}") |
| if a.scope != b.scope: |
| r.append(f"scope {a.scope} ≠ {b.scope}") |
| if comparison_type in ("level", "growth") and ADJUSTED in (a.accounting_basis, b.accounting_basis): |
| |
| import definitions as _defs |
| ok_d, why_d = _defs.definitions_compatible(a, b) |
| if not ok_d: |
| r.append(why_d) |
| if comparison_type == "level" and a.period_class == "FORWARD" and b.period_class == "FORWARD": |
| import definitions as _defs |
| ok_p, why_p = _defs.periods_overlap(a, b) |
| if not ok_p: |
| r.append(why_p) |
| if comparison_type in ("level", "growth", "ratio"): |
| if (a.currency or "") != (b.currency or "") and a.unit not in ("ratio", "x") and b.unit not in ("ratio", "x"): |
| r.append(f"currency {a.currency} ≠ {b.currency}") |
| if a.per_share and b.per_share and a.security_basis != b.security_basis: |
| r.append(f"security basis {a.security_basis} ≠ {b.security_basis}") |
| if not _same_perimeter(a, b): |
| r.append(f"perimeter {a.perimeter} ≠ {b.perimeter}") |
| if comparison_type == "ratio": |
| if SPOT not in (a.period_type, b.period_type) and a.period_class != b.period_class: |
| r.append("a multiple needs a spot price or matching periods") |
| return (not r), r |
|
|
|
|
| def require_compatible(a, b, comparison_type): |
| ok, reasons = compatible(a, b, comparison_type) |
| if not ok: |
| raise IncompatibleMetrics(f"{a.metric_id if a else None} vs {b.metric_id if b else None} " |
| f"({comparison_type}): " + "; ".join(reasons)) |
| return True |
|
|
|
|
| |
| SELECTION_POLICY = { |
| "EPS_GROWTH_BASE": ("historical.eps_adjusted_ttm", |
| "the EPS forecast is anchored to sell-side estimates, which are on the adjusted basis"), |
| "FORWARD_PE_DENOMINATOR": ("forecast.eps", "the reconciled 12-month EPS forecast"), |
| "TRAILING_PE_DENOMINATOR": ("historical.eps_adjusted_ttm", |
| "same basis as the forward EPS, so trailing and forward multiples compare like for like"), |
| "GAAP_PE_DENOMINATOR": ("historical.eps_gaap_ttm", "reported net income per diluted share"), |
| "STORY_START_REVENUE": ("historical.revenue_ttm", "the reconciled historical revenue every model uses"), |
| "STORY_START_MARGIN": ("historical.operating_margin_ttm", |
| "the reconciled historical operating margin every model uses"), |
| "FORECAST_GROWTH_BASE_REVENUE": ("historical.revenue_ttm", "the reconciled historical revenue"), |
| "FORECAST_ACCOUNTING_CHECK": ("forecast.operating_margin vs forecast.net_margin", |
| "both from the one reconciled forecast state, same period and basis"), |
| "INTRINSIC_VALUE": ("valuation.intrinsic_value", "the value produced by the engine selected for this issuer"), |
| "MARKET_PRICING": ("pricing.market_composite", "multiples and analyst targets, blended"), |
| "HEADLINE_RETURN": ("blend.headline_return", "the blended model-implied 12-month price return"), |
| "PM_RETURN": ("scenarios.pm_return", "the panel's probability-weighted scenario return"), |
| "DIVIDEND_YIELD": ("market.dividend_yield", "trailing dividends paid ÷ market capitalisation"), |
| "UNCERTAINTY_RANGE": ("uncertainty.scenario_range", "bear and bull scenario returns"), |
| } |
|
|
|
|
| |
| STAGES = ("INITIAL", "DATA_VALIDATED", "FORECAST_VALIDATED", "ANALYSIS_FINALIZED", "REPORT_VALIDATED", "RENDERED") |
| NAMESPACES = ("identity", "issuer", "market", "historical", "consensus", "segments", "guidance", "forecast", "valuation", |
| "pricing", "quant", "macro", "technical", "sentiment", "insiders", "ownership", "leadership", "corporate_actions", |
| "context", "relations", "evidence", "scenarios", "blend", "recommendation", "uncertainty", "claims", "provenance") |
| |
| SEALS = { |
| "DATA_VALIDATED": ("identity", "issuer", "market", "historical", "consensus", "segments", "guidance"), |
| "FORECAST_VALIDATED": ("forecast", "valuation", "pricing", "quant", "macro", "technical", "sentiment", |
| "insiders", "ownership", "leadership", "corporate_actions", "context", "relations", "evidence"), |
| "ANALYSIS_FINALIZED": NAMESPACES, |
| } |
|
|
|
|
| def _freeze(o): |
| """Read-only deep copy: dicts become mapping proxies, lists become tuples.""" |
| if isinstance(o, (dict, MappingProxyType)): |
| return MappingProxyType({k: _freeze(v) for k, v in o.items()}) |
| if isinstance(o, (list, tuple)): |
| return tuple(_freeze(v) for v in o) |
| if isinstance(o, set): |
| return frozenset(o) |
| return o |
|
|
|
|
| def _jsonable(o): |
| if isinstance(o, Metric): |
| return o.to_dict() |
| if isinstance(o, (dict, MappingProxyType)): |
| return {k: _jsonable(v) for k, v in o.items()} |
| if isinstance(o, (list, tuple)): |
| return [_jsonable(v) for v in o] |
| if isinstance(o, float) and o != o: |
| return None |
| return o |
|
|
|
|
| class Namespace: |
| __slots__ = ("_name", "_data", "_state") |
|
|
| def __init__(self, name, state): |
| object.__setattr__(self, "_name", name) |
| object.__setattr__(self, "_data", {}) |
| object.__setattr__(self, "_state", state) |
|
|
| def __getattr__(self, k): |
| try: |
| return self._data[k] |
| except KeyError: |
| raise AttributeError(f"{self._name}.{k} is not set") from None |
|
|
| def __setattr__(self, k, v): |
| self._state._check_writable(self._name, k) |
| self._data[k] = v |
|
|
| def get(self, k, default=None): |
| return self._data.get(k, default) |
|
|
| def set(self, k, v): |
| setattr(self, k, v) |
| return v |
|
|
| def keys(self): |
| return self._data.keys() |
|
|
| def items(self): |
| return self._data.items() |
|
|
| def __contains__(self, k): |
| return k in self._data |
|
|
|
|
| class AnalysisState: |
| |
| _stage: str |
| _sealed: set |
| _frozen: bool |
| _history: list |
| version: str | None |
| identity: Namespace; issuer: Namespace; market: Namespace; historical: Namespace |
| consensus: Namespace; forecast: Namespace; valuation: Namespace; pricing: Namespace |
| quant: Namespace; macro: Namespace; technical: Namespace; sentiment: Namespace |
| insiders: Namespace; leadership: Namespace; corporate_actions: Namespace; context: Namespace |
| scenarios: Namespace; blend: Namespace; recommendation: Namespace; uncertainty: Namespace |
| claims: Namespace; provenance: Namespace |
|
|
| def __init__(self, ticker, run_id=None): |
| object.__setattr__(self, "_stage", "INITIAL") |
| object.__setattr__(self, "_sealed", set()) |
| object.__setattr__(self, "_frozen", False) |
| object.__setattr__(self, "_history", []) |
| object.__setattr__(self, "version", None) |
| for ns in NAMESPACES: |
| object.__setattr__(self, ns, Namespace(ns, self)) |
| self.identity.ticker = ticker |
| self.identity.run_id = run_id |
|
|
| |
| def __setattr__(self, k, v): |
| raise AttributeError("write through a namespace, e.g. state.forecast.eps = …") |
|
|
| def _check_writable(self, ns, key): |
| if self._frozen: |
| raise SealedStateError(f"state is frozen at {self._stage}; cannot set {ns}.{key}") |
| if ns in self._sealed: |
| raise SealedStateError(f"{ns} was sealed when its validation stage passed; cannot set {ns}.{key}") |
|
|
| |
| def get(self, path, default=None): |
| """Resolve 'forecast.eps' or 'valuation.story.assumptions.margin_target'.""" |
| parts = path.split(".") |
| ns = getattr(self, parts[0], None) |
| if not isinstance(ns, Namespace): |
| return default |
| cur = ns.get(parts[1]) if len(parts) > 1 else ns |
| for p in parts[2:]: |
| if cur is None: |
| return default |
| if isinstance(cur, Metric): |
| cur = getattr(cur, p, None) |
| elif isinstance(cur, (dict, MappingProxyType)): |
| cur = cur.get(p) |
| elif isinstance(cur, (list, tuple)) and p.isdigit(): |
| i = int(p) |
| cur = cur[i] if i < len(cur) else None |
| else: |
| cur = getattr(cur, p, None) |
| return default if cur is None else cur |
|
|
| def value(self, path, default=None): |
| v = self.get(path) |
| if isinstance(v, Metric): |
| return default if v.value is None else v.value |
| return default if v is None else v |
|
|
| def select(self, use_case): |
| """The canonical metric for a use case (SELECTION_POLICY). Consumers never choose.""" |
| path, _reason = SELECTION_POLICY[use_case] |
| return self.get(path) |
|
|
| |
| @property |
| def stage(self): |
| return self._stage |
|
|
| def require_stage(self, stage): |
| if STAGES.index(self._stage) < STAGES.index(stage): |
| raise StageOrderError(f"requires {stage}; state is at {self._stage}") |
|
|
| def advance(self, stage, validation): |
| """Enter ``stage`` only in order and only when its validation passed.""" |
| cur = STAGES.index(self._stage) |
| nxt = STAGES.index(stage) |
| if nxt != cur + 1: |
| raise StageOrderError(f"cannot move from {self._stage} to {stage}") |
| if not getattr(validation, "passed", False): |
| raise StageOrderError(f"{stage} validation did not pass: {getattr(validation, 'blocking', None)}") |
| for ns in SEALS.get(stage, ()): |
| if ns not in self._sealed: |
| n = getattr(self, ns) |
| object.__setattr__(n, "_data", {k: _freeze(v) for k, v in n._data.items()}) |
| self._sealed.add(ns) |
| self._history.append({"stage": stage, "checks": len(getattr(validation, "checks", []) or [])}) |
| object.__setattr__(self, "_stage", stage) |
| if stage == "ANALYSIS_FINALIZED": |
| object.__setattr__(self, "_frozen", True) |
| object.__setattr__(self, "version", self.content_hash()) |
| return self |
|
|
| def mark(self, stage): |
| """Post-finalization stages (REPORT_VALIDATED, RENDERED) record progress without writes.""" |
| if STAGES.index(stage) != STAGES.index(self._stage) + 1: |
| raise StageOrderError(f"cannot move from {self._stage} to {stage}") |
| object.__setattr__(self, "_stage", stage) |
|
|
| @property |
| def frozen(self): |
| return self._frozen |
|
|
| |
| def to_dict(self): |
| return {ns: _jsonable(dict(getattr(self, ns)._data)) for ns in NAMESPACES} |
|
|
| def content_hash(self): |
| blob = json.dumps(self.to_dict(), sort_keys=True, default=str) |
| return hashlib.sha256(blob.encode()).hexdigest()[:16] |
|
|
| def snapshot(self): |
| """A deep, JSON-safe copy (for the QA record).""" |
| return copy.deepcopy(self.to_dict()) |
|
|
