ramseshk
392bde44a0
Bug: /api/backtest/{name} only looked in backtests/results/,
but all historical backtests are saved in backtests/results/historical/.
Fix: check HISTORICAL_DIR first, then fall back to BACKTEST_DIR.
This fixes SPX backtest detail showing zero prices/fees.
562 lines
22 KiB
Python
562 lines
22 KiB
Python
"""
|
|
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()
|