Coverage for src / monte_neo / core / evolution_ai.py: 93%
110 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-28 16:27 +0200
1"""AI-Driven Evolutionary Engine.
3Advanced evolutionary optimization with symbolic regression and heuristic-based mutation.
4"""
6from __future__ import annotations
8from dataclasses import dataclass
10import numpy as np
11import pandas as pd
13from monte_neo.indicators.base import BaseIndicator
14from monte_neo.indicators.code_gen import CodeGenerator
15from monte_neo.indicators.dynamic import DynamicIndicator
16from monte_neo.metrics.calculator import MetricsCalculator
17from monte_neo.utils.logger import get_logger
19logger = get_logger(__name__)
21@dataclass
22class EvolutionStats:
23 generation: int
24 best_fitness: float
25 avg_fitness: float
26 diversity_score: float
28class AIEvolutionEngine:
29 """Advanced evolution engine with AI-inspired heuristics."""
31 def __init__(
32 self,
33 population_size: int = 100,
34 mutation_rate: float = 0.2,
35 crossover_rate: float = 0.8,
36 metrics_calc: MetricsCalculator | None = None,
37 initial_capital: float = 100000.0,
38 leverage: float = 1.0
39 ):
40 self.population_size = population_size
41 self.mutation_rate = mutation_rate
42 self.crossover_rate = crossover_rate
43 self.metrics_calc = metrics_calc or MetricsCalculator(initial_capital=initial_capital, leverage=leverage)
44 self.rng = np.random.default_rng()
45 self.code_gen = CodeGenerator(self.rng)
47 # Heuristics: Map weaknesses to potential fixes
48 self.heuristics = {
49 "high_drawdown": ["trend_filter", "volatility_exit"],
50 "low_winrate": ["mean_reversion_filter", "tighter_sl"],
51 "instability": ["smoothing", "higher_timeframe_confirmation"]
52 }
54 def evolve(self, data: pd.DataFrame, target_metrics: dict[str, float], generations: int = 10) -> BaseIndicator:
55 """Runs the AI-driven evolution process."""
56 population = self._initialize_population()
58 for gen in range(generations):
59 fitness_scores = self._evaluate_population(population, data, target_metrics)
61 # Sort by fitness
62 population = [p for p, f in sorted(zip(population, fitness_scores), key=lambda x: x[1], reverse=True)]
63 best_fitness = fitness_scores[0]
65 logger.info(f"Gen {gen}: Best Fitness = {best_fitness:.4f}")
67 # Selection & Breeding
68 new_population = population[:int(self.population_size * 0.1)] # Elitism 10%
70 while len(new_population) < self.population_size:
71 if self.rng.random() < self.crossover_rate:
72 parent1, parent2 = self.rng.choice(population[:20], size=2)
73 child = self._crossover(parent1, parent2)
74 else:
75 parent = self.rng.choice(population[:20])
76 child = self._mutate(parent)
77 new_population.append(child)
79 population = new_population
81 return population[0]
83 def _initialize_population(self) -> list[BaseIndicator]:
84 pop: list[BaseIndicator] = []
85 for _ in range(self.population_size):
86 ind = DynamicIndicator()
87 ind.set_parameter("source_code", self.code_gen.generate_code())
88 pop.append(ind)
89 return pop
91 def _evaluate_population(self, population: list[BaseIndicator], data: pd.DataFrame, targets: dict[str, float]) -> list[float]:
92 scores = []
93 for ind in population:
94 try:
95 signals = ind.generate_signals(data)
96 metrics = self.metrics_calc.calculate_all(data, signals)
98 # Multi-objective fitness score
99 score = 0.0
101 # 1. Performance (Profit Factor, Sharpe)
102 pf = metrics.get("profit_factor", 0)
103 sharpe = metrics.get("sharpe_ratio", 0)
105 # Weighted contribution
106 score += pf * 0.3
107 score += sharpe * 0.3
109 # 2. Robustness (Low Drawdown)
110 mdd = metrics.get("max_drawdown", 1.0)
111 score -= mdd * 0.2
113 # 3. Efficiency (Win Rate)
114 wr = metrics.get("win_rate", 0)
115 score += wr * 0.1
117 # 4. Target Matching (Proximity to user-defined targets)
118 # If a metric is provided in targets, penalize deviation
119 target_bonus = 0.0
120 for target_name, target_val in targets.items():
121 if target_name in metrics:
122 actual = metrics[target_name]
123 # Normalized distance (capped at 1.0)
124 if target_val != 0:
125 dist = abs(actual - target_val) / abs(target_val)
126 target_bonus += max(0, 0.2 * (1.0 - min(1.0, dist)))
127 score += target_bonus
129 # 5. Complexity Penalty (Occam's Razor)
130 formula_len = len(ind.get_formula())
131 score -= (formula_len / 1000.0) * 0.05 # Reduced penalty for AI evolution
133 scores.append(max(0.001, score)) # Ensure non-zero
134 except Exception as e:
135 logger.warning(f"Evaluation failed for indicator: {e}")
136 scores.append(0.0)
137 return scores
139 def _crossover(self, p1: BaseIndicator, p2: BaseIndicator) -> BaseIndicator:
140 """Tree-based crossover of indicator formulas."""
141 from monte_neo.utils.ast_utils import crossover_trees
143 f1 = p1.get_formula()
144 f2 = p2.get_formula()
146 try:
147 new_formula = crossover_trees(f1, f2)
148 except Exception:
149 new_formula = f1 # Fallback
151 child = DynamicIndicator()
152 child.set_parameter("source_code", new_formula)
153 return child
155 def _mutate(self, p: BaseIndicator) -> BaseIndicator:
156 """Heuristic-based mutation with symbolic tree manipulation."""
157 formula = p.get_formula()
159 # 1. Subtree Replacement
160 if self.rng.random() < 0.5:
161 new_part = self.code_gen.generate_code(depth=1)
162 # Wrap in a random operation
163 op = self.rng.choice(["+", "-", "*", "/"])
164 new_formula = f"({formula} {op} {new_part})"
165 # 2. Parameter Tweak
166 else:
167 # Look for integers in the formula and tweak them
168 import re
169 numbers = re.findall(r'\d+', formula)
170 if numbers:
171 target = self.rng.choice(numbers)
172 try:
173 val = int(target)
174 new_val = max(2, val + self.rng.integers(-5, 6))
175 new_formula = formula.replace(target, str(new_val), 1)
176 except Exception:
177 new_formula = formula
178 else:
179 new_formula = formula
181 child = DynamicIndicator()
182 child.set_parameter("source_code", new_formula)
183 return child