28  滚动测试范例

这里做一个滚动测试的范例。

Code 
# 以下都是例行公事，直接拷贝即可

import pandas as pd
import numpy as np 

# 导入matplotlib.pyplot绘图库，其中plt.plot()是最常用的绘图函数之一
import matplotlib.pyplot as plt 

import seaborn as sns

sns.set_theme() # 默认用seaborn的绘图样式

plt.rcParams["font.sans-serif"]=["Hiragino Sans GB", "Microsoft YaHei", "WenQuanYi Micro Hei"] #设置字体。如果不设置，中文可能会乱码。这里采用冬青黑、微软雅黑和文泉驿微米黑，一般情况都没问题。
plt.rcParams["axes.unicode_minus"]=False #该语句解决图像中的“-”负号的乱码问题

# 绘图使用'svg'后端：svg是矢量格式，可以任意缩放均保持清晰，各种屏幕的显示良好。
%config InlineBackend.figure_formats = ['svg']

28.1 滚动测试

滚动测试：向后看一段时间，找出模型在过去一段时间的最优参数，利用这个参数，向前做一段时间的交易。

例如：向后看3个月（黄色格），向前交易2个月（绿色格）

  1. 测试1、2、3月，找出这3个月的最优参数，用来执行4、5月的交易。
  2. 测试3、4、5月，找出这3个月的最优参数，用来执行6、7月的交易。
  3. 如此类推，最后把4、5、6、7 …的交易结果组合起来，就是最终的滚动测试的结果。

28.2 需求

  1. 用单均线系统，进行滚动测试。
  2. 最优参数（均线周期），由最高的夏普比率决定。
  3. 向后和向前的时间是整月，数量由外部决定。
import akshare as ak

计算移动平均和夏普比率的函数

def sma(x,period):
    return x.rolling(period).mean()

def sharpe_ratio(lrets, rf = 0.03):
    return (np.mean(lrets) * 250 - rf) / (np.std(lrets) * np.sqrt(250))

28.3 简单数据处理

# 读取宁德时代
# 见：https://akshare.xyz/data/stock/stock.html#id20

df = ak.stock_zh_a_hist(symbol="300750",start_date='20180101',end_date='20241212', adjust='qfq')

df
日期 股票代码 开盘 收盘 最高 最低 成交量 成交额 振幅 涨跌幅 涨跌额 换手率
0 2018-06-11 300750 9.64 12.99 12.99 9.64 788 2.845471e+06 48.98 89.91 6.15 0.04
1 2018-06-12 300750 15.00 15.00 15.00 15.00 266 1.058375e+06 0.00 15.47 2.01 0.01
2 2018-06-13 300750 17.21 17.21 17.21 17.21 450 1.972314e+06 0.00 14.73 2.21 0.02
3 2018-06-14 300750 19.64 19.64 19.64 19.64 743 3.578184e+06 0.00 14.12 2.43 0.03
4 2018-06-15 300750 22.32 22.32 22.32 22.32 2565 1.359503e+07 0.00 13.65 2.68 0.12
... ... ... ... ... ... ... ... ... ... ... ... ...
1576 2024-12-06 300750 256.00 266.55 269.90 255.00 330381 8.719648e+09 5.82 4.12 10.55 0.85
1577 2024-12-09 300750 266.99 268.05 269.66 264.41 224561 6.005103e+09 1.97 0.56 1.50 0.58
1578 2024-12-10 300750 281.28 270.57 281.50 269.33 329044 9.035442e+09 4.54 0.94 2.52 0.84
1579 2024-12-11 300750 273.96 267.10 279.50 266.52 194008 5.262015e+09 4.80 -1.28 -3.47 0.50
1580 2024-12-12 300750 265.90 273.71 275.38 265.20 223295 6.055502e+09 3.81 2.47 6.61 0.57

1581 rows × 12 columns

df['日期'] = pd.to_datetime(df['日期'])
df.set_index('日期',inplace = True)
df = df[['涨跌幅']]/100
df.fillna(0,inplace=True)
df['price'] = (df['涨跌幅'] + 1).cumprod() # 用涨跌幅反算出复权价，上市日前一天为1。
df['lrets'] = np.log(df['涨跌幅'] + 1)
df.tail()
涨跌幅 price lrets
日期
2024-12-06 0.0412 38.989447 0.040374
2024-12-09 0.0056 39.207788 0.005584
2024-12-10 0.0094 39.576342 0.009356
2024-12-11 -0.0128 39.069764 -0.012883
2024-12-12 0.0247 40.034788 0.024400

28.4 构造滚动窗口

需求：构造出回看月份，和测试月份的序列

例如：回看[4,5,6]月，交易[7,8]月；回看[6,7,8]月，交易[9,10]月，那么

train_time = [[4,5,6],[6,7,8]]
test_time = [[7,8],[9,10]]

相同的索引i，就可以取出回看区间和交易区间。

# 获得所有月份

all_months = df.resample('M').last().index.strftime('%Y-%m')
all_months
/tmp/ipykernel_2262/2225266909.py:3: FutureWarning: 'M' is deprecated and will be removed in a future version, please use 'ME' instead.
  all_months = df.resample('M').last().index.strftime('%Y-%m')
Index(['2018-06', '2018-07', '2018-08', '2018-09', '2018-10', '2018-11',
       '2018-12', '2019-01', '2019-02', '2019-03', '2019-04', '2019-05',
       '2019-06', '2019-07', '2019-08', '2019-09', '2019-10', '2019-11',
       '2019-12', '2020-01', '2020-02', '2020-03', '2020-04', '2020-05',
       '2020-06', '2020-07', '2020-08', '2020-09', '2020-10', '2020-11',
       '2020-12', '2021-01', '2021-02', '2021-03', '2021-04', '2021-05',
       '2021-06', '2021-07', '2021-08', '2021-09', '2021-10', '2021-11',
       '2021-12', '2022-01', '2022-02', '2022-03', '2022-04', '2022-05',
       '2022-06', '2022-07', '2022-08', '2022-09', '2022-10', '2022-11',
       '2022-12', '2023-01', '2023-02', '2023-03', '2023-04', '2023-05',
       '2023-06', '2023-07', '2023-08', '2023-09', '2023-10', '2023-11',
       '2023-12', '2024-01', '2024-02', '2024-03', '2024-04', '2024-05',
       '2024-06', '2024-07', '2024-08', '2024-09', '2024-10', '2024-11',
       '2024-12'],
      dtype='object', name='日期')
train_time = [] # 回看的月份
test_time = [] # 前向交易的月份

# 回看周期
back_n = 2

# 前向交易周期
test_n = 1

# 遍历所有月份
# 把第0月到back_n-1月，放入train_time
# 把back_n月到back_n + test_n月，放入test_time
for i in range(0,len(all_months) - (back_n + test_n) + 1,test_n):
    train_time.append(all_months[i:(i+back_n)])
    test_time.append(all_months[i+back_n:(i+back_n+test_n)])

print(train_time[0])
print(test_time[0])
print(train_time[1])
print(test_time[1])
Index(['2018-06', '2018-07'], dtype='object', name='日期')
Index(['2018-08'], dtype='object', name='日期')
Index(['2018-07', '2018-08'], dtype='object', name='日期')
Index(['2018-09'], dtype='object', name='日期')

28.5 构造一个测试窗口的最优化函数

需求：给出测试的时间段，问哪个均线的夏普比率最高？

思路：

  1. 暴力测试全部均线
  2. 返回均线周期，和对应时间段的夏普比率
# 对于指定的起点和终点，找出最佳的均线

def get_best_ma(start, end):
    
    # 暴力优化，穷举60,70,...250的所有周期
    ma_period_list = range(60,250,10)

    result = []

    for ma_period in ma_period_list:
        
        # 单均线交易系统，很简单
        df['ma'] = sma(df.price, ma_period)
        df['pos'] = np.where(df.price > df.ma,1,0) # 只允许做多
        df['traded'] = df['lrets'] * df.pos.shift() 

        # 计算测试区间[start:end]的夏普比率
        result.append((ma_period, sharpe_ratio(df.loc[start:end].traded)))

    res = pd.DataFrame(result)
    res.columns = ['ma_period','sharpe']
    return res

get_best_ma('2021-05','2021-11').sort_values(['sharpe','ma_period'],ascending=False).head()
ma_period sharpe
18 240 2.026973
17 230 2.026973
16 220 2.026973
15 210 2.026973
14 200 2.026973

28.6 构造一个窗口测试函数

需求：对于一个测试窗口i，包括训练集（回看最优）和测试集（前向交易），计算前向交易的结果

def test_a_window(i):
    """测试一个窗口i。
    例如，寻找7月和8月的最佳均线，用来执行9月的交易，返回9月的交易结果。
    """

    # 前面算好的测试用窗口i，取起点和终点的时间
    start = train_time[i][0]
    end = train_time[i][-1]
    # 问，这段时间内，什么均线的参数最优？
    res = get_best_ma(train_time[i][0],train_time[i][-1]
                      ).sort_values(['sharpe','ma_period'],ascending=False).iloc[0]
    
    #print(f"在{start}到{end}之间，最佳的均线是{int(res['ma_period'])}，区间年化夏普比率是{round(res['sharpe'],4)}。")

    # 利用上述最优的参数，我们进行交易
    ma_period = int(res['ma_period'])
    df['ma'] = sma(df.price, ma_period)
    df['pos'] = np.where(df.price > df.ma,1,0) 
    df['traded'] = df['lrets'] * df.pos.shift() 
    #print(f"使用上述设置，在{list(test_time[i])}内的对数收益率是{round(df.loc[test_time[i][0]:test_time[i][-1]].traded.sum(),4)}")

    # 返回测试集对应时间段的持仓情况。
    return df['pos'].loc[test_time[i][0]:test_time[i][-1]]

test_a_window(4).head()
/tmp/ipykernel_2262/2749032695.py:5: RuntimeWarning: divide by zero encountered in scalar divide
  return (np.mean(lrets) * 250 - rf) / (np.std(lrets) * np.sqrt(250))
日期
2018-12-03    1
2018-12-04    1
2018-12-05    1
2018-12-06    1
2018-12-07    1
Name: pos, dtype: int64

28.7 测试所有的窗口

最后把所有滚动交易的结果组合在一起

pos_final = []
for i in range(len(train_time)):
    pos_final.append(test_a_window(i))

pos_final = pd.concat(pos_final)
/tmp/ipykernel_2262/2749032695.py:5: RuntimeWarning: divide by zero encountered in scalar divide
  return (np.mean(lrets) * 250 - rf) / (np.std(lrets) * np.sqrt(250))

28.8 结果

df.loc[:,'strategy'] = df.lrets * pos_final.shift() # 注意不要作弊
df[['lrets','strategy']].loc['2020':].cumsum().apply(np.exp).plot();

计算一些比率

df[['lrets','strategy']].loc['2020':].agg("sum")
lrets       1.659712
strategy    1.944169
dtype: float64
df[['lrets','strategy']].loc['2020':].agg(sharpe_ratio)
lrets       0.638204
strategy    0.884721
dtype: float64