熟悉我的老铁都知道，我的本职工作是金融搬砖客（banker的正确翻译，嗯），业余爱好是编程，这两项的交集很容易联想到量化投资。
果不其然，我又要跟自己较劲了。
简单谈一下我对量化投资的看法：
- （1）不要误解量化投资就是机器自动投资，在下单之前有大量的建模、参数调优、回测，甚至读paper的人类工作要做；
- （2）不要神化量化投资，不存在一劳永逸的投资模型，而且资金规模与模型有效性往往成反比；
- （3）要有保留地相信回测结果，因为令人满意的回测结果都是基于部分历史样本的过拟合，面对VUCA时代的市场变化，我们永远面临样本不足的问题；
- （4）可以利用量化策略来执行“反人性”的操作，通过统计规律+投资纪律获得优于市场β的超额收益。
说干就干，在研究了多本教材之后，我找到了量化投资学习方面的路线图：
- （1）从股票指数的择时模型入手，千万不要误会择时容易成功，只是因为择时策略学习门槛低，“低买高卖”的思路简单清晰，各类技术指标成熟容易捕捉。
- （2）择时模型成熟后可以直接应用于股票和期货的调仓模型，同时大量写代码、写注释，补充统计学知识。
- （3）再深入就是股票择股模型，做足单因子分析功课，然后切换到多因子学习，以期发现高质量的Alpha收益因子。
今天是系列开篇，我们就来见识一下股指择时模型中二八轮动策略的威力。
01 二八轮动策略建模
所谓二八轮动，指的是大盘股与小盘股在不同时期的投资回报率有显著差异，比如2019年开始的白马股行情、2021年春季后开始的创业板行情，以及2022年下半年开始的中证1000行情。
根据这一事实，通过动量指标找到某一阶段上涨领先的股指，选择买入，如果该股指动量被另一股指超越则换仓，如果所有股指均变为下跌态势则清仓买入国债过冬。
中证指数25日环比变动实际应用时，我选择的是沪深300ETF和中证1000ETF进行轮动，使用baostock库获取指数的日线数据，动量计算窗口是25天，具体Python代码我放在文末。
02 二八轮动策略回测
这一策略效果如何呢？我在聚宽平台上对模型在2016年至今的时间区间内进行了回测，结果如下：
沪深300&中证1000轮动回测结果通过观察至少可以得到三个结论：
- （1）策略收益年化收益 6.97%（如果考虑清仓期间的国债收益总体超 8% 问题不大），较沪深300指数超额收益 37.65%，夏普比率 0.185，最大回撤 23.50%，算不上优秀，但跑赢指数不成问题。
- （2）策略优点是可以避开单边下跌行情，很明显的一段是22年以来的下跌。
- （3）策略缺点是无法在指数深跌反弹时快速上车，后期可以增加超涨买入指标。
03 二八轮动策略应用
策略有了，如何应用呢？我决定将现在的指数ETF定投策略改为二八轮动策略。根据指标最新提示，当下应该买入并持有中证1000ETF。
在今后的文章中，我会将轮动的最新提示放下末尾，供各位读者参考。即便不拿来做投资决策，你也可以通过图形一眼看出市场风格走向。
欢迎关注！
附：策略Python代码
import numpy as np
import pandas as pd
import baostock as bs
from datetime import date, timedelta
import matplotlib.pyplot as plt
plt.rcParams['axes.unicode_minus'] = False
plt.rcParams['font.sans-serif']=['simhei']
yesterday = (date.today() + timedelta(days=-1)).strftime('%Y-%m-%d')
lastyear = (date.today() + timedelta(days=-365)).strftime('%Y-%m-%d')
def bs_k_data(code_val, start_date, end_date, frequency, adjustflag):
# 登录baostock
lg = bs.login()
print('login respond error_code:' + lg.error_code)
print('login respond error_msg :' + lg.error_msg)
# 获取日线行情
fields = 'date,open,high,low,close,volume,pctChg'
df_bs = bs.query_history_k_data_plus(code_val, fields,
start_date=start_date, end_date=end_date,
frequency=frequency, adjustflag=adjustflag) #adjustflag: 3:默认不复权，1:后复权，2:前复权
print('query_history_k_data_plus respond error_code:' + df_bs.error_code)
print('query_history_k_data_plus respond error_msg :' + df_bs.error_msg)
# query_history_k_data返回的是可迭代对象，因此需要调用get_row_data方法将每行数据汇总为列表，再转换为DataFrame格式
data_list = []
while (df_bs.error_code == '0') & df_bs.next():
data_list.append(df_bs.get_row_data())
result = pd.DataFrame(data_list, columns=df_bs.fields)
result.close = result.close.astype('float64')
result.open = result.open.astype('float64')
result.low = result.low.astype('float64')
result.high = result.high.astype('float64')
result.volume = result.volume.astype('float64')/100
result.date = pd.DatetimeIndex(result.date)
result.set_index('date', drop=True, inplace=True)
result.index = result.index.set_names('Date')
result_data = {'Open':result.open, 'Close':result.close,
'High':result.high, 'Low':result.low,
'Volume':result.volume, 'pctChg':result.pctChg}
df_daily_stock = pd.DataFrame(result_data)
df_daily_stock['diff'] = df_daily_stock['Close'].diff()
df_daily_stock['signal'] = np.where(df_daily_stock['diff'] > 0, 0, 1)
df_daily_stock = df_daily_stock.round(2)
bs.logout()
return df_daily_stock
def index_singal(start_date=lastyear, end_date=yesterday):
# 获取指数日线
df_sh300 = bs_k_data('sh.000300', start_date, end_date,'d','3')
df_sh500 = bs_k_data('sh.000905', start_date, end_date,'d','3')
df_sh1000 = bs_k_data('sh.000852', start_date, end_date,'d','3')
# 计算增长动量
df_all = pd.DataFrame()
df_all['hs300'] = df_sh300['Close']
df_all['zz500'] = df_sh500['Close']
df_all['zz1000'] = df_sh1000['Close']
df_all['hs300_25d'] = df_sh300['Close'].shift(25)
df_all['zz500_25d'] = df_sh500['Close'].shift(25)
df_all['zz1000_25d'] = df_sh1000['Close'].shift(25)
df_all['hs300_inc'] = (df_all['hs300'] - df_all['hs300_25d']) / df_all['hs300_25d']
df_all['zz500_inc'] = (df_all['zz500'] - df_all['zz500_25d']) / df_all['zz500_25d']
df_all['zz1000_inc'] = (df_all['zz1000'] - df_all['zz1000_25d']) / df_all['zz1000_25d']
print(df_all.tail(10))
# print(df_all.corr())
# 作图演示
plt.figure(figsize=(12, 6), dpi=100)
plt.plot(df_all['hs300_inc'], color='navy', label='沪深300')
plt.plot(df_all['zz500_inc'], color='c', label='中证500')
plt.plot(df_all['zz1000_inc'], color='orange', label='中证1000')
plt.axhline(0, c='grey', ls='--')
plt.legend(loc='best')
plt.title('中证指数环比25日变动')
plt.show()
# 判断买入卖出
if (df_all['hs300_inc'][-1] <= 0 and df_all['zz500_inc'][-1] <= 0 and df_all['zz1000_inc'][-1] <= 0):
return 'sell_all_securities'
elif df_all['hs300_inc'][-1] - df_all['zz1000_inc'][-1] > 0.005:
return 'buy_ETF300'
elif df_all['zz1000_inc'][-1] - df_all['hs300_inc'][-1] > 0.005:
return 'buy_ETF1000'
index_singal()
