ftdt-quant-lab/dashboard/vbt_server.py

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