# source: https://raw.githubusercontent.com/hankniel/freqtrade/83ba4265808f598d5e1f810daa10ebea1726a645/user_data/strategies/technical/AHFTLong.py """ Anomalous Holonomy Field Theory Strategy Source: https://www.tradingview.com/script/YRfaiBIW-Anomalous-Holonomy-Field-Theory/ """ # --- Do not remove these libs --- from freqtrade.strategy import IStrategy, stoploss_from_absolute, Trade from pandas import DataFrame from datetime import datetime import numpy as np import pandas as pd # -------------------------------- from numba import njit import talib.abstract as ta @njit(cache=True, parallel=True) def calculate_robust_signal( close, ema20, ema50, ema60, rsi, macd, macd_signal, atr, volume, vol_avg, vwap, roc, score_amplifier ): """Calculate holonomy field signal strength - exact Pine Script translation""" n = len(close) signal = np.zeros(n) for i in range(10, n): # Skip first 10 for ROC calculation if np.isnan(close[i]) or np.isnan(ema20[i]) or np.isnan(ema50[i]) or np.isnan(ema60[i]): continue # HTF trend (using EMA60 as higher timeframe proxy) htf_trend = ema60[i] # Rate of change component - exact Pine Script: (close - close[10]) / close[10] * 100 if i >= 10: roc_val = (close[i] - close[i-10]) / close[i-10] * 100 else: roc_val = 0.0 # VWAP distance component vwap_distance = ((close[i] - vwap[i]) / close[i] * 100) if not np.isnan(vwap[i]) else 0 vwap_contrib = max(-2, min(2, vwap_distance)) # Base signal calculation - exact Pine Script logic curr_signal = 0.0 # Bullish conditions if close[i] > htf_trend and close[i] > ema20[i]: curr_signal += 1.5 if close[i] > ema50[i] else 1.0 if roc_val > 0: curr_signal += roc_val / 10 if volume[i] > vol_avg[i] * 1.2: curr_signal += 0.5 # Bearish conditions elif close[i] < htf_trend and close[i] < ema20[i]: curr_signal -= 1.5 if close[i] < ema50[i] else 1.0 if roc_val < 0: curr_signal += roc_val / 10 # roc is negative, so this reduces signal if volume[i] > vol_avg[i] * 1.2: curr_signal -= 0.5 # Add VWAP contribution curr_signal += vwap_contrib # RSI enhancements - exact Pine Script conditions if not np.isnan(rsi[i]): if rsi[i] < 30 and curr_signal > 0: curr_signal += 1.5 elif rsi[i] < 40 and curr_signal > 0: curr_signal += 0.5 elif rsi[i] > 70 and curr_signal < 0: curr_signal -= 1.5 elif rsi[i] > 60 and curr_signal < 0: curr_signal -= 0.5 # MACD confirmation if not np.isnan(macd[i]) and not np.isnan(macd_signal[i]): if macd[i] > macd_signal[i] and curr_signal > 0: curr_signal += 0.5 elif macd[i] < macd_signal[i] and curr_signal < 0: curr_signal -= 0.5 # Scale by amplifier - exact Pine Script formula signal[i] = curr_signal * score_amplifier / 10 return signal @njit(cache=True, parallel=True) def calculate_position_size(account_size, base_risk, adaptive_risk_multiplier, signal_strength, signal_threshold, dynamic_sizing, market_regime_bullish, win_rate, execution_mode, atr_current, atr_avg, close_price): """Calculate elite position size - exact Pine Script translation""" base_risk_amount = account_size * (base_risk / 100) * adaptive_risk_multiplier # Signal strength multiplier if dynamic_sizing: raw_signal_mult = 0.5 + (abs(signal_strength) / signal_threshold) * 1.5 signal_mult = max(0.5, min(3.0, raw_signal_mult)) else: signal_mult = 1.0 # Regime multiplier if (market_regime_bullish and signal_strength > 0) or (not market_regime_bullish and signal_strength < 0): regime_mult = 1.2 else: regime_mult = 0.8 # Performance multiplier if win_rate > 0.65: perf_mult = 1.5 elif win_rate > 0.55: perf_mult = 1.2 elif win_rate < 0.45: perf_mult = 0.5 else: perf_mult = 1.0 # Mode multiplier if execution_mode == 2: # Aggressive mode_mult = 1.8 elif execution_mode == 0: # Conservative mode_mult = 0.6 else: # Adaptive mode_mult = 1.0 # Volatility adjustment vol_mult = atr_current / atr_avg if atr_avg > 0 else 1.0 if vol_mult > 1.2: vol_adj = 0.8 elif vol_mult < 0.8: vol_adj = 1.3 else: vol_adj = 1.0 # Final risk calculation final_risk = base_risk_amount * signal_mult * regime_mult * perf_mult * mode_mult * vol_adj # Convert to position size (simplified for crypto/stocks) stop_percent = (atr_current * 1.5) / close_price * 100 if stop_percent > 0: position_val = final_risk / (stop_percent / 100) shares = position_val / close_price return max(0.001, shares) else: return 0.001 @njit(cache=True) def calculate_smart_stop(high, low, close, atr, execution_mode, is_long, lookback=10): """Calculate smart stop loss - exact Pine Script translation""" # Recent range calculation recent_high = 0.0 recent_low = 999999.0 start_idx = max(0, len(high) - lookback) for i in range(start_idx, len(high)): if high[i] > recent_high: recent_high = high[i] if low[i] < recent_low: recent_low = low[i] recent_range = recent_high - recent_low # Pivot point (simplified) if len(close) > 1: pivot_point = (high[-2] + low[-2] + close[-2]) / 3 else: pivot_point = close[-1] # Mode multiplier if execution_mode == 2: # Aggressive stop_mode_mult = 0.8 elif execution_mode == 0: # Conservative stop_mode_mult = 1.5 else: # Adaptive stop_mode_mult = 1.0 current_close = close[-1] current_atr = atr[-1] if is_long: # Support level calculation support_level = min(recent_low, pivot_point - recent_range * 0.382) atr_stop = current_close - current_atr * 1.5 * stop_mode_mult return max(atr_stop, support_level) else: # Resistance level calculation resistance_level = max(recent_high, pivot_point + recent_range * 0.382) atr_stop = current_close + current_atr * 2 * stop_mode_mult return min(atr_stop, resistance_level) class Github_hankniel_freqtrade__AHFTLong__20251014_102948(IStrategy): INTERFACE_VERSION: int = 3 # Minimal ROI designed for the strategy. # This attribute will be overridden if the config file contains "minimal_roi" # minimal_roi = { # "60": 0.01, # "30": 0.03, # "20": 0.04, # "0": 0.05 # } use_custom_roi = True use_custom_stoploss = True # This attribute will be overridden if the config file contains "stoploss" stoploss = -0.99 # Optimal timeframe for the strategy timeframe = '15m' # trailing stoploss trailing_stop = False # run "populate_indicators" only for new candle process_only_new_candles = True # Experimental settings (configuration will overide these if set) use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = False # Short can_short = True # Optional order type mapping order_types = { 'entry': 'limit', 'exit': 'limit', 'stoploss': 'market', 'stoploss_on_exchange': True } # ======================================== # HYPEROPTABLE PARAMETERS # ======================================== # Signal Configuration signal_sensitivity: float = 2.5 score_amplifier: float = 50.0 signal_confirmation: bool = True min_signal_separation: int = 5 # Execution Settings execution_mode: int = 1 # 0=Conservative, 1=Adaptive, 2=Aggressive dynamic_sizing: bool = True smart_entries: bool = True advanced_exits: bool = True # Advanced Configuration anomaly_threshold: float = 30.0 portal_threshold: float = 0.7 # Risk Management account_size: float = 100000.0 base_risk: float = 0.8 # Percentage max_daily_trades: int = 12 # ATR Configuration atr_period: int = 14 atr_multiplier: float = 1.5 # Fibonacci levels fib_range_period: int = 55 # Learning System (simplified tracking) enable_learning: bool = False # Disabled for simplicity learning_speed: float = 0.15 performance_tracking: int = 50 custom_leverage: int = 1 # Required startup_candles startup_candle_count: int = 100 # ======================================== # INTERNAL STATE TRACKING # ======================================== def __init__(self, config: dict): super().__init__(config) # Simple performance tracking self.trade_performance = [] self.adaptive_risk_multiplier = 1.0 self.win_rate = 0.5 self.market_regime_bullish = True self.daily_trades = 0 self.daily_pnl = 0.0 self.last_signal_bar = 0 def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Adds several different TA indicators to the given DataFrame Performance Note: For the best performance be frugal on the number of indicators you are using. Let uncomment only the indicator you are using in your strategies or your hyperopt configuration, otherwise you will waste your memory and CPU usage. """ # Get hyperopt parameters signal_sensitivity = self.signal_sensitivity score_amplifier = self.score_amplifier atr_period = self.atr_period fib_range_period = self.fib_range_period # ======================================== # CORE INDICATORS # ======================================== dataframe['rsi'] = ta.RSI(dataframe['close'], timeperiod=14) dataframe['ema20'] = ta.EMA(dataframe['close'], timeperiod=20) dataframe['ema50'] = ta.EMA(dataframe['close'], timeperiod=50) dataframe['ema60'] = ta.EMA(dataframe['close'], timeperiod=60) # HTF trend proxy dataframe['sma20'] = ta.SMA(dataframe['close'], timeperiod=20) dataframe['atr'] = ta.ATR(dataframe['high'], dataframe['low'], dataframe['close'], timeperiod=atr_period) # MACD dataframe['macd'], dataframe['macd_signal'], dataframe['macd_histogram'] = ta.MACD( dataframe['close'], fastperiod=12, slowperiod=26, signalperiod=9 ) # Volume indicators dataframe['vol_avg'] = ta.SMA(dataframe['volume'], timeperiod=20) # VWAP calculation dataframe['hlc3'] = (dataframe['high'] + dataframe['low'] + dataframe['close']) / 3 dataframe['vwap'] = (dataframe['hlc3'] * dataframe['volume']).cumsum() / dataframe['volume'].cumsum() # Rate of Change dataframe['roc'] = ta.ROC(dataframe['close'], timeperiod=10) # ======================================== # FIBONACCI LEVELS FOR ADDITIONAL RESONANCE # ======================================== dataframe['range_high'] = dataframe['high'].rolling(window=fib_range_period).max() dataframe['range_low'] = dataframe['low'].rolling(window=fib_range_period).min() dataframe['price_range'] = dataframe['range_high'] - dataframe['range_low'] # ======================================== # HOLONOMY FIELD CALCULATION # ======================================== signal_strength = calculate_robust_signal( dataframe['close'].values, dataframe['ema20'].values, dataframe['ema50'].values, dataframe['ema60'].values, dataframe['rsi'].values, dataframe['macd'].values, dataframe['macd_signal'].values, dataframe['atr'].values, dataframe['volume'].values, dataframe['vol_avg'].values, dataframe['vwap'].values, dataframe['roc'].values, score_amplifier ) dataframe['holonomy_signal'] = signal_strength # ======================================== # MARKET REGIME FILTER # ======================================== volatility_regime = dataframe['atr'] / dataframe['close'] * 100 vol_percentile_30 = volatility_regime.rolling(100).quantile(0.30) vol_percentile_70 = volatility_regime.rolling(100).quantile(0.70) low_vol = volatility_regime < vol_percentile_30 high_vol = volatility_regime > vol_percentile_70 # Mode multiplier if self.execution_mode == 0: # Conservative mode_multiplier = 1.3 elif self.execution_mode == 2: # Aggressive mode_multiplier = 0.7 else: # Adaptive mode_multiplier = 1.0 # Dynamic threshold base_threshold = signal_sensitivity dataframe['signal_threshold'] = np.where( low_vol, base_threshold * 0.7 * mode_multiplier, np.where(high_vol, base_threshold * 1.3 * mode_multiplier, base_threshold * mode_multiplier) ) # ======================================== # SIGNAL CALCULATION # ======================================== dataframe['strong_long'] = dataframe['holonomy_signal'] >= (dataframe['signal_threshold'] * 1.2) dataframe['strong_short'] = dataframe['holonomy_signal'] <= (-dataframe['signal_threshold'] * 1.2) # Signal quality rating abs_signal = np.abs(dataframe['holonomy_signal']) dataframe['signal_quality'] = np.where( abs_signal >= dataframe['signal_threshold'] * 1.5, 3, # ELITE np.where(abs_signal >= dataframe['signal_threshold'] * 1.2, 2, # STRONG np.where(abs_signal >= dataframe['signal_threshold'], 1, 0)) # GOOD/WEAK ) # ======================================== # MARKET REGIME DETECTION # ======================================== dataframe['market_regime_bullish'] = ( (dataframe['close'] > dataframe['sma20']) & (dataframe['sma20'] > dataframe['ema50']) ) # ======================================== # POSITION SIZING PREPARATION # ======================================== dataframe['atr_avg'] = ta.SMA(dataframe['atr'], timeperiod=50) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Based on TA indicators, populates the buy signal for the given dataframe :param dataframe: DataFrame :return: DataFrame with buy column """ # Entry timing logic min_bars_between = self.min_signal_separation if self.execution_mode == 2: # Aggressive min_bars_between = max(1, min_bars_between // 2) elif self.execution_mode == 0: # Conservative min_bars_between = min_bars_between * 2 # Base entry conditions base_conditions = ( (dataframe['strong_long'] == 1) & (dataframe['signal_quality'] >= 1) # At least GOOD quality ) # Signal confirmation filters if self.signal_confirmation and self.execution_mode == 0: # Conservative confirmation_conditions = ( (dataframe['close'] > dataframe['ema20']) & (dataframe['rsi'] > 30) & (dataframe['rsi'] < 70) ) base_conditions = base_conditions & confirmation_conditions # Smart entries - prevent over-trading if self.smart_entries: # Simple signal separation (can't replicate exact bar_index logic in FreqTrade) signal_changed = ( (dataframe['strong_long'] == 1) & (~(dataframe['strong_long'].shift(1) == 1)) ) base_conditions = base_conditions & signal_changed dataframe.loc[base_conditions, ['enter_long', 'enter_tag']] = (1, 'AHFT') return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Based on TA indicators, populates the sell signal for the given dataframe :param dataframe: DataFrame :return: DataFrame with buy column """ # Exit long condition - holonomy field reversal conditions = ( (dataframe['strong_short'] == 1) & (dataframe['signal_quality'] >= 1) # At least GOOD quality reversal ) dataframe.loc[conditions, ['exit_long', 'exit_tag']] = (1, 'AHFT') return dataframe def custom_stake_amount(self, pair: str, current_time: datetime, current_rate: float, proposed_stake: float, min_stake: float | None, max_stake: float, leverage: float, entry_tag: str | None, side: str, **kwargs) -> float: """ Custom position sizing based on AHFT calculations """ if not self.dynamic_sizing: return proposed_stake try: dataframe, _ = self.dp.get_analyzed_dataframe(pair=pair, timeframe=self.timeframe) current_candle = dataframe.iloc[-1].squeeze() # Get current values signal_strength = current_candle['holonomy_signal'] signal_threshold = current_candle['signal_threshold'] atr_current = current_candle['atr'] atr_avg = current_candle['atr_avg'] market_regime_bullish = current_candle['market_regime_bullish'] # Calculate position size using AHFT formula position_multiplier = calculate_position_size( self.account_size, self.base_risk, self.adaptive_risk_multiplier, signal_strength, signal_threshold, self.dynamic_sizing, market_regime_bullish, self.win_rate, self.execution_mode, atr_current, atr_avg if not pd.isna(atr_avg) else atr_current, current_rate ) # Calculate final stake base_stake = self.account_size * (self.base_risk / 100) final_stake = base_stake * position_multiplier * self.adaptive_risk_multiplier # Apply bounds if min_stake is not None: final_stake = max(final_stake, min_stake) final_stake = min(final_stake, max_stake) return final_stake except Exception as e: # Fallback to proposed stake return proposed_stake def custom_roi( self, pair: str, trade: Trade, current_time: datetime, trade_duration: int, entry_tag: str | None, side: str, **kwargs ) -> float | None: """ Custom ROI logic based on AHFT execution mode and ATR """ try: # Get last candle dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = dataframe.iloc[-1].squeeze() # ATR based roi with execution mode multipliers atr = last_candle['atr'] last_close = last_candle['close'] # Target multiplier based on execution mode if self.execution_mode == 2: # Aggressive target_multiplier = 4.0 elif self.execution_mode == 0: # Conservative target_multiplier = 2.0 else: # Adaptive target_multiplier = 3.0 # Calculate ROI based on ATR and target multiplier atr_roi = (atr * self.atr_multiplier * target_multiplier) / last_close roi = atr_roi * trade.leverage return roi except Exception as e: # Fallback ROI return None def custom_stoploss(self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, after_fill: bool, **kwargs) -> float | None: """ Custom stoploss logic based on AHFT smart stop calculation """ try: # Get recent data for stop calculation dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) if len(dataframe) < 20: return None # Use last 20 candles for smart stop calculation recent_data = dataframe.tail(20) # Calculate smart stop smart_stop_price = calculate_smart_stop( recent_data['high'].values, recent_data['low'].values, recent_data['close'].values, recent_data['atr'].values, self.execution_mode, not trade.is_short, # is_long lookback=10 ) # Convert to stoploss ratio stoploss_value = stoploss_from_absolute( smart_stop_price, current_rate=current_rate, is_short=trade.is_short, leverage=trade.leverage ) return stoploss_value except Exception as e: # Fallback to ATR-based stop try: dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = dataframe.iloc[-1].squeeze() close = last_candle['close'] atr = last_candle['atr'] if not trade.is_short: stoploss_price = close - (atr * self.atr_multiplier) else: stoploss_price = close + (atr * self.atr_multiplier) stoploss_value = stoploss_from_absolute( stoploss_price, current_rate=current_rate, is_short=trade.is_short, leverage=trade.leverage ) return stoploss_value except: return None def leverage( self, pair: str, current_time: datetime, current_rate: float, proposed_leverage: float, max_leverage: float, entry_tag: str | None, side: str, **kwargs ) -> float: """ Customize leverage per trade """ return self.custom_leverage # def confirm_trade_exit(self, pair: str, trade: Trade, order_type: str, amount: float, # rate: float, time_in_force: str, exit_reason: str, # current_time: datetime, **kwargs) -> bool: # """ # Track trade performance for adaptive risk management # """ # if trade.close_profit is not None: # # Simple performance tracking # self.trade_performance.append(trade.close_profit) # # Keep only recent trades # if len(self.trade_performance) > self.performance_tracking: # self.trade_performance.pop(0) # # Update win rate # if len(self.trade_performance) >= 10: # wins = sum(1 for p in self.trade_performance if p > 0) # self.win_rate = wins / len(self.trade_performance) # # Simple adaptive risk adjustment # if len(self.trade_performance) >= 20: # avg_win = np.mean([p for p in self.trade_performance if p > 0]) # avg_loss = np.mean([abs(p) for p in self.trade_performance if p < 0]) # if avg_loss > 0: # win_loss_ratio = avg_win / avg_loss # if win_loss_ratio < 1.5: # self.adaptive_risk_multiplier = max(0.5, self.adaptive_risk_multiplier - 0.1) # elif win_loss_ratio > 2.5: # self.adaptive_risk_multiplier = min(1.5, self.adaptive_risk_multiplier + 0.05) # # Update market regime # try: # dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) # if not dataframe.empty: # current_candle = dataframe.iloc[-1] # self.market_regime_bullish = current_candle['market_regime_bullish'] # except: # pass # return True