# source: https://raw.githubusercontent.com/jaredrsommer/freqtradestrategies/63cd46ea087ed8ed01f7d54d8cfcf1dbed166b04/HurstCycleV5%20(2).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 typing import Dict, List, Optional, Union from datetime import datetime, timedelta from freqtrade.persistence import Trade from freqtrade.strategy import (BooleanParameter, CategoricalParameter, DecimalParameter, IStrategy, IntParameter, RealParameter, merge_informative_pair) import pandas_ta as pta from datetime import datetime, timedelta from freqtrade.persistence import Trade import pandas_ta as pta import logging logger = logging.getLogger(__name__) class Github_jaredrsommer_freqtradestrategies__HurstCycleV5__20260509_225308(IStrategy): timeframe = '1h' minimal_roi = {"0": 1} stoploss = -0.99 trailing_stop = False trailing_stop_positive = 0.04 can_short = True position_adjustment_enable = True 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(70, 150, default=250, space='buy', optimize=True, load=True) l_window_size = IntParameter(20, 50, default=42, space='buy', optimize=True, load=True) rsi_period = IntParameter(7, 21, default=21, space="buy", optimize=True, load=True) def custom_stake_amount(self, pair: str, current_time: datetime, current_rate: float, proposed_stake: float, min_stake: float, max_stake: float, leverage: float, entry_tag: Optional[str], side: str, **kwargs) -> float: # Dividiamo lo stake proposto in tre parti uguali return proposed_stake / 3 def adjust_trade_position(self, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, min_stake: float, max_stake: float, **kwargs) -> Optional[float]: dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe) last_candle = dataframe.iloc[-1].squeeze() trade_duration = (current_time - trade.open_date_utc).seconds / 60 # Contiamo il numero di ordini già eseguiti per questo trade filled_entries = trade.nr_of_successful_entries # Se abbiamo già 3 entrate, non ne aggiungiamo altre if filled_entries >= 3: return None if current_profit > 0: return None # Verifichiamo il tempo trascorso dall'ultima entrata if filled_entries > 0 and trade_duration < 240: return None if filled_entries > 1 and trade_duration < 240 * 1.5: return None if filled_entries > 2 and trade_duration < 240 * 2.25: return None # Verifichiamo se ci sono segnali di entrata nelle ultime candele long_entry_signals = dataframe['enter_long'].rolling(window=3).sum().iloc[-1] short_entry_signals = dataframe['enter_short'].rolling(window=3).sum().iloc[-1] # Se abbiamo segnali di entrata e il prezzo è sotto il VTL up if not trade.is_short and long_entry_signals > 0: logger.info(f"{trade.pair} DCA Long Signal Detected") # Calcoliamo il nuovo stake (uguale allo stake iniziale) return trade.stake_amount if trade.is_short and short_entry_signals > 0: logger.info(f"{trade.pair} DCA Short Signal Detected") # Calcoliamo il nuovo stake (uguale allo stake iniziale) return trade.stake_amount return None def linear_regression_channel(self, data: pd.Series, window: int, num_dev: float): """ Calcola la linea di regressione e il canale di deviazione standard (bande superiore e inferiore). :param data: Serie di dati (prezzi di chiusura) :param window: Lunghezza della finestra di regressione :param num_dev: Numero di deviazioni standard per le bande :return: Linea centrale (regressione), banda superiore e banda inferiore """ # Lista per contenere i valori di output lr_channel = {'mid': [], 'upper': [], 'lower': []} for i in range(window, len(data)): # Seleziona la finestra corrente y = data[i-window:i] # Calcola l'indice del tempo per la finestra x = np.arange(window) # Regressione lineare sui dati della finestra A = np.vstack([x, np.ones(len(x))]).T slope, intercept = np.linalg.lstsq(A, y, rcond=None)[0] # Calcola la linea centrale (y = mx + b) y_line = intercept + slope * x[-1] # Calcola la deviazione standard residuals = y - (intercept + slope * x) std_dev = np.std(residuals) # Linea centrale, banda superiore e inferiore lr_channel['mid'].append(y_line) lr_channel['upper'].append(y_line + num_dev * std_dev) lr_channel['lower'].append(y_line - num_dev * std_dev) # Riempire i valori iniziali con NaN per mantenere la lunghezza della serie uguale lr_channel['mid'] = [np.nan] * window + lr_channel['mid'] lr_channel['upper'] = [np.nan] * window + lr_channel['upper'] lr_channel['lower'] = [np.nan] * window + lr_channel['lower'] return pd.DataFrame(lr_channel) 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) # Troughs and Crests dataframe['trough'] = dataframe['filtered_close'].rolling(self.h2).min() dataframe['crest'] = dataframe['filtered_close'].rolling(self.h2).max() dataframe['is_trough'] = np.where(dataframe['filtered_close'] == dataframe['trough'], 1, 0) dataframe['is_crest'] = np.where(dataframe['filtered_close'] == dataframe['crest'], 1, 0) # VTL based on last two troughs/crests # Initialize VTL columns dataframe['vtl_up'] = np.nan dataframe['vtl_down'] = np.nan # Group troughs and crests within a window (self.h2 candles) group_window = self.h2 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, 'is_trough'] == 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, 'is_crest'] == 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, 'vtl_up'] = 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, 'vtl_down'] = 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 # 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) dataframe['rsi'] = pta.rsi(dataframe['close'], length=self.rsi_period.value) dataframe['rsi_fast'] = pta.rsi(dataframe['close'], length=4) dataframe['rsi1'] = ((pta.rsi(dataframe['close'], length=self.rsi_period.value*4) -50) *2.15) +50 dataframe['rsi4'] = ((pta.rsi(dataframe['close'], length=self.rsi_period.value*4*4) -50) *3.75) +50 dataframe['rsi8'] = ((pta.rsi(dataframe['close'], length=self.rsi_period.value*4*8) -50) *5) +50 regression_channel = self.linear_regression_channel(dataframe['rsi_fast'], window=150, num_dev=1.0) dataframe['lr_mid_rsi'] = regression_channel['mid'] dataframe['lr_upper_rsi'] = regression_channel['upper'] dataframe['lr_lower_rsi'] = regression_channel['lower'] return dataframe def populate_entry_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame: dataframe.loc[ # (dataframe['is_trough'] == 1) & # (dataframe['filtered_close'] < dataframe['fld_short']) & # (dataframe['filtered_close'] < dataframe['agreeance_lower']) & # (dataframe['converging'] == True) & # (dataframe['trend'] == 1) & ~(dataframe['vtl_up'] > dataframe['vtl_down']) & (dataframe['lr_mid_rsi'] < 35) & (dataframe['close'].rolling(5).min() < dataframe['vtl_up']) & (dataframe['close'] > dataframe['vtl_up']), 'enter_long'] = 1 dataframe.loc[ # (dataframe['is_trough'] == 1) & # (dataframe['filtered_close'] < dataframe['fld_short']) & # (dataframe['filtered_close'] < dataframe['agreeance_lower']) & # (dataframe['converging'] == True) & # (dataframe['trend'] == 1) & ~(dataframe['vtl_up'] > dataframe['vtl_down']) & (dataframe['lr_mid_rsi'] > 65) & (dataframe['close'].rolling(5).max() > dataframe['vtl_down']) & (dataframe['close'] < dataframe['vtl_down']), 'enter_short'] = 1 return dataframe def populate_exit_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame: dataframe.loc[ # (dataframe['enter_long'].shift(1) == 1) & (dataframe['close'] > dataframe['vtl_down']), 'exit_long'] = 1 dataframe.loc[ # (dataframe['enter_long'].shift(1) == 1) & (dataframe['close'] < dataframe['vtl_up']), 'exit_short'] = 1 return dataframe def leverage(self, pair: str, current_time, current_rate: float, proposed_leverage: float, max_leverage: float, side: str, **kwargs) -> float: return 5.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