Step 2: Risk and Stops - Generating reports with pdf. It is missing to reorder the PDF.
This commit is contained in:
444
src/calibration/reports/risk_report.py
Normal file
444
src/calibration/reports/risk_report.py
Normal file
@@ -0,0 +1,444 @@
|
||||
# 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):
|
||||
|
||||
fig, ax = plt.subplots(figsize=(6, 4))
|
||||
|
||||
ax.hist(
|
||||
[d * 100 for d in stop_distances],
|
||||
bins=40,
|
||||
alpha=0.7,
|
||||
)
|
||||
|
||||
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):
|
||||
# Align and be robust
|
||||
ts, ps = _align_xy(timestamps, position_sizes)
|
||||
if not ps:
|
||||
return None
|
||||
|
||||
x = list(range(len(ps))) # robust axis (avoid matplotlib categorical date issues)
|
||||
y = [p * 100 for p in ps]
|
||||
|
||||
fig, ax = plt.subplots(figsize=(6, 4))
|
||||
ax.plot(x, y, linewidth=0.8)
|
||||
|
||||
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):
|
||||
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))
|
||||
ax.plot(x, y, linewidth=0.8)
|
||||
|
||||
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))
|
||||
|
||||
img_buffer = _create_stop_histogram(stop_distances)
|
||||
|
||||
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)
|
||||
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))
|
||||
|
||||
img_buffer = _create_effective_risk_plot(timestamps, effective_risks)
|
||||
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)
|
||||
Reference in New Issue
Block a user