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-08-08 21:03:07 +08:00
parent 8f22306766
commit 05942fed8e
4 changed files with 363 additions and 316 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ChanLun.py
+1 -197
View File
@@ -32,7 +32,6 @@ class ChanLun():
time30 = 30
time60 = 60
time4h = 240
last_peak = {'high': 0, 'low': float('inf')}
def create_all_data(self, dataframe, ticker_indicator):
all_data = dict()
all_data['1m'] = dataframe
@@ -78,15 +77,6 @@ class ChanLun():
#print(klc.start_time, klc.end_time,klc.next.start_time, klc.next.end_time,klc.fx, "BOTTOM")
return Chan_FX_TYPE.BOTTOM
return Chan_FX_TYPE.UNKNOWN
def get_macd(self, df):
fast = 8
slow = 15
period = 2
macd = ta.MACD(df, fastperiod=fast, slowperiod=slow, signalperiod=period)
df['macd'] = macd['macd']
df['macdsignal'] = macd['macdsignal']
df['macdhist'] = macd['macdhist']
return df
def plot_dataframe(self, dataframe):
klc_list = self.get_klc_list(dataframe)
bi_list= self.cal_bi_list(klc_list)
@@ -200,13 +190,6 @@ class ChanLun():
for df in df_list:
state_list.append(self.get_klc_state_list(df))
return state_list
def print_data(self, dataframe):
klc_list = self.get_klc_list(dataframe)
bi_list = self.cal_bi_list(klc_list)
seg_list = self.get_seg_list(bi_list)
bsp_list, zs_list = self.calculate_zs(bi_list, seg_list)
bi_macd_div_list = self.get_bi_macd_div_list(bi_list, dataframe)
seg_macd_div_list = self.get_seg_macd_div_list(seg_list, dataframe)
def resample_bsp_list(self, bsp_list, dataframe):
bsp_index = 0
resampled_bsp_list = []
@@ -287,10 +270,6 @@ class ChanLun():
c,
v
]
if h > self.last_peak['high']:
self.last_peak['high'] = h
if l < self.last_peak['low']:
self.last_peak['low'] = l
#klu = KLU(self.create_item_dict(item_data, GetColumnNameFromFieldList(fields)))
klu = ChanKLU(time_str, o, h, l, c, v)
#print(klu.time, klu.open, klu.high, klu.low, klu.close, klu.volume)
@@ -314,31 +293,6 @@ class ChanLun():
return df['volume_ratio']
def calculate_zs(self, bi_list, seg_list):
return self.get_zs_list(bi_list, seg_list)
def get_full_klc_list(self, dataframe):
klc_list = self.get_klc_list(dataframe)
bi_list = self.cal_bi_list(klc_list)
return klc_list
def print_bi_klc(self, dataframe):
klc_list = self.get_klc_list(dataframe)
bi_list = self.cal_bi_list(klc_list)
rsi_list = dataframe['rsi']
fx_list = []
for klc in klc_list:
if klc.klc_fx_type != Chan_KLC_FX.UNKNOWN:
if klc.bi.dir == Chan_BI_DIR.UP and klc.volume_ratio > 2:
print(klc.start_time, klc.end_time, klc.klc_fx_type, klc.rsi, klc.volume_ratio)
fx_list.append(klc)
elif klc.bi.dir == Chan_BI_DIR.DOWN and klc.volume_ratio > 2:
print(klc.start_time, klc.end_time, klc.klc_fx_type, klc.rsi, klc.volume_ratio)
fx_list.append(klc)
bi_start_index_list = []
for bi in bi_list:
bi_start_index_list.append(bi.start_klc.index)
if bi.end_klc:
bi.cal_macdhist()
bi.cal_macd_div()
print("Bi:", bi.start_time, bi.end_time, bi.dir, bi.macd_hist, bi.macd_div)
klc_index_count = 0
def get_seg_list(self, bi_list):
seg_list = []
up_bi_list = []
@@ -617,157 +571,7 @@ class ChanLun():
break
"""
return seg_list
def get_bi_zs_list(self, bi_list):
"""识别笔中枢列表
与线段中枢不同，笔中枢是由连续的同向笔构成，是更细粒度的中枢结构
Args:
bi_list: 笔列表
Returns:
bi_zs_list: 笔中枢列表
"""
bi_zs_list = []
if len(bi_list) < 3: # 至少需要3个笔才能形成中枢
print("笔数量不足，无法形成中枢")
return bi_zs_list
last_zs = None
first_bi_out = None
in_again = False
bi_out_count = 0
# 遍历所有笔，识别中枢
for i in range(2, len(bi_list)):
# 确保当前笔和前两个笔都是完成的
if not bi_list[i].end_klc or not bi_list[i-1].end_klc or not bi_list[i-2].end_klc:
continue
current_bi = bi_list[i]
prev_bi = bi_list[i-1]
prev_prev_bi = bi_list[i-2]
# 如果没有中枢或上一个中枢已完成
if len(bi_zs_list) == 0 or (last_zs and last_zs.is_sure):
# 检查是否是三个连续同向笔
if (current_bi.dir == prev_bi.dir == prev_prev_bi.dir):
# 创建潜在中枢
if current_bi.dir == Chan_BI_DIR.UP:
# 向上的三笔区间定义中枢
# 中枢的上沿：取三个笔的终点的最小值
# 中枢的下沿：取三个笔的起点的最大值
zd = max(prev_prev_bi.start_klc.low, prev_bi.start_klc.low, current_bi.start_klc.low)
zg = min(prev_prev_bi.end_klc.high, prev_bi.end_klc.high, current_bi.end_klc.high)
# 确保中枢有效(上沿大于下沿)
if zg > zd:
print(f"发现向上笔中枢: 起始时间={prev_prev_bi.start_klc.start_time}, ZG={zg}, ZD={zd}")
zs = ChanZS(prev_prev_bi.start_klc, zg, zd)
zs.start_bi = prev_prev_bi
zs.start_idx = i-2
zs.end_bi = current_bi
zs.end_idx = i
zs.end_klc = current_bi.end_klc
zs.type = "BI_ZS"
zs.direction = Chan_ZS_DIR.UP
zs.sure_time = None # 中枢尚未确认完成
bi_zs_list.append(zs)
last_zs = zs
else:
# 向下的三笔区间定义中枢
# 中枢的上沿：取三个笔的起点的最小值
# 中枢的下沿：取三个笔的终点的最大值
zg = min(prev_prev_bi.start_klc.high, prev_bi.start_klc.high, current_bi.start_klc.high)
zd = max(prev_prev_bi.end_klc.low, prev_bi.end_klc.low, current_bi.end_klc.low)
# 确保中枢有效(上沿大于下沿)
if zg > zd:
print(f"发现向下笔中枢: 起始时间={prev_prev_bi.start_klc.start_time}, ZG={zg}, ZD={zd}")
zs = ChanZS(prev_prev_bi.start_klc, zg, zd)
zs.start_bi = prev_prev_bi
zs.start_idx = i-2
zs.end_bi = current_bi
zs.end_idx = i
zs.end_klc = current_bi.end_klc
zs.type = "BI_ZS"
zs.direction = Chan_ZS_DIR.DOWN
zs.sure_time = None # 中枢尚未确认完成
bi_zs_list.append(zs)
last_zs = zs
# 处理已有的未完成中枢
elif last_zs and not last_zs.is_sure:
# 当前笔与中枢最后一笔方向相同，可能延伸中枢
if current_bi.dir == prev_bi.dir:
if last_zs.direction == Chan_ZS_DIR.UP and current_bi.dir == Chan_BI_DIR.UP:
# 检查是否仍在中枢内：向上时终点高价在中枢区间内
if current_bi.end_klc.high >= last_zs.zd and current_bi.end_klc.high <= last_zs.zg:
print(f"延伸向上笔中枢: 终点时间={current_bi.end_klc.end_time}")
# 延伸中枢
last_zs.end_klc = current_bi.end_klc
last_zs.end_bi = current_bi
last_zs.end_idx = i
else:
# 笔离开中枢，记录第一个离开的笔
if not first_bi_out:
first_bi_out = current_bi
bi_out_count += 1
print(f"笔离开向上中枢: 时间={current_bi.end_klc.end_time}, 价格={current_bi.end_klc.high}, 中枢上沿={last_zs.zg}")
else:
if not in_again:
# 第二次离开，确认中枢完成
print(f"确认向上笔中枢完成: 时间={current_bi.end_klc.end_time}")
last_zs.is_sure = True
last_zs.sure_bi = current_bi
last_zs.sure_time = current_bi.end_klc.end_time
elif last_zs.direction == Chan_ZS_DIR.DOWN and current_bi.dir == Chan_BI_DIR.DOWN:
# 检查是否仍在中枢内：向下时终点低价在中枢区间内
if current_bi.end_klc.low <= last_zs.zg and current_bi.end_klc.low >= last_zs.zd:
print(f"延伸向下笔中枢: 终点时间={current_bi.end_klc.end_time}")
# 延伸中枢
last_zs.end_klc = current_bi.end_klc
last_zs.end_bi = current_bi
last_zs.end_idx = i
else:
# 笔离开中枢，记录第一个离开的笔
if not first_bi_out:
first_bi_out = current_bi
bi_out_count += 1
print(f"笔离开向下中枢: 时间={current_bi.end_klc.end_time}, 价格={current_bi.end_klc.low}, 中枢下沿={last_zs.zd}")
else:
if not in_again:
# 第二次离开，确认中枢完成
print(f"确认向下笔中枢完成: 时间={current_bi.end_klc.end_time}")
last_zs.is_sure = True
last_zs.sure_bi = current_bi
last_zs.sure_time = current_bi.end_klc.end_time
# 方向改变，判断是否破坏中枢
else:
# 方向改变可能导致重新进入中枢或破坏中枢
# 向上中枢被向下笔破坏：低点低于中枢下沿
# 向下中枢被向上笔破坏：高点高于中枢上沿
if (last_zs.direction == Chan_ZS_DIR.UP and current_bi.end_klc.low < last_zs.zd) or \
(last_zs.direction == Chan_ZS_DIR.DOWN and current_bi.end_klc.high > last_zs.zg):
# 破坏中枢
print(f"笔中枢被破坏: 方向={current_bi.dir}, 时间={current_bi.end_klc.end_time}")
last_zs.is_sure = True
last_zs.sure_bi = current_bi
last_zs.sure_time = current_bi.end_klc.end_time
elif first_bi_out:
# 重新进入中枢
print(f"笔重新进入中枢: 时间={current_bi.end_klc.end_time}")
in_again = True
first_bi_out = None
# 延伸中枢
last_zs.end_klc = current_bi.end_klc
last_zs.end_bi = current_bi
last_zs.end_idx = i
# 打印识别结果
print(f"笔中枢识别完成，共找到 {len(bi_zs_list)} 个笔中枢")
return bi_zs_list
#-------------------------------------------------------------------
def cal_bi_list(self, klc_list):
bi_list = []
last_top = None