fithia2/libs/backtest/low_vol_cache.py

"""Disk-based cache for low_vol anomaly ranked-universe results.

Mirrors libs/backtest/xsmom_cache.py exactly — just renamed and keyed on the
low-vol params. Caches the full pre-ranked universe per rebalance date so that
re-runs with different top_n values can skip the expensive per-symbol bar scan.

Cache key: (snapshot_fingerprint, param_hash, LOW_VOL_CACHE_VERSION)
Cache file: <snapshot_dir>/.runtime_cache/low_vol_v{N}__{param_hash}.parquet

The cache stores rows that have already passed all universe-quality gates
(min_price, min_adv, volatility_min). Only top_n selection is deferred to read
time so a single cache file serves runs with different top_n values.
"""
from __future__ import annotations

import datetime as dt
import hashlib
import json
import os
from pathlib import Path
from typing import Any
from uuid import uuid4

import pyarrow as pa
import pyarrow.parquet as pq

from libs.common.logging import get_logger

logger = get_logger(__name__)

LOW_VOL_CACHE_VERSION = 1

# Params that determine the cached content.
# top_n is NOT here — it only slices the cached ranking.
_CACHE_KEY_PARAMS = (
    "lowvol_lookback_days",
    "lowvol_min_avg_dollar_volume",
    "lowvol_min_price",
    "lowvol_volatility_min",
)

_SCHEMA = pa.schema([
    pa.field("decision_date", pa.string()),
    pa.field("symbol", pa.string()),
    pa.field("realized_volatility", pa.float64()),
    pa.field("avg_dollar_volume_20d", pa.float64()),
    pa.field("last_close", pa.float64()),
    pa.field("last_bar_date", pa.string()),
    pa.field("last_bar_timestamp_iso", pa.string()),
])


def compute_snapshot_fingerprint(snapshot_dir: Path) -> str:
    """SHA-256 fingerprint of snapshot manifest + parquet files (size + mtime_ns)."""
    parts: list[str] = [f"low_vol_v{LOW_VOL_CACHE_VERSION}"]
    manifest = snapshot_dir / "manifest.json"
    if manifest.exists():
        s = manifest.stat()
        parts.append(f"manifest:{s.st_size}:{s.st_mtime_ns}")
    for pq_path in sorted(snapshot_dir.glob("*.parquet")):
        s = pq_path.stat()
        parts.append(f"{pq_path.name}:{s.st_size}:{s.st_mtime_ns}")
    return hashlib.sha256("|".join(parts).encode()).hexdigest()[:32]


def build_param_hash(engine: Any) -> str:
    """16-char hex hash of the cache-key params from an engine config."""
    params = {k: getattr(engine, k, None) for k in _CACHE_KEY_PARAMS}
    raw = json.dumps(params, sort_keys=True, default=str)
    return hashlib.sha256(raw.encode()).hexdigest()[:16]


class LowVolRankCache:
    """Per-snapshot-dir disk cache for low-vol rebalance-day ranked universes.

    Lifecycle:
      1. Construct once per backtest run (lazy, on first rebalance day).
      2. Pass to build_candidates() on every call.
      3. Call close() after the backtest loop to flush new rows to disk.
    """

    def __init__(
        self,
        snapshot_dir: Path,
        snapshot_fingerprint: str,
        param_hash: str,
    ) -> None:
        self._cache_dir = snapshot_dir / ".runtime_cache"
        self._fingerprint = snapshot_fingerprint
        self._param_hash = param_hash
        self._cache_path = (
            self._cache_dir
            / f"low_vol_v{LOW_VOL_CACHE_VERSION}__{param_hash}.parquet"
        )
        # None = not yet attempted; {} = loaded (possibly empty due to miss)
        self._by_date: dict[str, list[dict[str, Any]]] | None = None
        self._new_rows: list[dict[str, Any]] = []

    # ------------------------------------------------------------------
    # Internal: lazy load
    # ------------------------------------------------------------------

    def _ensure_loaded(self) -> None:
        if self._by_date is not None:
            return
        if not self._cache_path.exists():
            logger.info(
                "low_vol_cache_miss",
                reason="file_missing",
                cache_file=str(self._cache_path),
            )
            self._by_date = {}
            return
        try:
            table = pq.read_table(str(self._cache_path))
            meta = table.schema.metadata or {}
            stored_fp = (meta.get(b"low_vol_fingerprint") or b"").decode()
            if stored_fp != self._fingerprint:
                logger.info(
                    "low_vol_cache_miss",
                    reason="fingerprint_mismatch",
                    cache_file=str(self._cache_path),
                )
                self._by_date = {}
                return
            by_date: dict[str, list[dict[str, Any]]] = {}
            for row in table.to_pylist():
                d = str(row["decision_date"])
                by_date.setdefault(d, []).append(row)
            self._by_date = by_date
            logger.info(
                "low_vol_cache_hit",
                cache_file=str(self._cache_path),
                dates=len(by_date),
                rows=table.num_rows,
            )
        except Exception as exc:
            logger.warning(
                "low_vol_cache_read_failed",
                cache_file=str(self._cache_path),
                error=str(exc),
            )
            self._by_date = {}

    # ------------------------------------------------------------------
    # Public API
    # ------------------------------------------------------------------

    def get_date(self, decision_date: dt.date) -> list[dict[str, Any]] | None:
        """Return cached ranked rows for decision_date, or None on miss."""
        self._ensure_loaded()
        return self._by_date.get(decision_date.isoformat())

    def save_date(self, decision_date: dt.date, rows: list[dict[str, Any]]) -> None:
        """Buffer ranked rows for this rebalance date. Flushed by close()."""
        self._new_rows.extend(rows)

    def close(self) -> None:
        """Flush buffered rows to disk via atomic write. No-op if nothing new."""
        if not self._new_rows:
            return
        all_rows: list[dict[str, Any]] = []
        if self._by_date:
            for date_rows in self._by_date.values():
                all_rows.extend(date_rows)
        all_rows.extend(self._new_rows)

        table = pa.Table.from_pylist(all_rows, schema=_SCHEMA)
        existing_meta = dict(table.schema.metadata or {})
        table = table.replace_schema_metadata(
            {**existing_meta, b"low_vol_fingerprint": self._fingerprint.encode()}
        )
        self._cache_dir.mkdir(parents=True, exist_ok=True)
        _write_table_atomic(table, self._cache_path)
        logger.info(
            "low_vol_cache_written",
            cache_file=str(self._cache_path),
            total_rows=len(all_rows),
            new_dates=len({r["decision_date"] for r in self._new_rows}),
        )
        self._new_rows = []


# ------------------------------------------------------------------
# Atomic write (mirrors libs/intraday/cache.py pattern)
# ------------------------------------------------------------------

def _write_table_atomic(table: pa.Table, path: Path) -> None:
    tmp = path.with_suffix(f".{uuid4().hex}.tmp")
    try:
        pq.write_table(table, str(tmp), compression="snappy")
        os.replace(str(tmp), str(path))
    except Exception:
        if tmp.exists():
            tmp.unlink(missing_ok=True)
        raise