"""Builtin strategy registry.""" from __future__ import annotations import inspect from dataclasses import dataclass from typing import Any, Callable from strategies.builtin import ma_crossover, rsi_reversion SignalFn = Callable[..., tuple[Any, Any]] @dataclass(frozen=True) class BuiltinStrategy: key: str display_name: str description: str module: Any generate_signals: SignalFn default_params: dict[str, Any] param_grid: dict[str, list[Any]] source_code: str def optimize(self, close, init_cash: float, fees: float, metric: str, grid_override: dict | None = None): if self.key == "ma_crossover": pool = (grid_override or {}).get("window_pool", self.param_grid.get("window_pool")) return self.module.optimize_vectorized( close, window_pool=pool, init_cash=init_cash, fees=fees, metric=metric, ) if self.key == "rsi_reversion": return self.module.optimize_grid( close, param_grid=grid_override or self.param_grid, init_cash=init_cash, fees=fees, metric=metric, ) raise NotImplementedError(f"No optimizer for {self.key}") def _register(module) -> BuiltinStrategy: return BuiltinStrategy( key=module.STRATEGY_KEY, display_name=module.DISPLAY_NAME, description=module.DESCRIPTION, module=module, generate_signals=module.generate_signals, default_params=dict(module.DEFAULT_PARAMS), param_grid=dict(module.PARAM_GRID), source_code=inspect.getsource(module), ) BUILTIN_STRATEGIES: dict[str, BuiltinStrategy] = { ma_crossover.STRATEGY_KEY: _register(ma_crossover), rsi_reversion.STRATEGY_KEY: _register(rsi_reversion), } def list_builtins() -> list[BuiltinStrategy]: return list(BUILTIN_STRATEGIES.values()) def get_builtin(key: str) -> BuiltinStrategy: if key not in BUILTIN_STRATEGIES: raise KeyError(f"Unknown builtin strategy: {key}") return BUILTIN_STRATEGIES[key]