Advanced Multi-Dimensional Scalping Strategy for FreqTrade
Timeframe
5m
Direction
Long Only
Stoploss
-1.5%
Trailing Stop
Yes
ROI
0m: 2.0%, 5m: 1.5%, 10m: 1.0%, 20m: 0.5%
Interface Version
3
Startup Candles
N/A
Indicators
8
freqtrade/freqtrade-strategies
Strategy 003 author@: Gerald Lonlas github@: https://github.com/freqtrade/freqtrade-strategies
# 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 AdvancedScalpingStrategy(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