""" 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")