QF-Lib Quant Report: full strategy performance analytics
Backend: strategies/quant_report.py
- equityCurve: daily PnL from trade history
- monthlyReturns: heatmap matrix (years x months)
- yearlyReturns: bar chart data with mean
- monthlyReturnDistribution: histogram bins
- qqPlot: theoretical vs observed quantiles
- rollingStats: 6-month rolling return + volatility
API: /api/quant-report/{name}
Computes full report from any backtest JSON file
Frontend: QuantReport.tsx
- Strategy Performance chart (equity curve, blue line)
- Monthly Returns heatmap (blue saturation)
- Yearly Returns bar chart with mean line
- Distribution histogram
- Normal QQ plot with diagonal reference
- Rolling Statistics (6-month, dual line)
- QF-Lib header with logo and metadata
- Access via QF-Lib Report button in detail view
This commit is contained in:
@@ -0,0 +1,248 @@
|
||||
"""
|
||||
QF-Lib Quant Analytics — computes full strategy performance report.
|
||||
|
||||
Produces JSON with:
|
||||
- equityCurve: daily equity from trade history
|
||||
- monthlyReturns: heatmap matrix (years × months)
|
||||
- yearlyReturns: bar chart data with mean
|
||||
- monthlyReturnDistribution: histogram bins
|
||||
- qqPlot: theoretical vs observed quantiles
|
||||
- rollingStats: 6-month rolling return + volatility
|
||||
"""
|
||||
|
||||
import json, math
|
||||
from datetime import datetime, timedelta
|
||||
from collections import defaultdict, OrderedDict
|
||||
from typing import Optional
|
||||
|
||||
MONTHS = ["Jan", "Feb", "Mar", "Apr", "May", "Jun",
|
||||
"Jul", "Aug", "Sep", "Oct", "Nov", "Dec"]
|
||||
|
||||
def compute_daily_equity(trades: list[dict], start_equity: float = 100.0) -> list[dict]:
|
||||
"""Build daily equity curve from trade PnL history."""
|
||||
daily = defaultdict(float)
|
||||
for t in trades:
|
||||
try:
|
||||
ts = t.get("time", "")
|
||||
if "T" in ts:
|
||||
date = ts[:10]
|
||||
elif " " in ts:
|
||||
date = ts.split(" ")[0]
|
||||
elif len(ts) >= 10:
|
||||
date = ts[:10]
|
||||
else:
|
||||
continue
|
||||
pnl = float(t.get("pnl", 0))
|
||||
daily[date] += pnl
|
||||
except (ValueError, KeyError):
|
||||
continue
|
||||
|
||||
dates = sorted(daily.keys())
|
||||
if not dates:
|
||||
return [{"date": "2024-01-01", "value": start_equity}]
|
||||
|
||||
equity = start_equity
|
||||
curve = []
|
||||
# Fill from first trade date to last
|
||||
first = datetime.strptime(dates[0], "%Y-%m-%d")
|
||||
last = datetime.strptime(dates[-1], "%Y-%m-%d")
|
||||
current = first
|
||||
while current <= last:
|
||||
d = current.strftime("%Y-%m-%d")
|
||||
if d in daily:
|
||||
equity += daily[d]
|
||||
curve.append({"date": d, "value": round(equity, 4)})
|
||||
current += timedelta(days=1)
|
||||
return curve
|
||||
|
||||
def compute_monthly_returns(equity_curve: list[dict]) -> dict:
|
||||
"""Compute monthly returns from daily equity curve."""
|
||||
if len(equity_curve) < 2:
|
||||
return {"years": [], "months": MONTHS, "matrix": []}
|
||||
|
||||
# Group by year-month
|
||||
monthly = OrderedDict()
|
||||
for pt in equity_curve:
|
||||
d = datetime.strptime(pt["date"], "%Y-%m-%d")
|
||||
ym = f"{d.year}-{d.month:02d}"
|
||||
if ym not in monthly:
|
||||
monthly[ym] = {"first": pt["value"], "last": pt["value"], "date": pt["date"]}
|
||||
monthly[ym]["last"] = pt["value"]
|
||||
monthly[ym]["date"] = pt["date"]
|
||||
|
||||
# Compute returns
|
||||
months_data = []
|
||||
prev_value = None
|
||||
for ym, data in monthly.items():
|
||||
if prev_value is not None and prev_value > 0:
|
||||
ret = ((data["last"] / prev_value) - 1) * 100
|
||||
else:
|
||||
ret = None
|
||||
prev_value = data["last"]
|
||||
year = int(ym[:4])
|
||||
month = int(ym[5:7])
|
||||
months_data.append({"year": year, "month": month, "return": ret})
|
||||
|
||||
if not months_data:
|
||||
return {"years": [], "months": MONTHS, "matrix": []}
|
||||
|
||||
years = sorted(set(m["year"] for m in months_data), reverse=True)
|
||||
matrix = []
|
||||
for yr in years:
|
||||
row = [None] * 12
|
||||
for m in months_data:
|
||||
if m["year"] == yr:
|
||||
v = m["return"]
|
||||
row[m["month"] - 1] = round(v, 1) if v is not None else None
|
||||
matrix.append(row)
|
||||
|
||||
return {"years": years, "months": MONTHS, "matrix": matrix}
|
||||
|
||||
def compute_yearly_returns(monthly_data: dict) -> tuple[list[dict], float]:
|
||||
"""Compute yearly returns from monthly returns matrix."""
|
||||
years = monthly_data.get("years", [])
|
||||
matrix = monthly_data.get("matrix", [])
|
||||
yearly = []
|
||||
|
||||
for i, yr in enumerate(years):
|
||||
total = 1.0
|
||||
row = matrix[i]
|
||||
has_data = False
|
||||
for v in row:
|
||||
if v is not None:
|
||||
total *= (1 + v / 100)
|
||||
has_data = True
|
||||
if has_data:
|
||||
ret = round((total - 1) * 100, 1)
|
||||
yearly.append({"year": yr, "return": ret})
|
||||
|
||||
if not yearly:
|
||||
return [], 0.0
|
||||
|
||||
mean = round(sum(r["return"] for r in yearly) / len(yearly), 1)
|
||||
return yearly, mean
|
||||
|
||||
def compute_return_distribution(monthly_data: dict) -> dict:
|
||||
"""Compute histogram of monthly returns for distribution chart."""
|
||||
matrix = monthly_data.get("matrix", [])
|
||||
all_returns = []
|
||||
for row in matrix:
|
||||
for v in row:
|
||||
if v is not None:
|
||||
all_returns.append(v)
|
||||
|
||||
if not all_returns:
|
||||
return {"bins": [], "mean": 0.0}
|
||||
|
||||
mean = round(sum(all_returns) / len(all_returns), 1)
|
||||
min_r, max_r = min(all_returns), max(all_returns)
|
||||
padding = 2
|
||||
min_r = math.floor(min_r) - padding
|
||||
max_r = math.ceil(max_r) + padding
|
||||
bin_width = max(1.0, round((max_r - min_r) / 10, 1))
|
||||
|
||||
bins = []
|
||||
current = min_r
|
||||
while current < max_r:
|
||||
end = current + bin_width
|
||||
count = sum(1 for r in all_returns if current <= r < end)
|
||||
bins.append({"start": round(current, 1), "end": round(end, 1), "count": count})
|
||||
current = end
|
||||
|
||||
return {"bins": bins, "mean": mean}
|
||||
|
||||
def compute_qq_plot(monthly_data: dict) -> dict:
|
||||
"""Compute QQ plot: theoretical vs observed quantiles for monthly returns."""
|
||||
matrix = monthly_data.get("matrix", [])
|
||||
all_returns = []
|
||||
for row in matrix:
|
||||
for v in row:
|
||||
if v is not None:
|
||||
all_returns.append(v)
|
||||
|
||||
if len(all_returns) < 10:
|
||||
return {"points": []}
|
||||
|
||||
import random
|
||||
random.seed(42)
|
||||
sorted_r = sorted(all_returns)
|
||||
n = len(sorted_r)
|
||||
mean_r = sum(sorted_r) / n
|
||||
# Sample std (using n-1)
|
||||
variance = sum((r - mean_r) ** 2 for r in sorted_r) / (n - 1) if n > 1 else 1
|
||||
std_r = math.sqrt(max(variance, 1e-10))
|
||||
|
||||
points = []
|
||||
for i in range(1, n + 1):
|
||||
p = i / (n + 1)
|
||||
# Approximate inverse normal (Abramowitz & Stegun approximation)
|
||||
t = math.sqrt(-2 * math.log(min(p, 1 - p)))
|
||||
c0 = 2.515517
|
||||
c1 = 0.802853
|
||||
c2 = 0.010328
|
||||
d1 = 1.432788
|
||||
d2 = 0.189269
|
||||
d3 = 0.001308
|
||||
sign = 1 if p >= 0.5 else -1
|
||||
theoretical = sign * (t - (c0 + c1 * t + c2 * t * t) / (1 + d1 * t + d2 * t * t + d3 * t * t * t))
|
||||
observed = (sorted_r[i - 1] - mean_r) / std_r
|
||||
points.append({
|
||||
"theoretical": round(theoretical, 3),
|
||||
"observed": round(observed, 3)
|
||||
})
|
||||
|
||||
return {"points": points}
|
||||
|
||||
def compute_rolling_stats(equity_curve: list[dict], window_days: int = 126) -> dict:
|
||||
"""Compute rolling 6-month (126 trading day) return and volatility."""
|
||||
roll = []
|
||||
values = [p["value"] for p in equity_curve]
|
||||
|
||||
for i in range(window_days, len(values)):
|
||||
past = values[i - window_days:i]
|
||||
cur_val = values[i]
|
||||
prev_val = values[i - window_days]
|
||||
|
||||
if prev_val > 0:
|
||||
# Rolling return: total return over window, annualized
|
||||
roll_ret = ((cur_val / prev_val) - 1)
|
||||
# Daily returns for volatility
|
||||
daily_rets = [(past[j] / past[j-1]) - 1 for j in range(1, len(past)) if past[j-1] > 0]
|
||||
if daily_rets:
|
||||
vol = math.sqrt(sum(r * r for r in daily_rets) / len(daily_rets)) * math.sqrt(365)
|
||||
else:
|
||||
vol = 0
|
||||
roll.append({
|
||||
"date": equity_curve[i]["date"],
|
||||
"rollingReturn": round(roll_ret * 100, 2),
|
||||
"rollingVolatility": round(vol * 100, 2)
|
||||
})
|
||||
|
||||
return {"windowMonths": 6, "series": roll}
|
||||
|
||||
def compute_quant_report(strategy_name: str, strategy_id: str, trades: list[dict],
|
||||
start_equity: float = 100.0) -> dict:
|
||||
"""Compute the full QF-Lib quant report."""
|
||||
equity = compute_daily_equity(trades, start_equity)
|
||||
monthly = compute_monthly_returns(equity)
|
||||
yearly, mean_yearly = compute_yearly_returns(monthly)
|
||||
distribution = compute_return_distribution(monthly)
|
||||
qq = compute_qq_plot(monthly)
|
||||
rolling = compute_rolling_stats(equity)
|
||||
|
||||
return {
|
||||
"meta": {
|
||||
"strategyName": strategy_name,
|
||||
"strategyId": strategy_id,
|
||||
"generatedAt": datetime.utcnow().isoformat() + "Z",
|
||||
"library": "QF-Lib",
|
||||
"version": "1.0.0"
|
||||
},
|
||||
"equityCurve": equity,
|
||||
"monthlyReturns": monthly,
|
||||
"yearlyReturns": yearly,
|
||||
"meanYearlyReturn": mean_yearly,
|
||||
"monthlyReturnDistribution": distribution,
|
||||
"qqPlot": qq,
|
||||
"rollingStats": rolling
|
||||
}
|
||||
Reference in New Issue
Block a user