我有一个时间序列“Ser”，想要使用滚动窗口计算波动率（标准差）。我的当前代码在这种形式下可以正确执行：
w = 10
for timestep in range(length):
subSer = Ser[timestep:timestep + w]
mean_i = np.mean(subSer)
vol_i = (np.sum((subSer - mean_i)**2) / len(subSer))**0.5
volList.append(w_i)
这对我来说似乎非常低效。Pandas是否具有内置功能来执行此类操作？
通常，[金融类] 的人会以价格百分比年化波动率的形式引用。
假设您在一个数据框架 df 中有每日价格，并且一年有252个交易日，以下内容可能是您想要的：
df.pct_change().rolling(window_size).std()*(252**0.5)
vol_year = vol_day * sqrt(252)一年被假定为252个交易日。约定俗成。 - jfaleiroddof = 0来计算标准差，就像上面的答案一样？谢谢。 - Rocky the Owl看起来你正在寻找 Series.rolling。你可以对结果对象应用std计算：
roller = Ser.rolling(w)
volList = roller.std(ddof=0)
如果您不打算再次使用rolling window对象，您可以编写一个一行代码的简短语句：
volList = Ser.rolling(w).std(ddof=0)
请记住，在这种情况下ddof=0是必要的，因为标准差的规范化是通过len(Ser)-ddof来计算的，并且在pandas中默认ddof为1。
ddof 是什么了！ :) - Jonas Byström"波动性"即使在金融领域也是一个含义模糊的词。最常被提及的波动类型是实现波动率，它是实现方差的平方根。与收益标准差相比，其关键区别在于：
有多种计算实现波动率的方法；然而，我下面实现的是两种最为常见的方法：
import numpy as np
window = 21 # trading days in rolling window
dpy = 252 # trading days per year
ann_factor = days_per_year / window
df['log_rtn'] = np.log(df['price']).diff()
# Var Swap (returns are not demeaned)
df['real_var'] = np.square(df['log_rtn']).rolling(window).sum() * ann_factor
df['real_vol'] = np.sqrt(df['real_var'])
# Classical (returns are demeaned, dof=1)
df['real_var'] = df['log_rtn'].rolling(window).var() * ann_factor
df['real_vol'] = np.sqrt(df['real_var'])
var() default is ddof=0, so you could probably just replace the first example with var() and second with var(ddof=1) - Baczek这里是一种使用NumPy的方法-
# From https://dev59.com/IWYq5IYBdhLWcg3wpyNE#14314054 by @Jaime
def moving_average(a, n=3) :
ret = np.cumsum(a, dtype=float)
ret[n:] = ret[n:] - ret[:-n]
return ret[n - 1:] / n
# From https://dev59.com/7FkS5IYBdhLWcg3wSk3b#40085052
def strided_app(a, L, S=1 ): # Window len = L, Stride len/stepsize = S
nrows = ((a.size-L)//S)+1
n = a.strides[0]
return np.lib.stride_tricks.as_strided(a, shape=(nrows,L), strides=(S*n,n))
def rolling_meansqdiff_numpy(a, w):
A = strided_app(a, w)
B = moving_average(a,w)
subs = A-B[:,None]
sums = np.einsum('ij,ij->i',subs,subs)
return (sums/w)**0.5
示例运行 -
In [202]: Ser = pd.Series(np.random.randint(0,9,(20)))
In [203]: rolling_meansqdiff_loopy(Ser, w=10)
Out[203]:
[2.6095976701399777,
2.3000000000000003,
2.118962010041709,
2.022374841615669,
1.746424919657298,
1.7916472867168918,
1.3000000000000003,
1.7776388834631178,
1.6852299546352716,
1.6881943016134133,
1.7578395831246945]
In [204]: rolling_meansqdiff_numpy(Ser.values, w=10)
Out[204]:
array([ 2.60959767, 2.3 , 2.11896201, 2.02237484, 1.74642492,
1.79164729, 1.3 , 1.77763888, 1.68522995, 1.6881943 ,
1.75783958])
运行时测试
循环方法 -
def rolling_meansqdiff_loopy(Ser, w):
length = Ser.shape[0]- w + 1
volList= []
for timestep in range(length):
subSer=Ser[timestep:timestep+w]
mean_i=np.mean(subSer)
vol_i=(np.sum((subSer-mean_i)**2)/len(subSer))**0.5
volList.append(vol_i)
return volList
时间 -
In [223]: Ser = pd.Series(np.random.randint(0,9,(10000)))
In [224]: %timeit rolling_meansqdiff_loopy(Ser, w=10)
1 loops, best of 3: 2.63 s per loop
# @Mad Physicist's vectorized soln
In [225]: %timeit Ser.rolling(10).std(ddof=0)
1000 loops, best of 3: 380 µs per loop
In [226]: %timeit rolling_meansqdiff_numpy(Ser.values, w=10)
1000 loops, best of 3: 393 µs per loop
这两种向量化方法相对于循环方法的加速比接近7000倍!