Quant Python 面试实战：从 NumPy 到 Backtest 的 30 道代码题精讲

1import numpy as np
2import time
3
4# 生成测试数据
5a = np.random.randn(10000)
6b = np.random.randn(10000)
7
8# 方法 1：Python 循环（慢）
9start = time.time()
10dot_loop = sum(a[i] * b[i] for i in range(len(a)))
11loop_time = time.time() - start
12
13# 方法 2：NumPy 向量化（快）
14start = time.time()
15dot_numpy = np.dot(a, b)
16numpy_time = time.time() - start
17
18print(f"Loop: {loop_time*1000:.2f}ms, NumPy: {numpy_time*1000:.2f}ms")
19print(f"Speedup: {loop_time/numpy_time:.0f}x")
20# 典型输出：Loop: 3.5ms, NumPy: 0.02ms, Speedup: 175x

1import numpy as np
2
3# 模拟日收益率数据（252 个交易日，5 支股票）
4np.random.seed(42)
5returns = np.random.randn(252, 5) * 0.01  # 1% 日波动率
6
7# 计算协方差矩阵（年化：乘以 252）
8cov_matrix = np.cov(returns.T) * 252
9print("年化协方差矩阵形状:", cov_matrix.shape)  # (5, 5)
10
11# 计算相关系数矩阵
12corr_matrix = np.corrcoef(returns.T)
13print("相关系数矩阵（对角线应为 1）:")
14print(np.round(corr_matrix, 3))
15
16# 验证：从协方差矩阵计算相关系数
17std_devs = np.sqrt(np.diag(cov_matrix))
18corr_from_cov = cov_matrix / np.outer(std_devs, std_devs)
19print("两种方法结果一致:", np.allclose(corr_matrix, corr_from_cov))

1import numpy as np
2
3np.random.seed(42)
4returns = np.random.randn(252, 10) * 0.01
5
6# 标准化（零均值）
7returns_centered = returns - returns.mean(axis=0)
8
9# SVD 分解（PCA 的核心）
10U, S, Vt = np.linalg.svd(returns_centered, full_matrices=False)
11
12# 解释方差比例
13explained_variance = S**2 / (len(returns) - 1)
14total_variance = explained_variance.sum()
15explained_ratio = explained_variance / total_variance
16
17print("前 3 个主成分解释方差比例:")
18for i in range(3):
19    print(f"  PC{i+1}: {explained_ratio[i]:.1%}")
20print(f"  累计: {explained_ratio[:3].sum():.1%}")
21
22# 投影到前 3 个主成分
23pc_scores = returns_centered @ Vt[:3].T
24print("主成分得分形状:", pc_scores.shape)  # (252, 3)

1import numpy as np
2
3# 模拟股价数据
4np.random.seed(42)
5prices = 100 * np.exp(np.cumsum(np.random.randn(252) * 0.01))
6
7def rolling_stats(arr, window):
8    """计算滚动均值和标准差（纯 NumPy 实现）"""
9    n = len(arr)
10    means = np.full(n, np.nan)
11    stds = np.full(n, np.nan)
12    
13    for i in range(window - 1, n):
14        window_data = arr[i - window + 1:i + 1]
15        means[i] = np.mean(window_data)
16        stds[i] = np.std(window_data, ddof=1)  # ddof=1 使用样本标准差
17    
18    return means, stds
19
20# 更高效的向量化实现（使用 stride tricks）
21def rolling_mean_vectorized(arr, window):
22    """向量化滚动均值"""
23    cumsum = np.cumsum(arr)
24    cumsum[window:] = cumsum[window:] - cumsum[:-window]
25    result = np.full(len(arr), np.nan)
26    result[window-1:] = cumsum[window-1:] / window
27    return result
28
29means, stds = rolling_stats(prices, 20)
30print(f"第 20–25 日的滚动均值: {means[19:25].round(2)}")

1import numpy as np
2
3def ols_regression(X, y):
4    """
5    OLS 线性回归
6    X: (n, p) 特征矩阵（不含截距列）
7    y: (n,) 目标向量
8    返回: coefficients, r_squared, t_stats, p_values
9    """
10    n, p = X.shape
11    
12    # 添加截距列
13    X_with_intercept = np.column_stack([np.ones(n), X])
14    
15    # OLS 估计：β = (X'X)^{-1} X'y
16    XtX = X_with_intercept.T @ X_with_intercept
17    Xty = X_with_intercept.T @ y
18    beta = np.linalg.solve(XtX, Xty)  # 比 inv 更稳定
19    
20    # 残差和 R²
21    y_pred = X_with_intercept @ beta
22    residuals = y - y_pred
23    ss_res = np.sum(residuals**2)
24    ss_tot = np.sum((y - y.mean())**2)
25    r_squared = 1 - ss_res / ss_tot
26    
27    # 标准误差和 t 统计量
28    sigma_sq = ss_res / (n - p - 1)  # 无偏估计
29    var_beta = sigma_sq * np.linalg.inv(XtX)
30    se_beta = np.sqrt(np.diag(var_beta))
31    t_stats = beta / se_beta
32    
33    return beta, r_squared, t_stats
34
35# 测试：Fama-French 单因子模型（CAPM）
36np.random.seed(42)
37n = 252
38market_returns = np.random.randn(n) * 0.01
39stock_returns = 0.5 + 1.2 * market_returns + np.random.randn(n) * 0.005
40
41beta, r2, t_stats = ols_regression(market_returns.reshape(-1, 1), stock_returns)
42print(f"Alpha: {beta[0]:.4f}, Beta: {beta[1]:.4f}")
43print(f"R²: {r2:.4f}")
44print(f"t-stats: Alpha={t_stats[0]:.2f}, Beta={t_stats[1]:.2f}")

1import numpy as np
2from scipy.stats import norm
3
4def black_scholes_call(S, K, T, r, sigma):
5    """Black-Scholes 解析解"""
6    d1 = (np.log(S/K) + (r + 0.5*sigma**2)*T) / (sigma*np.sqrt(T))
7    d2 = d1 - sigma*np.sqrt(T)
8    return S*norm.cdf(d1) - K*np.exp(-r*T)*norm.cdf(d2)
9
10def monte_carlo_call(S, K, T, r, sigma, n_simulations=100000):
11    """Monte Carlo 期权定价"""
12    np.random.seed(42)
13    
14    # 模拟到期时股价（GBM）
15    Z = np.random.standard_normal(n_simulations)
16    S_T = S * np.exp((r - 0.5*sigma**2)*T + sigma*np.sqrt(T)*Z)
17    
18    # 计算期权收益
19    payoffs = np.maximum(S_T - K, 0)
20    
21    # 折现期望收益
22    price = np.exp(-r*T) * np.mean(payoffs)
23    std_error = np.exp(-r*T) * np.std(payoffs) / np.sqrt(n_simulations)
24    
25    return price, std_error
26
27# 参数
28S, K, T, r, sigma = 100, 100, 1, 0.05, 0.20
29
30bs_price = black_scholes_call(S, K, T, r, sigma)
31mc_price, mc_se = monte_carlo_call(S, K, T, r, sigma)
32
33print(f"Black-Scholes 价格: £{bs_price:.4f}")
34print(f"Monte Carlo 价格: £{mc_price:.4f} ± £{mc_se:.4f}")
35print(f"误差: {abs(mc_price - bs_price):.4f}")

1import numpy as np
2from statsmodels.tsa.stattools import adfuller
3
4# 模拟两种时间序列
5np.random.seed(42)
6n = 500
7
8# 非平稳序列（随机游走）
9random_walk = np.cumsum(np.random.randn(n))
10
11# 平稳序列（AR(1) with |φ| < 1）
12ar1 = np.zeros(n)
13for t in range(1, n):
14    ar1[t] = 0.7 * ar1[t-1] + np.random.randn()
15
16# ADF 检验
17for name, series in [("随机游走", random_walk), ("AR(1)", ar1)]:
18    result = adfuller(series)
19    print(f"\n{name}:")
20    print(f"  ADF 统计量: {result[0]:.4f}")
21    print(f"  p-value: {result[1]:.4f}")
22    is_stationary = result[1] < 0.05; print("  结论: " + ("平稳" if is_stationary else "非平稳"))

1import numpy as np
2import pandas as pd
3
4def momentum_backtest(prices_df, lookback=252, holding=21):
5    """
6    动量策略回测
7    prices_df: DataFrame，行为日期，列为股票
8    lookback: 动量计算窗口（交易日）
9    holding: 持仓周期（交易日）
10    """
11    returns = prices_df.pct_change()
12    
13    # 计算动量信号（过去 lookback 日收益率）
14    momentum = prices_df.pct_change(lookback)
15    
16    # 每 holding 天重新平衡
17    portfolio_returns = []
18    
19    for t in range(lookback, len(prices_df) - holding, holding):
20        # 选择动量最强的前 20% 股票（做多）
21        signal = momentum.iloc[t]
22        n_stocks = len(signal.dropna())
23        n_long = max(1, n_stocks // 5)
24        
25        long_stocks = signal.nlargest(n_long).index
26        
27        # 等权重持仓
28        period_returns = returns.iloc[t:t+holding][long_stocks].mean(axis=1)
29        portfolio_returns.extend(period_returns.tolist())
30    
31    portfolio_returns = np.array(portfolio_returns)
32    
33    # 计算绩效指标
34    annual_return = np.mean(portfolio_returns) * 252
35    annual_vol = np.std(portfolio_returns) * np.sqrt(252)
36    sharpe = annual_return / annual_vol
37    
38    # 最大回撤
39    cumulative = np.cumprod(1 + portfolio_returns)
40    running_max = np.maximum.accumulate(cumulative)
41    drawdowns = (cumulative - running_max) / running_max
42    max_drawdown = drawdowns.min()
43    
44    return {
45        'annual_return': annual_return,
46        'annual_vol': annual_vol,
47        'sharpe_ratio': sharpe,
48        'max_drawdown': max_drawdown
49    }
50
51# 模拟测试
52np.random.seed(42)
53n_days, n_stocks = 1260, 50  # 5 年数据，50 支股票
54prices = pd.DataFrame(
55    100 * np.exp(np.cumsum(np.random.randn(n_days, n_stocks) * 0.01, axis=0)),
56    columns=[f'Stock_{i}' for i in range(n_stocks)]
57)
58
59results = momentum_backtest(prices)
60for key, val in results.items():
61    print(f"{key}: {val:.2%}")

1import numpy as np
2from statsmodels.regression.linear_model import OLS
3from statsmodels.tsa.stattools import coint
4
5def pairs_trading_signal(price_a, price_b, window=60, entry_z=2.0, exit_z=0.5):
6    """
7    配对交易信号生成
8    返回：+1（做多价差），-1（做空价差），0（无仓位）
9    """
10    n = len(price_a)
11    signals = np.zeros(n)
12    
13    for t in range(window, n):
14        # 在滚动窗口内估计对冲比率
15        y = price_a[t-window:t]
16        x = price_b[t-window:t]
17        
18        # OLS 回归：price_a = β × price_b + ε
19        beta = np.cov(y, x)[0, 1] / np.var(x)
20        spread = y - beta * x
21        
22        # 计算当前价差的 z-score
23        spread_mean = np.mean(spread)
24        spread_std = np.std(spread)
25        current_spread = price_a[t] - beta * price_b[t]
26        z_score = (current_spread - spread_mean) / spread_std
27        
28        # 生成信号
29        if z_score > entry_z:
30            signals[t] = -1  # 价差过高，做空价差（卖 A 买 B）
31        elif z_score < -entry_z:
32            signals[t] = 1   # 价差过低，做多价差（买 A 卖 B）
33        elif abs(z_score) < exit_z:
34            signals[t] = 0   # 价差回归，平仓
35        else:
36            signals[t] = signals[t-1]  # 保持前一仓位
37    
38    return signals
39
40# 测试
41np.random.seed(42)
42n = 500
43common_factor = np.cumsum(np.random.randn(n) * 0.01)
44price_a = 100 * np.exp(common_factor + np.cumsum(np.random.randn(n) * 0.005))
45price_b = 100 * np.exp(common_factor + np.cumsum(np.random.randn(n) * 0.005))
46
47signals = pairs_trading_signal(price_a, price_b)
48print(f"做多信号次数: {(signals == 1).sum()}")
49print(f"做空信号次数: {(signals == -1).sum()}")