python - 计算*滚动* Pandas 系列的最大回撤

Question

编写一个计算时间序列最大回撤的函数非常容易。用 O(n) 时间而不是 O(n^2) 时间编写它需要一点思考。但这并不是那么糟糕。这将起作用:

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt

def max_dd(ser):
    max2here = pd.expanding_max(ser)
    dd2here = ser - max2here
    return dd2here.min()

让我们建立一个简短的系列来尝试一下:

np.random.seed(0)
n = 100
s = pd.Series(np.random.randn(n).cumsum())
s.plot()
plt.show()

正如预期的那样，max_dd(s) 最终显示在 -17.6 左右。好，好，大。现在说我有兴趣计算这个系列的滚动回撤。 IE。对于每一步，我想从指定长度的前一个子系列中计算最大回撤。使用 pd.rolling_apply 很容易做到这一点。它是这样工作的:

rolling_dd = pd.rolling_apply(s, 10, max_dd, min_periods=0)
df = pd.concat([s, rolling_dd], axis=1)
df.columns = ['s', 'rol_dd_10']
df.plot()

这非常有效。但是感觉很慢。 pandas 或其他工具包中是否有特别灵活的算法可以快速执行此操作？我尝试编写一些定制的东西:它跟踪各种中间数据(观察到的最大值的位置、先前发现的回撤位置)以减少大量冗余计算。它确实节省了一些时间，但不是很多，也不是尽可能多的。

我认为这是因为 Python/Numpy/Pandas 中的所有循环开销。但是我目前在 Cython 中的流利程度还不够，无法真正知道如何从那个角度开始攻击它。我希望有人以前试过这个。或者，也许有人可能想看看我的“手工”代码并愿意帮助我将其转换为 Cython。

---

编辑:对于任何想要查看这里提到的所有功能(以及其他一些功能!)的人，请查看 iPython 笔记本:http://nbviewer.ipython.org/gist/8one6/8506455

它展示了解决这个问题的一些方法是如何关联的，检查它们是否给出了相同的结果，并展示了它们在不同大小的数据上的运行时间。

如果有人感兴趣，我在帖子中提到的“定制”算法是 rolling_dd_custom。我认为如果在 Cython 中实现，这可能是一个非常快速的解决方案。

Answer 1

这是滚动最大回撤函数的 numpy 版本。 windowed_view 是一个单行函数的包装器，它使用 numpy.lib.stride_tricks.as_strided 来制作 1d 数组的内存高效 2d 窗口 View (完整代码如下) .一旦我们有了这个窗口 View ，计算基本上与您的 max_dd 相同，但为 numpy 数组编写，并沿第二个轴应用(即 axis=1) .

def rolling_max_dd(x, window_size, min_periods=1):
    """Compute the rolling maximum drawdown of `x`.

    `x` must be a 1d numpy array.
    `min_periods` should satisfy `1 <= min_periods <= window_size`.

    Returns an 1d array with length `len(x) - min_periods + 1`.
    """
    if min_periods < window_size:
        pad = np.empty(window_size - min_periods)
        pad.fill(x[0])
        x = np.concatenate((pad, x))
    y = windowed_view(x, window_size)
    running_max_y = np.maximum.accumulate(y, axis=1)
    dd = y - running_max_y
    return dd.min(axis=1)

这是一个完整的演示函数的脚本:

import numpy as np
from numpy.lib.stride_tricks import as_strided
import pandas as pd
import matplotlib.pyplot as plt


def windowed_view(x, window_size):
    """Creat a 2d windowed view of a 1d array.

    `x` must be a 1d numpy array.

    `numpy.lib.stride_tricks.as_strided` is used to create the view.
    The data is not copied.

    Example:

    >>> x = np.array([1, 2, 3, 4, 5, 6])
    >>> windowed_view(x, 3)
    array([[1, 2, 3],
           [2, 3, 4],
           [3, 4, 5],
           [4, 5, 6]])
    """
    y = as_strided(x, shape=(x.size - window_size + 1, window_size),
                   strides=(x.strides[0], x.strides[0]))
    return y


def rolling_max_dd(x, window_size, min_periods=1):
    """Compute the rolling maximum drawdown of `x`.

    `x` must be a 1d numpy array.
    `min_periods` should satisfy `1 <= min_periods <= window_size`.

    Returns an 1d array with length `len(x) - min_periods + 1`.
    """
    if min_periods < window_size:
        pad = np.empty(window_size - min_periods)
        pad.fill(x[0])
        x = np.concatenate((pad, x))
    y = windowed_view(x, window_size)
    running_max_y = np.maximum.accumulate(y, axis=1)
    dd = y - running_max_y
    return dd.min(axis=1)


def max_dd(ser):
    max2here = pd.expanding_max(ser)
    dd2here = ser - max2here
    return dd2here.min()


if __name__ == "__main__":
    np.random.seed(0)
    n = 100
    s = pd.Series(np.random.randn(n).cumsum())

    window_length = 10

    rolling_dd = pd.rolling_apply(s, window_length, max_dd, min_periods=0)
    df = pd.concat([s, rolling_dd], axis=1)
    df.columns = ['s', 'rol_dd_%d' % window_length]
    df.plot(linewidth=3, alpha=0.4)

    my_rmdd = rolling_max_dd(s.values, window_length, min_periods=1)
    plt.plot(my_rmdd, 'g.')

    plt.show()

该图显示了您的代码生成的曲线。绿点由 rolling_max_dd 计算。

时序比较，n = 10000 和 window_length = 500:

In [2]: %timeit rolling_dd = pd.rolling_apply(s, window_length, max_dd, min_periods=0)
1 loops, best of 3: 247 ms per loop

In [3]: %timeit my_rmdd = rolling_max_dd(s.values, window_length, min_periods=1)
10 loops, best of 3: 38.2 ms per loop

rolling_max_dd 大约快 6.5 倍。对于较小的窗口长度，加速效果更好。例如，使用 window_length = 200，它几乎快了 13 倍。

要处理 NA，您可以在将数组传递给 rolling_max_dd 之前使用 fillna 方法预处理 Series。