516 lines
14 KiB
Python
516 lines
14 KiB
Python
# src/calibration/reports/risk_report.py
|
||
import matplotlib.pyplot as plt
|
||
from reportlab.platypus import Image
|
||
import io
|
||
|
||
from datetime import datetime
|
||
from pathlib import Path
|
||
from typing import Dict, Any
|
||
|
||
from reportlab.platypus import (
|
||
SimpleDocTemplate,
|
||
Paragraph,
|
||
Spacer,
|
||
Table,
|
||
TableStyle,
|
||
)
|
||
from reportlab.lib.styles import ParagraphStyle, getSampleStyleSheet
|
||
from reportlab.lib import colors
|
||
from reportlab.lib.pagesizes import A4
|
||
from reportlab.lib.units import mm
|
||
from reportlab.platypus import PageBreak
|
||
|
||
# ============================================================
|
||
# HELPERS
|
||
# ============================================================
|
||
|
||
def _create_stop_histogram(
|
||
stop_distances,
|
||
p50=None,
|
||
p90=None,
|
||
p99=None,
|
||
):
|
||
|
||
fig, ax = plt.subplots(figsize=(6, 4))
|
||
|
||
ax.hist(
|
||
[d * 100 for d in stop_distances],
|
||
bins=40,
|
||
alpha=0.7,
|
||
)
|
||
|
||
# Percentile lines
|
||
if p50 is not None:
|
||
ax.axvline(p50 * 100, linestyle=":", linewidth=1)
|
||
|
||
if p90 is not None:
|
||
ax.axvline(p90 * 100, linestyle=":", linewidth=1)
|
||
|
||
if p99 is not None:
|
||
ax.axvline(p99 * 100, linestyle=":", linewidth=1)
|
||
|
||
ax.set_title("Stop Distance Distribution")
|
||
ax.set_xlabel("Stop Distance (%)")
|
||
ax.set_ylabel("Frequency")
|
||
|
||
buf = io.BytesIO()
|
||
plt.tight_layout()
|
||
plt.savefig(buf, format="png")
|
||
plt.close(fig)
|
||
buf.seek(0)
|
||
|
||
return buf
|
||
|
||
def _create_position_size_plot(
|
||
timestamps,
|
||
position_sizes,
|
||
*,
|
||
max_position_fraction=None,
|
||
):
|
||
ts, ps = _align_xy(timestamps, position_sizes)
|
||
if not ps:
|
||
return None
|
||
|
||
x = list(range(len(ps)))
|
||
y = [p * 100 for p in ps]
|
||
|
||
fig, ax = plt.subplots(figsize=(6, 4))
|
||
|
||
ax.plot(x, y, linewidth=0.8)
|
||
|
||
if max_position_fraction is not None:
|
||
ax.axhline(
|
||
max_position_fraction * 100,
|
||
linestyle="--",
|
||
linewidth=1,
|
||
)
|
||
|
||
ax.set_title("Position Size Over Time")
|
||
ax.set_ylabel("Position Size (% of equity)")
|
||
ax.set_xlabel("Samples")
|
||
|
||
buf = io.BytesIO()
|
||
plt.tight_layout()
|
||
plt.savefig(buf, format="png", dpi=150)
|
||
plt.close(fig)
|
||
buf.seek(0)
|
||
|
||
return buf
|
||
|
||
|
||
def _create_effective_risk_plot(
|
||
timestamps,
|
||
effective_risks,
|
||
*,
|
||
risk_target=None,
|
||
p50=None,
|
||
p90=None,
|
||
p99=None,
|
||
):
|
||
ts, er = _align_xy(timestamps, effective_risks)
|
||
if not er:
|
||
return None
|
||
|
||
x = list(range(len(er)))
|
||
y = [r * 100 for r in er]
|
||
|
||
fig, ax = plt.subplots(figsize=(6, 4))
|
||
|
||
# Main curve
|
||
ax.plot(x, y, linewidth=0.8)
|
||
|
||
# Risk target line
|
||
if risk_target is not None:
|
||
ax.axhline(
|
||
risk_target * 100,
|
||
linestyle="--",
|
||
linewidth=1,
|
||
)
|
||
|
||
ax.set_title("Effective Risk Over Time")
|
||
ax.set_ylabel("Effective Risk (%)")
|
||
ax.set_xlabel("Samples")
|
||
|
||
buf = io.BytesIO()
|
||
plt.tight_layout()
|
||
plt.savefig(buf, format="png", dpi=150)
|
||
plt.close(fig)
|
||
buf.seek(0)
|
||
|
||
return buf
|
||
|
||
def _align_xy(x, y):
|
||
"""
|
||
Ensures x and y have the same length.
|
||
Trims to the shortest length and drops None/NaN pairs.
|
||
Returns (x_aligned, y_aligned).
|
||
"""
|
||
if not x or not y:
|
||
return [], []
|
||
|
||
n = min(len(x), len(y))
|
||
x = x[:n]
|
||
y = y[:n]
|
||
|
||
x2, y2 = [], []
|
||
for xi, yi in zip(x, y):
|
||
if yi is None:
|
||
continue
|
||
try:
|
||
# Filter NaN
|
||
if isinstance(yi, float) and yi != yi:
|
||
continue
|
||
except Exception:
|
||
pass
|
||
x2.append(xi)
|
||
y2.append(yi)
|
||
|
||
return x2, y2
|
||
|
||
# ============================================================
|
||
# Footer (page number)
|
||
# ============================================================
|
||
|
||
def _add_footer(canvas, doc):
|
||
canvas.saveState()
|
||
footer_text = f"Trading Bot · Calibration Report · Page {doc.page}"
|
||
canvas.setFont("Helvetica", 8)
|
||
canvas.drawRightString(200 * mm, 10 * mm, footer_text)
|
||
canvas.restoreState()
|
||
|
||
|
||
# ============================================================
|
||
# Main PDF generator
|
||
# ============================================================
|
||
|
||
def generate_risk_report_pdf(
|
||
*,
|
||
output_path: Path,
|
||
context: Dict[str, Any],
|
||
config: Dict[str, Any],
|
||
results: Dict[str, Any],
|
||
):
|
||
styles = getSampleStyleSheet()
|
||
|
||
title_style = styles["Title"]
|
||
heading_style = styles["Heading2"]
|
||
normal_style = styles["Normal"]
|
||
|
||
story = []
|
||
|
||
# ============================================================
|
||
# TITLE
|
||
# ============================================================
|
||
|
||
story.append(
|
||
Paragraph(
|
||
"Calibration Report · Step 2 · Risk & Stops",
|
||
title_style,
|
||
)
|
||
)
|
||
story.append(Spacer(1, 12))
|
||
|
||
story.append(
|
||
Paragraph(
|
||
f"Generated at: {datetime.utcnow().strftime('%Y-%m-%d %H:%M:%S')} UTC",
|
||
normal_style,
|
||
)
|
||
)
|
||
story.append(Spacer(1, 24))
|
||
|
||
# ============================================================
|
||
# CONTEXT
|
||
# ============================================================
|
||
|
||
story.append(Paragraph("1. Context", heading_style))
|
||
story.append(Spacer(1, 8))
|
||
|
||
context_table = Table(
|
||
[[k, str(v)] for k, v in context.items()],
|
||
colWidths=[180, 300],
|
||
)
|
||
|
||
context_table.setStyle(
|
||
TableStyle([
|
||
("GRID", (0, 0), (-1, -1), 0.5, colors.grey),
|
||
("BACKGROUND", (0, 0), (-1, 0), colors.whitesmoke),
|
||
])
|
||
)
|
||
|
||
story.append(context_table)
|
||
story.append(Spacer(1, 24))
|
||
|
||
# ============================================================
|
||
# CONFIGURATION
|
||
# ============================================================
|
||
|
||
story.append(Paragraph("2. Configuration", heading_style))
|
||
story.append(Spacer(1, 8))
|
||
|
||
config_table = Table(
|
||
[[k, str(v)] for k, v in config.items()],
|
||
colWidths=[180, 300],
|
||
)
|
||
|
||
config_table.setStyle(
|
||
TableStyle([
|
||
("GRID", (0, 0), (-1, -1), 0.5, colors.grey),
|
||
])
|
||
)
|
||
|
||
story.append(config_table)
|
||
story.append(Spacer(1, 24))
|
||
|
||
# ============================================================
|
||
# VALIDATION RESULT
|
||
# ============================================================
|
||
|
||
story.append(Paragraph("3. Risk Validation Result", heading_style))
|
||
story.append(Spacer(1, 8))
|
||
|
||
status = results.get("status", "unknown").upper()
|
||
|
||
status_color = {
|
||
"OK": colors.green,
|
||
"WARNING": colors.orange,
|
||
"FAIL": colors.red,
|
||
}.get(status, colors.black)
|
||
|
||
status_table = Table(
|
||
[["Status", status]],
|
||
colWidths=[180, 300],
|
||
)
|
||
|
||
status_table.setStyle(
|
||
TableStyle([
|
||
("TEXTCOLOR", (1, 0), (1, 0), status_color),
|
||
("GRID", (0, 0), (-1, -1), 0.5, colors.grey),
|
||
])
|
||
)
|
||
|
||
story.append(status_table)
|
||
story.append(Spacer(1, 12))
|
||
|
||
story.append(
|
||
Paragraph(
|
||
results.get("message", ""),
|
||
normal_style,
|
||
)
|
||
)
|
||
story.append(Spacer(1, 24))
|
||
|
||
# ============================================================
|
||
# CHECKS
|
||
# ============================================================
|
||
|
||
checks = results.get("checks", {})
|
||
|
||
if checks:
|
||
story.append(Paragraph("4. Detailed Checks", heading_style))
|
||
story.append(Spacer(1, 8))
|
||
|
||
check_rows = [["Check", "Status", "Message"]]
|
||
|
||
for name, data in checks.items():
|
||
check_rows.append([
|
||
name,
|
||
data.get("status", "").upper(),
|
||
data.get("message", ""),
|
||
])
|
||
|
||
check_table = Table(
|
||
check_rows,
|
||
colWidths=[150, 80, 250],
|
||
)
|
||
|
||
check_table.setStyle(
|
||
TableStyle([
|
||
("BACKGROUND", (0, 0), (-1, 0), colors.lightgrey),
|
||
("GRID", (0, 0), (-1, -1), 0.5, colors.grey),
|
||
])
|
||
)
|
||
|
||
story.append(check_table)
|
||
story.append(Spacer(1, 24))
|
||
|
||
# ============================================================
|
||
# STOP DISTANCE STATISTICS
|
||
# ============================================================
|
||
|
||
stop_metrics = (
|
||
results.get("checks", {})
|
||
.get("stop_sanity", {})
|
||
.get("metrics", {})
|
||
)
|
||
|
||
if stop_metrics:
|
||
story.append(Paragraph("5. Stop Distance Statistics", heading_style))
|
||
story.append(Spacer(1, 8))
|
||
|
||
stats_table = Table(
|
||
[
|
||
["Metric", "Value (%)"],
|
||
["P50 (typical)", f"{stop_metrics.get('p50', 0) * 100:.2f}%"],
|
||
["P90 (wide)", f"{stop_metrics.get('p90', 0) * 100:.2f}%"],
|
||
["P99 (extreme)", f"{stop_metrics.get('p99', 0) * 100:.2f}%"],
|
||
],
|
||
colWidths=[180, 180],
|
||
)
|
||
|
||
stats_table.setStyle(
|
||
TableStyle([
|
||
("BACKGROUND", (0, 0), (-1, 0), colors.lightgrey),
|
||
("GRID", (0, 0), (-1, -1), 0.5, colors.grey),
|
||
])
|
||
)
|
||
|
||
story.append(stats_table)
|
||
story.append(Spacer(1, 24))
|
||
|
||
# ============================================================
|
||
# RISK SIZING BREAKDOWN (WITH FORMULAS)
|
||
# ============================================================
|
||
|
||
sizing_metrics = (
|
||
results.get("checks", {})
|
||
.get("sizer_feasibility", {})
|
||
.get("metrics", {})
|
||
)
|
||
|
||
stop_metrics = (
|
||
results.get("checks", {})
|
||
.get("stop_sanity", {})
|
||
.get("metrics", {})
|
||
)
|
||
|
||
if sizing_metrics and stop_metrics:
|
||
story.append(Paragraph("6. Risk Sizing Breakdown", heading_style))
|
||
story.append(Spacer(1, 8))
|
||
|
||
equity = context.get("Account equity", 0)
|
||
risk_pct = config.get("Risk per trade (%)", 0)
|
||
max_pos_pct = config.get("Max position fraction (%)", 0)
|
||
|
||
p50 = stop_metrics.get("p50", 0)
|
||
ideal_position = sizing_metrics.get("ideal_position_median", 0)
|
||
max_position_value = sizing_metrics.get("max_position_value", 0)
|
||
effective_position = sizing_metrics.get("effective_position_median", 0)
|
||
effective_risk = sizing_metrics.get("effective_risk_median", 0)
|
||
|
||
breakdown_lines = [
|
||
f"Position size (ideal) = {equity:,.0f} × {risk_pct:.2f}% ÷ {p50*100:.2f}% ≈ {ideal_position:,.0f}",
|
||
f"Max position size = {equity:,.0f} × {max_pos_pct:.2f}% = {max_position_value:,.0f}",
|
||
f"Effective position = min({ideal_position:,.0f}, {max_position_value:,.0f}) = {effective_position:,.0f}",
|
||
f"Effective risk = {effective_position:,.0f} × {p50*100:.2f}% ÷ {equity:,.0f} ≈ {effective_risk*100:.2f}%",
|
||
]
|
||
|
||
for line in breakdown_lines:
|
||
story.append(Paragraph(line, normal_style))
|
||
story.append(Spacer(1, 4))
|
||
|
||
story.append(Spacer(1, 24))
|
||
|
||
# ============================================================
|
||
# STOP CONFIGURATION DETAILS
|
||
# ============================================================
|
||
|
||
stop_snapshot = results.get("config_snapshot", {}).get("stop", {})
|
||
|
||
story.append(Paragraph("7. Stop Configuration Details", heading_style))
|
||
story.append(Spacer(1, 8))
|
||
|
||
for k, v in stop_snapshot.items():
|
||
story.append(Paragraph(f"{k}: {v}", normal_style))
|
||
story.append(Spacer(1, 4))
|
||
|
||
# ============================================================
|
||
# STOP DISTANCE HISTOGRAM
|
||
# ============================================================
|
||
|
||
series = results.get("series", {})
|
||
stop_distances = series.get("stop_distances")
|
||
|
||
if stop_distances:
|
||
story.append(Paragraph("6. Stop Distance Distribution", heading_style))
|
||
story.append(Spacer(1, 8))
|
||
|
||
stop_metrics = (
|
||
results.get("checks", {})
|
||
.get("stop_sanity", {})
|
||
.get("metrics", {})
|
||
)
|
||
|
||
img_buffer = _create_stop_histogram(
|
||
stop_distances,
|
||
p50=stop_metrics.get("p50"),
|
||
p90=stop_metrics.get("p90"),
|
||
p99=stop_metrics.get("p99"),
|
||
)
|
||
|
||
img = Image(img_buffer, width=400, height=250)
|
||
story.append(img)
|
||
story.append(Spacer(1, 24))
|
||
|
||
# ============================================================
|
||
# POSITION SIZE OVER TIME
|
||
# ============================================================
|
||
|
||
position_sizes = series.get("position_size_pct")
|
||
timestamps = series.get("timestamps")
|
||
|
||
if position_sizes and timestamps:
|
||
|
||
story.append(PageBreak())
|
||
story.append(Paragraph("7. Position Size Over Time", heading_style))
|
||
story.append(Spacer(1, 8))
|
||
|
||
img_buffer = _create_position_size_plot(
|
||
timestamps,
|
||
position_sizes,
|
||
max_position_fraction=config.get("Max position fraction (%)", 0) / 100,
|
||
)
|
||
|
||
if img_buffer:
|
||
img = Image(img_buffer, width=400, height=250)
|
||
story.append(img)
|
||
story.append(Spacer(1, 24))
|
||
|
||
# ============================================================
|
||
# EFFECTIVE RISK OVER TIME
|
||
# ============================================================
|
||
|
||
effective_risks = series.get("effective_risk_pct")
|
||
|
||
if effective_risks and timestamps:
|
||
|
||
story.append(Paragraph("8. Effective Risk Over Time", heading_style))
|
||
story.append(Spacer(1, 8))
|
||
|
||
risk_target = config.get("Risk per trade (%)", 0) / 100
|
||
|
||
img_buffer = _create_effective_risk_plot(
|
||
timestamps,
|
||
effective_risks,
|
||
risk_target=risk_target,
|
||
)
|
||
|
||
if img_buffer:
|
||
img = Image(img_buffer, width=400, height=250)
|
||
story.append(img)
|
||
story.append(Spacer(1, 24))
|
||
|
||
# ============================================================
|
||
# BUILD
|
||
# ============================================================
|
||
|
||
doc = SimpleDocTemplate(
|
||
str(output_path),
|
||
pagesize=A4,
|
||
rightMargin=36,
|
||
leftMargin=36,
|
||
topMargin=36,
|
||
bottomMargin=36,
|
||
)
|
||
|
||
doc.build(story, onFirstPage=_add_footer, onLaterPages=_add_footer)
|