| """
|
| strategy/selector.py — Regime detector and strategy selector.
|
| Determines if market is trending or choppy, selects appropriate strategy.
|
| """
|
| from __future__ import annotations
|
|
|
| import logging
|
|
|
| import numpy as np
|
| import pandas as pd
|
|
|
| import config
|
| from contracts import TechnicalScore, PriceTarget
|
| from signals.technical import compute_atr, compute_ema
|
| from strategy.base import Strategy
|
| from strategy.momentum import MomentumStrategy
|
| from strategy.mean_reversion import MeanReversionStrategy
|
| from strategy.swing import SwingStrategy
|
|
|
| logger = logging.getLogger("trading_system.selector")
|
|
|
| _momentum = MomentumStrategy()
|
| _mean_reversion = MeanReversionStrategy()
|
| _swing = SwingStrategy()
|
|
|
|
|
| def detect_regime(df_1h: pd.DataFrame) -> str:
|
| """Detect market regime from hourly bars.
|
|
|
| Uses ADX-like directional calculation and ATR trend to determine:
|
| - "trending": strong directional movement → use momentum
|
| - "choppy": range-bound, low directional movement → use mean reversion
|
|
|
| Args:
|
| df_1h: 1-hour OHLCV bars (need at least 30 bars)
|
|
|
| Returns: "trending" or "choppy"
|
| """
|
| if df_1h is None or len(df_1h) < 30:
|
| return "trending"
|
|
|
| close = df_1h["close"]
|
|
|
|
|
| ema20 = compute_ema(close, 20)
|
| ema_vals = ema20.dropna()
|
| if len(ema_vals) >= 5:
|
| recent_slope = (float(ema_vals.iloc[-1]) - float(ema_vals.iloc[-5])) / float(ema_vals.iloc[-5])
|
| else:
|
| recent_slope = 0.0
|
|
|
|
|
| atr = compute_atr(df_1h, 14)
|
| if not atr.dropna().empty:
|
| atr_pct = float(atr.iloc[-1]) / float(close.iloc[-1])
|
| else:
|
| atr_pct = 0.0
|
|
|
|
|
| recent_range = float(close.tail(10).max() - close.tail(10).min())
|
| wider_range = float(close.tail(30).max() - close.tail(30).min())
|
| range_ratio = recent_range / wider_range if wider_range > 0 else 0.5
|
|
|
|
|
| trending_score = 0
|
| if abs(recent_slope) > 0.005:
|
| trending_score += 1
|
| if atr_pct > 0.008:
|
| trending_score += 1
|
| if range_ratio > 0.6:
|
| trending_score += 1
|
|
|
| regime = "trending" if trending_score >= 2 else "choppy"
|
| logger.info(
|
| "Regime detection: slope=%.4f, atr_pct=%.4f, range_ratio=%.3f -> %s",
|
| recent_slope, atr_pct, range_ratio, regime,
|
| )
|
| return regime
|
|
|
|
|
| def select_strategy(df_1h: pd.DataFrame | None = None) -> Strategy:
|
| """Select the best strategy based on config or market regime."""
|
|
|
| strategy_mode = getattr(config, "STRATEGY_MODE", "auto")
|
| if strategy_mode == "swing":
|
| logger.info("Selected: SwingStrategy (configured via STRATEGY_MODE)")
|
| return _swing
|
|
|
|
|
| regime = detect_regime(df_1h)
|
| if regime == "trending":
|
| logger.info("Selected: MomentumStrategy (trending regime)")
|
| return _momentum
|
| else:
|
| logger.info("Selected: MeanReversionStrategy (choppy regime)")
|
| return _mean_reversion
|
|
|
|
|
| def generate_signal(
|
| symbol: str,
|
| df_5m: pd.DataFrame,
|
| df_1h: pd.DataFrame | None = None,
|
| df_1d: pd.DataFrame | None = None,
|
| open_positions: dict[str, str] | None = None,
|
| corr_matrix: pd.DataFrame | None = None,
|
| ) -> tuple[TechnicalScore, PriceTarget]:
|
| """Auto-select strategy based on regime and generate signal."""
|
| strategy = select_strategy(df_1h)
|
| return strategy.generate_signal(
|
| symbol, df_5m, df_1h, df_1d, open_positions, corr_matrix
|
| )
|
|
|
