百倍加速！Python量化策略的算法性能提升指南

作者：用python的交易员
链接：https://zhuanlan.zhihu.com/p/24168485

性能问题

Python 在2016年里可以说是风靡国内量化投资圈，目前整个生态链已经初具规模：

• 交易：vn.py、easytrader、at_py

• 数据：tushare

• 回测：rqalpha

• 在线平台：UQER、RiceQuant、JoinQuant

随着用户越来越多，Python 语言的性能问题也就逐渐成为整个社区关注的重点，经常遇到新手问：Python 写的量化交易程序是不是很慢啊？

在他们心中，Python 估计是这个样子：

（即使作为破旧自行车，我也深表怀疑这辆能不能骑好吧）

网上关于如何提升 Python 程序性能的文章不少，但大多不成体系只是非常简单的例子，总有点隔靴搔痒的感觉，和现实中应用的距离比较远。

作者一看，填补市场空白（装逼）的机会来了！！

在这篇文章里，将会通过实际的例子展示如何对一段量化策略常用的代码实现百倍加速。

一段常用的代码

接下来要用的例子相信几乎所有做量化策略的人都写过类似的代码：对时间序列求算术移动平均值。

这里我们先初始化即将用到的数据：10 万个数据点（随机整数），遍历计算窗口为 500 的算术移动平均值，每种算法运行 10 次求平均耗时。

1. # 这个测试目标在于仿造一个类似于实盘中，不断有新的数据推送过来，

2. # 然后需要计算移动平均线数值，这么一个比较常见的任务。

3. from __future__ import division

4. import time

5. import random

6. # 生成测试用的数据

7. data = []

8. data_length = 100000    # 总数据量

9. ma_length = 500         # 移动均线的窗口

10. test_times = 10         # 测试次数

11. for i in range(data_length):

12.    data.append(random.randint(1, 100))

在每次测试中，我们都通过遍历测试用数据的方式来模拟实盘中策略不断收到新数据推送的情况（同样适用于事件驱动的回测模式），将计算出的移动平均值不断保存到一个列表 list 中作为最终结果返回。

测试用电脑的配置情况：Core i7-6700K 4.0G/16G/Windows 7。

第一步我们以最简单、最原始的方式来计算移动平均值：

1. # 计算500期的移动均线，并将结果保存到一个列表里返回

2. def ma_basic(data, ma_length):

3.    # 用于保存均线输出结果的列表

4.    ma = []

5.    # 计算均线用的数据窗口

6.    data_window = data[:ma_length]

7.    # 测试用数据（去除了之前初始化用的部分）

8.    test_data = data[ma_length:]

9.    # 模拟实盘不断收到新数据推送的情景，遍历历史数据计算均线

10.    for new_tick in test_data:

11.        # 移除最老的数据点并增加最新的数据点

12.        data_window.pop(0)

13.        data_window.append(new_tick)

14.        # 遍历求均线

15.        sum_tick = 0

16.        for tick in data_window:

17.            sum_tick += tick

18.        ma.append(sum_tick/ma_length)

19.    # 返回数据

20.    return ma

21. # 运行测试

22. start = time.time()

23. for i in range(test_times):

24.    result = ma_basic(data, ma_length)

25. time_per_test = (time.time()-start)/test_times

26. time_per_point = time_per_test/(data_length - ma_length)

27. print u'单次耗时：%s秒' %time_per_test

28. print u'单个数据点耗时：%s微秒' %(time_per_point*1000000)

29. print u'最后10个移动平均值：', result[-10:]

单次耗时指的是遍历完整个测试数据计算移动平均值所需的时间，单个数据点耗时指的是遍历过程中每个数据点的平均计算耗时，最后 10 个移动平均值用于和后续的算法进行比对，保证计算结果的正确性。

ma_basic测试结果

• 单次耗时：1.15699999332秒

• 单个数据点耗时：11.6281406364微秒

大约 10 万个数据点（说大约因为有 500 个用于初始化了），这个测试结果不能说很好但也还过得去。考虑到一个简单的双均线 CTA 策略（Double SMA Strategy），每个数据点来了后会进行两次均线计算，通常均线窗口不会超过 500，且比较两根均线交叉情况的算法开销更低，估计策略单纯在信号计算方面的耗时会在 30 微秒以内，对于一个通常跑在 1 分钟线甚至更高时间周期上的策略而言已经是绰绰有余。

有了起点，下面来试着一步步提升性能。

试试 NumPy？

用 Python 做数值运算性能不够的时候，很多人的第一反应就是上 NumPy：之前的 ma_basic 里，比较慢的地方应该在每一个新的数据点加入到 data_window 中后遍历求平均值的代码，那么改用 numpy.array 数组来求和应该性能就会有所提升了吧？

1. # 改用numpy（首先是一种常见的错误用法）

2. import numpy as np

3. def ma_numpy_wrong(data, ma_length):

4.    ma = []

5.    data_window = data[:ma_length]

6.    test_data = data[ma_length:]

7.    for new_tick in test_data:

8.        data_window.pop(0)

9.        data_window.append(new_tick)

10.        # 使用numpy求均线，注意这里本质上每次循环

11.        # 都在创建一个新的numpy数组对象，开销很大

12.        data_array = np.array(data_window)

13.        ma.append(data_array.mean())

14.    return ma

ma_numpy_wrong 测试结果

• 单次耗时：2.11879999638秒

• 单个数据点耗时：21.2944723254微秒

WTF?! 用 NumPy 后居然反而速度降低了一半（耗时增加到了快 2 倍）！

这里的写法是一个非常常见的 NumPy 错误用法，问题就出在：

data_array = np.array(data_window)

由于 NumPy 中的对象大多实现得比较复杂（提供了丰富的功能），所以其对象创建和销毁的开销都非常大。上面的这句代码意味着在计算每一个新数据点时，都要创建一个新的 array 对象，并且仅使用一次后就会销毁，使用  array.mean 方法求均值带来的性能提升还比不上 array 对象创建和销毁带来的额外开销。

正确的用法是把 np.array 作为 data_window 时间序列的容器，每计算一个新的数据点时，使用底层数据偏移来实现数据更新：

1. # numpy的正确用法

2. def ma_numpy_right(data, ma_length):

3.    ma = []

4.    # 用numpy数组来缓存计算窗口内的数据

5.    data_window = np.array(data[:ma_length])

6.    test_data = data[ma_length:]

7.    for new_tick in test_data:

8.        # 使用numpy数组的底层数据偏移来实现数据更新

9.        data_window[0:ma_length-1] = data_window[1:ma_length]

10.        data_window[-1] = new_tick

11.        ma.append(data_window.mean())

12.    return ma

ma_numpy_right 测试结果

• 单次耗时：0.614300012589秒

• 单个数据点耗时：6.17386947325微秒 

速度比 ma_basic 提高了大约 2 倍，看来 NumPy 也就这么回事了。

JIT神器：Numba

关心过 Python 性能的朋友应该都听过 PyPy 的大名，通过重新设计的 Python 解释器，PyPy 内建的 JIT 技术号称可以将 Python 程序的速度提高几十倍（相比于CPython），可惜由于兼容性的问题并不适合于量化策略开发这一领域。

幸运的是，我们还有 Anaconda 公司推出的 Numba。Numba 允许用户使用基于 LLVM 的 JIT 技术，对程序内想要提高性能的部分（函数）进行局部优化。同时 Numba 在设计理念上更加务实：可以直接在 CPython 中使用，和其他常用的 Python 模块的兼容性良好，并且最爽的是使用方法傻瓜到了极点：

1. # 使用numba加速，ma_numba函数和ma_basic完全一样

2. import numba

3. @numba.jit

4. def ma_numba(data, ma_length):

5. ma = []

6. data_window = data[:ma_length]

7. test_data = data[ma_length:]

8. for new_tick in test_data:

9.    data_window.pop(0)

10.    data_window.append(new_tick)

11.    sum_tick = 0

12.    for tick in data_window:

13.        sum_tick += tick

14.    ma.append(sum_tick/ma_length)

15. return ma

ma_numba 测试结果

• 单次耗时：0.043700003624秒

• 单个数据点耗时：0.439196016321微秒

OMG！就加了一行 @numba.jit，性能竟然提高了 26 倍！这估计是按照代码修改行数算，性价比最高的优化方案了。

改写算法

从编程哲学的角度来看，想提高计算机程序的速度，一个最基本的原则就是降低算法复杂度。看到这里估计早就有量化老手 ma_basic 不爽了，弄个复杂度 O(N) 的算法来算平均值，就不能缓存下求和的结果，把复杂度降低到 O(1) 么？

1. # 将均线计算改写为高速算法

2. def ma_online(data, ma_length):

3.    ma = []

4.    data_window = data[:ma_length]

5.    test_data = data[ma_length:]

6.    # 缓存的窗口内数据求和结果

7.    sum_buffer = 0

8.    for new_tick in test_data:

9.        old_tick = data_window.pop(0)

10.        data_window.append(new_tick)

11.        # 如果缓存结果为空，则先通过遍历求第一次结果

12.        if not sum_buffer:

13.            sum_tick = 0

14.            for tick in data_window:

15.                sum_tick += tick

16.            ma.append(sum_tick/ma_length)

17.            # 将求和结果缓存下来

18.            sum_buffer = sum_tick

19.        else:

20.            # 这里的算法将计算复杂度从O(n)降低到了O(1)

21.            sum_buffer = sum_buffer - old_tick + new_tick

22.            ma.append(sum_buffer/ma_length)

23.    return ma

ma_online 测试结果

• 单次耗时：0.0348000049591秒

• 单个数据点耗时：0.349748793559微秒

哲学果然才是最强大的力量！！！

(索罗斯：其实我是个哲学家。)

改写算法后的 ma_online 无需 JIT 就超越了 ma_numba，将性能提高到了 33 倍（对比 ma_basic），如果再把 numba 加上会如何？

1. # 高速算法和numba结合，ma_online_numba函数和ma_online完全一样

2. @numba.jit

3. def ma_online_numba(data, ma_length):

4.    ma = []

5.    data_window = data[:ma_length]

6.    test_data = data[ma_length:]

7.    sum_buffer = 0

8.    for new_tick in test_data:

9.        old_tick = data_window.pop(0)

10.        data_window.append(new_tick)

11.        if not sum_buffer:

12.            sum_tick = 0

13.            for tick in data_window:

14.                sum_tick += tick

15.            ma.append(sum_tick/ma_length)

16.            sum_buffer = sum_tick

17.        else:

18.            sum_buffer = sum_buffer - old_tick + new_tick

19.            ma.append(sum_buffer/ma_length)

20.    return ma

ma_online_numba 测试结果

• 单次耗时：0.0290000200272秒

• 单个数据点耗时：0.29145748771微秒

尽管性能进一步提升了到了 40 倍，不过相比较于 ma_numba 对比 ma_basic 的提升没有那么明显，果然哲学的力量还是太强大了。

终极武器：Cython

到目前为止使用纯 Python 环境下的优化方法我们已经接近了极限，想要再进一步就得发挥 Python 胶水语言的特性了：使用其他扩展语言。由于 CPython 虚拟机的开发语言是 C，因此在性能提升方面的扩展语言主要选择就是 C/C++，相关的工具包括 ctypes、cffi、Swig、Boost.Python 等，尽管功能十分强大，不过以上工具都无一例外的需要用户拥有 C/C++ 语言相关的编程能力，对于很多 Python 用户而言是个比较麻烦的事。

好在 Python 社区对于偷懒的追求是永无止境的，Cython 这一终极武器应运而生。关于 Cython 的详细介绍可以去看，简单来它的主要作用就是允许用户以非常接近 Python 的语法来实现非常接近 C 的性能。

先来试试最简单的方法：完全不修改任何代码，只是把函数放到 .pyx 文件里，调用 Cython 编译成 .pyd 扩展模块。

1. # 基础的cython加速

2. def ma_cython(data, ma_length):

3.    ma = []

4.    data_window = data[:ma_length]

5.    test_data = data[ma_length:]

6.    for new_tick in test_data:

7.        data_window.pop(0)

8.        data_window.append(new_tick)

9.        sum_tick = 0

10.        for tick in data_window:

11.            sum_tick += tick

12.        ma.append(sum_tick/ma_length)

13.    return ma

ma_cython测试结果

• 单次耗时：0.600800013542秒

• 单个数据点耗时：6.03819109088微秒

ma_cython 和 ma_basic 的代码完全相同，简单使用 Cython 编译后性能提高了大约 1 倍，不过这和之前我们已经达成的优化效果比可以说是毫无吸引力。

Cython 官方的 Quick Start 里，第一步是教会用户如何去编译程序，第二步就是如何使用静态声明来大幅提高性能，所以我们的下一步就是：静态声明+高速算法。

1. # cython和高速算法

2. def ma_cython_online(data, ma_length):

3.    # 静态声明变量

4.    cdef int sum_buffer, sum_tick, old_tick, new_tick

5.    ma = []

6.    data_window = data[:ma_length]

7.    test_data = data[ma_length:]

8.    sum_buffer = 0

9.    for new_tick in test_data:

10.        old_tick = data_window.pop(0)

11.        data_window.append(new_tick)

12.        if not sum_buffer:

13.            sum_tick = 0

14.            for tick in data_window:

15.                sum_tick += tick

16.            ma.append(sum_tick/ma_length)

17.            sum_buffer = sum_tick

18.        else:

19.            sum_buffer = sum_buffer - old_tick + new_tick

20.            ma.append(sum_buffer/ma_length)

21.    return ma

ma_cython_online 测试结果

• 单次耗时：0.00980000495911秒

• 单个数据点耗时：0.0984925121518微秒

117倍！！！比 ma_online_numba 的速度还提高了接近 3 倍，98 纳秒的计算速度已经足以满足大部分毫秒级别高频策略的延时需求。

主要的功臣是这行：

cdef int sum_buffer, sum_tick, old_tick, new_tick

把函数中用到的变量静态声明成int类型后，Cython 在编译时无需再考虑 Python 对象的动态性特点，可以把整个函数高度优化成类似静态语言的实现，从而达到了接近 C 语言的运行性能，再加上复杂度 O(1 ) 的高速算法，有这个级别的性能提升也就不足为奇了。

附上简单的 Cython 使用指南：

1. 把要使用 Cython 编译的函数放到 .pyx 文件中，比如 test.pyx

2. 创建 setup.py 用于设置相关编译选项

3. 打开 cmd 或者 terminal 进入到 test.pyx 和 setup.py 所在的文件夹

4. 运行 python setup.py build_ext --inplace，执行编译

5. 若编译成功则在当前文件夹下会出现 test.pyd

6. 打开 python 像使用其他模块一样载入 （import）test.pyd 使用

感谢 Numba、Cython 和哲学的强大力量，作者最终装逼成功，从自己挖的“百倍加速”这个坑里爬了出来，不用当标题党了。

最终的算法性能对比图：

最后做一些总结吧：

1. 现实工作中遇到需要优化 Python 程序性能时，首先要做的就是去寻找程序里延时较大的热点代码，找到了问题所在，解决方案才有意义；

2. 所有的优化工作都应该基于测试来一步步推进，同样的优化方法对于不同类型的代码效果可能是截然相反的，同时错误的优化方法还不如不要优化（比如 ma_numpy_wrong）；

3. 只需增加一句代码（@numba.jit）就能实现加速的 Numba 无疑是性价比最高的优化方案，值得优先尝试，不过需要注意 numba 的 JIT 技术局限性比较大（主要针对数值计算相关的逻辑）；

4. 学习如何降低算法复杂度和编写更高效的算法，可以在潜移默化中提高自己的编程水平，在长期而言是对 Quant 或者程序员最有价值的优化方法；

5. 如果其他优化方法都无法达到令你满意的性能水平，试试 Cython （记得一定要加静态声明）;

6. 一个好的程序架构设计非常重要，把功能不同的计算逻辑分解到不同的函数里，适当降低每个函数的代码行数，会有助于后期的性能优化工作。

文章中的所有代码以及测试用的 Jupyter Noteboo 都可以在 vn.py 项目的 Github 仓库下载，记得点 Star！