525 lines
21 KiB
Python
525 lines
21 KiB
Python
import sys
|
|
import os
|
|
#sys.setrecursionlimit(1000000) #例如这里设置为一百万
|
|
#sys.path.append(os.path.abspath("/freqtrade/user_data/Chan"))
|
|
sys.path.append(os.path.abspath("/Users/jack/Project/freqtrade/user_data/Chan"))
|
|
import numpy as np
|
|
from datetime import timedelta
|
|
from pandas import DataFrame
|
|
from ChanEnum import Chan_FX_TYPE, Chan_KLINE_DIR, Chan_BI_DIR, Chan_SEG_DIR, Chan_ZS_DIR, Chan_BSP_DIR, Chan_BSP_TYPE, Chan_KLC_FX
|
|
from ChanKLU import ChanKLU
|
|
from ChanKLC import ChanKLC
|
|
from ChanBI import ChanBI
|
|
from ChanSBI import ChanSBI
|
|
from ChanSEG import ChanSEG
|
|
from ChanZS import ChanZS
|
|
from ChanBSP import ChanBSP
|
|
import talib.abstract as ta
|
|
import pandas as pd
|
|
import matplotlib.pyplot as plt
|
|
from matplotlib.dates import DateFormatter, date2num
|
|
import matplotlib.patches as patches
|
|
from technical.util import resample_to_interval
|
|
from decimal import Decimal
|
|
from ChanLun import ChanLun
|
|
import xgboost as xgb
|
|
from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score, classification_report
|
|
|
|
class ChanLunClassifier:
|
|
def __init__(self, dataframe: DataFrame):
|
|
self.dataframe = dataframe
|
|
self.model = None
|
|
chan = ChanLun()
|
|
|
|
def train_model(self, dataframe=None, data_file_path=None, model_file_path='chan_xgb_model.json', use_cv=False, custom_params=None, model_name=None):
|
|
"""
|
|
使用dataframe前80%的数据训练XGBoost模型
|
|
:param dataframe: 输入的DataFrame，如果为None则使用初始化时的dataframe
|
|
:param data_file_path: 特征数据保存路径，可选
|
|
:param model_file_path: 模型保存路径
|
|
:param use_cv: 是否使用交叉验证寻找最佳参数
|
|
:param custom_params: 自定义模型参数
|
|
:return: 训练好的模型
|
|
"""
|
|
if dataframe is None:
|
|
dataframe = self.dataframe
|
|
|
|
# 分割数据集，前80%用于训练
|
|
train_size = int(len(dataframe) * 0.8)
|
|
train_df = dataframe.iloc[:train_size].copy()
|
|
|
|
# 获取训练集特征和标签
|
|
save_csv = True if data_file_path else False
|
|
X_train, y_train = self.get_feature_data(train_df, save_csv=save_csv, csv_path=data_file_path if data_file_path else 'feature_data.csv')
|
|
|
|
if len(X_train) == 0:
|
|
print("没有提取到足够的特征数据进行训练")
|
|
return None
|
|
|
|
# 保存特征数据的步骤已经移到get_feature_data方法中处理
|
|
# 以下是原有代码
|
|
#{'eta': 0.03, 'max_depth': 4, 'subsample': 0.8, 'colsample_bytree': 0.8, 'gamma': 0.1, 'min_child_weight': 3, 'alpha': 1, 'lambda': 3},
|
|
# 默认XGBoost参数
|
|
default_params = {
|
|
'objective': 'binary:logistic',
|
|
'max_depth': 4,
|
|
'eta': 0.03,
|
|
'subsample': 0.8,
|
|
'colsample_bytree': 0.8,
|
|
'eval_metric': 'auc',
|
|
'gamma': 0.1,
|
|
'min_child_weight': 3,
|
|
'alpha': 1, # L1正则化
|
|
'lambda': 3, # L2正则化
|
|
'scale_pos_weight': 1
|
|
}
|
|
|
|
# 使用自定义参数覆盖默认参数
|
|
if custom_params:
|
|
for key, value in custom_params.items():
|
|
default_params[key] = value
|
|
|
|
params = default_params
|
|
dtrain = xgb.DMatrix(X_train, label=y_train)
|
|
|
|
# 如果使用交叉验证寻找最佳参数
|
|
if use_cv:
|
|
from sklearn.model_selection import GridSearchCV, RandomizedSearchCV
|
|
from sklearn.metrics import make_scorer, accuracy_score, f1_score
|
|
import numpy as np
|
|
|
|
# 转换为sklearn兼容格式
|
|
xgb_model = xgb.XGBClassifier(
|
|
objective=params['objective'],
|
|
max_depth=params['max_depth'],
|
|
learning_rate=params['eta'],
|
|
subsample=params['subsample'],
|
|
colsample_bytree=params['colsample_bytree'],
|
|
gamma=params['gamma'],
|
|
min_child_weight=params['min_child_weight'],
|
|
reg_alpha=params['alpha'],
|
|
reg_lambda=params['lambda'],
|
|
scale_pos_weight=params['scale_pos_weight'],
|
|
use_label_encoder=False,
|
|
eval_metric='auc'
|
|
)
|
|
|
|
# 参数网格
|
|
param_grid = {
|
|
'max_depth': [3, 5, 7, 9],
|
|
'learning_rate': [0.01, 0.05, 0.1, 0.2],
|
|
'subsample': [0.6, 0.8, 1.0],
|
|
'colsample_bytree': [0.6, 0.8, 1.0],
|
|
'min_child_weight': [1, 3, 5],
|
|
'gamma': [0, 0.1, 0.2],
|
|
'n_estimators': [50, 100, 200]
|
|
}
|
|
|
|
# 使用随机搜索寻找最佳参数（比网格搜索快）
|
|
random_search = RandomizedSearchCV(
|
|
estimator=xgb_model,
|
|
param_distributions=param_grid,
|
|
n_iter=10, # 随机尝试的参数组合数
|
|
scoring=make_scorer(f1_score),
|
|
cv=5,
|
|
verbose=1,
|
|
n_jobs=-1,
|
|
random_state=42
|
|
)
|
|
|
|
print("进行交叉验证参数搜索...")
|
|
random_search.fit(X_train, y_train)
|
|
|
|
# 获取最佳参数
|
|
best_params = random_search.best_params_
|
|
print(f"最佳参数: {best_params}")
|
|
|
|
# 使用最佳参数更新模型参数
|
|
params['max_depth'] = best_params['max_depth']
|
|
params['eta'] = best_params['learning_rate']
|
|
params['subsample'] = best_params['subsample']
|
|
params['colsample_bytree'] = best_params['colsample_bytree']
|
|
params['min_child_weight'] = best_params['min_child_weight']
|
|
params['gamma'] = best_params['gamma']
|
|
num_round = best_params['n_estimators']
|
|
|
|
# 使用最佳参数训练最终模型
|
|
self.model = xgb.train(params, dtrain, num_round)
|
|
else:
|
|
# 标准训练（不使用交叉验证）
|
|
# 使用早停机制避免过拟合
|
|
# 分割训练集为训练和验证
|
|
eval_size = int(len(X_train) * 0.2)
|
|
X_eval = X_train[-eval_size:]
|
|
y_eval = y_train[-eval_size:]
|
|
X_train_part = X_train[:-eval_size]
|
|
y_train_part = y_train[:-eval_size]
|
|
|
|
dtrain_part = xgb.DMatrix(X_train_part, label=y_train_part)
|
|
deval = xgb.DMatrix(X_eval, label=y_eval)
|
|
|
|
# 评估列表
|
|
evallist = [(dtrain_part, 'train'), (deval, 'eval')]
|
|
|
|
# 训练模型，使用早停
|
|
num_round = 1000 # 设置较大的轮数，让早停机制决定何时停止
|
|
self.model = xgb.train(
|
|
params,
|
|
dtrain_part,
|
|
num_round,
|
|
evallist,
|
|
early_stopping_rounds=50, # 50轮内评估指标无改善则停止
|
|
verbose_eval=True
|
|
)
|
|
|
|
# 使用全部训练数据重新训练最终模型，使用最佳轮数
|
|
# best_rounds = self.model.best_ntree_limit
|
|
# 兼容新版本的XGBoost
|
|
if hasattr(self.model, 'best_ntree_limit'):
|
|
best_rounds = self.model.best_ntree_limit
|
|
elif hasattr(self.model, 'best_iteration'):
|
|
best_rounds = self.model.best_iteration
|
|
elif hasattr(self.model, 'best_ntree_idx'):
|
|
best_rounds = self.model.best_ntree_idx
|
|
else:
|
|
# 如果都不存在，使用默认值
|
|
best_rounds = num_round
|
|
print(f"最佳轮数: {best_rounds}")
|
|
|
|
# 使用全部训练数据和最佳轮数训练最终模型
|
|
self.model = xgb.train(params, dtrain, best_rounds)
|
|
|
|
# 保存模型
|
|
if model_file_path:
|
|
self.model.save_model(model_name + model_file_path)
|
|
|
|
# 特征重要性分析
|
|
if hasattr(self.model, 'get_score'):
|
|
importance = self.model.get_score(importance_type='gain')
|
|
print("\n特征重要性 (gain):")
|
|
for key, value in sorted(importance.items(), key=lambda x: x[1], reverse=True):
|
|
print(f"{key}: {value}")
|
|
|
|
return self.model
|
|
def load_model(self, model_name=None, model_file_path='chan_xgb_model.json'):
|
|
if model_name:
|
|
self.model = xgb.Booster()
|
|
self.model.load_model(model_name + model_file_path)
|
|
else:
|
|
self.model = xgb.Booster()
|
|
self.model.load_model(model_file_path)
|
|
def find_best_params(self, dataframe=None, save_csv=False, csv_path_prefix='param_'):
|
|
"""
|
|
寻找最佳参数组合
|
|
:param dataframe: 输入的DataFrame，如果为None则使用初始化时的dataframe
|
|
:param save_csv: 是否保存特征数据到CSV文件
|
|
:param csv_path_prefix: CSV文件保存路径前缀，会自动添加参数信息
|
|
:return: 最佳参数
|
|
"""
|
|
# 不同参数组合
|
|
param_combinations = [
|
|
# 低学习率，深树
|
|
{'eta': 0.01, 'max_depth': 8, 'subsample': 0.8, 'colsample_bytree': 0.8, 'gamma': 0, 'min_child_weight': 1},
|
|
# 中等学习率，中等树深度
|
|
{'eta': 0.05, 'max_depth': 5, 'subsample': 0.7, 'colsample_bytree': 0.7, 'gamma': 0.1, 'min_child_weight': 3},
|
|
# 高学习率，浅树
|
|
{'eta': 0.1, 'max_depth': 3, 'subsample': 0.6, 'colsample_bytree': 0.6, 'gamma': 0.2, 'min_child_weight': 5},
|
|
# 正则化较强 best here
|
|
{'eta': 0.03, 'max_depth': 4, 'subsample': 0.8, 'colsample_bytree': 0.8, 'gamma': 0.1, 'min_child_weight': 3, 'alpha': 1, 'lambda': 3},
|
|
# 正则化较弱
|
|
{'eta': 0.08, 'max_depth': 6, 'subsample': 0.9, 'colsample_bytree': 0.9, 'gamma': 0, 'min_child_weight': 1, 'alpha': 0, 'lambda': 0.5},
|
|
]
|
|
|
|
best_score = 0
|
|
best_params = None
|
|
best_model = None
|
|
|
|
for i, params in enumerate(param_combinations):
|
|
print(f"\n尝试参数组合: {params}")
|
|
# 生成CSV文件名，包含一些参数信息
|
|
param_info = f"eta{params['eta']}_depth{params['max_depth']}"
|
|
train_csv_path = f"{csv_path_prefix}train_{param_info}.csv" if save_csv else None
|
|
|
|
model = self.train_model(dataframe=dataframe, data_file_path=train_csv_path, custom_params=params)
|
|
|
|
# 分割数据集，后20%用于测试
|
|
if dataframe is None:
|
|
dataframe = self.dataframe
|
|
|
|
train_size = int(len(dataframe) * 0.8)
|
|
test_df = dataframe.iloc[train_size:].copy()
|
|
|
|
# 获取测试集特征和标签
|
|
test_csv_path = f"{csv_path_prefix}test_{param_info}.csv" if save_csv else None
|
|
X_test, y_test = self.get_validate_feature_data(test_df, save_csv=save_csv, csv_path=test_csv_path)
|
|
|
|
if len(X_test) == 0:
|
|
print("没有提取到足够的测试特征数据")
|
|
continue
|
|
|
|
# 预测
|
|
dtest = xgb.DMatrix(X_test)
|
|
y_pred_prob = model.predict(dtest)
|
|
y_pred = [1 if p > 0.5 else 0 for p in y_pred_prob]
|
|
|
|
# 计算F1分数
|
|
f1 = f1_score(y_test, y_pred, zero_division=0)
|
|
print(f"F1分数: {f1:.4f}")
|
|
|
|
if f1 > best_score:
|
|
best_score = f1
|
|
best_params = params
|
|
best_model = model
|
|
|
|
print(f"\n最佳参数组合 (F1={best_score:.4f}):")
|
|
print(best_params)
|
|
self.model = best_model
|
|
|
|
return best_params
|
|
|
|
def get_feature_data(self, dataframe, save_csv=False, csv_path='feature_data.csv'):
|
|
"""
|
|
从dataframe提取特征数据
|
|
:param dataframe: 输入的DataFrame
|
|
:param save_csv: 是否保存特征数据到CSV文件
|
|
:param csv_path: CSV文件保存路径
|
|
:return: 特征矩阵X和标签y
|
|
"""
|
|
# 使用ChanLun获取bi_list
|
|
klc_list = self.chan.get_klc_list(dataframe)
|
|
bi_list = self.chan.cal_bi_list(klc_list)
|
|
seg_list = self.chan.get_seg_list(bi_list)
|
|
# 筛选方向为UP的bi的起始klc
|
|
feature_data = []
|
|
labels = []
|
|
feature_keys = [] # 用于保存特征名称
|
|
|
|
bi_index = 1
|
|
sample_list = []
|
|
for klc in klc_list:
|
|
if klc.klc_fx_type != Chan_KLC_FX.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:
|
|
#klc.set_bi(bi)
|
|
|
|
# 提取特征
|
|
features = klc.get_feature_data()
|
|
|
|
# 保存第一个样本的特征名称，用于CSV列名
|
|
if len(feature_keys) == 0:
|
|
feature_keys = list(features.keys())
|
|
|
|
# 将特征转换为模型可用的格式
|
|
feature_vec = []
|
|
for key, value in features.items():
|
|
if isinstance(value, (int, float)):
|
|
feature_vec.append(value)
|
|
else:
|
|
feature_vec.append(0)
|
|
|
|
# 判断这个bi是否赚钱（这里简单定义为：如果bi的结束价格高于起始价格，则标记为1，否则为0）
|
|
# 这个标签定义可以根据实际需求修改
|
|
matched = False
|
|
for bi in bi_list:
|
|
if bi.end_klc and bi.end_klc.index == klc.index:
|
|
#print(bi.start_time, bi.start_klc.start_time, bi.dir)
|
|
label = 1
|
|
matched = True
|
|
break
|
|
if not matched:
|
|
label = 0
|
|
|
|
feature_data.append(feature_vec)
|
|
labels.append(label)
|
|
|
|
# 如果需要保存到CSV
|
|
if save_csv:
|
|
# 创建DataFrame保存特征数据
|
|
# 只保留数值型特征
|
|
numeric_feature_keys = [key for i, key in enumerate(feature_keys)
|
|
if i < len(feature_data[0]) if isinstance(feature_data[0][i], (int, float))]
|
|
|
|
# 创建特征数据的DataFrame
|
|
df_features = pd.DataFrame(feature_data, columns=numeric_feature_keys)
|
|
# 添加标签列
|
|
df_features['label'] = labels
|
|
# 添加时间信息便于分析
|
|
if len(sample_list) > 0:
|
|
times = [klc.start_time for klc in sample_list]
|
|
df_features['time'] = times
|
|
|
|
# 保存到CSV
|
|
df_features.to_csv(csv_path, index=False)
|
|
print(f"特征数据已保存到 {csv_path}")
|
|
|
|
# 在return前添加
|
|
positive_count = np.sum(labels)
|
|
print(f"正样本数量: {positive_count}, 负样本数量: {len(labels) - positive_count}")
|
|
print("Trainning data: ", len(feature_data), klc_list[-1].start_time, klc_list[-1].klc_fx_type , "---------------------")
|
|
return np.array(feature_data), np.array(labels)
|
|
def get_validate_feature_data(self, dataframe, save_csv=False, csv_path='validate_feature_data.csv'):
|
|
"""
|
|
从dataframe提取特征数据
|
|
:param dataframe: 输入的DataFrame
|
|
:param save_csv: 是否保存特征数据到CSV文件
|
|
:param csv_path: CSV文件保存路径
|
|
:return: 特征矩阵X和标签y
|
|
"""
|
|
# 使用ChanLun获取bi_list
|
|
klc_list = self.chan.get_klc_list(dataframe)
|
|
bi_list = self.chan.cal_bi_list(klc_list)
|
|
seg_list = self.chan.get_seg_list(bi_list)
|
|
# 筛选方向为UP的bi的起始klc
|
|
feature_data = []
|
|
labels = []
|
|
feature_keys = [] # 用于保存特征名称
|
|
|
|
bi_index = 1
|
|
sample_list = []
|
|
for klc in klc_list:
|
|
if klc.klc_fx_type != Chan_KLC_FX.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]
|
|
# 提取特征
|
|
features = klc.get_feature_data()
|
|
|
|
# 保存第一个样本的特征名称，用于CSV列名
|
|
if len(feature_keys) == 0:
|
|
feature_keys = list(features.keys())
|
|
|
|
# 将特征转换为模型可用的格式
|
|
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)
|
|
seg = seg_list[bi_index]
|
|
matched = False
|
|
for bi in bi_list:
|
|
if bi.end_klc and bi.end_klc.index == klc.index:
|
|
label = 1
|
|
matched = True
|
|
break
|
|
if not matched:
|
|
label = 0
|
|
|
|
feature_data.append(feature_vec)
|
|
labels.append(label)
|
|
|
|
# 如果需要保存到CSV
|
|
if save_csv:
|
|
# 创建DataFrame保存特征数据
|
|
# 只保留数值型特征
|
|
numeric_feature_keys = [key for i, key in enumerate(feature_keys)
|
|
if i < len(feature_data[0]) if isinstance(feature_data[0][i], (int, float))]
|
|
|
|
# 创建特征数据的DataFrame
|
|
df_features = pd.DataFrame(feature_data, columns=numeric_feature_keys)
|
|
# 添加标签列
|
|
df_features['label'] = labels
|
|
# 添加时间信息便于分析
|
|
if len(sample_list) > 0:
|
|
times = [klc.start_time for klc in sample_list]
|
|
df_features['time'] = times
|
|
|
|
# 保存到CSV
|
|
df_features.to_csv(csv_path, index=False)
|
|
print(f"验证特征数据已保存到 {csv_path}")
|
|
|
|
print("Validating data: ", len(feature_data), klc_list[-1].start_time, klc_list[-1].klc_fx_type , "---------------------")
|
|
return np.array(feature_data), np.array(labels)
|
|
def validate_model(self, dataframe=None, save_csv=False, csv_path='validate_feature_data.csv'):
|
|
"""
|
|
使用dataframe后20%的数据验证模型
|
|
:param dataframe: 输入的DataFrame，如果为None则使用初始化时的dataframe
|
|
:param save_csv: 是否保存特征数据到CSV文件
|
|
:param csv_path: CSV文件保存路径
|
|
:return: 验证结果
|
|
"""
|
|
if self.model is None:
|
|
print("模型尚未训练，请先调用train_model方法")
|
|
return None
|
|
|
|
if dataframe is None:
|
|
dataframe = self.dataframe
|
|
|
|
# 分割数据集，后20%用于测试
|
|
train_size = int(len(dataframe) * 0.8)
|
|
test_df = dataframe.iloc[train_size:].copy()
|
|
|
|
# 获取测试集特征和标签
|
|
X_test, y_test = self.get_validate_feature_data(test_df, save_csv=save_csv, csv_path=csv_path)
|
|
|
|
if len(X_test) == 0:
|
|
print("没有提取到足够的测试特征数据")
|
|
return None
|
|
|
|
# 预测
|
|
dtest = xgb.DMatrix(X_test)
|
|
y_pred_prob = self.model.predict(dtest)
|
|
y_pred = [1 if p > 0.5 else 0 for p in y_pred_prob]
|
|
|
|
# 计算评估指标
|
|
accuracy = accuracy_score(y_test, y_pred)
|
|
precision = precision_score(y_test, y_pred, zero_division=0)
|
|
recall = recall_score(y_test, y_pred, zero_division=0)
|
|
f1 = f1_score(y_test, y_pred, zero_division=0)
|
|
|
|
# 打印评估报告
|
|
print("模型评估结果:")
|
|
print(f"准确率: {accuracy:.4f}")
|
|
print(f"精确率: {precision:.4f}")
|
|
print(f"召回率: {recall:.4f}")
|
|
print(f"F1分数: {f1:.4f}")
|
|
print("\n分类报告:")
|
|
print(classification_report(y_test, y_pred, zero_division=0))
|
|
|
|
return {
|
|
'accuracy': accuracy,
|
|
'precision': precision,
|
|
'recall': recall,
|
|
'f1': f1,
|
|
'y_test': y_test,
|
|
'y_pred': y_pred,
|
|
'y_pred_prob': y_pred_prob
|
|
}
|
|
|
|
def predict(self, klc):
|
|
"""
|
|
使用训练好的模型预测单个KLC
|
|
:param klc: 需要预测的ChanKLC对象
|
|
:return: 预测结果（概率值）
|
|
"""
|
|
if self.model is None:
|
|
print("模型尚未训练，请先调用train_model方法")
|
|
return None
|
|
|
|
# 提取特征
|
|
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(self.model.predict(dtest)[0])
|
|
|
|
def get_decimal(self, value):
|
|
return Decimal("{:.4f}".format(value))
|
|
|