SELECT A BACKTEST
-
Choose from sidebar or configure params and run a new test
+
Choose from sidebar or configure and run a new test
diff --git a/dashboard/vbt_server.py b/dashboard/vbt_server.py
new file mode 100644
index 0000000..734da6a
--- /dev/null
+++ b/dashboard/vbt_server.py
@@ -0,0 +1,204 @@
+"""
+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")