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

添加第二类卖卖点

Browse Source
This commit is contained in:
jackyu66git
2025-06-27 02:39:47 +08:00
parent 1c168c27bd
commit 770655c53f
3 changed files with 73 additions and 56 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ChanKLC.py
+6
View File
@@ -40,6 +40,12 @@ class ChanKLC():
self.candle_dir = klu.candle_dir
self.range = klu.range
self.strength = klu.strength
self.last_top_klc = None
self.last_bottom_klc = None
def set_last_top_klu(self, last_top_klc):
self.last_top_klc = last_top_klc
def set_last_bottom_klc(self, last_bottom_klc):
self.last_bottom_klc = last_bottom_klc
def set_klc_fx_type(self, klc_fx_type):
#print(self.start_time, klc_fx_type, self.get_feature_data()['klu_macd'], self.get_feature_data()['klu_macdhist'], self.get_feature_data()['klu_rsi'])
self.klc_fx_type = klc_fx_type
ChanLun.py
+13 -28
View File
@@ -92,20 +92,11 @@ class ChanLun():
klc_list = self.get_klc_list(dataframe)
bi_list= self.cal_bi_list(klc_list)
def get_klc_state_list(self, dataframe):
klu_list = self.get_klu_list(dataframe)
klc_list = self.get_klc_list(dataframe)
bi_list= self.cal_bi_list(klc_list)
state_list = []
if len(klc_list) > 0:
klc_index = 0
# 添加趋势强度判断
dataframe['trend_strength'] = abs(dataframe['close'].pct_change(20))
# 添加波动率判断
dataframe['volatility'] = dataframe['close'].pct_change().rolling(window=20).std()
# 添加成交量趋势
dataframe['volume_trend'] = dataframe['volume'].rolling(window=20).mean()
for index in range(0, len(dataframe)):
if klc_index == len(klc_list):
klc_index = len(klc_list) - 1
@@ -113,27 +104,18 @@ class ChanLun():
if klc.end_klu:
if klc.end_klu.idx == index:
klc_index += 1
features = klc.get_feature_data()
if (klc.klc_fx_type == Chan_KLC_FX.TOP1 or klc.klc_fx_type == Chan_KLC_FX.TOP2) and klc.bi.dir == Chan_BI_DIR.UP:
if klc.klc_fx_type == Chan_KLC_FX.TOP3:
state_list.append("10")
#print(klc.start_time, klc.klc_fx_type)
elif (klc.klc_fx_type == Chan_KLC_FX.BOTTOM1 or klc.klc_fx_type == Chan_KLC_FX.BOTTOM2) and klc.bi.dir == Chan_BI_DIR.DOWN:
print(klc.start_time, klc.end_time, klc.klc_fx_type)
elif klc.klc_fx_type == Chan_KLC_FX.BOTTOM3:
state_list.append("-10")
#print(klc.end_time, klc.klc_fx_type, klc.bi.start_time, klc.bi.dir)
print(klc.start_time, klc.end_time, klc.klc_fx_type)
else:
state_list.append("00")
else:
state_list.append("00")
else:
state_list.append("00")
klu = klu_list[index]
if klu.volume_ratio > 4.0:
if klu.close < klu.open:
state_list[-1] = "99"
print(klu.time, klu.volume_ratio, "99")
else:
state_list[-1] = "-99"
print(klu.time, klu.volume_ratio, "-99")
else:
for index in range(0, len(dataframe)):
state_list.append("00")
@@ -405,7 +387,6 @@ class ChanLun():
up_sbi_list = []
last_up_sbi = ChanSBI(last_up_bi, len(up_sbi_list), last_up_bi.dir)
up_sbi_list.append(last_up_sbi)
last_seg.add_bi(bi)
#up_sbi_list.append(last_up_sbi)
#print(last_up_bi.start_time, last_up_sbi.start_bi.start_time, "Reset up sbi list 1")
#print(bi.start_time, look_for_top, "UP 2")
@@ -434,7 +415,6 @@ class ChanLun():
up_sbi_list = []
last_up_sbi = ChanSBI(last_up_bi, len(up_sbi_list), last_up_bi.dir)
up_sbi_list.append(last_up_sbi)
last_seg.add_bi(bi)
#up_sbi_list.append(last_up_sbi)
#print(last_up_bi.start_time, last_up_sbi.start_bi.start_time, "Reset up sbi list 3")
last_down_sbi = down_sbi
@@ -449,8 +429,9 @@ class ChanLun():
down_sbi.set_pre(last_down_sbi)
down_sbi_list.append(down_sbi)
last_down_sbi = down_sbi
last_seg.add_bi(bi)
#print(bi.start_time, look_for_top, "UP 4")
last_seg.add_bi(bi)
else:
last_down_sbi = ChanSBI(bi, len(down_sbi_list), bi.dir)
down_sbi_list.append(last_down_sbi)
@@ -466,8 +447,9 @@ class ChanLun():
up_sbi.set_pre(last_up_sbi)
up_sbi_list.append(up_sbi)
last_up_sbi = up_sbi
last_seg.add_bi(bi)
#print(bi.start_time, look_for_top, "UP 6")
last_seg.add_bi(bi)
# Last seg is down
else:
if bi.dir == Chan_BI_DIR.UP:
@@ -499,7 +481,6 @@ class ChanLun():
down_sbi_list = []
last_down_sbi = ChanSBI(last_down_bi, len(down_sbi_list), last_down_bi.dir)
down_sbi_list.append(last_down_sbi)
last_seg.add_bi(bi)
#down_sbi_list.append(last_down_sbi)
#print(last_down_bi.start_time, last_down_sbi.start_bi.start_time, "Reset down sbi list 1")
#print(bi.start_time, look_for_top, "DOWN 2")
@@ -528,7 +509,6 @@ class ChanLun():
down_sbi_list = []
last_down_sbi = ChanSBI(last_down_bi, len(down_sbi_list), last_down_bi.dir)
down_sbi_list.append(last_down_sbi)
last_seg.add_bi(bi)
#down_sbi_list.append(last_down_sbi)
#print(last_down_bi.start_time, last_down_sbi.start_bi.start_time, "Reset down sbi list 3")
last_up_sbi = up_sbi
@@ -586,6 +566,7 @@ class ChanLun():
else:
last_down_bi = bi
down_bi_list.append(bi)
"""
if len(seg_list) > 1:
seg = seg_list[-1]
last_seg = seg_list[-2]
@@ -635,6 +616,7 @@ class ChanLun():
last_seg = seg
last_seg_bi = bi_list[i]
break
"""
return seg_list
def get_bi_zs_list(self, bi_list):
@@ -837,6 +819,7 @@ class ChanLun():
bi_list[-1].add_klc(klc)
klc.set_bi(bi_list[-1])
klc.set_klc_fx_type(Chan_KLC_FX.TOP3)
klc.set_last_top_klu(last_top)
#print(klc.start_time, klc.fx, "二类卖点Sell 1")
else:
# A new top found
@@ -957,6 +940,8 @@ class ChanLun():
bi_list[-1].add_klc(klc)
klc.set_bi(bi_list[-1])
klc.set_klc_fx_type(Chan_KLC_FX.BOTTOM3)
klc.set_last_bottom_klc(last_bottom)
#print(last_bottom.start_time, last_bottom.end_time, "--------------------------------1")
#print(klc.start_time, klc.fx, "二类买点Buy 1")
else:
# A new bottom found
strategies/ChanLun_BTC_30.py
+54 -28
View File
@@ -1,6 +1,6 @@
# --- Do not remove these libs ---
from statistics import median
from freqtrade.strategy import IStrategy
from freqtrade.strategy import IStrategy, stoploss_from_absolute
import sys
import os
# 添加父目录到系统路径
@@ -61,9 +61,11 @@ class ChanLun_BTC_30(IStrategy):
"2400": 0.025,
"3600": 0
}
can_short = False
can_short = True
lev = 2.0
stoploss = -0.5
use_custom_stoploss = True
trailing_stop = False
trailing_stop_positive = 0.025
trailing_stop_positive_offset = 0.045
@@ -77,10 +79,12 @@ class ChanLun_BTC_30(IStrategy):
time30 = 30
time60 = 60
time4h = 240
time30 = 60
last_time = datetime.now()
chan = ChanLun()
chanpy = ChanPY()
classifier = ChanLunClassifier(None)
last_order = None
last_trade = None
def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
@@ -107,9 +111,9 @@ class ChanLun_BTC_30(IStrategy):
#self.chan.plot_dual(dataframe_5, dataframe_30)
chanpy_state = self.chanpy.get_bsp_state(dataframe_5)
dataframe_5['chanpy_state'] = chanpy_state
state_list, fx_list = self.chan.get_klc_strength_list(dataframe_30)
dataframe_30['state'] = state_list
dataframe_30['fx'] = fx_list
state_list = self.chan.get_klc_state_list(dataframe_60)
dataframe_60['state'] = state_list
dataframe_60['fx'] = state_list
#bi_list_1 = self.chan.get_bi_list(dataframe)
#bi_list_5 = self.chan.get_bi_list(dataframe_5)
@@ -126,7 +130,7 @@ class ChanLun_BTC_30(IStrategy):
print("-------------------------------------------------------------------------------")
self.last_time = datetime.now()
dataframe = resampled_merge(dataframe, dataframe_5)
dataframe = resampled_merge(dataframe, dataframe_30)
dataframe = resampled_merge(dataframe, dataframe_60)
#dataframe = resampled_merge(dataframe, dataframe_30)
#dataframe = resampled_merge(dataframe, dataframe_60)
#dataframe = resampled_merge(dataframe, dataframe_4h)
@@ -188,7 +192,7 @@ class ChanLun_BTC_30(IStrategy):
new_exitprice = proposed_rate - 50
return new_exitprice
def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float,
def confirm_trade_entry1(self, pair: str, order_type: str, amount: float, rate: float,
time_in_force: str, current_time: datetime, entry_tag: str | None,
side: str, **kwargs) -> bool:
if self.last_trade:
@@ -207,7 +211,20 @@ class ChanLun_BTC_30(IStrategy):
#if self.last_trade:
#print(self.last_trade.open_date, current_time, self.last_trade.open_date + timedelta(minutes=self.time5))
return True
def custom_stoploss1(self, pair: str, trade: Trade, current_time: datetime,
current_rate: float, current_profit: float, after_fill: bool,
**kwargs) -> float | None:
last_high = trade.get_custom_data(key="entry_candle_high")
last_low = trade.get_custom_data(key="entry_candle_low")
# Convert absolute price to percentage relative to current_rate
if last_high:
return stoploss_from_absolute(last_high, current_rate, is_short=trade.is_short)
if last_low:
return stoploss_from_absolute(last_low, current_rate, is_short=trade.is_short)
# return maximum stoploss value, keeping current stoploss price unchanged
return None
def custom_exit1(self, pair: str, trade: Trade, current_time: datetime, current_rate: float,
current_profit: float, **kwargs):
#dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe)
@@ -229,15 +246,15 @@ class ChanLun_BTC_30(IStrategy):
"""
if trade.is_short:
last_high = trade.get_custom_data(key="entry_candle_high")
if current_rate > last_high:
if last_high and current_rate > last_high:
#print(trade.open_date, last_high, current_rate, "Relay Top FX exit")
return "Relay Top FX exit"
else:
last_low = trade.get_custom_data(key="entry_candle_low")
if current_rate < last_low:
if last_low and current_rate < last_low:
#print(trade.open_date, last_low, current_rate, "Relay Bottom FX exit")
return "Relay Bottom FX exit"
def order_filled1(self, pair: str, trade: Trade, order: Order, current_time: datetime, **kwargs) -> None:
def order_filled(self, pair: str, trade: Trade, order: Order, current_time: datetime, **kwargs) -> None:
"""
Called right after an order fills.
Will be called for all order types (entry, exit, stoploss, position adjustment).
@@ -250,17 +267,26 @@ class ChanLun_BTC_30(IStrategy):
# Obtain pair dataframe (just to show how to access it)
dataframe, _ = self.dp.get_analyzed_dataframe(trade.pair, self.timeframe)
#last_candle = dataframe.iloc[-1].squeeze()
klc_list = self.chan.get_klc_list(resample_to_interval(dataframe, self.get_ticker_indicator() * 30))
klc_list = self.chan.get_klc_list(resample_to_interval(dataframe, self.get_ticker_indicator() * self.time30))
bi_list = self.chan.cal_bi_list(klc_list)
last_high = klc_list[-2].high
last_low = klc_list[-2].low
if trade.is_short:
if (trade.nr_of_successful_entries == 1) and (order.ft_order_side == trade.entry_side):
trade.set_custom_data(key="entry_candle_high", value=last_high)
if self.last_order is None:
if trade.is_short and klc_list[-2].last_top_klc:
if (trade.nr_of_successful_entries == 1) and (order.ft_order_side == trade.entry_side):
last_high = klc_list[-2].last_top_klc.high
print(klc_list[-2].start_time, "--------------------------------", order.order_date, order.side, last_high)
trade.set_custom_data(key="entry_candle_high", value=last_high)
else:
if (trade.nr_of_successful_entries == 1) and (order.ft_order_side == trade.entry_side) and klc_list[-2].last_bottom_klc:
last_low = klc_list[-2].last_bottom_klc.low
trade.set_custom_data(key="entry_candle_low", value=last_low)
print(klc_list[-2].start_time, "--------------------------------", order.order_date, order.side, last_low)
#print(trade.open_date, trade.close_date, last_high, last_low, order.ft_order_side, klc_list[-2].start_time, klc_list[-2].end_time)
self.last_order = order
else:
if (trade.nr_of_successful_entries == 1) and (order.ft_order_side == trade.entry_side):
trade.set_custom_data(key="entry_candle_low", value=last_low)
#print(trade.open_date, trade.close_date, last_high, last_low, order.ft_order_side, klc_list[-2].start_time, klc_list[-2].end_time)
if self.last_order.side != order.side:
self.last_order = None
trade.set_custom_data(key="entry_candle_high", value=None)
trade.set_custom_data(key="entry_candle_low", value=None)
self.last_trade = trade
return None
def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame:
@@ -272,8 +298,8 @@ class ChanLun_BTC_30(IStrategy):
dataframe.loc[
(
#(dataframe['state'] == "-30")
(dataframe[state_str].shift(shift_time) > strength) &
(dataframe[fx_str].shift(shift_time) == -1)
(dataframe[state_str].shift(shift_time) == "-10")
#(dataframe[fx_str].shift(shift_time) == -1)
#(dataframe[chanpy_state_str].shift(shift_time+30) == 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") &
@@ -284,8 +310,8 @@ class ChanLun_BTC_30(IStrategy):
dataframe.loc[
(
#(dataframe['state'] == "-30")
(dataframe[state_str].shift(shift_time) > strength) &
(dataframe[fx_str].shift(shift_time) == 1)
(dataframe[state_str].shift(shift_time) == "10")
#(dataframe[fx_str].shift(shift_time) == 1)
#(dataframe[chanpy_state_str].shift(shift_time+30) == -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") &
@@ -303,8 +329,8 @@ class ChanLun_BTC_30(IStrategy):
dataframe.loc[
(
#(dataframe['state']== "30")
(dataframe[state_str].shift(shift_time) > strength) &
(dataframe[fx_str].shift(shift_time) == 1)
(dataframe[state_str].shift(shift_time) == "101")
#(dataframe[fx_str].shift(shift_time) == 1)
#(dataframe[chanpy_state_str].shift(shift_time+30) == -1)
#(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time30)] == "10") &
#(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time60)] == "10")
@@ -313,8 +339,8 @@ class ChanLun_BTC_30(IStrategy):
dataframe.loc[
(
#(dataframe['state']== "30")
(dataframe[state_str].shift(shift_time) > strength) &
(dataframe[fx_str].shift(shift_time) == -1)
(dataframe[state_str].shift(shift_time) == "-101")
#(dataframe[fx_str].shift(shift_time) == -1)
#(dataframe[chanpy_state_str].shift(shift_time+30) == 1)
#(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time30)] == "10") &
#(dataframe['resample_{}_state'.format(self.get_ticker_indicator()*self.time60)] == "10")