scripts/seed_trackrecord.py

"""Backfill the track record by forecasting through recent history, honestly.

The point of a backfill is to give a new model a track record without waiting
months for one. The danger is that a backfill is trivially easy to fake: run
the model over history it has already seen, or let it peek at the bar it is
predicting, and it will look extraordinary.

So this script forecasts *as of* each historical timestamp. The context is
sliced at that timestamp by `runtime.load_context`, and the adapter re-checks
the slice against `issued_ts` on every call -- a context bar dated after the
issue moment raises rather than producing a flattering number. Nothing here
opts out of that check; there is no fast path that skips it.

Every row written carries `backfilled: true`, and the UI badges those entries
wherever they appear. A backfilled record is evidence of a kind, but it is not
the same evidence as a forecast made before the outcome existed, and the
difference is on the page rather than in a footnote.

    HF_TOKEN=$(cat ../../key.txt) python scripts/seed_trackrecord.py
    python scripts/seed_trackrecord.py --dry-run --models chronos-bolt-tiny
"""

from __future__ import annotations

import argparse
import json
import logging
import os
import sys
from pathlib import Path

import pandas as pd

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from src import config, runtime, trackrecord        # noqa: E402
from src.store import ArenaStore, now_utc           # noqa: E402

logging.basicConfig(level=logging.INFO, format="%(levelname)-7s %(message)s")
log = logging.getLogger("backfill")

# How many forecasts to seed per (model, asset, timeframe).
TARGET_PER_SERIES = 60

# Space the issue points out so the windows do not all overlap: consecutive
# forecasts sharing 95% of their context would give 60 nearly identical
# observations and a coverage figure with far less evidence behind it than the
# count suggests.
STRIDE = {"1h": 12, "1d": 3}


def checkpoint_path(root: Path) -> Path:
    return root / "backfill_checkpoint.json"


def load_checkpoint(root: Path) -> dict:
    p = checkpoint_path(root)
    if p.exists():
        try:
            return json.loads(p.read_text())
        except Exception:
            log.warning("unreadable checkpoint; starting fresh")
    return {"done": {}}


def save_checkpoint(root: Path, state: dict) -> None:
    checkpoint_path(root).parent.mkdir(parents=True, exist_ok=True)
    checkpoint_path(root).write_text(json.dumps(state, indent=2, sort_keys=True))


def series_key(model: str, asset: str, tf: str) -> str:
    return f"{model}|{asset}|{tf}"


def backfill_series(store: ArenaStore, model_slug: str, asset: str, tf: str,
                    registry: dict, target: int, horizon: int,
                    seed: int, dry_run: bool) -> int:
    """Issue `target` as-of forecasts, oldest first. Returns rows archived."""
    prices = store.get_prices(asset, tf)
    if not len(prices):
        log.warning("  no prices for %s %s", asset, tf)
        return 0

    ts = pd.to_datetime(prices["ts"], utc=True).reset_index(drop=True)
    stride = STRIDE.get(tf, 6)

    # Only issue points whose horizon has already elapsed -- an unresolved
    # backfill row would sit in the archive forever without ever scoring.
    latest_resolvable = len(ts) - horizon - 1
    earliest = config.DEFAULT_CONTEXT_BARS + 1
    if latest_resolvable <= earliest:
        log.warning("  not enough history for %s %s (%d bars)", asset, tf, len(ts))
        return 0

    points = list(range(latest_resolvable, earliest, -stride))[:target]
    points.reverse()

    written = 0
    for i, idx in enumerate(points):
        as_of = ts.iloc[idx]
        try:
            run = runtime.run_forecast(
                store, model_slug, asset, tf, horizon=horizon,
                n_samples=config.DEFAULT_N_SAMPLES, seed=seed, as_of=as_of,
                registry=registry, archive=not dry_run, backfilled=True)
        except runtime.ForecastUnavailable as e:
            log.warning("  %s: %s", as_of.date(), e)
            continue
        written += run.archived_rows
        if (i + 1) % 10 == 0:
            log.info("  %d/%d issued (%s)", i + 1, len(points), as_of.date())
    return written


def main() -> int:
    ap = argparse.ArgumentParser()
    ap.add_argument("--dry-run", action="store_true")
    ap.add_argument("--models", nargs="*", default=None,
                    help="model slugs; default is every CPU-tier enrolled model")
    ap.add_argument("--assets", nargs="*", default=list(config.SEED_ASSETS))
    ap.add_argument("--timeframes", nargs="*", default=list(config.SEED_TIMEFRAMES))
    ap.add_argument("--target", type=int, default=TARGET_PER_SERIES)
    ap.add_argument("--seed", type=int, default=42)
    ap.add_argument("--include-gpu", action="store_true",
                    help="also seed GPU-tier models (slow on CPU)")
    ap.add_argument("--local-root", default=None)
    args = ap.parse_args()

    store = ArenaStore(local_root=args.local_root,
                       offline=not os.environ.get("HF_TOKEN"))
    registry = store.get_registry()
    models = registry.get("models", {})
    if not models:
        log.error("registry is empty; run scripts/bootstrap_registry.py first")
        return 1

    if args.models:
        chosen = [m for m in args.models if m in models]
        missing = set(args.models) - set(chosen)
        if missing:
            log.error("not enrolled: %s", ", ".join(sorted(missing)))
            return 1
    else:
        chosen = [slug for slug, entry in sorted(models.items())
                  if args.include_gpu
                  or entry.get("capabilities", {}).get("hardware") == "cpu"]

    root = Path(store.local_root)
    checkpoint = load_checkpoint(root)
    total = 0

    for model_slug in chosen:
        log.info("model %s", model_slug)
        for asset in args.assets:
            for tf in args.timeframes:
                key = series_key(model_slug, asset, tf)
                if checkpoint["done"].get(key):
                    log.info("  %s %s already done, skipping", asset, tf)
                    continue
                horizon = config.DEFAULT_HORIZON.get(tf, 24)
                log.info("  %s %s (h=%d, target=%d)", asset, tf, horizon, args.target)
                written = backfill_series(store, model_slug, asset, tf, registry,
                                          args.target, horizon, args.seed,
                                          args.dry_run)
                total += written
                if not args.dry_run:
                    checkpoint["done"][key] = {
                        "rows": written, "ts": now_utc().isoformat()}
                    save_checkpoint(root, checkpoint)
                    # Commit per series rather than at the end: a run that dies
                    # halfway must not throw away hours of forecasting.
                    store.flush(f"arena: backfill {model_slug} {asset} {tf}")

    log.info("archived %d forecast rows", total)

    if not args.dry_run:
        log.info("resolving")
        for model_slug in chosen:
            for asset in args.assets:
                for tf in args.timeframes:
                    trackrecord.resolve(store, model_slug, asset, tf)
        trackrecord.regenerate_standings(store)
        store.flush("arena: resolve backfill and rebuild standings")
        log.info("standings rebuilt")

    return 0


if __name__ == "__main__":
    raise SystemExit(main())