157 lines
5.1 KiB
Python
157 lines
5.1 KiB
Python
import json
|
||
from typing import Dict, TypedDict
|
||
import sys
|
||
import os
|
||
sys.path.append(os.path.abspath("/Users/jack/Project/chan.py"))
|
||
sys.path.append(os.path.abspath("/Users/jack/Project/chan.py/Debug"))
|
||
import xgboost as xgb
|
||
|
||
from Chan import CChan
|
||
from ChanConfig import CChanConfig
|
||
from ChanModel.Features import CFeatures
|
||
from Common.CEnum import AUTYPE, DATA_SRC, KL_TYPE, BSP_TYPE, MACD_ALGO
|
||
from Common.CTime import CTime
|
||
from Plot.PlotDriver import CPlotDriver
|
||
from Bi.Bi import CBi
|
||
from FeatureDict import FeatureDict
|
||
class T_SAMPLE_INFO(TypedDict):
|
||
feature: CFeatures
|
||
is_buy: bool
|
||
open_time: CTime
|
||
|
||
|
||
def plot(chan, plot_marker):
|
||
plot_config = {
|
||
"plot_kline": True,
|
||
"plot_bi": True,
|
||
"plot_seg": True,
|
||
"plot_zs": True,
|
||
"plot_bsp": True,
|
||
"plot_marker": True,
|
||
"plot_macd": True,
|
||
}
|
||
plot_para = {
|
||
"figure": {
|
||
"x_range": 4000,
|
||
},
|
||
"marker": {
|
||
"markers": plot_marker
|
||
}
|
||
}
|
||
plot_driver = CPlotDriver(
|
||
chan,
|
||
plot_config=plot_config,
|
||
plot_para=plot_para,
|
||
)
|
||
plot_driver.save2img("sell.png")
|
||
|
||
if __name__ == "__main__":
|
||
"""
|
||
本demo主要演示如何记录策略产出的买卖点的特征
|
||
然后将这些特征作为样本，训练一个模型(以XGB为demo)
|
||
用于预测买卖点的准确性
|
||
|
||
请注意，demo训练预测都用的是同一份数据，这是不合理的，仅仅是为了演示
|
||
"""
|
||
code = "BTC/USDT:USDT"
|
||
begin_time = "2024-12-1"
|
||
end_time = "2024-12-4"
|
||
data_src = DATA_SRC.CCXT
|
||
lv_list = [KL_TYPE.K_1M]
|
||
|
||
config = CChanConfig({
|
||
"trigger_step": True, # 打开开关！
|
||
"bi_strict": True,
|
||
"skip_step": 0,
|
||
"divergence_rate": float("inf"),
|
||
"bsp2_follow_1": False,
|
||
"bsp3_follow_1": False,
|
||
"min_zs_cnt": 0,
|
||
"bs1_peak": False,
|
||
"macd_algo": "peak",
|
||
"bs_type": '1,2,3a,1p,2s,3b',
|
||
"print_warning": True,
|
||
"zs_algo": "normal",
|
||
"cal_rsi": True,
|
||
"cal_kdj": True,
|
||
})
|
||
|
||
chan = CChan(
|
||
code=code,
|
||
begin_time=begin_time,
|
||
end_time=end_time,
|
||
data_src=data_src,
|
||
lv_list=lv_list,
|
||
config=config,
|
||
autype=AUTYPE.QFQ,
|
||
)
|
||
|
||
bsp_dict: Dict[int, T_SAMPLE_INFO] = {} # 存储策略产出的bsp的特征
|
||
|
||
# 跑策略，保存买卖点的特征
|
||
index = 0
|
||
for chan_snapshot in chan.step_load():
|
||
last_klu = chan_snapshot[0][-1][-1]
|
||
bsp_list = chan_snapshot.get_bsp()
|
||
if not bsp_list:
|
||
continue
|
||
last_bsp = bsp_list[-1]
|
||
cur_lv_chan = chan_snapshot[0]
|
||
if BSP_TYPE.T1 in last_bsp.type or BSP_TYPE.T1P in last_bsp.type:
|
||
if last_bsp.klu.idx not in bsp_dict and cur_lv_chan[-2].idx == last_bsp.klu.klc.idx and not last_bsp.is_buy:
|
||
# 假如策略是：买卖点分形第三元素出现时交易
|
||
bsp_dict[last_bsp.klu.idx] = {
|
||
"feature": last_bsp.features,
|
||
"is_buy": last_bsp.is_buy,
|
||
"open_time": last_klu.time,
|
||
}
|
||
bsp_dict[last_bsp.klu.idx]['feature'].add_feat(FeatureDict().stragety_feature(last_bsp)) # 开仓K线特征
|
||
print(index, last_bsp.klu.time, last_bsp.is_buy)
|
||
index += 1
|
||
|
||
# 生成libsvm样本特征
|
||
bsp_academy = [bsp.klu.idx for bsp in chan.get_bsp()]
|
||
feature_meta = {} # 特征meta
|
||
cur_feature_idx = 0
|
||
plot_marker = {}
|
||
fid = open("sell_feature.libsvm", "w")
|
||
for bsp_klu_idx, feature_info in bsp_dict.items():
|
||
label = int(bsp_klu_idx in bsp_academy) # 以买卖点识别是否准确为label
|
||
features = [] # List[(idx, value)]
|
||
for feature_name, value in feature_info['feature'].items():
|
||
if feature_name not in feature_meta:
|
||
feature_meta[feature_name] = cur_feature_idx
|
||
cur_feature_idx += 1
|
||
features.append((feature_meta[feature_name], value))
|
||
features.sort(key=lambda x: x[0])
|
||
feature_str = " ".join([f"{idx}:{value}" for idx, value in features])
|
||
fid.write(f"{label} {feature_str}\n")
|
||
plot_marker[feature_info["open_time"].to_str()] = ("√" if label else "×", "down" if feature_info["is_buy"] else "up")
|
||
fid.close()
|
||
|
||
with open("feature.meta", "w") as fid:
|
||
# meta保存下来，实盘预测时特征对齐用
|
||
fid.write(json.dumps(feature_meta))
|
||
|
||
# 画图检查label是否正确
|
||
plot(chan, plot_marker)
|
||
|
||
# load sample file & train model
|
||
dtrain = xgb.DMatrix("sell_feature.libsvm?format=libsvm") # load sample
|
||
param = {'max_depth': 2, 'eta': 0.3, 'objective': 'binary:logistic', 'eval_metric': 'auc'}
|
||
evals_result = {}
|
||
bst = xgb.train(
|
||
param,
|
||
dtrain=dtrain,
|
||
num_boost_round=50,
|
||
evals=[(dtrain, "train")],
|
||
evals_result=evals_result,
|
||
verbose_eval=True,
|
||
)
|
||
bst.save_model("sell_model.json")
|
||
|
||
# load model
|
||
model = xgb.Booster()
|
||
model.load_model("sell_model.json")
|
||
# predict
|
||
print(model.predict(dtrain)) |