import json import sqlite3 from datetime import datetime, timezone from typing import Optional, Tuple import numpy as np import pandas as pd import plotly.graph_objects as go _TUNING_DDL = """ CREATE TABLE IF NOT EXISTS tuning_results ( id INTEGER PRIMARY KEY AUTOINCREMENT, ts TEXT NOT NULL, symbol TEXT NOT NULL, interval TEXT NOT NULL, lookback INTEGER, pred_len INTEGER, stride INTEGER, sample_count INTEGER, max_anchors INTEGER, start_date TEXT, end_date TEXT, best_T REAL, best_top_p REAL, best_pnl_pct REAL, best_hit_rate REAL, best_rmse REAL, grid_json TEXT, UNIQUE(symbol, interval) ON CONFLICT REPLACE ) """ def init_tuning_table(db_path: str) -> None: with sqlite3.connect(db_path) as c: c.execute(_TUNING_DDL) def save_tuning(db_path: str, symbol: str, interval: str, best: dict, results: list, params: dict) -> None: ts = datetime.now(timezone.utc).isoformat(timespec="seconds") grid_json = json.dumps(results, default=float) with sqlite3.connect(db_path) as c: c.execute( "INSERT INTO tuning_results " "(ts, symbol, interval, lookback, pred_len, stride, sample_count, " " max_anchors, start_date, end_date, best_T, best_top_p, " " best_pnl_pct, best_hit_rate, best_rmse, grid_json) " "VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", (ts, symbol, interval, int(params["lookback"]), int(params["pred_len"]), int(params["stride"]), int(params["sample_count"]), int(params["max_anchors"]), params.get("start_date", "") or "", params.get("end_date", "") or "", float(best["T"]), float(best["top_p"]), float(best["total_return_pct"]), float(best["hit_rate"]), float(best["mean_rmse"]), grid_json), ) def load_tuning(db_path: str, symbol: str, interval: str) -> Optional[dict]: with sqlite3.connect(db_path) as c: c.row_factory = sqlite3.Row row = c.execute( "SELECT * FROM tuning_results WHERE symbol=? AND interval=? " "ORDER BY id DESC LIMIT 1", (symbol, interval), ).fetchone() return dict(row) if row else None BACKTEST_COST_BP = 1.0 # 1 bp per round-trip trade def backtest_core(predict_fn, fetch_fn, symbol: str, interval: str, start_date: str, end_date: str, lookback: int, pred_len: int, stride: int, T: float, top_p: float, sample_count: int, max_anchors: int) -> dict: lookback = int(lookback) pred_len = int(pred_len) stride = max(int(stride), 1) sample_count = max(int(sample_count), 1) max_anchors = max(int(max_anchors), 1) if lookback + pred_len > 512: raise ValueError(f"lookback + pred_len must be ≤ 512 (got {lookback + pred_len})") full = fetch_fn(symbol.upper(), interval, 5000) sd = (start_date or "").strip() ed = (end_date or "").strip() if sd: try: full = full[full["timestamps"] >= pd.Timestamp(sd)] except Exception as e: raise ValueError(f"Bad start_date '{sd}': {e}") if ed: try: full = full[full["timestamps"] <= pd.Timestamp(ed)] except Exception as e: raise ValueError(f"Bad end_date '{ed}': {e}") full = full.reset_index(drop=True) if len(full) < lookback + pred_len + 1: raise ValueError(f"Not enough bars in window (got {len(full)}, need ≥ {lookback + pred_len + 1})") last_anchor = len(full) - pred_len anchors = list(range(lookback, last_anchor, stride)) if not anchors: raise ValueError("No anchors to evaluate (window too tight)") if len(anchors) > max_anchors: idx = np.linspace(0, len(anchors) - 1, max_anchors).astype(int) anchors = [anchors[i] for i in idx] cost = BACKTEST_COST_BP / 1e4 rows = [] for ai in anchors: x_df = full.iloc[ai - lookback:ai].reset_index(drop=True) future_df = full.iloc[ai:ai + pred_len].reset_index(drop=True) x_timestamp = x_df["timestamps"] y_timestamp = future_df["timestamps"] pred_df = predict_fn( df=x_df[["open","high","low","close","volume","amount"]], x_timestamp=x_timestamp, y_timestamp=y_timestamp, pred_len=pred_len, T=T, top_p=top_p, sample_count=sample_count, verbose=False, ) forecast_close = pred_df["close"].values realized_close = future_df["close"].values last_close = float(x_df["close"].iloc[-1]) rmse = float(np.sqrt(np.mean((forecast_close - realized_close) ** 2))) forecast_dir = 1 if forecast_close[-1] > last_close else -1 realized_dir = 1 if realized_close[-1] > last_close else -1 hit = int(forecast_dir == realized_dir) realized_ret = realized_close[-1] / last_close - 1.0 trade_ret = forecast_dir * realized_ret - cost rows.append({ "anchor_ts": x_timestamp.iloc[-1], "last_close": last_close, "forecast_close": float(forecast_close[-1]), "realized_close": float(realized_close[-1]), "rmse": rmse, "hit": hit, "trade_pnl": trade_ret, }) df = pd.DataFrame(rows) df["cum_pnl"] = (1.0 + df["trade_pnl"]).cumprod() - 1.0 df["hit_rate_running"] = df["hit"].expanding().mean() equity = (1.0 + df["trade_pnl"]).cumprod() rmax = equity.cummax() max_dd = float(((equity - rmax) / rmax).min()) sharpe = float(df["trade_pnl"].mean() / df["trade_pnl"].std()) if df["trade_pnl"].std() > 0 else 0.0 return { "anchors": int(len(df)), "mean_rmse": float(df["rmse"].mean()), "hit_rate": float(df["hit_rate_running"].iloc[-1]), "total_return_pct": float(df["cum_pnl"].iloc[-1] * 100.0), "max_dd_pct": float(max_dd * 100.0), "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 def apply_tuning(db_path: str, symbol: str, interval: str, use_tuned: bool) -> Tuple[Optional[float], Optional[float]]: if not use_tuned: return (None, None) row = load_tuning(db_path, symbol, interval) if row is None: return (None, None) return (float(row["best_T"]), float(row["best_top_p"]))