以整个数组而不是单个元素来思考
向量化就是把计算写成对整个数组的运算,让 NumPy 在快速的编译代码中运行循环。a + b 逐元素相加,a * b 相乘,a ** 2 对每个元素平方,np.sqrt(a) 求每个元素的平方根。这些都是通用函数(ufunc)。写起来更短、读起来更清楚,而且在大数据上往往比等价的 Python for 循环快五十倍,因为 for 循环在每个元素上都要付出解释器的开销。
import numpy as np
a = np.array([1.0, 2.0, 3.0])
b = np.array([10.0, 20.0, 30.0])
print(a + b) # [11. 22. 33.]
print(a * b) # [10. 40. 90.]
print(a ** 2) # [1. 4. 9.]
print(np.sqrt(a)) # [1. 1.41421356 1.73205081]
# The loop version of a * b, which vectorisation replaces
result = np.empty(3)
for i in range(3):
result[i] = a[i] * b[i]
print(result) # [10. 40. 90.]无需循环的收益率:用切片错位
需要把每个元素与相邻元素比较的计算,要用到数组错开一位的第二份拷贝,而切片可以免费提供。p[1:] 是今天的价格,p[:-1] 是昨天的价格,长度都是 n - 1,所以简单收益率就是 p[1:] / p[:-1] - 1。同一对切片还能给出差分(p[1:] - p[:-1],或 np.diff(p))。累计和与累计积是 np.cumsum 和 np.cumprod;增长路径是 np.cumprod(1 + returns)。整个数组的比较(r > 0)和 np.where(条件, 为真时, 为假时) 取代循环里的 if。
import numpy as np
p = np.array([100.0, 102.0, 101.0, 103.0])
returns = p[1:] / p[:-1] - 1 # today / yesterday - 1, no loop
print(np.round(returns, 4)) # [ 0.02 -0.0098 0.0198]
print(np.diff(p)) # [ 2. -1. 2.]
print(np.cumprod(1 + np.array([0.01, 0.02]))) # [1.01 1.0302]
# np.where replaces if/else inside a loop
r = np.array([0.01, -0.02, 0.03, -0.01])
print(np.where(r > 0, "up", "down")) # ['up' 'down' 'up' 'down']
print(np.where(r > 0, r, 0.0)) # [0.01 0. 0.03 0. ]沿轴聚合与矩阵乘积
sum、mean、std、min、max 等函数会把数组聚合。对于一个二维收益率数组(每行一天,每列一个资产),axis=0 沿行方向聚合,得到每列一个值(每个资产随时间的平均),而 axis=1 得到每行一个值(每天跨资产的平均)。形状为 (天数, 资产数) 的数组与长度为 资产数 的权重向量的矩阵乘积 R @ w,一次调用就给出每天的组合收益:这就是向量章节的点积,同时对所有行完成。
import numpy as np
# 2 days x 3 assets
R = np.array([[0.02, -0.01, 0.03],
[0.01, 0.00, 0.02]])
w = np.array([0.5, 0.3, 0.2])
print(R.shape) # (2, 3)
print(R.mean(axis=0)) # [ 0.015 -0.005 0.025] one value per asset
print(R.sum(axis=1)) # [0.04 0.03] one value per day
print(R @ w) # [0.013 0.009] portfolio return each day
# Standardise: how many standard deviations from the mean?
z = np.array([1.0, 2.0, 3.0, 4.0])
print((z - z.mean()) / z.std()) # [-1.34164079 -0.4472136 0.4472136 1.34164079]用 cumsum 求移动平均,以及向量化的局限
窗口为 w 的移动平均看似需要循环,但借助累计和就不需要:在 np.cumsum(x) 前面放一个 0,任意连续 w 个值的和就是两个累计和之差 cs[w:] - cs[:-w],再除以 w。结果有 len(x) - w + 1 个值。要向量化的是算术,而不是依赖上一步结果的逻辑(例如取决于昨天成交的仓位):这类逻辑留在循环里,或之后改用 Numba 之类的编译工具。还要留意内存,因为每个中间数组都是完整的副本。
import numpy as np
x = np.array([1.0, 2.0, 4.0, 8.0, 16.0])
w = 3
cs = np.concatenate(([0.0], np.cumsum(x))) # [0, 1, 3, 7, 15, 31]
print((cs[w:] - cs[:-w]) / w) # [2.33333333 4.66666667 9.33333333]
# Check against the plain definition
print([x[i:i + w].mean() for i in range(len(x) - w + 1)]) # three values, the samenp.allclose(快, 慢) 对比。还要留意边界情况:空数组、只有一个元素、常数序列(标准差为零,z 分数会除以零)以及 nan 值,需要明确的处理决定,而不是崩溃或悄悄给出错误的数。- 整个数组的运算(
a + b、a ** 2、np.sqrt(a))取代循环;条件用np.where。 - 错位切片:
p[1:] / p[:-1] - 1;cumsum和cumprod用于累计和与增长。 axis=0沿行聚合(每列一个值),axis=1沿列聚合(每行一个值);R @ w是每行的点积。
下一节:广播,让不同形状的数组可以一起运算的规则。
实验:评分内容
- 用
simple_returns一次算出所有收益率25 分 - 用
portfolio_returns得到每天的组合收益25 分 - 用
moving_average不用循环平滑序列25 分 - 用
zscore对数值标准化25 分