策略思路：传统网格策略需要设定参数为网格间距、仓位权重。
相当于把收盘价分成数个网格、价格移动到下个网格则成交。
传统网格策略需提前设置格子数量、若价格走到无格子的位置时则不产生交易。
本文主要参考网格策略的思想，取消格子数量限制，选用日内Tick级别高频数据，专注于下单策略及成交情况。
引入必要库、读取数据(可以使用Tick级别、分钟级别，不支持日级别)
import pandas as pdimport warningsimport datetimeimport numpy as npfrom pyecharts import options as optsfrom pyecharts.globals import ThemeTypefrom pyecharts.charts import Kline, Line,Grid,Scatterdata = pd.read_csv('你的数据源.csv')data.index = data['tradetime']# 打上时间标签为后续记录做准备data['time'] = data.index.map(lambda x: x.time())
初始化必要参数、可以将后续代码设置为函数
# 设置买买双边报单价格sell_price = data['close'][0]*1.01buy_price = data['close'][0]*0.99# 记录每日最初股票持仓、现金持仓s_inday = 5000cash_inday = 1000000# 记录实时股票持仓、s = s_indayc = cash_inday# 初始化卖单买单记录sell = []buy = []
遍历所有数据，记录交易情况，网格间距为1%
# 最后一个时点要做平仓、循环时不遍历for i in range(0, len(data)-1):# 假设时点的最高价>卖单报价，且有足够股票持仓则卖单成交if ((data['high'][i]>sell_price) & (s_inday > 200)):# 记录成交时点、报单成交价格、成交数量sell.append([data.index[i], sell_price.round(2), 200])# 更新当日可出售股票持仓记录、全部股票持仓、现金持仓记录s_inday -= 200s -= 200c += 200*data['close'][i]*0.998# 重置双边报价buy_price = sell_price *0.99sell_price = sell_price*1.01# 假设时点的最低价<卖单报价，且有足够现金持仓则买单成交elif ((data['low'][i]<buy_price) & (cash_inday > data.close[i]*200)):# 记录成交时点、报单成交价格、成交数量buy.append([data.index[i], buy_price.round(2),200])# 更新当日可用于购买现金持仓记录、全部股票持仓、现金持仓记录cash_inday -= data.close[i]*200c -= data.close[i]*200s += 200# 重置双边报价sell_price = buy_price*1.01buy_price = buy_price*0.99# 每日15:00确定次日可用股票、及现金持仓 if data['time'][i]==datetime.time(15, 0):s_inday =scash_inday =c
平仓、数据整理
# 循环结束后平仓if len(sell) - len(buy)> 0:buy.append([data.index[-1], data['close'][-1],(len(sell) - len(buy))*200])elif len(sell) - len(buy)< 0:sell.append([data.index[-1], data['close'][-1],(len(buy) - len(sell))*200])# 整理数据sell = pd.DataFrame(sell,columns = ['卖出时间','卖出价格', '卖出数量'])sell.index = sell['卖出时间']buy = pd.DataFrame(buy,columns = ['买入时间','买入价格', '买入数量'])buy.index = buy['买入时间']data = data.join(sell[['卖出价格', '卖出数量']])data = data.join(buy[['买入价格', '买入数量']])
绘图查看买点卖点
data_plot = dataochl = data_plot[['open','close','low','high']]k = [ochl.iloc[i].tolist() for i in range(len(ochl))]# K线图K = (Kline().add_xaxis(data_plot.index.tolist()).add_yaxis("kline",k).set_global_opts(yaxis_opts=opts.AxisOpts(is_scale=True),xaxis_opts=opts.AxisOpts(is_scale=True),title_opts=opts.TitleOpts(title="网格交易策略"),datazoom_opts=[opts.DataZoomOpts(type_='inside')]))#买点卖点散点图S_ = (Scatter().add_xaxis(data_plot.index.tolist()).add_yaxis("卖出",data_plot['卖出价格'].tolist()).add_yaxis("买入",data_plot['买入价格'].tolist()).set_global_opts(yaxis_opts=opts.AxisOpts(is_scale=True),xaxis_opts=opts.AxisOpts(is_scale=True),datazoom_opts=[opts.DataZoomOpts(type_='inside')]))K.overlap(S_) #将均线叠加到K线上K.render_notebook()
整理数据查看盈利情况
data = data.fillna(0)data['敞口'] = (data['买入数量'].cumsum()-data['卖出数量'].cumsum())data['实现盈利'] = (data['卖出数量']*data['卖出价格']).cumsum()*0.998-(data['买入数量']*data['买入价格']).cumsum()data['净值'] = (data['敞口']+5000)*data['close']+(data['实现盈利']+1000000)data['净值'] = data['净值'].round(2)data['净值'][-1]-data['净值'][0]# 输出289001.5
绘图查看净值变化、股价变化
# 用了一年的Tick级别数据回测，为方便展示做了数据降频df = data[['close','净值']].resample('1D').last()df = df.dropna()L_ = (# 换个可爱的风格 :DLine(init_opts=opts.InitOpts(theme=ThemeType.MACARONS, width='900px', height='500px')).add_xaxis(df.index.tolist()).add_yaxis("收盘价",df['close'].tolist(),yaxis_index=1).add_yaxis("净值",df['净值'].tolist()).extend_axis(yaxis=opts.AxisOpts(axislabel_opts=opts.LabelOpts(formatter="{value}元"))) # 添加一条蓝色的y轴.set_global_opts(yaxis_opts=opts.AxisOpts(is_scale=True),xaxis_opts=opts.AxisOpts(is_scale=True),datazoom_opts=[opts.DataZoomOpts(type_='inside')]))L_.render_notebook()
网格策略受单边行情影响严重、可以看出策略运行一段时间后就没有再发生交易、由于敞口原因与股价呈现同比波动。最后能得到正盈利主要是因为回测区间的整体涨跌幅度不大。
考虑到上述情况、可以采取按一定周期（每月、每周）平仓并重新开始策略的方案:
将之前的计算过程封装入函数
def grid_strategy(data):sell_price = data['close'][0]*1.01buy_price = data['close'][0]*0.99# 记录每日开盘持有股票、现金的初始值s_inday = 10000cash_inday = 10000000s = s_indayc = cash_indaysell = []buy = []for i in range(0, len(data)-1): if ((data['high'][i]>sell_price) & (s_inday > 200)):sell.append([data.index[i], sell_price.round(2), 200])s_inday -= 200s -= 200c += 200*data['close'][i]*0.998buy_price = sell_price *0.99sell_price = sell_price*1.01elif ((data['low'][i]<buy_price) & (cash_inday > data.close[i]*200)):buy.append([data.index[i], buy_price.round(2),200])cash_inday -= data.close[i]*200c -= data.close[i]*200s += 200sell_price = buy_price*1.01buy_price = buy_price*0.99if data['time'][i]==datetime.time(15, 0):s_inday =scash_inday =c# 循环结束后平仓if len(sell) - len(buy)> 0:buy.append([data.index[-1], data['close'][-1],(len(sell) - len(buy))*200])elif len(sell) - len(buy)< 0:sell.append([data.index[-1], data['close'][-1],(len(buy) - len(sell))*200])# 整理数据sell = pd.DataFrame(sell,columns = ['卖出时间','卖出价格', '卖出数量'])sell.index = sell['卖出时间']buy = pd.DataFrame(buy,columns = ['买入时间','买入价格', '买入数量'])buy.index = buy['买入时间']data = data.join(sell[['卖出价格', '卖出数量']])data = data.join(buy[['买入价格', '买入数量']])profit = (sell['卖出价格'] * sell['卖出数量']).sum() - (buy['买入价格'] * buy['买入数量']).sum()fee = (sell['卖出价格'] * sell['卖出数量']).sum() * 0.002net_profit = profit - feereturn data
重新整理数据、并打上月份标签
data = pd.read_csv('你的数据源.csv')data.index = data['tradetime']data['time'] = data.index.map(lambda x: x.time())data['month'] = data.index.map(lambda x: x.strftime('%Y.%m'))
利用循环跑出计算结果
也可以将盈亏计算过程写入函数、返回值改为区间盈亏
使用pandas 分组聚合的方式来计算:data.groupby('month').apply(grid_strategy)
results = pd.DataFrame()for month in data['month'].unique():df = data[(data['month']==month)]result = grid_strategy(df)results = pd.concat([results, result])
整理并画图、此处排除持仓影响、专注于净盈亏
data = results.fillna(0)data['敞口'] = (data['买入数量'].cumsum()-data['卖出数量'].cumsum())data['实现盈利'] = (data['卖出数量']*data['卖出价格']).cumsum()*0.998-(data['买入数量']*data['买入价格']).cumsum()data['净盈亏'] = (data['敞口'])*data['close']+(data['实现盈利'])data['净盈亏'] = data['净盈亏'].round(2)data['净盈亏'][-1]-data['净盈亏'][0]# 输出-65330.16df = data[['close','净盈亏']].resample('1D').last()df = df.dropna()L_ = (Line(init_opts=opts.InitOpts(theme=ThemeType.MACARONS, width='900px', height='500px')).add_xaxis(df.index.tolist()).add_yaxis("收盘价",df['close'].tolist(),yaxis_index=1).add_yaxis("净值",df['净盈亏'].tolist()).extend_axis(yaxis=opts.AxisOpts(axislabel_opts=opts.LabelOpts(formatter="{value}元"))) # 添加一条蓝色的y轴.set_global_opts(yaxis_opts=opts.AxisOpts(is_scale=True),xaxis_opts=opts.AxisOpts(is_scale=True),title_opts=opts.TitleOpts(title="网格交易策略--净盈亏、股价变化情况"),datazoom_opts=[opts.DataZoomOpts(type_='inside')]))L_.render_notebook()
可以看出震荡行情时表现良好、单边行情时不适用
最后附上DolphinDBAPI代码，选用逐笔成交聚合成3秒K线的方法
symbol = '688001'起始日期 = '.01.01'截止日期 = '.12.31'script = '''select tradetime, open, high, low, closefrom (select first(price) as open, max(price) as high, min(price) as low, last(price) as close, sum(volume) as volume from (select symbol, TradingTime, TradeVolume as volume, TradePrice as pricefrom loadTable("dfs://CSMAR_SEL2_TRANSACTION_H", 'data') where Symbol in ['{}'],tradingdate between {}:{},tradingtime.minute() between 09:30m:15:01m) group by symbol, dailyAlignedBar(TradingTime, 09:30:00.000, 3*1000) as Tradetime)'''.format(str(symbol), 起始日期, 截止日期)x = s.run(script)x.index =x['tradetime']data = x
如果觉得《【Python】 网格策略回测（日内高频数据）》对你有帮助，请点赞、收藏，并留下你的观点哦！
