""" Dashboard backend — WebSocket metrics server. Reads live metrics from a shared JSON file (written by the live node) and serves backtest results from disk. Streams everything to connected dashboard clients via WebSocket. Architecture: - /ws — WebSocket for real-time streaming - /backtests — list available backtest results - /backtest/{name} — serve specific backtest result - / — static HTML dashboard Usage: python dashboard/server.py --port 9175 """ import asyncio import json import os import time import threading from pathlib import Path from typing import Optional from fastapi import FastAPI, WebSocket, WebSocketDisconnect from fastapi.staticfiles import StaticFiles from fastapi.responses import FileResponse, JSONResponse import sys sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) from config.fee_tiers import get_perp_fees, PERPS_TIERS, STAKING_TIERS, STRATEGY_FEE_MODELS from common.risk import risk_summary from strategies.quant_report import compute_quant_report # ═══════════════════════════════════════════════════════════ # Memory guard: check RSS via /proc, force GC at 256MB, # log warning at 384MB, hard exit at 512MB. # RLIMIT_AS disabled — Python heap needs virtual headroom. # ═══════════════════════════════════════════════════════════ import gc, os as _os MEM_SOFT_LIMIT = 256 * 1024 * 1024 # 256 MB — force GC MEM_WARN_LIMIT = 384 * 1024 * 1024 # 384 MB — log warning MEM_HARD_LIMIT = 512 * 1024 * 1024 # 512 MB — terminate def check_memory(): """Check RSS, force GC if over soft limit, raise if over hard limit.""" try: with open("/proc/self/status") as f: for line in f: if line.startswith("VmRSS:"): rss_kb = int(line.split()[1]) rss = rss_kb * 1024 if rss > MEM_HARD_LIMIT: print(f"[CRIT] RSS {rss_kb // 1024}MB > 512MB — exiting", flush=True) _os._exit(1) if rss > MEM_SOFT_LIMIT: gc.collect() gc.collect() return except Exception: pass import uvicorn # ═══════════════════════════════════════════════════════════ # Constants # ═══════════════════════════════════════════════════════════ METRICS_FILE = "/tmp/ftdt-metrics.json" PAPER_METRICS_FILE = "/tmp/ftdt-paper-metrics.json" BACKTEST_DIR = "/home/debian/ftdt-quant-lab/backtests/results" HISTORICAL_DIR = "/home/debian/ftdt-quant-lab/backtests/results/historical" STATIC_DIR = Path(__file__).parent / "static" # Ensure backtest dir exists os.makedirs(BACKTEST_DIR, exist_ok=True) # ═══════════════════════════════════════════════════════════ # App # ═══════════════════════════════════════════════════════════ app = FastAPI(title="FTDT Quant Lab Dashboard") connected_clients: set[WebSocket] = set() paper_clients: set[WebSocket] = set() loop: Optional[asyncio.AbstractEventLoop] = None # ═══════════════════════════════════════════════════════════ # Metrics reader # ═══════════════════════════════════════════════════════════ def read_metrics() -> dict: """Read the shared metrics file written by the live node.""" try: if os.path.exists(METRICS_FILE): with open(METRICS_FILE) as f: return json.load(f) except (json.JSONDecodeError, IOError): pass return _empty_metrics() def _empty_metrics() -> dict: return { "timestamp": time.time(), "wallet": "0x...", "total_equity": 898.0, "total_pnl": 0.0, "total_pnl_pct": 0.0, "equity_history": [], "strategies": {}, "trades": [], "status": "starting", } def read_paper_metrics() -> dict: """Read paper trading metrics file.""" try: if os.path.exists(PAPER_METRICS_FILE): with open(PAPER_METRICS_FILE) as f: return json.load(f) except (json.JSONDecodeError, IOError): pass return {"status": "waiting", "mode": "paper", "strategies": {}, "trades": [], "equity_history": [], "total_pnl": 0, "total_equity": 100000} # ═══════════════════════════════════════════════════════════ # Background broadcaster # ═══════════════════════════════════════════════════════════ async def broadcast_to_client(ws: WebSocket, payload: str): try: await ws.send_text(payload) except Exception: connected_clients.discard(ws) def broadcast_loop(): """Continuously read metrics and broadcast to all clients.""" while True: time.sleep(1) check_memory() data = read_metrics() payload = json.dumps(data, default=str) for ws in list(connected_clients): if loop: asyncio.run_coroutine_threadsafe( broadcast_to_client(ws, payload), loop ) # Also broadcast paper metrics paper_data = read_paper_metrics() paper_payload = json.dumps(paper_data, default=str) for ws in list(paper_clients): if loop: asyncio.run_coroutine_threadsafe( broadcast_to_client(ws, paper_payload), loop ) # ═══════════════════════════════════════════════════════════ # WebSocket # ═══════════════════════════════════════════════════════════ @app.websocket("/ws") async def websocket_endpoint(websocket: WebSocket): await websocket.accept() connected_clients.add(websocket) try: # Send initial state immediately data = read_metrics() await websocket.send_text(json.dumps(data, default=str)) while True: await asyncio.sleep(30) except WebSocketDisconnect: connected_clients.discard(websocket) @app.websocket("/ws/paper") async def paper_websocket_endpoint(websocket: WebSocket): await websocket.accept() paper_clients.add(websocket) try: data = read_paper_metrics() await websocket.send_text(json.dumps(data, default=str)) while True: await asyncio.sleep(30) except WebSocketDisconnect: paper_clients.discard(websocket) # ═══════════════════════════════════════════════════════════ # ═══════════════════════════════════════════════════════════ # REST metrics endpoints — polled by Next.js dashboard # ═══════════════════════════════════════════════════════════ @app.get("/api/metrics") async def get_metrics_rest(): return JSONResponse(read_metrics()) @app.get("/api/metrics/paper") async def get_paper_metrics_rest(): return JSONResponse(read_paper_metrics()) # Backtest endpoints # ═══════════════════════════════════════════════════════════ @app.get("/api/backtests") async def list_backtests(): """List all saved backtest results.""" results = [] if os.path.isdir(BACKTEST_DIR): for fname in sorted(os.listdir(BACKTEST_DIR), reverse=True): if fname.endswith(".json"): fpath = os.path.join(BACKTEST_DIR, fname) try: with open(fpath) as f: data = json.load(f) results.append({ "name": fname.replace(".json", ""), "strategy": data.get("strategy", "unknown"), "start": data.get("start_time"), "end": data.get("end_time"), "sharpe": data.get("sharpe", 0), "sortino": data.get("sortino", 0), "pnl_pct": data.get("pnl_pct", 0), "max_dd": data.get("max_dd", 0), "win_rate": data.get("win_rate", 0) or recalc_win_rate(data.get("trades", [])) or 0, "total_trades": data.get("total_trades", 0), }) except (json.JSONDecodeError, IOError): pass return JSONResponse(results) @app.get("/api/backtest/{name}") async def get_backtest(name: str): """Get full backtest result data — checks historical dir first.""" # Try historical subdirectory first (where dashboard saves backtests) fpath = os.path.join(HISTORICAL_DIR, f"{name}.json") if not os.path.exists(fpath): fpath = os.path.join(BACKTEST_DIR, f"{name}.json") if os.path.exists(fpath): with open(fpath) as f: return JSONResponse(json.load(f)) return JSONResponse({"error": "not found"}, status_code=404) def recalc_win_rate(trades): """Fallback win rate when stored value is 0.""" if not trades: return 0.0 wins = sum(1 for t in trades if (t.get("pnl_net") or t.get("pnl_gross") or t.get("pnl", 0)) > 0) return round(wins / len(trades), 4) if trades else 0.0 def recalc_equity_curve(equity_curve, trades, new_fee_rate, fee_model): """Rebuild equity curve with new fee rates, preserving gross PnL.""" if not equity_curve or not trades: return equity_curve fee_deltas = {} cum_delta = 0.0 for t in trades: sz = t.get("size", 0) px = t.get("price", 0) old_fee = t.get("fee", 0) new_fee = sz * px * new_fee_rate * 2 delta = old_fee - new_fee cum_delta += delta fee_deltas[t.get("time", "")] = cum_delta new_curve = [] delta_idx = 0 trade_times = list(fee_deltas.keys()) cum = 0.0 for pt in equity_curve: pt_time = pt.get("t", "") while delta_idx < len(trade_times) and trade_times[delta_idx] <= pt_time: cum = fee_deltas[trade_times[delta_idx]] delta_idx += 1 new_curve.append({"t": pt_time, "v": round(pt.get("v", 0) + cum, 6)}) return new_curve @app.get("/api/backtest/{name}/recalc") async def recalc_backtest(name: str, fee_tier: int = 0, staking_tier: str = "none"): """Recalculate backtest PnL with different fee tier.""" fpath = os.path.join(BACKTEST_DIR, f"{name}.json") if not os.path.exists(fpath): fpath = os.path.join(HISTORICAL_DIR, f"{name}.json") if not os.path.exists(fpath): return JSONResponse({"error": "not found"}, status_code=404) with open(fpath) as f: data = json.load(f) fee_model = STRATEGY_FEE_MODELS.get(data.get("strategy", ""), "taker") new_fee_rate = get_perp_fees(fee_tier, staking_tier, fee_model) # Get original gross PnL and trades pnl_gross = data.get("pnl_gross", data.get("pnl", 0)) trades = data.get("trades", []) # Recalculate fees with new rate new_fees = 0.0 new_trades = [] for t in trades: sz = t.get("size", 0) px = t.get("price", 0) orig_fee = t.get("fee", 0) new_fee = sz * px * new_fee_rate * 2 # entry + exit new_fees += new_fee new_trades.append({**t, "fee": round(new_fee, 6), "pnl_net": round(t.get("pnl_gross", t.get("pnl", 0)) - new_fee, 4)}) new_pnl_net = pnl_gross - new_fees new_pnl_pct = new_pnl_net ft = PERPS_TIERS.get(fee_tier, PERPS_TIERS[0]) st = STAKING_TIERS.get(staking_tier, STAKING_TIERS["none"]) eff_taker = get_perp_fees(fee_tier, staking_tier, "taker") eff_maker = get_perp_fees(fee_tier, staking_tier, "maker") return JSONResponse({ "strategy": data.get("strategy"), "fee_tier": ft["name"], "staking_tier": st["name"], "effective_taker_pct": round(eff_taker * 100, 4), "effective_maker_pct": round(eff_maker * 100, 4), "fee_model": fee_model, "pnl_gross": round(pnl_gross, 4), "pnl_gross_pct": round(pnl_gross, 4), "pnl_net": round(new_pnl_net, 4), "pnl_net_pct": round(new_pnl_pct, 4), "fees_total": round(new_fees, 4), "total_trades": len(new_trades), "equity_curve": recalc_equity_curve( data.get("equity_curve", []), data.get("trades", []), new_fee_rate, fee_model ), "trades": new_trades[-100:], "sharpe": data.get("sharpe", 0), "sortino": data.get("sortino", 0), "max_dd": data.get("max_dd", 0), "win_rate": data.get("win_rate", 0) or recalc_win_rate(data.get("trades", [])) or 0, "num_periods": data.get("num_periods", 720), }) @app.get("/api/backtests/historical") async def list_historical_backtests(): """List historical (real data) backtest results.""" results = [] d = HISTORICAL_DIR if os.path.isdir(d): for fname in sorted(os.listdir(d), reverse=True): if fname.endswith(".json"): fpath = os.path.join(d, fname) try: with open(fpath) as f: data = json.load(f) results.append({ "name": fname.replace(".json", ""), "strategy": data.get("strategy", "unknown"), "coin": data.get("coin", "?"), "start": data.get("start_time"), "end": data.get("end_time"), "sharpe": data.get("sharpe", 0), "sortino": data.get("sortino", 0), "pnl_pct": data.get("pnl_pct", 0), "max_dd": data.get("max_dd", 0), "win_rate": data.get("win_rate", 0) or recalc_win_rate(data.get("trades", [])) or 0, "total_trades": data.get("total_trades", 0), "data_source": "Hyperliquid Mainnet", }) except (json.JSONDecodeError, IOError): pass return JSONResponse(results) @app.get("/api/backtest/historical/{name}") async def get_historical_backtest(name: str): """Get full historical backtest result.""" fpath = os.path.join(HISTORICAL_DIR, f"{name}.json") if os.path.exists(fpath): with open(fpath) as f: return JSONResponse(json.load(f)) return JSONResponse({"error": "not found"}, status_code=404) @app.get("/api/backtest/{name}/csv") async def get_backtest_csv(name: str): """Download backtest trades as CSV.""" from fastapi.responses import Response fpath = os.path.join(BACKTEST_DIR, f"{name}.json") if not os.path.exists(fpath): return JSONResponse({"error": "not found"}, status_code=404) with open(fpath) as f: data = json.load(f) trades = data.get("trades", []) # Build CSV with headers header = "time,side,size,price,pnl_gross,pnl_net,fee\n" rows = [] for t in trades: rows.append(f"{t.get('time','')},{t.get('side','')},{t.get('size','')},{t.get('price','')},{t.get('pnl_gross',t.get('pnl',''))},{t.get('pnl_net',t.get('pnl',''))},{t.get('fee','0')}") csv_content = header + "\n".join(rows) return Response( content=csv_content, media_type="text/csv", headers={"Content-Disposition": f"attachment; filename={name}_trades.csv"} ) @app.get("/api/risk") async def get_risk_metrics(): """Compute risk analytics from the latest paper metrics.""" paper = read_paper_metrics() equity_history = paper.get("equity_history", []) strategy_equity = paper.get("strategy_equity", {}) if not equity_history: return JSONResponse({"error": "no equity history available"}, status_code=404) summary = risk_summary(equity_history, strategy_equity) # Build a compact correlation text summary for the frontend corr = summary.get("correlation", {}) corr_summary = [] names = sorted(corr.keys()) for i, n1 in enumerate(names): for n2 in names[i + 1:]: val = corr.get(n1, {}).get(n2, 0) if abs(val) > 0.3: # only show meaningful correlations corr_summary.append({ "pair": f"{n1} ↔ {n2}", "correlation": round(val, 3), "level": "high" if abs(val) > 0.7 else "medium", }) corr_summary.sort(key=lambda x: -abs(x["correlation"])) return JSONResponse({ "portfolio": { "var_95": summary["var_95"], "cvar_95": summary["cvar_95"], "max_drawdown": summary["max_drawdown"], "calmar_ratio": summary["calmar_ratio"], "sharpe": summary["sharpe"], "sortino": summary["sortino"], "num_observations": summary["num_observations"], }, "per_strategy": summary.get("per_strategy", {}), "correlation_summary": corr_summary, "correlation_matrix": corr, }) # ═══════════════════════════════════════════════════════════ # Static # ═══════════════════════════════════════════════════════════ @app.get("/") async def root(): return FileResponse(STATIC_DIR / "index.html") app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static") # ═══════════════════════════════════════════════════════════ # Main # ═══════════════════════════════════════════════════════════ @app.get("/api/quant-report/{name}") async def get_quant_report(name: str): """Compute full QF-Lib quant report from a backtest file. Accepts strategy name and auto-maps to filename prefix. """ # Strategy name → file prefix mapping NAME_MAP = { "order book imbalance": "ofi", "avellaneda-stoikov": "avellaneda", "funding rate arb": "funding_arb", "iceberg detection": "iceberg", "momentum breakout": "momentum", "mean reversion": "mean_rev", "kalman pairs": "kalman_pairs", "pairs trading": "pairs", } name_lower = name.lower() prefix = NAME_MAP.get(name_lower, name_lower.replace(" ", "_")) # Build candidate paths candidates = [] exact_path = os.path.join(BACKTEST_DIR, name) hist_exact = os.path.join(HISTORICAL_DIR, name) candidates.extend([exact_path, hist_exact]) # Try exact match for path in candidates: if os.path.exists(path): backtest_path = path break else: # Fuzzy match: find files starting with the mapped prefix fuzzy = [] for d in [BACKTEST_DIR, HISTORICAL_DIR]: if not os.path.exists(d): continue for f in os.listdir(d): f_clean = f.lower() # Match by prefix, then prefer BTC/ETH files if f_clean.startswith(f"{prefix}_"): fuzzy.append(os.path.join(d, f)) if fuzzy: backtest_path = fuzzy[0] else: return JSONResponse({"error": f"Backtest '{name}' not found"}, status_code=404) try: with open(backtest_path) as f: data = json.load(f) trades = data.get("trades", data.get("trade_history", [])) strategy_name = data.get("name", data.get("strategy", name)) strategy_id = data.get("id", name) report = compute_quant_report(strategy_name, strategy_id, trades, 100.0) return JSONResponse(report) except Exception as e: return JSONResponse({"error": str(e)}, status_code=500) def main(): import argparse parser = argparse.ArgumentParser() parser.add_argument("--port", type=int, default=9175) parser.add_argument("--host", default="0.0.0.0") args = parser.parse_args() global loop loop = asyncio.new_event_loop() asyncio.set_event_loop(loop) # Start metrics broadcaster broadcaster = threading.Thread(target=broadcast_loop, daemon=True) broadcaster.start() print(f"FTDT Quant Lab Dashboard") print(f" http://{args.host}:{args.port}") print(f" WebSocket: ws://{args.host}:{args.port}/ws") print(f" Backtests: /api/backtests") uvicorn.run(app, host=args.host, port=args.port, log_level="warning") # Serve Next.js assets at /_next/static/ _next_dir = Path(__file__).parent / "static" / "_next" if _next_dir.is_dir(): app.mount("/_next", StaticFiles(directory=str(_next_dir)), name="next_assets") if __name__ == "__main__": main()