signal-platform/scripts/research_snapshot_manifest.py

"""Completion manifest for research.sqlite — cheap race guard.

The 2026-07-18 21:14 breadth run fired while ``extend_snapshot_universe`` was
still (or had just been) building the snapshot. Harness and shared-filter
recomputes agree on *complete* data, so the orphaned +0.0575 was incomplete
universe, not a code path bug.

Same class of protection as calendar-truncation assertions in the research
matrix: refuse to read results from a half-built artifact.

Layout
------
Sidecar path: ``<snapshot>.manifest.json`` next to the sqlite file
(e.g. ``backtest_snapshots/research.sqlite.manifest.json``).
"""

from __future__ import annotations

import json
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

from sqlalchemy import create_engine, text

MANIFEST_SCHEMA_VERSION = 1


def manifest_path_for(snapshot: Path) -> Path:
    """Sidecar path for a research snapshot."""
    return Path(str(snapshot) + ".manifest.json")


def _count_snapshot(snapshot: Path) -> dict[str, int]:
    engine = create_engine(
        f"sqlite:///{snapshot.resolve().as_posix()}",
        future=True,
    )
    try:
        with engine.connect() as conn:
            ticker_n = int(conn.execute(text("SELECT COUNT(*) FROM tickers")).scalar_one())
            ohlcv_n = int(
                conn.execute(text("SELECT COUNT(*) FROM ohlcv_records")).scalar_one()
            )
            try:
                rank_only_n = int(
                    conn.execute(text("SELECT COUNT(*) FROM research_rank_only")).scalar_one()
                )
            except Exception:
                rank_only_n = 0
    finally:
        engine.dispose()
    return {
        "ticker_count": ticker_n,
        "ohlcv_row_count": ohlcv_n,
        "rank_only_count": rank_only_n,
    }


def write_completion_manifest(
    snapshot: Path,
    *,
    complete: bool,
    sources: dict[str, str] | None = None,
    history_days: int | None = None,
    min_bars: int | None = None,
    fetch_ok: int | None = None,
    fetch_fail: int | None = None,
    limit: int | None = None,
    extra: dict[str, Any] | None = None,
) -> Path:
    """Write (or overwrite) the sidecar completion manifest for *snapshot*."""
    snapshot = Path(snapshot)
    counts = _count_snapshot(snapshot) if snapshot.exists() else {
        "ticker_count": 0,
        "ohlcv_row_count": 0,
        "rank_only_count": 0,
    }
    payload: dict[str, Any] = {
        "schema_version": MANIFEST_SCHEMA_VERSION,
        "snapshot": snapshot.name,
        "snapshot_resolved": str(snapshot.resolve()) if snapshot.exists() else str(snapshot),
        "complete": bool(complete),
        "finished_at": datetime.now(timezone.utc).isoformat(),
        **counts,
        "sources": sources or {},
        "history_days": history_days,
        "min_bars": min_bars,
        "fetch_ok": fetch_ok,
        "fetch_fail": fetch_fail,
        "limit": limit,
    }
    if extra:
        payload["extra"] = extra
    path = manifest_path_for(snapshot)
    path.write_text(json.dumps(payload, indent=2, default=str) + "\n", encoding="utf-8")
    return path


def clear_manifest(snapshot: Path) -> None:
    """Remove any existing completion manifest (start of a rebuild)."""
    path = manifest_path_for(Path(snapshot))
    if path.exists():
        path.unlink()


def load_manifest(snapshot: Path) -> dict[str, Any] | None:
    path = manifest_path_for(Path(snapshot))
    if not path.exists():
        return None
    return json.loads(path.read_text(encoding="utf-8"))


def assert_research_snapshot_complete(snapshot: Path) -> dict[str, Any]:
    """Refuse breadth-mode work unless the extender finished cleanly.

    Raises ``SystemExit`` with a clear message on any failure (missing
    manifest, incomplete flag, or live counts that no longer match the
    recorded totals — e.g. a mid-run overwrite of the sqlite file).
    """
    snapshot = Path(snapshot)
    if not snapshot.exists():
        raise SystemExit(
            f"Research snapshot missing: {snapshot}\n"
            "Build it with: python scripts/extend_snapshot_universe.py"
        )

    path = manifest_path_for(snapshot)
    if not path.exists():
        raise SystemExit(
            f"Research snapshot completion manifest missing: {path}\n"
            "Refusing breadth run — this is the guard that would have caught "
            "the 2026-07-18 21:14 race against a half-built research.sqlite.\n"
            "Re-run extend_snapshot_universe.py to completion (no --limit), "
            "or for a trusted existing full snapshot:\n"
            "  python -c \"from pathlib import Path; "
            "from scripts.research_snapshot_manifest import write_completion_manifest; "
            f"write_completion_manifest(Path(r'{snapshot}'), complete=True)\""
        )

    try:
        manifest = json.loads(path.read_text(encoding="utf-8"))
    except json.JSONDecodeError as exc:
        raise SystemExit(f"Corrupt research snapshot manifest {path}: {exc}") from exc

    if not manifest.get("complete"):
        raise SystemExit(
            f"Research snapshot marked incomplete in {path}\n"
            f"(finished_at={manifest.get('finished_at')}, limit={manifest.get('limit')}).\n"
            "Re-run extend_snapshot_universe.py without --limit until Done."
        )

    live = _count_snapshot(snapshot)
    mismatches: list[str] = []
    for key in ("ticker_count", "ohlcv_row_count", "rank_only_count"):
        recorded = manifest.get(key)
        if recorded is None:
            mismatches.append(f"{key}: missing in manifest")
            continue
        if int(recorded) != int(live[key]):
            mismatches.append(
                f"{key}: manifest={recorded} live={live[key]}"
            )
    if mismatches:
        raise SystemExit(
            "Research snapshot does not match its completion manifest "
            f"({path}). Likely a partial rewrite or concurrent extend:\n  - "
            + "\n  - ".join(mismatches)
            + "\nRe-run extend_snapshot_universe.py to completion."
        )

    return {**manifest, "live_counts": live}