# source: https://raw.githubusercontent.com/jaredrsommer/freqtradestrategies/9d66f71ce21f2f65feaad6bb967215be22a3cfd6/HurstCycleV6.py import math import numpy as np from freqtrade.strategy import IStrategy import pandas as pd import talib from scipy.fft import fft from technical import qtpylib from freqtrade.strategy import (BooleanParameter, CategoricalParameter, DecimalParameter, IStrategy, IntParameter, RealParameter, merge_informative_pair) class Github_jaredrsommer_freqtradestrategies__HurstCycleV6__20260518_195114(IStrategy): timeframe = '15m' minimal_roi = {"0": 1} stoploss = -0.15 trailing_stop = True trailing_stop_positive = 0.015 can_short = False base_cycle_period = 20 filter_weights = [1, 2, 4, 8, 4] # New parameter for convergence threshold (tunable) convergence_threshold = 0.005 # 0.5% of price as max spread ### Hyperoptable parameters ### u_window_size = IntParameter(250, 350, default=250, space='buy', optimize=True, load=True) l_window_size = IntParameter(20, 50, default=42, space='buy', optimize=True, load=True) def populate_indicators(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame: pair = metadata['pair'] # Heikin-Ashi Candles heikinashi = qtpylib.heikinashi(dataframe) dataframe['ha_open'] = heikinashi['open'] dataframe['ha_close'] = heikinashi['close'] dataframe['ha_high'] = heikinashi['high'] dataframe['ha_low'] = heikinashi['low'] # Initialize Hurst Cycles cycle_period = 80 harmonics = [0, 0, 0] harmonics[0] = 40 harmonics[1] = 27 harmonics[2] = 20 if len(dataframe) < self.u_window_size.value: raise ValueError(f"Insufficient data points for FFT: {len(dataframe)}. Need at least {self.u_window_size.value} data points.") # Perform FFT freq, power = perform_fft(dataframe['ha_close'], window_size=self.u_window_size.value) if len(freq) == 0 or len(power) == 0: raise ValueError("FFT resulted in zero or invalid frequencies.") positive_mask = (1 / freq > self.l_window_size.value) & (1 / freq < self.u_window_size.value) positive_freqs = freq[positive_mask] positive_power = power[positive_mask] if len(positive_power) == 0: raise ValueError("No positive frequencies meet the filtering criteria.") cycle_periods = 1 / positive_freqs power_threshold = 0 if len(positive_power) == 0 else 0.01 * np.max(positive_power) significant_indices = positive_power > power_threshold significant_periods = cycle_periods[significant_indices] significant_power = positive_power[significant_indices] dominant_freq_index = np.argmax(significant_power) dominant_freq = positive_freqs[dominant_freq_index] cycle_period = int(np.abs(1 / dominant_freq)) if dominant_freq != 0 else 100 if cycle_period == np.inf: raise ValueError("No dominant frequency found.") harmonics = [cycle_period / (i + 1) for i in range(1, 4)] self.cp = int(cycle_period) self.h0 = int(harmonics[0]) self.h1 = int(harmonics[1]) self.h2 = int(harmonics[2]) # EWMA for cycles dataframe['cp'] = dataframe['ha_close'].ewm(span=self.cp).mean() dataframe['h0'] = dataframe['ha_close'].ewm(span=self.h0).mean() dataframe['h1'] = dataframe['ha_close'].ewm(span=self.h1).mean() dataframe['h2'] = dataframe['ha_close'].ewm(span=self.h2).mean() # Peak-to-Peak Movement rolling_windowc = dataframe['ha_close'].rolling(self.cp) rolling_windowh0 = dataframe['ha_close'].rolling(self.h0) rolling_windowh1 = dataframe['ha_close'].rolling(self.h1) rolling_windowh2 = dataframe['ha_close'].rolling(self.h2) ptp_valuec = rolling_windowc.apply(lambda x: np.ptp(x)) ptp_valueh0 = rolling_windowh0.apply(lambda x: np.ptp(x)) ptp_valueh1 = rolling_windowh1.apply(lambda x: np.ptp(x)) ptp_valueh2 = rolling_windowh2.apply(lambda x: np.ptp(x)) dataframe['cycle_move'] = ptp_valuec / dataframe['ha_close'] dataframe['h0_move'] = ptp_valueh0 / dataframe['ha_close'] dataframe['h1_move'] = ptp_valueh1 / dataframe['ha_close'] dataframe['h2_move'] = ptp_valueh2 / dataframe['ha_close'] dataframe['cycle_move_mean'] = dataframe['cycle_move'].rolling(self.cp).mean() dataframe['h0_move_mean'] = dataframe['h0_move'].rolling(self.cp).mean() dataframe['h1_move_mean'] = dataframe['h1_move'].rolling(self.cp).mean() dataframe['h2_move_mean'] = dataframe['h2_move'].rolling(self.cp).mean() # Numerical Filter close = dataframe['ha_close'].values weights = np.array(self.filter_weights) / sum(self.filter_weights) filtered = np.convolve(close, weights, mode='valid') dataframe['filtered_close'] = pd.Series(filtered, index=dataframe.index[-len(filtered):]) # FLDs half_short = math.ceil(self.h2 / 2) half_mid = math.ceil(self.h0 / 2) half_long = math.ceil(self.cp / 2) fld_short_base = dataframe['filtered_close'].rolling(window=self.h2, center=True).mean() fld_mid_base = dataframe['filtered_close'].rolling(window=self.h0, center=True).mean() fld_long_base = dataframe['filtered_close'].rolling(window=self.cp, center=True).mean() dataframe['fld_short'] = fld_short_base.shift(half_short) dataframe['fld_mid'] = fld_mid_base.shift(half_mid) dataframe['fld_long'] = fld_long_base.shift(half_long) for fld in ['fld_short', 'fld_mid', 'fld_long']: last_valid_idx = dataframe[fld].last_valid_index() if last_valid_idx is not None: last_valid_row = dataframe.index.get_loc(last_valid_idx) if last_valid_row < len(dataframe) - 1 and last_valid_row > 0: for i in range(last_valid_row + 1, len(dataframe)): prev_slope = (dataframe[fld].iloc[last_valid_row] - dataframe[fld].iloc[last_valid_row - 1]) dataframe[fld].iloc[i] = dataframe[fld].iloc[last_valid_row] + \ prev_slope * (i - last_valid_row) # Convergence-Based Agreeance Band # Calculate spread between FLDs fld_spread = dataframe[['fld_short', 'fld_mid', 'fld_long']].max(axis=1) - \ dataframe[['fld_short', 'fld_mid', 'fld_long']].min(axis=1) # Relative spread as a percentage of current price relative_spread = fld_spread / dataframe['filtered_close'] # Define convergence condition dataframe['converging'] = relative_spread < self.convergence_threshold # Agreeance Band: Mean of FLDs ± a small buffer when converging band_center = dataframe[['fld_short', 'fld_mid', 'fld_long']].mean(axis=1) band_width = dataframe['filtered_close'] * 0.01 # 1% of price as buffer dataframe['agreeance_upper'] = np.where(dataframe['converging'], band_center + band_width, np.nan) dataframe['agreeance_lower'] = np.where(dataframe['converging'], band_center - band_width, np.nan) # Generate VTLs for each harmonic harmonics_dict = {'cp': self.cp, 'h0': self.h0, 'h1': self.h1, 'h2': self.h2} for harmonic_name, harmonic_period in harmonics_dict.items(): self._create_vtl_for_harmonic(dataframe, harmonic_name, harmonic_period) # Calculate slopes for each VTL for harmonic_name in harmonics_dict.keys(): # Calculate slopes for up VTLs dataframe[f'vtl_up_{harmonic_name}_slope'] = dataframe[f'vtl_up_{harmonic_name}'].diff() # Calculate slopes for down VTLs dataframe[f'vtl_down_{harmonic_name}_slope'] = dataframe[f'vtl_down_{harmonic_name}'].diff() # Sum and average the VTL slopes # Sum of all up VTL slopes divided by 4 up_slope_cols = [f'vtl_up_{harmonic}_slope' for harmonic in harmonics_dict.keys()] dataframe['vtl_up_slopes_sum'] = dataframe[up_slope_cols].sum(axis=1) dataframe['vtl_up_slopes_avg'] = dataframe['vtl_up_slopes_sum'] / 4 # Sum of all down VTL slopes divided by 4 down_slope_cols = [f'vtl_down_{harmonic}_slope' for harmonic in harmonics_dict.keys()] dataframe['vtl_down_slopes_sum'] = dataframe[down_slope_cols].sum(axis=1) dataframe['vtl_down_slopes_avg'] = dataframe['vtl_down_slopes_sum'] / 4 # Combined average of all VTL slopes (up and down combined) dataframe['vtl_all_slopes_avg'] = ((dataframe['vtl_up_slopes_sum'] + dataframe['vtl_down_slopes_sum']) / dataframe['filtered_close']) * 100 # Slopes for cycle sync dataframe['fld_short_slope'] = dataframe['fld_short'].diff() dataframe['fld_mid_slope'] = dataframe['fld_mid'].diff() dataframe['fld_long_slope'] = dataframe['fld_long'].diff() # Trend dataframe['trend'] = np.where(dataframe['filtered_close'] > dataframe['cp'], 1, -1) return dataframe def _create_vtl_for_harmonic(self, dataframe: pd.DataFrame, harmonic_name: str, harmonic_period: int): """Create VTL lines for a specific harmonic period""" # Troughs and Crests for this harmonic dataframe[f'trough_{harmonic_name}'] = dataframe['filtered_close'].rolling(harmonic_period).min() dataframe[f'crest_{harmonic_name}'] = dataframe['filtered_close'].rolling(harmonic_period).max() dataframe[f'is_trough_{harmonic_name}'] = np.where( dataframe['filtered_close'] == dataframe[f'trough_{harmonic_name}'], 1, 0) dataframe[f'is_crest_{harmonic_name}'] = np.where( dataframe['filtered_close'] == dataframe[f'crest_{harmonic_name}'], 1, 0) # Initialize VTL columns for this harmonic dataframe[f'vtl_up_{harmonic_name}'] = np.nan dataframe[f'vtl_down_{harmonic_name}'] = np.nan # Group troughs and crests within a window group_window = harmonic_period trough_groups = [] crest_groups = [] current_trough_group = [] current_crest_group = [] last_trough_idx = None last_crest_idx = None # Identify trough and crest groups for idx in dataframe.index: if dataframe.at[idx, f'is_trough_{harmonic_name}'] == 1: if last_trough_idx is None or (dataframe.index.get_loc(idx) - dataframe.index.get_loc(last_trough_idx)) <= group_window: current_trough_group.append(idx) else: if len(current_trough_group) > 0: prices = [dataframe.at[i, 'filtered_close'] for i in current_trough_group] min_idx = current_trough_group[np.argmin(prices)] trough_groups.append(min_idx) current_trough_group = [idx] last_trough_idx = idx if dataframe.at[idx, f'is_crest_{harmonic_name}'] == 1: if last_crest_idx is None or (dataframe.index.get_loc(idx) - dataframe.index.get_loc(last_crest_idx)) <= group_window: current_crest_group.append(idx) else: if len(current_crest_group) > 0: prices = [dataframe.at[i, 'filtered_close'] for i in current_crest_group] max_idx = current_crest_group[np.argmax(prices)] crest_groups.append(max_idx) current_crest_group = [idx] last_crest_idx = idx # Add final groups if len(current_trough_group) > 0: prices = [dataframe.at[i, 'filtered_close'] for i in current_trough_group] min_idx = current_trough_group[np.argmin(prices)] trough_groups.append(min_idx) if len(current_crest_group) > 0: prices = [dataframe.at[i, 'filtered_close'] for i in current_crest_group] max_idx = current_crest_group[np.argmax(prices)] crest_groups.append(max_idx) # Price bounds price_min = dataframe['filtered_close'].min() price_max = dataframe['filtered_close'].max() # Store VTL segments: [start_idx, end_idx, slope, intercept] up_vtl_segments = [] down_vtl_segments = [] # Create VTL segments for trough groups for i in range(1, len(trough_groups)): x1 = dataframe.index.get_loc(trough_groups[i-1]) y1 = dataframe.at[trough_groups[i-1], 'filtered_close'] x2 = dataframe.index.get_loc(trough_groups[i]) y2 = dataframe.at[trough_groups[i], 'filtered_close'] slope = (y2 - y1) / (x2 - x1) if x2 != x1 else 0 intercept = y1 - slope * x1 up_vtl_segments.append([trough_groups[i-1], None, slope, intercept]) # Create VTL segments for crest groups for i in range(1, len(crest_groups)): x1 = dataframe.index.get_loc(crest_groups[i-1]) y1 = dataframe.at[crest_groups[i-1], 'filtered_close'] x2 = dataframe.index.get_loc(crest_groups[i]) y2 = dataframe.at[crest_groups[i], 'filtered_close'] slope = (y2 - y1) / (x2 - x1) if x2 != x1 else 0 intercept = y1 - slope * x1 down_vtl_segments.append([crest_groups[i-1], None, slope, intercept]) # Apply VTL segments across the DataFrame for idx in dataframe.index: current_x = dataframe.index.get_loc(idx) current_price = dataframe.at[idx, 'filtered_close'] # Apply vtl_up for i, segment in enumerate(up_vtl_segments): start_idx, end_idx, slope, intercept = segment if idx >= start_idx and (end_idx is None or idx <= end_idx): vtl_value = slope * current_x + intercept if price_min <= vtl_value <= price_max: dataframe.at[idx, f'vtl_up_{harmonic_name}'] = vtl_value # Check for break (price below VTL) if current_price < vtl_value and i + 1 < len(up_vtl_segments): segment[1] = idx # End this segment else: segment[1] = idx if end_idx is None else end_idx # Apply vtl_down for i, segment in enumerate(down_vtl_segments): start_idx, end_idx, slope, intercept = segment if idx >= start_idx and (end_idx is None or idx <= end_idx): vtl_value = slope * current_x + intercept if price_min <= vtl_value <= price_max: dataframe.at[idx, f'vtl_down_{harmonic_name}'] = vtl_value # Check for break (price above VTL) if current_price > vtl_value and i + 1 < len(down_vtl_segments): segment[1] = idx else: segment[1] = idx if end_idx is None else end_idx def populate_entry_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame: # Entry conditions for each harmonic VTL entry_conditions = [] # CP (dominant cycle) entry condition cp_entry = ( (dataframe['close'] < dataframe['vtl_up_h0']) ) entry_conditions.append(cp_entry) # H0 (longest harmonic) entry condition h0_entry = ( (dataframe['close'] < dataframe['vtl_up_h0']) ) entry_conditions.append(h0_entry) # H1 (medium harmonic) entry condition h1_entry = ( (dataframe['close'] < dataframe['vtl_up_h1']) ) entry_conditions.append(h1_entry) # H2 (shortest harmonic) entry condition h2_entry = ( (dataframe['close'] < dataframe['vtl_up_h2']) ) entry_conditions.append(h2_entry) # Combined entry: any harmonic VTL breach with trend confirmation dataframe.loc[ cp_entry | h0_entry | h1_entry | h2_entry, 'enter_long' ] = 1 # Optional: Individual entry signals for each harmonic dataframe.loc[cp_entry, 'enter_tag'] = 'cp' dataframe.loc[h0_entry, 'enter_tag'] = 'h0' dataframe.loc[h1_entry, 'enter_tag'] = 'h1' dataframe.loc[h2_entry, 'enter_tag'] = 'h2' return dataframe def populate_exit_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame: # Exit conditions for each harmonic VTL exit_conditions = [] # CP (dominant cycle) exit condition cp_exit = (dataframe['close'] > dataframe['vtl_down_cp']) exit_conditions.append(cp_exit) # H0 (longest harmonic) exit condition h0_exit = (dataframe['close'] > dataframe['vtl_down_h0']) exit_conditions.append(h0_exit) # H1 (medium harmonic) exit condition h1_exit = (dataframe['close'] > dataframe['vtl_down_h1']) exit_conditions.append(h1_exit) # H2 (shortest harmonic) exit condition h2_exit = (dataframe['close'] > dataframe['vtl_down_h2']) exit_conditions.append(h2_exit) # Combined exit: any harmonic VTL breach upward dataframe.loc[ cp_exit | h0_exit | h1_exit | h2_exit, 'exit_long' ] = 1 # Optional: Individual exit signals for each harmonic dataframe.loc[cp_exit, 'exit_tag'] = 'cp' dataframe.loc[h0_exit, 'exit_tag'] = 'h0' dataframe.loc[h1_exit, 'exit_tag'] = 'h1' dataframe.loc[h2_exit, 'exit_tag'] = 'h2' return dataframe def leverage(self, pair: str, current_time, current_rate: float, proposed_leverage: float, max_leverage: float, side: str, **kwargs) -> float: return 3.0 def perform_fft(price_data, window_size=None): if window_size is not None: price_data = price_data.rolling(window=window_size, center=True).mean().dropna() normalized_data = (price_data - np.mean(price_data)) / np.std(price_data) n = len(normalized_data) fft_data = np.fft.fft(normalized_data) freq = np.fft.fftfreq(n) power = np.abs(fft_data) ** 2 power[np.isinf(power)] = 0 return freq, power