# source: https://raw.githubusercontent.com/mkajnar/HighProfitStrategy/7b9ec9775ecafaee4b197863900ac2f19d9761d9/HPStrategyV7UltraDCACSL.py import logging import os import sys import json from functools import reduce import numpy import numpy as np # noqa import pandas as pd # noqa from pandas import DataFrame from typing import Optional, Union, List from pandas_ta import stdev from freqtrade.enums import ExitCheckTuple from freqtrade.persistence import Trade, Order from freqtrade.strategy import (BooleanParameter, CategoricalParameter, DecimalParameter, IStrategy, IntParameter, informative) from datetime import timedelta, datetime, timezone import talib.abstract as ta import freqtrade.vendor.qtpylib.indicators as qtpylib import pandas_ta as pta class Github_mkajnar_HighProfitStrategy__HPStrategyV7UltraDCACSL__20240310_013741(IStrategy): INTERFACE_VERSION = 3 timeframe = '5m' leverage_value = 3 stoploss = -0.3 csl = {} minimal_roi = { "0": 0.007 * leverage_value } allprofits = {} candle_open_prices = {} max_open_trades = 20 process_only_new_candles = True startup_candle_count = 50 trailing_stop = True trailing_only_offset_is_reached = False trailing_stop_positive = 0.003 * leverage_value trailing_stop_positive_offset = 0.007 * leverage_value use_exit_signal = False use_custom_stoploss = True ignore_roi_if_entry_signal = False position_adjustment_enable = True custom_tp_pct = DecimalParameter(0.005, 0.50, default=0.005, decimals=3, space='sell', optimize=True) dca_threshold_pct_k = DecimalParameter(0.90, 0.99, default=0.94, decimals=2, space='buy', optimize=position_adjustment_enable) dca_threshold_pct = DecimalParameter(0.1, 0.5, default=0.3 / leverage_value, decimals=2, space='buy', optimize=position_adjustment_enable) stoploss = dca_threshold_pct.value - 1.25 dca_multiplier = DecimalParameter(0.5, 3, default=2.71, decimals=2, space='buy', optimize=position_adjustment_enable) dca_limit = IntParameter(1, 5, default=1, space='buy', optimize=position_adjustment_enable) donchian_period = IntParameter(5, 50, default=23, space='buy', optimize=True) rsi_treshold = IntParameter(10, 50, default=35, space='buy', optimize=True) cci_treshold = IntParameter(-200, -10, default=-100, space='buy', optimize=True) kick_off_threshold = DecimalParameter(-0.99, 0, default=-0.32, decimals=2, space='sell', optimize=True) pct3_buy_threshold = DecimalParameter(-0.999, -0.005, default=-0.01, decimals=3, space='buy', optimize=True) red_candles_before_buy = IntParameter(1, 5, default=1, space='buy', optimize=True) candle_time_threshold = DecimalParameter(0.01, 1, default=0.36, decimals=2, space='buy', optimize=False) trade_timeout = IntParameter(1, 48, default=12, space='sell', optimize=True) max_tradable_ratio = DecimalParameter(0.01, 1, default=0.75, space='buy', optimize=True) max_trades = IntParameter(1, 30, default=max_open_trades, space='buy', optimize=True) exit_profit_offset = 0.001 exit_profit_only = False order_types = { 'entry': 'market', 'exit': 'market', 'stoploss': 'market', 'stoploss_on_exchange': False } def calc_donchian_channels(self, dataframe, period: int): dataframe["upperDon"] = dataframe["high"].rolling(period).max() dataframe["lowerDon"] = dataframe["low"].rolling(period).min() dataframe["midDon"] = (dataframe["upperDon"] + dataframe["lowerDon"]) / 2 return dataframe def mid_don_cross_over(self, dataframe): dataframe["position_m"] = np.nan dataframe["position_m"] = np.where(dataframe["close"] > dataframe["midDon"], 1, dataframe["position_m"]) dataframe["position_m"] = dataframe["position_m"].ffill().fillna(0) return dataframe def don_channel_breakout(self, dataframe): dataframe["position_b"] = np.nan dataframe["position_b"] = np.where(dataframe["close"] > dataframe["upperDon"].shift(1), 1, dataframe["position_b"]) dataframe["position_b"] = dataframe["position_b"].ffill().fillna(0) return dataframe def don_reversal(self, dataframe): dataframe["position_r"] = np.nan dataframe["position_r"] = np.where(dataframe["close"] < dataframe["lowerDon"].shift(1), 1, dataframe["position_r"]) dataframe["position_r"] = dataframe["position_r"].ffill().fillna(0) return dataframe 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_price_change_coefficient(self, dataframe: DataFrame, candles_count: int = 1) -> DataFrame: dataframe['price_change_pct'] = dataframe['close'].pct_change(periods=candles_count) dataframe['price_change_coeff'] = dataframe['price_change_pct'].apply( lambda x: max(min(x, 0.99), -0.99) if x != 0 else 0.01 * np.sign(x) ) dataframe[f'price_change_coeff_{candles_count}'] = dataframe['price_change_coeff'].rolling( window=candles_count).mean() * self.leverage_value return dataframe def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14) dataframe['cci'] = ta.CCI(dataframe, timeperiod=20) # Swing high/low dataframe = self.calc_swings(dataframe) dataframe = self.calc_donchian_channels(dataframe=dataframe, period=self.donchian_period.value) dataframe = self.mid_don_cross_over(dataframe=dataframe) dataframe = self.don_reversal(dataframe=dataframe) dataframe = self.don_channel_breakout(dataframe=dataframe) dataframe = self.calculate_price_change_coefficient(dataframe, 3) return dataframe def calc_swings(self, dataframe): dataframe['swing_low'] = (dataframe['close'].shift(2) > dataframe['close'].shift(1)) & \ (dataframe['close'].shift(1) < dataframe['close']).astype(int) dataframe['swing_high'] = (dataframe['close'].shift(2) < dataframe['close'].shift(1)) & \ (dataframe['close'].shift(1) > dataframe['close']).astype(int) return dataframe def check_red_candles(self, dataframe: DataFrame, n: int) -> DataFrame: red_candle = dataframe['close'] < dataframe['open'] dataframe.loc[:, 'red_candles_in_row'] = red_candle.rolling(window=n).sum() == n return dataframe def timeframe_to_minutes(self, timeframe: str) -> int: 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}") def is_candle_open_more_than_threshold(self, dataframe: DataFrame, threshold: float) -> DataFrame: candle_timeframe_minutes = self.timeframe_to_minutes(timeframe=self.timeframe) t_time = pd.to_timedelta(candle_timeframe_minutes * threshold, unit='minutes') current_time = pd.Timestamp.utcnow() dataframe.loc[:, 'time_since_open'] = current_time - dataframe['date'] dataframe.loc[:, 'is_open_more_than_threshold'] = dataframe['time_since_open'] >= t_time return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: self.check_red_candles(dataframe, self.red_candles_before_buy.value) self.is_candle_open_more_than_threshold(dataframe, threshold=self.candle_time_threshold.value) dataframe.loc[ ( # ((dataframe['position_r'] == 1) | # (dataframe['swing_low'] == 1) | # (dataframe['position_m'] == 1) | # (dataframe['position_b'] == 1)) & dataframe['red_candles_in_row'] & dataframe['is_open_more_than_threshold'] & (dataframe['rsi'] < self.rsi_treshold.value) & (dataframe['cci'] < self.cci_treshold.value) & ( ( (dataframe['price_change_coeff_3'] < -0.003) & (dataframe['price_change_coeff_3'] > -0.025) ) | ( (dataframe['price_change_coeff_3'] > 0.003) & (dataframe['price_change_coeff_3'] < 0.025) ) ) ), ['enter_long', 'enter_tag'] ] = (1, 'swing_low') return dataframe def custom_exit(self, pair: str, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> Optional[Union[str, bool]]: dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = dataframe.iloc[-1].squeeze() if last_candle['price_change_coeff_3'] <= self.kick_off_threshold.value: return 'kick_off' def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: dataframe.loc[ ( (dataframe['swing_high'] == 1) ), ['exit_long', 'exit_tag'] ] = (1, 'swing_high') dataframe.loc[ ( (dataframe['price_change_coeff_3'] <= self.kick_off_threshold.value) ), ['exit_long', 'exit_tag'] ] = (1, 'kick_off') return dataframe 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: dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) profit_ratio = trade.calc_profit_ratio(rate) if 'swing' in exit_reason or 'trailing' in exit_reason: confirm_pf = profit_ratio > self.exit_profit_offset return confirm_pf if 'stop_loss' in exit_reason or 'kick_off' in exit_reason or 'force' in exit_reason: return True return True def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> float: try: if pair in self.csl.keys(): o = json.loads(self.csl[pair]) return o.get('sl', self.stoploss) except: pass return self.stoploss 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: max_total_stake = max_stake * self.max_tradable_ratio.value return min(max_stake, max_total_stake / self.max_trades.value) 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]: dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe) last_candle = dataframe.iloc[-1] candle_open_price = last_candle['open'] if self.candle_open_prices.get(trade.pair, None) == candle_open_price: return None if trade.open_rate != current_rate: try: if trade.pair in self.csl.keys(): o = json.loads(self.csl[trade.pair]) tp = o.get('tp', None) if tp is not None and current_rate >= tp: logging.info(f"Hit {trade.pair} to TP: {self.csl[trade.pair]}, at {current_rate}") o['tp'] = None o['sl'] = None self.csl[trade.pair] = json.dumps(o) return -trade.stake_amount except: pass filled_entries = trade.select_filled_orders(trade.entry_side) if len(filled_entries) > self.dca_limit.value: return None new_stop_loss_price = trade.open_rate * (1 + self.stoploss) tp_price = trade.open_rate * (1 + self.custom_tp_pct.value) info = { "sl": new_stop_loss_price, "tp": tp_price } self.csl[trade.pair] = json.dumps(info) # logging.info(f"Updated CSL: {self.csl[trade.pair]}") try: if current_rate <= (trade.open_rate * (1 - self.dca_threshold_pct.value)): self.candle_open_prices[trade.pair] = candle_open_price p = trade.stake_amount * self.dca_multiplier.value # logging.info( # f"[ADJ] {trade.pair} current SL adjusted to {self.csl[trade.pair]}, REBUY {p} USDT") return p except Exception as e: pass return None