量化练手：基于北向资金的量化策略
量化练手
最近在学习量化，在使用各种量化框架之前，试着不基于其他框架，只基于pandas和 matplotlib，编写一个量化策略，做量化学习的练手，同时也是对不熟悉的python做个练手。
量化策略
参考北向资金的每日净流入情况，决定是否买入沪深300指数。
具体策略如下：
1、获得北向资金每日净流入数，北向资金净流入为 沪港通与深港通净流入数据之和。
2、生成北向资金每日净流入资金的布林线；（此处采用100天均值，1.5倍标准差）
3、如果当日净流入高于布林线，则认为买入沪深300指数，次日按开盘价买入90%的仓位，如果连续高出，只买入一次；
4、如果当日净流入低于布林线，则认为卖出，次日按开盘价清仓。
5、交易手续费为万分之二；
6、参考一些指数基本的收费规则，如果持有天数低于7天，卖出时手续费会很高，我设置了trade_interval，默认为7天，也就是至少持仓7天，才会卖出。
数据来源
北向资金数据与指数数据来爬取自tt_fund。
beixiang_20211001.csv
market_index_20211001.csv
结果分析
北向资金布林线情况：
策略运行结果， 其中benchMark为沪深指数蓝线，红线为策略的运行结果。
从 2016-12-07开始至2021-10-01，总共执行买操作25次，卖操作25次.
10000的起初资金，最终市值为16436.075215， 近5年 64%的收益，高于沪深300 40%的收益。
详细代码
github地址 https://github.com/quentinxxz/QuantLib/tree/main/beixiang
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Created on Tue Sep 21 15:14:53 2021
@author: quentinxxz
"""
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Created on Sun Sep 19 20:42:39 2021
@author: quentinxxz
"""
import pandas as pd
import datetime
import matplotlib.pyplot as plt
# 数据准备处理
def prepareBeixiangData():
#爬下来的北向资金数据
df = pd.read_csv('beixiang_20211001.csv')
df['datetime']= pd.to_datetime(df['datetime'] ,format="%Y-%m-%d")
# 1表示沪股通， 3为深股通, 4为
df_1 = df[df['marketType'] == 1]
df_3 = df[df['marketType'] == 3]
df = pd.merge(df_1,df_3,suffixes=('_1', '_3'),how='left',on='datetime' )
# 合并沪股通与深股通数据
df['total_in']=df['total_in_1']+df['total_in_3']
df['total_out']=df['total_out_1']+df['total_out_3']
df['total_net_in']=df['total_net_in_1']+df['total_net_in_3']
df['grand_total_in']=df['grand_total_in_1']+df['grand_total_in_3']
df['today_balance']=df['today_balance_1']+ df['today_balance_3']
df=df.loc[:,['datetime','total_in','total_out','total_net_in','grand_total_in','today_balance']]
# 按日期排序
df.set_index('datetime',drop=False,append=False,inplace=True)
df.sort_index(ascending=True,inplace=True)
#df.sort_values(by='datetime',ascending=True,inplace= True)
# 计算布林线
df['std'] = df['total_net_in'].rolling(100).std()
df['mean'] = df['total_net_in'].rolling(100).mean()
df['lower_interval'] = df['mean'] - 1.5*df['std']
df['upper_interval'] = df['mean'] + 1.5*df['std']
# 选取一段数据做展示
#s_date = datetime.datetime.strptime('20190401', '%Y%m%d')
#e_date = datetime.datetime.strptime('20200508', '%Y%m%d')
# d_t = df[(df['datetime']<e_date) & (df['datetime']>s_date) ]
# d_t.plot.line(x='datetime',y=['total_net_in','mean','lower_interval','upper_interval'])
return df
# 准备沪深300指数
def prepareShangZheng():
df= pd.read_csv('market_index_20211001.csv')
df['datetime']= pd.to_datetime(df['datetime'])
df.set_index('datetime',drop=False,append=False,inplace=True)
df=df[df['code']==300]
df.sort_index(ascending=True,inplace=True)
return df
# 计算买卖时机，deprecated
def updateBuyAndSell(bx):
bx.loc[bx['lower_interval']>bx['total_net_in'], 'trade']=1
bx.loc[bx['upper_interval']<bx['total_net_in'], 'trade']=-1
# 交易回测
# bx为北向资金数据，
# benchmark为 回测参考 本例中采用上证指烽，
# cost_ration为手续费比例，默认万二，
# init_cap为初始资金，
# delay代表 基于北向资金分析数据，几日后实施交易，至少为1天
def backTrade(bx,benchmark,cost_ration=0.0002,init_cap=10000,delay=1,trade_interval=7):
count=0
trade_records = pd.DataFrame(columns=['operation','trade_val','trade_vol','acc_val_end','acc_vol_end','total_value','capital','order_cost'],index=['datetime'])
for pos in range(len(bx)) :
#取对应的交易日
next_trade_pos = pos+ delay
if next_trade_pos >= len(bx):
break
trade_day = bx.iloc[next_trade_pos]['datetime']
# 如果北向资金净流入，低于布林线，做卖出操作 operation -1代表卖出
if pd.isnull(bx.iloc[pos]['lower_interval'])==False and bx.iloc[pos]['lower_interval']> bx.iloc[pos]['total_net_in']:
record=pd.DataFrame({'operation':-1},index=[trade_day])
count=count+1
# 如果北向资金净流入，高于布林线，做买入操作 operation 1代表买入
elif pd.isnull(bx.iloc[pos]['upper_interval'])==False and bx.iloc[pos]['upper_interval'] < bx.iloc[pos]['total_net_in']:
count=count+1
record=pd.DataFrame({'operation':1},index=[trade_day])
# 其他情况，不做操作
else:
record=pd.DataFrame({'operation':0},index=[trade_day])
trade_records= pd.concat([trade_records, record])
# 删除首行
trade_records.drop(['datetime'],inplace=True)
# 初始时持币
holdCash =True
capital = init_cap
acc_vol_end =0
acc_val_end =0
total_val = capital
sellCount =0
buyCount = 0
delaySell=False
for index ,row in trade_records.iterrows():
if row['operation'] ==1:
delaySell=False
# 持币且operation 为1时，买入
if row['operation'] ==1 and holdCash==True:
#按开盘价买入，收盘价计算价格，每次交易90%本金,
trade_val = capital*0.9 # 交易额
trade_vol = trade_val/benchmark.loc[index,'price_end'] # 交易量
order_cost = trade_val * cost_ration # 交易手续费
capital = capital-trade_val-order_cost #剩余本金
acc_vol_end = acc_vol_end + trade_vol # 收盘账户持仓数量
acc_val_end = acc_vol_end * benchmark.loc[index,'price_end'] # 收盘账户持仓金额
total_val= acc_val_end + capital # 收盘总金额
print( "buy, date:%s, trade_val：%.2f, acc_val_end:%.2f total_val:%.2f"%(index, trade_val,acc_val_end,total_val))
record=pd.DataFrame({'operation':1,
'trade_val':trade_val,
'trade_vol': trade_vol,
'acc_vol_end': acc_vol_end,
'acc_val_end': acc_val_end,
'total_value':total_val,
'capital':capital,
'order_cost':order_cost},
index=[index])
trade_records.update(record)
holdCash =False
buyCount=buyCount+1
buyDay= index #纪录购买时
elif (delaySell or row['operation'] == -1) and holdCash== False:
if index-buyDay< datetime.timedelta(days=trade_interval) :
print('delaySell,buyDay:%s, currentDay:%s'%(buyDay,index))
delaySell =True
acc_val_end = acc_vol_end * benchmark.loc[index,'price_end'] # 按收盘计算
total_val= acc_val_end + capital
record=pd.DataFrame({'operation':0,
'trade_val':0,
'trade_vol': 0,
'acc_vol_end': acc_vol_end ,
'acc_val_end': acc_val_end,
'total_value':total_val,
'capital':capital,
'order_cost':0},
index=[index])
trade_records.update(record)
else :
#按收盘价卖出，收盘价计算金额，卖出则卖空
trade_val = acc_vol_end *benchmark.loc[index,'price_end'] # 按开盘价卖出全部
trade_vol = acc_val_end
order_cost = trade_val * cost_ration
capital =capital+trade_val-order_cost
acc_vol_end = acc_val_end- trade_vol
acc_val_end = acc_vol_end * benchmark.loc[index,'price_end'] # 按收盘计算
total_val= acc_val_end + capital
record=pd.DataFrame({'operation':-1,
'trade_val':trade_val,
'trade_vol': trade_vol,
'acc_vol_end': acc_vol_end,
'acc_val_end': acc_val_end,
'total_value':total_val,
'capital':capital,
'order_cost':order_cost},
index=[index])
print( "sell, date:%s, trade_val：%.2f, acc_val_end:%.2f total_val:%.2f"%(index, trade_val,acc_val_end,total_val))
trade_records.update(record)
holdCash= True
sellCount=sellCount+1
else :
acc_val_end = acc_vol_end * benchmark.loc[index,'price_end'] # 按收盘计算
total_val= acc_val_end + capital
record=pd.DataFrame({'operation':0,
'trade_val':0,
'trade_vol': 0,
'acc_vol_end': acc_vol_end ,
'acc_val_end': acc_val_end,
'total_value':total_val,
'capital':capital,
'order_cost':0},
index=[index])
trade_records.update(record)
#print("common_day, date:%s,acc_val_end:%.2f,total_val: %.2f "%(index,acc_val_end,total_val))
print("sellCount:%d,buy_count:%d "%(sellCount,buyCount))
return trade_records
# 获取北向数据
bx= prepareBeixiangData()
# 绘制布林线
#bx.plot(x='datetime',y=['total_net_in','lower_interval','upper_interval','mean'] )
# 获取上证指数为参考
bm= prepareShangZheng()
# 进行回测
records= backTrade(bx,bm,delay=0)
#开始比较时间
s_date = datetime.datetime.strptime('20161206', '%Y%m%d')
bm_selected =bm[bm['datetime']>s_date]
# 将参考指数归一化，初始为10000
norm_bm_price = bm_selected['price_end']/ bm_selected.iloc[0]['price_end']*10000
# 将操作纪录归一化，初始为10000
records_select = records[records.index>s_date]
norm_records_total_val = records_select['total_value']/ records_select.iloc[0]['total_value']*10000
norm_bm_price.plot.line(subplots= True,x='datetime',y='price_end',style='b',label='benchMark')
norm_records_total_val.plot.line(subplots=True,style='r',label='beixiang')
plt.grid(True)
plt.show()参考资料
《年化46%的北向资金+20日涨幅的创业板策略》https://www.joinquant.com/view/community/detail/a5fb2c8b9644ea49441cfcfffb735f13
20210101首发于3dobe.com
本站链接：http://3dobe.com/archives/263/