💬 Forum

HPStrategyV6

🏆 League #35 / 1909

remiotore/ccxt-freqtrade/strategies/HPStrategyV6.py · ★3 · ⑂2 · first seen 2026-07-16 · repo updated 2026-01-11 · ⬇ 1 download

Basics mode: spot timeframe: 15m interface version: 3
Settings has minimal roi trailing dca custom stoploss hyperopt hyperopt params: 2
Indicators ATR CCI RSI pandas_ta talib
Concepts dca trailing
15 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
import logging
import numpy as np  # noqa
import pandas as pd  # noqa
from pandas import DataFrame
from typing import Optional, Union
from freqtrade.persistence import Trade, Order
from freqtrade.strategy import (BooleanParameter, CategoricalParameter, DecimalParameter,
                                IStrategy, IntParameter)


import datetime
import talib.abstract as ta
import freqtrade.vendor.qtpylib.indicators as qtpylib
import pandas_ta as pta

class HPStrategyV6(IStrategy):
    INTERFACE_VERSION = 3
    timeframe = '15m'
    leverage_value = 3

    minimal_roi = {
        "0": 0.03
    }

    last_dca_timeframe = {}
    max_entry_position_adjustment = 5
    max_dca_multiplier = 5.5
    open_trade_limit = 6
    position_adjustment_enable = True
    dca_threshold_pct = DecimalParameter(0.01, 0.20, default=0.04 * leverage_value, decimals=2, space='buy',
                                         optimize=position_adjustment_enable)
    rolling_ha_treshold = IntParameter(3, 10, default=9, space='buy', optimize=True)
    trailing_stop = True
    trailing_only_offset_is_reached = True
    trailing_stop_positive = 0.003
    trailing_stop_positive_offset = 0.01
    stoploss = -0.15 * leverage_value
    use_exit_signal = True
    ignore_roi_if_entry_signal = True
    use_custom_stoploss = True

    order_types = {
        'entry': 'market',
        'exit': 'market',
        'stoploss': 'market',
        'stoploss_on_exchange': False
    }

    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:
        return self.leverage_value

    def calculate_heiken_ashi(self, dataframe):
        if dataframe.empty:
            raise ValueError("DataFrame je prázdný")
        heiken_ashi = pd.DataFrame(index=dataframe.index)
        heiken_ashi['HA_Close'] = (dataframe['open'] + dataframe['high'] + dataframe['low'] + dataframe['close']) / 4
        heiken_ashi['HA_Open'] = heiken_ashi['HA_Close'].shift(1)
        heiken_ashi['HA_Open'].iloc[0] = heiken_ashi['HA_Close'].iloc[0]
        heiken_ashi['HA_High'] = heiken_ashi[['HA_Open', 'HA_Close']].join(dataframe['high'], how='inner').max(axis=1)
        heiken_ashi['HA_Low'] = heiken_ashi[['HA_Open', 'HA_Close']].join(dataframe['low'], how='inner').min(axis=1)

        heiken_ashi['HA_Close'] = heiken_ashi['HA_Close'].rolling(window=self.rolling_ha_treshold.value).mean()
        heiken_ashi['HA_Open'] = heiken_ashi['HA_Open'].rolling(window=self.rolling_ha_treshold.value).mean()
        heiken_ashi['HA_High'] = heiken_ashi['HA_High'].rolling(window=self.rolling_ha_treshold.value).mean()
        heiken_ashi['HA_Low'] = heiken_ashi['HA_Low'].rolling(window=self.rolling_ha_treshold.value).mean()

        return heiken_ashi

    def should_already_sell(self, dataframe):
        heiken_ashi = self.calculate_heiken_ashi(dataframe)
        last_candle = heiken_ashi.iloc[-1]
        if last_candle['HA_Close'] > last_candle['HA_Open']:
            return False
        else:
            return True

    def adjust_entry_price(self, trade: Trade, order: Optional[Order], pair: str,
                           current_time: datetime, proposed_rate: float, current_order_rate: float,
                           entry_tag: Optional[str], side: str, **kwargs) -> float:

        return proposed_rate

    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:
        return (Trade.get_open_trade_count() < self.open_trade_limit)

    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:
        force_reasons = ['force_sell', 'force_exit']
        dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe)
        if 'trailing' in exit_reason:
            return rate > trade.open_rate + self.trailing_stop_positive_offset * self.leverage_value
        if exit_reason in force_reasons:
            return True
        should_already_sell = self.should_already_sell(dataframe)
        if should_already_sell:
            return True
        return False

    def custom_stoploss(self, pair: str, trade: 'Trade', current_time: 'datetime', current_rate: float,
                        current_profit: float, **kwargs) -> float:
        dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
        atr = dataframe.iloc[-1]['atr']
        atr_multiplier = 3
        stop_loss_atr = atr * atr_multiplier
        stop_loss_percentage = -stop_loss_atr / current_rate
        return max(stop_loss_percentage, self.stoploss)

    def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe['cci'] = ta.CCI(dataframe, timeperiod=14)
        dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14)
        dataframe['atr'] = ta.ATR(dataframe, timeperiod=14)
        return dataframe

    def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe.loc[(dataframe['cci'] < -100) &
                      (dataframe['rsi'] < 30) &
                      (dataframe['volume'] > 0), ['enter_long', 'enter_tag']] = (1, 'cci_buy')
        return dataframe

    def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
        dataframe.loc[(dataframe['cci'] > 100) &
                      (dataframe['rsi'] > 70) &
                      (dataframe['volume'] > 0), ['exit_long', 'exit_tag']] = (1, 'cci_sell')
        return dataframe


    def custom_stake_amount(self, pair: str, current_time: datetime, current_rate: float,
                            proposed_stake: float, min_stake: Optional[float], max_stake: float,
                            leverage: float, entry_tag: Optional[str], side: str,
                            **kwargs) -> float:
        return proposed_stake / self.max_dca_multiplier



    def adjust_trade_position(self, trade: Trade, current_time: datetime,
                              current_rate: float, current_profit: float,
                              min_stake: Optional[float], max_stake: float,
                              current_entry_rate: float, current_exit_rate: float,
                              current_entry_profit: float, current_exit_profit: float,
                              **kwargs) -> Optional[float]:

        last_dca_time = self.last_dca_timeframe.get(trade.id,
                                                    datetime.datetime.min.replace(tzinfo=datetime.timezone.utc))
        if current_time - last_dca_time < datetime.timedelta(minutes=self.timeframe_to_minutes(self.timeframe)):
            return None  # Skip DCA if already done in the current timeframe

        if current_profit > self.dca_threshold_pct.value and trade.nr_of_successful_exits == 0:

            return -(trade.stake_amount / 2)
        if current_profit > -self.dca_threshold_pct.value:
            return None

        dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe)

        last_candle = dataframe.iloc[-1].squeeze()
        previous_candle = dataframe.iloc[-2].squeeze()
        if last_candle['close'] < previous_candle['close']:
            return None

        filled_entries = trade.select_filled_orders(trade.entry_side)
        count_of_entries = trade.nr_of_successful_entries
        try:
            stake_amount = filled_entries[0].stake_amount
            stake_amount = stake_amount * (1 + (count_of_entries * 0.25))

            self.last_dca_timeframe[trade.id] = current_time
            return stake_amount
        except Exception as exception:
            return None
        return None

    def timeframe_to_minutes(self, timeframe: str) -> int:
        """Convert a timeframe string to minutes."""
        if 'm' in timeframe:
            return int(timeframe.replace('m', ''))
        elif 'h' in timeframe:
            return int(timeframe.replace('h', '')) * 60
        elif 'd' in timeframe:
            return int(timeframe.replace('d', '')) * 1440
        else:
            raise ValueError(f"Unsupported timeframe: {timeframe}")