fithia2/libs/backtest/breakout_52w_cache.py

"""Disk-based cache for breakout_52w ranked-universe results.

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

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

The cache stores rows that have already passed all universe-quality gates
(min_price, min_adv, is_52w_breakout). 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__)

BREAKOUT_52W_CACHE_VERSION = 1

# Params that determine the cached content.
# top_n is NOT here — it only slices the cached ranking.
_CACHE_KEY_PARAMS = (
    "breakout_52w_lookback_days",
    "breakout_52w_min_avg_dollar_volume",
    "breakout_52w_min_price",
)

_SCHEMA = pa.schema([
    pa.field("decision_date", pa.string()),
    pa.field("symbol", pa.string()),
    pa.field("volume_ratio_20d", pa.float64()),
    pa.field("avg_dollar_volume_20d", pa.float64()),
    pa.field("last_close", pa.float64()),
    pa.field("prior_252d_max_high", 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"breakout_52w_v{BREAKOUT_52W_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 Breakout52wRankCache:
    """Per-snapshot-dir disk cache for breakout-52w 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"breakout_52w_v{BREAKOUT_52W_CACHE_VERSION}__{param_hash}.parquet"
        )
        self._by_date: dict[str, list[dict[str, Any]]] | None = None
        self._new_rows: list[dict[str, Any]] = []

    def _ensure_loaded(self) -> None:
        if self._by_date is not None:
            return
        if not self._cache_path.exists():
            logger.info(
                "breakout_52w_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"breakout_52w_fingerprint") or b"").decode()
            if stored_fp != self._fingerprint:
                logger.info(
                    "breakout_52w_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(
                "breakout_52w_cache_hit",
                cache_file=str(self._cache_path),
                dates=len(by_date),
                rows=table.num_rows,
            )
        except Exception as exc:
            logger.warning(
                "breakout_52w_cache_read_failed",
                cache_file=str(self._cache_path),
                error=str(exc),
            )
            self._by_date = {}

    def get_date(self, decision_date: dt.date) -> list[dict[str, Any]] | None:
        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:
        self._new_rows.extend(rows)

    def close(self) -> None:
        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"breakout_52w_fingerprint": self._fingerprint.encode()}
        )
        self._cache_dir.mkdir(parents=True, exist_ok=True)
        _write_table_atomic(table, self._cache_path)
        logger.info(
            "breakout_52w_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 = []


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