momentum-system/nightly_tune.py

#!/usr/bin/env python3
"""Nightly self-tuning loop.

Run after the close: backfill the latest bars, sweep the ORB strategy parameters
over the accumulated gapper sessions, and write the best config to
data/learned_params.json. run_live_paper.py reads that file on boot, so the
strategy re-tunes itself as data grows — no manual intervention.

    PYTHONPATH=. /home/philip/.venvs/momentum/bin/python nightly_tune.py

Only writes a config that is actually PROFITABLE on the sample (>0 P&L, >=5
trades); otherwise it leaves the existing learned_params.json untouched so a thin
or bad day can't degrade the live config.
"""

from __future__ import annotations

import itertools
import json
import os
from datetime import datetime, timezone

from dotenv import load_dotenv

from research.multi_schema import open_research_db

load_dotenv()

# config parity with the live loop: DB-backed knobs (config_overrides) beat
# compose env — a bare env would replay under DEFAULT exits (the 4.5x-baseline
# landmine, orderflow-shelf-verdicts 2026-07-17)
from runtime.env_overrides import load as _load_cfg_overrides  # noqa: E402
_load_cfg_overrides()


def _write_heartbeat(outcome: str) -> None:
    """Record that the tuner RAN (regardless of whether it found a profitable
    config), so the boot staleness check can tell 'tuner not running' (real
    problem) from 'tuner ran, no profitable ORB config, kept params' (expected
    when the strategy has no edge — the common case). Best-effort."""
    try:
        path = os.path.join(os.environ.get("DATA_DIR", "./data"), "tuner_heartbeat.json")
        os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
        with open(path, "w") as f:
            json.dump({"ran_at": datetime.now(timezone.utc).isoformat(),
                       "outcome": outcome}, f)
    except Exception:  # noqa: BLE001 — a heartbeat must never break tuning
        pass
    # ALSO record in the DB: the file lives on one container's volume, so a
    # DATA_DIR mismatch / stale scheduler image makes it silently unverifiable
    # from anywhere else (the 2026-07-22 "tuner NOT running" alert autopsy).
    # The DB row is queryable remotely and survives container churn.
    try:
        con = open_research_db("market")
        import uuid
        con.execute(
            "INSERT INTO telemetry_events (id, event_type, source, reason, metadata_json) "
            "VALUES (?, ?, ?, ?, ?)",
            [str(uuid.uuid4()), "tuner_heartbeat", "nightly_tune", outcome,
             json.dumps({"ran_at": datetime.now(timezone.utc).isoformat()})])
    except Exception:  # noqa: BLE001
        pass


def main():
    # Tune the profitable setup only.
    os.environ["STRATEGY_SETUPS"] = "opening_range_break"
    _write_heartbeat("started")

    # 1) accumulate the latest sessions (best-effort; never abort tuning on this)
    try:
        import backfill_history
        backfill_history.main()
    except Exception as exc:  # noqa: BLE001
        print(f"backfill skipped: {exc}", flush=True)

    # imported after backfill so STRATEGY_SETUPS is already set for classify_setup
    from learn_params import load_sessions, run_combo

    con = open_research_db("market")
    sessions = load_sessions(con)
    print(f"tuning over {len(sessions)} gapper sessions", flush=True)
    if len(sessions) < 10:
        _write_heartbeat("kept:too_few_sessions")
        print("too few sessions — keeping current learned_params.json")
        return

    grid = {
        "warmup": [3, 6],
        "min_bars": [3, 5],
        "eval_every": [1],
        "ready": [55, 60, 65, 70],
        "gap_min": [0.05],
        "rvol_min": [2.0],
    }
    keys = list(grid)
    best = None
    for combo in itertools.product(*grid.values()):
        p = dict(zip(keys, combo))
        if p["warmup"] < p["min_bars"]:
            continue
        m = run_combo(sessions, **p)
        if m["trades"] < 5 or m["pnl"] <= 0:
            continue
        # score: profit, lightly favouring earlier entries on ties
        score = m["pnl"] - (m["entry_min"] * 50.0)
        if best is None or score > best[2]:
            best = (p, m, score)

    if best is None:
        _write_heartbeat("kept:no_profitable_config")
        print("no profitable ORB config on this sample — keeping current params")
        return

    p, m, _ = best
    out = {
        "min_bars": p["min_bars"],
        "ready_score_pct": float(p["ready"]),
        "setups": "opening_range_break",
        "as_of": datetime.now(timezone.utc).isoformat(),
        "sessions": len(sessions),
        "pnl": round(m["pnl"]),
        "win": round(m["win"], 3),
        "avg_R": round(m["avgR"], 3),
        "entry_min": round(m["entry_min"], 1),
    }
    path = os.path.join(os.environ.get("DATA_DIR", "./data"), "learned_params.json")
    os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
    # Atomic write: boot reads this on every restart, so a torn write (crash
    # mid-dump) would leave the live loop parsing half a JSON. tmp -> fsync ->
    # os.replace makes the swap atomic on POSIX.
    tmp = path + ".tmp"
    with open(tmp, "w") as f:
        json.dump(out, f, indent=2)
        f.flush()
        os.fsync(f.fileno())
    os.replace(tmp, path)
    _write_heartbeat("wrote:profitable")
    print(f"wrote {path}:\n{json.dumps(out, indent=2)}", flush=True)


if __name__ == "__main__":
    main()