一個基於機器學習的打板策略分享（六）: 番外之打臉篇!
一個基於機器學習的打板策略分享（六）: 番外之打臉篇!
版權所有:光明與黑暗([email protected])
轉載請註明出處: https://www.zhihu.com/people/guang-ming-yu-hei-an-22-2/activities
說句老實話,我是真的不想寫這個番外的,尤其寫的還是打臉打到啪啪響的番外.不過本著自己約的炮含淚也要打完的宗旨,當然你也可以理解為一個四十歲男人的偏執.最終還是有了這篇番外篇的誕生.
一個基於機器學習的打板策略分享系列出來後,有不少量化專業人士跟我通過各種渠道進行了交流.在溝通和實踐的過程中,我發現代碼里至少有兩處不當的地方,下面開始打面:
第一下打左邊臉,問題出在這一句:
df_minute_line[obv]=talib.OBV(df_minute_line[abs_close].values,df_minute_line[abs_vol].values)
在本系列的第一篇我引入了所謂的相對於上一交易日收市價的變化價和相對於自由流通股本的成交量百分比意圖將信號標準化,同時引入OBV試圖得出一個價量加權結合的時間序列二維向量.但可能因為OBV的內部實現對交易價格和成交量也做了相對上一時刻的按比例變換(以某時點為基期，逐時點累計每時點上市股票總成交量，若下個時點指數或股票上漲 ，則基期OBV加上該時點成交量為本時點OBV),因此在拉升觸及漲停的判斷過程中OBV反而令真正漲停的和佯攻漲停的信號非常相似,反而降低了模型的預測準確率.
因此我決定放棄OBV,將價和量分別做信號提取和統一.
第二下打右邊臉,問題出在這三句:
ts_train=to_time_series_dataset(stock_minute_train_data,dtype=np.float32)
where_are_nan = np.isnan(ts_train)
ts_train[where_are_nan] = 0
這裡必須先說明一下開始這麼寫的原因:tslearn要求監督分類輸入的是一個定長數組.對於漲停來說,因為漲停時點是不定的,因此輸入不可能是定長的.那有什麼辦法讓它等長呢?我第一時間想到的就是在數據後面補0.乍看沒有問題,但是在實時預測中,除了漲停外經常會出現跌停或者價格波動不大的誤報信號,歸根結底就是因為補0後連續的若干個0也被當做信號的判斷因素之一.
那咋辦呢?我想到了一個辦法,就是對信號進行重新採樣:
所以我對build_train_data方法做了一些修改:
1)先用小波變換截取信號:
A2,D6,D5,D4=pywt.wavedec(df_minute_temp[abs_close].values, db4,mode=symmetric,level=3)
stock_minute_train_close.append(to_time_series(A2))
2)然後用TimeSeriesResampler 對信號重新採樣為統一長度
tsr_close=TimeSeriesResampler(sz=24)
stock_minute_train_close=tsr_close.fit_transform(stock_minute_train_close)
此刻,作者的臉是這樣的:
幸好,身為不作死就不會死星人我是絕對不會善罷甘休的.因此有了我埋頭苦思的若干分鐘和飄散在空中的幾根白髮.終於打板系列進化到V2版了!
在文章的最後,作為補充,我放出訓練模型和實時預測的絕大部分代碼,希望對各位有所啟發,歡迎各位與我郵件或知乎討論!
LimitUpTacticsTrain.py
# -*- coding: UTF-8 -*-
#1)不要問我tushare、pandas、pytdx、tslearn、pywt怎麼安裝這些問題,請自行百度； 一般pip install 即可。
#2)本程序需要300以上tushare積分才可運行,積分獲取方法請參考https://tushare.pro/document/1?doc_id=13 ,當然你通過https://tushare.pro/register?reg=100017 註冊tushare我會謝謝你的。
#3)這是訓練模型的程序,策略原理請參考https://zhuanlan.zhihu.com/c_1100040485648797696 的一個基於機器學習的打板策略分享系列，歡迎在知乎交流。
import tushare as ts
import pandas as pd
import numpy as np
import datetime
import pytdx
import pywt
import tslearn
import talib
import pickle
import os
import math
import peakutils
from scipy import stats as st
from datetime import datetime
from pytdx.hq import TdxHq_API
from tslearn.utils import save_timeseries_txt ,load_timeseries_txt,to_time_series_dataset,to_time_series
from tslearn.neighbors import KNeighborsTimeSeriesClassifier
from tslearn.clustering import TimeSeriesKMeans,silhouette_score,GlobalAlignmentKernelKMeans
from tslearn.preprocessing import TimeSeriesResampler
from tqdm import *
import warnings
#忽略警告信息
warnings.filterwarnings("ignore")
#☆常量定義,改這句即可運行.請支持tushare和米哥.
ts_token=請填你tushare的token值
#初始化當天日期
today=str(datetime.now().strftime(%Y%m%d))
#pytdx伺服器地址
pytdx_server=119.147.212.81
#pytdx伺服器埠
pytdx_port=7709
#初始化TS_API介面
def get_ts_api():
ts.set_token(ts_token)
pro = ts.pro_api()
return pro
#初始化pytdx介面
def get_pytdx_api():
api = TdxHq_API(auto_retry=True, raise_exception=True,heartbeat=True)
return api
#獲取全部股票每日重要的基本面指標,主要用到收市價和自由流通股本
def build_stock_daily_basic(trade_date):
pro=get_ts_api()
df_stock_daily_basic=pro.daily_basic(trade_date=trade_date)
return df_stock_daily_basic
#獲取股票日線行情,主要用到漲跌幅和最高價和最低價
def build_stock_daily(trade_date):
pro=get_ts_api()
df_stock_daily=pro.daily(trade_date=trade_date)
return df_stock_daily
#使用pytdx獲取歷史分鐘線數據
def get_stock_minute_data(api,stock_code,pre_close,free_share,stock_date):
df_minute_line=api.to_df(api.get_history_minute_time_data(select_market_code(stock_code),stock_code[0:6],stock_date))
#https://zhuanlan.zhihu.com/p/62339900
df_minute_line[abs_close]=df_minute_line[price]/pre_close
df_minute_line[abs_vol]=df_minute_line[vol]/(free_share*100)
#去掉特殊數值
df_minute_line=df_minute_line.replace(np.NaN,0)
df_minute_line=df_minute_line.replace(np.inf,0)
#刪掉無效欄位
df_minute_line=df_minute_line.drop(columns=[price,vol], axis=1)
#截取從開市到close首次到達最高點的分鐘線
index_abs_close_max=df_minute_line[df_minute_line[abs_close]==df_minute_line[abs_close].max()].index.min()+1
df_minute_line=df_minute_line.iloc[0:index_abs_close_max]
#重構索引
df_minute_line=df_minute_line.reset_index(drop=True)
return df_minute_line
#根據股票代碼確認是深市還是滬市
def select_market_code(code):
market_code=0
if code.startswith(60,0,2):
market_code=1
return market_code
#訓練分鐘數據 https://zhuanlan.zhihu.com/p/62784517
def train_minute_data():
#載入訓練數據
close_x=load_timeseries_txt(stock_minute_train_close.txt)
y=np.loadtxt(stock_minute_label_data.csv,delimiter=,,dtype=int)
#如果已有模型文件從本地載入
if os.path.exists(LimitUpTacticsClose.h5):
fr_close=open(LimitUpTacticsClose.h5, rb)
clf_close = pickle.load(fr_close, fix_imports=True)
#如果本地無模型文件新創建一個
else:
clf_close=KNeighborsTimeSeriesClassifier(n_neighbors =2,metric="dtw")
clf_close.fit(close_x, y)
#保存模型文件到本地
with open(LimitUpTacticsClose.h5, wb) as fw_close:
pickle.dump(clf_close, fw_close,fix_imports=True)
#載入文件
df_top_three=pd.read_csv(vol_top_three.csv,encoding=gbk)
#對價格變數進行四捨五入
df_top_three[range]=round(df_top_three[abs_close],2)
#構建空白DataFrame
df_step=pd.DataFrame()
#獲得各價格段的80%置信區間
for name, group in df_top_three.groupby(range):
mean=np.mean(group[abs_vol].values)
std=st.sem(group[abs_vol].values)
#按80%的置信區間取上下界
conf_intveral = st.norm.interval(0.8, loc=mean, scale=std)
#判斷元組的上下界元素都不為Nan
if(name!=np.nan)and(math.isnan(conf_intveral[0])==False)and(math.isnan(conf_intveral[1])==False):
df_step=df_step.append(pd.DataFrame(data=[[name,conf_intveral[0],conf_intveral[1]]],columns=[range,lower_limit,upper_limit]), ignore_index=True)
df_step.to_csv(df_step.csv,encoding=gbk,index=False)
#構建訓練數據並訓練
def build_train_data(start_date,end_date):
pro=get_ts_api()
api=get_pytdx_api()
stock_minute_train_close=[]
stock_minute_train_vol=[]
stock_minute_label_data=[]
#獲取起止日期間的交易日
df_trade_cal=pro.trade_cal(exchange=, start_date=start_date, end_date=end_date)
df_trade_cal=df_trade_cal[(df_trade_cal[is_open]==1)]
tsr_close=TimeSeriesResampler(sz=24)
tsr_vol=TimeSeriesResampler(sz=10)
df_top_three=pd.DataFrame(columns=[stock_code,trade_date,abs_vol,abs_close])
with api.connect(pytdx_server, pytdx_port):
for j in tqdm(range(0,len(df_trade_cal))):
cal_date=df_trade_cal.iloc[j][cal_date]
df_stock_daily=build_stock_daily(cal_date)
df_stock_daily_basic=build_stock_daily_basic(cal_date)
for i in tqdm(range(0,len(df_stock_daily))):
ts_code=df_stock_daily.iloc[i][ts_code]
pct_chg=df_stock_daily.iloc[i][pct_chg]
#非新股上市
if(pct_chg<11)and(pct_chg>-11):
#獲取上一交易日收市價
pre_close=df_stock_daily.iloc[i][pre_close]
#獲取股票自由流通股本
free_share=df_stock_daily_basic[df_stock_daily_basic[ts_code]==ts_code][free_share].values[0]
df_minute_temp=get_stock_minute_data(api, ts_code, pre_close, free_share, cal_date)
if(len(df_minute_temp)>0):
#漲幅大於9%的取成交量前三的分鐘線成交量數據
if(pct_chg>=9):
#取成交量前三的分鐘線成交量數據
df_temp=df_minute_temp.sort_values(by=abs_vol,ascending=False).head(3)
df_temp[stock_code]=ts_code
df_temp[trade_date]=pd.to_datetime(cal_date)
df_temp[abs_close]=df_temp[abs_close]-1
df_top_three=df_top_three.append(df_temp, ignore_index=True)
stock_minute_label_data.append(round(pct_chg,0))
#提取價格的高頻信號
A2,D6,D5,D4=pywt.wavedec(df_minute_temp[abs_close].values, db4,mode=symmetric,level=3)
stock_minute_train_close.append(to_time_series(A2))
#重新採樣為24個點的信號
stock_minute_train_close=tsr_close.fit_transform(stock_minute_train_close)
ts_train_close=to_time_series_dataset(stock_minute_train_close,dtype=np.float32)
where_are_nan = np.isnan(ts_train_close)
ts_train_close[where_are_nan] = 0
save_timeseries_txt(stock_minute_train_close.txt, ts_train_close)
#保存標籤數據
np.savetxt(stock_minute_label_data.csv, stock_minute_label_data,delimiter=,,fmt=%i)
#保存成交量前三的匯總數據
df_top_three.to_csv(vol_top_three.csv,encoding=gbk,index=False)
if __name__ == "__main__":
#☆如果你能填為19940808並訓練成功,請務必發一份模型文件給我,哪怕是20190101也好
build_train_data(20190101, today)
train_minute_data()
LimitUpTacticsPredictProba.py
# -*- coding: UTF-8 -*-
#1)不要問我tushare、pandas、pytdx、tslearn、pywt怎麼安裝這些問題,請自行百度； 一般pip install 即可。
#2)本程序需要300以上tushare積分才可運行,積分獲取方法請參考https://tushare.pro/document/1?doc_id=13 ,當然你通過https://tushare.pro/register?reg=100017 註冊tushare我會謝謝你的。
#3)這是預測概率的程序,策略原理請參考https://zhuanlan.zhihu.com/c_1100040485648797696 的一個基於機器學習的打板策略分享系列，歡迎在知乎交流。
import tushare as ts
import pandas as pd
import numpy as np
import datetime
import pytdx
import pywt
import tslearn
import talib
import pickle
import os
import multiprocessing
from datetime import datetime
from pytdx.hq import TdxHq_API
from tslearn.neighbors import KNeighborsTimeSeriesClassifier
from tslearn.preprocessing import TimeSeriesResampler
from tslearn.utils import save_timeseries_txt ,load_timeseries_txt,to_time_series_dataset,to_time_series
import warnings
#忽略警告信息
warnings.filterwarnings("ignore")
#使用CPU核心設置
thread_count=int(multiprocessing.cpu_count()-2)
#☆常量定義,改這句即可運行.請支持tushare和米哥.
ts_token=請填你tushare的token值
#初始化當天日期
today=str(datetime.now().strftime(%Y%m%d))
#pytdx伺服器地址
pytdx_server=119.147.212.81
#pytdx伺服器埠
pytdx_port=7709
#概率閾值
probability_threshold=0.5
#初始化TS_API介面
def get_ts_api():
ts.set_token(ts_token)
pro = ts.pro_api()
return pro
#初始化pytdx介面
def get_pytdx_api():
api = TdxHq_API(auto_retry=True, raise_exception=True,heartbeat=True)
return api
#傳入五個參數，分別是實例化的TdxHq_API對象、股票代碼、上一交易日的收市價，上一交易日自由流通股本數和要獲取的分鐘線的日期
def get_stock_minute_data(api,stock_code,pre_close,free_share,stock_date):
df_minute_line=api.to_df(api.get_security_bars(8, get_market_code(stock_code),stock_code[0:6], 0, 800))
df_minute_line[datetime]=pd.to_datetime(df_minute_line[datetime])
df_minute_line=df_minute_line[(df_minute_line[datetime]>=stock_date)]
df_minute_line[abs_close]=df_minute_line[close]/pre_close
df_minute_line[abs_vol]=df_minute_line[vol]/(free_share*100)
#去掉特殊數值
df_minute_line=df_minute_line.replace(np.NaN,0)
df_minute_line=df_minute_line.replace(np.inf,0)
#刪掉無用欄位
df_minute_line=df_minute_line.drop(columns=[open,close,high,low,vol,amount,year,month,day,hour,minute,datetime], axis=1)
#重構索引
df_minute_line=df_minute_line.reset_index(drop=True)
return df_minute_line
#根據股票代碼確認是深市還是滬市
def get_market_code(code):
market_code=0
if code.startswith(60,0,2):
market_code=1
return market_code
#獲取上一交易日
def get_last_trade_date(stock_date):
pro=get_ts_api()
df_trade_cal=pro.trade_cal()
df_trade_cal=df_trade_cal[(df_trade_cal[is_open]==1)&(df_trade_cal[cal_date]<str(stock_date))]
return df_trade_cal.tail(1)[cal_date].values[0]
#☆請自行實現你的股票池方法
def get_stock_pool():
#return [600336.SH,002255.SZ,000720.SZ,601689.SH,000413.SZ,300749.SZ,603679.SH]
return [600758.SH]
#獲取某隻股票上一交易日的收市價和自由流通股本
def get_stock_daily_basic(stock_date):
pro=get_ts_api()
df_stock_daily_basic=pro.daily_basic(trade_date=stock_date)
return df_stock_daily_basic
#獲取股票日線行情,主要用到漲跌幅和最高價和最低價
def get_stock_daily(trade_date):
pro=get_ts_api()
df_stock_daily=pro.daily(trade_date=trade_date)
return df_stock_daily
#☆在此實現你自己對於該股票今天是否會漲停的判斷,例如我會去5擋委買看看
def your_extra_judge(api,stock_code,pre_close,df_step,free_share):
#某個不可描述的迷之介面
df_real_time=******
df_real_time=df_real_time.replace(to_replace=, value=0)
#五檔委買價格數組
price_range=[df_real_time[bid1][0],df_real_time[bid2][0],df_real_time[bid3][0],df_real_time[bid4][0],df_real_time[bid5][0]]
#五檔委買手數數組
vol_range=[df_real_time[bid_vol1][0],df_real_time[bid_vol2][0],df_real_time[bid_vol3][0],df_real_time[bid_vol4][0],df_real_time[bid_vol5][0]]
flag=False
for i in range(0,len(price_range)):
#計算當前相對價格
abs_price=round(price_range[i]/pre_close-1,2)
#查找大於當前相對價格的檔位
df_step_sub=df_step[df_step[range]>=abs_price]
#計算相對委買力度
abs_vol=vol_range[i]/(free_share*100)
for j in range(0,len(df_step_sub)):
if(abs_vol>=df_step_sub.iloc[j][lower_limit])and(abs_price>0):
flag=True
print(flag)
return flag
#實時預測
def real_time_predict_proba():
#載入自選股票池,請自行實現
stock_pool=get_stock_pool()
#獲取上一交易日
last_trade_date=get_last_trade_date(today)
#獲取上一交易日的基本信息
stock_daily_basic=get_stock_daily_basic(last_trade_date)
stock_daily=get_stock_daily(last_trade_date)
multiprocessing.freeze_support()
df_step=pd.read_csv(df_step.csv,encoding=gbk)
pool = multiprocessing.Pool(processes = min(thread_count,len(stock_pool)))
#載入模型文件
if os.path.exists(LimitUpTacticsClose.h5):
fr_close=open(LimitUpTacticsClose.h5, rb)
clf_close = pickle.load(fr_close, fix_imports=True)
#循環啟動後台進程
for i, stock_code_temp in enumerate(stock_pool):
stock_daily_basic_temp=stock_daily_basic[(stock_daily_basic[ts_code]==stock_code_temp)]
stock_daily_temp=stock_daily[(stock_daily[ts_code]==stock_code_temp)]
if(len(stock_daily_basic_temp)>0)and(len(stock_daily_temp)>0):
#獲取上一交易日收市價
pre_close=stock_daily_basic_temp[close].values[0]
#獲取上一交易自由流通股本
free_share=stock_daily_basic_temp[float_share].values[0]
#多進程版預測
pool.apply_async(sub_process_predict_proba, (stock_code_temp,pre_close,free_share,today,clf_close,df_step, ))
pool.close()
pool.join()
#多進程版預測
def sub_process_predict_proba(stock_code,pre_close,free_share,stock_date,clf_close,df_step):
api=get_pytdx_api()
tsr_close=TimeSeriesResampler(sz=24)
try:
with api.connect(pytdx_server, pytdx_port):
df_minute_line=get_stock_minute_data(api, stock_code, pre_close, free_share, stock_date)
A2,D6,D5,D4=pywt.wavedec(df_minute_line[abs_close].values, db4,mode=symmetric,level=3)
probability_close=clf_close.predict_proba(tsr_close.fit_transform(A2))
max_index=probability_close[0].argmax(axis=0)
print(probability_close[0])
if(probability_close[0][max_index]>=probability_threshold)and(max_index>=11)and(your_extra_judge(api,stock_code,pre_close,df_step,free_share)):
print(股票代碼為,stock_code,的股票當前還不趕緊買入更待何時!)
except Exception as e:
print(e)
pass
if __name__ == "__main__":
real_time_predict_proba()
推薦閱讀：
