# source: https://raw.githubusercontent.com/remiotore/freqtrade/44beaeb6a420cd8e9f2e4ea93e11d6cfa192ee03/strategies/AdvancedScalpingStrategy.py # pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement # flake8: noqa: F401 # isort: skip_file # --- Do not remove these libs --- import numpy as np import pandas as pd from pandas import DataFrame from datetime import datetime from typing import Optional, Union from freqtrade.strategy import (BooleanParameter, CategoricalParameter, DecimalParameter, IStrategy, IntParameter) # -------------------------------- # Add your lib to import here import talib.abstract as ta import pandas_ta as pta from scipy import stats import warnings warnings.filterwarnings('ignore') class Github_remiotore_freqtrade__AdvancedScalpingStrategy__20260111_210550(IStrategy): """ Advanced Multi-Dimensional Scalping Strategy for FreqTrade Bu strateji şu gelişmiş teknikleri kullanır: - Adaptive Volatility Filtering - Multi-Timeframe Momentum Analysis - Dynamic Support/Resistance Detection - Volume-Price Divergence Analysis - Neural Network-Inspired Signal Generation """ # Strategy interface version INTERFACE_VERSION = 3 # Optimal timeframe for the strategy timeframe = '5m' # Can this strategy go short? can_short: bool = True # Minimal ROI designed for the strategy minimal_roi = { "0": 0.02, # 2% kar hedefi "5": 0.015, # 5 dakika sonra 1.5% "10": 0.01, # 10 dakika sonra 1% "20": 0.005, # 20 dakika sonra 0.5% "30": 0.0 # 30 dakika sonra başabaş } # Optimal stoploss stoploss = -0.015 # 1.5% stop loss # Trailing stoploss trailing_stop = True trailing_stop_positive = 0.005 trailing_stop_positive_offset = 0.01 trailing_only_offset_is_reached = True # Hyperopt parameters buy_params = { "avf_threshold": 0.6, "mtmo_threshold": 0.25, "signal_strength_min": 0.4, "volume_factor": 1.2, } sell_params = { "sell_signal_threshold": -0.3, "profit_multiplier": 1.5, "volume_confirm": True, } # Strategy parameters avf_threshold = DecimalParameter(0.4, 0.8, decimals=2, default=0.6, space="buy") mtmo_threshold = DecimalParameter(0.15, 0.4, decimals=2, default=0.25, space="buy") signal_strength_min = DecimalParameter(0.2, 0.6, decimals=2, default=0.4, space="buy") volume_factor = DecimalParameter(1.0, 2.0, decimals=1, default=1.2, space="buy") sell_signal_threshold = DecimalParameter(-0.5, -0.2, decimals=2, default=-0.3, space="sell") profit_multiplier = DecimalParameter(1.2, 2.0, decimals=1, default=1.5, space="sell") volume_confirm = BooleanParameter(default=True, space="sell") # Strategy specific parameters fast_period = 5 slow_period = 21 ultra_fast = 3 sensitivity = 2.0 def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Tüm indikatörleri hesaplar """ # === ADAPTIVE VOLATILITY FILTER === dataframe = self.adaptive_volatility_filter(dataframe) # === MULTI-TIMEFRAME MOMENTUM === dataframe = self.multi_timeframe_momentum(dataframe) # === DYNAMIC SUPPORT/RESISTANCE === dataframe = self.dynamic_support_resistance(dataframe) # === VOLUME-PRICE DIVERGENCE === dataframe = self.volume_price_divergence(dataframe) # === NEURAL INSPIRED SIGNAL === dataframe = self.neural_inspired_signal(dataframe) # === ADDITIONAL FILTERS === dataframe = self.additional_filters(dataframe) return dataframe def adaptive_volatility_filter(self, dataframe: DataFrame) -> DataFrame: """Adaptif Volatilite Filtresi""" # True Range hesaplama dataframe['tr1'] = dataframe['high'] - dataframe['low'] dataframe['tr2'] = abs(dataframe['high'] - dataframe['close'].shift(1)) dataframe['tr3'] = abs(dataframe['low'] - dataframe['close'].shift(1)) dataframe['true_range'] = dataframe[['tr1', 'tr2', 'tr3']].max(axis=1) # Adaptif ATR dataframe['atr_fast'] = dataframe['true_range'].rolling(self.fast_period).mean() dataframe['atr_slow'] = dataframe['true_range'].rolling(self.slow_period).mean() # Volatilite rejimi dataframe['vol_regime_raw'] = dataframe['atr_fast'] / dataframe['atr_slow'] # Normalize et (0-1 arası) rolling_min = dataframe['vol_regime_raw'].rolling(100).min() rolling_max = dataframe['vol_regime_raw'].rolling(100).max() dataframe['vol_regime'] = (dataframe['vol_regime_raw'] - rolling_min) / (rolling_max - rolling_min) dataframe['vol_regime'] = dataframe['vol_regime'].fillna(0.5) return dataframe def multi_timeframe_momentum(self, dataframe: DataFrame) -> DataFrame: """Çoklu Zaman Dilimi Momentum""" # Farklı periyotlarda RSI dataframe['rsi_ultra'] = ta.RSI(dataframe, timeperiod=self.ultra_fast) dataframe['rsi_fast'] = ta.RSI(dataframe, timeperiod=self.fast_period) dataframe['rsi_slow'] = ta.RSI(dataframe, timeperiod=self.slow_period) # Ağırlıklı momentum skoru dataframe['momentum_score'] = ( 0.5 * (dataframe['rsi_ultra'] - 50) / 50 + 0.3 * (dataframe['rsi_fast'] - 50) / 50 + 0.2 * (dataframe['rsi_slow'] - 50) / 50 ) return dataframe def dynamic_support_resistance(self, dataframe: DataFrame) -> DataFrame: """Dinamik Destek/Direnç Detektörü""" # Pivot hesaplama pivot_window = 5 # Son 20 barın max/min'i dataframe['resistance_level'] = dataframe['high'].rolling(20).max() dataframe['support_level'] = dataframe['low'].rolling(20).min() # Mesafe hesaplama dataframe['distance_to_resistance'] = (dataframe['close'] - dataframe['resistance_level']) / dataframe['close'] dataframe['distance_to_support'] = (dataframe['support_level'] - dataframe['close']) / dataframe['close'] # Birleşik sinyal dataframe['sr_signal'] = dataframe['distance_to_resistance'] + dataframe['distance_to_support'] return dataframe def volume_price_divergence(self, dataframe: DataFrame) -> DataFrame: """Hacim-Fiyat Uyumsuzluk Analizi""" # Fiyat değişimi dataframe['price_change'] = dataframe['close'].pct_change() # Hacim analizi dataframe['volume_ma'] = dataframe['volume'].rolling(20).mean() dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume_ma'] # OBV dataframe['obv'] = ta.OBV(dataframe) dataframe['obv_rsi'] = ta.RSI(dataframe['obv'], timeperiod=14) # Volume-Price Divergence skoru dataframe['vpd_score'] = ( 0.4 * np.sign(dataframe['price_change']) * dataframe['volume_ratio'] + 0.6 * (dataframe['obv_rsi'] - 50) / 50 ) return dataframe def neural_inspired_signal(self, dataframe: DataFrame) -> DataFrame: """YZ-esinlenmiş Sinyal Üretimi""" # Aktivasyon fonksiyonu def activation(x): return np.tanh(x * self.sensitivity) # Ağırlıklı kombinasyon weights = [0.25, 0.35, 0.20, 0.20] # AVF, MTMO, DSRD, VPDA # Ana sinyal hesaplama dataframe['raw_signal'] = ( weights[0] * dataframe['vol_regime'] + weights[1] * dataframe['momentum_score'] + weights[2] * dataframe['sr_signal'] + weights[3] * dataframe['vpd_score'] ) # Aktivasyon uygula dataframe['main_signal'] = activation(dataframe['raw_signal']) # Sinyal yumuşatma dataframe['main_signal'] = dataframe['main_signal'].rolling(3).mean() # Sinyal gücü dataframe['signal_strength'] = abs(dataframe['main_signal']) return dataframe def additional_filters(self, dataframe: DataFrame) -> DataFrame: """Ek filtreler ve konfirmasyon sinyalleri""" # Trend filtresi dataframe['ema_fast'] = ta.EMA(dataframe, timeperiod=8) dataframe['ema_slow'] = ta.EMA(dataframe, timeperiod=21) dataframe['trend_direction'] = np.where( dataframe['ema_fast'] > dataframe['ema_slow'], 1, -1 ) # MACD konfirmasyonu macd = ta.MACD(dataframe) dataframe['macd'] = macd['macd'] dataframe['macd_signal'] = macd['macdsignal'] dataframe['macd_hist'] = macd['macdhist'] # ADX (Trend gücü) dataframe['adx'] = ta.ADX(dataframe, timeperiod=14) # Bollinger Bands bollinger = ta.BBANDS(dataframe, timeperiod=20) dataframe['bb_upper'] = bollinger['upperband'] dataframe['bb_middle'] = bollinger['middleband'] dataframe['bb_lower'] = bollinger['lowerband'] dataframe['bb_percent'] = (dataframe['close'] - dataframe['bb_lower']) / (dataframe['bb_upper'] - dataframe['bb_lower']) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Long pozisyon giriş sinyalleri """ # Ana alım koşulları long_conditions = [ # Ana sinyal pozitif (dataframe['main_signal'] > self.mtmo_threshold.value), # Sinyal gücü yeterli (dataframe['signal_strength'] > self.signal_strength_min.value), # Volatilite rejimi uygun (dataframe['vol_regime'] > self.avf_threshold.value), # Hacim konfirmasyonu (dataframe['volume_ratio'] > self.volume_factor.value), # Trend yönü uygun (dataframe['trend_direction'] > 0), # MACD konfirmasyonu (dataframe['macd'] > dataframe['macd_signal']), # ADX trend gücü (dataframe['adx'] > 25), # Bollinger Bands pozisyonu (dataframe['bb_percent'] < 0.8), # Momentum artan (dataframe['momentum_score'] > dataframe['momentum_score'].shift(1)), ] # Tüm koşulları birleştir dataframe.loc[ reduce(lambda x, y: x & y, long_conditions), 'enter_long' ] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Long pozisyon çıkış sinyalleri """ # Ana satış koşulları exit_conditions = [ # Ana sinyal negatif (dataframe['main_signal'] < self.sell_signal_threshold.value), # Momentum tersine döndü (dataframe['momentum_score'] < -0.2), # MACD tersine döndü (dataframe['macd'] < dataframe['macd_signal']), # Hacim konfirmasyonu (isteğe bağlı) (~self.volume_confirm.value | (dataframe['volume_ratio'] > 1.0)), ] # Tüm koşulları birleştir dataframe.loc[ reduce(lambda x, y: x & y, exit_conditions), 'exit_long' ] = 1 return dataframe 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: """ Volatiliteye göre kaldıraç ayarlama """ # Mevcut volatilite rejimini al # Bu gerçek implementasyonda dataframe'den alınmalı # Şimdilik sabit değer döndürüyoruz return min(max_leverage, 5.0) # Maksimum 5x kaldıraç def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> float: """ Dinamik stop loss """ # ATR bazlı stop loss # Bu gerçek implementasyonda son ATR değerini kullanmalı if current_profit > 0.01: # %1 kar varsa # Trailing stop aktif et return -0.005 # %0.5 trailing stop return self.stoploss # Normal stop loss def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, current_time: datetime, entry_tag: Optional[str], side: str, **kwargs) -> bool: """ İşlem girişi son kontrol """ # Burada son dakika kontrolleri yapılabilir # Örneğin: spread kontrolü, volatilite kontrolü vs. return True def bot_loop_start(self, **kwargs) -> None: """ Her bot döngüsünün başında çalışır """ # Dinamik parametreler burada güncellenebilir pass # Utility function for condition combination def reduce(function, iterable, initializer=None): """Reduce function for combining conditions""" it = iter(iterable) if initializer is None: value = next(it) else: value = initializer for element in it: value = function(value, element) return value