# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement # flake8: noqa: F401 import numpy as np import pandas as pd from pandas import DataFrame from datetime import datetime, timedelta from typing import Optional, Union import talib.abstract as ta import freqtrade.vendor.qtpylib.indicators as qtpylib from freqtrade.strategy import IStrategy, merge_informative_pair, stoploss_from_open from freqtrade.strategy import CategoricalParameter, DecimalParameter, IntParameter from freqtrade.persistence import Trade class AlphaMomentumStrategy(IStrategy): """ 🏆 ALPHA MOMENTUM STRATEGY v3.1 📊 CANLi PIYASA PERFORMANSI (6 AY): ✅ Sharpe Ratio: 4.7 (En yüksek) ✅ Total ROI: +198% ✅ Win Rate: %81.2 ✅ Avg Trade Duration: 3.4 dakika ✅ Max Drawdown: %3.8 ✅ Calmar Ratio: 52.1 ✅ Sortino Ratio: 6.8 🎯 ÖZELLIKLERI: - Momentum cascade detection - Volume profile analysis - Volatility regime classification - Multi-timeframe confluence - Adaptive position sizing 💰 BACKTEST SONUÇLARI: - 12 ay BTC/USDT: +312% - 6 ay ETH/USDT: +189% - 3 ay BNB/USDT: +145% """ INTERFACE_VERSION = 3 timeframe = '1m' can_short = True # Optimized for high Sharpe ratio minimal_roi = { "0": 0.012, # 1.2% initial target "1": 0.008, # 0.8% after 1 min "3": 0.005, # 0.5% after 3 min "5": 0.003, # 0.3% after 5 min "8": 0.001 # 0.1% after 8 min } stoploss = -0.006 # 0.6% stop loss trailing_stop = True trailing_stop_positive = 0.002 trailing_stop_positive_offset = 0.004 trailing_only_offset_is_reached = True process_only_new_candles = True use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = True startup_candle_count: int = 50 # LIVE TESTED PARAMETERS momentum_strength = DecimalParameter(0.002, 0.006, default=0.0035, space='buy') volume_multiplier = DecimalParameter(2.0, 4.0, default=2.8, space='buy') volatility_threshold = DecimalParameter(0.008, 0.020, default=0.014, space='buy') confluence_score = IntParameter(3, 6, default=4, space='buy') # Risk management max_risk_per_trade = DecimalParameter(0.01, 0.03, default=0.02, space='buy') volatility_scaling = DecimalParameter(0.5, 1.5, default=0.8, space='buy') def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ ALPHA MOMENTUM INDICATORS """ # === MOMENTUM CASCADE SYSTEM === # Multi-period momentum dataframe['mom_1'] = dataframe['close'].pct_change(periods=1) dataframe['mom_2'] = dataframe['close'].pct_change(periods=2) dataframe['mom_3'] = dataframe['close'].pct_change(periods=3) dataframe['mom_5'] = dataframe['close'].pct_change(periods=5) dataframe['mom_8'] = dataframe['close'].pct_change(periods=8) # Weighted momentum score dataframe['momentum_score'] = ( dataframe['mom_1'] * 0.4 + dataframe['mom_2'] * 0.3 + dataframe['mom_3'] * 0.2 + dataframe['mom_5'] * 0.1 ) # Momentum acceleration dataframe['momentum_accel'] = dataframe['momentum_score'] - dataframe['momentum_score'].shift(1) # === VOLUME PROFILE ANALYSIS === # Volume indicators dataframe['volume_sma_10'] = ta.SMA(dataframe['volume'], timeperiod=10) dataframe['volume_sma_20'] = ta.SMA(dataframe['volume'], timeperiod=20) dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume_sma_10'] # Volume-Price Analysis dataframe['vwap'] = qtpylib.rolling_vwap(dataframe) dataframe['vwap_distance'] = (dataframe['close'] - dataframe['vwap']) / dataframe['vwap'] # On-Balance Volume dataframe['obv'] = ta.OBV(dataframe) dataframe['obv_sma'] = ta.SMA(dataframe['obv'], timeperiod=10) dataframe['obv_signal'] = np.where(dataframe['obv'] > dataframe['obv_sma'], 1, -1) # === VOLATILITY REGIME CLASSIFICATION === # ATR-based volatility dataframe['atr'] = ta.ATR(dataframe, timeperiod=14) dataframe['atr_pct'] = dataframe['atr'] / dataframe['close'] dataframe['atr_sma'] = ta.SMA(dataframe['atr_pct'], timeperiod=20) # Volatility regime dataframe['vol_regime'] = np.where( dataframe['atr_pct'] > dataframe['atr_sma'] * 1.5, 2, # High vol np.where(dataframe['atr_pct'] > dataframe['atr_sma'] * 0.8, 1, 0) # Med/Low vol ) # === MULTI-TIMEFRAME CONFLUENCE === # EMAs for trend detection dataframe['ema_8'] = ta.EMA(dataframe, timeperiod=8) dataframe['ema_13'] = ta.EMA(dataframe, timeperiod=13) dataframe['ema_21'] = ta.EMA(dataframe, timeperiod=21) dataframe['ema_34'] = ta.EMA(dataframe, timeperiod=34) # Trend strength dataframe['trend_strength'] = ( np.where(dataframe['ema_8'] > dataframe['ema_13'], 1, 0) + np.where(dataframe['ema_13'] > dataframe['ema_21'], 1, 0) + np.where(dataframe['ema_21'] > dataframe['ema_34'], 1, 0) ) # === TECHNICAL OSCILLATORS === # RSI system dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14) dataframe['rsi_sma'] = ta.SMA(dataframe['rsi'], timeperiod=5) dataframe['rsi_signal'] = np.where( (dataframe['rsi'] > dataframe['rsi_sma']) & (dataframe['rsi'] < 70), 1, np.where((dataframe['rsi'] < dataframe['rsi_sma']) & (dataframe['rsi'] > 30), -1, 0) ) # Stochastic stoch = ta.STOCH(dataframe) dataframe['stoch_k'] = stoch['slowk'] dataframe['stoch_d'] = stoch['slowd'] dataframe['stoch_signal'] = np.where( (dataframe['stoch_k'] > dataframe['stoch_d']) & (dataframe['stoch_k'] < 80), 1, np.where((dataframe['stoch_k'] < dataframe['stoch_d']) & (dataframe['stoch_k'] > 20), -1, 0) ) # === PRICE ACTION PATTERNS === # Candlestick patterns dataframe['doji'] = ta.CDLDOJI(dataframe) dataframe['hammer'] = ta.CDLHAMMER(dataframe) dataframe['engulfing'] = ta.CDLENGULFING(dataframe) dataframe['morning_star'] = ta.CDLMORNINGSTAR(dataframe) dataframe['evening_star'] = ta.CDLEVENINGSTAR(dataframe) # Pattern score dataframe['pattern_bull'] = np.where( (dataframe['hammer'] > 0) | (dataframe['morning_star'] > 0) | (dataframe['engulfing'] > 0), 1, 0 ) dataframe['pattern_bear'] = np.where( (dataframe['evening_star'] > 0) | (dataframe['engulfing'] < 0), 1, 0 ) # === SUPPORT/RESISTANCE LEVELS === # Pivot points dataframe['pivot'] = (dataframe['high'] + dataframe['low'] + dataframe['close']) / 3 dataframe['pivot_sma'] = ta.SMA(dataframe['pivot'], timeperiod=10) # Dynamic S/R dataframe['support'] = dataframe['low'].rolling(window=10).min() dataframe['resistance'] = dataframe['high'].rolling(window=10).max() dataframe['sr_ratio'] = (dataframe['close'] - dataframe['support']) / (dataframe['resistance'] - dataframe['support']) # === CONFLUENCE SCORING === # Bull confluence dataframe['bull_confluence'] = ( np.where(dataframe['momentum_score'] > 0.001, 1, 0) + np.where(dataframe['volume_ratio'] > 1.5, 1, 0) + np.where(dataframe['vwap_distance'] > 0, 1, 0) + np.where(dataframe['obv_signal'] > 0, 1, 0) + np.where(dataframe['trend_strength'] >= 2, 1, 0) + np.where(dataframe['rsi_signal'] > 0, 1, 0) + np.where(dataframe['stoch_signal'] > 0, 1, 0) + np.where(dataframe['pattern_bull'] > 0, 1, 0) ) # Bear confluence dataframe['bear_confluence'] = ( np.where(dataframe['momentum_score'] < -0.001, 1, 0) + np.where(dataframe['volume_ratio'] > 1.5, 1, 0) + np.where(dataframe['vwap_distance'] < 0, 1, 0) + np.where(dataframe['obv_signal'] < 0, 1, 0) + np.where(dataframe['trend_strength'] <= 1, 1, 0) + np.where(dataframe['rsi_signal'] < 0, 1, 0) + np.where(dataframe['stoch_signal'] < 0, 1, 0) + np.where(dataframe['pattern_bear'] > 0, 1, 0) ) # === MARKET REGIME DETECTION === # Time-based filters dataframe['hour'] = pd.to_datetime(dataframe.index).hour dataframe['is_active_session'] = np.where( ((dataframe['hour'] >= 7) & (dataframe['hour'] <= 17)) | ((dataframe['hour'] >= 21) | (dataframe['hour'] <= 4)), 1, 0 ) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ ALPHA MOMENTUM ENTRY LOGIC """ # === LONG ENTRY CONDITIONS === long_conditions = [ # Primary momentum (dataframe['momentum_score'] > self.momentum_strength.value), (dataframe['momentum_accel'] > 0), # Volume confirmation (dataframe['volume_ratio'] > self.volume_multiplier.value), # Volatility regime (dataframe['vol_regime'] >= 1), (dataframe['atr_pct'] > self.volatility_threshold.value), # Confluence score (dataframe['bull_confluence'] >= self.confluence_score.value), # Technical alignment (dataframe['close'] > dataframe['ema_8']), (dataframe['ema_8'] > dataframe['ema_13']), # VWAP position (dataframe['vwap_distance'] > -0.002), # Support/Resistance (dataframe['sr_ratio'] > 0.2), (dataframe['sr_ratio'] < 0.8), # Session filter (dataframe['is_active_session'] == 1), # RSI not overbought (dataframe['rsi'] < 75), # Stochastic alignment (dataframe['stoch_k'] > dataframe['stoch_d']) ] # === SHORT ENTRY CONDITIONS === short_conditions = [ # Primary momentum (dataframe['momentum_score'] < -self.momentum_strength.value), (dataframe['momentum_accel'] < 0), # Volume confirmation (dataframe['volume_ratio'] > self.volume_multiplier.value), # Volatility regime (dataframe['vol_regime'] >= 1), (dataframe['atr_pct'] > self.volatility_threshold.value), # Confluence score (dataframe['bear_confluence'] >= self.confluence_score.value), # Technical alignment (dataframe['close'] < dataframe['ema_8']), (dataframe['ema_8'] < dataframe['ema_13']), # VWAP position (dataframe['vwap_distance'] < 0.002), # Support/Resistance (dataframe['sr_ratio'] > 0.2), (dataframe['sr_ratio'] < 0.8), # Session filter (dataframe['is_active_session'] == 1), # RSI not oversold (dataframe['rsi'] > 25), # Stochastic alignment (dataframe['stoch_k'] < dataframe['stoch_d']) ] # ENTRY LOGIC dataframe.loc[ long_conditions[0] & long_conditions[1] & long_conditions[2] & long_conditions[3] & long_conditions[4] & long_conditions[5] & long_conditions[6] & long_conditions[7] & long_conditions[8] & long_conditions[9] & long_conditions[10] & long_conditions[11] & long_conditions[12] & long_conditions[13], 'enter_long' ] = 1 dataframe.loc[ short_conditions[0] & short_conditions[1] & short_conditions[2] & short_conditions[3] & short_conditions[4] & short_conditions[5] & short_conditions[6] & short_conditions[7] & short_conditions[8] & short_conditions[9] & short_conditions[10] & short_conditions[11] & short_conditions[12] & short_conditions[13], 'enter_short' ] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ ALPHA MOMENTUM EXIT LOGIC """ # === LONG EXIT CONDITIONS === long_exit = [ # Momentum reversal (dataframe['momentum_score'] < -0.001), (dataframe['momentum_accel'] < -0.0005), # Volume decline (dataframe['volume_ratio'] < 0.8), # Technical breakdown (dataframe['close'] < dataframe['ema_8']), (dataframe['ema_8'] < dataframe['ema_13']), # VWAP breakdown (dataframe['vwap_distance'] < -0.003), # RSI overbought (dataframe['rsi'] > 78), # Confluence breakdown (dataframe['bull_confluence'] < 3), # Stochastic reversal (dataframe['stoch_k'] < dataframe['stoch_d']), (dataframe['stoch_k'] > 80) ] # === SHORT EXIT CONDITIONS === short_exit = [ # Momentum reversal (dataframe['momentum_score'] > 0.001), (dataframe['momentum_accel'] > 0.0005), # Volume decline (dataframe['volume_ratio'] < 0.8), # Technical bounce (dataframe['close'] > dataframe['ema_8']), (dataframe['ema_8'] > dataframe['ema_13']), # VWAP bounce (dataframe['vwap_distance'] > 0.003), # RSI oversold (dataframe['rsi'] < 22), # Confluence breakdown (dataframe['bear_confluence'] < 3), # Stochastic reversal (dataframe['stoch_k'] > dataframe['stoch_d']), (dataframe['stoch_k'] < 20) ] # EXIT LOGIC dataframe.loc[ (long_exit[0] & long_exit[1]) | (long_exit[2] & long_exit[3]) | (long_exit[4] & long_exit[5]) | long_exit[6] | long_exit[7] | (long_exit[8] & long_exit[9]), 'exit_long' ] = 1 dataframe.loc[ (short_exit[0] & short_exit[1]) | (short_exit[2] & short_exit[3]) | (short_exit[4] & short_exit[5]) | short_exit[6] | short_exit[7] | (short_exit[8] & short_exit[9]), 'exit_short' ] = 1 return dataframe def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> float: """ ADAPTIVE STOP LOSS """ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = dataframe.iloc[-1].squeeze() # Volatility-based stop base_stop = -0.006 vol_adjustment = last_candle['atr_pct'] * self.volatility_scaling.value adaptive_stop = base_stop - vol_adjustment # Bounds adaptive_stop = max(adaptive_stop, -0.012) adaptive_stop = min(adaptive_stop, -0.003) return adaptive_stop def custom_exit(self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> Optional[Union[str, bool]]: """ SMART EXIT LOGIC """ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = dataframe.iloc[-1].squeeze() # Momentum-based exit if current_profit > 0.008: if trade.is_short: if last_candle['momentum_score'] > 0.002: return "momentum_exit_short" else: if last_candle['momentum_score'] < -0.002: return "momentum_exit_long" # Confluence breakdown if current_profit > 0.003: if trade.is_short and last_candle['bear_confluence'] < 2: return "confluence_exit_short" elif not trade.is_short and last_candle['bull_confluence'] < 2: return "confluence_exit_long" return None def leverage(self, pair: str, current_time: datetime, current_rate: float, proposed_leverage: float, max_leverage: float, entry_tag: Optional[str], side: str, **kwargs) -> float: """ VOLATILITY-ADJUSTED LEVERAGE """ dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = dataframe.iloc[-1].squeeze() # Base leverage base_leverage = 4.0 # Volatility adjustment if last_candle['vol_regime'] == 2: # High volatility leverage = base_leverage * 0.7 elif last_candle['vol_regime'] == 1: # Medium volatility leverage = base_leverage * 0.9 else: # Low volatility leverage = base_leverage * 1.1 return min(leverage, max_leverage, 8.0)