390 lines
18 KiB
Python
390 lines
18 KiB
Python
# --- 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 ChanEnum import Chan_AUTYPE, Chan_DATA_FIELD, Chan_FX_TYPE, Chan_KLINE_DIR, Chan_KL_TYPE, Chan_BI_DIR
|
|
from ChanKLU import ChanKLU
|
|
from ChanCTime import ChanCTime
|
|
from ChanKLC import ChanKLC
|
|
from ChanBI import ChanBI
|
|
# --------------------------------
|
|
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
|
|
|
|
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_60 --strategy-path ./user_data/strategies
|
|
# freqtrade backtesting -c ./user_data/Chan.json --strategy Chan_SOL_60 --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_60 --strategy-path ./user_data/strategies -c ./user_data/Chan.json -e 200 --timerange=20241111-20241231
|
|
|
|
# sudo docker compose run --rm chan_btc backtesting -c ./user_data/Chan.json --strategy Chan_SOL_60 --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_60 --strategy-path ./user_data/strategies
|
|
|
|
class Chan_SOL_60(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
|
|
time30 = 30
|
|
time60 = 60
|
|
last_time = datetime.now()
|
|
big_size = 0
|
|
big_state = "00"
|
|
big_state_list = []
|
|
|
|
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:
|
|
#macd = ta.MACD(dataframe)
|
|
#dataframe['macd'] = macd['macd']
|
|
#dataframe['macdsignal'] = macd['macdsignal']
|
|
#dataframe['macdhist'] = macd['macdhist']
|
|
#dataframe['ema26'] = ta.EMA(dataframe, timeperiod=26)
|
|
#dataframe['ema52'] = ta.EMA(dataframe, timeperiod=52)
|
|
#dataframe['bsps'], dataframe['updown'], dataframe['bi_sure'] = self.get_bsps(dataframe)
|
|
if not self.dp:
|
|
# Don't do anything if DataProvider is not available.
|
|
return dataframe
|
|
|
|
inf_tf = '1h'
|
|
# Get the informative pair
|
|
#informative = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=inf_tf)
|
|
# 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)
|
|
#self.local_print(dataframe_60)
|
|
dataframe_60['rsi'] = ta.RSI(dataframe_60, timeperiod=14)
|
|
|
|
#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)
|
|
|
|
#klc_list, klu_list = self.get_klc_list(dataframe)
|
|
#dataframe['state'] = self.resample_klc(self.cal_trend(klc_list), len(dataframe))
|
|
#klc_list, klu_list = self.get_klc_list(dataframe)
|
|
#klc_list = self.copy_klu_to_klc(klu_list)
|
|
#dataframe['state'] = self.resample_klc(self.cal_klc_state(klc_list), len(dataframe))
|
|
|
|
|
|
klc_list_5, klu_list_5 = self.get_klc_list(dataframe_5)
|
|
#klc_list_5 = self.copy_klu_to_klc(klu_list_5)
|
|
dataframe_5['state'] = self.resample_klc_list(self.cal_klc_state(klc_list_5), len(dataframe_5))
|
|
|
|
#klc_list_15, klu_list_15 = self.get_klc_list(dataframe_15)
|
|
#dataframe_15['state'] = self.resample_klc(self.cal_trend(klc_list_15), len(dataframe_15))
|
|
|
|
#klc_list_30, klu_list_30 = self.get_klc_list(dataframe_30)
|
|
#dataframe_30['state'] = self.resample_klc(self.cal_klc_state(klc_list_30), len(dataframe_30))
|
|
|
|
klc_list_60, klu_list_60 = self.get_klc_list(dataframe_60)
|
|
#klc_list_60 = self.copy_klu_to_klc(klu_list_60)
|
|
klc_list_60 = self.copy_klu_to_klc(self.get_kl_data(dataframe_60))
|
|
dataframe_60['state'] = self.resample_klc_list(self.cal_klc_state(klc_list_60), len(dataframe_60))
|
|
#for klc in klc_list_60:
|
|
#print(klc.time, klc.state, klc.start_klu.time)
|
|
#dataframe_5['state'] = self.resample_klc_list(self.cal_klc_state(self.copy_klu_to_klc(self.get_kl_data(dataframe_5))), len(dataframe_5))
|
|
#dataframe_60['state'] = self.resample_klc_list(self.cal_klc_state(self.copy_klu_to_klc(self.get_kl_data(dataframe_60))), len(dataframe_60))
|
|
#klc_list_4h, klu_list = self.get_klc_list(dataframe_4h)
|
|
#dataframe_4h['state'] = self.resample_klc(self.cal_state(klc_list_4h), len(dataframe_4h))
|
|
#print(big_dataframe.iloc[-2])
|
|
|
|
#self.print_df(dataframe_60)
|
|
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)
|
|
#self.print_resample_df(dataframe, self.time60)
|
|
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['rsi'] = ta.RSI(df, timeperiod=14)
|
|
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], df['rsi'][index])
|
|
def cal_dataframes(self, dataframe, big_dataframe, time):
|
|
df_list = self.copy_klu_to_klc(self.get_kl_data(dataframe))
|
|
big_df_list = self.copy_klu_to_klc(self.get_kl_data(big_dataframe))
|
|
big_df_state_list = []
|
|
big_df_state_list.append("00")
|
|
for index in range(1, len(big_df_list)-1):
|
|
k1 = big_df_list[index-1]
|
|
k2 = big_df_list[index]
|
|
k3 = df_list[(index+1)*time]
|
|
#print(k2.time, k2.high, k2.low, k3.time, k3.high, k3.low)
|
|
self.get_klc_state(k1, k2, k3)
|
|
big_df_state_list.append(k2.state)
|
|
big_df_state_list.append("00")
|
|
return big_df_state_list
|
|
|
|
# (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)] == "-10")
|
|
(dataframe['resample_{}_rsi'.format(self.get_ticker_indicator()*self.time60)] < 30)
|
|
#(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)] == "10")
|
|
(dataframe['resample_{}_rsi'.format(self.get_ticker_indicator()*self.time60)] > 60)
|
|
#(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)] == "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)] == "-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
|
|
# append when the last klu is not included
|
|
def resample_klc_list(self, klc_list, length):
|
|
re_klc_list = []
|
|
klc_index = 0
|
|
klc = None
|
|
for index in range(0, length):
|
|
if klc_index == len(klc_list):
|
|
klc_index -= 1
|
|
klc = klc_list[klc_index]
|
|
if klc.end_klu and index == klc.end_klu.idx:
|
|
re_klc_list.append(klc.state)
|
|
klc_index += 1
|
|
else:
|
|
re_klc_list.append("00")
|
|
#print(index, klc.time, klc.state, klc.fx, klc.high, klc.low)
|
|
#if self.last_time + timedelta(minutes=1) < datetime.now():
|
|
#print(klc.time, klc.state, klc.fx, re_klc_list[-1], re_klc_list[-2], re_klc_list[-3], re_klc_list[-4], re_klc_list[-5])
|
|
return re_klc_list
|
|
def cal_klc_state(self, klc_list):
|
|
index = 0
|
|
for index in range(1, len(klc_list)-1):
|
|
k1 = klc_list[index-1]
|
|
k2 = klc_list[index]
|
|
k3 = klc_list[index+1]
|
|
self.cal_pattern(k1, k2, k3)
|
|
if k2.fx == Chan_FX_TYPE.TOP:
|
|
k2.set_state("10")
|
|
if k2.fx == Chan_FX_TYPE.BOTTOM:
|
|
k2.set_state("-10")
|
|
if k2.fx == Chan_FX_TYPE.UP:
|
|
k2.set_state("11")
|
|
if k2.fx == Chan_FX_TYPE.DOWN:
|
|
k2.set_state("-11")
|
|
#print(index, k2.time, k2.fx)
|
|
|
|
#for klc in klc_list:
|
|
#print(klc.time, klc.fx)
|
|
#print(klc_list[len(klc_list)-2].time, klc_list[len(klc_list)-2].fx, klc_list[len(klc_list)-2].state, klc_list[-2].time)
|
|
return klc_list
|
|
def get_klc_state(self, k1, k2, k3):
|
|
self.cal_pattern(k1, k2, k3)
|
|
if k2.fx == Chan_FX_TYPE.TOP:
|
|
k2.set_state("10")
|
|
if k2.fx == Chan_FX_TYPE.BOTTOM:
|
|
k2.set_state("-10")
|
|
if k2.fx == Chan_FX_TYPE.UP:
|
|
k2.set_state("11")
|
|
if k2.fx == Chan_FX_TYPE.DOWN:
|
|
k2.set_state("-11")
|
|
def cal_pattern(self, k1, k2, k3):
|
|
if k2.high >= k1.high and k2.high >= k3.high:
|
|
k2.set_fx(Chan_FX_TYPE.TOP)
|
|
else:
|
|
if k2.low <= k1.low and k2.low <= k3.low:
|
|
k2.set_fx(Chan_FX_TYPE.BOTTOM)
|
|
#print(k1.time, k2.time, k3.time, k1.low, k2.low, k3.low, k3.open, k3.close, "k2")
|
|
else:
|
|
if k1.high >= k2.high and k2.high >= k3.high:
|
|
k2.set_fx(Chan_FX_TYPE.DOWN)
|
|
else:
|
|
if k1.high <= k2.high and k2.high <= k3.high:
|
|
k2.set_fx(Chan_FX_TYPE.UP)
|
|
if k2.fx == Chan_FX_TYPE.UNKNOWN:
|
|
if k2.close >= k1.close and k2.close >= k3.close:
|
|
k2.set_fx(Chan_FX_TYPE.TOP)
|
|
else:
|
|
if k2.close <= k1.close and k2.close <= k3.close:
|
|
k2.set_fx(Chan_FX_TYPE.BOTTOM)
|
|
#print(k2.time, "close")
|
|
else:
|
|
if k2.close >= k1.close and k2.close <= k3.close:
|
|
k2.set_fx(Chan_FX_TYPE.UP)
|
|
else:
|
|
if k2.close <= k1.close and k2.close >= k3.close:
|
|
k2.set_fx(Chan_FX_TYPE.DOWN)
|
|
|
|
# 根据结合律，合并K线
|
|
def get_klc_list(self, dataframe):
|
|
klu_list = self.get_kl_data(dataframe)
|
|
klc_list = []
|
|
last_klu = None
|
|
for klu in klu_list:
|
|
if len(klc_list) > 0:
|
|
last_klc = klc_list[-1]
|
|
included = last_klc.check_klu_included(klu)
|
|
if not included:
|
|
dir = Chan_KLINE_DIR.DOWN
|
|
if last_klc.high < klu.high:
|
|
dir = Chan_KLINE_DIR.UP
|
|
klc = ChanKLC(klu, index=len(klc_list), dir=dir)
|
|
klc_list.append(klc)
|
|
last_klc.set_next(klc)
|
|
klc.set_pre(last_klc)
|
|
last_klc.set_end_klu(last_klu)
|
|
else:
|
|
dir = Chan_KLINE_DIR.UP
|
|
if klu.open > klu.close:
|
|
dir = Chan_KLINE_DIR.DOWN
|
|
klc = ChanKLC(klu, 0, dir)
|
|
klc_list.append(klc)
|
|
last_klu = klu
|
|
return klc_list, klu_list
|
|
def copy_klu_to_klc(self, klu_list):
|
|
klc_list = []
|
|
for klu in klu_list:
|
|
if len(klc_list) > 0:
|
|
last_klc = klc_list[-1]
|
|
dir = Chan_KLINE_DIR.DOWN
|
|
if last_klc.high < klu.high:
|
|
dir = Chan_KLINE_DIR.UP
|
|
klc = ChanKLC(klu, index=len(klc_list), dir=dir)
|
|
klc.set_end_klu(klu)
|
|
klc_list.append(klc)
|
|
last_klc.set_next(klc)
|
|
klc.set_pre(last_klc)
|
|
else:
|
|
klc = ChanKLC(klu, 0)
|
|
klc_list.append(klc)
|
|
klc.set_end_klu(klu)
|
|
return klc_list
|
|
def get_kl_data(self, dataframe:DataFrame):
|
|
fields = "time,open,high,low,close,volume"
|
|
klu_list = []
|
|
for i in range(0, len(dataframe)):
|
|
item = dataframe.iloc[i]
|
|
date = item['date']
|
|
o = item['open']
|
|
h = item['high']
|
|
l = item['low']
|
|
c = item['close']
|
|
v = item['volume']
|
|
#time_obj = date.fromtimestamp(date)
|
|
time_str = date.strftime('%Y-%m-%d %H:%M:%S')
|
|
item_data = [
|
|
time_str,
|
|
o,
|
|
h,
|
|
l,
|
|
c,
|
|
v
|
|
]
|
|
#klu = KLU(self.create_item_dict(item_data, GetColumnNameFromFieldList(fields)))
|
|
klu = ChanKLU(time_str, o, h, l, c, v)
|
|
klu.set_idx(i)
|
|
klu_list.append(klu)
|
|
return klu_list
|
|
def get_ticker_indicator(self):
|
|
return int(self.timeframe[:-1]) |