From 6889e06a86e15a14cf1f89f8565c0f3a6881eea6 Mon Sep 17 00:00:00 2001 From: ramseshk <45832522+ramseshk@users.noreply.github.com> Date: Fri, 7 Aug 2026 14:14:40 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20VBT=20dashboard=20overhaul=20=E2=80=94?= =?UTF-8?q?=20pagination,=20caching,=20LTTB=20downsampling,=20deep=20links?= =?UTF-8?q?,=20export,=20more=20metrics?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - Merge vbt_server.py into server.py (eliminate duplicated VBT API) - Add server-side pagination (offset/limit) with metadata (total, has_more) - Add server-side ?asset= filtering to results endpoint - Add JSON file caching with 5s TTL to avoid re-parsing on every request - Add LTTB (Largest-Triangle-Three-Buckets) downsampling for equity curves - Add pre-computed drawdown curve to result detail response - Add /api/vbt/result/{filename}/csv endpoint for trade export - Add Calmar ratio and expectancy to results metadata - Rebuild vbt.html frontend with: - URL hash deep-linking (#filename) for bookmarkable views - JSON and CSV export buttons in detail panel - More metrics: Calmar, Sortino, Expectancy, End Equity (10 total) - Running backtest progress indicator with elapsed seconds - Pagination controls (prev/next) with page info - Filter/sort changes auto-apply (no manual refresh needed) - Better error states with retry buttons - Run strategy selector independent of filter --- dashboard/server.py | 262 ++++++++++++++++++++++++++------------ dashboard/static/vbt.html | 257 ++++++++++++++++++++++++++----------- dashboard/vbt_server.py | 204 ----------------------------- 3 files changed, 359 insertions(+), 364 deletions(-) delete mode 100644 dashboard/vbt_server.py diff --git a/dashboard/server.py b/dashboard/server.py index 85a2b0b..50fc496 100644 --- a/dashboard/server.py +++ b/dashboard/server.py @@ -462,32 +462,82 @@ async def get_risk_metrics(): # VBT Dashboard API — VectorBT backtest results browser # ═══════════════════════════════════════════════════════════ +_vbt_meta_cache: dict[str, dict] = {} # filename → parsed summary dict +_vbt_full_cache: dict[str, dict] = {} # filename → full result dict +_vbt_cache_time: float = 0.0 # epoch of last cache rebuild +_VBT_CACHE_TTL = 5.0 # seconds before re-scan + +def _refresh_vbt_cache(): + """Scan results dirs once and populate caches.""" + global _vbt_meta_cache, _vbt_full_cache, _vbt_cache_time + now = time.time() + if now - _vbt_cache_time < _VBT_CACHE_TTL: + return + new_meta: dict[str, dict] = {} + new_full: dict[str, dict] = {} + for d in [BACKTEST_DIR, HISTORICAL_DIR]: + if not os.path.isdir(d): + continue + for fname in sorted(os.listdir(d)): + if not fname.endswith(".json"): + continue + if fname in new_meta: + continue + fpath = os.path.join(d, fname) + try: + with open(fpath) as f: + data = json.load(f) + n = _normalize_vbt_fields(data) + asset = _infer_asset(n.get("strategy", ""), fname) + new_meta[fname] = { + "filename": fname, + "strategy": n.get("strategy", "unknown"), + "asset": asset, + "engine": n.get("engine", "vectorbt"), + "interval": n.get("interval", "1h"), + "sharpe": n.get("sharpe", 0), + "sortino": n.get("sortino", 0), + "total_return_pct": n["total_return_pct"], + "max_drawdown_pct": n["max_drawdown_pct"], + "calmar": round((n["total_return_pct"] / max(abs(n["max_drawdown_pct"]), 0.01)), 2), + "win_rate": n.get("win_rate", 0), + "profit_factor": n["profit_factor"], + "expectancy": n.get("expectancy", 0), + "total_trades": n["total_trades"], + "n_bars": n["n_bars"], + "generated_at": n.get("generated_at", ""), + "has_equity_curve": bool(n.get("equity_curve")), + } + new_full[fname] = n + except (json.JSONDecodeError, IOError): + pass + _vbt_meta_cache = new_meta + _vbt_full_cache = new_full + _vbt_cache_time = now + + def _normalize_vbt_fields(data: dict) -> dict: """Normalise old/new backtest file field names to a consistent schema.""" out = dict(data) - # total_return_pct if "total_return_pct" not in out: out["total_return_pct"] = out.get("pnl_pct", out.get("ann_return_pct", 0)) if out.get("total_return_pct") is None: out["total_return_pct"] = 0 - # max_drawdown_pct if "max_drawdown_pct" not in out: dd = out.get("max_dd_pct", out.get("max_dd")) if dd is not None and isinstance(dd, (int, float)) and abs(dd) < 1: - dd = dd * 100 # decimal → percent + dd = dd * 100 out["max_drawdown_pct"] = dd or 0 if out.get("max_drawdown_pct") is None: out["max_drawdown_pct"] = 0 - # n_bars if "n_bars" not in out: out["n_bars"] = out.get("num_periods", 0) if out.get("n_bars") is None: out["n_bars"] = 0 - # profit_factor → compute from trades if missing if "profit_factor" not in out and "trades" in out: trades = out.get("trades", []) if trades: @@ -500,24 +550,14 @@ def _normalize_vbt_fields(data: dict) -> dict: for t in trades if (t.get("pnl_net", t.get("pnl_gross", t.get("pnl", 0))) or 0) < 0 )) out["profit_factor"] = round(gross_win / gross_loss, 3) if gross_loss > 0 else 0 - elif out.get("pnl_gross") is not None and out.get("fees_total") is not None: - # Synthetic: approximate PF from gross/fees relationship - pnl_gross = out.get("pnl_gross", 0) - fees = out.get("fees_total", 0) - if fees > 0: - wins = pnl_gross + fees if pnl_gross > 0 else fees - losses = fees if pnl_gross > 0 else fees - pnl_gross - out["profit_factor"] = round(wins / losses, 3) if losses > 0 else 0 if "profit_factor" not in out: out["profit_factor"] = 0 - # total_trades if "total_trades" not in out: out["total_trades"] = len(out.get("trades", [])) if out.get("total_trades") is None: out["total_trades"] = 0 - # win_rate → compute from trades if missing/zero if not out.get("win_rate") and "trades" in out: trades = out.get("trades", []) if trades: @@ -527,67 +567,80 @@ def _normalize_vbt_fields(data: dict) -> dict: return out +def _lttb_downsample(points: list[dict], target: int) -> list[dict]: + """Largest-Triangle-Three-Buckets downsampling for visual fidelity.""" + n = len(points) + if n <= target or target < 3: + return points + bucket_size = (n - 2) / (target - 2) + result = [points[0]] + a = 0 + for i in range(target - 2): + avg_x_start = int((i + 0) * bucket_size) + 1 + avg_x_end = int((i + 1) * bucket_size) + 1 + avg_range = points[avg_x_start:avg_x_end] + avg_x = sum(p.get("t", 0) if isinstance(p.get("t"), (int, float)) else 0 for p in avg_range) / max(len(avg_range), 1) + avg_y = sum(p["v"] for p in avg_range) / max(len(avg_range), 1) + range_offs = int((i + 1) * bucket_size) + 1 + range_to = int((i + 2) * bucket_size) + 1 + max_area = -1.0 + for j in range(range_offs, min(range_to + 1, n)): + pa = result[a] + pc = points[j] + area = abs((pa.get("t", 0) if isinstance(pa.get("t"), (int, float)) else a) * (pc["v"] - avg_y) + + pc.get("t", 0) * (avg_y - pa["v"]) + + avg_x * (pa["v"] - pc["v"])) * 0.5 + if area > max_area: + max_area = area + next_a = j + result.append(points[next_a]) + a = next_a + result.append(points[-1]) + return result + + @app.get("/api/vbt/results") async def list_vbt_results( strategy: str = "", interval: str = "", + asset: str = "", sort: str = "date", limit: int = 100, + offset: int = 0, ): - """List VectorBT backtest results with full metrics and filtering.""" - results = [] - for d in [BACKTEST_DIR, HISTORICAL_DIR]: - if not os.path.isdir(d): - continue - for fname in sorted(os.listdir(d), reverse=True): - if not fname.endswith(".json"): - continue - if strategy and strategy not in fname: - continue - fpath = os.path.join(d, fname) - try: - with open(fpath) as f: - data = json.load(f) - n = _normalize_vbt_fields(data) - if interval and n.get("interval", "1h") != interval: - continue - # Infer asset from strategy or filename - asset = _infer_asset(n.get("strategy", ""), fname) - results.append({ - "filename": fname, - "strategy": n.get("strategy", "unknown"), - "asset": asset, - "engine": n.get("engine", "vectorbt"), - "interval": n.get("interval", "1h"), - "sharpe": n.get("sharpe", 0), - "sortino": n.get("sortino", 0), - "total_return_pct": n["total_return_pct"], - "max_drawdown_pct": n["max_drawdown_pct"], - "win_rate": n.get("win_rate", 0), - "profit_factor": n["profit_factor"], - "total_trades": n["total_trades"], - "n_bars": n["n_bars"], - "generated_at": n.get("generated_at", ""), - "has_equity_curve": bool(n.get("equity_curve")), - }) - except (json.JSONDecodeError, IOError): - pass - if len(results) >= limit: - break + """List VectorBT backtest results with full metrics, pagination, and server-side filtering.""" + _refresh_vbt_cache() + results = list(_vbt_meta_cache.values()) - # Sort - if sort == "sharpe": - results.sort(key=lambda r: r.get("sharpe", -999), reverse=True) - elif sort == "return": - results.sort(key=lambda r: r.get("total_return_pct", -999), reverse=True) - elif sort == "dd": - results.sort(key=lambda r: -abs(r.get("max_drawdown_pct", 999)), reverse=True) - elif sort == "trades": - results.sort(key=lambda r: r.get("total_trades", 0), reverse=True) - else: # date + if strategy: + results = [r for r in results if strategy in r.get("filename", "")] + if interval: + results = [r for r in results if r.get("interval") == interval] + if asset: + results = [r for r in results if r.get("asset", "") == asset or r.get("asset", "").endswith("/" + asset)] + + sort_keys = { + "sharpe": ("sharpe", True), + "return": ("total_return_pct", True), + "dd": ("max_drawdown_pct", False), + "trades": ("total_trades", True), + "calmar": ("calmar", True), + } + if sort in sort_keys: + key, rev = sort_keys[sort] + results.sort(key=lambda r: r.get(key, -999 if rev else 999), reverse=rev) + else: results.sort(key=lambda r: r.get("generated_at", ""), reverse=True) - return JSONResponse(results[:limit]) + total = len(results) + page = results[offset:offset + limit] + return JSONResponse({ + "results": page, + "total": total, + "offset": offset, + "limit": limit, + "has_more": (offset + limit) < total, + }) def _infer_asset(strategy_name: str, filename: str) -> str: @@ -622,20 +675,32 @@ def _infer_asset(strategy_name: str, filename: str) -> str: @app.get("/api/vbt/result/{filename}") async def get_vbt_result(filename: str): - """Get full VBT backtest result including equity curve.""" - for d in [BACKTEST_DIR, HISTORICAL_DIR]: - fpath = os.path.join(d, filename) - if os.path.exists(fpath): - with open(fpath) as f: - data = json.load(f) - data = _normalize_vbt_fields(data) - # Ensure equity curve is compact for transport - ec = data.get("equity_curve", []) - if ec and len(ec) > 500: - step = len(ec) // 500 - data["equity_curve"] = ec[::step] - return JSONResponse(data) - return JSONResponse({"error": "not found"}, status_code=404) + """Get full VBT backtest result including equity curve with LTTB downsampling.""" + _refresh_vbt_cache() + if filename in _vbt_full_cache: + data = dict(_vbt_full_cache[filename]) + else: + for d in [BACKTEST_DIR, HISTORICAL_DIR]: + fpath = os.path.join(d, filename) + if os.path.exists(fpath): + with open(fpath) as f: + data = json.load(f) + data = _normalize_vbt_fields(data) + break + else: + return JSONResponse({"error": "not found"}, status_code=404) + + ec = data.get("equity_curve", []) + if ec and len(ec) > 500: + data["equity_curve"] = _lttb_downsample(ec, 500) + if ec: + values = [p["v"] for p in data["equity_curve"]] + peak = values[0] if values else 0 + for i, v in enumerate(values): + peak = max(peak, v) + values[i] = round(-((peak - v) / peak * 100) if peak > 0 else 0, 2) + data["drawdown_curve"] = values + return JSONResponse(data) @app.get("/api/vbt/run") @@ -650,7 +715,6 @@ async def run_vbt_backtest( try: from backtests.vbt_runner import VBTBacktestRunner runner = VBTBacktestRunner() - from datetime import datetime ts = datetime.now().strftime("%Y%m%d-%H%M%S") coin_suffix = f"_{coin}" if coin else "" result = runner.run_strategy( @@ -664,6 +728,7 @@ async def run_vbt_backtest( with open(fpath, "w") as f: json.dump(result, f, default=str) result["filename"] = fname + _vbt_cache_time = 0.0 return JSONResponse(result) return JSONResponse({"error": "no results generated"}, status_code=500) except Exception as e: @@ -700,11 +765,41 @@ async def list_vbt_strategies(): {"key": "grid_mm", "name": "Grid Market Making", "coins": ["BTC"]}, {"key": "composite_mm", "name": "Composite MM", "coins": ["BTC"]}, {"key": "iceberg", "name": "Iceberg Detection", "coins": ["BTC"]}, - {"key": "momentum", "name": "Momentum Breakout", "coins": ["ETH"]}, - {"key": "mean_rev", "name": "Mean Reversion", "coins": ["ETH"]}, + {"key": "momentum", "name": "Momentum Breakout", "coins": ["BTC", "ETH"]}, + {"key": "mean_rev", "name": "Mean Reversion", "coins": ["BTC", "ETH"]}, ]) +@app.get("/api/vbt/result/{filename}/csv") +async def get_vbt_csv(filename: str): + """Download VBT backtest trades as CSV.""" + from fastapi.responses import Response + _refresh_vbt_cache() + if filename in _vbt_full_cache: + data = _vbt_full_cache[filename] + else: + for d in [BACKTEST_DIR, HISTORICAL_DIR]: + fpath = os.path.join(d, filename) + if os.path.exists(fpath): + with open(fpath) as f: + data = json.load(f) + data = _normalize_vbt_fields(data) + break + else: + return JSONResponse({"error": "not found"}, status_code=404) + trades = data.get("trades", []) + header = "time,side,size,entry_px,exit_px,pnl,return_pct,duration\n" + rows = [] + for t in trades: + rows.append(f"{t.get('time','')},{t.get('side','')},{t.get('size','')},{t.get('entry_px','')},{t.get('exit_px','')},{t.get('pnl','')},{t.get('return_pct','')},{t.get('duration','')}") + csv_content = header + "\n".join(rows) + return Response( + content=csv_content, + media_type="text/csv", + headers={"Content-Disposition": f"attachment; filename={filename}_trades.csv"} + ) + + # ═══════════════════════════════════════════════════════════ # Static # ═══════════════════════════════════════════════════════════ @@ -802,6 +897,7 @@ def main(): print(f" http://{args.host}:{args.port}") print(f" WebSocket: ws://{args.host}:{args.port}/ws") print(f" Backtests: /api/backtests") + print(f" VBT Dashboard: /vbt") uvicorn.run(app, host=args.host, port=args.port, log_level="warning") diff --git a/dashboard/static/vbt.html b/dashboard/static/vbt.html index f7b0b25..c1a1b19 100644 --- a/dashboard/static/vbt.html +++ b/dashboard/static/vbt.html @@ -8,41 +8,48 @@

FTDT QUANT LAB — VBT

-
● LIVE · Hyperliquid Mainnet
+
Hyperliquid · VectorBT
- @@ -82,8 +93,10 @@ select:focus{outline:none;border-color:#555} + + - @@ -91,71 +104,130 @@ select:focus{outline:none;border-color:#555}
- - + - +
- - - +
- +
+ + + +
+
+ + +
+
Loading...
+

SELECT A BACKTEST

-

Choose from sidebar or configure and run a new test

+

Choose from the sidebar or configure and run a new test

- + \ No newline at end of file diff --git a/dashboard/vbt_server.py b/dashboard/vbt_server.py deleted file mode 100644 index 734da6a..0000000 --- a/dashboard/vbt_server.py +++ /dev/null @@ -1,204 +0,0 @@ -""" -Minimal VBT dashboard server — no live trading, no memory guard, no broadcast. -Just serves the VBT dashboard HTML and backtest API endpoints. -""" -import json, os, sys -from pathlib import Path -from datetime import datetime - -project_root = str(Path(__file__).resolve().parent.parent) -sys.path.insert(0, project_root) - -from fastapi import FastAPI -from fastapi.staticfiles import StaticFiles -from fastapi.responses import FileResponse, JSONResponse - -RESULTS_DIR = Path(project_root) / "backtests" / "results" -HISTORICAL_DIR = RESULTS_DIR / "historical" -STATIC_DIR = Path(project_root) / "dashboard" / "static" -os.makedirs(RESULTS_DIR, exist_ok=True) - -app = FastAPI(title="FTDT VBT Dashboard") - - -# ── Field normalization ────────────────────────────────────────── - -def _normalize(data: dict) -> dict: - out = dict(data) - if "total_return_pct" not in out: - out["total_return_pct"] = out.get("pnl_pct", out.get("ann_return_pct", 0)) - if out.get("total_return_pct") is None: - out["total_return_pct"] = 0 - if "max_drawdown_pct" not in out: - dd = out.get("max_dd_pct", out.get("max_dd")) - if dd is not None and isinstance(dd, (int, float)) and abs(dd) < 1: - dd = dd * 100 - out["max_drawdown_pct"] = dd or 0 - if out.get("max_drawdown_pct") is None: - out["max_drawdown_pct"] = 0 - if "n_bars" not in out: - out["n_bars"] = out.get("num_periods", 0) - if out.get("n_bars") is None: - out["n_bars"] = 0 - if "profit_factor" not in out: - trades = out.get("trades", []) - if trades: - gross_win = sum(t.get("pnl", t.get("pnl_net", t.get("pnl_gross", 0))) or 0 - for t in trades if (t.get("pnl", t.get("pnl_net", t.get("pnl_gross", 0))) or 0) > 0) - gross_loss = abs(sum(t.get("pnl", t.get("pnl_net", t.get("pnl_gross", 0))) or 0 - for t in trades if (t.get("pnl", t.get("pnl_net", t.get("pnl_gross", 0))) or 0) < 0)) - out["profit_factor"] = round(gross_win / gross_loss, 3) if gross_loss > 0 else 0 - else: - out["profit_factor"] = 0 - if "total_trades" not in out: - out["total_trades"] = len(out.get("trades", [])) - if out.get("total_trades") is None: - out["total_trades"] = 0 - if not out.get("win_rate") and "trades" in out: - trades = out.get("trades", []) - if trades: - wins = sum(1 for t in trades if (t.get("pnl", t.get("pnl_net", t.get("pnl_gross", 0))) or 0) > 0) - out["win_rate"] = round(wins / len(trades), 3) - return out - - -def _infer_asset(strategy_name: str, filename: str) -> str: - name = (strategy_name + " " + filename).lower() - for key, asset in { - "pairs": "BTC/ETH", "order book": "BTC", "obi": "BTC", - "iceberg": "BTC", "hurst": "BTC", "vpin": "BTC", - "avellaneda": "BTC", "as_mm": "BTC", "grid": "BTC", - "composite": "BTC", "funding": "BTC", "kalman": "BTC/ETH", - "cartea": "BTC", "gueant": "BTC", "hawkes": "BTC", - "deep lob": "BTC", "queue": "BTC", - }.items(): - if key in name: - return asset - return "BTC" if "btc" in name or "eth" not in name else "ETH" - - -# ── REST API ──────────────────────────────────────────────────── - -@app.get("/api/vbt/results") -async def list_results(strategy: str = "", interval: str = "", sort: str = "date", limit: int = 200): - results = [] - for d in [RESULTS_DIR, HISTORICAL_DIR]: - if not os.path.isdir(d): - continue - for fname in sorted(os.listdir(d), reverse=True): - if not fname.endswith(".json"): - continue - if strategy and strategy not in fname: - continue - try: - with open(os.path.join(d, fname)) as f: - n = _normalize(json.load(f)) - if interval and n.get("interval", "1h") != interval: - continue - results.append({ - "filename": fname, - "strategy": n.get("strategy", "unknown"), - "asset": _infer_asset(n.get("strategy", ""), fname), - "engine": n.get("engine", "vectorbt"), - "interval": n.get("interval", "1h"), - "sharpe": n.get("sharpe", 0), - "sortino": n.get("sortino", 0), - "total_return_pct": n["total_return_pct"], - "max_drawdown_pct": n["max_drawdown_pct"], - "win_rate": n.get("win_rate", 0), - "profit_factor": n["profit_factor"], - "total_trades": n["total_trades"], - "n_bars": n["n_bars"], - "generated_at": n.get("generated_at", ""), - "has_equity_curve": bool(n.get("equity_curve")), - }) - except (json.JSONDecodeError, IOError): - pass - if len(results) >= limit: - break - - sort_keys = { - "sharpe": ("sharpe", True), "return": ("total_return_pct", True), - "dd": ("max_drawdown_pct", False), "trades": ("total_trades", True), - } - if sort in sort_keys: - key, rev = sort_keys[sort] - results.sort(key=lambda r: r.get(key, -999 if rev else 999), reverse=rev) - else: - results.sort(key=lambda r: r.get("generated_at", ""), reverse=True) - return JSONResponse(results[:limit]) - - -@app.get("/api/vbt/result/{filename}") -async def get_result(filename: str): - for d in [RESULTS_DIR, HISTORICAL_DIR]: - fpath = os.path.join(d, filename) - if os.path.exists(fpath): - with open(fpath) as f: - data = _normalize(json.load(f)) - ec = data.get("equity_curve", []) - if ec and len(ec) > 500: - data["equity_curve"] = ec[::len(ec)//500] - return JSONResponse(data) - return JSONResponse({"error": "not found"}, status_code=404) - - -@app.get("/api/vbt/run") -async def run_backtest(strategy: str = "pairs", interval: str = "1h", limit: int = 500, coin: str = ""): - try: - from backtests.vbt_runner import VBTBacktestRunner - runner = VBTBacktestRunner() - ts = datetime.now().strftime("%Y%m%d-%H%M%S") - result = runner.run_strategy(strategy=strategy, interval=interval, limit=limit) - if result: - if coin: - result["asset"] = coin.upper() - fname = f"{strategy}_{'' if not coin else coin+'_'}vbt_{ts}.json" - fpath = RESULTS_DIR / fname - with open(fpath, "w") as f: - json.dump(result, f, default=str) - result["filename"] = fname - return JSONResponse(result) - return JSONResponse({"error": "no results"}, status_code=500) - except Exception as e: - return JSONResponse({"error": str(e)}, status_code=500) - - -@app.get("/api/vbt/strategies") -async def list_strategies(): - return JSONResponse([ - {"key": "pairs", "name": "Pairs Trading", "coins": ["BTC", "ETH"]}, - {"key": "hurst_vpin", "name": "Hurst VPIN", "coins": ["BTC"]}, - {"key": "as_mm", "name": "Avellaneda-Stoikov", "coins": ["BTC"]}, - {"key": "obi", "name": "Order Book Imbalance", "coins": ["BTC"]}, - {"key": "grid_mm", "name": "Grid Market Making", "coins": ["BTC"]}, - {"key": "composite_mm", "name": "Composite MM", "coins": ["BTC"]}, - {"key": "iceberg", "name": "Iceberg Detection", "coins": ["BTC"]}, - ]) - - -# ── Static ────────────────────────────────────────────────────── - -@app.get("/vbt") -async def vbt_page(): - return FileResponse(STATIC_DIR / "vbt.html") - - -@app.get("/") -async def root(): - return FileResponse(STATIC_DIR / "vbt.html") - - -app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static") - - -# ── Main ──────────────────────────────────────────────────────── - -if __name__ == "__main__": - import uvicorn, argparse - p = argparse.ArgumentParser() - p.add_argument("--port", type=int, default=9175) - p.add_argument("--host", default="0.0.0.0") - args = p.parse_args() - print(f"VBT Dashboard → http://{args.host}:{args.port}/vbt") - uvicorn.run(app, host=args.host, port=args.port, log_level="error")