prediction_engine/simulation/scenario_cache.py

"""LRU scenario cache keyed by state hash.

Caches simulation results to avoid recomputing identical
or near-identical market states within the same event window.
"""

from __future__ import annotations

import hashlib
import json
import time
from collections import OrderedDict
from dataclasses import dataclass, field
from typing import Any, Dict, List, Optional


@dataclass
class CachedScenario:
    """A cached simulation result."""
    state_hash: str
    branches: List[List[Dict[str, float]]]
    scores: List[float]
    expected_value: float
    var_95: float
    timestamp: float = field(default_factory=time.time)
    hit_count: int = 0


class ScenarioCache:
    """LRU cache for simulation results.

    Keyed by a hash of the market state vector. Evicts least
    recently used entries when capacity is exceeded.
    """

    def __init__(self, capacity: int = 1024, ttl_seconds: float = 300.0):
        self.capacity = capacity
        self.ttl = ttl_seconds
        self._cache: OrderedDict[str, CachedScenario] = OrderedDict()
        self._hits = 0
        self._misses = 0

    def state_hash(self, state: Dict[str, Any]) -> str:
        """Compute deterministic hash of a market state."""
        serialized = json.dumps(state, sort_keys=True, default=str)
        return hashlib.sha256(serialized.encode()).hexdigest()[:16]

    def get(self, state: Dict[str, Any]) -> Optional[CachedScenario]:
        """Retrieve cached scenario, or None on miss."""
        h = self.state_hash(state)
        if h in self._cache:
            entry = self._cache[h]
            if time.time() - entry.timestamp > self.ttl:
                del self._cache[h]
                self._misses += 1
                return None
            self._cache.move_to_end(h)
            entry.hit_count += 1
            self._hits += 1
            return entry
        self._misses += 1
        return None

    def put(self, state: Dict[str, Any], scenario: CachedScenario) -> None:
        """Store a scenario result."""
        h = self.state_hash(state)
        scenario.state_hash = h
        if h in self._cache:
            self._cache.move_to_end(h)
        self._cache[h] = scenario
        while len(self._cache) > self.capacity:
            self._cache.popitem(last=False)

    @property
    def hit_rate(self) -> float:
        total = self._hits + self._misses
        return self._hits / total if total > 0 else 0.0

    def clear(self) -> None:
        self._cache.clear()
        self._hits = 0
        self._misses = 0