ramseshk
9d817ac2fa
Backend (vbt_runner.py):
- Add asset (BTC/ETH) to each trade record
- Compute per-trade fee using HL taker rate (0.05%)
entry_fee = size * entry_px * fee_rate
exit_fee = size * exit_px * fee_rate
- Add pnl_gross (before fees) and pnl_net (after fees)
- Add fee_rate field for transparency
Server (server.py):
- Normalize old backtest trades: add missing asset, fee,
pnl_net, pnl_gross fields
- Holyliquid default fee rate: 0.05% taker
Frontend (vbt.html):
- Trade log table now shows:
Time | Side + Asset | Size | Entry | Exit | Fee | PnL (net) | Duration
- Asset shown as inline badge in Side column
- Fee column with explicit USD amount
- PnL now explicitly labeled 'net' (after fees)
- Fallback to old 'pnl' field for legacy backtest files
1009 lines
39 KiB
Python
1009 lines
39 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))
|
|
|
|
# Fix BACKTEST_DIR — auto-detect local path if deployed dir doesn't exist
|
|
_default_results = str(Path(__file__).resolve().parent.parent / "backtests" / "results")
|
|
BACKTEST_DIR = _default_results if os.path.isdir(_default_results) else "/home/debian/ftdt-quant-lab/backtests/results"
|
|
HISTORICAL_DIR = BACKTEST_DIR + "/historical" if os.path.isdir(BACKTEST_DIR + "/historical") else BACKTEST_DIR
|
|
|
|
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 = 512 * 1024 * 1024 # 512 MB — force GC
|
|
MEM_WARN_LIMIT = 768 * 1024 * 1024 # 768 MB — log warning
|
|
MEM_HARD_LIMIT = 2048 * 1024 * 1024 # 2 GB — 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"
|
|
STATIC_DIR = Path(__file__).parent / "static"
|
|
|
|
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/v2")
|
|
async def get_metrics_v2():
|
|
"""Read the v2 node metrics file."""
|
|
try:
|
|
v2_file = "/tmp/ftdt-metrics-v2.json"
|
|
if os.path.exists(v2_file):
|
|
with open(v2_file) as f:
|
|
return JSONResponse(json.load(f))
|
|
except (json.JSONDecodeError, IOError):
|
|
pass
|
|
return JSONResponse({"status": "no_data"})
|
|
|
|
@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,
|
|
})
|
|
|
|
# ═══════════════════════════════════════════════════════════
|
|
# VBT Dashboard API — VectorBT backtest results browser
|
|
# ═══════════════════════════════════════════════════════════
|
|
|
|
_vbt_meta_cache: dict[str, dict] = {} # filename → parsed summary dict
|
|
_vbt_full_cache: dict[str, dict] = {} # filename → full result dict
|
|
_vbt_cache_time: float = 0.0 # epoch of last cache rebuild
|
|
_VBT_CACHE_TTL = 5.0 # seconds before re-scan
|
|
|
|
def _refresh_vbt_cache():
|
|
"""Scan results dirs once and populate caches."""
|
|
global _vbt_meta_cache, _vbt_full_cache, _vbt_cache_time
|
|
now = time.time()
|
|
if now - _vbt_cache_time < _VBT_CACHE_TTL:
|
|
return
|
|
new_meta: dict[str, dict] = {}
|
|
new_full: dict[str, dict] = {}
|
|
for d in [BACKTEST_DIR, HISTORICAL_DIR]:
|
|
if not os.path.isdir(d):
|
|
continue
|
|
for fname in sorted(os.listdir(d)):
|
|
if not fname.endswith(".json"):
|
|
continue
|
|
if fname in new_meta:
|
|
continue
|
|
fpath = os.path.join(d, fname)
|
|
try:
|
|
with open(fpath) as f:
|
|
data = json.load(f)
|
|
n = _normalize_vbt_fields(data)
|
|
asset = _infer_asset(n.get("strategy", ""), fname)
|
|
new_meta[fname] = {
|
|
"filename": fname,
|
|
"strategy": n.get("strategy", "unknown"),
|
|
"asset": asset,
|
|
"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"],
|
|
"calmar": round((n["total_return_pct"] / max(abs(n["max_drawdown_pct"]), 0.01)), 2),
|
|
"win_rate": n.get("win_rate", 0),
|
|
"profit_factor": n["profit_factor"],
|
|
"expectancy": n.get("expectancy", 0),
|
|
"total_trades": n["total_trades"],
|
|
"n_bars": n["n_bars"],
|
|
"generated_at": n.get("generated_at", ""),
|
|
"has_equity_curve": bool(n.get("equity_curve")),
|
|
}
|
|
new_full[fname] = n
|
|
except (json.JSONDecodeError, IOError):
|
|
pass
|
|
_vbt_meta_cache = new_meta
|
|
_vbt_full_cache = new_full
|
|
_vbt_cache_time = now
|
|
|
|
|
|
def _normalize_vbt_fields(data: dict) -> dict:
|
|
"""Normalise old/new backtest file field names to a consistent schema."""
|
|
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 and "trades" in out:
|
|
trades = out.get("trades", [])
|
|
if trades:
|
|
gross_win = sum(
|
|
t.get("pnl_net", t.get("pnl_gross", t.get("pnl", 0)))
|
|
for t in trades if (t.get("pnl_net", t.get("pnl_gross", t.get("pnl", 0))) or 0) > 0
|
|
)
|
|
gross_loss = abs(sum(
|
|
t.get("pnl_net", t.get("pnl_gross", t.get("pnl", 0)))
|
|
for t in trades if (t.get("pnl_net", t.get("pnl_gross", t.get("pnl", 0))) or 0) < 0
|
|
))
|
|
out["profit_factor"] = round(gross_win / gross_loss, 3) if gross_loss > 0 else 0
|
|
if "profit_factor" not in out:
|
|
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_net", t.get("pnl_gross", t.get("pnl", 0))) or 0) > 0)
|
|
out["win_rate"] = round(wins / len(trades), 3)
|
|
|
|
# Normalize trade records: add asset, fee, pnl_net/pnl_gross for old files
|
|
trades = out.get("trades", [])
|
|
if trades:
|
|
asset = _infer_asset(out.get("strategy", ""), "")
|
|
hl_fee = 0.0005 # default Hyperliquid taker rate
|
|
for t in trades:
|
|
if not t.get("asset"):
|
|
t["asset"] = asset
|
|
if not t.get("fee"):
|
|
entry_px = float(t.get("entry_px", 0))
|
|
exit_px = float(t.get("exit_px", 0))
|
|
size = float(t.get("size", 0))
|
|
notional_entry = size * entry_px
|
|
notional_exit = size * exit_px
|
|
t["fee"] = round((notional_entry + notional_exit) * hl_fee, 6)
|
|
t["fee_rate"] = f"{hl_fee*100:.3f}%"
|
|
if not t.get("pnl_net"):
|
|
t["pnl_net"] = t.get("pnl", 0)
|
|
if not t.get("pnl_gross"):
|
|
pnl_net = float(t.get("pnl_net", t.get("pnl", 0)))
|
|
fee = float(t.get("fee", 0))
|
|
t["pnl_gross"] = round(pnl_net + fee, 4)
|
|
|
|
return out
|
|
|
|
|
|
def _lttb_downsample(points: list[dict], target: int) -> list[dict]:
|
|
"""Largest-Triangle-Three-Buckets downsampling for visual fidelity."""
|
|
n = len(points)
|
|
if n <= target or target < 3:
|
|
return points
|
|
bucket_size = (n - 2) / (target - 2)
|
|
result = [points[0]]
|
|
a = 0 # index of last point in result (within original points array)
|
|
for i in range(target - 2):
|
|
avg_start = int((i + 0) * bucket_size) + 1
|
|
avg_end = int((i + 1) * bucket_size) + 1
|
|
avg_range = points[avg_start:avg_end]
|
|
avg_y = sum(p["v"] for p in avg_range) / max(len(avg_range), 1)
|
|
avg_x = int((avg_start + avg_end - 1) / 2)
|
|
range_offs = int((i + 1) * bucket_size) + 1
|
|
range_to = int((i + 2) * bucket_size) + 1
|
|
max_area = -1.0
|
|
next_pt = range_offs
|
|
for j in range(range_offs, min(range_to + 1, n)):
|
|
pa = points[a]
|
|
pc = points[j]
|
|
area = abs((a - j) * (pc["v"] - avg_y) +
|
|
(j - avg_x) * (avg_y - pa["v"]) +
|
|
(avg_x - a) * (pa["v"] - pc["v"])) * 0.5
|
|
if area > max_area:
|
|
max_area = area
|
|
next_pt = j
|
|
result.append(points[next_pt])
|
|
a = next_pt
|
|
result.append(points[-1])
|
|
return result
|
|
|
|
|
|
@app.get("/api/vbt/results")
|
|
async def list_vbt_results(
|
|
strategy: str = "",
|
|
interval: str = "",
|
|
asset: str = "",
|
|
sort: str = "date",
|
|
limit: int = 100,
|
|
offset: int = 0,
|
|
):
|
|
"""List VectorBT backtest results with full metrics, pagination, and server-side filtering."""
|
|
_refresh_vbt_cache()
|
|
results = list(_vbt_meta_cache.values())
|
|
|
|
if strategy:
|
|
results = [r for r in results if strategy in r.get("filename", "")]
|
|
if interval:
|
|
results = [r for r in results if r.get("interval") == interval]
|
|
if asset:
|
|
results = [r for r in results if r.get("asset", "") == asset or r.get("asset", "").endswith("/" + asset)]
|
|
|
|
sort_keys = {
|
|
"sharpe": ("sharpe", True),
|
|
"return": ("total_return_pct", True),
|
|
"dd": ("max_drawdown_pct", False),
|
|
"trades": ("total_trades", True),
|
|
"calmar": ("calmar", 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)
|
|
|
|
total = len(results)
|
|
page = results[offset:offset + limit]
|
|
return JSONResponse({
|
|
"results": page,
|
|
"total": total,
|
|
"offset": offset,
|
|
"limit": limit,
|
|
"has_more": (offset + limit) < total,
|
|
})
|
|
|
|
|
|
def _infer_asset(strategy_name: str, filename: str) -> str:
|
|
"""Infer the trading asset from strategy name or filename."""
|
|
name = (strategy_name + " " + filename).lower()
|
|
coin_map = {
|
|
"pairs": "BTC/ETH",
|
|
"order book": "BTC",
|
|
"obi": "BTC",
|
|
"iceberg": "BTC",
|
|
"momentum": "ETH" if "eth" in name else "BTC",
|
|
"mean rev": "ETH" if "eth" in name else "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",
|
|
}
|
|
for key, asset in coin_map.items():
|
|
if key in name:
|
|
return asset
|
|
return "BTC"
|
|
|
|
|
|
@app.get("/api/vbt/result/{filename}")
|
|
async def get_vbt_result(filename: str):
|
|
"""Get full VBT backtest result including equity curve with LTTB downsampling."""
|
|
_refresh_vbt_cache()
|
|
if filename in _vbt_full_cache:
|
|
data = dict(_vbt_full_cache[filename])
|
|
else:
|
|
for d in [BACKTEST_DIR, HISTORICAL_DIR]:
|
|
fpath = os.path.join(d, filename)
|
|
if os.path.exists(fpath):
|
|
with open(fpath) as f:
|
|
data = json.load(f)
|
|
data = _normalize_vbt_fields(data)
|
|
break
|
|
else:
|
|
return JSONResponse({"error": "not found"}, status_code=404)
|
|
|
|
ec = data.get("equity_curve", [])
|
|
if ec and len(ec) > 500:
|
|
data["equity_curve"] = _lttb_downsample(ec, 500)
|
|
if ec:
|
|
values = [p["v"] for p in data["equity_curve"]]
|
|
peak = values[0] if values else 0
|
|
for i, v in enumerate(values):
|
|
peak = max(peak, v)
|
|
values[i] = round(-((peak - v) / peak * 100) if peak > 0 else 0, 2)
|
|
data["drawdown_curve"] = values
|
|
return JSONResponse(data)
|
|
|
|
|
|
@app.get("/api/vbt/run")
|
|
async def run_vbt_backtest(
|
|
strategy: str = "pairs",
|
|
interval: str = "1h",
|
|
limit: int = 500,
|
|
coin: str = "",
|
|
testnet: bool = False,
|
|
):
|
|
"""Run a new VectorBT backtest and return results."""
|
|
try:
|
|
from backtests.vbt_runner import VBTBacktestRunner
|
|
runner = VBTBacktestRunner()
|
|
ts = datetime.now().strftime("%Y%m%d-%H%M%S")
|
|
coin_suffix = f"_{coin}" if coin else ""
|
|
result = runner.run_strategy(
|
|
strategy=strategy, interval=interval, testnet=testnet, limit=limit
|
|
)
|
|
if result:
|
|
if coin:
|
|
result["asset"] = coin.upper()
|
|
fname = f"{strategy}{coin_suffix}_vbt_{ts}.json"
|
|
fpath = os.path.join(BACKTEST_DIR, fname)
|
|
with open(fpath, "w") as f:
|
|
json.dump(result, f, default=str)
|
|
result["filename"] = fname
|
|
_vbt_cache_time = 0.0
|
|
return JSONResponse(result)
|
|
return JSONResponse({"error": "no results generated"}, status_code=500)
|
|
except Exception as e:
|
|
return JSONResponse({"error": str(e)}, status_code=500)
|
|
|
|
|
|
@app.get("/api/vbt/sweep")
|
|
async def run_vbt_sweep(strategy: str = "pairs"):
|
|
"""Run parameter sweep and return heatmap data."""
|
|
try:
|
|
from backtests.vbt_runner import VBTBacktestRunner
|
|
runner = VBTBacktestRunner()
|
|
df = runner.param_sweep(strategy=strategy)
|
|
if df is not None and not df.empty:
|
|
rows = df.to_dict(orient="records")
|
|
return JSONResponse({
|
|
"strategy": strategy,
|
|
"results": rows,
|
|
"best": max(rows, key=lambda r: r.get("sharpe", -999)),
|
|
})
|
|
return JSONResponse({"error": "no sweep results"}, status_code=500)
|
|
except Exception as e:
|
|
return JSONResponse({"error": str(e)}, status_code=500)
|
|
|
|
|
|
@app.get("/api/vbt/strategies")
|
|
async def list_vbt_strategies():
|
|
"""List available strategies for VBT backtesting."""
|
|
return JSONResponse([
|
|
{"key": "pairs", "name": "Pairs Trading", "coins": ["BTC", "ETH"]},
|
|
{"key": "hurst_vpin", "name": "Hurst VPIN", "coins": ["BTC"]},
|
|
{"key": "as_mm", "name": "Avellaneda-Stoikov MM", "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"]},
|
|
{"key": "momentum", "name": "Momentum Breakout", "coins": ["BTC", "ETH"]},
|
|
{"key": "mean_rev", "name": "Mean Reversion", "coins": ["BTC", "ETH"]},
|
|
])
|
|
|
|
|
|
@app.get("/api/vbt/result/{filename}/csv")
|
|
async def get_vbt_csv(filename: str):
|
|
"""Download VBT backtest trades as CSV."""
|
|
from fastapi.responses import Response
|
|
_refresh_vbt_cache()
|
|
if filename in _vbt_full_cache:
|
|
data = _vbt_full_cache[filename]
|
|
else:
|
|
for d in [BACKTEST_DIR, HISTORICAL_DIR]:
|
|
fpath = os.path.join(d, filename)
|
|
if os.path.exists(fpath):
|
|
with open(fpath) as f:
|
|
data = json.load(f)
|
|
data = _normalize_vbt_fields(data)
|
|
break
|
|
else:
|
|
return JSONResponse({"error": "not found"}, status_code=404)
|
|
trades = data.get("trades", [])
|
|
header = "time,side,size,entry_px,exit_px,pnl,return_pct,duration\n"
|
|
rows = []
|
|
for t in trades:
|
|
rows.append(f"{t.get('time','')},{t.get('side','')},{t.get('size','')},{t.get('entry_px','')},{t.get('exit_px','')},{t.get('pnl','')},{t.get('return_pct','')},{t.get('duration','')}")
|
|
csv_content = header + "\n".join(rows)
|
|
return Response(
|
|
content=csv_content,
|
|
media_type="text/csv",
|
|
headers={"Content-Disposition": f"attachment; filename={filename}_trades.csv"}
|
|
)
|
|
|
|
|
|
# ═══════════════════════════════════════════════════════════
|
|
# Static
|
|
# ═══════════════════════════════════════════════════════════
|
|
|
|
@app.get("/")
|
|
async def root():
|
|
return FileResponse(STATIC_DIR / "index.html")
|
|
|
|
|
|
@app.get("/vbt")
|
|
async def vbt_dashboard():
|
|
return FileResponse(STATIC_DIR / "vbt.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")
|
|
print(f" VBT Dashboard: /vbt")
|
|
|
|
uvicorn.run(app, host=args.host, port=args.port, log_level="warning")
|
|
|
|
|
|
|
|
# ── Next.js Dashboard (served at /cv/) ──────────────────────
|
|
|
|
|
|
@app.get("/cv/api/backtests/historical")
|
|
async def cv_historical_backtests():
|
|
return await list_historical_backtests()
|
|
|
|
@app.get("/cv/api/backtest/historical/{name}")
|
|
async def cv_historical_detail(name: str):
|
|
return await get_historical_backtest(name)
|
|
|
|
@app.get("/cv/api/backtest/{name}")
|
|
async def cv_backtest_detail(name: str):
|
|
return await get_backtest(name)
|
|
|
|
@app.get("/cv/api/backtest/{name}/recalc")
|
|
async def cv_backtest_recalc(name: str, fee_tier: int = 0, staking_tier: str = "none"):
|
|
return await recalc_backtest(name, fee_tier, staking_tier)
|
|
|
|
@app.get("/cv/api/risk")
|
|
async def cv_risk():
|
|
return await get_risk_metrics()
|
|
|
|
@app.websocket("/cv/ws")
|
|
async def cv_websocket(websocket):
|
|
await websocket_endpoint(websocket)
|
|
|
|
@app.websocket("/cv/ws/paper")
|
|
async def cv_paper_websocket(websocket):
|
|
await paper_websocket_endpoint(websocket)
|
|
|
|
@app.get("/cv/api/metrics/v2")
|
|
async def cv_metrics_v2():
|
|
return await get_metrics_v2()
|
|
|
|
|
|
# Mount Next.js static assets only (not the whole /cv/ path)
|
|
_next_out = Path(__file__).parent.parent / "dashboard-next" / "out"
|
|
_next_static = _next_out / "_next" if _next_out.is_dir() else None
|
|
if _next_static and _next_static.is_dir():
|
|
app.mount("/cv/_next", StaticFiles(directory=str(_next_static)), name="cv_static")
|
|
|
|
# Serve Next.js HTML pages individually
|
|
_serve_next_static = _next_out.is_dir()
|
|
@app.get("/cv")
|
|
async def cv_index():
|
|
if _serve_next_static:
|
|
return FileResponse(_next_out / "index.html")
|
|
return FileResponse(STATIC_DIR / "index.html")
|
|
|
|
@app.get("/cv/{path:path}")
|
|
async def cv_catchall(path: str):
|
|
if _serve_next_static:
|
|
fpath = _next_out / path
|
|
if fpath.is_file():
|
|
return FileResponse(fpath)
|
|
if (fpath / "index.html").is_file():
|
|
return FileResponse(fpath / "index.html")
|
|
# Try .html extension
|
|
html_path = _next_out / f"{path}.html"
|
|
if html_path.is_file():
|
|
return FileResponse(html_path)
|
|
return FileResponse(STATIC_DIR / "index.html")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|