# source: https://raw.githubusercontent.com/karen-claros1212/ares-trading-system/62f87cb5035c19fd68c0b398052619f1ac141f67/ares/strategies/AresAIStrategy.py """ Github_karen_claros1212_ares_trading_system__AresAIStrategy__20260506_042859 — Freqtrade strategy with adaptive Market Regime detection. Integrates Market Regime features as FreqAI training features and adaptive signal filtering based on detected market conditions. Basado en: ARES v3.0 God Tier Upgrade — Salto 4 """ from functools import reduce from typing import Dict import pandas as pd import talib.abstract as ta from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter, CategoricalParameter from ares.analysis.market_regime import MarketRegimeDetector class Github_karen_claros1212_ares_trading_system__AresAIStrategy__20260506_042859(IStrategy): """ Strategy adaptativa que usa Market Regime Detector para ajustar parámetros de entrada y salida según el régimen actual. """ INTERFACE_VERSION = 3 timeframe = "1h" can_short = True # ROI dinámico — se ajusta por régimen minimal_roi = { "0": 0.10, "60": 0.05, "120": 0.02, "240": 0.01, } stoploss = -0.05 # ── Parámetros optimizables ────────────────────────────── buy_rsi = IntParameter(25, 40, default=30, space="buy") sell_rsi = IntParameter(65, 80, default=70, space="sell") buy_adx_threshold = IntParameter(20, 35, default=25, space="buy") regime_safety_multiplier = DecimalParameter( 0.3, 1.5, default=1.0, space="buy", description="Multiplicador de tamaño según régimen" ) # ── Features para FreqAI ───────────────────────────────── startup_candle_count = 100 def populate_indicators(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame: """Calcula indicadores técnicos + features de régimen de mercado.""" # Indicadores base dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14) dataframe['adx'] = ta.ADX(dataframe, timeperiod=14) dataframe['ema_20'] = ta.EMA(dataframe, timeperiod=20) dataframe['ema_50'] = ta.EMA(dataframe, timeperiod=50) dataframe['ema_200'] = ta.EMA(dataframe, timeperiod=200) # MACD macd = ta.MACD(dataframe) dataframe['macd'] = macd['macd'] dataframe['macd_signal'] = macd['macdsignal'] dataframe['macd_hist'] = macd['macdhist'] # Bollinger Bands bollinger = ta.BBANDS(dataframe, timeperiod=20, nbdevup=2.0, nbdevdn=2.0) dataframe['bb_upper'] = bollinger['upperband'] dataframe['bb_middle'] = bollinger['middleband'] dataframe['bb_lower'] = bollinger['lowerband'] dataframe['bb_width'] = ( (dataframe['bb_upper'] - dataframe['bb_lower']) / dataframe['bb_middle'] * 100 ) # ATR para stop-loss dinámico dataframe['atr_14'] = ta.ATR(dataframe, timeperiod=14) dataframe['atr'] = dataframe['atr_14'] dataframe['atr_pct'] = dataframe['atr_14'] / dataframe['close'] * 100 # SuperTrend dataframe['supertrend'], dataframe['supertrend_direction'] = self._calculate_supertrend( dataframe, period=10, multiplier=3.0 ) # ★ NUEVO: Market Regime como feature para FreqAI y agentes LLM regime_detector = MarketRegimeDetector() funding_rate = self._get_funding_rate(metadata.get('pair', '')) # Aplicar detección de régimen vela por vela regime_codes = { "strong_trend_up": 0, "weak_trend_up": 1, "ranging": 2, "weak_trend_down": 3, "strong_trend_down": 4, "high_volatility": 5, "crash": 6, "euphoria": 7 } dataframe['market_regime_code'] = 2 # default: ranging dataframe['regime_confidence'] = 0.0 dataframe['regime_size_multiplier'] = 1.0 dataframe['regime_allows_entry'] = 1 # Calcular régimen sobre ventanas deslizantes for i in range(100, len(dataframe)): window = dataframe.iloc[i - 100: i + 1] if len(window) < 50: continue try: regime_analysis = regime_detector.detect( df=window, funding_rate=funding_rate ) code = regime_codes.get(regime_analysis.regime.value, 2) dataframe.loc[dataframe.index[i], 'market_regime_code'] = code dataframe.loc[dataframe.index[i], 'regime_confidence'] = regime_analysis.confidence dataframe.loc[dataframe.index[i], 'regime_size_multiplier'] = ( regime_analysis.position_size_multiplier ) dataframe.loc[dataframe.index[i], 'regime_allows_entry'] = ( 0 if code in [6, 7] else 1 ) except Exception: pass # Contar señales recientes (para penalización overtrading en RL) dataframe['recent_trade_count_20'] = self._count_recent_signals(dataframe, window=20) return dataframe def populate_entry_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame: """Genera señales de entrada condicionadas al régimen.""" dataframe.loc[ ( # Condiciones de entrada base (dataframe['rsi'] < self.buy_rsi.value) & (dataframe['adx'] > self.buy_adx_threshold.value) & (dataframe['close'] > dataframe['ema_20']) & (dataframe['volume'] > dataframe['volume'].rolling(20).mean() * 1.2) & # ★ Filtro de régimen: NO entrar en crash o euphoria (dataframe['regime_allows_entry'] == 1) & # Régimen trending: más confianza ( (dataframe['market_regime_code'].isin([0, 1, 4])) | ( (dataframe['market_regime_code'] == 2) & # ranging (dataframe['regime_confidence'] > 0.7) # solo en extremos ) ) ), ['enter_long', 'enter_tag'] ] = (1, 'ares_long_regime_filtered') # Entradas short dataframe.loc[ ( (dataframe['rsi'] > self.sell_rsi.value) & (dataframe['adx'] > self.buy_adx_threshold.value) & (dataframe['close'] < dataframe['ema_20']) & (dataframe['regime_allows_entry'] == 1) & (dataframe['market_regime_code'].isin([3, 4, 5])) ), ['enter_short', 'enter_tag'] ] = (1, 'ares_short_regime_filtered') return dataframe def populate_exit_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame: """Genera señales de salida.""" dataframe.loc[ ( (dataframe['rsi'] > 75) | (dataframe['market_regime_code'] == 6) # crash detectado ), ['exit_long', 'exit_tag'] ] = (1, 'ares_exit_rsi_or_crash') dataframe.loc[ ( (dataframe['rsi'] < 30) | (dataframe['market_regime_code'] == 7) # euphoria ends ), ['exit_short', 'exit_tag'] ] = (1, 'ares_exit_rsi_or_euphoria') return dataframe def custom_stoploss(self, pair: str, trade, current_time, current_rate, current_profit, **kwargs) -> float: """Stop-loss dinámico según régimen.""" dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe is not None and not dataframe.empty: last_candle = dataframe.iloc[-1] regime_code = last_candle.get('market_regime_code', 2) # Ajustar stop según volatilidad del régimen if regime_code in [5, 6]: # alta volatilidad o crash return -0.08 # stop más amplio elif regime_code == 0: # strong trend up if current_profit > 0.03: return current_rate * 0.97 # trailing stop return -0.04 elif regime_code in [3, 4]: # bearish return -0.03 # stop más agresivo return self.stoploss def leverage(self, pair: str, current_time, current_rate, proposed_leverage, max_leverage, **kwargs) -> float: """Ajusta leverage según régimen.""" dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if dataframe is not None and not dataframe.empty: regime_code = dataframe.iloc[-1].get('market_regime_code', 2) if regime_code in [0, 4]: # tendencias fuertes return min(2.0, max_leverage) elif regime_code == 5: # alta volatilidad return 1.0 return 1.0 # ── Métodos auxiliares ──────────────────────────────────── @staticmethod def _calculate_supertrend(df: pd.DataFrame, period: int = 10, multiplier: float = 3.0): """Calcula SuperTrend.""" hl2 = (df['high'] + df['low']) / 2 atr = ta.ATR(df, timeperiod=period) upper_band = hl2 + (multiplier * atr) lower_band = hl2 - (multiplier * atr) supertrend = upper_band.copy() direction = pd.Series(1, index=df.index) for i in range(period, len(df)): if df['close'].iloc[i] > upper_band.iloc[i - 1]: direction.iloc[i] = 1 elif df['close'].iloc[i] < lower_band.iloc[i - 1]: direction.iloc[i] = -1 else: direction.iloc[i] = direction.iloc[i - 1] if direction.iloc[i] == 1 and lower_band.iloc[i] < lower_band.iloc[i - 1]: lower_band.iloc[i] = lower_band.iloc[i - 1] if direction.iloc[i] == -1 and upper_band.iloc[i] > upper_band.iloc[i - 1]: upper_band.iloc[i] = upper_band.iloc[i - 1] supertrend.iloc[i] = lower_band.iloc[i] if direction.iloc[i] == 1 else upper_band.iloc[i] return supertrend, direction @staticmethod def _get_funding_rate(pair: str) -> float: """Obtiene funding rate (place holder — integrar con exchange).""" return 0.0 @staticmethod def _count_recent_signals(df: pd.DataFrame, window: int = 20) -> pd.Series: """Cuenta señales recientes para penalización de overtrading.""" # Placeholder: retorna 0 si no hay columna de señal return pd.Series(0, index=df.index)