✨ AI Search 🔒
💬 Forum

HurstCycle3

🏆 League #1905 / 1922
♡

jaredrsommer/freqtradestrategies/HurstCycle3.py · ★22 · ⑂9 · first seen 2026-07-16 · repo updated 2026-05-18 · ⬇ 1 download

Basics mode: spot timeframe: 4h
Settings stoploss: -0.04 has minimal roi trailing hyperopt hyperopt params: 1
Indicators ATR Heikin_Ashi SMA pandas_ta scipy talib technical
Concepts breakout trailing
Methods leverage
8 related strategies (⧉ identical code, ≈ similar name)

Each tile is a different kind of check — from an instant code lint to full sandboxed backtests and forward tests on recent data. Not sure what a check actually proves? See the FAQ →

Source

Download Raw 
  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
from freqtrade.strategy import IStrategy
import pandas as pd
import numpy as np
import talib
from scipy.fft import fft
from typing import Dict, List
import logging
import numpy as np
import pandas as pd
from technical import qtpylib
from pandas import DataFrame, Series
from datetime import datetime, timezone
from typing import Optional
from functools import reduce
import talib.abstract as ta
import talib
import pandas_ta as pta
from freqtrade.strategy import (BooleanParameter, CategoricalParameter, DecimalParameter, 
                                IStrategy, IntParameter, RealParameter, merge_informative_pair)
import freqtrade.vendor.qtpylib.indicators as qtpylib
from freqtrade.persistence import Trade
import warnings
from pandas.errors import PerformanceWarning
from scipy.signal import savgol_filter
import math

class HurstCycle3(IStrategy):
    timeframe = '4h'
    minimal_roi = {"0": 0.05, "60": 0.03, "120": 0.01}
    stoploss = -0.04
    trailing_stop = True
    trailing_stop_positive = 0.015
    trailing_stop_positive_offset = 0.02

    # Hyperparameters to optimize
    base_cycle_period = 20
    envelope_factor_short = 1.2
    envelope_factor_long = 2.0
    filter_weights = [1, 2, 3, 2, 1]  # Symmetric weights for smoothing (p. 177)
    convergence_threshold = DecimalParameter(0.002, 0.01, default=0.005, space='buy', optimize=True)

    def populate_indicators(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        # Spectral Analysis
        heikinashi = qtpylib.heikinashi(dataframe)

        dataframe['ha_open'] = heikinashi['open']
        dataframe['ha_close'] = heikinashi['close']
        dataframe['ha_high'] = heikinashi['high']
        dataframe['ha_low'] = heikinashi['low']

        close = dataframe['ha_close'].values
        n = min(len(close), 500)
        yf = fft(close[-n:])
        freq = np.fft.fftfreq(n)
        power = np.abs(yf) ** 2
        dominant_idx = np.argmax(power[1:n//2]) + 1
        dominant_period = int(1 / freq[dominant_idx]) if freq[dominant_idx] != 0 else self.base_cycle_period
        self.short_cycle = max(10, min(30, dominant_period))
        self.long_cycle = self.short_cycle * 2

        # Numerical Filter
        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 (for convergence check)
        half_short = math.ceil(self.short_cycle / 4)
        half_mid = math.ceil(self.short_cycle / 2)
        half_long = math.ceil(self.long_cycle / 2)
        dataframe['filtered_close_smooth'] = dataframe['filtered_close'].rolling(window=2).mean()
        fld_short_base = dataframe['filtered_close'].rolling(window=int(self.short_cycle/2), center=True).mean()
        fld_mid_base = dataframe['filtered_close'].rolling(window=self.short_cycle, center=True).mean()
        fld_long_base = dataframe['filtered_close'].rolling(window=self.long_cycle, 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
        fld_spread = dataframe[['fld_short', 'fld_mid', 'fld_long']].max(axis=1) - \
                     dataframe[['fld_short', 'fld_mid', 'fld_long']].min(axis=1)
        relative_spread = fld_spread / dataframe['filtered_close']
        dataframe['converging'] = relative_spread < self.convergence_threshold.value
        band_center = dataframe[['fld_short', 'fld_mid', 'fld_long']].mean(axis=1)
        band_width = dataframe['filtered_close'] * 0.01
        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)


        # [Rest of indicators: envelopes, VTLs, trend...]
        dataframe['sma_short'] = talib.SMA(dataframe['filtered_close'], timeperiod=self.short_cycle)
        dataframe['sma_long'] = talib.SMA(dataframe['filtered_close'], timeperiod=self.long_cycle)
        dataframe['atr_short'] = talib.ATR(dataframe['high'], dataframe['low'], dataframe['filtered_close'], timeperiod=self.short_cycle)
        dataframe['atr_long'] = talib.ATR(dataframe['high'], dataframe['low'], dataframe['filtered_close'], timeperiod=self.long_cycle)
        dataframe['upper_env_short'] = dataframe['sma_short'] + self.envelope_factor_short * dataframe['atr_short']
        dataframe['lower_env_short'] = dataframe['sma_short'] - self.envelope_factor_short * dataframe['atr_short']
        dataframe['upper_env_long'] = dataframe['sma_long'] + self.envelope_factor_long * dataframe['atr_long']
        dataframe['lower_env_long'] = dataframe['sma_long'] - self.envelope_factor_long * dataframe['atr_long']
        dataframe['trough'] = dataframe['filtered_close'].rolling(self.long_cycle).min()
        dataframe['crest'] = dataframe['filtered_close'].rolling(self.long_cycle).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)
        troughs = dataframe[dataframe['is_trough'] == 1].index
        crests = dataframe[dataframe['is_crest'] == 1].index
        if len(troughs) >= 2:
            t1, t2 = troughs[-2], troughs[-1]
            dataframe['vtl_up_slope'] = (dataframe['filtered_close'][t2] - dataframe['filtered_close'][t1]) / (t2 - t1)
        if len(crests) >= 2:
            c1, c2 = crests[-2], crests[-1]
            dataframe['vtl_down_slope'] = (dataframe['filtered_close'][c2] - dataframe['filtered_close'][c1]) / (c2 - c1)
        dataframe['trend'] = np.where(dataframe['filtered_close'] > dataframe['sma_long'], 1, -1)

        # Calculate distances from filtered_close to each line
        dataframe['dist_upper_short'] = dataframe['upper_env_short'] - dataframe['filtered_close']
        dataframe['dist_lower_short'] = dataframe['filtered_close'] - dataframe['lower_env_short']
        dataframe['dist_upper_long'] = dataframe['upper_env_long'] - dataframe['filtered_close']
        dataframe['dist_lower_long'] = dataframe['filtered_close'] - dataframe['lower_env_long']
        dataframe['dist_fld_short'] = abs(dataframe['filtered_close'] - dataframe['fld_short'])
        dataframe['dist_fld_mid'] = abs(dataframe['filtered_close'] - dataframe['fld_mid'])
        dataframe['dist_fld_long'] = abs(dataframe['filtered_close'] - dataframe['fld_long'])

        # Normalize distances as percentage of filtered_close
        for col in ['dist_upper_short', 'dist_lower_short', 'dist_upper_long', 
                    'dist_lower_long', 'dist_fld_short', 'dist_fld_mid', 'dist_fld_long']:
            dataframe[f'{col}_norm'] = dataframe[col] / dataframe['filtered_close']

        # Calculate spread metrics for nesting detection
        envelope_spread_short = dataframe['upper_env_short'] - dataframe['lower_env_short']
        envelope_spread_long = dataframe['upper_env_long'] - dataframe['lower_env_long']
        fld_spread = dataframe[['fld_short', 'fld_mid', 'fld_long']].max(axis=1) - \
                     dataframe[['fld_short', 'fld_mid', 'fld_long']].min(axis=1)
        
        # Normalized spreads
        dataframe['env_spread_short_norm'] = envelope_spread_short / dataframe['filtered_close']
        dataframe['env_spread_long_norm'] = envelope_spread_long / dataframe['filtered_close']
        dataframe['fld_spread_norm'] = fld_spread / dataframe['filtered_close']

        # Nesting detection
        # We'll consider lines "nesting" when:
        # 1. All distances are within a small threshold
        # 2. Spreads between lines are compressing
        nesting_threshold = 0.01  # 1% of price, adjustable
        dataframe['is_nesting'] = (
            (dataframe['dist_upper_short_norm'] < nesting_threshold) &
            (dataframe['dist_lower_short_norm'] < nesting_threshold) &
            (dataframe['dist_upper_long_norm'] < nesting_threshold) &
            (dataframe['dist_lower_long_norm'] < nesting_threshold) &
            (dataframe['dist_fld_short_norm'] < nesting_threshold) &
            (dataframe['dist_fld_mid_norm'] < nesting_threshold) &
            (dataframe['dist_fld_long_norm'] < nesting_threshold)
        )

        # Nesting trend (are lines coming together?)
        lookback = 5  # Lookback period to detect convergence
        dataframe['nesting_score'] = 0.0
        for col in ['env_spread_short_norm', 'env_spread_long_norm', 'fld_spread_norm']:
            spread_change = dataframe[col] - dataframe[col].shift(lookback)
            # Negative change means spreads are compressing
            dataframe['nesting_score'] += np.where(spread_change < 0, 1, 0)
        # Normalize score (0-3, where 3 means all spreads are compressing)
        dataframe['nesting_score'] = dataframe['nesting_score'] / 3.0

        # Additional metric: average normalized distance to all lines
        distance_cols = [col for col in dataframe.columns if col.startswith('dist_') and col.endswith('_norm')]
        dataframe['avg_line_distance'] = dataframe[distance_cols].mean(axis=1)
        dataframe['avg_dist_mean'] = ta.SMA(dataframe['avg_line_distance'], 200)

        return dataframe
    def populate_entry_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        # Long Entry: Trough, below FLD, short envelope breach, VTL breakout
        # dataframe.loc[
        #     # (dataframe['is_trough'] == 1) &
        #     (dataframe['filtered_close'] > dataframe['filtered_close_smooth']) &
        #     (dataframe['filtered_close'].shift() < dataframe['filtered_close_smooth'].shift()),
        #     # (dataframe['trend'] == 1),
        #     'enter_long'] = 1

        dataframe.loc[
            # (dataframe['is_trough'] == 1) &
            (dataframe['ha_close'] > dataframe['filtered_close']) &
            (dataframe['ha_close'].shift() < dataframe['filtered_close'].shift()) &
            (dataframe['ha_close'] > dataframe['ha_open']),
    
            # (dataframe['trend'] == 1),
            'enter_long'] = 1

        # # Short Entry: Crest, above FLD, short envelope breach, VTL breakout
        # dataframe.loc[
        #     (dataframe['is_crest'] == 1) &
        #     (dataframe['filtered_close'] > dataframe['fld_short']) &
        #     (dataframe['filtered_close'] > dataframe['upper_env_short']) &
        #     (dataframe['trend'] == -1) &
        #     (dataframe['vtl_down_slope'].notna() & (dataframe['vtl_down_slope'] < 0)),
        #     'enter_short'] = 1

        return dataframe

    def populate_exit_trend(self, dataframe: pd.DataFrame, metadata: dict) -> pd.DataFrame:
        # Exit Long: Hit FLD or upper long envelope
        # dataframe.loc[
        #     (dataframe['filtered_close'] < dataframe['filtered_close_smooth']) &
        #     (dataframe['filtered_close'].shift() > dataframe['filtered_close_smooth'].shift()),
        #     'exit_long'] = 1

        dataframe.loc[
            (dataframe['ha_open'] > dataframe['filtered_close']) &
            (dataframe['ha_open'].shift() < dataframe['filtered_close'].shift()) &
            (dataframe['ha_close'] < dataframe['ha_open']),
            'exit_long'] = 1

        # # Exit Short: Hit FLD or lower long envelope
        # dataframe.loc[
        #     (dataframe['enter_short'].shift(1) == 1) &
        #     ((dataframe['filtered_close'] <= dataframe['fld_long']) | 
        #      (dataframe['filtered_close'] <= dataframe['lower_env_long'])),
        #     '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 3.0