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

Add some classifier code

Browse Source
This commit is contained in:
jackyu66git
2025-04-23 10:39:04 +08:00
parent 70e14c2ea3
commit 48d5647ebf
14 changed files with 139 additions and 47 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.DS_Store
Vendored
BIN
View File
Binary file not shown.
ChanBI.py
+18 -1
View File
@@ -18,10 +18,25 @@ class ChanBI():
self.start_time = klc.start_time
self.macd_hist = 0
self.macd_div = 0
def set_macd_hist(self, macd_hist):
def set_macdhist(self, macd_hist):
self.macd_hist = macd_hist
def set_macd_div(self, macd_div):
self.macd_div = macd_div
def cal_macd_div(self):
self.macd_div = 0.0
if self.pre and self.pre.pre:
if self.pre.pre.macd_hist == 0:
self.macd_div = 0.0
else:
self.macd_div = self.macd_hist / self.pre.pre.macd_hist
def cal_macdhist(self):
self.macd_hist = 0
for klc in self.klc_list:
for klu in klc.klus:
if self.dir == Chan_BI_DIR.UP and klu.macdhist > 0:
self.macd_hist += klu.macdhist
if self.dir == Chan_BI_DIR.DOWN and klu.macdhist < 0:
self.macd_hist -= klu.macdhist
def check_overlap(self):
if self.next and self.next.next:
if self.dir == Chan_BI_DIR.UP:
@@ -61,6 +76,8 @@ class ChanBI():
break
if not added:
self.klc_list.append(klc)
self.cal_macdhist()
self.cal_macd_div()
def append_klc_list(self, klc_list):
self.klc_list.append(klc_list)
def update_bi(self, klc):
ChanKLC.py
+15 -6
View File
@@ -26,6 +26,9 @@ class ChanKLC():
self.close = klu.close
self.volume = klu.volume
self.macdhist = 0
self.bi_macdhist = 0
self.bi_macd_div = 0.0
self.bi = None
def add_klu(self, klu):
self.klus.append(klu)
def set_end_klu(self, klu):
@@ -116,10 +119,10 @@ class ChanKLC():
if klc.index == self.index:
break
else:
if bi.dir == Chan_BI_DIR.UP:
self.macdhist += klc.get_macdhist()
else:
self.macdhist -= klc.get_macdhist()
if bi.dir == Chan_BI_DIR.UP and klc.get_macdhist() > 0:
self.bi_macdhist += klc.get_macdhist()
elif bi.dir == Chan_BI_DIR.DOWN and klc.get_macdhist() < 0:
self.bi_macdhist -= klc.get_macdhist()
def get_macdhist(self):
self.macdhist = 0
for klu in self.klus:
@@ -158,8 +161,14 @@ class ChanKLC():
features['klc_fx'] = 0 if self.fx == Chan_FX_TYPE.UNKNOWN else 1 if self.fx == Chan_FX_TYPE.TOP else 2 #7
features['klc_klus'] = len(self.klus) #8
features['klc_volume'] = self.volume #9
features['klc_pre_fx'] = (0 if self.pre.fx == Chan_FX_TYPE.UNKNOWN else 1 if self.pre.fx == Chan_FX_TYPE.TOP else 2) if self.pre else 0
features['klc_pre_fx'] = (0 if self.pre.fx == Chan_FX_TYPE.UNKNOWN else 1 if self.pre.fx == Chan_FX_TYPE.TOP else 2) if self.pre else 0 #10
features['klc_pre_pre_fx'] = (0 if self.pre.pre.fx == Chan_FX_TYPE.UNKNOWN else 1 if self.pre.pre.fx == Chan_FX_TYPE.TOP else 2) if self.pre and self.pre.pre else 0 #11
features['klc_pre_pre_pre_fx'] = (0 if self.pre.pre.pre.fx == Chan_FX_TYPE.UNKNOWN else 1 if self.pre.pre.pre.fx == Chan_FX_TYPE.TOP else 2) if self.pre and self.pre.pre and self.pre.pre.pre else 0 #12
# New Add 20250422
#features['klc_macdhist'] = self.get_macdhist() #11
#features['klc_bi_macdhist'] = self.bi_macdhist #12
#features['klc_bi_macd_div'] = self.bi_macd_div #13
#features['klc_bi_dir'] = 1 if self.bi.dir == Chan_BI_DIR.UP else -1 #14
# ===== 2.1 K线形态因子 =====
# K线实体大小
ChanLun.py
+82 -31
View File
@@ -15,6 +15,9 @@ from matplotlib.dates import DateFormatter, date2num
import matplotlib.patches as patches
from technical.util import resample_to_interval
from decimal import Decimal
import xgboost as xgb
import numpy as np
class ChanLun():
timeframes = ["5m", "15m", "30m", "60m", "4h"]
times = {
@@ -29,7 +32,6 @@ class ChanLun():
time30 = 30
time60 = 60
time4h = 240
def create_all_data(self, dataframe, ticker_indicator):
all_data = dict()
all_data['1m'] = dataframe
@@ -910,8 +912,8 @@ class ChanLun():
#self.print_zs(zs_list)
return zs_list
def get_bi_macd_hist_list(self, bi_list, dataframe):
bi_macd_hist_list = []
def get_bi_macdhist_list(self, bi_list, dataframe):
bi_macdhist_list = []
for bi in bi_list:
start_index = bi.start_klc.start_klu.index
if bi.end_klc:
@@ -925,12 +927,12 @@ class ChanLun():
total_macd_hist += macd_hist
if bi.dir == Chan_BI_DIR.DOWN and macd_hist < 0:
total_macd_hist -= macd_hist
bi_macd_hist_list.append(abs(total_macd_hist))
bi.set_macd_hist(total_macd_hist)
return bi_macd_hist_list, bi_list
bi_macdhist_list.append(abs(total_macd_hist))
bi.set_macdhist(total_macd_hist)
return bi_macdhist_list, bi_list
def get_seg_macd_hist_list(self, seg_list, dataframe):
seg_macd_hist_list = []
def get_seg_macdhist_list(self, seg_list, dataframe):
seg_macdhist_list = []
for seg in seg_list:
start_index = seg.start_bi.start_klc.start_klu.index
if seg.end_bi:
@@ -944,46 +946,46 @@ class ChanLun():
total_macd_hist += macd_hist
if seg.dir == Chan_SEG_DIR.DOWN and macd_hist < 0:
total_macd_hist -= macd_hist
seg_macd_hist_list.append(abs(total_macd_hist))
seg.set_macd_hist(total_macd_hist)
return seg_macd_hist_list, seg_list
seg_macdhist_list.append(abs(total_macd_hist))
seg.set_macdhist(total_macd_hist)
return seg_macdhist_list, seg_list
def get_bi_macd_div_list(self, bi_list, dataframe):
bi_macd_div_list = []
bi_macd_hist_list, bi_list = self.get_bi_macd_hist_list(bi_list, dataframe)
bi_macdhist_list, bi_list = self.get_bi_macdhist_list(bi_list, dataframe)
for index in range(2, len(bi_list)):
if bi_macd_hist_list[index-2] == 0:
if bi_macdhist_list[index-2] == 0:
bi_macd_div = 0.0
if index > 3 and bi_macd_hist_list[index-4] > 0.0:
bi_macd_div = bi_macd_hist_list[index]/bi_macd_hist_list[index-4]
if index > 3 and bi_macdhist_list[index-4] > 0.0:
bi_macd_div = bi_macdhist_list[index]/bi_macdhist_list[index-4]
else:
bi_macd_div = bi_macd_hist_list[index]/bi_macd_hist_list[index-2]
bi_macd_div = bi_macdhist_list[index]/bi_macdhist_list[index-2]
if bi_macd_div < 0.01:
if index > 3 and bi_macd_hist_list[index-4] > 0.0:
bi_macd_div = bi_macd_hist_list[index]/bi_macd_hist_list[index-4]
if index > 3 and bi_macdhist_list[index-4] > 0.0:
bi_macd_div = bi_macdhist_list[index]/bi_macdhist_list[index-4]
bi_macd_div = self.get_decimal(bi_macd_div)
bi_macd_div_list.append(bi_macd_div)
bi_list[index].set_macd_div(bi_macd_div)
#print(bi_list[index].start_klc.start_time, self.get_decimal(bi_macd_hist_list[index]), self.get_decimal(bi_macd_hist_list[index - 1]), self.get_decimal(bi_macd_div))
#print(bi_list[index].start_klc.start_time, self.get_decimal(bi_macdhist_list[index]), self.get_decimal(bi_macdhist_list[index - 1]), self.get_decimal(bi_macd_div))
return bi_macd_div_list, bi_list
def get_seg_macd_div_list(self, seg_list, dataframe):
seg_macd_div_list = []
seg_macd_hist_list, seg_list = self.get_seg_macd_hist_list(seg_list, dataframe)
seg_macdhist_list, seg_list = self.get_seg_macdhist_list(seg_list, dataframe)
for index in range(2, len(seg_list)):
if seg_macd_hist_list[index-2] == 0:
if seg_macdhist_list[index-2] == 0:
seg_macd_div = 0.0
if index > 3 and seg_macd_hist_list[index-4] > 0.0:
seg_macd_div = seg_macd_hist_list[index]/seg_macd_hist_list[index - 4]
if index > 3 and seg_macdhist_list[index-4] > 0.0:
seg_macd_div = seg_macdhist_list[index]/seg_macdhist_list[index - 4]
else:
seg_macd_div = seg_macd_hist_list[index]/seg_macd_hist_list[index - 2]
seg_macd_div = seg_macdhist_list[index]/seg_macdhist_list[index - 2]
if seg_macd_div < 0.01:
if index > 3 and seg_macd_hist_list[index-4] > 0.0:
seg_macd_div = seg_macd_hist_list[index]/seg_macd_hist_list[index - 4]
if index > 3 and seg_macdhist_list[index-4] > 0.0:
seg_macd_div = seg_macdhist_list[index]/seg_macdhist_list[index - 4]
seg_macd_div = self.get_decimal(seg_macd_div)
seg_macd_div_list.append(seg_macd_div)
seg_list[index].set_macd_div(seg_macd_div)
#print(seg_list[index].start_bi.start_klc.start_time, self.get_decimal(seg_macd_hist_list[index]), self.get_decimal(seg_macd_hist_list[index - 1]), self.get_decimal(seg_macd_div))
#print(seg_list[index].start_bi.start_klc.start_time, self.get_decimal(seg_macdhist_list[index]), self.get_decimal(seg_macdhist_list[index - 1]), self.get_decimal(seg_macd_div))
return seg_macd_div_list, seg_list
def get_macd_div_list(self, dataframe):
@@ -1596,7 +1598,7 @@ class ChanLun():
for index in range(0, len(big_bi_macd_div)):
bi_macd_div = big_bi_macd_div[index]
bi = big_bi[index + 2]
if bi.end_klc:
if bi.end_klc and False:
text_index = bi.end_klc.end_klu.index
if bi.dir == Chan_BI_DIR.UP:
ax1.text(big_dates_num[text_index], bi.end_klc.high+1, bi_macd_div, color='red', fontsize=10, alpha=0.6)
@@ -1605,12 +1607,24 @@ class ChanLun():
for index in range(0, len(big_seg_macd_div)):
seg_macd_div = big_seg_macd_div[index]
seg = big_seg[index + 2]
if seg.end_bi:
if seg.end_bi and False:
text_index = seg.end_bi.end_klc.end_klu.index
if seg.dir == Chan_SEG_DIR.UP:
ax1.text(big_dates_num[text_index], seg.end_bi.end_klc.high+1, seg_macd_div, color='red', fontsize=14, alpha=0.6)
else:
ax1.text(big_dates_num[text_index], seg.end_bi.end_klc.low-1, seg_macd_div, color='green', fontsize=14, alpha=0.6)
big_klc = self.get_klc_list(big_df)
model = xgb.Booster()
model.load_model("30m_modelchan_xgb_model.json")
for klc in big_klc:
predict = self.predict(klc, model)
if predict > 0.35 and klc.fx == Chan_FX_TYPE.BOTTOM:
text_index = klc.end_klu.index
ax1.text(big_dates_num[text_index], klc.high+1, predict, color='red', fontsize=14, alpha=0.6)
if predict > 0.35 and klc.fx == Chan_FX_TYPE.TOP:
text_index = klc.end_klu.index
ax1.text(big_dates_num[text_index], klc.low-1, predict, color='green', fontsize=14, alpha=0.6)
"""
# 绘制大周期买卖点
marker_styles = {
@@ -1748,7 +1762,7 @@ class ChanLun():
for index in range(0, len(small_bi_macd_div)):
bi_macd_div = small_bi_macd_div[index]
bi = small_bi[index + 2]
if bi.end_klc:
if bi.end_klc and False:
text_index = bi.end_klc.end_klu.index
if bi.dir == Chan_BI_DIR.UP:
ax2.text(small_dates_num[text_index], bi.end_klc.high+1, bi_macd_div, color='red', fontsize=10, alpha=0.6)
@@ -1757,12 +1771,23 @@ class ChanLun():
for index in range(0, len(small_seg_macd_div)):
seg_macd_div = small_seg_macd_div[index]
seg = small_seg[index + 2]
if seg.end_bi:
if seg.end_bi and False:
text_index = seg.end_bi.end_klc.end_klu.index
if seg.dir == Chan_SEG_DIR.UP:
ax2.text(small_dates_num[text_index], seg.end_bi.end_klc.high+1, seg_macd_div, color='red', fontsize=14, alpha=0.8)
else:
ax2.text(small_dates_num[text_index], seg.end_bi.end_klc.low-1, seg_macd_div, color='green', fontsize=14, alpha=0.8)
small_klc = self.get_klc_list(small_df)
model = xgb.Booster()
model.load_model("5m_modelchan_xgb_model.json")
for klc in small_klc:
predict = self.predict(klc, model)
if predict > 0.35 and klc.fx == Chan_FX_TYPE.BOTTOM:
text_index = klc.end_klu.index
ax2.text(small_dates_num[text_index], klc.high+1, predict, color='red', fontsize=14, alpha=0.6)
if predict > 0.35 and klc.fx == Chan_FX_TYPE.TOP:
text_index = klc.end_klu.index
ax2.text(small_dates_num[text_index], klc.low-1, predict, color='green', fontsize=14, alpha=0.6)
"""
# 绘制小周期买卖点
for idx, point in small_buy_sell_points.items():
@@ -1807,6 +1832,32 @@ class ChanLun():
plt.tight_layout()
plt.show()
def predict(self, klc, model):
"""
使用训练好的模型预测单个KLC
:param klc: 需要预测的ChanKLC对象
:return: 预测结果（概率值）
"""
# 提取特征
features = klc.get_feature_data()
feature_vec = []
# 与get_feature_data保持一致，只使用相同的特征集
for key, value in features.items():
if isinstance(value, (int, float)):
feature_vec.append(value)
else:
feature_vec.append(0)
# 转换为模型输入格式
dtest = xgb.DMatrix(np.array([feature_vec]))
# 预测
return self.get_decimal(model.predict(dtest)[0])
def get_decimal(self, value):
return Decimal("{:.2f}".format(value))
def plot(self, dataframe, bi_list, seg_list, zs_list=None, buy_sell_points=None, divergence_points=None):
"""
绘制缠论分析图表，包括K线、笔、线段、中枢、买卖点和MACD背驰
ChanLun_Classifier.py
+23 -8
View File
@@ -281,13 +281,18 @@ class ChanLunClassifier:
# 使用ChanLun获取bi_list
bi_list = self.chan.cal_bi_list(self.chan.get_klc_list(dataframe))
klc_list = self.chan.get_klc_list(dataframe)
seg_list = self.chan.get_seg_list(bi_list)
# 筛选方向为UP的bi的起始klc
feature_data = []
labels = []
bi_index = 0
sample_list = []
for klc in klc_list:
if bi_index == len(bi_list):
if klc.fx != Chan_FX_TYPE.UNKNOWN:
sample_list.append(klc)
for klc in sample_list:
if bi_index >= len(bi_list):
bi_index = len(bi_list) - 1
bi = bi_list[bi_index]
if klc.end_klu and bi.end_klc and klc.start_klu.index >= bi.start_klc.start_klu.index and klc.end_klu.index <= bi.end_klc.end_klu.index:
@@ -306,9 +311,11 @@ class ChanLunClassifier:
# 判断这个bi是否赚钱（这里简单定义为：如果bi的结束价格高于起始价格，则标记为1，否则为0）
# 这个标签定义可以根据实际需求修改
if bi.start_klc.index == klc.index:
seg = seg_list[bi_index]
if bi.start_klc.index == klc.index and bi.dir == Chan_BI_DIR.UP:
#if klc.index == seg.start_bi.start_klc.index and seg.dir == Chan_SEG_DIR.UP:
label = 1
bi_index += 1
bi_index += 2
else:
label = 0
@@ -325,12 +332,17 @@ class ChanLunClassifier:
# 使用ChanLun获取bi_list
bi_list = self.chan.cal_bi_list(self.chan.get_klc_list(dataframe))
klc_list = self.chan.get_klc_list(dataframe)
seg_list = self.chan.get_seg_list(bi_list)
# 筛选方向为UP的bi的起始klc
feature_data = []
labels = []
bi_index = 0
sample_list = []
for klc in klc_list:
if bi_index == len(bi_list):
if klc.fx != Chan_FX_TYPE.UNKNOWN:
sample_list.append(klc)
for klc in sample_list:
if bi_index >= len(bi_list):
bi_index = len(bi_list) - 1
bi = bi_list[bi_index]
# 提取特征
@@ -344,10 +356,11 @@ class ChanLunClassifier:
feature_vec.append(value)
else:
feature_vec.append(0)
if bi.start_klc.index == klc.index:
seg = seg_list[bi_index]
if bi.start_klc.index == klc.index and bi.dir == Chan_BI_DIR.UP:
#if klc.index == seg.start_bi.start_klc.index and seg.dir == Chan_SEG_DIR.UP:
label = 1
bi_index += 1
bi_index += 2
else:
label = 0
@@ -433,6 +446,8 @@ class ChanLunClassifier:
dtest = xgb.DMatrix(np.array([feature_vec]))
# 预测
return self.model.predict(dtest)[0]
return self.get_decimal(self.model.predict(dtest)[0])
def get_decimal(self, value):
return Decimal("{:.2f}".format(value))
ChanSEG.py
+1 -1
View File
@@ -26,7 +26,7 @@ class ChanSEG():
self.sure_time = None
self.macd_hist = 0
self.macd_div = 0
def set_macd_hist(self, macd_hist):
def set_macdhist(self, macd_hist):
self.macd_hist = macd_hist
def set_macd_div(self, macd_div):
self.macd_div = macd_div
__pycache__/ChanBI.cpython-312.pyc
BIN
View File
Binary file not shown.
__pycache__/ChanBSP.cpython-312.pyc
BIN
View File
Binary file not shown.
__pycache__/ChanKLC.cpython-312.pyc
BIN
View File
Binary file not shown.
__pycache__/ChanKLU.cpython-312.pyc
BIN
View File
Binary file not shown.
__pycache__/ChanLun.cpython-312.pyc
BIN
View File
Binary file not shown.
__pycache__/ChanLun_Classifier.cpython-312.pyc
BIN
View File
Binary file not shown.
__pycache__/ChanSEG.cpython-312.pyc
BIN
View File
Binary file not shown.
__pycache__/ChanZS.cpython-312.pyc
BIN
View File
Binary file not shown.