59 lines
2.1 KiB
Python
59 lines
2.1 KiB
Python
"""Import dinamico delle classi Strategy da scripts/strategies/."""
|
|
from __future__ import annotations
|
|
|
|
import importlib
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from src.strategies.base import Strategy
|
|
|
|
PROJECT_ROOT = Path(__file__).resolve().parents[2]
|
|
STRATEGIES_DIR = PROJECT_ROOT / "scripts" / "strategies"
|
|
|
|
_REGISTRY: dict[str, type[Strategy]] = {}
|
|
|
|
# Solo strategie con edge netto validato out-of-sample (fee-aware).
|
|
# La famiglia squeeze-breakout (SQ/MT/ML/AD/CM/PD) e' stata spostata in
|
|
# scripts/waste/: l'edge storico era un artefatto di look-ahead
|
|
# (vedi scripts/analysis/oos_validation.py).
|
|
MODULE_MAP = {
|
|
"DIP01_dip_buy": ("DIP01_dip_buy", "Dip01DipBuy"),
|
|
"MR01_bollinger_fade": ("MR01_bollinger_fade", "BollingerFade"),
|
|
"MR02_donchian_fade": ("MR02_donchian_fade", "DonchianFade"),
|
|
"MR07_return_reversal": ("MR07_return_reversal", "ReturnReversal"),
|
|
# SH01 Shape-ML: generate_signals fa walk-forward (riallena il modello) -> pesante
|
|
# per-tick. Caricabile per backtest; per il live serve un worker con retraining
|
|
# periodico (come il legacy signal_engine), NON lo StrategyWorker a regola fissa.
|
|
"SH01_shape_ml": ("SH01_shape_ml", "ShapeMLStrategy"),
|
|
}
|
|
|
|
|
|
def load_strategy(name: str) -> Strategy:
|
|
"""Carica e istanzia una Strategy per nome."""
|
|
if name in _REGISTRY:
|
|
return _REGISTRY[name]()
|
|
|
|
if name not in MODULE_MAP:
|
|
raise ValueError(f"Strategia sconosciuta: {name}. Disponibili: {list(MODULE_MAP)}")
|
|
|
|
module_file, class_name = MODULE_MAP[name]
|
|
module_path = STRATEGIES_DIR / f"{module_file}.py"
|
|
|
|
if not module_path.exists():
|
|
raise FileNotFoundError(f"File strategia non trovato: {module_path}")
|
|
|
|
if str(PROJECT_ROOT) not in sys.path:
|
|
sys.path.insert(0, str(PROJECT_ROOT))
|
|
|
|
spec = importlib.util.spec_from_file_location(f"strategies.{module_file}", module_path)
|
|
module = importlib.util.module_from_spec(spec)
|
|
spec.loader.exec_module(module)
|
|
|
|
cls = getattr(module, class_name)
|
|
_REGISTRY[name] = cls
|
|
return cls()
|
|
|
|
|
|
def list_available() -> list[str]:
|
|
return list(MODULE_MAP.keys())
|