jack/Chan
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add strength list to check the bsp

Browse Source
This commit is contained in:
Porter
2025-05-25 12:15:16 +08:00
parent cdc40f795c
commit fb2104edf3
6 changed files with 62 additions and 132 deletions
Download Patch File Download Diff File Expand all files Collapse all files
strategies/ChanLun_SOL_5.py
+35 -130
View File
@@ -22,7 +22,7 @@ logger = logging.getLogger(__name__)
# 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 SOL/USDT:USDT --timerange=20250405-
# 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-
@@ -35,9 +35,9 @@ class ChanLun_SOL_5(IStrategy):
# This attribute will be overridden if the config file contains "minimal_roi"
# 30m and 1h
minimal_roi = {
"0": 0.10,
"360": 0.05,
"640": 0.025,
"0": 0.30,
"360": 0.2,
"640": 0.1,
"1200": 0
}
# 5m and 15m
@@ -61,8 +61,8 @@ class ChanLun_SOL_5(IStrategy):
"3600": 0
}
can_short = True
lev = 1.0
stoploss = -0.3 * lev
lev = 20.0
stoploss = -0.3
trailing_stop = False
trailing_stop_positive = 0.025
trailing_stop_positive_offset = 0.045
@@ -78,7 +78,7 @@ class ChanLun_SOL_5(IStrategy):
time30 = 30
time60 = 60
time4h = 240
time5 = 60
time5 = 15
last_time = datetime.now()
big_size = 0
big_state = "00"
@@ -110,92 +110,24 @@ class ChanLun_SOL_5(IStrategy):
dataframe_60 = self.add_indicators(dataframe_60)
dataframe_4h = self.add_indicators(dataframe_4h)
dataframe_1d = self.add_indicators(dataframe_1d)
dataframe_60['state'] = self.chan.get_klc_state_list(dataframe_60)
#dataframe_5['state'] = self.chan.plot_dataframe(dataframe_5)
#self.chan.print_data(dataframe_5)
#self.chan.cal_qjt(dataframe, dataframe_5)
#self.classifier.train_model(dataframe_4h, model_name="4h_model")
"""
self.classifier.train_model(dataframe_5, model_name="5m_model")
self.classifier.train_model(dataframe_30, model_name="30m_model")
self.classifier.train_model(dataframe_60, model_name="1h_model")
self.classifier.train_model(dataframe_4h, model_name="4h_model")
self.classifier.train_model(dataframe, model_name="1m_model")
self.classifier.train_model(dataframe_1d, model_name="1d_model")
"""
model_name = "1m_model"
df = dataframe
#self.classifier.find_best_params(df, model_name=model_name)
if self.classifier.model is None and False:
#self.classifier.train_model(df, model_name=model_name, data_file_path=model_name + '_feature_data.csv')
self.classifier.load_model(model_name=model_name)
klc_list = self.chan.get_klc_list(df)
bi_list = self.chan.cal_bi_list(klc_list)
top_avg = 0
bottom_avg = 0
top_count = 0
bottom_count = 0
for index in range(int(len(klc_list) * 0.8), len(klc_list)):
klc = klc_list[index]
features = klc.get_feature_data()
if self.classifier.predict(klc) > 0.5 and (klc.klc_fx_type == Chan_KLC_FX.BOTTOM1 or klc.klc_fx_type == Chan_KLC_FX.BOTTOM2) and features['klc_rsi'] < 40:
print(klc.end_time, klc.fx, self.classifier.predict(klc), features['klc_volume_ratio'], features['klc_macdhist'], features['klc_rsi'])
bottom_avg += self.classifier.predict(klc)
bottom_count += 1
if self.classifier.predict(klc) > 0.5 and (klc.klc_fx_type == Chan_KLC_FX.TOP1 or klc.klc_fx_type == Chan_KLC_FX.TOP2) and features['klc_rsi'] > 50:
print(klc.end_time, klc.fx, self.classifier.predict(klc), features['klc_volume_ratio'], features['klc_macdhist'], features['klc_rsi'])
top_avg += self.classifier.predict(klc)
top_count += 1
if bottom_count > 0:
bottom_avg /= bottom_count
if top_count > 0:
top_avg /= top_count
print('Bottom:', bottom_avg, 'Top:', top_avg)
print("-------------------------------------------------------------------------------")
"""
self.print_xgb(dataframe, "1m_model")
self.print_xgb(dataframe_5, "5m_model")
self.print_xgb(dataframe_30, "30m_model")
self.print_xgb(dataframe_60, "1h_model")
self.print_xgb(dataframe_4h, "4h_model")
print("-------------------------------------------------------------------------------")
"""
#classifier.train_model(use_cv=False)
#classifier.validate_model(dataframe)
#self.chan.get_bsp_list(dataframe)
#dataframe_5['state'] = self.chan.cal_klu_state(dataframe_5)
#dataframe_15['state'] = self.chan.cal_klu_state(dataframe_15)
#dataframe_30['state'] = self.chan.cal_klu_state(dataframe_30)
#dataframe_60['state'] = self.chan.cal_klu_state(dataframe_60)
#dataframe_4h['state'] = self.chan.resample_klc_list(dataframe_4h)
#self.print_bi_klc_fx(dataframe_60, "60m_model")
#self.chan.plot_dual(dataframe_5, dataframe_30)
dataframe['rsi'] = ta.RSI(dataframe, timeperiod=14)
#self.print_macd_div_list(dataframe)
#self.print_resample_df(dataframe, 1, 50)
#self.chan.get_bi_list(dataframe_30)
if self.last_time + timedelta(minutes=1) < datetime.now() and False:
#print(informative.iloc[-1])
#self.print_klc(dataframe, "1m: ")
#self.print_klc(dataframe_5, "5m: ")
#self.print_klc(dataframe_30, "30m:")
#self.log_macd_div_list(dataframe)
self.print_xgb(dataframe, "1m_model")
self.print_xgb(dataframe_5, "5m_model")
self.print_xgb(dataframe_30, "30m_model")
self.print_xgb(dataframe_60, "60m_model")
self.print_xgb(dataframe_4h, "4h_model")
#self.print_xgb(dataframe_1d, "1d_model")
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_15)
#dataframe = resampled_merge(dataframe, dataframe_30)
dataframe = resampled_merge(dataframe, dataframe_60)
#dataframe = resampled_merge(dataframe, dataframe_60)
#dataframe = resampled_merge(dataframe, dataframe_4h)
return dataframe
def print_bi_klc_fx(self, dataframe, model_name):
@@ -284,7 +216,7 @@ class ChanLun_SOL_5(IStrategy):
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_amount(self, pair: str, current_time: datetime, current_rate: float,
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:
@@ -292,7 +224,7 @@ class ChanLun_SOL_5(IStrategy):
# 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_position(self, trade: Trade, current_time: datetime,
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,
@@ -364,16 +296,16 @@ class ChanLun_SOL_5(IStrategy):
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*6) < dataframe_date:
if dataframe.iloc[-self.time5]['resample_{}_state'.format(self.get_ticker_indicator()*self.time5)] == "-10" and last_entry.side == "buy":
#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"
if last_entry.order_filled_utc + timedelta(minutes=self.time5*6) < dataframe_date:
if dataframe.iloc[-self.time5]['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 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)
@@ -472,58 +404,31 @@ class ChanLun_SOL_5(IStrategy):
# (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)
close_str = 'resample_{}_close'.format(self.get_ticker_indicator()*self.time5)
volume_str = 'resample_{}_volume'.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) == "-10") &
(dataframe[close_str].pct_change().abs() < 0.05) &
(dataframe[close_str] > dataframe[close_str].shift(self.time5)) &
(dataframe[volume_str] > dataframe[volume_str].rolling(window=20).mean())
(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) == "10") &
(dataframe[close_str].pct_change().abs() < 0.05) &
(dataframe[close_str] < dataframe[close_str].shift(self.time5)) &
(dataframe[volume_str] > dataframe[volume_str].rolling(window=20).mean())
#(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)
close_str = 'resample_{}_close'.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) == "10") |
(dataframe[close_str] < dataframe[close_str].rolling(window=20).min()) |
(dataframe[close_str] > dataframe[close_str].rolling(window=20).max()*1.05)
(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) == "-10") |
(dataframe[close_str] > dataframe[close_str].rolling(window=20).max()) |
(dataframe[close_str] < dataframe[close_str].rolling(window=20).min()*0.95)
#(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,