"""Disk-based cache for EarningsRunup per-symbol-per-day feature inputs.

Phase 2 mirror of ``libs/backtest/xsmom_cache.py``. ER re-evaluates the same
per-symbol-per-day inputs (attention z-scores, dollar-volume z-scores, 20-day
momentum, upcoming-earnings reaction date) on every backtest run for a 900+
symbol universe. Caching these eliminates the dominant compute cost.

Design notes
------------
* **Cache key excludes engine threshold params.** All threshold filters
  (`attention_zscore_20d_min`, `dollar_volume_zscore_20d_min`,
  `days_to_earnings_min/max`, `momentum_20d_min`, `min_avg_dollar_volume`) are
  re-applied at read time via ``evaluate_trigger`` so a single cache file can
  serve runs with different ER threshold configs.

* **The MISS path uses a fixed wide population window** (``CACHE_MAX_DAYS_TO_EARNINGS``)
  rather than the engine's ``days_to_earnings_max``. This guarantees that any
  future run with a wider window still gets a clean MISS-or-HIT, never a silent
  truncation. Engine-level ``days_to_earnings_max`` is then applied as a filter
  at read time.

* **Single shared cache across ER engines per run.** Cache key has no engine
  identity, so one cache instance is shared by every ER engine on a given date.
  Within a date, the first engine's MISS path populates an in-memory snapshot
  so subsequent engines (same date) HIT immediately without re-scanning.

Cache key: (snapshot_fingerprint, attention_provider_version, ER_CACHE_VERSION)
Cache file: <snapshot_dir>/.runtime_cache/er_v{N}__{param_hash}.parquet
"""
from __future__ import annotations

import datetime as dt
import hashlib
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__)

ER_CACHE_VERSION = 1

# Fixed wide population window. All MISS-path rows are gathered for symbols
# with upcoming earnings within this many trading days. Threshold filters
# (including engine-specific dmax) are applied post-cache.
CACHE_MAX_DAYS_TO_EARNINGS = 20

_SCHEMA = pa.schema([
    pa.field("decision_date", pa.string()),
    pa.field("symbol", pa.string()),
    pa.field("days_to_earnings", pa.int32()),
    pa.field("reaction_date", pa.string()),
    pa.field("momentum_20d", pa.float64()),  # nullable
    pa.field("dollar_volume_zscore_20d", pa.float64()),
    pa.field("attention_zscore_20d", 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"er_v{ER_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 compute_attention_provider_version(attention_service: Any) -> str:
    """Derive a 16-char stable hash for the attention provider.

    AttentionFilterService has no explicit ``version`` attribute; we hash the
    base URL + scoring_model since those uniquely determine the z-score
    distribution Oracle returns. Stable across runs with identical config.
    """
    base_url = getattr(attention_service, "_base_url", "") or ""
    scoring_model = getattr(attention_service, "_scoring_model", "") or ""
    raw = f"{base_url}|{scoring_model}"
    return hashlib.sha256(raw.encode()).hexdigest()[:16]


def build_param_hash(
    snapshot_fingerprint: str,
    attention_provider_version: str,
) -> str:
    """16-char hex hash combining all cache-key params."""
    raw = (
        f"snap={snapshot_fingerprint}|"
        f"att={attention_provider_version}|"
        f"max_d2e={CACHE_MAX_DAYS_TO_EARNINGS}"
    )
    return hashlib.sha256(raw.encode()).hexdigest()[:16]


class ErInputsCache:
    """Per-snapshot-dir disk cache for EarningsRunup per-symbol-per-day inputs.

    Lifecycle:
      1. Construct once per backtest run (lazy, on first ER scheduling call).
      2. Pass to ``build_earnings_runup_candidates`` on every call.
      3. Call ``close()`` after the backtest loop to flush new rows to disk.

    Cache is shared across all ER engines because the cache key has no
    engine identity. Within a single date, the MISS path immediately
    publishes its rows into the in-memory map so subsequent engine calls
    (same date) hit synchronously.
    """

    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"er_v{ER_CACHE_VERSION}__{param_hash}.parquet"
        )
        # None = not yet loaded; {} = loaded (possibly empty due to miss)
        self._by_date: dict[str, list[dict[str, Any]]] | None = None
        self._new_rows: list[dict[str, Any]] = []
        # Track which dates are populated for THIS run (HIT or just-saved).
        # Used so the second engine on the same date sees the first engine's
        # MISS-path output without going through disk.
        self._dates_populated_this_run: set[str] = set()

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

    def _ensure_loaded(self) -> None:
        if self._by_date is not None:
            return
        if not self._cache_path.exists():
            logger.info(
                "er_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"er_fingerprint") or b"").decode()
            if stored_fp != self._fingerprint:
                logger.info(
                    "er_cache_miss",
                    reason="fingerprint_mismatch",
                    cache_file=str(self._cache_path),
                    stored_fp=stored_fp[:16],
                    current_fp=self._fingerprint[:16],
                )
                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
            # Dates loaded from disk also count as "populated this run" so
            # within-run calls hit the in-memory map.
            self._dates_populated_this_run.update(by_date.keys())
            logger.info(
                "er_cache_hit",
                cache_file=str(self._cache_path),
                dates=len(by_date),
                rows=table.num_rows,
            )
        except Exception as exc:
            logger.warning(
                "er_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 rows for decision_date, or None on miss.

        Only returns rows for dates that are confirmed populated (either
        loaded from disk or saved earlier in this run). A date with zero rows
        legitimately populated returns ``[]`` (HIT, empty universe);
        ``None`` only means MISS-not-yet-populated.
        """
        self._ensure_loaded()
        key = decision_date.isoformat()
        if key not in self._dates_populated_this_run:
            return None
        return self._by_date.get(key, [])  # type: ignore[union-attr]

    def save_date(self, decision_date: dt.date, rows: list[dict[str, Any]]) -> None:
        """Buffer rows for this decision_date and publish to in-memory map.

        Within-run subsequent calls to ``get_date`` for the same date will see
        these rows immediately (in-memory), without disk round-trip.
        """
        self._ensure_loaded()
        key = decision_date.isoformat()
        self._new_rows.extend(rows)
        # Publish to in-memory map so subsequent ER engines on the same date HIT.
        if self._by_date is not None:
            existing = self._by_date.setdefault(key, [])
            existing.extend(rows)
        self._dates_populated_this_run.add(key)

    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:
            # _by_date already contains the newly-saved rows from save_date,
            # so we DON'T extend with _new_rows again — that would duplicate.
            for date_rows in self._by_date.values():
                all_rows.extend(date_rows)
        else:
            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"er_fingerprint": self._fingerprint.encode()}
        )
        self._cache_dir.mkdir(parents=True, exist_ok=True)
        _write_table_atomic(table, self._cache_path)
        logger.info(
            "er_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 xsmom_cache.py / 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


__all__ = [
    "ER_CACHE_VERSION",
    "CACHE_MAX_DAYS_TO_EARNINGS",
    "ErInputsCache",
    "build_param_hash",
    "compute_attention_provider_version",
    "compute_snapshot_fingerprint",
]
