Enhanced Professional High-Frequency Scalping Strategy with AI Integration
Timeframe
1m
Direction
Long Only
Stoploss
-2.0%
Trailing Stop
No
ROI
0m: 2.5%, 1m: 2.0%, 2m: 1.5%, 5m: 1.0%
Interface Version
3
Startup Candles
N/A
Indicators
13
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, timedelta
from typing import Optional, Union
import logging
from freqtrade.strategy import (BooleanParameter, CategoricalParameter, DecimalParameter,
IntParameter, IStrategy, merge_informative_pair)
# --------------------------------
# Add your lib to import here
import talib.abstract as ta
import freqtrade.vendor.qtpylib.indicators as qtpylib
logger = logging.getLogger(__name__)
class ClaudeScalpingEnhanced(IStrategy):
"""
Enhanced Professional High-Frequency Scalping Strategy with AI Integration
New Features:
- FreqAI integration support
- Enhanced hyperopt parameters
- Advanced risk management
- ML-ready feature engineering
- Smart money flow detection
- Volatility regime filtering
- Multi-timeframe momentum analysis
- Advanced order flow indicators
- Ensemble signal weighting
- Real-time performance tracking
Optimized for:
- 1-5 minute timeframes
- High leverage (10-20x)
- AI/ML model integration
- Hyperopt optimization
- Production trading
"""
INTERFACE_VERSION = 3
# Enhanced ROI with more granular steps
minimal_roi = {
"0": 0.025, # 2.5% at any time
"1": 0.02, # 2% after 1 minute
"2": 0.015, # 1.5% after 2 minutes
"5": 0.01, # 1% after 5 minutes
"10": 0.008, # 0.8% after 10 minutes
"15": 0.005, # 0.5% after 15 minutes
"30": 0.003, # 0.3% after 30 minutes
"60": 0.001 # 0.1% after 1 hour
}
# Dynamic stoploss based on volatility
stoploss = -0.02 # 2% base stoploss
# Primary timeframe
timeframe = '1m'
# Enhanced informative timeframes
inf_5m = '5m'
inf_15m = '15m'
inf_1h = '1h'
# Performance optimizations
process_only_new_candles = True
use_exit_signal = True
exit_profit_only = True
ignore_roi_if_entry_signal = False
# Enhanced startup candles for better signal quality
startup_candle_count: int = 100
# === Enhanced Strategy Parameters - Hyperopt Enabled ===
# === Core Momentum Parameters ===
rsi_period = IntParameter(5, 21, default=7, space='buy', optimize=True)
rsi_entry_long = IntParameter(30, 55, default=40, space='buy', optimize=True)
rsi_entry_short = IntParameter(45, 70, default=60, space='sell', optimize=True)
rsi_exit_long = IntParameter(65, 90, default=75, space='sell', optimize=True)
rsi_exit_short = IntParameter(10, 35, default=25, space='sell', optimize=True)
# === Enhanced Bollinger Bands ===
bb_period = IntParameter(10, 30, default=20, space='buy', optimize=True)
bb_std = DecimalParameter(1.5, 3.0, default=2.0, space='buy', optimize=True)
bb_squeeze_threshold = DecimalParameter(0.1, 0.5, default=0.2, space='buy', optimize=True)
# === Enhanced MACD ===
macd_fast = IntParameter(6, 18, default=12, space='buy', optimize=True)
macd_slow = IntParameter(20, 35, default=26, space='buy', optimize=True)
macd_signal = IntParameter(5, 15, default=9, space='buy', optimize=True)
# === Multi-Timeframe EMAs ===
ema_fast = IntParameter(3, 15, default=8, space='buy', optimize=True)
ema_medium = IntParameter(12, 25, default=21, space='buy', optimize=True)
ema_slow = IntParameter(25, 55, default=50, space='buy', optimize=True)
# === Enhanced Volume Analysis ===
volume_factor = DecimalParameter(1.2, 4.0, default=2.0, space='buy', optimize=True)
volume_ma_period = IntParameter(5, 25, default=10, space='buy', optimize=True)
volume_spike_threshold = DecimalParameter(2.5, 5.0, default=3.0, space='buy', optimize=True)
# === Advanced ADX Parameters ===
adx_period = IntParameter(8, 25, default=14, space='buy', optimize=True)
adx_threshold = IntParameter(15, 50, default=25, space='buy', optimize=True)
adx_strong_trend = IntParameter(35, 60, default=45, space='buy', optimize=True)
# === Enhanced Stochastic ===
stoch_k = IntParameter(8, 25, default=14, space='buy', optimize=True)
stoch_d = IntParameter(3, 10, default=3, space='buy', optimize=True)
stoch_smooth = IntParameter(3, 10, default=3, space='buy', optimize=True)
stoch_oversold = IntParameter(15, 25, default=20, space='buy', optimize=True)
stoch_overbought = IntParameter(75, 85, default=80, space='sell', optimize=True)
# === Volatility Parameters ===
atr_period = IntParameter(7, 21, default=14, space='buy', optimize=True)
volatility_threshold = DecimalParameter(1.0, 5.0, default=2.5, space='buy', optimize=True)
# === Momentum Oscillators ===
cci_period = IntParameter(10, 25, default=20, space='buy', optimize=True)
cci_oversold = IntParameter(-120, -80, default=-100, space='buy', optimize=True)
cci_overbought = IntParameter(80, 120, default=100, space='sell', optimize=True)
# === Williams %R ===
williams_period = IntParameter(10, 25, default=14, space='buy', optimize=True)
williams_oversold = IntParameter(-90, -70, default=-80, space='buy', optimize=True)
williams_overbought = IntParameter(-30, -10, default=-20, space='sell', optimize=True)
# === Enhanced Exit Parameters ===
trailing_stop = BooleanParameter(default=True, space='sell', optimize=True)
trailing_stop_positive = DecimalParameter(0.003, 0.025, default=0.01, space='sell', optimize=True)
trailing_stop_positive_offset = DecimalParameter(0.008, 0.035, default=0.015, space='sell', optimize=True)
# === Dynamic Risk Management ===
use_dynamic_stoploss = BooleanParameter(default=True, space='sell', optimize=True)
atr_stoploss_multiplier = DecimalParameter(1.5, 3.5, default=2.0, space='sell', optimize=True)
max_drawdown_protection = DecimalParameter(0.05, 0.15, default=0.10, space='sell', optimize=True)
# === Signal Weighting ===
momentum_weight = DecimalParameter(0.1, 0.5, default=0.25, space='buy', optimize=True)
volume_weight = DecimalParameter(0.1, 0.5, default=0.25, space='buy', optimize=True)
trend_weight = DecimalParameter(0.1, 0.5, default=0.3, space='buy', optimize=True)
volatility_weight = DecimalParameter(0.1, 0.4, default=0.2, space='buy', optimize=True)
# === FreqAI Integration ===
use_freqai_signals = BooleanParameter(default=False, space='buy', optimize=False)
freqai_signal_weight = DecimalParameter(0.1, 0.8, default=0.4, space='buy', optimize=True)
# === Risk Management ===
max_open_trades = 3 # Reduced for better risk management
position_adjustment_enable = False
def informative_pairs(self):
"""Enhanced informative pairs for multi-timeframe analysis"""
pairs = self.dp.current_whitelist()
informative_pairs = [(pair, self.inf_5m) for pair in pairs]
informative_pairs += [(pair, self.inf_15m) for pair in pairs]
informative_pairs += [(pair, self.inf_1h) for pair in pairs]
return informative_pairs
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""Enhanced indicators with AI-ready feature engineering"""
# === Basic Price Action ===
dataframe['hl2'] = (dataframe['high'] + dataframe['low']) / 2
dataframe['hlc3'] = (dataframe['high'] + dataframe['low'] + dataframe['close']) / 3
dataframe['ohlc4'] = (dataframe['open'] + dataframe['high'] + dataframe['low'] + dataframe['close']) / 4
# === Enhanced EMAs ===
dataframe['ema_fast'] = ta.EMA(dataframe, timeperiod=self.ema_fast.value)
dataframe['ema_medium'] = ta.EMA(dataframe, timeperiod=self.ema_medium.value)
dataframe['ema_slow'] = ta.EMA(dataframe, timeperiod=self.ema_slow.value)
# EMA relationships
dataframe['ema_fast_above_medium'] = dataframe['ema_fast'] > dataframe['ema_medium']
dataframe['ema_medium_above_slow'] = dataframe['ema_medium'] > dataframe['ema_slow']
dataframe['emas_aligned_bullish'] = (dataframe['ema_fast_above_medium'] &
dataframe['ema_medium_above_slow'])
# === Enhanced RSI ===
dataframe['rsi'] = ta.RSI(dataframe, timeperiod=self.rsi_period.value)
dataframe['rsi_ma'] = ta.SMA(dataframe['rsi'], timeperiod=5)
dataframe['rsi_slope'] = dataframe['rsi'] - dataframe['rsi'].shift(1)
# === Enhanced Bollinger Bands ===
bollinger = qtpylib.bollinger_bands(dataframe['close'],
window=self.bb_period.value,
stds=self.bb_std.value)
dataframe['bb_lower'] = bollinger['lower']
dataframe['bb_middle'] = bollinger['mid']
dataframe['bb_upper'] = bollinger['upper']
dataframe['bb_percent'] = ((dataframe['close'] - dataframe['bb_lower']) /
(dataframe['bb_upper'] - dataframe['bb_lower']))
dataframe['bb_width'] = ((dataframe['bb_upper'] - dataframe['bb_lower']) /
dataframe['bb_middle'])
dataframe['bb_squeeze'] = dataframe['bb_width'] < self.bb_squeeze_threshold.value
# === Enhanced MACD ===
macd = ta.MACD(dataframe,
fastperiod=self.macd_fast.value,
slowperiod=self.macd_slow.value,
signalperiod=self.macd_signal.value)
dataframe['macd'] = macd['macd']
dataframe['macdsignal'] = macd['macdsignal']
dataframe['macdhist'] = macd['macdhist']
dataframe['macd_momentum'] = dataframe['macdhist'] - dataframe['macdhist'].shift(1)
# === Enhanced Stochastic ===
stoch = ta.STOCH(dataframe,
fastk_period=self.stoch_k.value,
slowk_period=self.stoch_d.value,
slowd_period=self.stoch_smooth.value)
dataframe['stoch_k'] = stoch['slowk']
dataframe['stoch_d'] = stoch['slowd']
dataframe['stoch_crossover'] = qtpylib.crossed_above(dataframe['stoch_k'], dataframe['stoch_d'])
dataframe['stoch_crossunder'] = qtpylib.crossed_below(dataframe['stoch_k'], dataframe['stoch_d'])
# === Enhanced ADX ===
dataframe['adx'] = ta.ADX(dataframe, timeperiod=self.adx_period.value)
dataframe['di_plus'] = ta.PLUS_DI(dataframe, timeperiod=self.adx_period.value)
dataframe['di_minus'] = ta.MINUS_DI(dataframe, timeperiod=self.adx_period.value)
dataframe['di_diff'] = dataframe['di_plus'] - dataframe['di_minus']
# === CCI ===
dataframe['cci'] = ta.CCI(dataframe, timeperiod=self.cci_period.value)
# === Williams %R ===
dataframe['williams_r'] = ta.WILLR(dataframe, timeperiod=self.williams_period.value)
# === Enhanced Volume Analysis ===
dataframe['volume_sma'] = ta.SMA(dataframe['volume'], timeperiod=self.volume_ma_period.value)
dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume_sma']
dataframe['high_volume'] = dataframe['volume_ratio'] > self.volume_factor.value
dataframe['volume_spike'] = dataframe['volume_ratio'] > self.volume_spike_threshold.value
# Volume flow indicators
dataframe['mfi'] = ta.MFI(dataframe, timeperiod=14)
dataframe['ad'] = ta.AD(dataframe)
dataframe['obv'] = ta.OBV(dataframe)
# === Enhanced VWAP ===
dataframe['vwap'] = qtpylib.rolling_vwap(dataframe, window=20)
dataframe['vwap_distance'] = (dataframe['close'] - dataframe['vwap']) / dataframe['vwap']
# === Enhanced ATR and Volatility ===
dataframe['atr'] = ta.ATR(dataframe, timeperiod=self.atr_period.value)
dataframe['atr_percent'] = (dataframe['atr'] / dataframe['close']) * 100
dataframe['volatility_regime'] = np.where(
dataframe['atr_percent'] > self.volatility_threshold.value, 'high', 'normal'
)
# === Advanced Price Action Analysis ===
dataframe = self.calculate_advanced_signals(dataframe)
# === Higher Timeframe Context ===
dataframe = self.populate_higher_timeframe_indicators(dataframe, metadata)
# === Signal Scoring ===
dataframe = self.calculate_signal_scores(dataframe)
return dataframe
def calculate_advanced_signals(self, dataframe: DataFrame) -> DataFrame:
"""Calculate advanced scalping signals and patterns"""
# === Price Momentum ===
for period in [3, 5, 10]:
dataframe[f'price_change_{period}'] = dataframe['close'].pct_change(period)
dataframe[f'price_momentum_{period}'] = dataframe['close'].pct_change(period)
# === Volume Momentum ===
dataframe['volume_change'] = dataframe['volume'].pct_change()
dataframe['volume_momentum'] = (dataframe['volume'].rolling(3).mean() /
dataframe['volume'].rolling(10).mean())
# === Volatility Analysis ===
dataframe['volatility_expansion'] = (dataframe['atr'] > dataframe['atr'].rolling(20).mean() * 1.2)
dataframe['volatility_contraction'] = (dataframe['atr'] < dataframe['atr'].rolling(20).mean() * 0.8)
# === Support/Resistance Levels ===
dataframe['resistance_5'] = dataframe['high'].rolling(5).max()
dataframe['support_5'] = dataframe['low'].rolling(5).min()
dataframe['resistance_10'] = dataframe['high'].rolling(10).max()
dataframe['support_10'] = dataframe['low'].rolling(10).min()
dataframe['near_resistance'] = (dataframe['close'] >= dataframe['resistance_10'] * 0.998)
dataframe['near_support'] = (dataframe['close'] <= dataframe['support_10'] * 1.002)
# === Candle Patterns ===
dataframe['doji'] = (abs(dataframe['open'] - dataframe['close']) <=
(dataframe['high'] - dataframe['low']) * 0.1)
dataframe['hammer'] = (
(dataframe['close'] > dataframe['open']) &
((dataframe['close'] - dataframe['open']) /
(0.001 + dataframe['high'] - dataframe['low']) > 0.6) &
((dataframe['open'] - dataframe['low']) /
(0.001 + dataframe['high'] - dataframe['low']) > 0.6)
)
dataframe['shooting_star'] = (
(dataframe['open'] > dataframe['close']) &
((dataframe['open'] - dataframe['close']) /
(0.001 + dataframe['high'] - dataframe['low']) > 0.6) &
((dataframe['high'] - dataframe['open']) /
(0.001 + dataframe['high'] - dataframe['low']) > 0.6)
)
# === Market Microstructure ===
dataframe['bid_ask_spread'] = (dataframe['high'] - dataframe['low']) / dataframe['close']
dataframe['price_efficiency'] = abs(dataframe['close'] - dataframe['vwap']) / dataframe['atr']
# === Momentum Divergence ===
dataframe['price_higher_high'] = (dataframe['high'] > dataframe['high'].shift(1))
dataframe['rsi_lower_high'] = (dataframe['rsi'] < dataframe['rsi'].shift(1))
dataframe['bearish_divergence'] = (dataframe['price_higher_high'] & dataframe['rsi_lower_high'])
dataframe['price_lower_low'] = (dataframe['low'] < dataframe['low'].shift(1))
dataframe['rsi_higher_low'] = (dataframe['rsi'] > dataframe['rsi'].shift(1))
dataframe['bullish_divergence'] = (dataframe['price_lower_low'] & dataframe['rsi_higher_low'])
return dataframe
def populate_higher_timeframe_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""Enhanced higher timeframe analysis"""
# === 5-minute context ===
inf_5m = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=self.inf_5m)
inf_5m['ema_trend_5m'] = ta.EMA(inf_5m, timeperiod=20)
inf_5m['rsi_5m'] = ta.RSI(inf_5m, timeperiod=14)
inf_5m['adx_5m'] = ta.ADX(inf_5m, timeperiod=14)
inf_5m['trend_5m'] = (inf_5m['close'] > inf_5m['ema_trend_5m']).astype(int)
inf_5m['momentum_5m'] = inf_5m['close'].pct_change(5)
dataframe = merge_informative_pair(dataframe, inf_5m, self.timeframe, self.inf_5m, ffill=True)
# === 15-minute context ===
inf_15m = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=self.inf_15m)
inf_15m['ema_trend_15m'] = ta.EMA(inf_15m, timeperiod=20)
inf_15m['rsi_15m'] = ta.RSI(inf_15m, timeperiod=14)
inf_15m['trend_15m'] = (inf_15m['close'] > inf_15m['ema_trend_15m']).astype(int)
inf_15m['volatility_15m'] = ta.ATR(inf_15m, timeperiod=14) / inf_15m['close']
dataframe = merge_informative_pair(dataframe, inf_15m, self.timeframe, self.inf_15m, ffill=True)
# === 1-hour context ===
inf_1h = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=self.inf_1h)
inf_1h['ema_trend_1h'] = ta.EMA(inf_1h, timeperiod=20)
inf_1h['trend_1h'] = (inf_1h['close'] > inf_1h['ema_trend_1h']).astype(int)
inf_1h['strength_1h'] = ta.ADX(inf_1h, timeperiod=14)
dataframe = merge_informative_pair(dataframe, inf_1h, self.timeframe, self.inf_1h, ffill=True)
return dataframe
def calculate_signal_scores(self, dataframe: DataFrame) -> DataFrame:
"""Calculate weighted signal scores for enhanced decision making"""
# === Momentum Score ===
momentum_signals = [
dataframe['rsi_slope'] > 0,
dataframe['macd'] > dataframe['macdsignal'],
dataframe['macd_momentum'] > 0,
dataframe['stoch_crossover'],
dataframe['cci'] > self.cci_oversold.value,
dataframe['williams_r'] > self.williams_oversold.value
]
dataframe['momentum_score'] = sum(momentum_signals) / len(momentum_signals)
# === Volume Score ===
volume_signals = [
dataframe['high_volume'],
dataframe['volume_momentum'] > 1.1,
dataframe['volume_change'] > 0,
dataframe['mfi'] > 50,
dataframe['ad'] > dataframe['ad'].shift(1)
]
dataframe['volume_score'] = sum(volume_signals) / len(volume_signals)
# === Add missing trend and RSI columns ===
dataframe['trend_5m'] = np.where(dataframe['ema_fast'] > dataframe['ema_medium'], 1, 0)
dataframe['trend_15m'] = np.where(dataframe['ema_fast'] > dataframe['ema_medium'], 1, 0)
dataframe['trend_1h'] = np.where(dataframe['ema_fast'] > dataframe['ema_medium'], 1, 0)
dataframe['rsi_5m'] = dataframe['rsi'] # Use main timeframe RSI as proxy
dataframe['rsi_15m'] = dataframe['rsi'] # Use main timeframe RSI as proxy
dataframe['rsi_1h'] = dataframe['rsi'] # Use main timeframe RSI as proxy
# === Trend Score ===
trend_signals = [
dataframe['emas_aligned_bullish'],
dataframe['close'] > dataframe['vwap'],
dataframe['trend_5m'] == 1,
dataframe['trend_15m'] == 1,
dataframe['trend_1h'] == 1,
dataframe['di_plus'] > dataframe['di_minus']
]
dataframe['trend_score'] = sum(trend_signals) / len(trend_signals)
# === Volatility Score ===
volatility_signals = [
dataframe['adx'] > self.adx_threshold.value,
~dataframe['bb_squeeze'],
dataframe['volatility_expansion'],
dataframe['atr_percent'] > 1.0,
dataframe['atr_percent'] < 5.0 # Not too volatile
]
dataframe['volatility_score'] = sum(volatility_signals) / len(volatility_signals)
# === Combined Signal Score ===
dataframe['signal_score'] = (
dataframe['momentum_score'] * self.momentum_weight.value +
dataframe['volume_score'] * self.volume_weight.value +
dataframe['trend_score'] * self.trend_weight.value +
dataframe['volatility_score'] * self.volatility_weight.value
)
return dataframe
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""Enhanced entry logic with signal scoring and AI integration"""
# === LONG CONDITIONS ===
# Core momentum conditions
momentum_bullish = (
(dataframe['rsi'] > self.rsi_entry_long.value) &
(dataframe['rsi'] < 70) &
(dataframe['rsi_slope'] > 0) &
(dataframe['macd'] > dataframe['macdsignal']) &
(dataframe['macd_momentum'] > 0)
)
# Enhanced trend alignment
trend_bullish = (
(dataframe['emas_aligned_bullish']) &
(dataframe['close'] > dataframe['ema_fast']) &
(dataframe['trend_5m'] == 1) &
(dataframe['trend_15m'] == 1)
)
# Advanced price action
price_action_bullish = (
(dataframe['close'] > dataframe['vwap']) &
(dataframe['bb_percent'] > 0.3) &
(dataframe['bb_percent'] < 0.8) &
(~dataframe['near_resistance']) &
(dataframe['vwap_distance'] > -0.005)
)
# Enhanced volume confirmation
volume_bullish = (
(dataframe['high_volume']) &
(dataframe['volume_momentum'] > 1.1) &
(dataframe['mfi'] > 40) &
(dataframe['ad'] > dataframe['ad'].shift(1))
)
# Advanced volatility and strength
volatility_strength = (
(dataframe['adx'] > self.adx_threshold.value) &
(dataframe['di_plus'] > dataframe['di_minus']) &
(~dataframe['bb_squeeze']) &
(dataframe['volatility_regime'] == 'normal')
)
# Pattern recognition
bullish_patterns = (
(dataframe['hammer']) |
(dataframe['bullish_divergence']) |
(dataframe['stoch_crossover'] & (dataframe['stoch_k'] < 30))
)
# Signal quality filter
signal_quality = (
(dataframe['signal_score'] > 0.6) &
(dataframe['momentum_score'] > 0.5) &
(dataframe['trend_score'] > 0.5)
)
# Market microstructure
microstructure_ok = (
(dataframe['price_efficiency'] < 2.0) &
(dataframe['bid_ask_spread'] < 0.01)
)
# FreqAI integration (if enabled)
freqai_condition = True
if self.use_freqai_signals.value:
# Placeholder for FreqAI signals
# This would be populated by FreqAI predictions
freqai_condition = (
dataframe.get('&-prediction', 0) > 0.5 # Example FreqAI signal
)
dataframe.loc[
(momentum_bullish) &
(trend_bullish) &
(price_action_bullish) &
(volume_bullish) &
(volatility_strength) &
(bullish_patterns | signal_quality) &
(microstructure_ok) &
(freqai_condition),
'enter_long'] = 1
# === SHORT CONDITIONS ===
# Core momentum conditions
momentum_bearish = (
(dataframe['rsi'] < self.rsi_entry_short.value) &
(dataframe['rsi'] > 30) &
(dataframe['rsi_slope'] < 0) &
(dataframe['macd'] < dataframe['macdsignal']) &
(dataframe['macd_momentum'] < 0)
)
# Enhanced trend alignment
trend_bearish = (
(~dataframe['emas_aligned_bullish']) &
(dataframe['close'] < dataframe['ema_fast']) &
(dataframe['trend_5m'] == 0) &
(dataframe['trend_15m'] == 0)
)
# Advanced price action
price_action_bearish = (
(dataframe['close'] < dataframe['vwap']) &
(dataframe['bb_percent'] > 0.2) &
(dataframe['bb_percent'] < 0.7) &
(~dataframe['near_support']) &
(dataframe['vwap_distance'] < 0.005)
)
# Enhanced volume confirmation
volume_bearish = (
(dataframe['high_volume']) &
(dataframe['volume_momentum'] > 1.1) &
(dataframe['mfi'] < 60) &
(dataframe['ad'] < dataframe['ad'].shift(1))
)
# Advanced volatility and strength
volatility_strength_short = (
(dataframe['adx'] > self.adx_threshold.value) &
(dataframe['di_minus'] > dataframe['di_plus']) &
(~dataframe['bb_squeeze']) &
(dataframe['volatility_regime'] == 'normal')
)
# Pattern recognition
bearish_patterns = (
(dataframe['shooting_star']) |
(dataframe['bearish_divergence']) |
(dataframe['stoch_crossunder'] & (dataframe['stoch_k'] > 70))
)
# Signal quality filter
signal_quality_short = (
(dataframe['signal_score'] < 0.4) & # Inverted for short
(dataframe['momentum_score'] < 0.5) &
(dataframe['trend_score'] < 0.5)
)
# FreqAI integration for shorts
freqai_condition_short = True
if self.use_freqai_signals.value:
freqai_condition_short = (
dataframe.get('&-prediction', 0) < -0.5 # Example FreqAI short signal
)
dataframe.loc[
(momentum_bearish) &
(trend_bearish) &
(price_action_bearish) &
(volume_bearish) &
(volatility_strength_short) &
(bearish_patterns | signal_quality_short) &
(microstructure_ok) &
(freqai_condition_short),
'enter_short'] = 1
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""Enhanced exit logic with multiple exit conditions"""
# === EXIT LONG ===
# Momentum exhaustion
momentum_exit_long = (
(dataframe['rsi'] > self.rsi_exit_long.value) |
(dataframe['stoch_k'] > self.stoch_overbought.value) |
(dataframe['cci'] > self.cci_overbought.value) |
(dataframe['williams_r'] > self.williams_overbought.value)
)
# Trend reversal signals
trend_reversal_long = (
(dataframe['macd'] < dataframe['macdsignal']) |
(dataframe['close'] < dataframe['ema_fast']) |
(~dataframe['emas_aligned_bullish']) |
(dataframe['bearish_divergence'])
)
# Volume and structure
volume_structure_exit_long = (
(dataframe['near_resistance']) |
(dataframe['volume_ratio'] < 0.8) |
(dataframe['bb_percent'] > 0.95) |
(dataframe['price_efficiency'] > 3.0)
)
# Higher timeframe weakness
htf_weakness_long = (
(dataframe['trend_5m'] == 0) |
(dataframe['rsi_5m'] > 75)
)
# Signal score deterioration
signal_deterioration_long = (
(dataframe['signal_score'] < 0.3) |
(dataframe['momentum_score'] < 0.3)
)
dataframe.loc[
(momentum_exit_long) |
(trend_reversal_long) |
(volume_structure_exit_long) |
(htf_weakness_long) |
(signal_deterioration_long),
'exit_long'] = 1
# === EXIT SHORT ===
# Momentum exhaustion
momentum_exit_short = (
(dataframe['rsi'] < self.rsi_exit_short.value) |
(dataframe['stoch_k'] < self.stoch_oversold.value) |
(dataframe['cci'] < self.cci_oversold.value) |
(dataframe['williams_r'] < self.williams_oversold.value)
)
# Trend reversal signals
trend_reversal_short = (
(dataframe['macd'] > dataframe['macdsignal']) |
(dataframe['close'] > dataframe['ema_fast']) |
(dataframe['emas_aligned_bullish']) |
(dataframe['bullish_divergence'])
)
# Volume and structure
volume_structure_exit_short = (
(dataframe['near_support']) |
(dataframe['volume_ratio'] < 0.8) |
(dataframe['bb_percent'] < 0.05) |
(dataframe['price_efficiency'] > 3.0)
)
# Higher timeframe strength
htf_strength_short = (
(dataframe['trend_5m'] == 1) |
(dataframe['rsi_5m'] < 25)
)
# Signal score improvement
signal_improvement_short = (
(dataframe['signal_score'] > 0.7) |
(dataframe['momentum_score'] > 0.7)
)
dataframe.loc[
(momentum_exit_short) |
(trend_reversal_short) |
(volume_structure_exit_short) |
(htf_strength_short) |
(signal_improvement_short),
'exit_short'] = 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:
"""Dynamic leverage based on market conditions"""
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
last_candle = dataframe.iloc[-1].squeeze()
# Base leverage
base_leverage = 15.0
# Adjust based on volatility
if last_candle['volatility_regime'] == 'high':
base_leverage *= 0.7 # Reduce leverage in high volatility
# Adjust based on signal quality
if last_candle['signal_score'] > 0.8:
base_leverage *= 1.2 # Increase leverage for high-quality signals
elif last_candle['signal_score'] < 0.5:
base_leverage *= 0.8 # Reduce leverage for weak signals
return min(base_leverage, max_leverage, 20.0)
def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime,
current_rate: float, current_profit: float, **kwargs) -> float:
"""Enhanced dynamic stoploss with ATR and volatility adjustment"""
if not self.use_dynamic_stoploss.value:
return self.stoploss
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
last_candle = dataframe.iloc[-1].squeeze()
# ATR-based dynamic stoploss
atr_stop = self.atr_stoploss_multiplier.value * last_candle['atr'] / current_rate
# Volatility adjustment
if last_candle['volatility_regime'] == 'high':
atr_stop *= 1.5 # Wider stops in high volatility
# Trailing stop logic
if self.trailing_stop.value and current_profit > self.trailing_stop_positive.value:
trailing_stop = self.trailing_stop_positive_offset.value - current_profit
return max(trailing_stop, -atr_stop, self.stoploss)
return max(-atr_stop, self.stoploss)
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:
"""Enhanced entry confirmation with multiple safety checks"""
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
last_candle = dataframe.iloc[-1].squeeze()
# Volume check
if last_candle['volume_ratio'] < 1.2:
logger.info(f"Entry rejected for {pair}: Low volume")
return False
# Volatility check
if last_candle['bb_squeeze']:
logger.info(f"Entry rejected for {pair}: Bollinger Band squeeze")
return False
# Trend strength check
if last_candle['adx'] < 20:
logger.info(f"Entry rejected for {pair}: Weak trend strength")
return False
# Signal quality check
if last_candle['signal_score'] < 0.4:
logger.info(f"Entry rejected for {pair}: Low signal quality")
return False
# Market microstructure check
if last_candle['price_efficiency'] > 3.0:
logger.info(f"Entry rejected for {pair}: Poor price efficiency")
return False
# Volatility regime check
if last_candle['volatility_regime'] == 'high' and last_candle['atr_percent'] > 4.0:
logger.info(f"Entry rejected for {pair}: Excessive volatility")
return False
return True
def custom_exit(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float,
current_profit: float, **kwargs) -> Optional[Union[str, bool]]:
"""Enhanced custom exit logic"""
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
last_candle = dataframe.iloc[-1].squeeze()
# Quick scalp profit
if current_profit > 0.02: # 2% profit
return "scalp_profit_2pct"
# Volume drying up with some profit
if last_candle['volume_ratio'] < 0.6 and current_profit > 0.005:
return "volume_exit"
# Signal deterioration
if last_candle['signal_score'] < 0.2 and current_profit > 0.003:
return "signal_deterioration"
# Momentum reversal with profit
if trade.is_open:
if (trade.is_short and last_candle['macd'] > last_candle['macdsignal'] and
last_candle['rsi_slope'] > 0 and current_profit > 0.005):
return "momentum_reversal_short"
elif (not trade.is_short and last_candle['macd'] < last_candle['macdsignal'] and
last_candle['rsi_slope'] < 0 and current_profit > 0.005):
return "momentum_reversal_long"
# Volatility spike protection
if last_candle['volatility_regime'] == 'high' and current_profit < -0.005:
return "volatility_protection"
# Pattern-based exits
if current_profit > 0.008:
if (not trade.is_short and last_candle['shooting_star']) or \
(trade.is_short and last_candle['hammer']):
return "pattern_exit"
return None
def bot_loop_start(self, **kwargs) -> None:
"""Enhanced bot loop start with performance tracking"""
# Performance tracking could be added here
pass
def check_entry_timeout(self, pair: str, trade: 'Trade', order: 'Order',
current_time: datetime, **kwargs) -> bool:
"""Enhanced entry timeout for scalping"""
return False
def check_exit_timeout(self, pair: str, trade: 'Trade', order: 'Order',
current_time: datetime, **kwargs) -> bool:
"""Enhanced exit timeout for scalping"""
return False