""" optimizer_loop.py Background thread that runs the full optimization loop and emits real-time SocketIO events to the dashboard. """ from __future__ import annotations import json import threading import time import uuid from datetime import datetime from pathlib import Path from typing import Any, Optional import yaml from loguru import logger # Project imports from data.models import Run, RunMetrics, Candidate, Hypothesis from data.store import DataStore from mt5.ini_builder import IniBuilder from mt5.runner import MT5Runner from mt5.report_parser import ReportParser from mt5.log_reader import TradeLogReader from analysis.reversal import ReversalAnalyzer from analysis.time_performance import TimePerformanceAnalyzer from analysis.entry_exit_quality import EntryExitQualityAnalyzer from analysis.equity_curve import EquityCurveAnalyzer from scoring.composite import CompositeScorer from mutation.engine import MutationEngine from validation.gate import ValidationGate from reports.writer import ReportWriter from analysis.ai_reasoner import AIReasoner from analysis.ai_reasoner_config import load_api_key import pandas as pd BASE_DIR = Path(__file__).parent KB_PATH = BASE_DIR / "mutation" / "knowledge_base.yaml" DB_PATH = BASE_DIR / "optimizer.db" RUNS_DIR = BASE_DIR / "runs" # Keep MANIFEST_PATH for MutationEngine (still uses it for advanced mode) MANIFEST_PATH = BASE_DIR / "mutation" / "param_manifest.yaml" class OptimizerLoop: """ Runs the full optimization pipeline in a background thread. Pushes events to the SocketIO broadcast channel so the dashboard updates in real time. """ def __init__(self, config_path: str, socketio, reports_dir: Path, auto_mode: bool = True): self.config_path = config_path self.socketio = socketio self.reports_dir = reports_dir self.auto_mode = auto_mode self.running = False self.paused = False self._stop_flag = False self._skip_flag = False self._pause_event = threading.Event() self._pause_event.set() # not paused initially # Live state (read by /api/status) self.iteration = 0 self.phase = "idle" self.best_score = 0.0 self.current_run_id: Optional[str] = None self.score_history: list[dict] = [] self.run_start_ts: Optional[float] = None self.session_tested_deltas: list[dict] = [] # dedup within this session only # AI Reasoning Layer api_key = load_api_key(config_path) self.ai_reasoner = AIReasoner(api_key=api_key) self._run_history: list[dict] = [] # accumulates across iterations for AI context with open(config_path) as f: self.cfg = yaml.safe_load(f) # ── Controls ────────────────────────────────────────────────────────────── def toggle_pause(self): self.paused = not self.paused if self.paused: self._pause_event.clear() self._emit("status_change", {"state": "paused"}) else: self._pause_event.set() self._emit("status_change", {"state": "running"}) def stop(self): self._stop_flag = True self._pause_event.set() self.running = False def skip_hypothesis(self): self._skip_flag = True def get_status(self) -> dict: elapsed = int(time.time() - self.run_start_ts) if self.run_start_ts else 0 return { "state": "running" if self.running else ("paused" if self.paused else "idle"), "iteration": self.iteration, "phase": self.phase, "best_score": round(self.best_score, 4), "run_id": self.current_run_id, "elapsed_s": elapsed, } # ── Main loop ───────────────────────────────────────────────────────────── def run(self): self.running = True self._stop_flag = False self.run_start_ts = time.time() cfg = self.cfg store, builder, runner, parser, log_rdr, analyzers, scorer, mutator, gate, writer = ( self._build_components() ) per = cfg["periods"] self._emit("status_change", {"state": "running", "phase": "baseline"}) self._emit("log", {"level": "info", "msg": "🚀 Optimizer started"}) # ── Baseline run ────────────────────────────────────────────────────── self.phase = "baseline" default_params = builder.default_params() baseline_id = f"baseline_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}" baseline_metrics, baseline_trades = self._execute_run( run_id=baseline_id, params=default_params, period_start=per["train_start"], period_end=per["train_end"], phase="baseline", hypothesis_id=None, store=store, builder=builder, runner=runner, parser=parser, log_rdr=log_rdr, analyzers=analyzers, scorer=scorer, ) if baseline_metrics is None: self._emit("error", {"msg": "Baseline run failed. Check MT5 configuration."}) self.running = False return # Write baseline report findings = self._run_analysis(baseline_id, baseline_trades, baseline_metrics, analyzers, store) # AI Reasoning — baseline self._emit("log", {"level": "info", "msg": "🤖 AI Reasoner analyzing baseline..."}) ai_insight = self.ai_reasoner.analyze( findings=findings, metrics=baseline_metrics, run_history=self._run_history, current_params=default_params, ) self._run_history.append({ "run_id": baseline_id, "score": round(baseline_metrics.composite_score, 4), "calmar": round(baseline_metrics.calmar_ratio, 4), "pf": round(baseline_metrics.profit_factor, 4), "phase": "baseline", "params": default_params, }) self._emit("ai_insight", ai_insight.to_dict()) self._emit("log", {"level": "info", "msg": f"🤖 AI: {ai_insight.headline}"}) writer.write(baseline_id, baseline_metrics, baseline_trades, findings, default_params) self.best_score = baseline_metrics.composite_score current_params = default_params.copy() current_metrics = baseline_metrics current_findings = findings # reuse in iter 1, avoid double-emit no_improve_count = 0 max_iter = cfg["optimization"]["max_iterations"] conv_win = cfg["optimization"]["convergence_window"] conv_thr = cfg["optimization"]["convergence_threshold"] # ── Iteration loop ──────────────────────────────────────────────────── while self.iteration < max_iter and not self._stop_flag: self._pause_event.wait() if self._stop_flag: break self.iteration += 1 self.phase = "analyze" self._emit("iteration_start", {"iteration": self.iteration}) self._emit("log", {"level": "info", "msg": f"━━ Iteration {self.iteration} ━━"}) # Load latest trades (for re-analysis) trades_df = store.load_trades(current_metrics.run_id) if trades_df.empty and not baseline_trades.empty: trades_df = baseline_trades # Analysis — reuse cached findings when re-analyzing same run_id if current_findings is not None and trades_df.empty: findings = current_findings else: findings = self._run_analysis( current_metrics.run_id, trades_df, current_metrics, analyzers, store ) current_findings = None # only reuse once if not findings: self._emit("log", {"level": "warn", "msg": "No actionable findings. Stopping."}) break # AI Reasoning — per iteration self._emit("log", {"level": "info", "msg": f"🤖 AI Reasoner analyzing iteration {self.iteration}..."}) ai_insight = self.ai_reasoner.analyze( findings=findings, metrics=current_metrics, run_history=self._run_history, current_params=current_params, ) self._emit("ai_insight", ai_insight.to_dict()) self._emit("log", {"level": "info", "msg": f"🤖 AI: {ai_insight.headline}"}) # Mutation proposals — only dedup within this session hypotheses = mutator.propose( findings=findings, current_params=current_params, recent_deltas=self.session_tested_deltas, max_proposals=cfg["mutation"]["max_hypotheses_per_cycle"], ) if not hypotheses: self._emit("log", {"level": "warn", "msg": "No new hypotheses. Stopping."}) break self._emit("hypotheses", {"items": [ { "id": h.hypothesis_id, "desc": h.description, "delta": h.param_delta, "strategy": h.strategy, } for h in hypotheses ]}) # Test each hypothesis iteration_best: Optional[RunMetrics] = None iteration_best_params: Optional[dict] = None iteration_best_hyp: Optional[Hypothesis] = None for idx, hyp in enumerate(hypotheses): self._pause_event.wait() if self._stop_flag or self._skip_flag: self._skip_flag = False break test_params = {**current_params, **hyp.param_delta} run_id = f"iter{self.iteration:03d}_h{idx+1}" store.save_hypothesis(hyp) self._emit("log", { "level": "info", "msg": f"Testing H{idx+1}: {hyp.description[:60]}" }) self._emit("hypothesis_testing", {"idx": idx+1, "desc": hyp.description}) test_metrics, test_trades = self._execute_run( run_id=run_id, params=test_params, period_start=per["train_start"], period_end=per["train_end"], phase="explore", hypothesis_id=hyp.hypothesis_id, store=store, builder=builder, runner=runner, parser=parser, log_rdr=log_rdr, analyzers=analyzers, scorer=scorer, ) if test_metrics is None: continue delta_score = test_metrics.composite_score - current_metrics.composite_score color = "green" if delta_score > 0 else "red" self._emit("log", { "level": "success" if delta_score > 0 else "warn", "msg": f"H{idx+1} score: {current_metrics.composite_score:.4f} → " f"{test_metrics.composite_score:.4f} ({delta_score:+.4f})" }) # Write per-run report h_findings = self._run_analysis(run_id, test_trades, test_metrics, analyzers, store) writer.write(run_id, test_metrics, test_trades, h_findings, test_params, hypothesis=hyp, baseline_score=current_metrics.composite_score) store.update_hypothesis_status(hyp.hypothesis_id, "tested", run_id) self.session_tested_deltas.append(hyp.param_delta) # track for session dedup # Emit score update for every run so chart populates self._emit("score_update", { "iteration": self.iteration, "run_id": run_id, "score": round(test_metrics.composite_score, 4), "calmar": round(test_metrics.calmar_ratio, 4), "pf": round(test_metrics.profit_factor, 4), "ts": datetime.utcnow().isoformat(), "promoted": False, }) if iteration_best is None or test_metrics.composite_score > iteration_best.composite_score: iteration_best = test_metrics iteration_best_params = test_params iteration_best_hyp = hyp if iteration_best is None: no_improve_count += 1 continue # Validation gate self.phase = "validate" gate_result = gate.run_is_check(iteration_best, baseline_score=current_metrics.composite_score) if not gate_result.passed: self._emit("log", {"level": "error", "msg": f"IS gate failed: {gate_result.reason}"}) store.update_hypothesis_status(iteration_best_hyp.hypothesis_id, "rejected") no_improve_count += 1 continue # Walk-forward — pass an executor so gate never imports 'main' self._emit("log", {"level": "info", "msg": "Running walk-forward validation..."}) _store, _builder, _runner, _parser, _log_rdr, _analyzers, _scorer = ( store, builder, runner, parser, log_rdr, analyzers, scorer ) def _wfv_executor(params, start, end, fold_id): m, _ = self._execute_run( run_id=fold_id, params=params, period_start=start, period_end=end, phase="wfv", hypothesis_id=None, store=_store, builder=_builder, runner=_runner, parser=_parser, log_rdr=_log_rdr, analyzers=_analyzers, scorer=_scorer, ) return m wfv = gate.run_walk_forward( params=iteration_best_params, executor=_wfv_executor, ) self._emit("log", { "level": "success" if wfv.passed else "warn", "msg": f"WFV: OOS/IS ratio = {wfv.oos_is_ratio:.2f} " f"({'PASS ✅' if wfv.passed else 'FAIL ❌'})" }) if not wfv.passed: store.update_hypothesis_status(iteration_best_hyp.hypothesis_id, "rejected") no_improve_count += 1 continue # Promote candidate candidate = Candidate( run_id=iteration_best.run_id, composite_score=iteration_best.composite_score, params=iteration_best_params, ) store.save_candidate(candidate) store.update_hypothesis_status(iteration_best_hyp.hypothesis_id, "validated") self._emit("candidate_promoted", { "candidate_id": candidate.candidate_id, "score": round(candidate.composite_score, 4), "delta": round(candidate.composite_score - self.best_score, 4), "params": candidate.params, }) self._emit("log", { "level": "success", "msg": f"✅ Candidate promoted! Score: {candidate.composite_score:.4f}" }) improvement = iteration_best.composite_score - self.best_score if improvement >= conv_thr: current_params = iteration_best_params current_metrics = iteration_best self.best_score = iteration_best.composite_score no_improve_count = 0 else: no_improve_count += 1 # Track run history for AI context self._run_history.append({ "run_id": iteration_best.run_id, "score": round(iteration_best.composite_score, 4), "calmar": round(iteration_best.calmar_ratio, 4), "pf": round(iteration_best.profit_factor, 4), "phase": "explore", "params": iteration_best_params, }) # Update score chart self.score_history.append({ "iteration": self.iteration, "score": round(self.best_score, 4), "calmar": round(current_metrics.calmar_ratio, 4), "pf": round(current_metrics.profit_factor, 4), "ts": datetime.utcnow().isoformat(), }) self._emit("score_update", self.score_history[-1]) if no_improve_count >= conv_win: self._emit("log", { "level": "warn", "msg": f"Converged: no improvement in {no_improve_count} iterations." }) break # ── Done ───────────────────────────────────────────────────────────── candidates = store.list_candidates() self._emit("optimization_complete", { "candidates": len(candidates), "best_score": round(self.best_score, 4), "iterations": self.iteration, }) self._emit("log", {"level": "success", "msg": "🏁 Optimization complete!"}) self.running = False self.phase = "idle" # ── Single run ──────────────────────────────────────────────────────────── def _execute_run( self, run_id, params, period_start, period_end, phase, hypothesis_id, store, builder, runner, parser, log_rdr, analyzers, scorer, ): cfg = self.cfg self.current_run_id = run_id run_dir = RUNS_DIR / run_id run_dir.mkdir(parents=True, exist_ok=True) self._emit("run_started", { "run_id": run_id, "phase": phase, "period": f"{period_start} → {period_end}", "params": {k: v for k, v in list(params.items())[:8]}, # first 8 for display }) 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=self._profile.ex5_file, ea_symbol=self._profile.symbol, ea_timeframe=self._profile.timeframe, ) run = Run( run_id=run_id, ea_name=self._profile.name, symbol=self._profile.symbol, timeframe=self._profile.timeframe, period_start=period_start, period_end=period_end, params=params, phase=phase, hypothesis_id=hypothesis_id, tester_model=cfg["mt5"]["tester_model"], ini_snapshot=ini_path.read_text(), ) store.save_run(run) self._emit("log", {"level": "info", "msg": f"⏳ MT5 running: {run_id}"}) result = runner.run( run_id, ini_path, run_dir / "report", log_csv_search_dir=Path(cfg["mt5"]["mql5_files_path"]), profile=self._profile, ) if not result.success: self._emit("run_failed", {"run_id": run_id, "error": result.error_message}) return None, pd.DataFrame() metrics, trades = parser.parse(result.report_xml, result.report_html) if metrics is None: self._emit("run_failed", {"run_id": run_id, "error": "Could not parse report"}) return None, pd.DataFrame() metrics.run_id = run_id if trades: trades = log_rdr.merge( trades, result.trade_log_csv, reversal_mfe_threshold_pips=cfg["analysis"]["reversal"]["mfe_threshold_pips"], ) trades_df = pd.DataFrame([t.model_dump() for t in trades]) else: trades_df = pd.DataFrame() if not trades_df.empty: if "result_class" in trades_df.columns: losers = trades_df[trades_df["net_money"] < 0] reversals = trades_df[trades_df["result_class"] == "reversal"] metrics.reversal_rate = len(reversals) / max(1, len(losers)) if "mfe_capture_ratio" in trades_df.columns: # dropna() first — if no MFE data, series is all-NaN, mean()=NaN # Use None (not NaN) so scorer uses its safe default of 0.5 cap_series = trades_df["mfe_capture_ratio"].dropna() metrics.avg_mfe_capture = float(cap_series.mean()) if not cap_series.empty else None metrics.composite_score = scorer.score(metrics) store.save_metrics(metrics) if not trades_df.empty: store.save_trades(run_id, trades) run.report_path = result.report_xml store.save_run(run) self._emit("run_complete", { "run_id": run_id, "phase": phase, "net_profit": round(metrics.net_profit, 2), "profit_factor": round(metrics.profit_factor, 3), "calmar": round(metrics.calmar_ratio, 3), "drawdown_pct": round(metrics.max_drawdown_pct * 100, 1), "win_rate": round(metrics.win_rate * 100, 1), "total_trades": metrics.total_trades, "score": round(metrics.composite_score, 4), "reversal_rate": round((metrics.reversal_rate or 0) * 100, 1), "mfe_capture": round((metrics.avg_mfe_capture or 0) * 100, 1), }) return metrics, trades_df def _run_analysis(self, run_id, trades_df, metrics, analyzers, store): all_findings = [] for az in analyzers: findings = az.run(trades_df, metrics, run_id) for f in findings: self._emit("finding", { "analyzer": f.analyzer, "severity": f.severity, "description": f.description, "confidence": round(f.confidence, 2), "impact": round(f.impact_estimate_pnl, 0), }) all_findings.extend(findings) all_findings.sort(key=lambda f: f.confidence, reverse=True) store.save_findings(all_findings) return all_findings # ── SocketIO emit helper ────────────────────────────────────────────────── def _emit(self, event: str, data: dict = {}): try: self.socketio.emit(event, data) except Exception as e: logger.debug(f"Emit error ({event}): {e}") # ── Component factory ───────────────────────────────────────────────────── def _build_components(self): cfg = self.cfg # ── EA Registry: load profile + schema ──────────────────────────────── from ea.registry import EARegistry reg = EARegistry(self.config_path) ea_name = cfg.get("ea", {}).get("name", "LEGSTECH_EA_V2") try: self._profile = reg.get(ea_name) schema = reg.get_schema(self._profile) logger.info(f"Using EA profile from registry: {ea_name} (mode={self._profile.mode})") except KeyError: # Profile not in registry — fall back to legacy manifest logger.warning(f"EA {ea_name!r} not in registry, using legacy manifest") from ea.registry import EAProfile self._profile = EAProfile( name=ea_name, ex5_file=cfg["ea"]["file"], set_template="", symbol=cfg["ea"]["symbol"], timeframe=cfg["ea"]["timeframe"], ) schema = None # builder will use manifest # ── Component construction ───────────────────────────────────────────── store = DataStore(DB_PATH, RUNS_DIR) if schema is not None: builder = IniBuilder(self.config_path, schema=schema) else: builder = IniBuilder(self.config_path, str(MANIFEST_PATH)) runner = MT5Runner(self.config_path) parser = ReportParser() log_rdr = TradeLogReader( broker_tz_offset_hours=cfg["broker"]["timezone_offset_hours"], ) analyzers = [ ReversalAnalyzer( mfe_threshold_pips=cfg["analysis"]["reversal"]["mfe_threshold_pips"], min_reversal_rate=cfg["analysis"]["reversal"]["min_reversal_rate"], permutation_n=cfg["analysis"]["reversal"]["permutation_n"], ), TimePerformanceAnalyzer( z_score_threshold=cfg["analysis"]["time_performance"]["z_score_threshold"], min_bucket_trades=cfg["analysis"]["time_performance"]["min_trades_per_bucket"], permutation_n=cfg["analysis"]["time_performance"]["permutation_n"], ), EntryExitQualityAnalyzer( poor_exit_threshold=cfg["analysis"]["entry_exit"]["poor_exit_quality"], poor_entry_threshold=cfg["analysis"]["entry_exit"]["poor_entry_quality"], ), EquityCurveAnalyzer( max_flatness=cfg["analysis"]["equity_curve"]["max_flatness_score"], min_r_squared=cfg["analysis"]["equity_curve"]["min_r_squared"], ), ] scorer = CompositeScorer(self.config_path) mutator = MutationEngine(KB_PATH, MANIFEST_PATH, dedup_lookback=cfg["mutation"]["dedup_lookback_runs"]) gate = ValidationGate(self.config_path) writer = ReportWriter(self.reports_dir) return store, builder, runner, parser, log_rdr, analyzers, scorer, mutator, gate, writer