# -*- coding: utf-8 -*- """ report.py ========= Generazione di report (XLSX + PNG) dei risultati della grid search. - write_full_report(df_results, agg, output_dir) → genera 4 file: 1) grid_search_full.xlsx (tutti i fold, una riga per (ISIN, combo, fold)) 2) grid_search_aggregate.xlsx (aggregato per combo, ordinato per Sharpe medio) 3) heatmap_*.png (heatmap delle metriche su coppie di parametri) 4) top_combos.xlsx (top-K combinazioni con confronto con baseline) L'idea è che il PM possa aprire SOLO grid_search_aggregate.xlsx e capire in 30 secondi quale set di parametri usare in produzione. """ from __future__ import annotations from pathlib import Path from typing import Optional import numpy as np import pandas as pd import matplotlib.pyplot as plt def write_full_report( df_results: pd.DataFrame, agg: pd.DataFrame, output_dir: Path, *, primary_metric: str = "Sharpe", top_k: int = 25, baseline_params: Optional[dict] = None, ) -> dict: """ Scrive tutti i file di output e ritorna un dict con i path. Parameters ---------- df_results : DataFrame "lungo" prodotto da run_grid_search agg : DataFrame aggregato prodotto da aggregate_results output_dir : Path primary_metric : str top_k : int Numero di top combinazioni da analizzare in dettaglio baseline_params : dict, opzionale Parametri della config attuale di produzione, per confronto. Esempio: {"Wp":60, "Ha":10, "k":25, "theta_entry":0.005, "sl_bps":300, "tp_bps":800, "trail_bps":300, "time_stop_bars":20, "decision_every":1, "min_holding_bars":0} """ output_dir = Path(output_dir) output_dir.mkdir(parents=True, exist_ok=True) paths = {} # === 1. Dataset completo === path_full = output_dir / "grid_search_full.xlsx" df_results.to_excel(path_full, index=False) paths["full"] = path_full print(f"[REPORT] Salvato {path_full} ({len(df_results):,} righe)") # === 2. Aggregato per combinazione === path_agg = output_dir / "grid_search_aggregate.xlsx" with pd.ExcelWriter(path_agg, engine="openpyxl") as xw: agg.to_excel(xw, sheet_name="ByCombo", index=False) if baseline_params is not None: baseline_row = _find_baseline_row(agg, baseline_params) if baseline_row is not None: pd.DataFrame([baseline_row]).to_excel(xw, sheet_name="Baseline", index=False) paths["aggregate"] = path_agg print(f"[REPORT] Salvato {path_agg}") # === 3. Top-K combinazioni === top = agg.head(top_k).copy() if baseline_params is not None: baseline_row = _find_baseline_row(agg, baseline_params) if baseline_row is not None: metric_mean = f"{primary_metric}_mean" base_metric = baseline_row.get(metric_mean, np.nan) top["delta_vs_baseline"] = top[metric_mean] - base_metric top["pct_improvement"] = (top[metric_mean] / base_metric - 1) * 100 if base_metric and base_metric != 0 else np.nan path_top = output_dir / "top_combos.xlsx" top.to_excel(path_top, index=False) paths["top"] = path_top print(f"[REPORT] Salvato {path_top}") # === 4. Heatmaps === heatmap_paths = _generate_heatmaps(agg, output_dir, primary_metric=primary_metric) paths.update(heatmap_paths) # === 5. Top combos equity comparison === # (Skipped if no per-fold equity data — left as hook for future enhancement) return paths def _find_baseline_row(agg: pd.DataFrame, baseline_params: dict) -> Optional[dict]: """Trova la riga corrispondente ai parametri baseline (se presente).""" mask = pd.Series(True, index=agg.index) for k, v in baseline_params.items(): if k not in agg.columns: continue if v is None: mask &= agg[k].isna() else: mask &= (agg[k] == v) if mask.sum() == 0: return None return agg[mask].iloc[0].to_dict() def _generate_heatmaps( agg: pd.DataFrame, output_dir: Path, *, primary_metric: str = "Sharpe", ) -> dict: """ Per ogni coppia "interessante" di parametri, genera una heatmap del primary_metric (mediato sugli altri parametri). """ paths = {} metric_col = f"{primary_metric}_mean" if metric_col not in agg.columns: return paths pairs_to_plot = [ ("decision_every", "tp_bps", "decision_vs_tp"), ("decision_every", "sl_bps", "decision_vs_sl"), ("tp_bps", "sl_bps", "tp_vs_sl"), ("Wp", "Ha", "wp_vs_ha"), ("decision_every", "min_holding_bars", "decision_vs_holding"), ("k", "theta_entry", "k_vs_theta"), ] for px, py, label in pairs_to_plot: if px not in agg.columns or py not in agg.columns: continue if agg[px].nunique() <= 1 or agg[py].nunique() <= 1: continue try: pivot = agg.pivot_table(index=py, columns=px, values=metric_col, aggfunc="mean") n_pivot = agg.pivot_table(index=py, columns=px, values="N_Trades_avg", aggfunc="mean") fig, ax = plt.subplots(figsize=(8, 5.5), dpi=120) im = ax.imshow(pivot.values, cmap="RdYlGn", aspect="auto") ax.set_xticks(range(len(pivot.columns))) ax.set_xticklabels(pivot.columns, fontsize=9) ax.set_yticks(range(len(pivot.index))) ax.set_yticklabels(pivot.index, fontsize=9) ax.set_xlabel(px, fontsize=10) ax.set_ylabel(py, fontsize=10) ax.set_title(f"{primary_metric} medio per {px} × {py} (annotazione: N_Trades_avg)", fontsize=10) # Annota celle: metrica sopra, n_trades sotto for i in range(pivot.shape[0]): for j in range(pivot.shape[1]): val = pivot.iloc[i, j] if not np.isfinite(val): continue n_val = n_pivot.iloc[i, j] if not n_pivot.empty else np.nan text = f"{val:.2f}" if np.isfinite(n_val): text += f"\nn={int(n_val)}" ax.text(j, i, text, ha="center", va="center", fontsize=7, color="black" if abs(val) < 0.8 else "white") fig.colorbar(im, ax=ax, label=primary_metric) plt.tight_layout() p = output_dir / f"heatmap_{label}.png" fig.savefig(p, dpi=120, bbox_inches="tight") plt.close(fig) paths[f"heatmap_{label}"] = p print(f"[REPORT] Salvata heatmap {p}") except Exception as e: print(f"[REPORT] Heatmap {label} fallita: {e}") return paths def summary_text_table(agg: pd.DataFrame, top_k: int = 10, primary_metric: str = "Sharpe") -> str: """ Restituisce una tabella testuale dei top-K combinazioni, formattata per la console. """ if agg is None or agg.empty: return "(nessun risultato)" cols_to_show = [ "Wp", "Ha", "k", "decision_every", "tp_bps", "sl_bps", "trail_bps", "min_holding_bars", f"{primary_metric}_mean", "Stability", "N_Trades_avg", "CAGR_avg", "MaxDD_avg", "Calmar_avg", ] cols_to_show = [c for c in cols_to_show if c in agg.columns] return agg[cols_to_show].head(top_k).to_string(index=False) __all__ = ["write_full_report", "summary_text_table"]