Advanced Price Prediction Strategy with Support/Resistance Detection
Timeframe
15m
Direction
Long Only
Stoploss
-6.0%
Trailing Stop
No
ROI
0m: 15.0%, 10m: 8.0%, 20m: 5.0%, 40m: 3.0%
Interface Version
3
Startup Candles
N/A
Indicators
14
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
from scipy import stats
from scipy.signal import find_peaks, find_peaks_cwt
from sklearn.preprocessing import StandardScaler
from sklearn.cluster import KMeans
import warnings
warnings.filterwarnings('ignore')
logger = logging.getLogger(__name__)
class ClaudePricePrediction(IStrategy):
"""
Advanced Price Prediction Strategy with Support/Resistance Detection
This strategy combines machine learning with sophisticated technical analysis:
- Multi-timeframe price prediction using FreqAI
- Dynamic support and resistance level detection
- Volume profile analysis for key price zones
- Advanced pattern recognition
- Statistical analysis of price movements
- Risk-adjusted position sizing
- Profitable entry/exit timing optimization
Features:
- Machine learning price targets
- Support/resistance breakout detection
- Volume-weighted price levels
- Fibonacci retracement levels
- Market regime detection
- Risk parity position sizing
- Advanced exit strategies
"""
INTERFACE_VERSION = 3
# Optimized ROI for ML predictions
minimal_roi = {
"0": 0.15, # 15% at any time (ML confidence high)
"10": 0.08, # 8% after 10 minutes
"20": 0.05, # 5% after 20 minutes
"40": 0.03, # 3% after 40 minutes
"80": 0.015, # 1.5% after 80 minutes
"120": 0.01 # 1% after 2 hours
}
# Dynamic stoploss based on volatility
stoploss = -0.06 # 6% base stoploss
# Multi-timeframe approach
timeframe = '15m'
# Informative timeframes for comprehensive analysis
inf_5m = '5m'
inf_1h = '1h'
inf_4h = '4h'
inf_1d = '1d'
# Process only new candles
process_only_new_candles = True
# Use ML-enhanced exit signals
use_exit_signal = True
exit_profit_only = False
ignore_roi_if_entry_signal = True
# Extended startup for comprehensive analysis
startup_candle_count: int = 400
# === Strategy Parameters - Hyperopt Optimized ===
# === ML Prediction Parameters ===
prediction_confidence_threshold = DecimalParameter(0.6, 0.9, default=0.75, space='buy', optimize=True)
prediction_horizon = IntParameter(12, 48, default=24, space='buy', optimize=True)
ml_signal_weight = DecimalParameter(0.3, 0.8, default=0.5, space='buy', optimize=True)
# === Support/Resistance Parameters ===
sr_lookback_period = IntParameter(50, 200, default=100, space='buy', optimize=True)
sr_touch_tolerance = DecimalParameter(0.002, 0.01, default=0.005, space='buy', optimize=True)
sr_strength_threshold = IntParameter(3, 8, default=5, space='buy', optimize=True)
sr_break_threshold = DecimalParameter(0.003, 0.015, default=0.008, space='buy', optimize=True)
# === Volume Profile Parameters ===
volume_profile_periods = IntParameter(100, 300, default=200, space='buy', optimize=True)
poc_deviation_threshold = DecimalParameter(0.01, 0.05, default=0.02, space='buy', optimize=True)
# === Fibonacci Parameters ===
fib_lookback = IntParameter(20, 100, default=50, space='buy', optimize=True)
fib_deviation = DecimalParameter(0.005, 0.02, default=0.01, space='buy', optimize=True)
# === Trend Parameters ===
trend_ema_fast = IntParameter(12, 26, default=20, space='buy', optimize=True)
trend_ema_slow = IntParameter(40, 80, default=60, space='buy', optimize=True)
trend_strength_threshold = DecimalParameter(0.02, 0.08, default=0.04, space='buy', optimize=True)
# === Risk Parameters ===
volatility_threshold = DecimalParameter(0.01, 0.05, default=0.03, space='buy', optimize=True)
risk_reward_ratio = DecimalParameter(1.5, 4.0, default=2.5, space='buy', optimize=True)
# === Exit Parameters ===
take_profit_multiplier = DecimalParameter(1.5, 3.5, default=2.0, space='sell', optimize=True)
trailing_stop_activation = DecimalParameter(0.02, 0.08, default=0.04, space='sell', optimize=True)
# Position sizing
max_open_trades = 3
position_adjustment_enable = True
def informative_pairs(self):
"""
Define informative pairs for comprehensive market analysis
"""
pairs = self.dp.current_whitelist()
informative_pairs = []
# Add all timeframes for current pairs
for pair in pairs:
informative_pairs.extend([
(pair, self.inf_5m),
(pair, self.inf_1h),
(pair, self.inf_4h),
(pair, self.inf_1d)
])
# Add major pairs for correlation analysis
major_pairs = ['BTC/USDT:USDT', 'ETH/USDT:USDT']
for pair in major_pairs:
if pair not in pairs:
informative_pairs.extend([
(pair, self.timeframe),
(pair, self.inf_1h),
(pair, self.inf_4h)
])
return informative_pairs
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Populate all indicators including ML features and S/R levels
"""
# === Basic Price Data ===
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
# === Trend Analysis ===
dataframe['ema_fast'] = ta.EMA(dataframe, timeperiod=self.trend_ema_fast.value)
dataframe['ema_slow'] = ta.EMA(dataframe, timeperiod=self.trend_ema_slow.value)
dataframe['ema_trend'] = dataframe['ema_fast'] > dataframe['ema_slow']
dataframe['trend_strength'] = abs((dataframe['ema_fast'] - dataframe['ema_slow']) / dataframe['close'])
# === Volatility ===
dataframe['atr'] = ta.ATR(dataframe, timeperiod=14)
dataframe['volatility'] = dataframe['atr'] / dataframe['close']
# === Volume Analysis ===
dataframe['volume_sma'] = ta.SMA(dataframe['volume'], timeperiod=20)
dataframe['volume_ratio'] = dataframe['volume'] / dataframe['volume_sma']
dataframe['vwap'] = qtpylib.vwap(dataframe)
# === Price Action Patterns ===
dataframe = self.calculate_price_patterns(dataframe)
# === Support/Resistance Levels ===
dataframe = self.calculate_support_resistance(dataframe)
# === Volume Profile ===
dataframe = self.calculate_volume_profile(dataframe)
# === Fibonacci Levels ===
dataframe = self.calculate_fibonacci_levels(dataframe)
# === Market Regime Detection ===
dataframe = self.detect_market_regime(dataframe)
# === Advanced Technical Indicators ===
dataframe = self.calculate_advanced_indicators(dataframe)
# === Higher Timeframe Analysis ===
dataframe = self.populate_higher_timeframe_data(dataframe, metadata)
# === FreqAI Features ===
dataframe = self.populate_freqai_features(dataframe)
return dataframe
def calculate_support_resistance(self, dataframe: DataFrame) -> DataFrame:
"""
Advanced support and resistance level detection
"""
# Calculate pivot points using different methods
dataframe = self.calculate_pivot_points(dataframe)
# Find significant highs and lows
high_peaks = find_peaks(dataframe['high'].values, distance=10, prominence=dataframe['atr'].mean())[0]
low_peaks = find_peaks(-dataframe['low'].values, distance=10, prominence=dataframe['atr'].mean())[0]
# Initialize S/R columns
dataframe['resistance_level'] = np.nan
dataframe['support_level'] = np.nan
dataframe['resistance_strength'] = 0
dataframe['support_strength'] = 0
dataframe['at_resistance'] = False
dataframe['at_support'] = False
dataframe['resistance_break'] = False
dataframe['support_break'] = False
lookback = min(len(dataframe), self.sr_lookback_period.value)
for i in range(lookback, len(dataframe)):
current_price = dataframe.iloc[i]['close']
tolerance = self.sr_touch_tolerance.value
# Get recent peaks for resistance
recent_highs = []
for peak in high_peaks:
if i - lookback <= peak < i:
peak_price = dataframe.iloc[peak]['high']
# Check if this level was tested multiple times
touches = 0
for j in range(max(0, peak - 20), min(len(dataframe), peak + 20)):
if abs(dataframe.iloc[j]['high'] - peak_price) / peak_price <= tolerance:
touches += 1
if touches >= self.sr_strength_threshold.value:
recent_highs.append((peak_price, touches))
# Get recent lows for support
recent_lows = []
for peak in low_peaks:
if i - lookback <= peak < i:
peak_price = dataframe.iloc[peak]['low']
# Check if this level was tested multiple times
touches = 0
for j in range(max(0, peak - 20), min(len(dataframe), peak + 20)):
if abs(dataframe.iloc[j]['low'] - peak_price) / peak_price <= tolerance:
touches += 1
if touches >= self.sr_strength_threshold.value:
recent_lows.append((peak_price, touches))
# Find closest resistance and support
if recent_highs:
resistance_candidates = [(level, strength) for level, strength in recent_highs if level > current_price]
if resistance_candidates:
resistance_level, resistance_strength = min(resistance_candidates, key=lambda x: x[0])
dataframe.iloc[i, dataframe.columns.get_loc('resistance_level')] = resistance_level
dataframe.iloc[i, dataframe.columns.get_loc('resistance_strength')] = resistance_strength
# Check if at resistance
if abs(current_price - resistance_level) / resistance_level <= tolerance:
dataframe.iloc[i, dataframe.columns.get_loc('at_resistance')] = True
# Check for resistance break
if current_price > resistance_level * (1 + self.sr_break_threshold.value):
dataframe.iloc[i, dataframe.columns.get_loc('resistance_break')] = True
if recent_lows:
support_candidates = [(level, strength) for level, strength in recent_lows if level < current_price]
if support_candidates:
support_level, support_strength = max(support_candidates, key=lambda x: x[0])
dataframe.iloc[i, dataframe.columns.get_loc('support_level')] = support_level
dataframe.iloc[i, dataframe.columns.get_loc('support_strength')] = support_strength
# Check if at support
if abs(current_price - support_level) / support_level <= tolerance:
dataframe.iloc[i, dataframe.columns.get_loc('at_support')] = True
# Check for support break
if current_price < support_level * (1 - self.sr_break_threshold.value):
dataframe.iloc[i, dataframe.columns.get_loc('support_break')] = True
return dataframe
def calculate_pivot_points(self, dataframe: DataFrame) -> DataFrame:
"""
Calculate various pivot point levels
"""
# Standard Pivot Points
dataframe['pivot'] = (dataframe['high'].shift(1) + dataframe['low'].shift(1) + dataframe['close'].shift(1)) / 3
dataframe['r1'] = 2 * dataframe['pivot'] - dataframe['low'].shift(1)
dataframe['s1'] = 2 * dataframe['pivot'] - dataframe['high'].shift(1)
dataframe['r2'] = dataframe['pivot'] + (dataframe['high'].shift(1) - dataframe['low'].shift(1))
dataframe['s2'] = dataframe['pivot'] - (dataframe['high'].shift(1) - dataframe['low'].shift(1))
# Fibonacci Pivot Points
dataframe['fib_r1'] = dataframe['pivot'] + 0.382 * (dataframe['high'].shift(1) - dataframe['low'].shift(1))
dataframe['fib_s1'] = dataframe['pivot'] - 0.382 * (dataframe['high'].shift(1) - dataframe['low'].shift(1))
dataframe['fib_r2'] = dataframe['pivot'] + 0.618 * (dataframe['high'].shift(1) - dataframe['low'].shift(1))
dataframe['fib_s2'] = dataframe['pivot'] - 0.618 * (dataframe['high'].shift(1) - dataframe['low'].shift(1))
return dataframe
def calculate_volume_profile(self, dataframe: DataFrame) -> DataFrame:
"""
Calculate Volume Profile and Point of Control (POC)
"""
lookback = min(len(dataframe), self.volume_profile_periods.value)
dataframe['poc_level'] = np.nan
dataframe['vah_level'] = np.nan # Value Area High
dataframe['val_level'] = np.nan # Value Area Low
dataframe['near_poc'] = False
for i in range(lookback, len(dataframe)):
# Get price and volume data for the lookback period
price_data = dataframe.iloc[i-lookback:i]['hlc3'].values
volume_data = dataframe.iloc[i-lookback:i]['volume'].values
# Create price bins
price_min, price_max = price_data.min(), price_data.max()
price_bins = np.linspace(price_min, price_max, 50)
# Calculate volume at each price level
volume_profile = []
for j in range(len(price_bins)-1):
bin_volume = 0
bin_center = (price_bins[j] + price_bins[j+1]) / 2
for k, price in enumerate(price_data):
if price_bins[j] <= price <= price_bins[j+1]:
bin_volume += volume_data[k]
volume_profile.append((bin_center, bin_volume))
if volume_profile:
# Find Point of Control (highest volume level)
poc_level = max(volume_profile, key=lambda x: x[1])[0]
dataframe.iloc[i, dataframe.columns.get_loc('poc_level')] = poc_level
# Calculate Value Area (70% of total volume)
total_volume = sum([x[1] for x in volume_profile])
target_volume = total_volume * 0.7
# Sort by volume and find value area
sorted_profile = sorted(volume_profile, key=lambda x: x[1], reverse=True)
cumulative_volume = 0
value_area_prices = []
for price, volume in sorted_profile:
cumulative_volume += volume
value_area_prices.append(price)
if cumulative_volume >= target_volume:
break
if value_area_prices:
vah = max(value_area_prices)
val = min(value_area_prices)
dataframe.iloc[i, dataframe.columns.get_loc('vah_level')] = vah
dataframe.iloc[i, dataframe.columns.get_loc('val_level')] = val
# Check if current price is near POC
current_price = dataframe.iloc[i]['close']
if abs(current_price - poc_level) / poc_level <= self.poc_deviation_threshold.value:
dataframe.iloc[i, dataframe.columns.get_loc('near_poc')] = True
return dataframe
def calculate_fibonacci_levels(self, dataframe: DataFrame) -> DataFrame:
"""
Calculate dynamic Fibonacci retracement levels
"""
lookback = min(len(dataframe), self.fib_lookback.value)
dataframe['fib_0'] = np.nan
dataframe['fib_236'] = np.nan
dataframe['fib_382'] = np.nan
dataframe['fib_500'] = np.nan
dataframe['fib_618'] = np.nan
dataframe['fib_786'] = np.nan
dataframe['fib_1000'] = np.nan
dataframe['near_fib_level'] = False
for i in range(lookback, len(dataframe)):
# Find swing high and low in lookback period
high_data = dataframe.iloc[i-lookback:i]['high']
low_data = dataframe.iloc[i-lookback:i]['low']
swing_high = high_data.max()
swing_low = low_data.min()
# Calculate Fibonacci levels
diff = swing_high - swing_low
dataframe.iloc[i, dataframe.columns.get_loc('fib_0')] = swing_high
dataframe.iloc[i, dataframe.columns.get_loc('fib_236')] = swing_high - 0.236 * diff
dataframe.iloc[i, dataframe.columns.get_loc('fib_382')] = swing_high - 0.382 * diff
dataframe.iloc[i, dataframe.columns.get_loc('fib_500')] = swing_high - 0.5 * diff
dataframe.iloc[i, dataframe.columns.get_loc('fib_618')] = swing_high - 0.618 * diff
dataframe.iloc[i, dataframe.columns.get_loc('fib_786')] = swing_high - 0.786 * diff
dataframe.iloc[i, dataframe.columns.get_loc('fib_1000')] = swing_low
# Check if price is near any Fibonacci level
current_price = dataframe.iloc[i]['close']
fib_levels = [
dataframe.iloc[i]['fib_236'],
dataframe.iloc[i]['fib_382'],
dataframe.iloc[i]['fib_500'],
dataframe.iloc[i]['fib_618'],
dataframe.iloc[i]['fib_786']
]
for fib_level in fib_levels:
if not pd.isna(fib_level):
if abs(current_price - fib_level) / fib_level <= self.fib_deviation.value:
dataframe.iloc[i, dataframe.columns.get_loc('near_fib_level')] = True
break
return dataframe
def detect_market_regime(self, dataframe: DataFrame) -> DataFrame:
"""
Detect market regime (trending, ranging, volatile)
"""
# Calculate regime indicators
dataframe['regime_adx'] = ta.ADX(dataframe, timeperiod=14)
dataframe['regime_atr_ratio'] = dataframe['atr'] / dataframe['close']
# Price efficiency (how much price moved vs total movement)
dataframe['price_efficiency'] = abs(dataframe['close'] - dataframe['close'].shift(20)) / (
dataframe['atr'].rolling(20).sum()
)
# Market regime classification
dataframe['market_regime'] = 'ranging' # Default
for i in range(20, len(dataframe)):
adx = dataframe.iloc[i]['regime_adx']
efficiency = dataframe.iloc[i]['price_efficiency']
volatility = dataframe.iloc[i]['regime_atr_ratio']
if adx > 25 and efficiency > 0.3:
dataframe.iloc[i, dataframe.columns.get_loc('market_regime')] = 'trending'
elif volatility > self.volatility_threshold.value:
dataframe.iloc[i, dataframe.columns.get_loc('market_regime')] = 'volatile'
else:
dataframe.iloc[i, dataframe.columns.get_loc('market_regime')] = 'ranging'
return dataframe
def calculate_price_patterns(self, dataframe: DataFrame) -> DataFrame:
"""
Detect important price patterns
"""
# Higher highs and higher lows (uptrend)
dataframe['higher_high'] = (
(dataframe['high'] > dataframe['high'].shift(1)) &
(dataframe['high'].shift(1) > dataframe['high'].shift(2))
)
dataframe['higher_low'] = (
(dataframe['low'] > dataframe['low'].shift(1)) &
(dataframe['low'].shift(1) > dataframe['low'].shift(2))
)
# Lower highs and lower lows (downtrend)
dataframe['lower_high'] = (
(dataframe['high'] < dataframe['high'].shift(1)) &
(dataframe['high'].shift(1) < dataframe['high'].shift(2))
)
dataframe['lower_low'] = (
(dataframe['low'] < dataframe['low'].shift(1)) &
(dataframe['low'].shift(1) < dataframe['low'].shift(2))
)
# Engulfing patterns
dataframe['bullish_engulfing'] = (
(dataframe['close'].shift(1) < dataframe['open'].shift(1)) & # Previous red
(dataframe['close'] > dataframe['open']) & # Current green
(dataframe['open'] < dataframe['close'].shift(1)) & # Opens below previous close
(dataframe['close'] > dataframe['open'].shift(1)) # Closes above previous open
)
dataframe['bearish_engulfing'] = (
(dataframe['close'].shift(1) > dataframe['open'].shift(1)) & # Previous green
(dataframe['close'] < dataframe['open']) & # Current red
(dataframe['open'] > dataframe['close'].shift(1)) & # Opens above previous close
(dataframe['close'] < dataframe['open'].shift(1)) # Closes below previous open
)
return dataframe
def calculate_advanced_indicators(self, dataframe: DataFrame) -> DataFrame:
"""
Calculate advanced technical indicators
"""
# RSI with multiple periods
dataframe['rsi_14'] = ta.RSI(dataframe, timeperiod=14)
dataframe['rsi_21'] = ta.RSI(dataframe, timeperiod=21)
# MACD
macd = ta.MACD(dataframe)
dataframe['macd'] = macd['macd']
dataframe['macdsignal'] = macd['macdsignal']
dataframe['macdhist'] = macd['macdhist']
# Stochastic
stoch = ta.STOCH(dataframe)
dataframe['stoch_k'] = stoch['slowk']
dataframe['stoch_d'] = stoch['slowd']
# Williams %R
dataframe['willr'] = ta.WILLR(dataframe, timeperiod=14)
# Commodity Channel Index
dataframe['cci'] = ta.CCI(dataframe, timeperiod=20)
# Money Flow Index
dataframe['mfi'] = ta.MFI(dataframe, timeperiod=14)
# Bollinger Bands
bollinger = qtpylib.bollinger_bands(dataframe['close'], window=20, stds=2)
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'])
return dataframe
def populate_higher_timeframe_data(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
Add higher timeframe analysis
"""
# 1H timeframe
inf_1h = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=self.inf_1h)
inf_1h['trend_1h'] = ta.EMA(inf_1h, timeperiod=20) < inf_1h['close']
inf_1h['rsi_1h'] = ta.RSI(inf_1h, timeperiod=14)
inf_1h['volume_1h'] = inf_1h['volume']
dataframe = merge_informative_pair(dataframe, inf_1h, self.timeframe, self.inf_1h, ffill=True)
# 4H timeframe
inf_4h = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=self.inf_4h)
inf_4h['trend_4h'] = ta.EMA(inf_4h, timeperiod=20) < inf_4h['close']
inf_4h['rsi_4h'] = ta.RSI(inf_4h, timeperiod=14)
inf_4h['atr_4h'] = ta.ATR(inf_4h, timeperiod=14)
dataframe = merge_informative_pair(dataframe, inf_4h, self.timeframe, self.inf_4h, ffill=True)
# Daily timeframe
inf_1d = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=self.inf_1d)
inf_1d['trend_1d'] = ta.EMA(inf_1d, timeperiod=20) < inf_1d['close']
inf_1d['volatility_1d'] = ta.ATR(inf_1d, timeperiod=14) / inf_1d['close']
dataframe = merge_informative_pair(dataframe, inf_1d, self.timeframe, self.inf_1d, ffill=True)
return dataframe
def populate_freqai_features(self, dataframe: DataFrame) -> DataFrame:
"""
Create features specifically for FreqAI ML model
"""
# === Price-based features ===
for period in [5, 10, 20, 50]:
dataframe[f'price_change_{period}'] = dataframe['close'].pct_change(period)
dataframe[f'high_low_ratio_{period}'] = (
dataframe['high'].rolling(period).max() / dataframe['low'].rolling(period).min() - 1
)
# === Volume-based features ===
for period in [5, 10, 20]:
dataframe[f'volume_sma_{period}'] = dataframe['volume'].rolling(period).mean()
dataframe[f'volume_ratio_{period}'] = dataframe['volume'] / dataframe[f'volume_sma_{period}']
# === Momentum features ===
for period in [7, 14, 21]:
dataframe[f'rsi_{period}'] = ta.RSI(dataframe, timeperiod=period)
dataframe[f'roc_{period}'] = ta.ROC(dataframe, timeperiod=period)
# === Volatility features ===
for period in [10, 20, 30]:
dataframe[f'atr_{period}'] = ta.ATR(dataframe, timeperiod=period)
dataframe[f'volatility_{period}'] = dataframe[f'atr_{period}'] / dataframe['close']
# === Pattern features ===
dataframe['doji'] = (
abs(dataframe['close'] - dataframe['open']) /
(dataframe['high'] - dataframe['low']) < 0.1
).astype(int)
dataframe['hammer'] = (
((dataframe['high'] - dataframe['low']) > 3 * (dataframe['open'] - dataframe['close'])) &
((dataframe['close'] - dataframe['low']) / (0.001 + dataframe['high'] - dataframe['low']) > 0.6) &
((dataframe['open'] - dataframe['low']) / (0.001 + dataframe['high'] - dataframe['low']) > 0.6)
).astype(int)
# === Support/Resistance features ===
dataframe['distance_to_support'] = (
(dataframe['close'] - dataframe['support_level']) / dataframe['close']
).fillna(0)
dataframe['distance_to_resistance'] = (
(dataframe['resistance_level'] - dataframe['close']) / dataframe['close']
).fillna(0)
# === Fibonacci features ===
dataframe['near_fib'] = dataframe['near_fib_level'].astype(int)
# === Market regime features ===
dataframe['regime_trending'] = (dataframe['market_regime'] == 'trending').astype(int)
dataframe['regime_ranging'] = (dataframe['market_regime'] == 'ranging').astype(int)
dataframe['regime_volatile'] = (dataframe['market_regime'] == 'volatile').astype(int)
# === Target variable for ML ===
# Predict future price movement
future_periods = self.prediction_horizon.value
dataframe['future_price'] = dataframe['close'].shift(-future_periods)
dataframe['price_change_future'] = (
(dataframe['future_price'] - dataframe['close']) / dataframe['close']
)
# Classification target (up/down/sideways)
dataframe['future_direction'] = 0 # sideways
dataframe.loc[dataframe['price_change_future'] > 0.02, 'future_direction'] = 1 # up
dataframe.loc[dataframe['price_change_future'] < -0.02, 'future_direction'] = -1 # down
# Regression target (actual price change)
dataframe['target_price_change'] = dataframe['price_change_future']
return dataframe
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
ML-enhanced entry signals with support/resistance confirmation
"""
# === ML Prediction Conditions ===
ml_bullish_prediction = (
(dataframe['do_predict'] == 1) &
(dataframe['DI_values'] > self.prediction_confidence_threshold.value) &
(dataframe['prediction'] > 0.02) # Predicting >2% price increase
)
ml_bearish_prediction = (
(dataframe['do_predict'] == 1) &
(dataframe['DI_values'] > self.prediction_confidence_threshold.value) &
(dataframe['prediction'] < -0.02) # Predicting >2% price decrease
)
# === Technical Confirmation ===
technical_bullish = (
(dataframe['trend_1h'] == 1) &
(dataframe['trend_4h'] == 1) &
(dataframe['ema_trend'] == 1) &
(dataframe['trend_strength'] > self.trend_strength_threshold.value) &
(dataframe['rsi_14'] > 30) & (dataframe['rsi_14'] < 70) &
(dataframe['volume_ratio'] > 1.2)
)
technical_bearish = (
(dataframe['trend_1h'] == 0) &
(dataframe['trend_4h'] == 0) &
(dataframe['ema_trend'] == 0) &
(dataframe['trend_strength'] > self.trend_strength_threshold.value) &
(dataframe['rsi_14'] > 30) & (dataframe['rsi_14'] < 70) &
(dataframe['volume_ratio'] > 1.2)
)
# === Support/Resistance Confirmation ===
sr_bullish = (
(dataframe['at_support']) |
(dataframe['support_break'] == False) |
(dataframe['resistance_break'] == True) |
(dataframe['near_poc'] & (dataframe['close'] > dataframe['vwap']))
)
sr_bearish = (
(dataframe['at_resistance']) |
(dataframe['resistance_break'] == False) |
(dataframe['support_break'] == True) |
(dataframe['near_poc'] & (dataframe['close'] < dataframe['vwap']))
)
# === Pattern Confirmation ===
pattern_bullish = (
(dataframe['bullish_engulfing']) |
(dataframe['hammer'] == 1) |
(dataframe['higher_low'] & dataframe['higher_high'])
)
pattern_bearish = (
(dataframe['bearish_engulfing']) |
(dataframe['lower_high'] & dataframe['lower_low'])
)
# === Risk Management Filters ===
risk_acceptable = (
(dataframe['volatility'] < self.volatility_threshold.value) &
(dataframe['market_regime'] != 'volatile')
)
# === Entry Conditions ===
# Long entries
dataframe.loc[
(ml_bullish_prediction) &
(technical_bullish) &
(sr_bullish) &
(pattern_bullish) &
(risk_acceptable) &
(dataframe['near_fib_level'] | dataframe['at_support']),
'enter_long'] = 1
# Short entries
dataframe.loc[
(ml_bearish_prediction) &
(technical_bearish) &
(sr_bearish) &
(pattern_bearish) &
(risk_acceptable) &
(dataframe['near_fib_level'] | dataframe['at_resistance']),
'enter_short'] = 1
return dataframe
def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
"""
ML-enhanced exit signals
"""
# === ML-based exits ===
ml_exit_long = (
(dataframe['do_predict'] == 1) &
(dataframe['DI_values'] > 0.6) &
(dataframe['prediction'] < -0.01) # Predicting price decline
)
ml_exit_short = (
(dataframe['do_predict'] == 1) &
(dataframe['DI_values'] > 0.6) &
(dataframe['prediction'] > 0.01) # Predicting price increase
)
# === Technical exits ===
technical_exit_long = (
(dataframe['rsi_14'] > 75) |
(dataframe['at_resistance']) |
(dataframe['bearish_engulfing']) |
(dataframe['ema_trend'] == 0) |
(dataframe['macd'] < dataframe['macdsignal'])
)
technical_exit_short = (
(dataframe['rsi_14'] < 25) |
(dataframe['at_support']) |
(dataframe['bullish_engulfing']) |
(dataframe['ema_trend'] == 1) |
(dataframe['macd'] > dataframe['macdsignal'])
)
# === Support/Resistance exits ===
sr_exit_long = (
(dataframe['resistance_break'] == False) |
(dataframe['support_break'] == True)
)
sr_exit_short = (
(dataframe['support_break'] == False) |
(dataframe['resistance_break'] == True)
)
# === Exit Conditions ===
dataframe.loc[
(ml_exit_long) |
(technical_exit_long) |
(sr_exit_long),
'exit_long'] = 1
dataframe.loc[
(ml_exit_short) |
(technical_exit_short) |
(sr_exit_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 ML confidence and volatility
"""
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
last_candle = dataframe.iloc[-1].squeeze()
# Base leverage on ML confidence
base_leverage = 5.0
if 'DI_values' in last_candle and not pd.isna(last_candle['DI_values']):
confidence = last_candle['DI_values']
# Higher confidence = higher leverage
leverage_multiplier = 1 + (confidence - 0.5) * 2
base_leverage *= leverage_multiplier
# Adjust for volatility
if 'volatility' in last_candle and not pd.isna(last_candle['volatility']):
volatility = last_candle['volatility']
if volatility > 0.03: # High volatility
base_leverage *= 0.7
elif volatility < 0.01: # Low volatility
base_leverage *= 1.3
return min(base_leverage, max_leverage, 10.0) # Max 10x leverage
def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime,
current_rate: float, current_profit: float, **kwargs) -> float:
"""
Dynamic stoploss based on support/resistance levels
"""
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
last_candle = dataframe.iloc[-1].squeeze()
# Default stoploss
stop_loss = self.stoploss
# Adjust based on support/resistance
if not trade.is_short: # Long position
if 'support_level' in last_candle and not pd.isna(last_candle['support_level']):
support_distance = (current_rate - last_candle['support_level']) / current_rate
# Set stop just below support
stop_loss = max(self.stoploss, -support_distance * 1.1)
else: # Short position
if 'resistance_level' in last_candle and not pd.isna(last_candle['resistance_level']):
resistance_distance = (last_candle['resistance_level'] - current_rate) / current_rate
# Set stop just above resistance
stop_loss = max(self.stoploss, -resistance_distance * 1.1)
# Trailing stop based on ML predictions
if current_profit > self.trailing_stop_activation.value:
if 'prediction' in last_candle and not pd.isna(last_candle['prediction']):
# If ML predicts continued favorable movement, trail closer
if (not trade.is_short and last_candle['prediction'] > 0) or \
(trade.is_short and last_candle['prediction'] < 0):
trail_distance = self.trailing_stop_activation.value * 0.5
else:
trail_distance = self.trailing_stop_activation.value
stop_loss = max(stop_loss, trail_distance - current_profit)
return 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:
"""
Final confirmation before entering trade
"""
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
last_candle = dataframe.iloc[-1].squeeze()
# Ensure ML model is providing predictions
if 'do_predict' not in last_candle or last_candle['do_predict'] != 1:
return False
# Ensure confidence is above threshold
if 'DI_values' not in last_candle or last_candle['DI_values'] < self.prediction_confidence_threshold.value:
return False
# Check risk/reward ratio
if side == "long":
if 'resistance_level' in last_candle and not pd.isna(last_candle['resistance_level']):
potential_profit = (last_candle['resistance_level'] - rate) / rate
potential_loss = abs(self.stoploss)
if potential_profit / potential_loss < self.risk_reward_ratio.value:
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]]:
"""
Custom exit logic based on ML predictions and S/R levels
"""
dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
last_candle = dataframe.iloc[-1].squeeze()
# ML-based exit
if 'prediction' in last_candle and 'DI_values' in last_candle:
if last_candle['DI_values'] > 0.7: # High confidence prediction
if not trade.is_short and last_candle['prediction'] < -0.015: # Predicting 1.5% decline
return "ml_reversal_prediction"
elif trade.is_short and last_candle['prediction'] > 0.015: # Predicting 1.5% increase
return "ml_reversal_prediction"
# Profit taking at resistance/support
if current_profit > 0.03: # 3% profit
if not trade.is_short and last_candle.get('at_resistance', False):
return "resistance_profit_taking"
elif trade.is_short and last_candle.get('at_support', False):
return "support_profit_taking"
# Risk management exit
if last_candle.get('market_regime') == 'volatile' and current_profit < -0.02:
return "volatile_market_exit"
return None