feat(autotune): run_autotune orchestrator with heatmap output

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

autotune.py

@@ -5,6 +5,7 @@ from typing import Optional
 
 import numpy as np
 import pandas as pd
+import plotly.graph_objects as go
 
 _TUNING_DDL = """
 CREATE TABLE IF NOT EXISTS tuning_results (
@@ -147,3 +148,95 @@ def backtest_core(predict_fn, fetch_fn,
         "sharpe":           sharpe,
         "per_anchor":       df,
     }
+
+
+GRID_T     = (0.5, 1.0, 1.5)
+GRID_TOP_P = (0.7, 0.85, 0.95)
+
+def run_autotune(*, predict_fn, fetch_fn, db_path: str,
+                 symbol: str, interval: str,
+                 start_date: str, end_date: str,
+                 lookback: int, pred_len: int, stride: int,
+                 sample_count: int, max_anchors: int):
+    params = dict(lookback=lookback, pred_len=pred_len, stride=stride,
+                  sample_count=sample_count, max_anchors=max_anchors,
+                  start_date=start_date, end_date=end_date)
+    results = []
+    for T in GRID_T:
+        for top_p in GRID_TOP_P:
+            try:
+                m = backtest_core(
+                    predict_fn, fetch_fn,
+                    symbol=symbol, interval=interval,
+                    start_date=start_date, end_date=end_date,
+                    lookback=lookback, pred_len=pred_len, stride=stride,
+                    T=T, top_p=top_p,
+                    sample_count=sample_count, max_anchors=max_anchors,
+                )
+                results.append({
+                    "T": T, "top_p": top_p, "error": None,
+                    "anchors": m["anchors"],
+                    "mean_rmse": m["mean_rmse"],
+                    "hit_rate": m["hit_rate"],
+                    "total_return_pct": m["total_return_pct"],
+                    "max_dd_pct": m["max_dd_pct"],
+                    "sharpe": m["sharpe"],
+                })
+            except Exception as e:
+                results.append({
+                    "T": T, "top_p": top_p, "error": str(e),
+                    "anchors": 0, "mean_rmse": float("nan"),
+                    "hit_rate": float("nan"),
+                    "total_return_pct": float("nan"),
+                    "max_dd_pct": float("nan"),
+                    "sharpe": float("nan"),
+                })
+
+    valid = [r for r in results if r["error"] is None]
+    if not valid:
+        first_err = results[0]["error"]
+        raise RuntimeError(f"All grid cells failed: {first_err}")
+
+    best = max(valid, key=lambda r: r["total_return_pct"])
+    save_tuning(db_path, symbol, interval, best, results, params)
+
+    grid_df = pd.DataFrame(results)
+    z = np.full((len(GRID_T), len(GRID_TOP_P)), np.nan)
+    for i, T in enumerate(GRID_T):
+        for j, top_p in enumerate(GRID_TOP_P):
+            row = next(r for r in results if r["T"] == T and r["top_p"] == top_p)
+            if row["error"] is None:
+                z[i, j] = row["total_return_pct"]
+
+    fig = go.Figure(data=go.Heatmap(
+        z=z, x=[str(p) for p in GRID_TOP_P], y=[str(t) for t in GRID_T],
+        colorscale="RdYlGn", zmid=0.0, colorbar=dict(title="P&L %"),
+    ))
+    fig.update_layout(
+        title=f"{symbol.upper()} {interval} — Auto-tune heatmap (P&L %)",
+        xaxis_title="top_p", yaxis_title="T",
+        template="plotly_dark", height=420,
+        margin=dict(l=20, r=20, t=60, b=20),
+    )
+    fig.add_annotation(
+        x=str(best["top_p"]), y=str(best["T"]),
+        text="★", showarrow=False, font=dict(size=24, color="black"),
+    )
+
+    cells_errored = sum(1 for r in results if r["error"] is not None)
+    summary = pd.DataFrame({
+        "Field": ["Symbol", "Interval", "Best T", "Best top_p",
+                  "Best P&L %", "Best hit rate %", "Best mean RMSE",
+                  "Anchors per cell", "Total cells", "Cells errored"],
+        "Value": [
+            symbol.upper(), interval,
+            f"{best['T']}", f"{best['top_p']}",
+            f"{best['total_return_pct']:+.3f}",
+            f"{best['hit_rate'] * 100.0:.2f}",
+            f"{best['mean_rmse']:.4f}",
+            f"{best['anchors']}", f"{len(results)}",
+            f"{cells_errored}",
+        ],
+    })
+
+    return fig, summary, grid_df

tests/test_autotune.py

@@ -134,3 +134,85 @@ def test_backtest_core_raises_when_lookback_plus_predlen_too_big():
                       lookback=400, pred_len=200, stride=50,
                       T=1.0, top_p=0.9,
                       sample_count=1, max_anchors=5)
+
+
+from unittest.mock import MagicMock
+from autotune import run_autotune, GRID_T, GRID_TOP_P
+
+def test_run_autotune_iterates_full_grid_and_picks_max(tmp_db, monkeypatch):
+    init_tuning_table(tmp_db)
+    calls = []
+
+    def fake_core(predict_fn, fetch_fn, *, symbol, interval, start_date, end_date,
+                  lookback, pred_len, stride, T, top_p, sample_count, max_anchors):
+        calls.append((T, top_p))
+        # Make T=1.0, top_p=0.85 win clearly
+        score = 9.0 if (T == 1.0 and top_p == 0.85) else 1.0
+        return {
+            "anchors": 5, "mean_rmse": 1.0,
+            "hit_rate": 0.5,
+            "total_return_pct": score,
+            "max_dd_pct": -1.0, "sharpe": 0.5,
+            "per_anchor": pd.DataFrame(),
+        }
+
+    monkeypatch.setattr("autotune.backtest_core", fake_core)
+    fig, summary, grid = run_autotune(
+        predict_fn=MagicMock(), fetch_fn=MagicMock(),
+        db_path=tmp_db, symbol="SPY", interval="5min",
+        start_date="", end_date="",
+        lookback=200, pred_len=30, stride=30,
+        sample_count=1, max_anchors=5,
+    )
+    assert sorted(calls) == sorted([(t, p) for t in GRID_T for p in GRID_TOP_P])
+    assert len(grid) == 9
+    best_row = summary[summary["Field"] == "Best T"].iloc[0]["Value"]
+    assert float(best_row) == 1.0
+
+    persisted = load_tuning(tmp_db, "SPY", "5min")
+    assert persisted["best_T"] == 1.0
+    assert persisted["best_top_p"] == 0.85
+    assert persisted["best_pnl_pct"] == 9.0
+
+def test_run_autotune_skips_failed_cells(tmp_db, monkeypatch):
+    init_tuning_table(tmp_db)
+
+    def flaky_core(predict_fn, fetch_fn, *, T, top_p, **kw):
+        if T == 0.5:
+            raise RuntimeError("fmp blew up")
+        return {"anchors": 5, "mean_rmse": 1.0, "hit_rate": 0.5,
+                "total_return_pct": 2.0 + top_p,
+                "max_dd_pct": -1.0, "sharpe": 0.5,
+                "per_anchor": pd.DataFrame()}
+
+    monkeypatch.setattr("autotune.backtest_core", flaky_core)
+    fig, summary, grid = run_autotune(
+        predict_fn=MagicMock(), fetch_fn=MagicMock(),
+        db_path=tmp_db, symbol="X", interval="5min",
+        start_date="", end_date="",
+        lookback=200, pred_len=30, stride=30,
+        sample_count=1, max_anchors=5,
+    )
+    errored = grid[grid["error"].notna()]
+    assert len(errored) == 3                      # all T=0.5 rows
+    valid = grid[grid["error"].isna()]
+    assert len(valid) == 6
+    persisted = load_tuning(tmp_db, "X", "5min")
+    assert persisted["best_T"] != 0.5            # never the failing T
+
+def test_run_autotune_raises_when_all_cells_fail(tmp_db, monkeypatch):
+    init_tuning_table(tmp_db)
+    def always_fail(*a, **kw):
+        raise RuntimeError("everything is broken")
+    monkeypatch.setattr("autotune.backtest_core", always_fail)
+
+    import pytest
+    with pytest.raises(RuntimeError, match="All grid cells failed"):
+        run_autotune(
+            predict_fn=MagicMock(), fetch_fn=MagicMock(),
+            db_path=tmp_db, symbol="X", interval="5min",
+            start_date="", end_date="",
+            lookback=200, pred_len=30, stride=30,
+            sample_count=1, max_anchors=5,
+        )
+    assert load_tuning(tmp_db, "X", "5min") is None