# --- Do not remove these libs --- from statistics import median from freqtrade.strategy import IStrategy import sys import os # 添加父目录到系统路径 sys.path.append(os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from ChanLun import ChanLun from ChanLun_Classifier import ChanLunClassifier from ChanEnum import Chan_FX_TYPE, Chan_KLC_FX, Chan_BI_DIR, Chan_KLC_FX # -------------------------------- from technical.util import resample_to_interval, resampled_merge import talib.abstract as ta from pandas import DataFrame from datetime import datetime, timedelta from freqtrade.persistence import Trade from typing import Optional import logging logger = logging.getLogger(__name__) ### Now you can use logger.info('asfd') to log # freqtrade plot-dataframe --strategy ChanLun_SOL_5 --datadir user_data/data/binance -c ./user_data/ChanLun_SOL.json --timerange=20250309- # freqtrade trade -c ./user_data/Chan/config/ChanLun_SOL.json --strategy ChanLun_SOL_5 --strategy-path ./user_data/Chan/strategies # freqtrade backtesting -c ./user_data/Chan/config/ChanLun_SOL.json --strategy ChanLun_SOL_5 --strategy-path ./user_data/Chan/strategies --timerange=20250416- # freqtrade download-data -c ./user_data/Chan/config/ChanLun_SOL.json -t 1m --pairs BTC/USDT:USDT --timerange=20250405- # freqtrade hyperopt --hyperopt-loss SharpeHyperOptLossDaily --spaces roi stoploss --strategy ChanLun_SOL_5 --strategy-path ./user _data/Chan/strategies -c ./user_data/Chan/config/ChanLun_SOL.json -e 200 --timerange=20250201-20250401 # sudo docker compose run --rm chan_btc backtesting -c ./user_data/Chan/config/ChanLun_SOL.json --strategy ChanLun_SOL --strategy-path ./user_data/Chan/strategies --timerange=20250101- # sudo docker compose run --rm chan_btc download-data -c ./user_data/Chan/config/ChanLun_SOL.json --pairs SOL/USDT:USDT -t 1m --timerange 20240101- # sudo docker compose run --rm chan_btc trade -c ./user_data/Chan/config/ChanLun_SOL.json --strategy ChanLun_SOL --strategy-path ./user_data/Chan/strategies class ChanLun_SOL_5(IStrategy): INTERFACE_VERSION: int = 3 # Minimal ROI designed for the strategy. # This attribute will be overridden if the config file contains "minimal_roi" # 30m and 1h minimal_roi = { "0": 0.30, "360": 0.2, "640": 0.1, "1200": 0 } # 5m and 15m minimal_roi_1 = { "0": 0.1, "60": 0.05, "120": 0.02, "240": 0 } # 15m and 30m minimal_roi_1 = { "0": 0.1, "240": 0.05, "480": 0.03, "600": 0 } minimal_roi_2 = { "0": 0.10, "1200": 0.05, "2400": 0.025, "3600": 0 } can_short = True lev = 20.0 stoploss = -0.3 trailing_stop = False trailing_stop_positive = 0.025 trailing_stop_positive_offset = 0.045 trailing_only_offset_is_reached = False position_adjustment_enable = True max_entry_position_adjustment = 3 max_dca_multiplier = 5.5 startup_candle_count = 600 time5 = 5 time15 = 15 time30 = 30 time60 = 60 time4h = 240 time5 = 15 last_time = datetime.now() big_size = 0 big_state = "00" big_state_list = [] chan = ChanLun() small_size = 0 small_state = "00" small_state_list = [] classifier = ChanLunClassifier(None) def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # resample our dataframes dataframe_5 = resample_to_interval(dataframe, self.get_ticker_indicator() * 5) dataframe_15 = resample_to_interval(dataframe, self.get_ticker_indicator() * 15) dataframe_30 = resample_to_interval(dataframe, self.get_ticker_indicator() * 30) dataframe_60 = resample_to_interval(dataframe, self.get_ticker_indicator() * 60) dataframe_4h = resample_to_interval(dataframe, self.get_ticker_indicator() * 240) #dataframe_1d = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe='1d') #dataframe_1w = resample_to_interval(dataframe_1d, self.get_ticker_indicator() * 10080) #dataframe_1m = resample_to_interval(dataframe_1d, self.get_ticker_indicator() * 43200) dataframe_1d = resample_to_interval(dataframe, self.get_ticker_indicator() * 1440) #dataframe_1w = resample_to_interval(dataframe, self.get_ticker_indicator() * 10080) #dataframe_1m = resample_to_interval(dataframe, self.get_ticker_indicator() * 43200) dataframe = self.add_indicators(dataframe) dataframe_5 = self.add_indicators(dataframe_5) dataframe_30 = self.add_indicators(dataframe_30) dataframe_60 = self.add_indicators(dataframe_60) dataframe_4h = self.add_indicators(dataframe_4h) dataframe_1d = self.add_indicators(dataframe_1d) #self.chan.plot_dual(dataframe_5, dataframe_30) dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14) state_list, fx_list = self.chan.get_klc_strength_list(dataframe_15) dataframe_15['state'] = state_list dataframe_15['fx'] = fx_list klc_list = self.chan.get_klc_list(dataframe_15) bi_list = self.chan.cal_bi_list(klc_list) if self.last_time + timedelta(minutes=1) < datetime.now(): print(state_list[-1], state_list[-2], state_list[-3], state_list[-4], state_list[-5]) print(fx_list[-1], fx_list[-2], fx_list[-3], fx_list[-4], fx_list[-5]) print(klc_list[-1].klc_fx_type, klc_list[-2].klc_fx_type, klc_list[-3].klc_fx_type, klc_list[-4].klc_fx_type, klc_list[-5].klc_fx_type) print("-------------------------------------------------------------------------------") self.last_time = datetime.now() #dataframe = resampled_merge(dataframe, dataframe_5) dataframe = resampled_merge(dataframe, dataframe_15) #dataframe = resampled_merge(dataframe, dataframe_30) #dataframe = resampled_merge(dataframe, dataframe_60) #dataframe = resampled_merge(dataframe, dataframe_4h) return dataframe def print_bi_klc_fx(self, dataframe, model_name): klc_list = self.chan.get_full_klc_list(dataframe) bi_list = self.chan.cal_bi_list(klc_list) fx_count_list = [] fx_count_list_up = [] fx_count_list_down = [] self.classifier.load_model(model_name) for bi in bi_list[1:-1]: fx_count = 0 if bi.end_klc: for index in range(bi.start_klc.index, bi.end_klc.index+1): klc = klc_list[index] features = klc.get_feature_data() if bi.dir == Chan_BI_DIR.UP and (klc.klc_fx_type == Chan_KLC_FX.TOP1 or klc.klc_fx_type == Chan_KLC_FX.TOP2): fx_count += 1 print(klc.start_time, klc.klc_fx_type, self.classifier.predict(klc), bi.dir, features['klc_volume_ratio'], features['klc_macdhist'], features['klc_rsi']) else: if bi.dir == Chan_BI_DIR.DOWN and (klc.klc_fx_type == Chan_KLC_FX.BOTTOM1 or klc.klc_fx_type == Chan_KLC_FX.BOTTOM2): fx_count += 1 print(klc.start_time, klc.klc_fx_type, self.classifier.predict(klc), bi.dir, features['klc_volume_ratio'], features['klc_macdhist'], features['klc_rsi']) fx_count_list.append(fx_count) if fx_count == 0: print("Not a bi: ", bi.start_time, bi.end_time, bi.dir) if bi.dir == Chan_BI_DIR.UP: fx_count_list_up.append(fx_count) else: fx_count_list_down.append(fx_count) #print(bi.start_time, bi.end_time, bi.dir, fx_count) avg_count = sum(fx_count_list) / len(fx_count_list) max_count = max(fx_count_list) min_count = min(fx_count_list) median_count = median(fx_count_list) print("Total bi:", len(bi_list), "AVG:", avg_count, "MAX:", max_count, "MIN:", min_count, "MEDIAN:", median_count) for index in range(0, max_count+1): index_count = fx_count_list.count(index) print("Total:", index, "COUNT:", index_count, "RATIO:", index_count/len(fx_count_list)) avg_count_up = sum(fx_count_list_up) / len(fx_count_list_up) avg_count_down = sum(fx_count_list_down) / len(fx_count_list_down) median_count_up = median(fx_count_list_up) median_count_down = median(fx_count_list_down) max_count_up = max(fx_count_list_up) min_count_up = min(fx_count_list_up) max_count_down = max(fx_count_list_down) min_count_down = min(fx_count_list_down) print("Total UP bi:", len(fx_count_list_up), "AVG:", avg_count_up, "MAX:", max_count_up, "MIN:", min_count_up, "MEDIAN:", median_count_up) for index in range(0, max_count_up+1): index_count = fx_count_list_up.count(index) print("UP:", index, "COUNT:", index_count, "RATIO:", index_count/len(fx_count_list_up)) print("Total DOWN bi:", len(fx_count_list_down), "AVG:", avg_count_down, "MAX:", max_count_down, "MIN:", min_count_down, "MEDIAN:", median_count_down) for index in range(0, max_count_down+1): index_count = fx_count_list_down.count(index) print("DOWN:", index, "COUNT:", index_count, "RATIO:", index_count/len(fx_count_list_down)) def print_klc_list(self, klc_list, bi_list): bi_index = 0 for klc in klc_list: if bi_index == len(bi_list): bi_index = len(bi_list) - 1 bi = bi_list[bi_index] if self.check_klc_in_bi(klc, bi): print("KLC in Bi: ", klc.start_time, klc.klc_fx_type, klc.bi.dir, klc.distance, bi.start_time, bi.dir) else: if bi.start_klc.index < klc.index: bi_index += 1 print("KLC not in Bi: ", klc.start_time, klc.klc_fx_type, klc.bi.dir, klc.distance, bi.start_time, bi.dir) else: if klc.bi: print("KLC not in Bi: ", klc.start_time, klc.klc_fx_type, klc.bi.dir, klc.distance, bi.start_time, bi.dir) else: print("KLC not in Bi: ", klc.start_time, klc.klc_fx_type, klc.distance, bi.start_time, bi.dir) def check_klc_in_bi(self, klc, bi): if klc.bi and klc.bi.index == bi.index: return True return False def print_xgb(self, dataframe, model_name): self.classifier.load_model(model_name) klc_list = self.chan.get_klc_list(dataframe) klc1 = klc_list[-1] klc2 = klc_list[-2] klc3 = klc_list[-3] if klc1.end_time == klc2.start_time: print(model_name, klc1.end_time, klc1.fx, self.classifier.predict(klc1)) else: print(model_name, klc1.start_time, klc1.fx, self.classifier.predict(klc1)) print(model_name, klc2.end_time, klc2.fx, self.classifier.predict(klc2)) print(model_name, klc3.end_time, klc3.fx, self.classifier.predict(klc3)) def custom_stake_amount1(self, pair: str, current_time: datetime, current_rate: float, proposed_stake: float, min_stake: float | None, max_stake: float, leverage: float, entry_tag: str | None, side: str, **kwargs) -> float: # We need to leave most of the funds for possible further DCA orders # This also applies to fixed stakes return proposed_stake / self.max_dca_multiplier def adjust_trade_position1(self, trade: Trade, current_time: datetime, current_rate: float, current_profit: float, min_stake: float | None, max_stake: float, current_entry_rate: float, current_exit_rate: float, current_entry_profit: float, current_exit_profit: float, **kwargs ) -> float | None | tuple[float | None, str | None]: """ Custom trade adjustment logic, returning the stake amount that a trade should be increased or decreased. This means extra entry or exit orders with additional fees. Only called when `position_adjustment_enable` is set to True. For full documentation please go to https://www.freqtrade.io/en/latest/strategy-advanced/ When not implemented by a strategy, returns None :param trade: trade object. :param current_time: datetime object, containing the current datetime :param current_rate: Current entry rate (same as current_entry_profit) :param current_profit: Current profit (as ratio), calculated based on current_rate (same as current_entry_profit). :param min_stake: Minimal stake size allowed by exchange (for both entries and exits) :param max_stake: Maximum stake allowed (either through balance, or by exchange limits). :param current_entry_rate: Current rate using entry pricing. :param current_exit_rate: Current rate using exit pricing. :param current_entry_profit: Current profit using entry pricing. :param current_exit_profit: Current profit using exit pricing. :param **kwargs: Ensure to keep this here so updates to this won't break your strategy. :return float: Stake amount to adjust your trade, Positive values to increase position, Negative values to decrease position. Return None for no action. Optionally, return a tuple with a 2nd element with an order reason """ #if trade.has_open_orders: # Only act if no orders are open #return #if current_profit > 0.05 and trade.nr_of_successful_exits == 0: # Take half of the profit at +5% #return -(trade.stake_amount / 2), "half_profit_5%" #if current_profit > -0.05: #return None # Obtain pair dataframe (just to show how to access it) dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe) # Only buy when not actively falling price. #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) last_entry = filled_entries[-1] count_of_entries = trade.nr_of_successful_entries # Allow up to 3 additional increasingly larger buys (4 in total) # Initial buy is 1x # If that falls to -5% profit, we buy 1.25x more, average profit should increase to roughly -2.2% # If that falls down to -5% again, we buy 1.5x more # If that falls once again down to -5%, we buy 1.75x more # Total stake for this trade would be 1 + 1.25 + 1.5 + 1.75 = 5.5x of the initial allowed stake. # That is why max_dca_multiplier is 5.5 # Hope you have a deep wallet! # This returns first order stake size #print(dataframe.iloc[-1]['resample_{}_state'.format(self.get_ticker_indicator()*self.time60)]) # This returns first order stake size stake_amount = filled_entries[0].stake_amount # This then calculates current safety order size stake_amount = stake_amount * (1 + (count_of_entries * 0.5)) dataframe_date = dataframe.iloc[-1]['date'] #print(stake_amount, "---------------------------------------------------") #if last_entry.order_filled_utc + timedelta(minutes=self.time5) < dataframe_date: #if dataframe.iloc[-self.time5*2]['resample_{}_state'.format(self.get_ticker_indicator()*self.time5)] > 1 and last_entry.side == "buy": #print(dataframe.iloc[-self.time5*2]) #print(stake_amount) #return stake_amount, "1/3rd_increase" #if last_entry.order_filled_utc + timedelta(minutes=self.time5*6) < dataframe_date: #if dataframe.iloc[-self.time5*2]['resample_{}_state'.format(self.get_ticker_indicator()*self.time5)] == "10" and last_entry.side == "sell": #print(dataframe.iloc[-self.time5]) #print(stake_amount) #return stake_amount, "1/3rd_increase" return None def log_macd_div_list(self, dataframe): bi_macd_div_list, bi_list, seg_macd_div_list, seg_list = self.chan.get_macd_div_list(dataframe) logger.info(f"BI MACD DIV LIST") for index in range(len(bi_list)-5, len(bi_list)): bi = bi_list[index] logger.info(f'{bi.start_time}, {bi.high}, {bi.low}, {bi.dir}, {bi.macd_div}') logger.info(f"SEG MACD DIV LIST") for index in range(len(seg_list)-5, len(seg_list)): seg = seg_list[index] logger.info(f'{seg.start_bi.start_time}, {seg.high}, {seg.low}, {seg.dir}, {seg.macd_div}') def print_fx(self, df, label=5): fx_list = self.chan.get_bsp_list(self.chan.get_klc_list(df)) fx1 = fx_list[-1] fx2 = fx_list[-2] fx3 = fx_list[-3] fx4 = fx_list[-4] fx5 = fx_list[-5] #print(fx1.end_time, fx1.fx, fx2.end_time, fx2.fx, fx3.end_time, fx3.fx) logger.info(f'\n{fx5.end_time} {fx5.state} {fx4.end_time} {fx4.state} {fx3.end_time} {fx3.state} {fx2.end_time} {fx2.state} {fx1.end_time} {fx1.state} TF: {label}') def print_klc(self, df, label): klc_list = self.chan.get_full_klc_list(df) log_str = f"" for index in range(len(klc_list)-5, len(klc_list)): klc = klc_list[index] fx = str(klc.fx).replace("Chan_FX_TYPE.", "") bi_dir = str(klc.bi.dir).replace("Chan_BI_DIR.", "") klc_fx_type = str(klc.klc_fx_type).replace("Chan_KLC_FX.", "") if klc.end_time: log_str += f"{klc.end_time}E, {bi_dir}, {klc_fx_type}, " else: log_str += f"{klc.start_time}S, {bi_dir}, {klc_fx_type}, " logger.info(label+log_str) def print_fx_list(self, df): bsp_list = self.chan.get_bsp_list(self.chan.get_klc_list(df)) for bsp in bsp_list: logger.info(f'{bsp.start_time}, {bsp.end_time}, {bsp.fx}') def print_df(self, df): for index in range(0, len(df)): state = 'state' rsi = 'rsi' logger.info(f'{df[state][index]}, {df[rsi][index]}, {df[state][index]}') def print_resample_df(self, dataframe, time, limit=10): df = dataframe.tail(limit) if limit > 0: if time == 1: for index in range(len(dataframe) - limit, len(dataframe)): cn1 = 'date' cn2 = 'rsi' cn3 = 'state' logger.info(f'{df[cn1][index]}, {df[cn2][index]}, {df[cn3][index]}') else: for index in range(0, limit): cn1 = 'resample_{}_date'.format(self.get_ticker_indicator()*time) cn2 = 'resample_{}_rsi'.format(self.get_ticker_indicator()*time) cn3 = 'resample_{}_state'.format(self.get_ticker_indicator()*time) logger.info(f'{df[cn1][index]}, {df[cn2][index]}, {df[cn3][index]}') def add_indicators(self, df): fast = 8 slow = 16 period = 6 macd = ta.MACD(df, fastperiod=fast, slowperiod=slow, signalperiod=period) df['macd'] = macd['macd'] df['macdsignal'] = macd['macdsignal'] df['macdhist'] = macd['macdhist'] df['ma5'] = ta.MA(df, timeperiod=5) df['ma10'] = ta.MA(df, timeperiod=10) df['ma30'] = ta.EMA(df, timeperiod=30) df['ma250'] = ta.MA(df, timeperiod=250) df['rsi'] = ta.RSI(df, timeperiod=14) df['volume_ratio'] = self.cal_volume_ratio(df) return df def cal_volume_ratio(self, dataframe, window=10): df = dataframe.copy() # 计算过去N根K线的平均成交量 df['avg_volume'] = df['volume'].rolling(window=window).mean() # 计算量比 df['volume_ratio'] = df['volume'] / df['avg_volume'] # 填充缺失值（前N根K线） df['volume_ratio'] = df['volume_ratio'].fillna(1.0) return df['volume_ratio'] def local_print(self, df): fast = 7 slow = 14 macd = ta.MACD(df, fast=fast, slow=slow) df['macd'] = macd['macd'] df['macdsignal'] = macd['macdsignal'] df['macdhist'] = macd['macdhist'] df['ema26'] = ta.EMA(df, timeperiod=26) df['ema52'] = ta.EMA(df, timeperiod=52) df['ma5'] = ta.MA(df, timeperiod=5) df['ma10'] = ta.MA(df, timeperiod=10) df['masub'] = df['ma5'].subtract(df['ma10']) #for index in range(0, len(df)): #print(df['date'][index], df['macdhist'][index], df['macd'][index], df['macdsignal'][index], df['masub'][index]) # (1,1) = 1, (1,0) = 2, (-1,1) = 3, (-1, 0) = 4, (0,0) = 0 def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: state_str = 'resample_{}_state'.format(self.get_ticker_indicator()*self.time5) fx_str = 'resample_{}_fx'.format(self.get_ticker_indicator()*self.time5) dataframe.loc[ ( #(dataframe['state'] == "-30") (dataframe[state_str].shift(self.time5*2) > 1.0) & (dataframe[fx_str].shift(self.time5*2) == -1) #(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time5)].shift(self.time5) == "-10") & #(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time30)] == "-10") & #(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time5)].shift(self.time5) == "-10") #(qtpylib.crossed_above(dataframe['macd'], dataframe['macdsignal'])) ), ['enter_long', 'enter_tag']] = (1, 'long_signal_chan') dataframe.loc[ ( #(dataframe['state'] == "-30") (dataframe[state_str].shift(self.time5*2) > 1.0) & (dataframe[fx_str].shift(self.time5*2) == 1) #(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time5)].shift(self.time5) == "-10") & #(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time30)] == "-10") & #(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time5)].shift(self.time5) == "-10") #(qtpylib.crossed_above(dataframe['macd'], dataframe['macdsignal'])) ), ['enter_short', 'enter_tag']] = (1, 'short_signal_chan') return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: state_str = 'resample_{}_state'.format(self.get_ticker_indicator()*self.time5) fx_str = 'resample_{}_fx'.format(self.get_ticker_indicator()*self.time5) dataframe.loc[ ( #(dataframe['state']== "30") (dataframe[state_str].shift(self.time5*2) > 1.0) & (dataframe[fx_str].shift(self.time5*2) == 1) #(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time30)] == "10") & #(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time60)] == "10") ), ['exit_long', 'exit_tag']] = (1, 'long_close_signal_chan') dataframe.loc[ ( #(dataframe['state']== "30") (dataframe[state_str].shift(self.time5*2) > 1.0) & (dataframe[fx_str].shift(self.time5*2) == -1) #(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time30)] == "10") & #(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time60)] == "10") ), ['exit_short', 'exit_tag']] = (1, 'short_close_signal_chan') 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.lev def get_ticker_indicator(self): return int(self.timeframe[:-1])