Spaces:
Running
Running
| 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"])) | |