feat: multi-strategy paper trader — N strategie in parallelo su testnet
- src/live/multi_runner.py: orchestratore con fetch raggruppato per asset/tf - src/live/strategy_worker.py: worker indipendente con stato persistente JSONL - src/live/strategy_loader.py: import dinamico classi Strategy - strategies.yml: config dichiarativa con defaults e override per strategia - Docker: container unico, strategies.yml montato come volume read-only - Supporta hot-add: aggiungi riga YAML + restart, storico intatto - Ogni strategia: €1000 USDC virtuale, equity tracking, Telegram notify Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
@@ -0,0 +1,51 @@
|
||||
"""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]] = {}
|
||||
|
||||
MODULE_MAP = {
|
||||
"SQ01_squeeze_base": ("SQ01_squeeze_base", "SqueezeBase"),
|
||||
"SQ02_antifake_vol": ("SQ02_squeeze_antifake_vol", "SqueezeAntifakeVol"),
|
||||
"SQ03_filtered": ("SQ03_squeeze_all_filters", "SqueezeFiltered"),
|
||||
"SQ04_ultimate": ("SQ04_squeeze_ultimate", "SqueezeUltimate"),
|
||||
"ML01_squeeze_gbm": ("ML01_squeeze_gbm", "SqueezeGBM"),
|
||||
}
|
||||
|
||||
|
||||
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())
|
||||
Reference in New Issue
Block a user