# --- Do not remove these libs --- from freqtrade.strategy import IStrategy from typing import Dict, List from functools import reduce from pandas import DataFrame, pandas import freqtrade.vendor.qtpylib.indicators as qtpylib import sys import os #sys.setrecursionlimit(1000000) #例如这里设置为一百万 #sys.path.append(os.path.abspath("/freqtrade/user_data/Chan")) sys.path.append(os.path.abspath("/Users/jack/Project/freqtrade/user_data/Chan")) #sys.path.append(os.path.abspath("/Users/jack/Documents/GitHub/freqtrade/user_data/Chan")) from ChanLun import ChanLun # -------------------------------- from technical.util import resample_to_interval, resampled_merge import talib.abstract as ta import freqtrade.vendor.qtpylib.indicators as qtpylib from datetime import datetime, timedelta, timezone from freqtrade.persistence import Trade, Order from typing import Optional from ChanPY import ChanPY import logging logger = logging.getLogger(__name__) ### Now you can use logger.info('asfd') to log # freqtrade trade -c ./user_data/Chan.json --strategy Chan_SOL_2 --strategy-path ./user_data/strategies # freqtrade backtesting -c ./user_data/Chan.json --strategy Chan_SOL_2 --strategy-path ./user_data/strategies --timerange=20250309- # freqtrade download-data -c ./user_data/Chan.json -t 1m --pairs SOL/USDT:USDT --timerange=20240101- # freqtrade hyperopt --hyperopt-loss SharpeHyperOptLossDaily --spaces roi stoploss --strategy Chan_SOL_2 --strategy-path ./user_data/strategies -c ./user_data/Chan.json -e 200 --timerange=20250101-20250215 # sudo docker compose run --rm chan_btc backtesting -c ./user_data/Chan.json --strategy Chan_SOL_2 --strategy-path ./user_data/strategies --timerange=20250101- # sudo docker compose run --rm chan_btc download-data -c ./user_data/Chan.json --pairs SOL/USDT:USDT -t 1m --timerange 20240101- # sudo docker compose run --rm chan_btc trade -c ./user_data/Chan.json --strategy Chan_SOL_2 --strategy-path ./user_data/strategies class Chan_SOL_2(IStrategy): INTERFACE_VERSION: int = 3 # Minimal ROI designed for the strategy. # This attribute will be overridden if the config file contains "minimal_roi" minimal_roi = { "0": 0.253, "120": 0.159, "240": 0.052, "360": 0 } can_short = True # Optimal stoploss designed for the strategy # This attribute will be overridden if the config file contains "stoploss" stoploss = -0.21 trailing_stop = False trailing_stop_positive = 0.015 trailing_stop_positive_offset = 0.043 trailing_only_offset_is_reached = False # Optimal timeframe for the strategy # timeframe = '15m' startup_candle_count = 600 time5 = 5 time15 = 15 time30 = 30 time60 = 60 time240 = 240 last_time = datetime.now() big_size = 0 big_state = "00" big_state_list = [] chanpy = ChanPY() chan = ChanLun() small_size = 0 small_state = "00" small_state_list = [] def informative_pairs(self): # get access to all pairs available in whitelist. pairs = self.dp.current_whitelist() # Assign tf to each pair so they can be downloaded and cached for strategy. informative_pairs = [(pair, '1h') for pair in pairs] # Optionally Add additional "static" pairs #informative_pairs += [("ETH/USDT:USDT", "5m"),("ETH/USDT:USDT", "15m"),] return informative_pairs 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) self.local_print(dataframe_5) dataframe_5['state'] = self.chan.resample_klc_list(dataframe_5) #dataframe_15['state'] = self.chan.resample_klc_list(dataframe_15) #dataframe_30['state'] = self.chan.resample_klc_list(dataframe_30) dataframe_60['state'] = self.chan.resample_klc_list(dataframe_60) #dataframe_4h['state'] = self.chan.resample_klc_list(dataframe_4h) #dataframe_5['bsps'], dataframe_5['updown'], dataframe_5['bi_sure'] = self.chanpy.get_bsps(dataframe_5) print("===================================================") #print(dataframe_60['high'].rolling(window).max()) #print(dataframe_60['low'].rolling(window).min()) #for index in range(0, len(dataframe_5)): #print(dataframe_5[dataframe_5['bi_sure'] == 1][dataframe_5['state'] == "-10"]) #print(dataframe_5[dataframe_5['bi_sure'] == 1][dataframe_5['state'] == "10"]) 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_df(self, df): for index in range(0, len(df)): print(df['date'][index], df['rsi'][index], df['state'][index]) def print_resample_df(self, df, time): for index in range(0, len(df)): 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) print(df[cn1][index], df[cn2][index], df[cn3][index]) 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: dataframe.loc[ ( #(dataframe['state'].shift(1) == "-10") & #((dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time5)] == "-11") | (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.time60)].shift(self.time60) == "11") & (dataframe['resample_{}_bsps'.format(self.get_ticker_indicator()*self.time5)].shift(self.time5) > 0) & (dataframe['resample_{}_bsps'.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'].shift(1) == "10") & #((dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time5)] == "11") | (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.time60)].shift(self.time60) == "-11") & (dataframe['resample_{}_bsps'.format(self.get_ticker_indicator()*self.time5)].shift(self.time5) < 0) #(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: dataframe.loc[ ( #(dataframe['state'].shift(1) == "10") & (dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time60)].shift(self.time60) == "10") #(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'].shift(1) == "-10") & (dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time60)].shift(self.time60) == "-10") #(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 1.0 def get_ticker_indicator(self): return int(self.timeframe[:-1])