""" Reporting module for backtest results. Handles summary printing, CSV exports, and plotting. """ from pathlib import Path import pandas as pd import plotly.graph_objects as go import vectorbt as vbt from plotly.subplots import make_subplots from engine.logging_config import get_logger logger = get_logger(__name__) class Reporter: """Reporter for backtest results with market-specific metrics.""" def __init__(self, output_dir: str = "backtest_logs"): self.output_dir = Path(output_dir) self.output_dir.mkdir(exist_ok=True) def print_summary(self, result) -> None: """ Print backtest summary to console via logger. Args: result: BacktestResult or vbt.Portfolio object """ (portfolio, market_type, leverage, funding_paid, liq_count, liq_loss, adjusted_return) = self._extract_result_data(result) # Extract period info idx = portfolio.wrapper.index start_date = idx[0].strftime("%Y-%m-%d") end_date = idx[-1].strftime("%Y-%m-%d") # Extract price info close = portfolio.close start_price = close.iloc[0].mean() if hasattr(close.iloc[0], 'mean') else close.iloc[0] end_price = close.iloc[-1].mean() if hasattr(close.iloc[-1], 'mean') else close.iloc[-1] price_change = ((end_price - start_price) / start_price) * 100 # Extract fees stats = portfolio.stats() total_fees = stats.get('Total Fees Paid', 0) raw_return = portfolio.total_return().mean() * 100 # Build summary summary_lines = [ "", "=" * 50, "BACKTEST RESULTS", "=" * 50, f"Market Type: [{market_type.upper()}]", f"Leverage: [{leverage}x]", f"Period: [{start_date} to {end_date}]", f"Price: [{start_price:,.2f} -> {end_price:,.2f} ({price_change:+.2f}%)]", ] # Show adjusted return if liquidations occurred if liq_count > 0 and adjusted_return is not None: summary_lines.append(f"Raw Return: [%{raw_return:.2f}] (before liq adjustment)") summary_lines.append(f"Adj Return: [%{adjusted_return:.2f}] (after liq losses)") else: summary_lines.append(f"Total Return: [%{raw_return:.2f}]") summary_lines.extend([ f"Sharpe Ratio: [{portfolio.sharpe_ratio().mean():.2f}]", f"Max Drawdown: [%{portfolio.max_drawdown().mean() * 100:.2f}]", f"Total Trades: [{portfolio.trades.count().mean():.0f}]", f"Win Rate: [%{portfolio.trades.win_rate().mean() * 100:.2f}]", f"Total Fees: [{total_fees:,.2f}]", ]) if funding_paid != 0: summary_lines.append(f"Funding Paid: [{funding_paid:,.2f}]") if liq_count > 0: summary_lines.append(f"Liquidations: [{liq_count}] (${liq_loss:,.2f} margin lost)") summary_lines.append("=" * 50) logger.info("\n".join(summary_lines)) def save_reports(self, result, filename_prefix: str) -> None: """ Save trade log, stats, and liquidation events to CSV files. Args: result: BacktestResult or vbt.Portfolio object filename_prefix: Prefix for output filenames """ (portfolio, market_type, leverage, funding_paid, liq_count, liq_loss, adjusted_return) = self._extract_result_data(result) # Save trades self._save_csv( data=portfolio.trades.records_readable, path=self.output_dir / f"{filename_prefix}_trades.csv", description="trade log" ) # Save stats with market-specific additions stats = portfolio.stats() stats['Market Type'] = market_type stats['Leverage'] = leverage stats['Total Funding Paid'] = funding_paid stats['Liquidations'] = liq_count stats['Liquidation Loss'] = liq_loss if adjusted_return is not None: stats['Adjusted Return'] = adjusted_return self._save_csv( data=stats, path=self.output_dir / f"{filename_prefix}_stats.csv", description="stats" ) # Save liquidation events if any if hasattr(result, 'liquidation_events') and result.liquidation_events: liq_df = pd.DataFrame([ { 'entry_time': e.entry_time, 'entry_price': e.entry_price, 'liquidation_time': e.liquidation_time, 'liquidation_price': e.liquidation_price, 'actual_price': e.actual_price, 'direction': e.direction, 'margin_lost_pct': e.margin_lost_pct } for e in result.liquidation_events ]) self._save_csv( data=liq_df, path=self.output_dir / f"{filename_prefix}_liquidations.csv", description="liquidation events" ) def plot(self, portfolio: vbt.Portfolio, show: bool = True) -> None: """Display portfolio plot.""" if show: portfolio.plot().show() def plot_wfa( self, wfa_results: pd.DataFrame, stitched_curve: pd.Series | None = None, show: bool = True ) -> None: """ Plot Walk-Forward Analysis results. Args: wfa_results: DataFrame with WFA window results stitched_curve: Stitched out-of-sample equity curve show: Whether to display the plot """ fig = make_subplots( rows=2, cols=1, shared_xaxes=False, vertical_spacing=0.1, subplot_titles=( "Walk-Forward Test Scores (Sharpe)", "Stitched Out-of-Sample Equity" ) ) fig.add_trace( go.Bar( x=wfa_results['window'], y=wfa_results['test_score'], name="Test Sharpe" ), row=1, col=1 ) if stitched_curve is not None: fig.add_trace( go.Scatter( x=stitched_curve.index, y=stitched_curve.values, name="Equity", mode='lines' ), row=2, col=1 ) fig.update_layout(height=800, title_text="Walk-Forward Analysis Report") if show: fig.show() def _extract_result_data(self, result) -> tuple: """ Extract data from BacktestResult or raw Portfolio. Returns: Tuple of (portfolio, market_type, leverage, funding_paid, liq_count, liq_loss, adjusted_return) """ if hasattr(result, 'portfolio'): return ( result.portfolio, result.market_type.value, result.leverage, result.total_funding_paid, result.liquidation_count, getattr(result, 'total_liquidation_loss', 0.0), getattr(result, 'adjusted_return', None) ) return (result, "unknown", 1, 0.0, 0, 0.0, None) def _save_csv(self, data, path: Path, description: str) -> None: """ Save data to CSV with consistent error handling. Args: data: DataFrame or Series to save path: Output file path description: Human-readable description for logging """ try: data.to_csv(path) logger.info("Saved %s to %s", description, path) except Exception as e: logger.error("Could not save %s: %s", description, e)