SVM-and-HS300-maste

# -*- coding: utf-8 -*- # 基于沪深300股票数据支持向量机SVM预测股票涨跌方向 数据多运行时间大约几分钟 可以快进 import pandas as pd import numpy as np import seaborn as sns import matplotlib.pyplot as plt from sklearn.svm import SVC from sklearn.model_selection import GridSearchCV as GSCV class SVM(): train_years=10 #训练样本的大小，十年数据 year_days=240 #一年有240个交易日 def __init__(self, momentum, span): self.m_span = momentum #动量的观测天数 self.span = span #波动性的观测天数 self.slippage = 0.0002 #滑点情况，共2个基点，买入卖出各1个基点 def data_pre(self): ''' 数据准备，构建数据特征，并标记类别标签 ''' data = self.data data['retn'] = data['close'].pct_change() span = self.span attr = pd.DataFrame() #构建七个数据特征，刻画当天，周度和月度的市场状况 attr['high_low'] = (data['high'] - data['low']) / data['open'] attr['close_open'] = data['close'] / data['open'] - 1 attr['vol_pct'] = data['volume'].pct_change() attr['pct_m'] = data['close'].shift(-self.m_span) / data['open'] - 1 for i in range(span, len(data)-span): now = data.index[i] begin = data.index[i-span] attr.loc[now, 'high_v'] = data.loc[now, 'high']\ / max(data.loc[begin:now, 'high']) - 1 attr.loc[now, 'low_v'] = data.loc[now, 'low']\ / min(data.loc[begin:now, 'low']) - 1 attr.loc[now, 'sigma'] = data.loc[begin:now, 'retn'].std()\ * self.year_days / span attr.dropna(inplace= True) #将特征标准化 for field in attr.columns: attr[field] = (attr[field] - attr[field].mean()) / attr[field].std() #上涨标记为1，下跌标记为-1 attr['label'] = np.where(attr['close_open'].shift(-1) > 0, 1, -1) return attr def data_split(self, begin_year): ''' 数据划分，十年数据为训练样本，下一年的数据为测试样本 Args: begin_year: 第几年，0表示第一年，以此类推 Returns: train_data: 训练集 test_data: 测试集 ''' begin = begin_year * self.year_days #训练集开始的位置 end = begin + self.train_years * self.year_days #训练集结束的位置 attrs = self.data_pre() train_data = attrs.iloc[begin : end, :] #训练集 test_data = attrs.iloc[end : end + self.year_days, :] #下一年数据为测试集 return train_data, test_data def gscv_para(self, C_list, gamma_list, x_train, y_train): ''' 用网格搜索和交叉验证调节参数，考虑类别不平衡的情况,k-fold的k为10 Args: C_list: C参数的备选列表 gamma_list: gamma参数的备选列表 x_train: 训练集中的特征数据 y_train: 训练集中的类别数据 Returns: 最优的C参数，gamma参数和最优时的score(平均值) 优化的标准是SVC的score值，score越高表示，表示参数越好 ''' clf = SVC(class_weight='balanced', cache_size=4000) gscv = GSCV(clf, param_grid={'C': C_list, 'gamma': gamma_list}, n_jobs=-1, cv=10, pre_dispatch=4) gscv.fit(x_train, y_train) return gscv.best_params_.values(), gscv.best_score_ def svm_fit(self, x_train, y_train): ''' 用网格搜索调参数 ''' C_list = list(range(1, 5)) gamma_list = np.linspace(0.01, 0.05, 5) (C, gamma), score = self.gscv_para(C_list, gamma_list, x_train, y_train) #可能存在参数在边界的情况，此时需要重置一下参数的备选范围 if C == 1 or C == 10: if C == 1: print('C touched the lower limit') C_list = np.linspace(0.2, 2, 10) else: print('C touched the upper limit') C_list = list(range(1, 10)) if gamma == 0.01 or gamma == 0.1: #C和gamma都在边界 if gamma == 0.01: print('gamma touched the lower limit') gamma_list = np.linspace(0.002, 0.02, 10) else: print('gamma touched the upper limit') gamma_list = np.linspace(0.02, 0.2, 10) (C, gamma), score = self.gscv_para(C_list, gamma_list, x_train, y_train) else: pass if gamma == 0.01 or gamma == 0.1: #仅仅是gamma在边界 if gamma == 0.01: print('gamma touched the lower limit') gamma_list = np.linspace(0.002, 0.02, 10) else: print('gamma touched the upper limit') gamma_list = np.linspace(0.02, 0.2, 10) (C, gamma), score = self.gscv_para(C_list, gamma_list, x_train, y_train) else: pass return C, gamma, score def predict(self, begin_year): ''' 用拟合的分类器对样本外的数据做预测 ''' train_data, test_data = self.data_split(begin_year) x_train, y_train = train_data.iloc[:, :-1], train_data.iloc[:, -1] x_test, y_test = test_data.iloc[:, :-1], test_data.iloc[:, -1] C, gamma, score = self.svm_fit(x_train, y_train) # C, gamma, score, _ = paras[begin_year] clf = SVC(C=C, gamma=gamma, class_weight='balanced', cache_size=4000) clf.fit(x_train, y_train) test_score = clf.score(x_test, y_test) #测试集上的准确率 para = C, gamma, score, test_score print('%d, %.4f, %.4f, %.4f' % para) #在测试集上的预测值 y_predict = pd.Series(clf.predict(x_test), index=y_test.index, name='predict') return para, pd.concat([y_test, y_predict], axis=1) def cum_retn(self, years): retns = {} paras = [] for i in range(years): para, label = self.predict(i) paras.append(para) for j in range(len(label)): date = label.index[j] if label.loc[date, 'label'] == label.loc[date, 'predict']: #预测正确，则获取多空收益 retn_i = abs(data.loc[date, 'close'] / data.loc[date, 'open'] - 1) retns[date] = retn_i - self.slippage else: #预测失误，产生损失 retn_i = -abs(data.loc[date, 'close'] / data.loc[date, 'open'] - 1) retns[date] = retn_i - self.slippage return paras, 1 + pd.Series(retns).sort_index().cumsum() if __name__ == '__main__': # 网络搜索耗费时间 ！！！ data = pd.read_excel('data/HS300_05_18.xlsx', index_col='date') SVM.data = data hs = SVM(5, 20) attr = hs.data_pre() plt.figure(figsize=(6, 5)) sns.heatmap(attr.iloc[:, :-1].corr()) plt.xticks(fontsize=12) plt.yticks(fontsize=12) plt.tight_layout() plt.savefig('attr_heatmap.png') total_years = int(len(data) / 240) + 1 years = total_years - SVM.train_years paras, nav = hs.cum_retn(years) def drawdown(nav): #计算最大回撤 dd = [] for i in range(1, len(nav)): max_i = max(nav[:i]) dd.append(min(0, nav[i] - max_i) / max_i) return dd Drawdown = pd.Series(drawdown(nav), index=nav.index[1:]) MaxDD = min(Drawdown) Total_return = nav[-1] Ann