跳转至

Python 实现多列滚动计算——以“更优波动率”为例

对单列数据进行滚动计算，可以使用常规的.rolling()。

如果需要对多列数据进行滚动计算，可以考虑下面两种方法：

  1. 引入外部包numpy_ext，使用其中的rollling_apply()方法。
  2. 在.rolling()中加入参数method='table'。

本文以方正金工发表的一篇研报中提出的计算“更优波动率”为例，实现了对多列数据进行滚动计算，并对上述两种方法总结如下：

  1. numpy_ext.rollling_apply()需要引入外部包numpy_ext，该方法接受需要进行滚动计算的多个 Series，并返回计算出的一个数组。
  2. .rolling(method='table')是 Pandas 内置的函数（需要升级到较新的版本），指定method='table'后，就可以对数据框中的多列进行滚动计算，并返回一个数据框。若返回的多列结果相同，我们只需要取出其中一列即可。
  3. .rolling(method='table')使用了engine='numba'，计算速度更快。

更优波动率的定义

“更优波动率”的改进之处在于，将原始由分钟收盘价来计算标准差，改为同时使用分钟开盘价、分钟最高价、分钟最低价和分钟收盘价来记录这一分钟的价格信息。

具体计算方法为：

第 t 分钟的“更优波动率”为，第 t-4、t-3、t-2、t-1、t 分钟的分钟开盘价、分钟最高价、分钟最低价和分钟收盘价，共 20 个价格数据求标准差，然后除以这 20 个价格数据的均值，最后将该比值取平方，作为 t 分钟的“更优波动率”。

导入包

Python
import pandas as pd
import numpy as np
import akshare as ak
from numpy_ext import rolling_apply

使用akshare获取数据

Python
# 获取分钟数据
df = ak.stock_zh_a_minute(symbol="sh000001", period="1")
# 将除'day'外的各列的字符串转换为数值
df = df.apply(pd.to_numeric, errors="ignore")

共获取 30000 条分钟频数据：

image-20230123000214570

使用numpy_ext计算更优波动率

定义计算更优波动率的函数

Python
# 计算更优波动率
def better_volatility_numpy_ext(o, h, l, c):
    # 将 4 列数据合并成一个数组
    data = np.array([o, h, l, c])
    # 计算 20 个数据的标准差
    std = np.std(data)
    # 计算 20 个数据的均值
    mean = np.mean(data)
    # 计算更优波动率
    return (std / mean) ** 2

这个函数接受 4 个 Series 作为参数，我们需要在函数内部将这 4 个 Series 拼接成一个数组，再按照更优波动率的计算逻辑，求出 20 个数据的标准差和均值，最后将两者相除再平方即可得到更优波动率。

应用numpy_ext.rollling_apply()

Python
better_volatility_using_numpy_ext = rolling_apply(
    better_volatility_numpy_ext, 5, df["open"], df["high"], df["low"], df["close"]
)

这里的window = 5，即滚动 5 分钟进行计算。

image-20230123000740853

耗时 7 秒，计算得到前 5 分钟的更优波动率为 1.56392876e-08。

使用pandas内置的.rolling(method='table')计算更优波动率

定义计算更优波动率的函数

Python
# 计算更优波动率
def better_volatility_pandas(data):
    # 计算 20 个数据的标准差
    std = np.std(data)
    # 计算 20 个数据的均值
    mean = np.mean(data)
    # 计算更优波动率
    return (std / mean) ** 2

与numpy_ext.rollling_apply()中使用的函数不同的是，这里接受的是一个数据框作为输入数据。

应用.rolling(method='table')

Python
better_volatility_using_pandas = (
    df[["open", "high", "low", "close"]]
    .rolling(5, method="table")
    .apply(better_volatility_pandas, raw=True, engine="numba")
)

这里的window = 5，即滚动 5 分钟进行计算。

image-20230123001228155

image-20230123001249308

耗时 3 秒，计算得到前 5 分钟的更优波动率也为 1.563929e-08。

这一条命令的输出是几列相同的数据（计算和输出的时候和data的形状是匹配的），而我们需要的只有一列，所以需要单独取出一列数据。

Python
better_volatility_using_pandas = better_volatility_using_pandas.iloc[:, 0].values

image-20230123001409649

手动验算结果

Python
# 手动验算，与上面的结果一致
std = df.iloc[:5][["open", "high", "low", "close"]].values.std()
mean = df.iloc[:5][["open", "high", "low", "close"]].values.mean()
better_volatility_using_numpy = (std / mean) ** 2
print(better_volatility_using_numpy)

image-20230123001450158

可以发现，两种计算方法的结果都是准确的。

总结

  1. numpy_ext.rollling_apply()需要引入外部包numpy_ext，该方法接受需要进行滚动计算的多个 Series，并返回计算出的一个数组。
  2. .rolling(method='table')是 Pandas 内置的函数（需要升级到较新的版本），指定method='table'后，就可以对数据框中的多列进行滚动计算，并返回一个数据框。若返回的多列结果相同，我们只需要取出其中一列即可。
  3. .rolling(method='table')使用了engine='numba'，计算速度更快。

评论