""" optimizer/pipeline.py The Smart 3-Phase Optimization Pipeline. Replaces optimizer_loop.py as the core orchestration engine. Receives a SessionConfig → runs Phase 1, 2, 3 → emits SocketIO events. Architecture: Phase 1: Broad Discovery (LHS samples, 20–30 runs) Phase 2: Deep Refinement (neighbor search around top 3) Phase 3: Validation (OOS backtest + sensitivity) Decision: RECOMMENDED / RISKY / NOT_RELIABLE + .set file download """ from __future__ import annotations import hashlib import os import random import time from datetime import datetime from pathlib import Path from typing import Optional import yaml from loguru import logger from ea.registry import EARegistry from ea.schema import ParameterSchema from mt5.ini_builder import IniBuilder from mt5.runner import MT5Runner from mt5.report_parser import ReportParser from data.models import Run from data.store import DataStore from reports.writer import ReportWriter from optimizer.session_config import SessionConfig from optimizer.lhs_sampler import LatinHypercubeSampler from optimizer.result_ranker import ResultRanker, RankedResult from optimizer.budget import BudgetManager import pandas as pd from analysis.equity_curve import EquityCurveAnalyzer from analysis.time_performance import TimePerformanceAnalyzer from analysis.ai_reasoner import AIReasoner from analysis.ai_reasoner_config import load_api_key BASE_DIR = Path(__file__).parent.parent RUNS_DIR = BASE_DIR / "runs" DB_PATH = BASE_DIR / "optimizer.db" class OptimizationPipeline: """ 3-phase autonomous optimization pipeline. Run in a background thread. Usage: pipeline = OptimizationPipeline("config.yaml", socketio, reports_dir) pipeline.configure(session_config) thread = threading.Thread(target=pipeline.run, daemon=True) thread.start() """ def __init__(self, config_path: str, socketio, reports_dir: Path): self.config_path = config_path self.socketio = socketio self.reports_dir = reports_dir with open(config_path) as f: self.cfg = yaml.safe_load(f) # Runtime state self.session: Optional[SessionConfig] = None self.running = False self._stop_flag = False self._phase = "idle" self._run_count = 0 self._total_runs = 0 self.best_result: Optional[RankedResult] = None self.phase1_results: list[RankedResult] = [] self.phase2_results: list[RankedResult] = [] self.final_result: Optional[RankedResult] = None self.verdict: Optional[str] = None self.best_set_path: Optional[Path] = None self.run_start_ts: Optional[float] = None # AI & analysis state self._ai_reasoner: Optional[AIReasoner] = None self._ai_insights: list[dict] = [] self._run_findings: dict[str, list] = {} self._run_insights: dict[str, dict] = {} self._baseline_metrics: Optional[dict] = None # Running best (updated each run so status can serve it live) self._live_best: Optional[RankedResult] = None # All completed runs (for history restoration) self._completed_runs: list[dict] = [] # ── Public API ──────────────────────────────────────────────────────────── def configure(self, session: SessionConfig) -> None: self.session = session def stop(self) -> None: self._stop_flag = True self._emit("status_change", {"state": "stopping"}) self._emit_early_termination( reason_code="user_stop", message="User requested stop. Finishing current run and exiting.", details={"phase": self._phase}, ) def get_status(self) -> dict: elapsed = int(time.time() - self.run_start_ts) if self.run_start_ts else 0 # Use final_result if set, otherwise best seen so far during the run best = self.final_result or self._live_best return { "state": "running" if self.running else "idle", "phase": self._phase, "run_count": self._run_count, "total_runs": self._total_runs, "best_score": round(best.score, 4) if best else 0.0, "verdict": self.verdict, "elapsed_s": elapsed, "ea_name": self.session.ea_name if self.session else "", "symbol": self.session.symbol if self.session else "", "timeframe": self.session.timeframe if self.session else "", "has_insight": bool(self._ai_insights), "insight_count": len(self._ai_insights), "latest_insight": self._ai_insights[-1] if self._ai_insights else None, "best_net_profit": round(best.net_profit, 2) if best else None, "best_calmar": round(best.calmar, 3) if best else None, "best_pf": round(best.profit_factor, 3) if best else None, "best_win_rate": round(best.win_rate * 100, 1) if best else None, "best_max_drawdown": round(best.max_drawdown * 100, 2) if best else None, "best_total_trades": best.total_trades if best else None, } # ── Main entry point ────────────────────────────────────────────────────── def run(self) -> None: self.running = True self._stop_flag = False self.run_start_ts = time.time() try: self._run_pipeline() except Exception as e: logger.exception(f"Pipeline crashed: {e}") self._emit("error", {"msg": f"Pipeline error: {e}"}) finally: self.running = False self._phase = "idle" self._emit("status_change", {"state": "idle"}) def _run_pipeline(self) -> None: cfg = self.session self._total_runs = cfg.total_budget_runs self._emit("status_change", {"state": "running", "phase": "setup"}) self._log("info", f"🚀 Smart Optimizer started — {cfg.ea_name} | {cfg.symbol} | {cfg.timeframe}") self._log("info", f"📋 Budget: {cfg.budget_minutes} min | Samples: {cfg.phase1_samples} Phase1 + {cfg.phase2_samples} Phase2 + {cfg.phase3_samples} Phase3") # ── Build components ───────────────────────────────────────────────── reg = EARegistry(self.config_path) profile = reg.get(cfg.ea_name) schema = reg.get_schema(profile) # Apply user's param selection if specified if cfg.selected_params: for p in schema.parameters.values(): if p.type != "fixed": p.optimize = p.name in cfg.selected_params builder = IniBuilder(self.config_path, schema=schema) runner = MT5Runner(self.config_path) parser = ReportParser() store = DataStore(DB_PATH, RUNS_DIR) writer = ReportWriter(self.reports_dir) sampler = LatinHypercubeSampler(seed=int(time.time())) ranker = ResultRanker(weights=cfg.scoring_weights) budget = BudgetManager(cfg.budget_minutes) # Initialize AI reasoning layer api_key = load_api_key(self.config_path) self._ai_reasoner = AIReasoner(api_key=api_key) self._ai_insights.clear() self._run_findings.clear() self._run_insights.clear() budget.start() # ── Phase 1: Broad Discovery ───────────────────────────────────────── if self._stop_flag: return self._phase = "phase1" self._emit("phase_start", {"phase": "phase1", "total": cfg.phase1_samples}) self._emit_thinking( f"Starting broad discovery with {cfg.phase1_samples} Latin-Hypercube samples. " f"Scanning parameter space to find profitable regions before focused refinement.", kind="reasoning", ) self._log("info", f"━━ Phase 1: Broad Discovery ({cfg.phase1_samples} configurations) ━━") samples = sampler.sample(schema, cfg.phase1_samples) phase1_raw: list[RankedResult] = [] for i, params in enumerate(samples): if self._stop_flag: break run_id = f"p1_{i+1:02d}_{datetime.utcnow().strftime('%H%M%S')}" self._log("info", f"[{i+1}/{cfg.phase1_samples}] Testing configuration {i+1}...") t0 = time.time() result = self._execute_run( run_id, params, cfg.train_start, cfg.train_end, "phase1", builder, runner, parser, store, writer, ranker, profile ) budget.record_run(time.time() - t0) phase1_raw.append(result) self._run_count += 1 # Track live best and completed runs for status/history APIs if self._live_best is None or result.score > self._live_best.score: self._live_best = result run_dict = self._make_run_dict(run_id, result, "phase1") self._completed_runs.append(run_dict) # Emit progress after each run self._emit("run_complete", { **run_dict, "run_number": i + 1, "total": cfg.phase1_samples, "progress_pct": round((i + 1) / cfg.phase1_samples * 100), "budget_summary": budget.summary(), }) if budget.is_exhausted(): self._log("warning", "⏱ Time budget reached during Phase 1") break # Rank Phase 1 results self.phase1_results = ranker.rank(phase1_raw) top5 = ranker.top_n(self.phase1_results, 5) n_passing = sum(1 for r in self.phase1_results if r.passing) self._log("info" if n_passing > 0 else "warning", f"Phase 1 complete: {n_passing}/{len(self.phase1_results)} profitable configurations found" ) # Emit Phase 1 summary for checkpoint UI self._emit("phase1_complete", { "total_tested": len(self.phase1_results), "n_passing": n_passing, "top_results": [self._result_to_dict(r) for r in top5], }) if not top5: self._emit("no_profitable_config", { "msg": ( "Phase 1 found no profitable configuration. " "Suggestions: try a different date range, check EA settings, " "or try a different timeframe." ) }) self._emit_early_termination( reason_code="no_profit", message="Optimization stopped early: Phase 1 found no profitable configuration.", details={ "phase": "phase1", "total_tested": len(self.phase1_results), "suggestions": [ "Try a different date range", "Check EA settings for issues", "Try a different timeframe", ], }, ) self._emit_thinking( "No profitable configuration found across the broad scan. " "Strategy is unstable under current EA settings — stopping optimization.", kind="warning", ) self._log("error", "❌ No profitable configuration found in Phase 1. Stopping.") return if self._stop_flag: return # ── Phase 2: Refinement (AI-Guided or Random Neighbor) ─────────────── if not budget.can_fit(3): self._log("warning", "⏱ Not enough budget for Phase 2 — using Phase 1 winner directly") self.final_result = top5[0] elif cfg.autonomous_mode and self._ai_reasoner and self._ai_reasoner.enabled: # ── AI-Guided Autonomous Loop ───────────────────────────────────── self._phase = "phase2" self._emit("phase_start", { "phase": "phase2", "total": cfg.autonomous_max_iterations, "mode": "autonomous", }) self._emit_thinking( f"Phase 1 found {len(top5)} promising candidates. Best Calmar is " f"{top5[0].calmar:.2f}. Switching to autonomous AI loop — I'll read each " f"result, decide what parameters to change, and iterate toward the targets.", kind="reasoning", ) self._log("info", f"━━ Phase 2: Autonomous AI Loop " f"(up to {cfg.autonomous_max_iterations} iterations) ━━" ) from optimizer.ai_guided_loop import AIGuidedLoop loop = AIGuidedLoop( pipeline=self, schema=schema, cfg=cfg, builder=builder, runner=runner, parser=parser, store=store, writer=writer, ranker=ranker, profile=profile, budget=budget, ) targets = { "min_profit_factor": cfg.target_profit_factor, "max_drawdown_pct": cfg.target_max_drawdown_pct, "min_calmar": cfg.target_min_calmar, } loop.run( seed_results=self.phase1_results, max_iterations=cfg.autonomous_max_iterations, targets=targets, ) self.phase2_results = loop.all_results all_candidates = [r for r in (self.phase1_results + loop.all_results) if r.passing] all_ranked = ranker.rank(all_candidates) if all_candidates else ranker.rank(self.phase1_results) self.final_result = all_ranked[0] if all_ranked else top5[0] self._emit("phase2_complete", { "best_run_id": self.final_result.run_id, "best_score": round(self.final_result.score, 4), "best_profit": round(self.final_result.net_profit, 2), "best_calmar": round(self.final_result.calmar, 3), "mode": "autonomous", "iterations": len(loop.all_results), }) self._log("info", f"AI Loop complete. Best: {self.final_result.run_id} " f"(PF={self.final_result.profit_factor:.2f}, " f"profit=${self.final_result.net_profit:.0f}, " f"calmar={self.final_result.calmar:.2f})" ) else: # ── Original Random-Neighbor Phase 2 ───────────────────────────── self._phase = "phase2" self._emit("phase_start", {"phase": "phase2", "total": cfg.phase2_samples}) self._emit_thinking( f"Phase 2 starting in classic mode: sampling {cfg.phase2_samples} neighbors " f"around the top {min(3, len(top5))} Phase-1 winners (no AI loop).", kind="info", ) self._log("info", f"━━ Phase 2: Deep Refinement (refining top {min(3, len(top5))} configs) ━━") top3 = top5[:3] phase2_raw = [] neighbors_per = cfg.phase2_samples // max(1, len(top3)) for rank_i, base_result in enumerate(top3): if self._stop_flag: break self._log("info", f" Refining config #{rank_i+1}: {base_result.run_id}") neighbors = sampler.sample_neighbors( base_result.params, schema, n_neighbors=neighbors_per, step_pct=0.20 ) for j, params in enumerate(neighbors): if self._stop_flag or budget.is_exhausted(): break run_id = f"p2_{rank_i+1}_{j+1:02d}_{datetime.utcnow().strftime('%H%M%S')}" t0 = time.time() result = self._execute_run( run_id, params, cfg.train_start, cfg.train_end, "phase2", builder, runner, parser, store, writer, ranker, profile ) budget.record_run(time.time() - t0) phase2_raw.append(result) self._run_count += 1 run_dict_p2 = self._make_run_dict(run_id, result, "phase2") self._completed_runs.append(run_dict_p2) if self._live_best is None or result.score > self._live_best.score: self._live_best = result self._emit("run_complete", { **run_dict_p2, "progress_pct": round(self._run_count / self._total_runs * 100), }) all_results = list(self.phase1_results) + ranker.rank(phase2_raw) all_ranked = ranker.rank( [r for r in all_results if r.passing] or list(self.phase1_results) ) self.phase2_results = ranker.rank(phase2_raw) self.final_result = all_ranked[0] if all_ranked else top5[0] self._emit("phase2_complete", { "best_run_id": self.final_result.run_id, "best_score": round(self.final_result.score, 4), "best_profit": round(self.final_result.net_profit, 2), "best_calmar": round(self.final_result.calmar, 3), }) self._log("info", f"Phase 2 complete. Best config: {self.final_result.run_id} " f"(profit=${self.final_result.net_profit:.0f}, calmar={self.final_result.calmar:.2f})" ) if self._stop_flag: return # ── Phase 3: Validation ─────────────────────────────────────────────── if not budget.can_fit(2): self._log("warning", "⏱ Not enough budget for Phase 3 validation — skipping OOS test") self._emit_early_termination( reason_code="budget_exhausted", message="Skipping validation: not enough time budget remaining.", details={"phase": "phase3"}, ) self.verdict = "RISKY" oos_result = None else: self._phase = "phase3" self._emit("phase_start", {"phase": "phase3", "total": cfg.phase3_samples}) # Count the validation runs we intend to run so progress makes sense opts = schema.optimizable() plan_sens = bool(opts) and budget.can_fit(2) planned_runs = 1 + (2 if plan_sens else 0) # 1 OOS + 2 sens self._emit("validation_start", { "best_run_id": self.final_result.run_id, "planned_runs": planned_runs, "oos_start": cfg.val_start, "oos_end": cfg.val_end, "train_start": cfg.train_start, "train_end": cfg.train_end, "sensitivity": plan_sens, }) self._emit_thinking( f"Entering validation phase. Testing best config {self.final_result.run_id} " f"on out-of-sample data ({cfg.val_start} → {cfg.val_end}) + sensitivity checks.", kind="reasoning", ) self._log("info", "━━ Phase 3: Validation (out-of-sample + sensitivity) ━━") # ── OOS test ── oos_id = f"oos_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}" self._log("info", f" OOS test: {cfg.val_start} → {cfg.val_end}") self._emit("validation_run_start", { "run_id": oos_id, "kind": "oos", "label": "Out-of-Sample", "description": f"Re-testing best config on unseen data: {cfg.val_start} → {cfg.val_end}", "index": 1, "total": planned_runs, "period_start": cfg.val_start, "period_end": cfg.val_end, "params": self.final_result.params, }) self._emit_thinking( "Running out-of-sample test: if the strategy holds up on data it wasn't " "optimized on, the edge is real — not curve-fit noise.", kind="hypothesis", ) t0 = time.time() oos_result = self._execute_run( oos_id, self.final_result.params, cfg.val_start, cfg.val_end, "phase3_oos", builder, runner, parser, store, writer, ranker, profile ) budget.record_run(time.time() - t0) self._run_count += 1 oos_dict = self._make_run_dict(oos_id, oos_result, "phase3_oos") self._completed_runs.append(oos_dict) self._emit("run_complete", {**oos_dict, "progress_pct": round(self._run_count / self._total_runs * 100)}) self._emit("validation_run_complete", { "run_id": oos_id, "kind": "oos", "label": "Out-of-Sample", "index": 1, "total": planned_runs, "net_profit": round(oos_result.net_profit, 2), "profit_factor": round(oos_result.profit_factor, 3), "calmar": round(oos_result.calmar, 3), "max_drawdown": round(oos_result.max_drawdown * 100, 2), "total_trades": oos_result.total_trades, "passing": oos_result.passing, }) # Thinking narration on OOS outcome oos_ratio = (oos_result.calmar / max(self.final_result.calmar, 0.001)) if oos_result.calmar else 0 if oos_result.net_profit > 0 and oos_ratio >= 0.50: self._emit_thinking( f"OOS profitable: ${oos_result.net_profit:.0f} with Calmar {oos_result.calmar:.2f} " f"(≈{oos_ratio*100:.0f}% of training performance). The edge generalizes.", kind="success", ) else: self._emit_thinking( f"OOS weak: profit ${oos_result.net_profit:.0f}, Calmar {oos_result.calmar:.2f} — " f"strategy likely overfit to the training period.", kind="warning", ) # ── Sensitivity test (2 runs: nudge top param up and down) ── sens_results = [] if opts and budget.can_fit(2): top_param = opts[0] # first optimizable param self._emit_thinking( f"Sensitivity check: nudging `{top_param.name}` ±20% to see if performance " f"survives small parameter drift.", kind="reasoning", ) for i, direction in enumerate([1, -1], start=2): if budget.is_exhausted() or self._stop_flag: break nudged = dict(self.final_result.params) current = float(nudged.get(top_param.name, top_param.default)) span = float(top_param.max) - float(top_param.min) nudged[top_param.name] = top_param.clamp(current + direction * span * 0.20) sens_id = f"sens_{direction}_{datetime.utcnow().strftime('%H%M%S')}" label = f"Sensitivity ({top_param.name} {'+' if direction > 0 else '-'}20%)" self._emit("validation_run_start", { "run_id": sens_id, "kind": "sensitivity", "label": label, "description": f"Nudging `{top_param.name}` to {nudged[top_param.name]} " f"(from {current}) to test parameter stability.", "index": i, "total": planned_runs, "period_start": cfg.train_start, "period_end": cfg.train_end, "params": nudged, }) t0 = time.time() sr = self._execute_run( sens_id, nudged, cfg.train_start, cfg.train_end, "phase3_sens", builder, runner, parser, store, writer, ranker, profile ) budget.record_run(time.time() - t0) sens_results.append(sr) self._run_count += 1 sens_dict = self._make_run_dict(sens_id, sr, "phase3_sens") self._completed_runs.append(sens_dict) self._emit("run_complete", {**sens_dict, "progress_pct": round(self._run_count / max(self._total_runs, 1) * 100)}) self._emit("validation_run_complete", { "run_id": sens_id, "kind": "sensitivity", "label": label, "index": i, "total": planned_runs, "net_profit": round(sr.net_profit, 2), "profit_factor": round(sr.profit_factor, 3), "calmar": round(sr.calmar, 3), "max_drawdown": round(sr.max_drawdown * 100, 2), "total_trades": sr.total_trades, "passing": bool(sr.passing), }) # Determine verdict self.verdict = self._determine_verdict(self.final_result, oos_result, sens_results) # Validation done — narrate the conclusion self._emit("validation_done", { "verdict": self.verdict, "oos_passing": bool(oos_result and oos_result.passing), "sens_passing": sum(1 for s in sens_results if s.passing), "sens_total": len(sens_results), }) verdict_narration = { "RECOMMENDED": "Validation passed on all fronts. Strategy is robust and ready for deployment.", "RISKY": "Validation mixed. Strategy may work but has stability concerns — proceed with care.", "NOT_RELIABLE": "Validation failed. Strategy is not reliable — likely overfit or unstable.", }.get(self.verdict, "Validation complete.") self._emit_thinking( verdict_narration, kind=("success" if self.verdict == "RECOMMENDED" else "warning" if self.verdict == "RISKY" else "warning"), ) # ── Generate .set output ───────────────────────────────────────────── self.best_set_path = self._write_set_file(self.final_result, schema, cfg) # ── Final emit ─────────────────────────────────────────────────────── self._emit("optimization_complete", { "verdict": self.verdict, "best_run_id": self.final_result.run_id, "score": round(self.final_result.score, 4), "net_profit": round(self.final_result.net_profit, 2), "calmar": round(self.final_result.calmar, 3), "profit_factor": round(self.final_result.profit_factor, 3), "win_rate": round(self.final_result.win_rate * 100, 1), "max_drawdown": round(self.final_result.max_drawdown * 100, 2), "total_trades": self.final_result.total_trades, "oos_profit": round(oos_result.net_profit, 2) if oos_result else None, "oos_calmar": round(oos_result.calmar, 3) if oos_result else None, "set_file_url": f"/download_set/{self.final_result.run_id}" if self.best_set_path else None, "total_runs": self._run_count, "elapsed_min": round((time.time() - self.run_start_ts) / 60, 1), }) verdict_icon = {"RECOMMENDED": "✅", "RISKY": "⚠️", "NOT_RELIABLE": "❌"}.get(self.verdict, "?") self._log("success" if self.verdict == "RECOMMENDED" else "warning", f"{verdict_icon} VERDICT: {self.verdict} | " f"Profit: ${self.final_result.net_profit:.0f} | " f"Calmar: {self.final_result.calmar:.2f}" ) # ── Single run executor ─────────────────────────────────────────────────── def _execute_run( self, run_id, params, period_start, period_end, phase, builder, runner, parser, store, writer, ranker, profile ) -> RankedResult: """Execute one MT5 backtest and return a RankedResult.""" # Demo mode short-circuit — generate synthetic metrics so judges can see the # AI loop without an MT5 install. Toggle with APEX_DEMO_MODE=1. if os.environ.get("APEX_DEMO_MODE", "").strip() in ("1", "true", "yes"): return self._execute_demo_run(run_id, params, phase, ranker) run_dir = RUNS_DIR / run_id run_dir.mkdir(parents=True, exist_ok=True) try: ini_path = builder.build( run_id=run_id, params=params, period_start=period_start, period_end=period_end, output_dir=run_dir, phase=phase, ea_file=profile.ex5_file, ea_symbol=profile.symbol, ea_timeframe=profile.timeframe, ) run = Run( run_id=run_id, ea_name=profile.name, symbol=profile.symbol, timeframe=profile.timeframe, period_start=period_start, period_end=period_end, params=params, phase=phase, tester_model=self.cfg["mt5"]["tester_model"], ini_snapshot=ini_path.read_text(), ) store.save_run(run) result = runner.run( run_id, ini_path, run_dir / "report", log_csv_search_dir=Path(self.cfg["mt5"]["mql5_files_path"]), profile=profile, ) if not result.success: return ranker.make_result(run_id, params, phase, None, error=result.error_message) metrics, trades = parser.parse(result.report_xml, result.report_html) if metrics is None: return ranker.make_result(run_id, params, phase, None, error="parse_failed") metrics.run_id = run_id # Run analysis & AI reasoning findings = self._analyze_run(run_id, metrics, trades or [], params) self._reason_about_run(run_id, metrics, findings, params) trades_df = pd.DataFrame() if trades: try: trades_df = pd.DataFrame([t.model_dump() for t in trades]) except Exception: try: trades_df = pd.DataFrame([vars(t) for t in trades]) except Exception: pass # Build RankedResult first so we have the ranked_score for summary.json ranked = ranker.make_result(run_id, params, phase, metrics) ai_insight = self._run_insights.get(run_id) writer.write( run_id, metrics, trades_df, findings, params, phase=phase, ranked_score=ranked.score, ai_insight=ai_insight, ) return ranked except Exception as e: logger.warning(f"[{run_id}] Run error: {e}") return ranker.make_result(run_id, params, phase, None, error=str(e)) # ── Demo mode (synthetic backtest) ──────────────────────────────────────── def _execute_demo_run(self, run_id: str, params: dict, phase: str, ranker) -> RankedResult: """ Generate a synthetic, deterministic-but-realistic RankedResult from the params hash. Lets the AI loop run end-to-end without MT5 installed. The metrics improve as parameters approach a hidden "sweet spot" so the AI can hill-climb in a way judges can observe. We also add small jitter to look organic. """ from data.models import RunMetrics # local import to avoid cycle # Hash params → stable seed per config key = ",".join(f"{k}={v}" for k, v in sorted(params.items())) seed = int(hashlib.md5(key.encode()).hexdigest()[:12], 16) rng = random.Random(seed) # Hidden sweet-spot hash drift: a deterministic "true score" between 0–1 # based on a smooth function of parameter values. Small param changes # produce small score changes — that's what lets the AI hill-climb. true_score = 0.0 for k, v in sorted(params.items()): try: fv = float(v) # Map value into [-1,1] using a stable hash, multiply by a gentle # bias toward "moderate" values (sweet spot is mid-range). slot = (int(hashlib.md5(k.encode()).hexdigest()[:8], 16) % 100) / 100.0 normalized = (fv % 100) / 100.0 true_score += 1.0 - abs(normalized - slot) except Exception: true_score += 0.5 if params: true_score /= len(params) true_score = max(0.05, min(0.98, true_score)) # Add jitter for realism (±10%) true_score *= rng.uniform(0.90, 1.10) true_score = max(0.05, min(0.99, true_score)) # Project onto realistic metric ranges profit_factor = round(0.7 + true_score * 1.8, 3) # 0.7–2.5 calmar = round(true_score * 1.4, 3) # 0.0–1.4 max_drawdown = round(28 - true_score * 22, 2) # 28%→6% win_rate = round(40 + true_score * 25, 1) # 40%→65% total_trades = int(80 + rng.random() * 220) # 80–300 net_profit = round((profit_factor - 1) * 5000 * (1 + rng.uniform(-0.2, 0.2)), 2) # Out-of-sample tends to be slightly worse (more realistic) if phase.startswith("phase3_oos"): profit_factor *= 0.85 calmar *= 0.80 net_profit *= 0.75 avg_trade = net_profit / max(total_trades, 1) winners = int(total_trades * (win_rate / 100.0)) losers = max(1, total_trades - winners) # Derive avg win/loss consistent with profit_factor: PF = (winners*avg_win)/(losers*|avg_loss|) avg_loss = -abs(avg_trade) * (1 + 1.5 / max(profit_factor, 0.5)) avg_win = (profit_factor * losers * abs(avg_loss)) / max(winners, 1) metrics = RunMetrics( run_id=run_id, net_profit=net_profit, profit_factor=profit_factor, calmar_ratio=calmar, max_drawdown_pct=max_drawdown / 100.0, max_drawdown_abs=round(net_profit * (max_drawdown / 100.0) * 1.2 + 200, 2), win_rate=win_rate / 100.0, total_trades=total_trades, recovery_factor=round(profit_factor * 1.5, 2), sharpe_ratio=round(true_score * 1.8, 2), avg_win=round(avg_win, 2), avg_loss=round(avg_loss, 2), largest_loss=round(avg_loss * 3.5, 2), expected_payoff=round(avg_trade, 2), ) # Simulate per-run latency so the dashboard feels alive — not instant. # Each run takes ~1.5s so a 10-iteration AI loop is ~15s total. delay = float(os.environ.get("APEX_DEMO_RUN_SECONDS", "1.5")) time.sleep(max(0.05, delay)) # Run AI reasoning if enabled — same flow as live mode try: self._reason_about_run(run_id, metrics, [], params) except Exception: pass return ranker.make_result(run_id, params, phase, metrics) # ── Verdict logic ───────────────────────────────────────────────────────── def _determine_verdict( self, best: RankedResult, oos: Optional[RankedResult], sens: list[RankedResult], ) -> str: """ RECOMMENDED: IS profitable + OOS profitable + not fragile RISKY: IS profitable but OOS weak OR fragile NOT_RELIABLE: IS marginal or OOS loss """ if best.calmar < 0.1 or best.net_profit <= 0: return "NOT_RELIABLE" oos_ok = False if oos and oos.net_profit > 0: # OOS degradation: acceptable if OOS calmar ≥ 50% of IS calmar oos_ratio = oos.calmar / max(best.calmar, 0.001) oos_ok = oos_ratio >= 0.50 elif oos is None: oos_ok = True # No OOS test — can't penalize # Sensitivity: fragile if any nudge drops Calmar by >50% fragile = any( s.passing and s.calmar < best.calmar * 0.50 for s in sens ) if sens else False if oos_ok and not fragile and best.calmar >= 0.30: return "RECOMMENDED" if oos_ok or (not fragile and best.calmar >= 0.20): return "RISKY" return "NOT_RELIABLE" # ── .set file output ────────────────────────────────────────────────────── def _write_set_file( self, result: RankedResult, schema: ParameterSchema, cfg: SessionConfig ) -> Optional[Path]: """Write a clean .set file for MT5 import.""" try: out_dir = self.reports_dir / result.run_id out_dir.mkdir(parents=True, exist_ok=True) out_path = out_dir / f"{cfg.ea_name}_optimized_{cfg.symbol}_{cfg.timeframe}.set" header = ( f"Optimized by MT5 Smart Optimizer\n" f"EA: {cfg.ea_name} | Symbol: {cfg.symbol} | TF: {cfg.timeframe}\n" f"Training: {cfg.train_start} – {cfg.train_end}\n" f"Verdict: {self.verdict}\n" f"Net Profit: ${result.net_profit:.2f} | Calmar: {result.calmar:.2f} | " f"Win Rate: {result.win_rate:.1f}%\n" f"Generated: {datetime.utcnow().strftime('%Y-%m-%d %H:%M UTC')}" ) content = schema.to_set_file(result.params, header_comment=header) out_path.write_text(content, encoding="utf-8") logger.info(f"Optimized .set file written: {out_path}") return out_path except Exception as e: logger.warning(f"Could not write .set file: {e}") return None # ── Helpers ─────────────────────────────────────────────────────────────── def _make_run_dict(self, run_id: str, result: RankedResult, phase: str) -> dict: """Canonical run dict used for both _completed_runs and run_complete emits. IMPORTANT — unit convention: win_rate and max_drawdown are emitted as PERCENTAGES (0–100), not fractions, so the dashboard can render them directly with `${val}%`. RankedResult stores these as fractions so we scale here at the API boundary. """ d = { "run_id": run_id, "phase": phase, "ts": datetime.utcnow().isoformat(), "net_profit": round(result.net_profit, 2), "calmar": round(result.calmar, 3), "profit_factor": round(result.profit_factor, 3), "win_rate": round(result.win_rate * 100, 1), "max_drawdown": round(result.max_drawdown * 100, 2), "total_trades": result.total_trades, "passing": bool(result.passing), "score": round(result.score, 4), "params": result.params, } insight = self._run_insights.get(run_id) if insight: d["ai_insight"] = insight return d def _result_to_dict(self, r: RankedResult) -> dict: d = { "run_id": r.run_id, "rank": r.rank, "score": round(r.score, 4), "net_profit": round(r.net_profit, 2), "calmar": round(r.calmar, 3), "profit_factor": round(r.profit_factor, 3), "win_rate": round(r.win_rate * 100, 1), # fraction → % "max_drawdown": round(r.max_drawdown * 100, 2), # fraction → % "total_trades": r.total_trades, "passing": bool(r.passing), "params": r.params, # full param dict "params_summary": self._params_summary(r.params), } # Attach AI insight if available insight = self._run_insights.get(r.run_id) if insight: d["ai_insight"] = insight return d @staticmethod def _params_summary(params: dict) -> str: """Show a few key params for display — works with any EA.""" parts = [] for k, v in list(params.items())[:8]: short = k.replace("Inp", "").replace("inp", "")[:12] parts.append(f"{short}={v}") return " | ".join(parts[:4]) def _analyze_run(self, run_id: str, metrics, trades: list, params: dict) -> list: """Run Tier 1 analyzers on trade data. Always available from HTML report.""" findings = [] if not trades: return findings try: trades_df = pd.DataFrame([t.model_dump() for t in trades]) except Exception: try: trades_df = pd.DataFrame([vars(t) for t in trades]) except Exception: return findings if trades_df.empty: return findings for analyzer_cls in (EquityCurveAnalyzer, TimePerformanceAnalyzer): try: az = analyzer_cls() result = az.analyze(trades_df) if isinstance(result, list): findings.extend(result) elif result is not None: findings.append(result) except Exception as e: logger.debug(f"[{run_id}] {analyzer_cls.__name__} skipped: {e}") self._run_findings[run_id] = findings return findings def _reason_about_run( self, run_id: str, metrics, findings: list, params: dict ) -> Optional[dict]: """Call AI reasoner and emit insight via SocketIO.""" if not self._ai_reasoner or not self._ai_reasoner.enabled: return None try: history = [ { "run_id": r.run_id, "score": r.score, "calmar": r.calmar, "pf": r.profit_factor, "phase": r.phase, } for r in (self.phase1_results + self.phase2_results)[-5:] ] insight = self._ai_reasoner.analyze( findings=findings, metrics=metrics, run_history=history, current_params=params, ) d = insight.to_dict() d["run_id"] = run_id self._ai_insights.append(d) self._run_insights[run_id] = d self._emit("ai_insight", d) return d except Exception as e: logger.warning(f"[{run_id}] AI reasoning failed: {e}") return None def get_latest_insight(self) -> Optional[dict]: """Return the most recent AI insight.""" return self._ai_insights[-1] if self._ai_insights else None def get_all_insights(self) -> list[dict]: return list(self._ai_insights) def _log(self, level: str, msg: str) -> None: getattr(logger, level, logger.info)(msg) self._emit("log", {"level": level, "msg": msg}) def _emit_thinking( self, msg: str, kind: str = "info", iteration: Optional[int] = None, meta: Optional[dict] = None, ) -> None: """ Emit an AI-thinking stream event — distinct from system logs. Shows up in the dashboard's "Live AI Thinking Feed". kind: 'info' | 'reasoning' | 'decision' | 'warning' | 'success' | 'hypothesis' """ payload = { "msg": msg, "kind": kind, "iteration": iteration, "phase": self._phase, "ts": datetime.utcnow().isoformat(), } if meta: payload["meta"] = meta self._emit("ai_thinking", payload) def _emit_early_termination(self, reason_code: str, message: str, details: dict = None) -> None: """ Surface an early stop to the user. reason_code examples: - 'no_profit' (Phase 1 found nothing) - 'targets_met' (AI loop hit all quality targets) - 'budget_exhausted'(time budget used up) - 'user_stop' (user clicked Stop) - 'stuck_escape' (optimizer stuck, bailing) """ payload = { "reason": reason_code, "message": message, "phase": self._phase, "ts": datetime.utcnow().isoformat(), } if details: payload["details"] = details self._emit("early_termination", payload) def _emit(self, event: str, data: dict = {}) -> None: try: self.socketio.emit(event, data) except Exception as e: logger.debug(f"Emit error ({event}): {e}")