第五节：NumPy与模拟股票数据

先认识这份案例数据

数据文件 stock_demo.csv 含有 4 只模拟股票、15 个日期和 60 条观测。数值是为了教学而合成的，不代表真实市场报价；它的优点是可以放心修改，且每次运行得到相同结果。

建议先把 CSV 下载到课程项目的 data/ 文件夹，再在 Notebook 中使用相对路径。这样换一台电脑时，项目结构仍然清楚。

1. NumPy 到底解决什么问题

Python 列表可以保存数字，但金融计算常常需要同时处理几百只股票、几千个日期和多个情景。NumPy 的 ndarray 是专门存储同类数字的数组，支持整列运算：写一次公式，就能对每一个元素执行。

import numpy as np

prices = np.array([10.0, 10.5, 10.2, 10.8])
returns = prices[1:] / prices[:-1] - 1
print(returns)
print("平均日收益率:", returns.mean())
print("日波动率:", returns.std(ddof=1))

这里的 prices[1:] 是从第二个价格开始，prices[:-1] 是到倒数第二个价格为止；两者对齐后，每个分子都对应下一个交易日，每个分母都对应前一个交易日。

2. 读取 CSV 并检查数组形状

第五节先用一个纯 NumPy 的小练习理解数组，再用 Pandas 读取 CSV。现实项目中通常由 Pandas 负责读取表格，由 NumPy 负责数值运算；这不是竞争关系，而是分工。

from pathlib import Path
import pandas as pd
import numpy as np

file_path = Path("data/stock_demo.csv")
df = pd.read_csv(file_path)
close_prices = df["close"].to_numpy()
print("记录数:", close_prices.size)
print("数组形状:", close_prices.shape)
print("前 5 个收盘价:", close_prices[:5])

此时数组是 60 行的一维数组，股票之间和日期之间混在一起。要计算每只股票自己的收益率，需要先把长表整理成“日期 × 股票”的矩阵。这一步会在第六节详细讲解；现在先用已知的 15×4 数组体验 NumPy。

close_matrix = df.pivot(index="date", columns="ticker", values="close").to_numpy()
print(close_matrix.shape)
print(close_matrix[0])

3. 用向量化计算收益率

最常用的简单收益率公式是：

r_t = P_t / P_(t-1) - 1。它表示价格从前一个观测日到当前观测日的相对变化，不是百分数；如果输出 0.01，阅读时要转成 1%。

returns_matrix = close_matrix[1:] / close_matrix[:-1] - 1
print("收益率矩阵形状:", returns_matrix.shape)
print("第一只股票前 3 个收益率:", returns_matrix[:3, 0])
print("第一天各股票收益率:", returns_matrix[0])

为什么从 15 个价格变成 14 个收益率？因为第一天没有“前一天价格”，无法计算第一天的变化。这种少一行不是错误，而是金融时间序列中的自然结果。

4. 创建、查看和改变 NumPy 数组

NumPy 数组有维度、形状、数据类型和轴。一维可以表示一只股票的时间序列，二维可以表示日期 × 股票，三维还可以表示多个情景 × 日期 × 股票。

import numpy as np

one_dim = np.array([12.8, 1683.2, 180.7, 42.2])
two_dim = np.array([[12.8, 1683.2], [12.9, 1690.4]])
zeros = np.zeros((2, 3))
sequence = np.arange(0, 10, 2)
points = np.linspace(0, 1, 5)

for name, array in [("one_dim", one_dim), ("two_dim", two_dim),
                    ("zeros", zeros), ("sequence", sequence),
                    ("points", points)]:
    print(name, "ndim=", array.ndim, "shape=", array.shape,
          "dtype=", array.dtype)
print("sequence:", sequence)
print("points:", points)

zeros 常用于初始化结果矩阵，arange 按固定步长生成序列，linspace 按数量均匀分割区间。需要独立副本时使用 .copy()，避免修改视图时意外改变原始价格。

prices = np.array([[12.79, 12.81, 12.86],
                   [1683.25, 1690.43, 1690.49]])
print("第一行:", prices[0])
print("第二列:", prices[:, 1])
prices_copy = prices.copy()
prices_copy[0, 0] = 99
print("原数组左上角:", prices[0, 0])
print("复制数组左上角:", prices_copy[0, 0])

5. 索引、切片与布尔筛选

索引从 0 开始，切片右端不包含，布尔数组可以把满足条件的观测一次性筛出来。多个条件用 & 表示“并且”、用 | 表示“或者”，每个条件都要加括号。

close = np.array([12.79, 12.81, 12.86, 12.84, 12.75])
print("前三个:", close[:3])
print("最后两个:", close[-2:])
print("倒序:", close[::-1])
mask = close > close.mean()
print("平均价格:", close.mean())
print("是否高于平均:", mask)
print("高于平均的价格:", close[mask])

下面的筛选同时使用第一列和第四列，模拟“银行股价格超过阈值且另一只股票价格也超过阈值”的条件。

prices = np.array([
    [12.79, 1683.25, 180.71, 42.19],
    [12.81, 1690.43, 181.61, 42.36],
    [12.86, 1690.49, 180.83, 42.07]
])
selected = prices[(prices[:, 0] > 12.80) & (prices[:, 3] > 42.10)]
print(selected)

6. 广播：让不同形状的数据批量计算

广播允许形状兼容的数组批量运算，不需要手动复制。二维收益矩阵形状为 (日期, 股票)，权重形状为 (股票,)，因此每一行都会自动使用同一套权重。

returns = np.array([
    [0.010, -0.005, 0.020, 0.004],
    [-0.002, 0.008, -0.010, 0.006]
])
weights = np.array([0.25, 0.25, 0.30, 0.20])
weighted = returns * weights
portfolio = weighted.sum(axis=1)
print("逐项加权:\n", weighted)
print("组合收益:", portfolio)

如果权重写成 3 个数字，形状无法对齐，就会得到维度错误。先打印 .shape，再修正数组，而不是盲目让 AI 改代码。

base_price = np.array([12.8, 1680.0, 180.0, 42.0])
change = np.array([0.01, -0.02, 0.03, 0.00])
scenario_prices = base_price * (1 + change)
print(scenario_prices)
scenario_change = np.array([
    [0.01, 0.01, 0.01, 0.01],
    [-0.02, 0.00, 0.03, 0.01],
    [-0.05, -0.03, 0.08, -0.02]
])
portfolio_value = 1_000_000 * (1 + scenario_change @ weights)
print(portfolio_value)

7. 统计函数：从价格数组读出风险特征

均值回答“典型水平”，中位数减少极端值影响，标准差回答“分散程度”，百分位数帮助识别尾部风险。累计收益要用连续复利。

daily_returns = returns_matrix[:, 0]
print("均值:", np.mean(daily_returns))
print("中位数:", np.median(daily_returns))
print("标准差:", np.std(daily_returns, ddof=1))
print("最小值:", np.min(daily_returns))
print("最大值:", np.max(daily_returns))
print("第 25 百分位:", np.percentile(daily_returns, 25))
print("累计收益:", np.prod(1 + daily_returns) - 1)

mean_by_stock = returns_matrix.mean(axis=0)
std_by_stock = returns_matrix.std(axis=0, ddof=1)
statistics = np.column_stack([mean_by_stock, std_by_stock,
                              returns_matrix.min(axis=0),
                              returns_matrix.max(axis=0)])
print("列依次为均值、标准差、最小值、最大值")
print(np.round(statistics, 6))

axis=0 沿日期方向压缩，每列得到一只股票的统计量；axis=1 沿股票方向压缩，每行得到一个日期的统计量。

8. 相关系数、协方差与风险分散

两只股票可能各自波动很大，但如果不是同方向变化，把它们放在一起可能降低组合波动。NumPy 可以计算协方差矩阵和相关系数矩阵。

covariance = np.cov(returns_matrix, rowvar=False, ddof=1)
correlation = np.corrcoef(returns_matrix, rowvar=False)
print("协方差矩阵:\n", np.round(covariance, 8))
print("相关系数矩阵:\n", np.round(correlation, 3))
portfolio_variance = weights @ covariance @ weights
portfolio_volatility = np.sqrt(portfolio_variance)
print("组合方差:", portfolio_variance)
print("组合日波动率:", portfolio_volatility)

协方差单位是收益率平方，相关系数已经标准化。公式 wᵀΣw 把权重向量、协方差矩阵和权重向量组合起来；* 是逐元素相乘，@ 才是矩阵乘法。

9. 线性代数：矩阵乘法与方程组

金融模型常写成矩阵形式。例如多资产组合收益是 R w，最小二乘估计中会出现矩阵转置和方程组。先学会查看形状，再逐渐学习模型。

R = np.array([[0.01, 0.02],
              [0.03, -0.01],
              [-0.02, 0.04]])
w = np.array([0.6, 0.4])
print("每期组合收益:", R @ w)
print("R 的转置形状:", R.T.shape)
print("R.T @ R:\n", R.T @ R)

A = np.array([[2.0, 1.0], [1.0, 3.0]])
b = np.array([100.0, 120.0])
x = np.linalg.solve(A, b)
print("解:", x)
print("验证 A @ x:", A @ x)

矩阵的每一行和每一列都应有清楚的经济含义。不要为了让代码运行而随意转置数组。

10. 随机数与可复现的模拟

模拟数据、蒙特卡洛分析和压力测试都需要随机数。课堂示例必须设置种子，这样每次运行得到同样结果，别人才能复核。

from numpy.random import default_rng

rng = default_rng(20260106)
shocks = rng.normal(loc=0.0005, scale=0.01, size=(3, 4))
print(np.round(shocks, 4))
print("每个情景平均冲击:", np.round(shocks.mean(axis=1), 4))

loc 是平均水平，scale 是波动规模，size=(3,4) 表示情景数和股票数。随机数只是把假设转化为可计算的样本。

11. 缺失值、无穷值和数据检查

缺失值 NaN 会传播到很多运算；零价格、负成交量和无穷值则可能说明数据口径或代码有问题。

sample = np.array([0.01, np.nan, 0.02, np.inf, -0.01])
print("是否缺失:", np.isnan(sample))
print("是否有限:", np.isfinite(sample))
print("普通均值:", np.mean(sample))
clean = sample[np.isfinite(sample)]
print("忽略异常后的均值:", np.mean(clean))
assert np.all(clean > -1), "收益率低于 -100%，请检查数据"
print("清洗后的数组:", clean)

清洗规则要写进 Notebook。研究中还应追溯缺失值和无穷值为什么出现，而不是直接把所有异常替换成 0。

12. 从 NumPy 计算回到金融判断

NumPy 的学习路线已经从一个数字扩展到数组、矩阵、统计量、随机情景和金融解释。Pandas 会负责把真实表格整理成适合这些计算的形状。

13. 练习：把数组能力用起来

15. 从单只股票到投资组合

假设把资金分成四份：平安银行 25%、贵州茅台 25%、宁德时代 30%、中国平安 20%。组合每天的收益率是各资产收益率按权重加总。

weights = np.array([0.25, 0.25, 0.30, 0.20])
portfolio_returns = returns_matrix @ weights

mean_daily = portfolio_returns.mean()
vol_daily = portfolio_returns.std(ddof=1)
annual_return = mean_daily * 252
annual_vol = vol_daily * np.sqrt(252)

print("组合平均日收益率:", round(mean_daily, 6))
print("年化收益率近似:", round(annual_return, 4))
print("年化波动率近似:", round(annual_vol, 4))

这里使用 252 个交易日进行年化只是教学中的近似。模拟数据只有 14 个日收益率，不能据此声称真实年化收益率就是这个数。代码的目的，是让你看清“价格 → 收益 → 权重 → 组合风险”的计算链条。

16. 用数组做一个小型压力测试

为了理解风险，可以把每只股票的收益波动放大或缩小，观察组合是否仍能承受。这里不模拟复杂市场，只练习“同一组数组被重复计算”。

scenarios = np.array([0.8, 1.0, 1.2])
for scale in scenarios:
    scenario_returns = returns_matrix * scale
    scenario_portfolio = scenario_returns @ weights
    print(f"波动情景 {scale:.1f}：组合标准差 = {scenario_portfolio.std(ddof=1):.6f}")

当 scale 从 0.8 变为 1.2，所有收益波动被放大，组合标准差也随之放大。这是一个非常简化的压力测试：真实压力测试还应改变相关性、跳跃风险和流动性，而不是只乘一个数。

17. 综合练习：先预测结果，再运行代码

下一节会把长表重新整理成 DataFrame，用筛选、分组、合并和透视表完成同一案例的业务分析。