上一篇结尾我们把 5 只股票一天的涨幅算了出来,代码不长,跑得也快。但请把场面放大一万倍:5000 只股票、每个 1200 个交易日、60 万个价格点。如果还用 for 循环挨个元素去除法, Python 一圈跑下来要等上好几秒甚至更久——而行情计算每天要做几百次这种活儿。慢的原因不是 Python 不行,而是「逐个元素打交道」这个思路不行。本篇请出的主角 NumPy,就是来解决这个问题的。
上一篇我们用 list、dict 等容器把数据装好、用循环和推导式做了批量计算,这是「用 Python 自己的力气干活」。本篇在其之上引入 NumPy 的 ndarray:数据整列装进一块连续内存,运算一次作用到整列,这种思路叫向量化(Vectorization)。你会亲眼看到——用计时器实测——同一个收益率计算,循环写法和向量化写法差出一个数量级的速度。这个思维转变是后面所有篇章的地基:第 4 篇的 Pandas 表格、第 18 篇的向量化回测,全是它的延伸。
NumPy 全名 Numerical Python,是 Python 科学计算的底座:它自己提供高效的数组与数学函数,Pandas、matplotlib、各类回测框架全都构建在它之上。所以哪怕你以后很少直接写 np. 开头的代码,也必须懂它。
本篇目标
- 能用
np.array、np.zeros、np.arange、np.linspace、np.full五种方式创建 ndarray,并说出彼此的适用场合。 - 能读写 ndarray 的
dtype(元素类型)与shape(形状),并解释「二维数组 = 行是股票、列是交易日」的约定。 - 能用下标、切片、布尔索引从数组里取数,并演示「切片是视图、改切片会动原数组」。
- 能解释广播(Broadcasting)规则,让
(2, 3)的价格矩阵与(3,)的权重向量直接相乘。 - 能实测「整列一步算收益率」与「for 循环逐个算」的耗时差,并用
np.allclose验证两种写法结果一致。
核心概念
ndarray:整列装进一块连续内存
ndarray(N-dimensional Array,N 维数组)是 NumPy 的核心数据结构:同一种类型的一批数,装在内存里一整块连续的空间中。这一点和 list 有本质区别——list 里每个元素都是独立对象,散落在内存各处,取一个元素要多绕一跳;ndarray 的元素紧挨着排,配上底层的 C 语言实现,整列运算时 CPU 可以流水线式地扫过去,这就是它快的根源。
创建 ndarray 的常用姿势有五种:np.array(列表) 把现成列表转成数组;np.zeros(n) 生成 n 个 0,常用来「预分配」结果容器;np.arange(1, 6) 类似 range,生成 1 到 5;np.linspace(0, 1, 6) 在 0 到 1 之间均匀取 6 个点,画坐标轴时常用;np.full((2, 3), 100.0) 生成 2 行 3 列、全填 100 的二维数组。
dtype 与 shape:数组的身份证
每个 ndarray 都有两个最重要的属性。dtype(Data Type,元素类型)说明数组里装的是什么类型的数:float64 是 64 位浮点数,行情价格最常用;int64 是整数。同一种类型才装得进同一块连续内存,这也是 ndarray 比 list 快的前提之一。shape(形状)描述数组的结构:一维数组 (5,) 是 5 个数的一排;二维数组 (2, 3) 是 2 行 3 列的表格。量化里最常用的约定是:二维价格矩阵,一行一只股票,一列一个交易日。
索引、切片与布尔索引
一维数组的下标和切片与 list 完全同规则:a[0] 首元素、a[-1] 末元素、a[1:4] 包头不包尾。二维数组用 [行, 列] 双下标:g[1, 2] 是第 2 行第 3 列;g[:, 0] 的冒号表示「所有行」,合起来取第 1 列。切片依然是视图(View):改切片元素,原数组同步变化——这是 NumPy 省内存的设计,但也是新手数据被「莫名改掉」的头号来源,需要副本时用 .copy()。
布尔索引(Boolean Indexing)是量化取数的利器:a > 10.3 会生成一个同形状的 True/False 数组,a[mask] 一步筛出所有满足条件的元素。「筛出今天涨幅超过 3% 的股票」这类需求,全靠它一句写完。
广播:形状不同也能运算
广播(Broadcasting)是 NumPy 自动「补齐形状」的规则:两个形状不同的数组运算时,NumPy 沿缺失的维度自动复制对齐。最常用的场景:(2, 3) 的价格矩阵乘以 (3,) 的权重向量,权重被自动应用到矩阵的每一行,一行代码完成「3 天持仓权重加权」。规则只有一条主干:从最后一维往前比对,维度相等或其中一方为 1 才能对齐,对不上就报错。不用背细节,记住「短的那一维会被自动复制」,剩下的靠实验理解。
常用统计函数:一行顶一个循环
sum 求和、mean 均值、std 标准差(Standard Deviation,衡量波动大小,下一篇讲收益率时会大量使用)、max/min 极值、median 中位数,这些函数都有一个 axis 参数:axis=0 沿列方向算(每列出一个结果),axis=1 沿行方向算(每行出一个结果)。二维价格矩阵上,prices.mean(axis=0) 得到「每天所有股票的均价」,prices.mean(axis=1) 得到「每只股票的均价」——方向记不清就动手试,试一次就忘不掉了。
向量化思维:整列一步算收益率
日收益率(Daily Return)定义为:
r(t) = P(t) / P(t-1) - 1
白话:今天的收盘价除以昨天的收盘价再减 1,就是今天涨(跌)了百分之几。手写循环的思路是「算第 t 天、存起来、算第 t+1 天」,要执行 N 次循环体;向量化的思路是「把『除以昨天的自己』这个动作一次性作用到整列」:
returns = prices[1:] / prices[:-1] - 1
prices[1:] 是「今天」的整列,prices[:-1] 是「昨天」的整列,两者逐元素相除再减 1,一句代码得到全部收益率。这行代码本系列会写到烂熟,第 9 篇讲收益率、第 18 篇写向量化回测都从它出发。
划重点:遇到循环先问一句「这个操作能不能整列做」。收益率、均线、标准化这些高频计算,NumPy 都有整列写法,速度通常快一到两个数量级。
划重点:
prices[1:]与prices[:-1]长度恰好都是 N−1,逐元素对齐后一个不差。理解这行切片,比背十遍广播规则都有用。
完整代码
依赖安装:本篇只需 uv pip install numpy(第 1 篇创建的 quant 环境通常已自带)。第一段代码过一遍 ndarray 的创建、属性、切片、广播与统计,保存为 numpy_basics.py 运行:
# numpy_basics.py —— ndarray 的创建、形状、切片、广播与统计
import numpy as np
# ---- 五种常用创建方式 ----
a = np.array([10.2, 10.5, 10.3, 10.9, 11.2]) # 列表 -> 一维数组
b = np.zeros(5) # 5 个 0, 常用来预分配结果容器
c = np.arange(1, 6) # [1 2 3 4 5], 类似 range
d = np.linspace(0, 1, 6) # 0 到 1 均匀取 6 个点
e = np.full((2, 3), 100.0) # 2 行 3 列全填 100
print(f"a = {a}, dtype = {a.dtype}")
# ---- dtype 与 shape: 数组的身份证 ----
f = np.array([1, 2, 3], dtype=np.float64) # 指定为 64 位浮点
g = np.array([[1, 2, 3], [4, 5, 6]]) # 二维: 2 行 3 列
print(f"f.dtype = {f.dtype}, f.shape = {f.shape}")
print(f"g.shape = {g.shape}, 行数 {g.shape[0]}, 列数 {g.shape[1]}")
# ---- 索引与切片: 一维同 list, 二维用 [行, 列] ----
print(f"第一个元素 {a[0]}, 最后一个 {a[-1]}")
print(f"切片 a[1:4] = {a[1:4]}") # 包头不包尾
print(f"g[0, 2] = {g[0, 2]}") # 第 0 行第 2 列
print(f"g 第 2 行 = {g[1]}, g 第 1 列 = {g[:, 0]}") # 冒号表示整行/整列
# ---- 切片是视图: 改切片会动原数组 ----
view = a[1:3] # 视图, 不是副本
view[0] = 99.0 # 通过视图赋值
print(f"改切片后 a = {a}") # 原数组跟着变了!
# ---- 布尔索引: 一句条件筛出目标元素 ----
mask = a > 10.3 # 每个位置的真假判断
print(f"条件 a > 10.3 筛出: {a[mask]}")
# ---- 广播: (2,3) 矩阵 与 (3,) 向量直接运算 ----
prices = np.array([[10.0, 20.0, 30.0], [40.0, 50.0, 60.0]]) # 2 只股票 × 3 天
weights = np.array([0.5, 0.3, 0.2]) # 3 天的仓位权重
weighted = prices * weights # 权重自动对齐到每一行
print(f"加权后每行求和: {weighted.sum(axis=1)}") # 每只股票的加权结果
# ---- 常用统计函数与 axis ----
print(f"总和 {prices.sum():.0f}, 均值 {prices.mean():.2f}, 标准差 {prices.std():.2f}")
print(f"最大 {prices.max()}, 最小 {prices.min()}, 中位数 {np.median(prices)}")
print(f"按列均值(axis=0): {prices.mean(axis=0)}") # 每个交易日所有股票的均价
第二段是本篇的重头戏:同一个收益率计算,循环写法与向量化写法各跑一遍并计时,保存为 vectorize_demo.py 运行:
# vectorize_demo.py —— 整列价格一步算收益率: 循环 vs 向量化计时对比
import time # 标准库计时模块
import numpy as np
rng = np.random.default_rng(seed=7) # 固定种子, 结果可复现
n_days = 250_000 # 相当于把 250 个交易日放大一千倍, 模拟大数据量
daily_ret = rng.normal(loc=0.0002, scale=0.02, size=n_days) # 模拟日收益率
prices = 100.0 * np.cumprod(1 + daily_ret) # 累乘还原成价格序列, 期初 100 元
# ---- 写法一: for 循环逐个元素计算 ----
start = time.perf_counter() # 高精度计时起点
loop_ret = np.empty(n_days - 1) # 预分配结果数组
for i in range(n_days - 1): # 逐个交易日
loop_ret[i] = prices[i + 1] / prices[i] - 1 # 今天/昨天 - 1
loop_time = time.perf_counter() - start
# ---- 写法二: 向量化, 整列一步算完 ----
start = time.perf_counter()
vector_ret = prices[1:] / prices[:-1] - 1 # 今天整列 / 昨天整列 - 1
vector_time = time.perf_counter() - start
print(f"数据量: {n_days} 个价格, 输出 {n_days - 1} 个日收益率")
print(f"循环版耗时: {loop_time:.4f} 秒")
print(f"向量化耗时: {vector_time:.4f} 秒")
print(f"向量化快约 {loop_time / vector_time:.0f} 倍")
print(f"两种写法结果一致: {np.allclose(loop_ret, vector_ret)}") # 逐元素近似相等
print(f"首日收益率 {vector_ret[0]:+.4%}, 均值 {vector_ret.mean():+.4%}, "
f"标准差 {vector_ret.std():.4%}")
第三段把向量化推广到二维:3 只股票 × 5 天的价格矩阵,一次算出全部收益率,保存为 matrix_returns.py 运行:
# matrix_returns.py —— 进阶: 用二维矩阵一次算多只股票的收益率
import numpy as np
rng = np.random.default_rng(seed=11) # 换个种子
# (3, 5) 的日收益率 -> 累乘成价格矩阵: 每行是一只股票 5 天的价格
prices = np.round(100 * np.cumprod(1 + rng.normal(0.001, 0.02, (3, 5)), axis=1), 2)
print(f"价格矩阵(行=股票, 列=交易日):\n{prices}")
rets = prices[:, 1:] / prices[:, :-1] - 1 # 广播到整个矩阵, 一步算完
print(f"收益率矩阵(3 只 × 4 天):\n{np.round(rets, 4)}")
print(f"每只股票的日均收益率: {np.round(rets.mean(axis=1), 4)}") # 沿行内平均
print(f"每只股票的日波动率: {np.round(rets.std(axis=1), 4)}")
best = np.unravel_index(np.argmax(rets), rets.shape) # 全矩阵最大值的行列位置
print(f"最大单日收益率出现在第 {best[0]} 只股票(下标从 0 数)第 {best[1] + 1} 天")
代码逐段解释
- 五种创建方式:
np.array从 list 转换最常用;np.zeros预分配在第二段代码里立刻派上用场(循环结果先占好位置);arange与linspace的区别是前者按步长、后者按个数;np.full造初始矩阵。 - dtype 与 shape:
f显式指定float64——价格、收益率这类连续量一律用浮点;g.shape返回(2, 3),shape[0]是行数、shape[1]是列数。以后调试「维度对不上」的错误,第一件事就是 print shape。 - 切片是视图:
view = a[1:3]拿到的不是复制品,改view[0]直接改到了a。要独立副本就写a[1:3].copy()。示例里我们故意利用这个特性展示「改切片、原数组变」,实际写策略时要警惕无意中污染原始数据。 - 布尔索引:
a > 10.3生成[False False False True True]形状的真假数组(对照被改后的a),a[mask]只把 True 位置的元素捞出来。多条件组合写mask = (a > 10) & (a < 11),注意要用&而不是and。 - 广播加权:
(2, 3)的prices乘(3,)的weights,NumPy 把权重向量沿行方向复制成 2 行再逐元素相乘,得到每只股票每天的「仓位 × 价格」,sum(axis=1)沿行求和就是每只股票的加权结果。这里 weights 恰好是按天加权的示意(权重和为 1)。 - 收益率两种写法:循环版提前
np.empty预分配,避免往列表里 append 上万次;向量化版用「今天整列 / 昨天整列 − 1」一句完成。np.allclose逐元素检查两者近似相等(浮点比较不能直接用==,呼应上一篇的坑 5),快不代表能算错,验证是向量化改造的固定收尾动作。 - 计时与结果解读:
time.perf_counter()取高精度时间戳,前后相减即耗时。示意输出里向量化快约 50 倍——这个倍数随机器和数据量浮动,但「一个数量级以上的差距」是稳定的。{x:+.4%}是百分比格式:自动乘 100、保留 4 位小数、带正负号。 - 二维推广:
prices[:, 1:]取「第 2 天到第 5 天」的所有行,prices[:, :-1]取「第 1 天到第 4 天」的所有行,相除后每行内部逐元素对齐,一次得到 3 只股票 × 4 天的收益率矩阵。np.argmax找全矩阵最大值的扁平位置,np.unravel_index把它还原成 (行, 列)——查「哪只股票哪天涨得最猛」就靠这一对搭档。
第一段代码的运行结果(示意输出):
a = [10.2 10.5 10.3 10.9 11.2], dtype = float64
f.dtype = float64, f.shape = (3,)
g.shape = (2, 3), 行数 2, 列数 3
第一个元素 10.2, 最后一个 11.2
切片 a[1:4] = [10.5 10.3 10.9]
g[0, 2] = 3
g 第 2 行 = [4 5 6], g 第 1 列 = [1 4]
改切片后 a = [10.2 99. 10.3 10.9 11.2]
条件 a > 10.3 筛出: [99. 10.9 11.2]
加权后每行求和: [17. 47.]
总和 210, 均值 35.00, 标准差 17.08
最大 60, 最小 10, 中位数 35.0
按列均值(axis=0): [25. 35. 45.]
第二段代码的运行结果(示意输出,耗时与倍数因机器而异,量级关系稳定;均值、标准差由模拟参数决定,约 +0.02% 与 2%):
数据量: 250000 个价格, 输出 249999 个日收益率
循环版耗时: 0.0861 秒
向量化耗时: 0.0017 秒
向量化快约 51 倍
两种写法结果一致: True
首日收益率 +1.3621%, 均值 +0.0210%, 标准差 2.0007%
第三段代码的运行结果(示意输出,随机数固定种子下可复现,数值随 NumPy 版本可能略有差异):
价格矩阵(行=股票, 列=交易日):
[[ 98.71 100.24 101.05 99.32 100.87]
[101.12 99.87 102.45 103.01 101.98]
[ 99.45 101.33 100.87 103.22 105.1 ]]
收益率矩阵(3 只 × 4 天):
[[ 0.0155 0.0081 -0.0171 0.0156]
[-0.0124 0.0258 0.0055 -0.01 ]
[ 0.0189 -0.0045 0.0233 0.0182]]
每只股票的日均收益率: [0.0055 0.0022 0.014 ]
每只股票的日波动率: [0.0134 0.0152 0.0108]
最大单日收益率出现在第 1 只股票(下标从 0 数)第 2 天
常见坑与小练习
常见坑:
- 把切片当副本用。现象:改了切片里的一个数,原始价格序列也变了,回测结果悄悄失真。原因:NumPy 切片默认是视图,与 list 切片(返回新列表)行为不同。避免:需要独立数据时显式
.copy();读到别人代码里的切片赋值时多留个心眼。 - 整数数组做除法结果类型想当然。现象:用整数价格算收益率,小数位对不上。原因:Python 3 的
/会转浮点,但dtype=np.int64的数组做其他某些运算(如取均值前的累加)时要注意溢出与精度。避免:行情数据统一用float64,加载时先astype(np.float64)。 axis方向记反。现象:想算「每只股票的均价」,写了mean(axis=0)得到的却是「每天的均价」。原因:axis 指的是「沿哪个方向压扁」——axis=0跨行压缩、结果按列出。避免:不确定就两种都跑一遍看 shape,(3,)还是(5,)一眼便知。- 布尔组合误用
and/or。现象:(a > 1) and (a < 5)直接报ValueError。原因:NumPy 数组的真假是逐元素的,必须用&、|,并且每个条件要加括号。避免:把and/or/not从数组运算的词汇表里删掉,换成& | ~。 - 计时只跑一次就下结论。现象:第一次运行显示「向量化反而慢」。原因:首次调用有初始化开销,系统瞬时负载也干扰计时。避免:正式对比时同一写法连跑多次取平均(后面篇章会用到
timeit),单次计时只用于量级判断。
小练习:
- 把
vectorize_demo.py里的n_days改成 25_000 和 2_500_000 各跑一次,记录两种写法的耗时和倍数,观察倍数随数据量的变化趋势。思路提示:数据越小倍数往往越不明显(固定开销占比高),把三次结果并排写下来,你会对「什么时候必须向量化」有体感。
- 在布尔索引那一段加一行:筛出
a中大于 10 且小于 11 的元素(用&组合两个条件),打印结果和对应的下标(提示:np.where)。思路提示:
mask = (a > 10) & (a < 11),a[mask]给值,np.where(mask)[0]给下标。 - 把第三段的价格矩阵扩成 5 只股票 × 20 天(改
(3, 5)为(5, 20)),计算每只股票的「区间累计涨幅」(最后一列 / 第一列 − 1),并找出累计涨幅最高的股票下标。思路提示:累计涨幅是
prices[:, -1] / prices[:, 0] - 1,配合np.argmax一行完成;注意这里用的是价格首尾比,不是日收益率的简单相加。
本文内容仅用于教学,不构成投资建议,实盘前请充分测试。