siloqy/prod/clean_arch/sim/mock_stack.py

"""Fully mocked DITA stack for isolated policy / trade / account testing.

The goal is to approximate the live PINK/BLUE boundaries without touching any
real exchange, Hazelcast cluster, ClickHouse instance, or logs.
"""

from __future__ import annotations

from collections import defaultdict
from copy import deepcopy
from dataclasses import dataclass, field
from datetime import datetime, timedelta
from typing import Any, Dict, Iterable, List, Optional
import math
import random

from prod.clean_arch.dita import (
    AccountProjection,
    AccountSnapshot,
    DecisionAction,
    DecisionConfig,
    DecisionContext,
    DecisionEngine,
    Intent,
    IntentContext,
    IntentEngine,
    TradeExecutor,
    TradePosition,
    TradeSide,
    TradeStage,
    DitaObservabilityNamespace,
    LEGACY_ANOMALY_SENSOR_KEY,
)
from prod.clean_arch.ports.data_feed import MarketSnapshot


@dataclass(frozen=True)
class NetworkProfile:
    """Fault model for the mock transport boundary."""

    drop_rate: float = 0.0
    duplicate_rate: float = 0.0
    jitter_ms: float = 0.0


@dataclass(frozen=True)
class ChaosProfile:
    """Deliberate anomaly model for aggressive failure simulation."""

    hang_entry_rate: float = 0.0
    hang_exit_rate: float = 0.0
    stale_account_rate: float = 0.0
    duplicate_terminal_rate: float = 0.0
    missing_terminal_rate: float = 0.0
    orphan_close_rate: float = 0.0
    reorder_account_rate: float = 0.0


INJECTED_ANOMALIES = {
    "hung_entry",
    "hung_exit",
    "network_drop",
    "missing_terminal_row",
    "duplicate_terminal_row",
    "orphaned_close",
    "account_write_reordered",
    "invalid_market_snapshot",
    "compound_fault",
}

EMERGENT_ANOMALIES = {
    "stale_account_projection",
    "closed_position_still_present",
    "rejected_order",
    "invalid_trade_state",
    "invalid_account_state",
    "exchange_executor_divergence",
    "decision_intent_mismatch",
    "duplicate_trade_id",
    "unknown_invariant_violation",
    "compound_fault",
}

ANOMALY_SENSOR_MAP = {
    "network_drop": "m7_transport_integrity",
    "hung_entry": "m8_execution_integrity",
    "hung_exit": "m8_execution_integrity",
    "missing_terminal_row": "m9_terminal_integrity",
    "duplicate_terminal_row": "m9_terminal_integrity",
    "orphaned_close": "m10_exchange_state_integrity",
    "closed_position_still_present": "m10_exchange_state_integrity",
    "stale_account_projection": "m11_account_projection_integrity",
    "account_write_reordered": "m11_account_projection_integrity",
    "rejected_order": "m12_order_acceptance_integrity",
    "invalid_market_snapshot": "m13_invariant_integrity",
    "invalid_trade_state": "m13_invariant_integrity",
    "invalid_account_state": "m13_invariant_integrity",
    "exchange_executor_divergence": "m13_invariant_integrity",
    "decision_intent_mismatch": "m13_invariant_integrity",
    "duplicate_trade_id": "m13_invariant_integrity",
    "unknown_invariant_violation": "m13_invariant_integrity",
    "compound_fault": "m13_invariant_integrity",
}

ANOMALY_SENSOR_SEVERITY = {
    "network_drop": 0.20,
    "hung_entry": 0.25,
    "hung_exit": 0.25,
    "missing_terminal_row": 0.30,
    "duplicate_terminal_row": 0.15,
    "orphaned_close": 0.30,
    "closed_position_still_present": 0.20,
    "stale_account_projection": 0.25,
    "account_write_reordered": 0.20,
    "rejected_order": 0.20,
    "invalid_market_snapshot": 0.25,
    "invalid_trade_state": 0.30,
    "invalid_account_state": 0.30,
    "exchange_executor_divergence": 0.35,
    "decision_intent_mismatch": 0.25,
    "duplicate_trade_id": 0.30,
    "unknown_invariant_violation": 0.40,
    "compound_fault": 0.35,
}


@dataclass
class AnomalySensorState:
    """MHS-shaped anomaly sensor surface for the DITA simulator.

    These are mock diagnostics, not live-system sensors. They provide a
    structured fault surface that can be written into HZ/CH-like mocks and
    validated in tests.
    """

    m7_transport_integrity: float = 1.0
    m8_execution_integrity: float = 1.0
    m9_terminal_integrity: float = 1.0
    m10_exchange_state_integrity: float = 1.0
    m11_account_projection_integrity: float = 1.0
    m12_order_acceptance_integrity: float = 1.0
    m13_invariant_integrity: float = 1.0

    def as_dict(self) -> Dict[str, float]:
        return {
            "m7_transport_integrity": self.m7_transport_integrity,
            "m8_execution_integrity": self.m8_execution_integrity,
            "m9_terminal_integrity": self.m9_terminal_integrity,
            "m10_exchange_state_integrity": self.m10_exchange_state_integrity,
            "m11_account_projection_integrity": self.m11_account_projection_integrity,
            "m12_order_acceptance_integrity": self.m12_order_acceptance_integrity,
            "m13_invariant_integrity": self.m13_invariant_integrity,
        }

    def degrade(self, sensor_name: str, amount: float = 0.1) -> None:
        if not hasattr(self, sensor_name):
            return
        current = float(getattr(self, sensor_name))
        setattr(self, sensor_name, max(0.0, min(1.0, current - max(0.0, amount))))

    def aggregate(self) -> float:
        values = list(self.as_dict().values())
        return sum(values) / max(1, len(values))

    def to_payload(
        self,
        timestamp: Optional[str] = None,
        anomaly_counts: Optional[Dict[str, int]] = None,
        origin_counts: Optional[Dict[str, int]] = None,
    ) -> Dict[str, Any]:
        payload = {
            "timestamp": timestamp,
            "rm_meta": round(self.aggregate(), 3),
            "status": "GREEN" if self.aggregate() >= 0.85 else "DEGRADED" if self.aggregate() >= 0.55 else "CRITICAL",
            **{key: round(value, 3) for key, value in self.as_dict().items()},
        }
        if anomaly_counts is not None:
            payload["anomaly_counts"] = deepcopy(anomaly_counts)
        if origin_counts is not None:
            payload["origin_counts"] = deepcopy(origin_counts)
        return payload


class MockNetwork:
    """Simple stochastic transport layer."""

    def __init__(self, profile: Optional[NetworkProfile] = None, seed: int = 7):
        self.profile = profile or NetworkProfile()
        self.rng = random.Random(seed)
        self.delivered = 0
        self.dropped = 0
        self.duplicated = 0

    def deliver(self, channel: str, payload: Dict[str, Any]) -> List[Dict[str, Any]]:
        """Return 0-2 copies of a payload, depending on the fault model."""
        roll = self.rng.random()
        if roll < self.profile.drop_rate:
            self.dropped += 1
            return []

        out = [deepcopy(payload)]
        self.delivered += 1

        if self.rng.random() < self.profile.duplicate_rate:
            self.duplicated += 1
            out.append(deepcopy(payload))
        return out


class MockHazelcast:
    """Dict-backed Hazelcast replacement."""

    def __init__(self):
        self.maps: Dict[str, Dict[str, Any]] = defaultdict(dict)
        self.history: Dict[str, List[Dict[str, Any]]] = defaultdict(list)

    def put(self, map_name: str, key: str, value: Any) -> None:
        self.maps[map_name][key] = deepcopy(value)
        self.history[map_name].append({key: deepcopy(value)})

    def get(self, map_name: str, key: str, default: Any = None) -> Any:
        return deepcopy(self.maps.get(map_name, {}).get(key, default))


class MockClickHouse:
    """Append-only table mock with bounded sample retention."""

    def __init__(self, capture_limit: int = 10_000):
        self.capture_limit = capture_limit
        self.tables: Dict[str, List[Dict[str, Any]]] = defaultdict(list)
        self.counts: Dict[str, int] = defaultdict(int)
        self.last_row: Dict[str, Dict[str, Any]] = {}

    def insert(self, table: str, row: Dict[str, Any]) -> None:
        self.counts[table] += 1
        payload = deepcopy(row)
        self.last_row[table] = payload
        if len(self.tables[table]) < self.capture_limit:
            self.tables[table].append(payload)

    def count(self, table: str) -> int:
        return self.counts.get(table, 0)


class MockLogSink:
    """Bounded log sink used for fuzzing and trace sampling."""

    def __init__(self, capture_limit: int = 5_000):
        self.capture_limit = capture_limit
        self.records: List[Dict[str, Any]] = []
        self.count = 0

    def emit(self, level: str, message: str, **fields: Any) -> None:
        self.count += 1
        if len(self.records) < self.capture_limit:
            self.records.append({"level": level, "message": message, "fields": deepcopy(fields)})


@dataclass
class ExchangeReceipt:
    trade_id: str
    status: str
    fill_price: float
    fill_size: float
    realized_pnl: float = 0.0
    fees: float = 0.0
    reason: str = ""
    partial: bool = False
    remaining_size: float = 0.0


class MockExchange:
    """Single-slot exchange simulator with entry/exit semantics."""

    def __init__(self, fee_bps: float = 2.0):
        self.fee_bps = fee_bps
        self.position: Optional[TradePosition] = None
        self.order_count = 0
        self.fill_count = 0
        self.rejected_count = 0
        self.realized_pnl = 0.0
        self.fees_paid = 0.0
        self.trade_history: List[ExchangeReceipt] = []

    def mark_price(self, price: float) -> None:
        if self.position is not None:
            self.position.mark_price(price)

    def reject(self, intent: Intent, reason: str) -> ExchangeReceipt:
        self.rejected_count += 1
        receipt = ExchangeReceipt(
            trade_id=intent.trade_id,
            status="REJECTED",
            fill_price=intent.reference_price,
            fill_size=0.0,
            reason=reason,
        )
        self.trade_history.append(receipt)
        return receipt

    def submit(self, intent: Intent) -> ExchangeReceipt:
        self.order_count += 1
        if intent.action == DecisionAction.ENTER:
            return self._enter(intent)
        if intent.action == DecisionAction.EXIT:
            return self._exit(intent)
        return self.reject(intent, "NOOP")

    def _enter(self, intent: Intent) -> ExchangeReceipt:
        if self.position is not None and not self.position.closed:
            return self.reject(intent, "POSITION_ALREADY_OPEN")

        position = TradePosition(
            trade_id=intent.trade_id,
            asset=intent.asset,
            side=intent.side,
            entry_price=intent.reference_price,
            entry_time=intent.timestamp,
            size=intent.target_size,
            leverage=intent.leverage,
            entry_velocity_divergence=float(intent.metadata.get("entry_velocity_divergence", intent.metadata.get("velocity_divergence", 0.0))),
            entry_irp_alignment=float(intent.metadata.get("entry_irp_alignment", intent.confidence)),
            current_price=intent.reference_price,
            initial_size=intent.target_size,
            exit_leg_ratios=tuple(intent.exit_leg_ratios),
        )
        self.position = position
        self.fill_count += 1
        fee = abs(position.entry_price * position.size) * (self.fee_bps / 10_000.0)
        self.fees_paid += fee
        receipt = ExchangeReceipt(
            trade_id=intent.trade_id,
            status="FILLED",
            fill_price=intent.reference_price,
            fill_size=intent.target_size,
            fees=fee,
            reason=intent.reason,
        )
        self.trade_history.append(receipt)
        return receipt

    def _exit(self, intent: Intent) -> ExchangeReceipt:
        if self.position is None or self.position.closed:
            return self.reject(intent, "NO_OPEN_POSITION")

        pos = self.position
        requested = intent.target_size if intent.target_size > 0 else pos.size
        fill_size = min(requested, pos.size)
        exit_price = intent.reference_price
        pnl_pct = (exit_price - pos.entry_price) / pos.entry_price if pos.entry_price > 0 else 0.0
        if pos.side == TradeSide.SHORT:
            pnl_pct = -pnl_pct
        pnl = pnl_pct * fill_size * pos.entry_price * pos.leverage
        fee = abs(exit_price * fill_size) * (self.fee_bps / 10_000.0)
        pnl_after_fee = pnl - fee

        pos.size = max(0.0, pos.size - fill_size)
        pos.exit_price = exit_price
        pos.realized_pnl += pnl_after_fee
        pos.closed = pos.size <= 1e-12
        pos.close_reason = intent.reason if pos.closed else f"PARTIAL_{intent.reason}"
        pos.mark_price(exit_price)

        if pos.closed:
            self.position = None
        self.fill_count += 1
        self.realized_pnl += pnl_after_fee
        self.fees_paid += fee
        receipt = ExchangeReceipt(
            trade_id=intent.trade_id,
            status="FILLED",
            fill_price=exit_price,
            fill_size=fill_size,
            realized_pnl=pnl_after_fee,
            fees=fee,
            reason=intent.reason,
            partial=not pos.closed,
            remaining_size=pos.size,
        )
        self.trade_history.append(receipt)
        return receipt

    def open_notional(self) -> float:
        if self.position is None:
            return 0.0
        return self.position.current_price * self.position.size

    def snapshot(self) -> Dict[str, Any]:
        return {
            "open_position": None if self.position is None else deepcopy(self.position),
            "order_count": self.order_count,
            "fill_count": self.fill_count,
            "rejected_count": self.rejected_count,
            "realized_pnl": self.realized_pnl,
            "fees_paid": self.fees_paid,
        }


@dataclass
class SimulationResult:
    """Summary from a simulation run."""

    steps: int
    trades_opened: int
    trades_closed: int
    trades_rejected: int
    capital_final: float
    equity_final: float
    open_notional_final: float
    decision_events: int
    trade_events: int
    account_events: int
    hazelcast_updates: int
    logs_emitted: int
    network_delivered: int
    network_dropped: int
    network_duplicated: int
    anomaly_counts: Dict[str, int] = field(default_factory=dict)
    anomaly_origin_counts: Dict[str, int] = field(default_factory=dict)
    injected_anomaly_counts: Dict[str, int] = field(default_factory=dict)
    emergent_anomaly_counts: Dict[str, int] = field(default_factory=dict)
    anomaly_sensor_payload: Dict[str, Any] = field(default_factory=dict)
    anomaly_samples: List[Dict[str, Any]] = field(default_factory=list)
    sample_policy_events: List[Dict[str, Any]] = field(default_factory=list)


class MockTradingStack:
    """Whole-stack mock: decision, intent, trade, account, HZ, CH, logs, network."""

    def __init__(
        self,
        policy: Optional[Any] = None,
        decision_engine: Optional[DecisionEngine] = None,
        intent_engine: Optional[IntentEngine] = None,
        capital: float = 25_000.0,
        runtime_namespace: str = "pink",
        strategy_namespace: str = "pink",
        event_namespace: str = "pink",
        network: Optional[MockNetwork] = None,
        hazelcast: Optional[MockHazelcast] = None,
        clickhouse: Optional[MockClickHouse] = None,
        logs: Optional[MockLogSink] = None,
        chaos: Optional[ChaosProfile] = None,
        observability: Optional[DitaObservabilityNamespace] = None,
    ):
        self.decision_engine = decision_engine or DecisionEngine()
        self.intent_engine = intent_engine or IntentEngine()
        if policy is not None and hasattr(policy, "decide"):
            self.decision_engine = policy
        self.trade_executor = TradeExecutor()
        self.exchange = MockExchange()
        self.network = network or MockNetwork()
        self.hz = hazelcast or MockHazelcast()
        self.ch = clickhouse or MockClickHouse()
        self.logs = logs or MockLogSink()
        self.chaos = chaos or ChaosProfile()
        self.anomaly_sensors = AnomalySensorState()
        self.account = AccountProjection(
            runtime_namespace=runtime_namespace,
            strategy_namespace=strategy_namespace,
            event_namespace=event_namespace,
            actor_name="clean_arch",
            exec_venue="bingx",
            data_venue="binance",
            ledger_authority="exchange",
            max_capital=1_000_000_000.0,
            snapshot=AccountSnapshot(capital=capital, equity=capital),
        )
        inferred_state_map = f"DOLPHIN_STATE_{str(runtime_namespace or 'pink').upper()}"
        if str(runtime_namespace or "").strip().lower() == "pink":
            inferred_state_map = "DOLPHIN_STATE_PINK"
        self.observability = observability or DitaObservabilityNamespace(
            runtime_namespace=runtime_namespace,
            feature_map="DOLPHIN_FEATURES",
            meta_health_map="DOLPHIN_META_HEALTH",
            state_map=inferred_state_map,
        )
        self._anomaly_sensor_key = self.observability.resolved_sensor_key()
        self.steps = 0
        self.trade_seq = 0
        self.decision_event_samples: List[Dict[str, Any]] = []
        self.anomaly_counts: Dict[str, int] = defaultdict(int)
        self.anomaly_origin_counts: Dict[str, int] = defaultdict(int)
        self.anomaly_samples: List[Dict[str, Any]] = []
        self._step_anomaly_names: List[str] = []
        self._seen_trade_ids: set[str] = set()
        self._equity_peak = capital

    @property
    def capital(self) -> float:
        return self.account.snapshot.capital

    @property
    def equity(self) -> float:
        return self.account.snapshot.equity

    @property
    def anomaly_sensor_key(self) -> str:
        return self._anomaly_sensor_key

    def step(self, snapshot: MarketSnapshot):
        """Run one market snapshot through the full mocked stack."""
        self.steps += 1
        self._step_anomaly_names = []
        self.exchange.mark_price(snapshot.price)
        self.trade_executor.position = self.exchange.position
        self.account.observe_position(self.exchange.position)
        self._emit_engine_status(snapshot, phase="pre")

        decision_context = DecisionContext(
            capital=self.account.snapshot.capital,
            open_positions=1 if self.exchange.position is not None and not self.exchange.position.closed else 0,
            trade_seq=self.trade_seq,
        )
        decision = self.decision_engine.decide(snapshot, decision_context, self.exchange.position)
        intent_context = IntentContext(
            capital=self.account.snapshot.capital,
            open_positions=decision_context.open_positions,
            trade_seq=self.trade_seq,
        )
        plan = self.intent_engine.plan(decision, intent_context, self.exchange.position)
        intent = plan.intent

        self._write_hz(snapshot, decision, intent)
        self._write_decision(snapshot, decision, intent)
        self._write_account(snapshot, decision, intent, stage=TradeStage.DECISION_CREATED)
        self._write_position_state(snapshot, decision, intent, stage=TradeStage.DECISION_CREATED)
        self._write_position_state(snapshot, decision, intent, stage=TradeStage.INTENT_CREATED)

        if decision.action == DecisionAction.ENTER and self._chance(self.chaos.hang_entry_rate):
            self._record_anomaly("hung_entry", snapshot, decision, intent, detail="entry dropped before execution", origin="injected")
            self._emit_engine_status(snapshot, phase="hung-entry", decision=decision, intent=intent)
            self._maybe_record_compound_fault(snapshot, decision, intent)
            return decision

        if decision.action == DecisionAction.EXIT and self.exchange.position is not None and self._chance(self.chaos.hang_exit_rate):
            self._record_anomaly("hung_exit", snapshot, decision, intent, detail="exit dropped before execution", origin="injected")
            self._emit_engine_status(snapshot, phase="hung-exit", decision=decision, intent=intent)
            self._maybe_record_compound_fault(snapshot, decision, intent)
            return decision

        position_before = self.trade_executor._clone_position(self.exchange.position)
        delivered = self.network.deliver("exchange", {"decision": decision.decision_id, "intent": intent.trade_id, "price": snapshot.price})
        result = None
        if not delivered:
            self.logs.emit("WARN", "network_drop", trade_id=intent.trade_id, action=intent.action.value)
            self._record_anomaly("network_drop", snapshot, decision, intent, detail="exchange delivery dropped", origin="injected")
        else:
            last_result = None
            for _payload in delivered:
                last_result = self.trade_executor.execute(intent, self.exchange, self.account.snapshot.capital)
                self.trade_executor.position = self.exchange.position
            result = last_result
        self.account.observe_position(self.exchange.position)

        if result is not None:
            for stage in result.stages:
                self._write_position_state(snapshot, decision, intent, stage=stage)

            if result.receipt is not None and intent.action == DecisionAction.EXIT and result.receipt.status == "FILLED":
                stale_projection = False
                if self._chance(self.chaos.reorder_account_rate):
                    self._record_anomaly(
                        "account_write_reordered",
                        snapshot,
                        decision,
                        intent,
                        detail="account row emitted before settlement",
                        origin="injected",
                    )
                    self._write_account(
                        snapshot,
                        decision,
                        intent,
                        stage=TradeStage.EXIT_ACKED,
                        position_override=position_before,
                        refresh=False,
                    )
                    stale_projection = True
                if self._chance(self.chaos.missing_terminal_rate):
                    self._record_anomaly(
                        "missing_terminal_row",
                        snapshot,
                        decision,
                        intent,
                        detail="close row suppressed",
                        origin="injected",
                    )
                else:
                    self.account.settle(result.receipt.realized_pnl, result.receipt.fees)
                    if self._chance(self.chaos.stale_account_rate):
                        self._write_account(
                            snapshot,
                            decision,
                            intent,
                            stage=TradeStage.EXIT_ACKED,
                            position_override=position_before,
                            refresh=False,
                        )
                        stale_projection = True
                    if self.exchange.position is None:
                        self._write_trade_close(snapshot, decision, intent, result)
                        if self._chance(self.chaos.duplicate_terminal_rate):
                            self._write_trade_close(snapshot, decision, intent, result)
                            self._record_anomaly(
                                "duplicate_terminal_row",
                                snapshot,
                                decision,
                                intent,
                                detail="close row duplicated",
                                origin="injected",
                            )
                    if not stale_projection:
                        self._write_account(snapshot, decision, intent, stage=TradeStage.EXIT_ACKED)
                    if self._chance(self.chaos.orphan_close_rate) and self.exchange.position is None:
                        ghost = self.trade_executor._clone_position(position_before)
                        if ghost is not None:
                            ghost.closed = True
                            ghost.close_reason = "ORPHANED_CLOSED"
                            self.exchange.position = ghost
                            self.trade_executor.position = ghost
                            self.account.observe_position(self.exchange.position)
                            self._record_anomaly(
                                "orphaned_close",
                                snapshot,
                                decision,
                                intent,
                                detail="exchange left open after terminal close",
                                origin="injected",
                            )
            elif result.receipt is not None and intent.action == DecisionAction.ENTER and result.receipt.status == "FILLED":
                if intent.trade_id in self._seen_trade_ids:
                    self._record_anomaly(
                        "duplicate_trade_id",
                        snapshot,
                        decision,
                        intent,
                        detail="trade_id reused on a new open",
                        origin="emergent",
                    )
                self._seen_trade_ids.add(intent.trade_id)
                self.trade_seq += 1
                self._write_account(snapshot, decision, intent, stage=TradeStage.POSITION_OPENED)
            elif result.receipt is not None and result.receipt.status == "REJECTED":
                self._record_anomaly("rejected_order", snapshot, decision, intent, detail=result.receipt.reason, origin="emergent")

        if self.ch.last_row.get("dolphin_pink.account_events"):
            last_account = self.ch.last_row["dolphin_pink.account_events"]
            if int(last_account.get("open_positions", 0)) != (0 if self.exchange.position is None else 1):
                self._record_anomaly("stale_account_projection", snapshot, decision, intent, detail=last_account, origin="emergent")

        if self.exchange.position is not None and self.exchange.position.closed:
            self._record_anomaly("closed_position_still_present", snapshot, decision, intent, detail=self.exchange.position.trade_id, origin="emergent")

        self._run_invariant_sweep(snapshot, decision, intent)
        self._maybe_record_compound_fault(snapshot, decision, intent)
        self.account.observe_position(self.exchange.position)
        self._equity_peak = max(self._equity_peak, self.account.snapshot.equity)
        self._emit_engine_status(snapshot, phase="post", decision=decision, intent=intent)
        return decision

    def run(self, snapshots: Iterable[MarketSnapshot], limit: Optional[int] = None) -> SimulationResult:
        processed = 0
        for snapshot in snapshots:
            self.step(snapshot)
            processed += 1
            if limit is not None and processed >= limit:
                break
        return self.summary(processed)

    def summary(self, steps: Optional[int] = None) -> SimulationResult:
        steps = self.steps if steps is None else steps
        position = self.exchange.position
        return SimulationResult(
            steps=steps,
            trades_opened=sum(1 for r in self.exchange.trade_history if r.status == "FILLED" and r.reason == "STRUCTURAL_DISLOCATION"),
            trades_closed=sum(1 for r in self.exchange.trade_history if r.status == "FILLED" and r.reason in {"TAKE_PROFIT", "MAX_HOLD", "MEAN_REVERSION", "CATASTROPHIC_LOSS"}),
            trades_rejected=self.exchange.rejected_count,
            capital_final=self.capital,
            equity_final=self.equity if position is None else self.equity + position.unrealized_pnl,
            open_notional_final=self.exchange.open_notional(),
            decision_events=self.ch.count("dolphin_pink.policy_events"),
            trade_events=self.ch.count("dolphin_pink.trade_events"),
            account_events=self.ch.count("dolphin_pink.account_events"),
            hazelcast_updates=sum(len(v) for v in self.hz.history.values()),
            logs_emitted=self.logs.count,
            network_delivered=self.network.delivered,
            network_dropped=self.network.dropped,
            network_duplicated=self.network.duplicated,
            anomaly_counts=dict(self.anomaly_counts),
            anomaly_origin_counts=dict(self.anomaly_origin_counts),
            injected_anomaly_counts={k: v for k, v in self.anomaly_counts.items() if k in INJECTED_ANOMALIES},
            emergent_anomaly_counts={k: v for k, v in self.anomaly_counts.items() if k in EMERGENT_ANOMALIES},
            anomaly_sensor_payload=self.anomaly_sensors.to_payload(
                timestamp=datetime.now().astimezone().isoformat(),
                anomaly_counts=dict(self.anomaly_counts),
                origin_counts=dict(self.anomaly_origin_counts),
            ),
            anomaly_samples=deepcopy(self.anomaly_samples[:100]),
            sample_policy_events=deepcopy(self.decision_event_samples[:50]),
        )

    def _apply_order(self, intent: Intent, snapshot: MarketSnapshot):
        delivered = self.network.deliver("exchange", {"intent": intent, "price": snapshot.price})
        if not delivered:
            self.logs.emit("WARN", "network_drop", trade_id=intent.trade_id, action=intent.action.value)
            return None
        return self.exchange.submit(intent)

    def _chance(self, rate: float) -> bool:
        return rate > 0.0 and self.network.rng.random() < rate

    def _record_anomaly(
        self,
        name: str,
        snapshot: MarketSnapshot,
        decision,
        intent: Intent,
        detail: Any,
        origin: str = "emergent",
    ) -> None:
        self.anomaly_counts[name] += 1
        self.anomaly_origin_counts[origin] += 1
        self._step_anomaly_names.append(name)
        sensor_name = ANOMALY_SENSOR_MAP.get(name)
        if sensor_name is not None:
            self.anomaly_sensors.degrade(sensor_name, ANOMALY_SENSOR_SEVERITY.get(name, 0.1))
        sensor_payload = self.anomaly_sensors.to_payload(
            timestamp=snapshot.timestamp.isoformat(),
            anomaly_counts=dict(self.anomaly_counts),
            origin_counts=dict(self.anomaly_origin_counts),
        )
        self.hz.put(self.observability.meta_health_map, self._anomaly_sensor_key, sensor_payload)
        self.hz.put(self.observability.feature_map, self._anomaly_sensor_key, sensor_payload)
        if self.observability.mirror_legacy_key and self._anomaly_sensor_key != LEGACY_ANOMALY_SENSOR_KEY:
            self.hz.put(self.observability.meta_health_map, LEGACY_ANOMALY_SENSOR_KEY, sensor_payload)
            self.hz.put(self.observability.feature_map, LEGACY_ANOMALY_SENSOR_KEY, sensor_payload)
        self.ch.insert(
            "dolphin_pink.anomaly_events",
            {
                "ts": snapshot.timestamp.isoformat(),
                "decision_id": decision.decision_id,
                "trade_id": intent.trade_id,
                "symbol": snapshot.symbol,
                "anomaly": name,
                "origin": origin,
                "sensor": sensor_name,
                "detail": detail,
                "rm_meta": sensor_payload["rm_meta"],
            },
        )
        if len(self.anomaly_samples) < 200:
            self.anomaly_samples.append(
                {
                    "anomaly": name,
                    "ts": snapshot.timestamp.isoformat(),
                    "symbol": snapshot.symbol,
                    "decision_id": decision.decision_id,
                    "trade_id": intent.trade_id,
                    "action": decision.action.value,
                    "reason": intent.reason,
                    "detail": detail,
                    "origin": origin,
                    "capital": self.capital,
                    "equity": self.equity,
                    "open_positions": 0 if self.exchange.position is None else 1,
                }
            )
        self.logs.emit("WARN", "anomaly", anomaly=name, origin=origin, trade_id=intent.trade_id, reason=intent.reason, detail=detail)

    def _maybe_record_compound_fault(self, snapshot: MarketSnapshot, decision, intent: Intent) -> None:
        """Promote multi-fault steps to an explicit combined-failure anomaly."""
        unique = [name for name in dict.fromkeys(self._step_anomaly_names) if name != "compound_fault"]
        if len(unique) < 2:
            return
        if "compound_fault" in unique:
            return
        detail = {
            "faults": unique[:8],
            "fault_count": len(unique),
        }
        self._record_anomaly("compound_fault", snapshot, decision, intent, detail=detail, origin="emergent")

    def _run_invariant_sweep(self, snapshot: MarketSnapshot, decision, intent: Intent) -> List[Dict[str, Any]]:
        """Run hard-state invariants that should never require heuristics.

        This sweep is intentionally false-positive resistant: it only checks
        directly contradictory states, not statistical expectations.
        """

        violations: List[Dict[str, Any]] = []

        def add(name: str, detail: Any, origin: str = "emergent") -> None:
            violations.append({"name": name, "detail": detail, "origin": origin})

        try:
            price = snapshot.price
            vdiv = snapshot.velocity_divergence
            if not isinstance(snapshot.symbol, str) or not snapshot.symbol.strip():
                add("invalid_market_snapshot", "missing or blank symbol", "injected")
            if not math.isfinite(price) or price <= 0:
                add("invalid_market_snapshot", {"price": price}, "injected")
            if vdiv is None or not math.isfinite(float(vdiv)):
                add("invalid_market_snapshot", {"velocity_divergence": vdiv}, "injected")

            acct = self.account.snapshot
            if (
                not math.isfinite(acct.capital)
                or not math.isfinite(acct.equity)
                or not math.isfinite(acct.open_notional)
                or acct.capital < 0
                or acct.equity < 0
                or acct.open_notional < 0
                or acct.open_positions not in (0, 1)
            ):
                add(
                    "invalid_account_state",
                    {
                        "capital": acct.capital,
                        "equity": acct.equity,
                        "open_notional": acct.open_notional,
                        "open_positions": acct.open_positions,
                    },
                    "emergent",
                )

            pos = self.exchange.position
            if pos is not None:
                if (
                    not isinstance(pos.trade_id, str)
                    or not pos.trade_id
                    or not isinstance(pos.asset, str)
                    or not pos.asset
                    or not math.isfinite(pos.entry_price)
                    or pos.entry_price <= 0
                    or not math.isfinite(pos.size)
                    or pos.size <= 0
                    or not math.isfinite(pos.leverage)
                    or pos.leverage <= 0
                    or not math.isfinite(pos.current_price)
                    or pos.current_price <= 0
                ):
                    add(
                        "invalid_trade_state",
                        {
                            "trade_id": pos.trade_id,
                            "asset": pos.asset,
                            "entry_price": pos.entry_price,
                            "size": pos.size,
                            "leverage": pos.leverage,
                            "current_price": pos.current_price,
                            "closed": pos.closed,
                        },
                        "emergent",
                    )

                executor_pos = self.trade_executor.position
                if executor_pos is None:
                    add("exchange_executor_divergence", "executor missing live position", "emergent")
                elif not self._positions_match(pos, executor_pos):
                    add(
                        "exchange_executor_divergence",
                        {
                            "exchange": self._position_snapshot(pos),
                            "executor": self._position_snapshot(executor_pos),
                        },
                        "emergent",
                    )
            elif self.trade_executor.position is not None:
                add("exchange_executor_divergence", "executor retains position while exchange is flat", "emergent")

            if decision.decision_id != intent.decision_id or decision.action != intent.action or decision.side != intent.side or decision.asset != intent.asset:
                add(
                    "decision_intent_mismatch",
                    {
                        "decision_id": decision.decision_id,
                        "intent_id": intent.decision_id,
                        "decision_action": decision.action.value,
                        "intent_action": intent.action.value,
                        "decision_side": decision.side.value,
                        "intent_side": intent.side.value,
                        "decision_asset": decision.asset,
                        "intent_asset": intent.asset,
                    },
                    "emergent",
                )
        except Exception as exc:  # pragma: no cover - safety net for the sweep itself
            add("unknown_invariant_violation", {"exception": repr(exc)}, "emergent")

        if violations:
            for violation in violations:
                name = violation["name"]
                detail = violation["detail"]
                origin = violation["origin"]
                if name not in ANOMALY_SENSOR_MAP:
                    self._record_anomaly("unknown_invariant_violation", snapshot, decision, intent, detail={"rule": name, "detail": detail}, origin="emergent")
                else:
                    self._record_anomaly(name, snapshot, decision, intent, detail=detail, origin=origin)
        return violations

    @staticmethod
    def _position_snapshot(position: TradePosition) -> Dict[str, Any]:
        return {
            "trade_id": position.trade_id,
            "asset": position.asset,
            "side": position.side.value,
            "entry_price": position.entry_price,
            "size": position.size,
            "leverage": position.leverage,
            "current_price": position.current_price,
            "closed": position.closed,
            "close_reason": position.close_reason,
        }

    @staticmethod
    def _positions_match(left: TradePosition, right: TradePosition) -> bool:
        return (
            left.trade_id == right.trade_id
            and left.asset == right.asset
            and left.side == right.side
            and math.isclose(left.entry_price, right.entry_price, rel_tol=0.0, abs_tol=1e-12)
            and math.isclose(left.size, right.size, rel_tol=0.0, abs_tol=1e-12)
            and math.isclose(left.leverage, right.leverage, rel_tol=0.0, abs_tol=1e-12)
            and left.closed == right.closed
            and left.close_reason == right.close_reason
        )

    def _write_hz(self, snapshot: MarketSnapshot, decision, intent: Intent) -> None:
        payload = {
            "timestamp": snapshot.timestamp.isoformat(),
            "symbol": snapshot.symbol,
            "price": snapshot.price,
            "velocity_divergence": snapshot.velocity_divergence,
            "decision": decision.action.value,
            "intent": intent.action.value,
            "reason": intent.reason,
        }
        self.hz.put(self.observability.feature_map, "latest_snapshot", payload)
        self.hz.put(
            self.observability.state_map,
            "dita_state",
            {
                "decision_id": decision.decision_id,
                "trade_id": intent.trade_id,
                "action": intent.action.value,
                "side": intent.side.value,
                "reason": intent.reason,
                "capital": self.capital,
                "anomaly_sensor_key": self._anomaly_sensor_key,
            },
        )
        sensor_payload = self.anomaly_sensors.to_payload(
            timestamp=snapshot.timestamp.isoformat(),
            anomaly_counts=dict(self.anomaly_counts),
            origin_counts=dict(self.anomaly_origin_counts),
        )
        self.hz.put(self.observability.meta_health_map, self._anomaly_sensor_key, sensor_payload)
        self.hz.put(self.observability.feature_map, self._anomaly_sensor_key, sensor_payload)
        if self.observability.mirror_legacy_key and self._anomaly_sensor_key != LEGACY_ANOMALY_SENSOR_KEY:
            self.hz.put(self.observability.meta_health_map, LEGACY_ANOMALY_SENSOR_KEY, sensor_payload)
            self.hz.put(self.observability.feature_map, LEGACY_ANOMALY_SENSOR_KEY, sensor_payload)

    def _write_decision(self, snapshot: MarketSnapshot, decision, intent: Intent) -> None:
        row = {
            "ts": snapshot.timestamp.isoformat(),
            "decision_id": decision.decision_id,
            "trade_id": intent.trade_id,
            "asset": decision.asset,
            "action": decision.action.value,
            "side": decision.side.value,
            "reason": decision.reason,
            "confidence": decision.confidence,
            "vel_div": decision.velocity_divergence,
            "irp_alignment": decision.irp_alignment,
            "stage": decision.stage.value,
        }
        for payload in self.network.deliver("clickhouse", row):
            self.ch.insert("dolphin_pink.policy_events", payload)
            self.ch.insert("dolphin_pink.v7_decision_events", payload)
            self.ch.insert("dolphin_pink.account_events", self._account_row(snapshot, decision, intent, TradeStage.DECISION_CREATED))

        if len(self.decision_event_samples) < 50:
            self.decision_event_samples.append(
                {
                    "decision_id": decision.decision_id,
                    "trade_id": intent.trade_id,
                    "action": decision.action.value,
                    "reason": decision.reason,
                    "stage": decision.stage.value,
                }
            )

    def _write_trade_close(self, snapshot: MarketSnapshot, decision, intent: Intent, result) -> None:
        if result.receipt is None:
            return
        receipt = result.receipt
        row = {
            "ts": snapshot.timestamp.isoformat(),
            "decision_id": decision.decision_id,
            "trade_id": intent.trade_id,
            "asset": decision.asset,
            "stage": TradeStage.TRADE_TERMINAL_WRITTEN.value,
            "action": decision.action.value,
            "side": intent.side.value,
            "reason": intent.reason,
            "price": receipt.fill_price,
            "size": receipt.fill_size,
            "pnl": receipt.realized_pnl,
            "fees": receipt.fees,
        }
        self.ch.insert("dolphin_pink.trade_events", row)
        self.ch.insert(
            "dolphin_pink.position_state",
            {
                "ts": snapshot.timestamp.isoformat(),
                "decision_id": decision.decision_id,
                "trade_id": intent.trade_id,
                "asset": decision.asset,
                "state": "CLOSED",
                "stage": TradeStage.TRADE_TERMINAL_WRITTEN.value,
                "reason": intent.reason,
            },
        )

    def _write_position_state(self, snapshot: MarketSnapshot, decision, intent: Intent, stage: TradeStage) -> None:
        position = self.exchange.position
        state = "FLAT"
        if position is not None and not position.closed:
            state = "OPEN"
        elif stage in {TradeStage.POSITION_CLOSED, TradeStage.TRADE_TERMINAL_WRITTEN}:
            state = "CLOSED"
        elif stage == TradeStage.POSITION_PARTIALLY_CLOSED:
            state = "OPEN"
        row = {
            "ts": snapshot.timestamp.isoformat(),
            "decision_id": decision.decision_id,
            "trade_id": intent.trade_id,
            "asset": decision.asset,
            "state": state,
            "stage": stage.value,
            "reason": intent.reason,
            "capital": self.capital,
            "equity": self.equity,
            "open_positions": 0 if position is None else 1,
        }
        self.ch.insert("dolphin_pink.position_state", row)

    def _emit_engine_status(self, snapshot: MarketSnapshot, phase: str, decision=None, intent: Optional[Intent] = None) -> None:
        position = self.exchange.position
        status = {
            "ts": snapshot.timestamp.isoformat(),
            "phase": phase,
            "symbol": snapshot.symbol,
            "price": snapshot.price,
            "capital": self.capital,
            "equity": self.equity if position is None else self.equity + position.unrealized_pnl,
            "open_positions": 0 if position is None else 1,
            "open_notional": self.exchange.open_notional(),
            "order_count": self.exchange.order_count,
            "fill_count": self.exchange.fill_count,
            "rejected_count": self.exchange.rejected_count,
            "capital_peak": self._equity_peak,
            "trade_seq": self.trade_seq,
            "network_delivered": self.network.delivered,
            "network_dropped": self.network.dropped,
            "network_duplicated": self.network.duplicated,
            "decision_stage": None if decision is None else decision.stage.value,
            "decision_action": None if decision is None else decision.action.value,
            "decision_reason": None if decision is None else decision.reason,
            "intent_action": None if intent is None else intent.action.value,
            "intent_reason": None if intent is None else intent.reason,
            "anomaly_rm_meta": round(self.anomaly_sensors.aggregate(), 3),
            "anomaly_sensors": self.anomaly_sensors.as_dict(),
        }
        self.logs.emit("INFO", "engine_status", **status)

    def _write_account(
        self,
        snapshot: MarketSnapshot,
        decision,
        intent: Intent,
        stage: TradeStage,
        position_override: Optional[TradePosition] = None,
        refresh: bool = True,
    ) -> None:
        row = self._account_row(snapshot, decision, intent, stage, position_override=position_override, refresh=refresh)
        self.ch.insert("dolphin_pink.account_events", row)

    def _account_row(
        self,
        snapshot: MarketSnapshot,
        decision,
        intent: Intent,
        stage: TradeStage,
        position_override: Optional[TradePosition] = None,
        refresh: bool = True,
    ) -> Dict[str, Any]:
        position = position_override if position_override is not None else self.exchange.position
        if refresh:
            self.account.observe_position(position)
        open_positions = 0 if position is None or position.closed else 1
        open_notional = 0.0 if position is None or position.closed else position.current_price * position.size
        leverage = 0.0 if position is None or position.closed else position.leverage
        return {
            "ts": snapshot.timestamp.isoformat(),
            "runtime_namespace": self.account.runtime_namespace,
            "strategy_namespace": self.account.strategy_namespace,
            "event_namespace": self.account.event_namespace,
            "actor_name": self.account.actor_name,
            "exec_venue": self.account.exec_venue,
            "data_venue": self.account.data_venue,
            "ledger_authority": self.account.ledger_authority,
            "capital": self.account.snapshot.capital,
            "equity": self.account.snapshot.equity,
            "open_positions": open_positions,
            "current_open_notional": open_notional,
            "current_account_leverage": leverage,
            "decision": decision.action.value,
            "reason": intent.reason,
            "decision_id": decision.decision_id,
            "trade_id": intent.trade_id,
            "symbol": snapshot.symbol,
            "side": intent.side.value,
            "stage": stage.value,
            "anomaly_rm_meta": self.anomaly_sensors.aggregate(),
            "anomaly_sensors": self.anomaly_sensors.as_dict(),
        }