Blame for engine/state.py

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

# ── period classes and bases ──────────────────────────────────────────────────────────
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"

# quality states
VALIDATED, RECONCILED, UNVERIFIED, DERIVED, MODEL_ESTIMATE, WITHHELD, UNAVAILABLE = (
    "VALIDATED", "RECONCILED", "UNVERIFIED", "DERIVED", "MODEL_ESTIMATE", "WITHHELD", "UNAVAILABLE")
PARTIALLY_RECONCILED = "PARTIALLY_RECONCILED"     # some, not all, of the periods a trailing figure sums were checked


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             # period end (ISO date) or as-of date
    scope: str = "CONSOLIDATED"
    perimeter: str = "consolidated"
    security_basis: str = "NA"            # ORDINARY_SHARE / ADS for per-share values
    per_share: bool = False
    source: str = ""
    source_date: str | None = None
    derivation: str | None = None
    quality_status: str = VALIDATED
    unit: str = "USD"                     # USD, ratio, x, per_share, count, pct_points
    # non-GAAP semantics: "ADJUSTED" alone is not a definition. Two adjusted figures compare only
    # when their definitions match or an explicit reconciliation exists (definitions.py)
    adjustment_definition_id: str | None = None   # e.g. PROVIDER_REPORTED_EPS, SELL_SIDE_CONSENSUS_EPS
    adjustment_provider: str | None = None
    adjustment_version: str | None = None
    excluded_items: tuple = ()
    definition_effective_date: str | None = None
    # the actual fiscal period, not just its class: NTM and FY1 are both "forward" but need not overlap
    period_start: str | None = None
    period_end: str | None = None
    forecast_horizon_days: int | None = None
    # what the number IS (quantities.py): a margin level and a margin change can share a decimal, never a type
    quantity_type: str | None = None              # LEVEL, GROWTH_RATE, MARGIN_LEVEL, MARGIN_CHANGE, …
    display_unit: str | None = None               # PERCENT, BASIS_POINTS, PERCENTAGE_POINTS, USD_PER_SHARE, …
    # the corporate action that defines this figure's perimeter (corporate_actions.py); None = no perimeter event
    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:          # NaN
        v = None
    if "quantity_type" not in kw:          # what the number IS follows from its concept (quantities.py)
        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


# perimeter: which business the figure describes. A completed separation leaves the trailing
# statements on the old perimeter while consensus and the forecast describe the continuing company.
CONSOLIDATED, PRE_SEPARATION, CONTINUING = "consolidated", "pre-separation", "continuing company"


def _same_perimeter(a, b):
    pa, pb = a.perimeter or CONSOLIDATED, b.perimeter or CONSOLIDATED
    # "consolidated" (no separation recorded) and "continuing company" describe the same business today
    norm = lambda p: CONTINUING if p == CONSOLIDATED else p
    return norm(pa) == norm(pb)


# ── compatibility: the ONLY way two metrics may be compared ───────────────────────────
COMPARISON_TYPES = ("level", "margin_relationship", "growth", "ratio", "benchmark", "valuation", "trend", "scenario", "path")
# a scenario concept is the same economic quantity under a stated condition (what today's price requires): it compares
# with its base concept, never with another quantity
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"):
        # scenario: a scenario figure against its base concept at the same horizon (break-even vs story revenue);
        # path: two levels of one quantity at two points of one path (starting revenue → break-even revenue)
        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":
        # the same quantity at two points in time (trailing vs forward): one concept, one basis, ONE business
        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):
        # two adjusted figures are comparable only on one adjustment definition (or a declared reconciliation)
        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


# ── canonical selection: one explicit choice per use case ─────────────────────────────
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 ────────────────────────────────────────────────────────────────────────────
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")
# what each stage validates, and therefore seals
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:
    # declared for readers and type checkers; set through object.__setattr__ in __init__
    _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

    # ── writes ────────────────────────────────────────────────────────────────────────
    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}")

    # ── reads ─────────────────────────────────────────────────────────────────────────
    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)

    # ── stages ────────────────────────────────────────────────────────────────────────
    @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

    # ── serialisation ─────────────────────────────────────────────────────────────────
    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())