Coverage for src / monte_neo / monte_carlo / sequential.py: 97%
87 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-28 16:27 +0200
1"""Sequential Monte Carlo execution module."""
3from __future__ import annotations
5import time
6from typing import TYPE_CHECKING
8import pandas as pd
9import questionary
10from rich.console import Console
11from rich.panel import Panel
12from rich.table import Table
14from monte_neo.monte_carlo.types import MCResult, MCStepResult
16if TYPE_CHECKING:
17 from monte_neo.indicators.base import BaseIndicator
18 from monte_neo.metrics.calculator import MetricsCalculator
19 from monte_neo.monte_carlo.engine import MonteCarloEngine
22from monte_neo.cli.styles import CUSTOM_STYLE
24console = Console()
27class SequentialMCRunner:
28 """Runner for sequential Monte Carlo methods."""
30 def __init__(self, engine: MonteCarloEngine):
31 """Initialize runner.
33 Args:
34 engine: Base MonteCarloEngine.
35 """
36 self.engine = engine
38 def run(
39 self,
40 data: pd.DataFrame,
41 indicator: BaseIndicator,
42 metrics_calc: MetricsCalculator,
43 target_metrics: dict[str, float],
44 interactive: bool = True,
45 ) -> MCResult:
46 """Run MC methods sequentially.
48 Args:
49 data: OHLCV data.
50 indicator: Indicator to test.
51 metrics_calc: Metrics calculator.
52 target_metrics: Target metrics.
53 interactive: Whether to ask for confirmation before each step.
55 Returns:
56 MCResult with sequential results.
57 """
58 start_time = time.time()
59 step_results = []
60 all_passed = True
62 # Define methods in requested order
63 methods = [
64 ("Walk-Forward Analysis", "walk_forward"),
65 ("Block Bootstrap", "block_bootstrap"),
66 ("Return Shuffling", "shuffling"),
67 ("Noise Injection", "noise"),
68 ("Sensitivity Analysis", "sensitivity"),
69 ]
71 enabled_methods = [
72 (display, key) for (display, key) in methods
73 if getattr(self.engine.config, f"use_{key}")
74 ]
76 # Educational descriptions for each method
77 descriptions = {
78 "walk_forward": "Validates strategy performance on 'future' data not used during training. Helps detect overfitting.",
79 "block_bootstrap": "Creates new market scenarios by shuffling historical data blocks. Tests strategy resilience to market regime changes.",
80 "shuffling": "Shuffles the sequence of returns, destroying temporal structure. If a strategy relies on real patterns, performance should degrade on shuffled data.",
81 "noise": "Adds random noise to OHLC prices. Tests strategy sensitivity to minor price changes and volatility.",
82 "sensitivity": "Varies indicator parameters within a small range (e.g., ±10%). A robust strategy should not break with small setting changes."
83 }
85 # Use rich table for sequential output if it's the main display
86 console.print(f"\n[bold yellow]🔍 Sequential MC Validation for: {indicator.name}[/]")
88 if not interactive:
89 console.print(f"[dim]Running in automated mode. All {len(enabled_methods)} steps will be executed.[/]")
91 table = Table(title="Monte Carlo Steps", show_header=True, header_style="bold magenta")
92 table.add_column("Step", justify="right")
93 table.add_column("Method", style="cyan")
94 table.add_column("Pass Rate", justify="right")
95 table.add_column("Status", justify="center")
97 total_steps = len(enabled_methods)
98 for idx, (display_name, method_key) in enumerate(enabled_methods, start=1):
99 console.print(f"\n[bold magenta]👉 Stage {idx}/{total_steps}: {display_name}[/]")
100 console.print(Panel(descriptions.get(method_key, ""), title="Educational Info", border_style="blue"))
102 # Run method
103 step_result = self._run_step(
104 display_name, method_key, data, indicator, metrics_calc, target_metrics
105 )
106 step_results.append(step_result)
108 # Update output
109 status = "[green]PASSED[/]" if step_result.passed else "[red]FAILED[/]"
110 table.add_row(f"{idx}/{total_steps}", display_name, f"{step_result.pass_rate:.1%}", status)
112 # Print current state
113 if interactive:
114 console.clear()
115 console.print(f"\n[bold yellow]🔍 Sequential MC Validation for: {indicator.name}[/]")
116 console.print(table)
118 # Print Detailed Advice for the current step
119 console.print(f"\n[bold cyan]💡 Analysis & Advice for {display_name}:[/]")
120 console.print(f"[italic]{step_result.advice}[/]")
122 # Show key metrics for this step
123 if step_result.metrics_summary:
124 pf = step_result.metrics_summary.get("profit_factor", {}).get("mean", 0.0)
125 sr = step_result.metrics_summary.get("sharpe_ratio", {}).get("mean", 0.0)
126 dd = step_result.metrics_summary.get("max_drawdown", {}).get("mean", 0.0)
127 console.print(f"[dim]Stats: PF={pf:.2f}, Sharpe={sr:.2f}, DD={dd:.1%}[/]")
128 console.print(
129 f"[dim]Pass Rate: {step_result.pass_rate:.1%} | "
130 f"Threshold: {self.engine.config.pass_threshold:.1%} | "
131 f"Iterations: {step_result.iterations}[/]"
132 )
134 if not step_result.passed:
135 all_passed = False
136 console.print(f"\n[bold red]❌ FAILED: {display_name} did not meet robustness criteria.[/]")
137 console.print("[red]To reach Production-Ready status, the indicator must pass all stages.[/]")
138 console.print("[red]Review the advice above and adjust your strategy parameters or logic.[/]")
140 if interactive:
141 if not questionary.confirm("Continue to next stage anyway (not recommended)?", default=False, style=CUSTOM_STYLE).ask():
142 break
143 # В неинтерактивном режиме продолжаем выполнение всех шагов для полного анализа
144 else:
145 console.print(f"\n[bold green]✅ STAGE PASSED: {display_name}[/]")
146 if interactive and idx < total_steps:
147 questionary.press_any_key_to_continue("Press any key to proceed to next stage...", style=CUSTOM_STYLE).ask()
149 elapsed = time.time() - start_time
150 total_enabled = max(1, len(enabled_methods))
151 total_executed = len(step_results)
152 pass_rate = (len([r for r in step_results if r.passed]) / total_executed) if total_executed else 0.0
153 total_iterations = sum(r.iterations for r in step_results)
155 # Populate high-level metrics summary from all steps
156 metrics_summary = {}
157 if step_results:
158 # For simplicity, use metrics from the last step or combine them
159 # Here we'll just take the last step's summary as the overall summary
160 metrics_summary = step_results[-1].metrics_summary
162 return MCResult(
163 passed=all_passed and total_executed > 0,
164 pass_rate=pass_rate,
165 iterations_run=total_iterations,
166 elapsed_time=elapsed,
167 metrics_summary=metrics_summary,
168 step_results=step_results,
169 )
171 def _run_step(
172 self,
173 name: str,
174 key: str,
175 data: pd.DataFrame,
176 indicator: BaseIndicator,
177 metrics_calc: MetricsCalculator,
178 target_metrics: dict[str, float],
179 ) -> MCStepResult:
180 """Run a single MC step."""
181 scenarios = []
182 iterations = self.engine.config.iterations
184 if key == "walk_forward":
185 scenarios = self.engine.scenario_builder.generate_walk_forward(data)
186 elif key == "block_bootstrap":
187 scenarios = self.engine.scenario_builder.generate_block_bootstrap(data, iterations)
188 elif key == "shuffling":
189 scenarios = self.engine.scenario_builder.generate_shuffling(data, iterations)
190 elif key == "noise":
191 scenarios = self.engine.scenario_builder.generate_noise(data, iterations)
192 elif key == "sensitivity":
193 # Sensitivity is special as it varies parameters, not data
194 return self._run_sensitivity_step(indicator, data, metrics_calc, target_metrics)
196 # Run backtests for scenarios
197 results = self.engine.gpu_engine.backtest_scenarios(
198 indicator,
199 scenarios,
200 executor=self.engine.executor,
201 use_sl_tp=self.engine.config.use_sl_tp,
202 sl_pct=self.engine.config.sl_pct,
203 tp_pct=self.engine.config.tp_pct,
204 )
206 passed_count = 0
207 for r in results:
208 metrics = r.get("metrics", {})
209 passed = True
210 for metric_name, target_value in target_metrics.items():
211 if metric_name not in metrics:
212 continue
213 actual = metrics[metric_name]
214 if metric_name in ["max_drawdown", "consecutive_losses"]:
215 if actual > target_value:
216 passed = False
217 break
218 else:
219 if actual < target_value:
220 passed = False
221 break
222 if passed:
223 passed_count += 1
224 pass_rate = passed_count / len(results) if results else 0.0
225 passed = pass_rate >= self.engine.config.pass_threshold
227 from monte_neo.monte_carlo.utils import summarize_metrics
228 summary = summarize_metrics(results)
229 advice = self._generate_advice(name, pass_rate, summary)
231 return MCStepResult(
232 method_name=name,
233 passed=passed,
234 pass_rate=pass_rate,
235 metrics_summary=summary,
236 advice=advice,
237 iterations=len(results),
238 )
240 def _run_sensitivity_step(
241 self,
242 indicator: BaseIndicator,
243 data: pd.DataFrame,
244 metrics_calc: MetricsCalculator,
245 target_metrics: dict[str, float],
246 ) -> MCStepResult:
247 """Run sensitivity analysis step."""
248 from monte_neo.monte_carlo.sensitivity import SensitivityAnalyzer
249 analyzer = SensitivityAnalyzer(variation_range=self.engine.config.sensitivity_range)
250 results = analyzer.analyze_all_parameters(indicator, data, metrics_calc)
251 report = analyzer.get_stability_report(results)
253 pass_rate = report["average_stability"]
254 passed = report["overall_stable"]
256 summary = {
257 "stability_score": {"mean": pass_rate},
258 "stable_params": {"mean": report["stable_parameters"]},
259 "total_params": {"mean": report["total_parameters"]},
260 }
262 advice = self._generate_advice("Sensitivity Analysis", pass_rate, summary)
264 return MCStepResult(
265 method_name="Sensitivity Analysis",
266 passed=passed,
267 pass_rate=pass_rate,
268 metrics_summary=summary,
269 advice=advice,
270 iterations=len(results),
271 )
273 def _generate_advice(self, method_name: str, pass_rate: float, summary: dict) -> str:
274 """Generate advice based on results."""
275 name = method_name.lower()
276 if pass_rate >= 0.95:
277 if "walk" in name:
278 return "Excellent stability over time. The strategy adapts well to different market regimes."
279 if "block" in name or "bootstrap" in name:
280 return "High statistical significance. The edge is likely not due to random price sequences."
281 if "shuffling" in name:
282 return "The strategy captures real market structure, not just random price distributions (Passed)."
283 if "noise" in name:
284 return "Robust against price execution noise and minor volatility spikes."
285 if "sensitivity" in name:
286 return "Parameters are well-tuned and stable. Not over-optimized for specific values."
287 return "Strategy passed this stage with high confidence."
289 if pass_rate >= 0.80:
290 return f"Strategy is mostly stable but shows some weakness in {method_name}. Consider slight adjustments."
292 if "walk" in name:
293 return "Strategy fails to maintain performance across different time periods. Risk of over-fitting to specific dates."
294 if "block" in name or "bootstrap" in name:
295 return "Low statistical significance. The strategy might be capturing noise or specific patterns that don't repeat."
296 if "shuffling" in name:
297 return "Performance is similar to random entry (Failed). The 'edge' might be an illusion of price distribution."
298 if "noise" in name:
299 return "Strategy is very sensitive to price noise. Might fail in real-market execution with slippage."
300 if "sensitivity" in name:
301 return "High sensitivity to parameter changes. Likely over-optimized (curve-fitted)."
303 return f"Strategy failed to meet robustness criteria in {method_name} (pass rate: {pass_rate:.1%})."