diff --git a/tests/unit/backtest/test_report_generator.py b/tests/unit/backtest/test_report_generator.py new file mode 100644 index 00000000..aee56e19 --- /dev/null +++ b/tests/unit/backtest/test_report_generator.py @@ -0,0 +1,805 @@ +"""Tests for report generator module. + +Issue #44: [BT-43] Report generator - PDF/HTML reports +""" + +import json +import tempfile +from datetime import datetime, timedelta +from decimal import Decimal +from pathlib import Path + +import pytest + +from tradingagents.backtest import ( + # Backtest engine + BacktestResult, + BacktestTrade, + BacktestSnapshot, + BacktestConfig, + OrderSide, + # Report generator + ReportFormat, + ReportSection, + ChartType, + ReportConfig, + ChartData, + ReportContent, + ReportResult, + ReportGenerator, + create_report_generator, +) + + +# ============================================================================ +# Test Fixtures +# ============================================================================ + +@pytest.fixture +def sample_config() -> BacktestConfig: + """Create sample backtest configuration.""" + return BacktestConfig( + initial_capital=Decimal("100000"), + ) + + +@pytest.fixture +def sample_trades() -> list[BacktestTrade]: + """Create sample trades for testing.""" + trades = [] + base_date = datetime(2023, 1, 3) + + # Winning trade + trades.append(BacktestTrade( + timestamp=base_date, + symbol="AAPL", + side=OrderSide.BUY, + quantity=Decimal("100"), + price=Decimal("150.00"), + commission=Decimal("1.00"), + slippage=Decimal("0.50"), + pnl=Decimal("500.00"), + )) + + # Losing trade + trades.append(BacktestTrade( + timestamp=base_date + timedelta(days=5), + symbol="GOOGL", + side=OrderSide.BUY, + quantity=Decimal("50"), + price=Decimal("100.00"), + commission=Decimal("1.00"), + slippage=Decimal("0.25"), + pnl=Decimal("-200.00"), + )) + + # Another winning trade + trades.append(BacktestTrade( + timestamp=base_date + timedelta(days=10), + symbol="MSFT", + side=OrderSide.BUY, + quantity=Decimal("75"), + price=Decimal("300.00"), + commission=Decimal("1.50"), + slippage=Decimal("0.75"), + pnl=Decimal("750.00"), + )) + + return trades + + +@pytest.fixture +def sample_snapshots() -> list[BacktestSnapshot]: + """Create sample snapshots for testing.""" + snapshots = [] + base_date = datetime(2023, 1, 1) + + for i in range(30): + equity = Decimal("100000") + Decimal(str(i * 100 - 50 * (i % 5))) + snapshots.append(BacktestSnapshot( + timestamp=base_date + timedelta(days=i), + cash=equity * Decimal("0.3"), + positions_value=equity * Decimal("0.7"), + total_value=equity, + )) + + return snapshots + + +@pytest.fixture +def sample_result(sample_trades, sample_snapshots) -> BacktestResult: + """Create sample backtest result.""" + return BacktestResult( + initial_capital=Decimal("100000"), + final_value=Decimal("101050"), + total_return=Decimal("1.05"), + total_trades=3, + winning_trades=2, + losing_trades=1, + win_rate=Decimal("66.67"), + profit_factor=Decimal("6.25"), + max_drawdown=Decimal("0.50"), + sharpe_ratio=Decimal("1.85"), + sortino_ratio=Decimal("2.50"), + total_commission=Decimal("3.50"), + total_slippage=Decimal("1.50"), + start_date=datetime(2023, 1, 1), + end_date=datetime(2023, 1, 30), + trades=sample_trades, + snapshots=sample_snapshots, + ) + + +# ============================================================================ +# Test Enums +# ============================================================================ + +class TestReportFormat: + """Tests for ReportFormat enum.""" + + def test_values(self): + """Test enum values.""" + assert ReportFormat.HTML.value == "html" + assert ReportFormat.PDF.value == "pdf" + assert ReportFormat.JSON.value == "json" + assert ReportFormat.MARKDOWN.value == "markdown" + + def test_all_formats(self): + """Test all formats exist.""" + formats = list(ReportFormat) + assert len(formats) == 4 + + +class TestReportSection: + """Tests for ReportSection enum.""" + + def test_values(self): + """Test enum values.""" + assert ReportSection.SUMMARY.value == "summary" + assert ReportSection.EQUITY_CURVE.value == "equity_curve" + assert ReportSection.DRAWDOWN.value == "drawdown" + assert ReportSection.TRADE_LIST.value == "trade_list" + + def test_all_sections(self): + """Test all sections exist.""" + sections = list(ReportSection) + assert len(sections) >= 8 + + +class TestChartType: + """Tests for ChartType enum.""" + + def test_values(self): + """Test enum values.""" + assert ChartType.LINE.value == "line" + assert ChartType.BAR.value == "bar" + assert ChartType.HEATMAP.value == "heatmap" + + +# ============================================================================ +# Test Data Classes +# ============================================================================ + +class TestReportConfig: + """Tests for ReportConfig dataclass.""" + + def test_default_creation(self): + """Test default configuration.""" + config = ReportConfig() + assert config.title == "Backtest Report" + assert config.author == "" + assert config.chart_width == 800 + assert config.chart_height == 400 + assert config.decimal_places == 2 + assert config.include_trade_list is True + assert config.max_trades_shown == 100 + assert config.include_timestamp is True + + def test_custom_creation(self): + """Test custom configuration.""" + config = ReportConfig( + title="My Report", + author="Test Author", + chart_width=1024, + chart_height=768, + decimal_places=4, + include_trade_list=False, + max_trades_shown=50, + ) + assert config.title == "My Report" + assert config.author == "Test Author" + assert config.chart_width == 1024 + assert config.chart_height == 768 + assert config.decimal_places == 4 + assert config.include_trade_list is False + assert config.max_trades_shown == 50 + + def test_color_scheme(self): + """Test default color scheme.""" + config = ReportConfig() + assert "primary" in config.color_scheme + assert "secondary" in config.color_scheme + assert "success" in config.color_scheme + assert "danger" in config.color_scheme + + def test_include_sections_default(self): + """Test default sections include all.""" + config = ReportConfig() + # Default includes all sections + assert len(config.include_sections) > 0 + + +class TestChartData: + """Tests for ChartData dataclass.""" + + def test_creation(self): + """Test chart data creation.""" + chart = ChartData( + chart_type=ChartType.LINE, + title="Test Chart", + x_data=[1, 2, 3], + y_data=[10, 20, 30], + x_label="X Axis", + y_label="Y Axis", + ) + assert chart.chart_type == ChartType.LINE + assert chart.title == "Test Chart" + assert chart.x_data == [1, 2, 3] + assert chart.y_data == [10, 20, 30] + assert chart.x_label == "X Axis" + assert chart.y_label == "Y Axis" + + def test_default_values(self): + """Test default values.""" + chart = ChartData( + chart_type=ChartType.BAR, + title="Test", + x_data=[], + y_data=[], + ) + assert chart.x_label == "" + assert chart.y_label == "" + assert chart.series_names == [] + assert chart.colors == [] + + +class TestReportContent: + """Tests for ReportContent dataclass.""" + + def test_creation(self): + """Test content creation.""" + now = datetime.now() + content = ReportContent( + title="Test Report", + generated_at=now, + ) + assert content.title == "Test Report" + assert content.generated_at == now + assert content.sections == {} + assert content.charts == {} + assert content.metadata == {} + + +class TestReportResult: + """Tests for ReportResult dataclass.""" + + def test_success_result(self): + """Test successful result.""" + result = ReportResult( + success=True, + output_path=Path("/tmp/report.html"), + format=ReportFormat.HTML, + file_size_bytes=1024, + generation_time_ms=500, + ) + assert result.success is True + assert result.output_path == Path("/tmp/report.html") + assert result.format == ReportFormat.HTML + assert result.file_size_bytes == 1024 + assert result.generation_time_ms == 500 + assert result.error is None + + def test_failure_result(self): + """Test failure result.""" + result = ReportResult( + success=False, + format=ReportFormat.PDF, + error="PDF library not available", + ) + assert result.success is False + assert result.output_path is None + assert result.error == "PDF library not available" + + +# ============================================================================ +# Test ReportGenerator +# ============================================================================ + +class TestReportGenerator: + """Tests for ReportGenerator class.""" + + def test_initialization(self): + """Test generator initialization.""" + generator = ReportGenerator() + assert generator.config is not None + assert generator.analyzer is not None + + def test_initialization_with_config(self): + """Test initialization with custom config.""" + config = ReportConfig(title="Custom Title") + generator = ReportGenerator(config=config) + assert generator.config.title == "Custom Title" + + def test_generate_html(self, sample_result): + """Test HTML generation.""" + generator = ReportGenerator() + + with tempfile.TemporaryDirectory() as tmpdir: + output_path = Path(tmpdir) / "report.html" + result = generator.generate_html(sample_result, output_path) + + assert result.success is True + assert result.output_path == output_path + assert result.format == ReportFormat.HTML + assert result.file_size_bytes > 0 + assert output_path.exists() + + def test_generate_html_content(self, sample_result): + """Test HTML content contains expected sections.""" + generator = ReportGenerator() + + with tempfile.TemporaryDirectory() as tmpdir: + output_path = Path(tmpdir) / "report.html" + generator.generate_html(sample_result, output_path) + + content = output_path.read_text() + + # Check structure + assert "" in content + assert "" in content + + # Check sections + assert "Summary" in content + assert "Trade Statistics" in content + assert "Risk Metrics" in content + + def test_generate_json(self, sample_result): + """Test JSON generation.""" + generator = ReportGenerator() + + with tempfile.TemporaryDirectory() as tmpdir: + output_path = Path(tmpdir) / "report.json" + result = generator.generate_json(sample_result, output_path) + + assert result.success is True + assert result.format == ReportFormat.JSON + assert output_path.exists() + + # Verify JSON is valid + data = json.loads(output_path.read_text()) + assert "report" in data + assert "backtest" in data + assert "performance" in data + assert "trades" in data + + def test_generate_json_structure(self, sample_result): + """Test JSON structure is correct.""" + generator = ReportGenerator() + + with tempfile.TemporaryDirectory() as tmpdir: + output_path = Path(tmpdir) / "report.json" + generator.generate_json(sample_result, output_path) + + data = json.loads(output_path.read_text()) + + # Check report metadata + assert "title" in data["report"] + assert "generated_at" in data["report"] + + # Check backtest data + assert "initial_capital" in data["backtest"] + assert "final_capital" in data["backtest"] + + # Check performance metrics + assert "total_return" in data["performance"] + assert "sharpe_ratio" in data["performance"] + + def test_generate_markdown(self, sample_result): + """Test Markdown generation.""" + generator = ReportGenerator() + + with tempfile.TemporaryDirectory() as tmpdir: + output_path = Path(tmpdir) / "report.md" + result = generator.generate_markdown(sample_result, output_path) + + assert result.success is True + assert result.format == ReportFormat.MARKDOWN + assert output_path.exists() + + def test_generate_markdown_content(self, sample_result): + """Test Markdown content structure.""" + generator = ReportGenerator() + + with tempfile.TemporaryDirectory() as tmpdir: + output_path = Path(tmpdir) / "report.md" + generator.generate_markdown(sample_result, output_path) + + content = output_path.read_text() + + # Check structure + assert "# " in content # Title + assert "## Summary" in content + assert "## Trade Statistics" in content + assert "## Risk Metrics" in content + assert "|" in content # Tables + + def test_generate_pdf_fallback(self, sample_result): + """Test PDF generation falls back to HTML if libraries unavailable.""" + generator = ReportGenerator() + + with tempfile.TemporaryDirectory() as tmpdir: + output_path = Path(tmpdir) / "report.pdf" + result = generator.generate_pdf(sample_result, output_path) + + # Should succeed but might fallback to HTML + assert result.success is True + # Either PDF worked or fell back to HTML + assert result.output_path is not None + + def test_generate_generic(self, sample_result): + """Test generic generate method.""" + generator = ReportGenerator() + + with tempfile.TemporaryDirectory() as tmpdir: + # Test HTML + html_path = Path(tmpdir) / "report.html" + result = generator.generate(sample_result, html_path, ReportFormat.HTML) + assert result.success is True + assert result.format == ReportFormat.HTML + + # Test JSON + json_path = Path(tmpdir) / "report.json" + result = generator.generate(sample_result, json_path, ReportFormat.JSON) + assert result.success is True + assert result.format == ReportFormat.JSON + + def test_custom_title(self, sample_result): + """Test custom report title.""" + config = ReportConfig(title="My Custom Report") + generator = ReportGenerator(config=config) + + with tempfile.TemporaryDirectory() as tmpdir: + output_path = Path(tmpdir) / "report.html" + generator.generate_html(sample_result, output_path) + + content = output_path.read_text() + assert "My Custom Report" in content + + def test_custom_author(self, sample_result): + """Test custom author in report.""" + config = ReportConfig(author="Test Author") + generator = ReportGenerator(config=config) + + with tempfile.TemporaryDirectory() as tmpdir: + output_path = Path(tmpdir) / "report.html" + generator.generate_html(sample_result, output_path) + + content = output_path.read_text() + assert "Test Author" in content + + def test_max_trades_shown(self, sample_result): + """Test max trades limit in report.""" + config = ReportConfig(max_trades_shown=1) + generator = ReportGenerator(config=config) + + with tempfile.TemporaryDirectory() as tmpdir: + output_path = Path(tmpdir) / "report.json" + generator.generate_json(sample_result, output_path) + + data = json.loads(output_path.read_text()) + # Should only show 1 trade + assert len(data["trade_list"]) <= 1 + + +class TestHTMLRendering: + """Tests for HTML rendering.""" + + def test_summary_metrics(self, sample_result): + """Test summary metrics in HTML.""" + generator = ReportGenerator() + + with tempfile.TemporaryDirectory() as tmpdir: + output_path = Path(tmpdir) / "report.html" + generator.generate_html(sample_result, output_path) + + content = output_path.read_text() + + # Check metrics are present + assert "Total Return" in content + assert "Win Rate" in content + assert "Sharpe Ratio" in content + + def test_trade_list_rendering(self, sample_result): + """Test trade list in HTML.""" + generator = ReportGenerator() + + with tempfile.TemporaryDirectory() as tmpdir: + output_path = Path(tmpdir) / "report.html" + generator.generate_html(sample_result, output_path) + + content = output_path.read_text() + + # Check trade table + assert "Trade List" in content + assert "Symbol" in content + assert "AAPL" in content + + def test_css_styling(self, sample_result): + """Test CSS is included.""" + generator = ReportGenerator() + + with tempfile.TemporaryDirectory() as tmpdir: + output_path = Path(tmpdir) / "report.html" + generator.generate_html(sample_result, output_path) + + content = output_path.read_text() + + # Check CSS is present + assert " + + +
+ {header} + {content} + {footer} +
+ +""" + + +# ============================================================================ +# Main Class +# ============================================================================ + +class ReportGenerator: + """Generator for backtest reports. + + Generates professional reports from backtest results in various formats + including HTML and PDF with embedded charts and analysis. + + Attributes: + config: Report configuration + analyzer: Results analyzer for metrics + + Example: + >>> generator = ReportGenerator() + >>> result = generator.generate_html(backtest_result, "report.html") + >>> if result.success: + ... print(f"Report saved to {result.output_path}") + """ + + def __init__( + self, + config: Optional[ReportConfig] = None, + ) -> None: + """Initialize report generator. + + Args: + config: Report configuration + """ + self.config = config or ReportConfig() + self.analyzer = ResultsAnalyzer() + + def generate_html( + self, + result: BacktestResult, + output_path: str | Path, + analysis: Optional[AnalysisResult] = None, + ) -> ReportResult: + """Generate HTML report. + + Args: + result: Backtest result to report on + output_path: Path to save HTML file + analysis: Pre-computed analysis (optional) + + Returns: + ReportResult with generation status + """ + start_time = datetime.now() + output_path = Path(output_path) + + try: + # Analyze results if not provided + if analysis is None: + analysis = self.analyzer.analyze(result) + + # Build content + content = self._build_content(result, analysis) + + # Generate HTML + html_content = self._render_html(content) + + # Write file + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text(html_content, encoding="utf-8") + + generation_time = int((datetime.now() - start_time).total_seconds() * 1000) + + return ReportResult( + success=True, + output_path=output_path, + format=ReportFormat.HTML, + file_size_bytes=output_path.stat().st_size, + generation_time_ms=generation_time, + ) + + except Exception as e: + logger.error(f"Failed to generate HTML report: {e}") + return ReportResult( + success=False, + format=ReportFormat.HTML, + error=str(e), + ) + + def generate_pdf( + self, + result: BacktestResult, + output_path: str | Path, + analysis: Optional[AnalysisResult] = None, + ) -> ReportResult: + """Generate PDF report. + + Note: Requires weasyprint or similar library for PDF generation. + Falls back to HTML if PDF libraries not available. + + Args: + result: Backtest result to report on + output_path: Path to save PDF file + analysis: Pre-computed analysis (optional) + + Returns: + ReportResult with generation status + """ + start_time = datetime.now() + output_path = Path(output_path) + + try: + # Analyze results if not provided + if analysis is None: + analysis = self.analyzer.analyze(result) + + # Build content + content = self._build_content(result, analysis) + + # Generate HTML first + html_content = self._render_html(content) + + # Try to convert to PDF + pdf_generated = self._html_to_pdf(html_content, output_path) + + if not pdf_generated: + # Fallback to HTML + html_path = output_path.with_suffix(".html") + html_path.write_text(html_content, encoding="utf-8") + + generation_time = int((datetime.now() - start_time).total_seconds() * 1000) + + return ReportResult( + success=True, + output_path=html_path, + format=ReportFormat.HTML, + file_size_bytes=html_path.stat().st_size, + generation_time_ms=generation_time, + warnings=["PDF generation not available, saved as HTML"], + ) + + generation_time = int((datetime.now() - start_time).total_seconds() * 1000) + + return ReportResult( + success=True, + output_path=output_path, + format=ReportFormat.PDF, + file_size_bytes=output_path.stat().st_size, + generation_time_ms=generation_time, + ) + + except Exception as e: + logger.error(f"Failed to generate PDF report: {e}") + return ReportResult( + success=False, + format=ReportFormat.PDF, + error=str(e), + ) + + def generate_json( + self, + result: BacktestResult, + output_path: str | Path, + analysis: Optional[AnalysisResult] = None, + ) -> ReportResult: + """Generate JSON report. + + Args: + result: Backtest result to report on + output_path: Path to save JSON file + analysis: Pre-computed analysis (optional) + + Returns: + ReportResult with generation status + """ + import json + + start_time = datetime.now() + output_path = Path(output_path) + + try: + # Analyze results if not provided + if analysis is None: + analysis = self.analyzer.analyze(result) + + # Build JSON structure + data = self._build_json_data(result, analysis) + + # Write file + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text( + json.dumps(data, indent=2, default=str), + encoding="utf-8", + ) + + generation_time = int((datetime.now() - start_time).total_seconds() * 1000) + + return ReportResult( + success=True, + output_path=output_path, + format=ReportFormat.JSON, + file_size_bytes=output_path.stat().st_size, + generation_time_ms=generation_time, + ) + + except Exception as e: + logger.error(f"Failed to generate JSON report: {e}") + return ReportResult( + success=False, + format=ReportFormat.JSON, + error=str(e), + ) + + def generate_markdown( + self, + result: BacktestResult, + output_path: str | Path, + analysis: Optional[AnalysisResult] = None, + ) -> ReportResult: + """Generate Markdown report. + + Args: + result: Backtest result to report on + output_path: Path to save Markdown file + analysis: Pre-computed analysis (optional) + + Returns: + ReportResult with generation status + """ + start_time = datetime.now() + output_path = Path(output_path) + + try: + # Analyze results if not provided + if analysis is None: + analysis = self.analyzer.analyze(result) + + # Build markdown content + md_content = self._build_markdown(result, analysis) + + # Write file + output_path.parent.mkdir(parents=True, exist_ok=True) + output_path.write_text(md_content, encoding="utf-8") + + generation_time = int((datetime.now() - start_time).total_seconds() * 1000) + + return ReportResult( + success=True, + output_path=output_path, + format=ReportFormat.MARKDOWN, + file_size_bytes=output_path.stat().st_size, + generation_time_ms=generation_time, + ) + + except Exception as e: + logger.error(f"Failed to generate Markdown report: {e}") + return ReportResult( + success=False, + format=ReportFormat.MARKDOWN, + error=str(e), + ) + + def generate( + self, + result: BacktestResult, + output_path: str | Path, + format: ReportFormat = ReportFormat.HTML, + analysis: Optional[AnalysisResult] = None, + ) -> ReportResult: + """Generate report in specified format. + + Args: + result: Backtest result to report on + output_path: Path to save report + format: Output format + analysis: Pre-computed analysis (optional) + + Returns: + ReportResult with generation status + """ + generators = { + ReportFormat.HTML: self.generate_html, + ReportFormat.PDF: self.generate_pdf, + ReportFormat.JSON: self.generate_json, + ReportFormat.MARKDOWN: self.generate_markdown, + } + + generator = generators.get(format, self.generate_html) + return generator(result, output_path, analysis) + + # ======================================================================== + # Content Building + # ======================================================================== + + def _build_content( + self, + result: BacktestResult, + analysis: AnalysisResult, + ) -> ReportContent: + """Build report content from backtest results.""" + content = ReportContent( + title=self.config.title, + generated_at=datetime.now(), + ) + + # Metadata + content.metadata = { + "start_date": result.start_date.isoformat() if result.start_date else None, + "end_date": result.end_date.isoformat() if result.end_date else None, + "initial_capital": str(result.initial_capital), + "final_capital": str(result.final_value), + "total_trades": result.total_trades, + "author": self.config.author, + } + + # Summary section + if ReportSection.SUMMARY in self.config.include_sections: + content.sections["summary"] = self._build_summary_section(result, analysis) + + # Statistics section + if ReportSection.STATISTICS in self.config.include_sections: + content.sections["statistics"] = self._build_statistics_section(analysis) + + # Risk metrics section + if ReportSection.RISK_METRICS in self.config.include_sections: + content.sections["risk_metrics"] = self._build_risk_section(analysis) + + # Trade list section + if ReportSection.TRADE_LIST in self.config.include_sections: + content.sections["trade_list"] = self._build_trade_list_section(result) + + # Monthly returns section + if ReportSection.MONTHLY_RETURNS in self.config.include_sections: + content.sections["monthly_returns"] = self._build_monthly_section(analysis) + + # Build charts + if ReportSection.EQUITY_CURVE in self.config.include_sections: + content.charts["equity_curve"] = self._build_equity_chart(result) + + if ReportSection.DRAWDOWN in self.config.include_sections: + content.charts["drawdown"] = self._build_drawdown_chart(result) + + return content + + def _build_summary_section( + self, + result: BacktestResult, + analysis: AnalysisResult, + ) -> dict[str, Any]: + """Build summary metrics section.""" + return { + "total_return": result.total_return, + "total_return_pct": result.total_return, + "final_capital": result.final_value, + "initial_capital": result.initial_capital, + "total_trades": result.total_trades, + "winning_trades": result.winning_trades, + "losing_trades": result.losing_trades, + "win_rate": result.win_rate, + "profit_factor": result.profit_factor, + "max_drawdown": result.max_drawdown, + "max_drawdown_pct": result.max_drawdown, + "sharpe_ratio": result.sharpe_ratio, + "sortino_ratio": result.sortino_ratio, + "total_commission": result.total_commission, + "total_slippage": result.total_slippage, + } + + def _build_statistics_section( + self, + analysis: AnalysisResult, + ) -> dict[str, Any]: + """Build trade statistics section.""" + stats = analysis.trade_statistics + return { + "total_trades": stats.total_trades, + "winning_trades": stats.winning_trades, + "losing_trades": stats.losing_trades, + "win_rate": stats.win_rate, + "profit_factor": stats.profit_factor, + "avg_win": stats.avg_win, + "avg_loss": stats.avg_loss, + "largest_win": stats.max_win, + "largest_loss": stats.max_loss, + "avg_trade_pnl": stats.avg_trade, + "median_trade_pnl": stats.median_trade, + "max_consecutive_wins": stats.max_consecutive_wins, + "max_consecutive_losses": stats.max_consecutive_losses, + "avg_holding_period_days": stats.avg_holding_period, + "expectancy": stats.expectancy, + } + + def _build_risk_section( + self, + analysis: AnalysisResult, + ) -> dict[str, Any]: + """Build risk metrics section.""" + risk = analysis.risk_metrics + return { + "sharpe_ratio": risk.sharpe_ratio, + "sortino_ratio": risk.sortino_ratio, + "calmar_ratio": risk.calmar_ratio, + "max_drawdown": risk.max_drawdown, + "max_drawdown_duration": risk.max_drawdown_duration, + "var_95": risk.var_95, + "cvar_95": risk.cvar_95, + "ulcer_index": risk.ulcer_index, + "recovery_factor": risk.recovery_factor, + } + + def _build_trade_list_section( + self, + result: BacktestResult, + ) -> list[dict[str, Any]]: + """Build trade list section.""" + trades = [] + for i, trade in enumerate(result.trades[:self.config.max_trades_shown]): + trades.append({ + "id": i + 1, + "timestamp": trade.timestamp.isoformat() if trade.timestamp else "", + "symbol": trade.symbol, + "side": trade.side.value if trade.side else "", + "quantity": str(trade.quantity), + "price": str(trade.price), + "value": str(trade.price * trade.quantity), + "commission": str(trade.commission), + "slippage": str(trade.slippage), + "pnl": str(trade.pnl), + }) + return trades + + def _build_monthly_section( + self, + analysis: AnalysisResult, + ) -> dict[str, Any]: + """Build monthly returns section.""" + monthly_data: dict[str, dict[str, Decimal]] = {} + + for breakdown in analysis.monthly_performance: + # Parse period string like "2023-01" into year and month + try: + parts = breakdown.period.split("-") + if len(parts) >= 2: + year = parts[0] + month = int(parts[1]) + else: + year = parts[0] + month = 0 + except (ValueError, AttributeError): + continue + + if year not in monthly_data: + monthly_data[year] = {} + + monthly_data[year][str(month)] = breakdown.return_pct + + return { + "monthly_returns": monthly_data, + "years": sorted(monthly_data.keys()), + } + + def _build_equity_chart( + self, + result: BacktestResult, + ) -> ChartData: + """Build equity curve chart data.""" + x_data = [] + y_data = [] + + for snapshot in result.snapshots: + x_data.append(snapshot.timestamp.isoformat() if snapshot.timestamp else "") + y_data.append(float(snapshot.total_value)) + + return ChartData( + chart_type=ChartType.LINE, + title="Equity Curve", + x_data=x_data, + y_data=y_data, + x_label="Date", + y_label="Portfolio Value", + colors=[self.config.color_scheme["primary"]], + ) + + def _build_drawdown_chart( + self, + result: BacktestResult, + ) -> ChartData: + """Build drawdown chart data.""" + x_data = [] + y_data = [] + + peak = ZERO + for snapshot in result.snapshots: + equity = snapshot.total_value + if equity > peak: + peak = equity + + drawdown = ((peak - equity) / peak * HUNDRED) if peak > ZERO else ZERO + + x_data.append(snapshot.timestamp.isoformat() if snapshot.timestamp else "") + y_data.append(float(drawdown)) + + return ChartData( + chart_type=ChartType.LINE, + title="Drawdown", + x_data=x_data, + y_data=y_data, + x_label="Date", + y_label="Drawdown %", + colors=[self.config.color_scheme["danger"]], + ) + + # ======================================================================== + # Rendering + # ======================================================================== + + def _render_html(self, content: ReportContent) -> str: + """Render content to HTML.""" + # Header + header_html = f""" +
+

{html.escape(content.title)}

+
+ Generated: {content.generated_at.strftime('%Y-%m-%d %H:%M:%S')} + {f' | Author: {html.escape(self.config.author)}' if self.config.author else ''} +
+
+ """ + + # Content sections + sections_html = [] + + # Summary section + if "summary" in content.sections: + sections_html.append(self._render_summary_html(content.sections["summary"])) + + # Equity curve chart + if "equity_curve" in content.charts: + sections_html.append(self._render_chart_html(content.charts["equity_curve"])) + + # Drawdown chart + if "drawdown" in content.charts: + sections_html.append(self._render_chart_html(content.charts["drawdown"])) + + # Statistics section + if "statistics" in content.sections: + sections_html.append(self._render_statistics_html(content.sections["statistics"])) + + # Risk metrics section + if "risk_metrics" in content.sections: + sections_html.append(self._render_risk_html(content.sections["risk_metrics"])) + + # Monthly returns section + if "monthly_returns" in content.sections: + sections_html.append(self._render_monthly_html(content.sections["monthly_returns"])) + + # Trade list section + if "trade_list" in content.sections: + sections_html.append(self._render_trade_list_html(content.sections["trade_list"])) + + # Footer + footer_html = f""" +
+ {html.escape(self.config.footer_text)} +
+ """ + + # Render full HTML + return HTML_TEMPLATE.format( + title=html.escape(content.title), + color_primary=self.config.color_scheme["primary"], + color_secondary=self.config.color_scheme["secondary"], + color_success=self.config.color_scheme["success"], + color_danger=self.config.color_scheme["danger"], + color_warning=self.config.color_scheme["warning"], + color_background=self.config.color_scheme["background"], + color_text=self.config.color_scheme["text"], + color_border=self.config.color_scheme["border"], + header=header_html, + content="\n".join(sections_html), + footer=footer_html, + ) + + def _render_summary_html(self, data: dict[str, Any]) -> str: + """Render summary section to HTML.""" + dp = self.config.decimal_places + + def format_value(val: Any, is_pct: bool = False) -> str: + if isinstance(val, Decimal): + formatted = f"{float(val):,.{dp}f}" + elif isinstance(val, (int, float)): + formatted = f"{val:,.{dp}f}" + else: + formatted = str(val) + return f"{formatted}%" if is_pct else formatted + + def value_class(val: Any) -> str: + try: + return "positive" if float(val) > 0 else "negative" if float(val) < 0 else "" + except (ValueError, TypeError): + return "" + + metrics = [ + ("Total Return", data.get("total_return_pct", 0), True, True), + ("Final Capital", data.get("final_capital", 0), False, False), + ("Total Trades", data.get("total_trades", 0), False, False), + ("Win Rate", data.get("win_rate", 0), True, False), + ("Profit Factor", data.get("profit_factor", 0), False, False), + ("Max Drawdown", data.get("max_drawdown_pct", 0), True, True), + ("Sharpe Ratio", data.get("sharpe_ratio", 0), False, False), + ("Sortino Ratio", data.get("sortino_ratio", 0), False, False), + ] + + cards_html = [] + for label, value, is_pct, has_sign in metrics: + val_class = value_class(value) if has_sign else "" + cards_html.append(f""" +
+
{format_value(value, is_pct)}
+
{html.escape(label)}
+
+ """) + + return f""" +
+

Summary

+
+ {''.join(cards_html)} +
+
+ """ + + def _render_statistics_html(self, data: dict[str, Any]) -> str: + """Render trade statistics section to HTML.""" + dp = self.config.decimal_places + + rows = [ + ("Total Trades", data.get("total_trades", 0)), + ("Winning Trades", data.get("winning_trades", 0)), + ("Losing Trades", data.get("losing_trades", 0)), + ("Win Rate", f"{float(data.get('win_rate', 0)):,.{dp}f}%"), + ("Profit Factor", f"{float(data.get('profit_factor', 0)):,.{dp}f}"), + ("Average Win", f"${float(data.get('avg_win', 0)):,.{dp}f}"), + ("Average Loss", f"${float(data.get('avg_loss', 0)):,.{dp}f}"), + ("Largest Win", f"${float(data.get('largest_win', 0)):,.{dp}f}"), + ("Largest Loss", f"${float(data.get('largest_loss', 0)):,.{dp}f}"), + ("Avg Trade P&L", f"${float(data.get('avg_trade_pnl', 0)):,.{dp}f}"), + ("Max Consecutive Wins", data.get("max_consecutive_wins", 0)), + ("Max Consecutive Losses", data.get("max_consecutive_losses", 0)), + ("Avg Holding Period", f"{float(data.get('avg_holding_period_days', 0)):,.1f} days"), + ("Expectancy", f"${float(data.get('expectancy', 0)):,.{dp}f}"), + ] + + rows_html = "\n".join([ + f"{html.escape(str(label))}{html.escape(str(value))}" + for label, value in rows + ]) + + return f""" +
+

Trade Statistics

+ + + + + + {rows_html} + +
MetricValue
+
+ """ + + def _render_risk_html(self, data: dict[str, Any]) -> str: + """Render risk metrics section to HTML.""" + dp = self.config.decimal_places + + rows = [ + ("Sharpe Ratio", f"{float(data.get('sharpe_ratio', 0)):,.{dp}f}"), + ("Sortino Ratio", f"{float(data.get('sortino_ratio', 0)):,.{dp}f}"), + ("Calmar Ratio", f"{float(data.get('calmar_ratio', 0)):,.{dp}f}"), + ("Max Drawdown %", f"{float(data.get('max_drawdown', 0)):,.{dp}f}%"), + ("Max Drawdown Duration", f"{data.get('max_drawdown_duration', 0)} days"), + ("VaR (95%)", f"${float(data.get('var_95', 0)):,.{dp}f}"), + ("CVaR (95%)", f"${float(data.get('cvar_95', 0)):,.{dp}f}"), + ("Ulcer Index", f"{float(data.get('ulcer_index', 0)):,.{dp}f}"), + ("Recovery Factor", f"{float(data.get('recovery_factor', 0)):,.{dp}f}"), + ] + + rows_html = "\n".join([ + f"{html.escape(str(label))}{html.escape(str(value))}" + for label, value in rows + ]) + + return f""" +
+

Risk Metrics

+ + + + + + {rows_html} + +
MetricValue
+
+ """ + + def _render_monthly_html(self, data: dict[str, Any]) -> str: + """Render monthly returns heatmap to HTML.""" + monthly_returns = data.get("monthly_returns", {}) + years = data.get("years", []) + + months = ["Jan", "Feb", "Mar", "Apr", "May", "Jun", + "Jul", "Aug", "Sep", "Oct", "Nov", "Dec"] + + # Header row + header_cells = ['
Year
'] + for month in months: + header_cells.append(f'
{month}
') + + # Data rows + data_rows = [] + for year in years: + row_cells = [f'
{year}
'] + year_data = monthly_returns.get(year, {}) + + for month_num in range(1, 13): + value = year_data.get(str(month_num), Decimal("0")) + val_float = float(value) + + # Color based on value + if val_float > 5: + color = "#15803d" # Dark green + elif val_float > 2: + color = "#22c55e" # Green + elif val_float > 0: + color = "#86efac" # Light green + elif val_float > -2: + color = "#fca5a5" # Light red + elif val_float > -5: + color = "#ef4444" # Red + else: + color = "#b91c1c" # Dark red + + cell_text = f"{val_float:+.1f}%" if value != ZERO else "-" + row_cells.append( + f'
{cell_text}
' + ) + + data_rows.append("".join(row_cells)) + + return f""" +
+

Monthly Returns

+
+ {''.join(header_cells)} + {''.join(data_rows)} +
+
+ """ + + def _render_trade_list_html(self, trades: list[dict[str, Any]]) -> str: + """Render trade list section to HTML.""" + if not trades: + return """ +
+

Trade List

+

No trades executed.

+
+ """ + + rows_html = [] + for trade in trades: + pnl = float(trade.get("pnl", 0)) + pnl_class = "positive" if pnl > 0 else "negative" if pnl < 0 else "" + + rows_html.append(f""" + + {html.escape(str(trade.get('id', '')))} + {html.escape(str(trade.get('timestamp', '')))} + {html.escape(str(trade.get('symbol', '')))} + {html.escape(str(trade.get('side', '')))} + {html.escape(str(trade.get('quantity', '')))} + ${float(trade.get('price', 0)):,.2f} + ${float(trade.get('value', 0)):,.2f} + ${pnl:,.2f} + + """) + + return f""" +
+

Trade List

+ + + + + + + + + + + + + + + {''.join(rows_html)} + +
#TimestampSymbolSideQuantityPriceValueP&L
+
+ """ + + def _render_chart_html(self, chart: ChartData) -> str: + """Render chart to HTML with SVG.""" + if not chart.x_data or not chart.y_data: + return f""" +
+

{html.escape(chart.title)}

+
+
No data available
+
+
+ """ + + # Generate simple SVG chart + svg = self._generate_svg_chart(chart) + + return f""" +
+

{html.escape(chart.title)}

+
+ {svg} +
+
+ """ + + def _generate_svg_chart(self, chart: ChartData) -> str: + """Generate SVG chart from chart data.""" + width = self.config.chart_width + height = self.config.chart_height + padding = 60 + + # Calculate data bounds + y_values = [float(y) for y in chart.y_data if y is not None] + if not y_values: + return '
No data available
' + + y_min = min(y_values) + y_max = max(y_values) + y_range = y_max - y_min if y_max != y_min else 1 + + # Scale data to SVG coordinates + plot_width = width - 2 * padding + plot_height = height - 2 * padding + + points = [] + for i, y in enumerate(y_values): + x = padding + (i / max(len(y_values) - 1, 1)) * plot_width + y_scaled = height - padding - ((y - y_min) / y_range) * plot_height + points.append(f"{x:.1f},{y_scaled:.1f}") + + polyline_points = " ".join(points) + color = chart.colors[0] if chart.colors else self.config.color_scheme["primary"] + + # Generate SVG + svg = f""" + + + + + + + + + + + + + {y_max:,.0f} + {y_min:,.0f} + + + + + + + + + + + {html.escape(chart.x_label)} + + + {html.escape(chart.y_label)} + + + """ + + return svg + + # ======================================================================== + # PDF Generation + # ======================================================================== + + def _html_to_pdf(self, html_content: str, output_path: Path) -> bool: + """Convert HTML to PDF. + + Returns True if successful, False if PDF libraries not available. + """ + try: + # Try weasyprint first + from weasyprint import HTML + HTML(string=html_content).write_pdf(output_path) + return True + except ImportError: + pass + + try: + # Try pdfkit as fallback + import pdfkit + pdfkit.from_string(html_content, str(output_path)) + return True + except ImportError: + pass + + logger.warning("No PDF library available (weasyprint or pdfkit)") + return False + + # ======================================================================== + # JSON Generation + # ======================================================================== + + def _build_json_data( + self, + result: BacktestResult, + analysis: AnalysisResult, + ) -> dict[str, Any]: + """Build JSON data structure from backtest results.""" + return { + "report": { + "title": self.config.title, + "generated_at": datetime.now().isoformat(), + "author": self.config.author, + }, + "backtest": { + "start_date": result.start_date.isoformat() if result.start_date else None, + "end_date": result.end_date.isoformat() if result.end_date else None, + "initial_capital": str(result.initial_capital), + "final_capital": str(result.final_value), + }, + "performance": { + "total_return": str(result.total_return), + "total_return_pct": str(result.total_return), + "win_rate": str(result.win_rate), + "profit_factor": str(result.profit_factor), + "max_drawdown": str(result.max_drawdown), + "max_drawdown_pct": str(result.max_drawdown), + "sharpe_ratio": str(result.sharpe_ratio), + "sortino_ratio": str(result.sortino_ratio), + }, + "trades": { + "total": result.total_trades, + "winning": result.winning_trades, + "losing": result.losing_trades, + "total_commission": str(result.total_commission), + "total_slippage": str(result.total_slippage), + }, + "trade_statistics": { + "avg_win": str(analysis.trade_statistics.avg_win), + "avg_loss": str(analysis.trade_statistics.avg_loss), + "largest_win": str(analysis.trade_statistics.max_win), + "largest_loss": str(analysis.trade_statistics.max_loss), + "expectancy": str(analysis.trade_statistics.expectancy), + "max_consecutive_wins": analysis.trade_statistics.max_consecutive_wins, + "max_consecutive_losses": analysis.trade_statistics.max_consecutive_losses, + }, + "risk_metrics": { + "sharpe_ratio": str(analysis.risk_metrics.sharpe_ratio), + "sortino_ratio": str(analysis.risk_metrics.sortino_ratio), + "calmar_ratio": str(analysis.risk_metrics.calmar_ratio), + "max_drawdown": str(analysis.risk_metrics.max_drawdown), + "var_95": str(analysis.risk_metrics.var_95), + "cvar_95": str(analysis.risk_metrics.cvar_95), + "ulcer_index": str(analysis.risk_metrics.ulcer_index), + }, + "equity_curve": [ + { + "timestamp": s.timestamp.isoformat() if s.timestamp else None, + "equity": str(s.total_value), + } + for s in result.snapshots + ], + "trade_list": [ + { + "timestamp": t.timestamp.isoformat() if t.timestamp else None, + "symbol": t.symbol, + "side": t.side.value if t.side else None, + "quantity": str(t.quantity), + "price": str(t.price), + "pnl": str(t.pnl), + } + for t in result.trades[:self.config.max_trades_shown] + ], + } + + # ======================================================================== + # Markdown Generation + # ======================================================================== + + def _build_markdown( + self, + result: BacktestResult, + analysis: AnalysisResult, + ) -> str: + """Build Markdown report from backtest results.""" + dp = self.config.decimal_places + lines = [] + + # Title + lines.append(f"# {self.config.title}") + lines.append("") + lines.append(f"**Generated:** {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") + if self.config.author: + lines.append(f"**Author:** {self.config.author}") + lines.append("") + + # Summary + lines.append("## Summary") + lines.append("") + lines.append(f"| Metric | Value |") + lines.append(f"|--------|-------|") + lines.append(f"| Initial Capital | ${float(result.initial_capital):,.{dp}f} |") + lines.append(f"| Final Capital | ${float(result.final_value):,.{dp}f} |") + lines.append(f"| Total Return | {float(result.total_return):,.{dp}f}% |") + lines.append(f"| Total Trades | {result.total_trades} |") + lines.append(f"| Win Rate | {float(result.win_rate):,.{dp}f}% |") + lines.append(f"| Profit Factor | {float(result.profit_factor):,.{dp}f} |") + lines.append(f"| Max Drawdown | {float(result.max_drawdown):,.{dp}f}% |") + lines.append(f"| Sharpe Ratio | {float(result.sharpe_ratio):,.{dp}f} |") + lines.append("") + + # Trade Statistics + lines.append("## Trade Statistics") + lines.append("") + stats = analysis.trade_statistics + lines.append(f"| Metric | Value |") + lines.append(f"|--------|-------|") + lines.append(f"| Winning Trades | {stats.winning_trades} |") + lines.append(f"| Losing Trades | {stats.losing_trades} |") + lines.append(f"| Average Win | ${float(stats.avg_win):,.{dp}f} |") + lines.append(f"| Average Loss | ${float(stats.avg_loss):,.{dp}f} |") + lines.append(f"| Largest Win | ${float(stats.max_win):,.{dp}f} |") + lines.append(f"| Largest Loss | ${float(stats.max_loss):,.{dp}f} |") + lines.append(f"| Max Consecutive Wins | {stats.max_consecutive_wins} |") + lines.append(f"| Max Consecutive Losses | {stats.max_consecutive_losses} |") + lines.append(f"| Expectancy | ${float(stats.expectancy):,.{dp}f} |") + lines.append("") + + # Risk Metrics + lines.append("## Risk Metrics") + lines.append("") + risk = analysis.risk_metrics + lines.append(f"| Metric | Value |") + lines.append(f"|--------|-------|") + lines.append(f"| Sharpe Ratio | {float(risk.sharpe_ratio):,.{dp}f} |") + lines.append(f"| Sortino Ratio | {float(risk.sortino_ratio):,.{dp}f} |") + lines.append(f"| Calmar Ratio | {float(risk.calmar_ratio):,.{dp}f} |") + lines.append(f"| Max Drawdown | {float(risk.max_drawdown):,.{dp}f}% |") + lines.append(f"| VaR (95%) | ${float(risk.var_95):,.{dp}f} |") + lines.append(f"| CVaR (95%) | ${float(risk.cvar_95):,.{dp}f} |") + lines.append(f"| Ulcer Index | {float(risk.ulcer_index):,.{dp}f} |") + lines.append("") + + # Recent Trades + if result.trades: + lines.append("## Recent Trades") + lines.append("") + lines.append("| # | Symbol | Side | Quantity | Price | P&L |") + lines.append("|---|--------|------|----------|-------|-----|") + + for i, trade in enumerate(result.trades[:20]): + pnl = float(trade.pnl) + pnl_str = f"${pnl:,.{dp}f}" if pnl >= 0 else f"-${abs(pnl):,.{dp}f}" + lines.append( + f"| {i + 1} | {trade.symbol} | {trade.side.value if trade.side else ''} | " + f"{float(trade.quantity):,.2f} | ${float(trade.price):,.{dp}f} | {pnl_str} |" + ) + lines.append("") + + # Footer + lines.append("---") + lines.append(f"*{self.config.footer_text}*") + + return "\n".join(lines) + + +# ============================================================================ +# Factory Functions +# ============================================================================ + +def create_report_generator( + title: str = "Backtest Report", + author: str = "", + include_trade_list: bool = True, + max_trades: int = 100, +) -> ReportGenerator: + """Create a configured report generator. + + Args: + title: Report title + author: Author name + include_trade_list: Whether to include trade list + max_trades: Maximum trades to show + + Returns: + Configured ReportGenerator + """ + config = ReportConfig( + title=title, + author=author, + include_trade_list=include_trade_list, + max_trades_shown=max_trades, + ) + return ReportGenerator(config=config)