Coverage for src / monte_neo / core / optimizer.py: 42%
126 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-28 16:27 +0200
1"""Parameter optimizer module.
3Optimizes indicator parameters using various strategies.
4"""
6from __future__ import annotations
8from collections.abc import Callable
9from dataclasses import dataclass, field
10from typing import TYPE_CHECKING
12import numpy as np
14from monte_neo.utils.cache import load_calibration, save_calibration
15from monte_neo.utils.logger import get_logger
17if TYPE_CHECKING:
18 import pandas as pd
20 from monte_neo.indicators.base import BaseIndicator
21 from monte_neo.metrics.calculator import MetricsCalculator
23logger = get_logger(__name__)
26@dataclass
27class OptimizationResult:
28 """Optimization result."""
30 best_params: dict
31 best_score: float
32 iterations: int
33 history: list = field(default_factory=list)
36class ParameterOptimizer:
37 """Optimize indicator parameters."""
39 def __init__(
40 self,
41 method: str = "random",
42 max_iterations: int = 1000,
43 random_seed: int | None = None,
44 ) -> None:
45 """Initialize optimizer.
47 Args:
48 method: Optimization method ('random', 'grid', 'genetic').
49 max_iterations: Maximum iterations.
50 random_seed: Random seed.
51 """
52 self.method = method
53 self.max_iterations = max_iterations
54 self.rng = np.random.default_rng(random_seed)
56 def optimize(
57 self,
58 indicator: BaseIndicator,
59 param_ranges: dict[str, tuple[int, int]],
60 data: pd.DataFrame,
61 metrics_calc: MetricsCalculator,
62 objective: str = "sharpe_ratio",
63 objective_func: Callable[[dict], float] | None = None,
64 ) -> OptimizationResult:
65 """Optimize indicator parameters.
67 Args:
68 indicator: Indicator to optimize.
69 param_ranges: Parameter ranges {name: (min, max)}.
70 data: OHLCV data.
71 metrics_calc: Metrics calculator.
72 objective: Metric to maximize.
73 objective_func: Custom objective function.
75 Returns:
76 OptimizationResult with best parameters.
77 """
78 # Try to load from cache
79 indicator_name = indicator.__class__.__name__
80 if hasattr(indicator, "source_code"):
81 # For dynamic indicators, use source code as part of the key
82 indicator_name += f"_{hash(indicator.source_code)}"
84 cached_params = load_calibration(indicator_name, data)
85 if cached_params:
86 logger.info(f"🚀 Using cached calibration for {indicator_name}")
87 # We don't have the history/iterations, so we return a simplified result
88 return OptimizationResult(
89 best_params=cached_params,
90 best_score=0.0, # Unknown but presumably good
91 iterations=0,
92 history=[],
93 )
95 if self.method == "random":
96 result = self._random_search(
97 indicator, param_ranges, data, metrics_calc, objective, objective_func
98 )
99 elif self.method == "grid":
100 result = self._grid_search(
101 indicator, param_ranges, data, metrics_calc, objective, objective_func
102 )
103 elif self.method == "genetic":
104 result = self._genetic_search(
105 indicator, param_ranges, data, metrics_calc, objective, objective_func
106 )
107 else:
108 raise ValueError(f"Unknown method: {self.method}")
110 # Save to cache
111 save_calibration(indicator_name, data, result.best_params)
112 return result
114 def _random_search(
115 self,
116 indicator: BaseIndicator,
117 param_ranges: dict,
118 data: pd.DataFrame,
119 metrics_calc: MetricsCalculator,
120 objective: str,
121 objective_func: Callable | None,
122 ) -> OptimizationResult:
123 """Random search optimization."""
124 best_params = {}
125 best_score = float("-inf")
126 history = []
128 for i in range(self.max_iterations):
129 # Generate random parameters
130 params = {}
131 for name, (min_val, max_val) in param_ranges.items():
132 params[name] = int(self.rng.integers(min_val, max_val + 1))
134 # Evaluate
135 score = self._evaluate(
136 indicator, params, data, metrics_calc, objective, objective_func
137 )
139 history.append({"iteration": i, "params": params, "score": score})
141 if score > best_score:
142 best_score = score
143 best_params = dict(params)
145 return OptimizationResult(
146 best_params=best_params,
147 best_score=best_score,
148 iterations=self.max_iterations,
149 history=history,
150 )
152 def _grid_search(
153 self,
154 indicator: BaseIndicator,
155 param_ranges: dict,
156 data: pd.DataFrame,
157 metrics_calc: MetricsCalculator,
158 objective: str,
159 objective_func: Callable | None,
160 ) -> OptimizationResult:
161 """Grid search optimization."""
162 from itertools import product
164 # Create grid
165 grid_points = {}
166 for name, (min_val, max_val) in param_ranges.items():
167 step = max(1, (max_val - min_val) // 10)
168 grid_points[name] = list(range(min_val, max_val + 1, step))
170 # Search
171 best_params = {}
172 best_score = float("-inf")
173 history = []
174 iterations = 0
176 for values in product(*grid_points.values()):
177 params = dict(zip(grid_points.keys(), values))
179 score = self._evaluate(
180 indicator, params, data, metrics_calc, objective, objective_func
181 )
183 history.append({"iteration": iterations, "params": params, "score": score})
184 iterations += 1
186 if score > best_score:
187 best_score = score
188 best_params = dict(params)
190 if iterations >= self.max_iterations:
191 break
193 return OptimizationResult(
194 best_params=best_params,
195 best_score=best_score,
196 iterations=iterations,
197 history=history,
198 )
200 def _genetic_search(
201 self,
202 indicator: BaseIndicator,
203 param_ranges: dict,
204 data: pd.DataFrame,
205 metrics_calc: MetricsCalculator,
206 objective: str,
207 objective_func: Callable | None,
208 ) -> OptimizationResult:
209 """Genetic algorithm optimization."""
210 population_size = 50
211 mutation_rate = 0.1
212 elite_ratio = 0.2
214 # Initialize population
215 population = []
216 for _ in range(population_size):
217 params = {}
218 for name, (min_val, max_val) in param_ranges.items():
219 params[name] = int(self.rng.integers(min_val, max_val + 1))
220 population.append(params)
222 best_params = {}
223 best_score = float("-inf")
224 history = []
225 generations = self.max_iterations // population_size
227 for gen in range(generations):
228 # Evaluate fitness
229 fitness = []
230 for params in population:
231 score = self._evaluate(
232 indicator, params, data, metrics_calc, objective, objective_func
233 )
234 fitness.append((params, score))
236 # Sort by fitness
237 fitness.sort(key=lambda x: x[1], reverse=True)
239 # Update best
240 if fitness[0][1] > best_score:
241 best_score = fitness[0][1]
242 best_params = dict(fitness[0][0])
244 history.append({"generation": gen, "best_score": best_score})
246 # Selection and reproduction
247 elite_count = int(population_size * elite_ratio)
248 new_population = [f[0] for f in fitness[:elite_count]]
250 while len(new_population) < population_size:
251 # Tournament selection
252 parent1 = self._tournament_select(fitness)
253 parent2 = self._tournament_select(fitness)
255 # Crossover and mutation
256 child = self._crossover(parent1, parent2, param_ranges)
257 child = self._mutate(child, param_ranges, mutation_rate)
258 new_population.append(child)
260 population = new_population
262 return OptimizationResult(
263 best_params=best_params,
264 best_score=best_score,
265 iterations=generations * population_size,
266 history=history,
267 )
269 def _evaluate(
270 self,
271 indicator: BaseIndicator,
272 params: dict,
273 data: pd.DataFrame,
274 metrics_calc: MetricsCalculator,
275 objective: str,
276 objective_func: Callable | None,
277 ) -> float:
278 """Evaluate parameters."""
279 indicator.set_parameters(params)
280 signals = indicator.generate_signals(data)
281 metrics = metrics_calc.calculate_all(data, signals)
283 if objective_func:
284 return objective_func(metrics)
286 return metrics.get(objective, 0)
288 def _tournament_select(self, fitness: list, k: int = 3) -> dict:
289 """Tournament selection."""
290 selected = self.rng.choice(
291 len(fitness), size=min(k, len(fitness)), replace=False
292 )
293 best = max(selected, key=lambda i: fitness[i][1])
294 return fitness[best][0]
296 def _crossover(self, p1: dict, p2: dict, ranges: dict) -> dict:
297 """Single-point crossover."""
298 child = {}
299 for name in ranges:
300 child[name] = p1[name] if self.rng.random() < 0.5 else p2[name]
301 return child
303 def _mutate(self, params: dict, ranges: dict, rate: float) -> dict:
304 """Mutation with given rate."""
305 for name, (min_val, max_val) in ranges.items():
306 if self.rng.random() < rate:
307 params[name] = int(self.rng.integers(min_val, max_val + 1))
308 return params