Simon Moisy
10bb371054
- Introduced `RegimeReversionStrategy` for ML-based regime detection and mean reversion trading. - Added feature engineering and model training logic within the new strategy. - Removed the deprecated `regime_detection.py` file to streamline the codebase. - Updated the strategy factory to include the new regime strategy configuration.
143 lines
4.2 KiB
Python
143 lines
4.2 KiB
Python
"""
|
|
Strategy factory for creating strategy instances with their parameters.
|
|
|
|
Centralizes strategy creation and parameter configuration.
|
|
"""
|
|
from dataclasses import dataclass, field
|
|
from typing import Any
|
|
|
|
import numpy as np
|
|
|
|
from strategies.base import BaseStrategy
|
|
|
|
|
|
@dataclass
|
|
class StrategyConfig:
|
|
"""
|
|
Configuration for a strategy including default and grid parameters.
|
|
|
|
Attributes:
|
|
strategy_class: The strategy class to instantiate
|
|
default_params: Parameters for single backtest runs
|
|
grid_params: Parameters for grid search optimization
|
|
"""
|
|
strategy_class: type[BaseStrategy]
|
|
default_params: dict[str, Any] = field(default_factory=dict)
|
|
grid_params: dict[str, Any] = field(default_factory=dict)
|
|
|
|
|
|
def _build_registry() -> dict[str, StrategyConfig]:
|
|
"""
|
|
Build the strategy registry lazily to avoid circular imports.
|
|
|
|
Returns:
|
|
Dictionary mapping strategy names to their configurations
|
|
"""
|
|
# Import here to avoid circular imports
|
|
from strategies.examples import MaCrossStrategy, RsiStrategy
|
|
from strategies.supertrend import MetaSupertrendStrategy
|
|
from strategies.regime_strategy import RegimeReversionStrategy
|
|
|
|
return {
|
|
"rsi": StrategyConfig(
|
|
strategy_class=RsiStrategy,
|
|
default_params={
|
|
'period': 14,
|
|
'rsi_lower': 30,
|
|
'rsi_upper': 70
|
|
},
|
|
grid_params={
|
|
'period': np.arange(10, 25, 2),
|
|
'rsi_lower': [20, 30, 40],
|
|
'rsi_upper': [60, 70, 80]
|
|
}
|
|
),
|
|
"macross": StrategyConfig(
|
|
strategy_class=MaCrossStrategy,
|
|
default_params={
|
|
'fast_window': 10,
|
|
'slow_window': 20
|
|
},
|
|
grid_params={
|
|
'fast_window': np.arange(5, 20, 5),
|
|
'slow_window': np.arange(20, 60, 10)
|
|
}
|
|
),
|
|
"meta_st": StrategyConfig(
|
|
strategy_class=MetaSupertrendStrategy,
|
|
default_params={
|
|
'period1': 12, 'multiplier1': 3.0,
|
|
'period2': 10, 'multiplier2': 1.0,
|
|
'period3': 11, 'multiplier3': 2.0
|
|
},
|
|
grid_params={
|
|
'multiplier1': [2.0, 3.0, 4.0],
|
|
'period1': [10, 12, 14],
|
|
'period2': 11, 'multiplier2': 2.0,
|
|
'period3': 12, 'multiplier3': 1.0
|
|
}
|
|
),
|
|
"regime": StrategyConfig(
|
|
strategy_class=RegimeReversionStrategy,
|
|
default_params={
|
|
'horizon': 96,
|
|
'z_window': 24,
|
|
'stop_loss': 0.06,
|
|
'take_profit': 0.05
|
|
},
|
|
grid_params={
|
|
'horizon': [72, 96, 120],
|
|
'stop_loss': [0.04, 0.06, 0.08]
|
|
}
|
|
)
|
|
}
|
|
|
|
|
|
# Module-level cache for the registry
|
|
_REGISTRY_CACHE: dict[str, StrategyConfig] | None = None
|
|
|
|
|
|
def get_registry() -> dict[str, StrategyConfig]:
|
|
"""Get the strategy registry, building it on first access."""
|
|
global _REGISTRY_CACHE
|
|
if _REGISTRY_CACHE is None:
|
|
_REGISTRY_CACHE = _build_registry()
|
|
return _REGISTRY_CACHE
|
|
|
|
|
|
def get_strategy_names() -> list[str]:
|
|
"""
|
|
Get list of available strategy names.
|
|
|
|
Returns:
|
|
List of strategy name strings
|
|
"""
|
|
return list(get_registry().keys())
|
|
|
|
|
|
def get_strategy(name: str, is_grid: bool = False) -> tuple[BaseStrategy, dict[str, Any]]:
|
|
"""
|
|
Create a strategy instance with appropriate parameters.
|
|
|
|
Args:
|
|
name: Strategy identifier (e.g., 'rsi', 'macross', 'meta_st')
|
|
is_grid: If True, return grid search parameters
|
|
|
|
Returns:
|
|
Tuple of (strategy instance, parameters dict)
|
|
|
|
Raises:
|
|
KeyError: If strategy name is not found in registry
|
|
"""
|
|
registry = get_registry()
|
|
|
|
if name not in registry:
|
|
available = ", ".join(registry.keys())
|
|
raise KeyError(f"Unknown strategy '{name}'. Available: {available}")
|
|
|
|
config = registry[name]
|
|
strategy = config.strategy_class()
|
|
params = config.grid_params if is_grid else config.default_params
|
|
|
|
return strategy, params.copy()
|