主题切换
20.2 编程与算法题
编程是量化面试的核心考察维度——毕竟量化交易的本质是将交易思想转化为代码。编程面试通常分为三个层次：LeetCode 风格的算法题、Pandas/NumPy 数据处理题、以及系统设计题。本章覆盖各层次的高频考点和实战技巧。
LeetCode高频题在量化面试中的变形
量化面试中的算法题与科技公司面试有显著差异——更注重数值计算、概率相关算法和高效的数据处理，而不是纯粹的图论或动态规划。
典型高频题
1. 蓄水池抽样（Reservoir Sampling）
从一个未知大小的数据流中均匀随机选取 k 个样本。这在处理 tick 级行情数据时非常实用。
此处有展示代码展开 ▼
python
import random
def reservoir_sampling(stream, k: int) -> list:
"""
蓄水池抽样算法
从流式数据中均匀地随机抽取 k 个元素
时间复杂度: O(n), 空间复杂度: O(k)
"""
reservoir = []
for i, item in enumerate(stream):
if i < k:
reservoir.append(item)
else:
# 以 k/(i+1) 的概率替换
j = random.randint(0, i)
if j < k:
reservoir[j] = item
return reservoir
# 验证均匀性
def verify_reservoir_sampling(n_trials: int = 10000,
stream_size: int = 100,
k: int = 10) -> dict:
"""验证蓄水池抽样的均匀性"""
counts = {i: 0 for i in range(stream_size)}
for _ in range(n_trials):
stream = range(stream_size)
sampled = reservoir_sampling(stream, k)
for item in sampled:
counts[item] += 1
expected = n_trials * k / stream_size
chi_sq = sum((c - expected) ** 2 / expected for c in counts.values())
return {'expected_count': expected, 'chi_squared': chi_sq, 'counts': counts}
random.seed(42)
# 1) 单次抽样看结果
sample = reservoir_sampling(range(1000), k=10)
print(f"从 0..999 的流中抽 10 个: {sorted(sample)}")
# 2) 均匀性验证
res = verify_reservoir_sampling(n_trials=10000, stream_size=100, k=10)
counts = res['counts']
expected = res['expected_count']
chi_sq = res['chi_squared']
print(f"\n均匀性验证 (10000 次试验, 流长 100, k=10):")
print(f" 每个元素的期望入选次数: {expected:.0f}")
print(f" 实际最少 / 最多: {min(counts.values())} / {max(counts.values())}")
print(f" 卡方统计量: {chi_sq:.2f} (自由度 99)")
# 自由度 99 时 chi2 的 95% 临界值约 123.2；均匀分布下 chi_sq 应在 99 附近
verdict = "通过（无法拒绝均匀假设）" if chi_sq < 123.2 else "未通过（分布可疑）"
print(f" 95% 临界值 123.2 → {verdict}")
print("\n 头部/尾部元素入选次数抽查（应大致相同，无位置偏好）:")
for i in [0, 1, 2, 49, 50, 97, 98, 99]:
bar = '#' * int(counts[i] / expected * 30)
print(f" 元素 {i:>3}: {counts[i]:>5} 次 {bar}")点击展开可浏览运行结果
从 0..999 的流中抽 10 个: [18, 81, 235, 288, 430, 538, 562, 599, 663, 989]
均匀性验证 (10000 次试验, 流长 100, k=10):
每个元素的期望入选次数: 1000
实际最少 / 最多: 916 / 1084
卡方统计量: 72.09 (自由度 99)
95% 临界值 123.2 → 通过（无法拒绝均匀假设）
头部/尾部元素入选次数抽查（应大致相同，无位置偏好）:
元素 0: 947 次 ############################
元素 1: 1032 次 ##############################
元素 2: 978 次 #############################
元素 49: 1084 次 ################################
元素 50: 1005 次 ##############################
元素 97: 999 次 #############################
元素 98: 1012 次 ##############################
元素 99: 965 次 ############################2. 快速幂（Fast Exponentiation）
在期权定价和蒙特卡洛模拟中经常需要计算
此处有展示代码展开 ▼
python
def fast_pow(base: float, exp: int) -> float:
"""
二分快速幂算法
时间复杂度: O(log n)
"""
result = 1.0
current = base
while exp > 0:
if exp & 1: # 当前位为1
result *= current
current *= current # 底数平方
exp >>= 1 # 右移一位
return result
def matrix_pow(matrix: list, exp: int) -> list:
"""矩阵快速幂（用于马尔可夫链状态转移等场景）"""
n = len(matrix)
# 单位矩阵
result = [[1.0 if i == j else 0.0 for j in range(n)] for i in range(n)]
base = [row[:] for row in matrix]
while exp > 0:
if exp & 1:
# result = result * base
new_result = [[0.0] * n for _ in range(n)]
for i in range(n):
for j in range(n):
new_result[i][j] = sum(result[i][k] * base[k][j] for k in range(n))
result = new_result
# base = base * base
new_base = [[0.0] * n for _ in range(n)]
for i in range(n):
for j in range(n):
new_base[i][j] = sum(base[i][k] * base[k][j] for k in range(n))
base = new_base
exp >>= 1
return result点击展开可浏览运行结果
📘 本段代码定义了 2 个函数/类：函数 `fast_pow`（二分快速幂算法）、函数 `matrix_pow`（矩阵快速幂（用于马尔可夫链状态转移等场景））。该片段为教学展示（未包含独立运行的输入数据），可在实战练习中结合真实数据调用。
3. 在线中位数（数据流中位数）
在实时风控中，需要维护滑动窗口内的中位数来进行异常检测。使用两个堆可以实现
此处有展示代码展开 ▼
python
import heapq
class MedianFinder:
"""数据流中位数维护器"""
def __init__(self):
self.max_heap = [] # 存储较小的一半（取负实现大根堆）
self.min_heap = [] # 存储较大的一半（小根堆）
def add(self, num: float):
"""添加一个数值"""
if not self.max_heap or num <= -self.max_heap[0]:
heapq.heappush(self.max_heap, -num)
else:
heapq.heappush(self.min_heap, num)
# 平衡两个堆（大小差不超过1）
if len(self.max_heap) > len(self.min_heap) + 1:
heapq.heappush(self.min_heap, -heapq.heappop(self.max_heap))
elif len(self.min_heap) > len(self.max_heap):
heapq.heappush(self.max_heap, -heapq.heappop(self.min_heap))
def median(self) -> float:
"""查询当前中位数"""
if len(self.max_heap) > len(self.min_heap):
return -self.max_heap[0]
else:
return (-self.max_heap[0] + self.min_heap[0]) / 2.0点击展开可浏览运行结果
📘 本段代码定义了 1 个函数/类：类 `MedianFinder`（数据流中位数维护器）。该片段为教学展示（未包含独立运行的输入数据），可在实战练习中结合真实数据调用。
数据处理题：Pandas/NumPy核心操作
这是量化面试中最具区分度的环节——考察你能否熟练地用 Python 处理真实的金融数据。
Pandas 核心技巧
1. 高效的多股票收益率计算
此处有展示代码展开 ▼
python
import pandas as pd
import numpy as np
def efficient_return_calculation(prices: pd.DataFrame,
periods: list = [1, 5, 20]) -> pd.DataFrame:
"""
快速计算多个周期的收益率并处理异常值
Parameters
----------
prices : pd.DataFrame
index=date, columns=stock_codes, values=close_price
periods : list
需要计算的收益周期列表
"""
result = pd.DataFrame(index=prices.index)
for p in periods:
# 使用 pct_change 的 fill_method=None 避免前视偏差
ret = prices.pct_change(p, fill_method=None)
# 异常值处理：Winsorize
for col in ret.columns:
lower = ret[col].quantile(0.01)
upper = ret[col].quantile(0.99)
ret[col] = ret[col].clip(lower, upper)
result[f'ret_{p}d'] = ret.stack()
return result点击展开可浏览运行结果
📘 本段代码定义了 1 个函数/类：函数 `efficient_return_calculation`（快速计算多个周期的收益率并处理异常值）。该片段为教学展示（未包含独立运行的输入数据），可在实战练习中结合真实数据调用。
2. 滚动因子计算
此处有展示代码展开 ▼
python
def rolling_factor_calculation(data: pd.DataFrame) -> pd.DataFrame:
"""
计算常用的滚动因子
演示如何用 rolling + transform 高效计算
"""
df = data.copy()
# 动量因子：最近20日累计收益
df['momentum_20d'] = df.groupby('code')['close'].transform(
lambda x: x.pct_change(20, fill_method=None)
)
# 波动率因子：最近20日收益标准差
df['volatility_20d'] = df.groupby('code')['close'].transform(
lambda x: x.pct_change().rolling(20).std()
)
# 换手率加权因子：最近5日与20日成交量的比值
df['volume_ratio'] = df.groupby('code')['volume'].transform(
lambda x: x.rolling(5).mean() / x.rolling(20).mean()
)
# 截面排名（cross-sectional rank）
for col in ['momentum_20d', 'volatility_20d', 'volume_ratio']:
df[f'{col}_rank'] = df.groupby('date')[col].rank(pct=True)
return df点击展开可浏览运行结果
📘 本段代码定义了 1 个函数/类：函数 `rolling_factor_calculation`（计算常用的滚动因子）。该片段为教学展示（未包含独立运行的输入数据），可在实战练习中结合真实数据调用。
3. DataFrame 合并与对齐的高级操作
此处有展示代码展开 ▼
python
import numpy as np
import pandas as pd
def advanced_merge_operations():
"""量化中最常被追问的三个 DataFrame 合并/对齐场景。"""
# ---------- 场景1：(date, code) 双键对齐，行情与因子的并集 ----------
quotes = pd.DataFrame({
'date': pd.to_datetime(['2024-01-02', '2024-01-02', '2024-01-03', '2024-01-03']),
'code': ['600519', '000001', '600519', '000001'],
'close': [1680.0, 10.5, 1702.0, 10.4],
})
factors = pd.DataFrame({
'date': pd.to_datetime(['2024-01-02', '2024-01-03', '2024-01-03']),
'code': ['600519', '600519', '300750'], # 000001 缺因子，300750 缺行情
'value': [0.82, 0.91, -0.35],
})
merged = quotes.merge(factors, on=['date', 'code'], how='outer')
print("【场景1】(date, code) 外连接 —— 缺口一目了然")
print(merged.to_string(index=False))
print(f" close 缺失 {merged['close'].isna().sum()} 行，"
f"value 缺失 {merged['value'].isna().sum()} 行")
# 因子按代码前值填充（注意先排序，且绝不能跨代码填充）
merged['value_ffill'] = merged.sort_values('date').groupby('code')['value'].ffill()
print(f" 按 code 分组 ffill 后 value 缺失降至 {merged['value_ffill'].isna().sum()} 行\n")
# ---------- 场景2：merge_asof —— 委托时间戳配最近一笔行情 ----------
orders = pd.DataFrame({
'timestamp': pd.to_datetime(['2024-01-02 09:30:05', '2024-01-02 09:30:12',
'2024-01-02 09:30:31']),
'code': ['600519', '600519', '600519'],
'side': ['BUY', 'SELL', 'BUY'],
})
prices = pd.DataFrame({
'timestamp': pd.to_datetime(['2024-01-02 09:30:00', '2024-01-02 09:30:10',
'2024-01-02 09:30:20', '2024-01-02 09:30:30']),
'code': ['600519'] * 4,
'price': [1680.0, 1681.5, 1680.8, 1682.2],
})
asof = pd.merge_asof(orders, prices, on='timestamp', by='code', direction='backward')
print("【场景2】merge_asof(direction='backward') —— 只用委托时刻已知的行情")
print(asof.to_string(index=False))
print(" 用 direction='forward' 就变成未来函数：拿到了下一笔还没发生的价格\n")
# ---------- 场景3：截面标准化，transform 优于 apply ----------
np.random.seed(7)
panel = pd.DataFrame({
'date': np.repeat(pd.to_datetime(['2024-01-02', '2024-01-03']), 4),
'code': ['600519', '000001', '300750', '601318'] * 2,
'factor': np.random.randn(8).round(3),
})
panel['z'] = panel.groupby('date')['factor'].transform(lambda x: (x - x.mean()) / x.std())
print("【场景3】groupby.transform 做截面 z-score（形状不变，可直接赋列）")
print(panel.to_string(index=False))
chk = panel.groupby('date')['z'].agg(['mean', 'std']).round(6)
print(" 每个截面 z 的均值/标准差:")
print(chk.to_string())
print(" apply 返回的是嵌套结构，需要 reset_index 才能贴回原表 —— 这是面试常考的差别")
return merged, asof, panel
advanced_merge_operations()点击展开可浏览运行结果
【场景1】(date, code) 外连接 —— 缺口一目了然
date code close value
2024-01-02 000001 10.5 NaN
2024-01-02 600519 1680.0 0.82
2024-01-03 000001 10.4 NaN
2024-01-03 300750 NaN -0.35
2024-01-03 600519 1702.0 0.91
close 缺失 1 行，value 缺失 2 行
按 code 分组 ffill 后 value 缺失降至 2 行
【场景2】merge_asof(direction='backward') —— 只用委托时刻已知的行情
timestamp code side price
2024-01-02 09:30:05 600519 BUY 1680.0
2024-01-02 09:30:12 600519 SELL 1681.5
2024-01-02 09:30:31 600519 BUY 1682.2
用 direction='forward' 就变成未来函数：拿到了下一笔还没发生的价格
【场景3】groupby.transform 做截面 z-score（形状不变，可直接赋列）
date code factor z
2024-01-02 600519 1.691 1.382308
2024-01-02 000001 -0.466 -0.957149
2024-01-02 300750 0.033 -0.415940
2024-01-02 601318 0.408 -0.009219
2024-01-03 600519 -0.789 -0.183790
2024-01-03 000001 0.002 0.764840
2024-01-03 300750 -0.001 0.761242
2024-01-03 601318 -1.755 -1.342293
每个截面 z 的均值/标准差:
mean std
date
2024-01-02 0.0 1.0
2024-01-03 0.0 1.0
apply 返回的是嵌套结构，需要 reset_index 才能贴回原表 —— 这是面试常考的差别NumPy 核心技巧
1. 向量化优于循环
此处有展示代码展开 ▼
python
# 错误做法
def slow_correlation_matrix(returns: np.ndarray) -> np.ndarray:
n = returns.shape[1]
corr = np.zeros((n, n))
for i in range(n):
for j in range(n):
corr[i, j] = np.corrcoef(returns[:, i], returns[:, j])[0, 1]
return corr
# 正确做法
def fast_correlation_matrix(returns: np.ndarray) -> np.ndarray:
return np.corrcoef(returns.T)点击展开可浏览运行结果
📘 本段代码定义了 2 个函数/类：函数 `slow_correlation_matrix`、函数 `fast_correlation_matrix`。该片段为教学展示（未包含独立运行的输入数据），可在实战练习中结合真实数据调用。
2. 广播（Broadcasting）机制
此处有展示代码展开 ▼
python
def broadcasting_examples():
"""
NumPy 广播机制是量化计算加速的关键
"""
# 示例：计算所有股票对的价格比率
prices = np.array([10, 20, 30, 40, 50]) # (5,)
ratios = prices[:, np.newaxis] / prices[np.newaxis, :] # (5, 5)
# 示例：按行/列标准化
returns = np.random.randn(100, 50) # 100天, 50只股票
row_mean = returns.mean(axis=1, keepdims=True) # (100, 1)
row_std = returns.std(axis=1, keepdims=True) # (100, 1)
normalized = (returns - row_mean) / row_std
return normalized点击展开可浏览运行结果
--- 结果 ---
--- broadcasting_examples() ---
ndarray array([[-1.49428528, -0.81031696, 1.53556644, ..., 0.85659982,
-0.77393081, 0.61654908],
[ 0.9614732 , 0.48771949, 0.24524353, ..., -1.75775581,
-0.36113835, 0.73517147],
[ 1.66584358, 0.85300898, -0.16735648, ..., -0.8679565 ,
-0.73868999, -0.98711315],系统设计题
量化面试中的系统设计题聚焦于真实交易系统的架构：
实时行情系统设计
需求：设计一个能处理 5000+ 只股票、每秒数万条 tick 数据的实时行情系统。
关键考虑：
- 数据接入：多路数据源冗余（防止单点故障），使用消息队列（Kafka/Pulsar）解耦。
- 数据存储：
- 热数据（最近数日）：内存数据库（Redis）
- 温数据（最近数月）：列式存储（ClickHouse/DolphinDB）
- 冷数据（历史存档）：对象存储（S3/MinIO）
- 数据分发：发布-订阅模式，策略系统按需订阅。
- 延迟优化：采用零拷贝（zero-copy）、内存映射文件（mmap）、内核旁路（kernel bypass）技术。
- 容错：数据源自动切换、断线重连和回补机制。
回测系统设计
核心模块：
┌──────────────┐ ┌──────────────┐ ┌──────────────┐
│ 数据引擎 │ │ 策略引擎 │ │ 执行模拟 │
│ - 历史数据 │───▶│ - 信号生成 │───▶│ - 订单管理 │
│ - 复权处理 │ │ - 组合构建 │ │ - 成交模拟 │
│ - 数据对齐 │ │ - 风控约束 │ │ - 成本估算 │
└──────────────┘ └──────────────┘ └──────────────┘
│ │
└────────── ┌──────────────┐ ─────────┘
│ 分析引擎 │
│ - 绩效统计 │
│ - 风险分析 │
│ - 归因报告 │
└──────────────┘面试要点：
- 事件驱动架构 vs 向量化架构的取舍
- 如何处理前视偏差（Look-ahead Bias）
- 生存偏差（Survivorship Bias）的修正方法
- 如何在保证灵活性的同时控制计算复杂度
Python常用技巧与陷阱
高频技巧
此处有展示代码展开 ▼
python
# 1. 使用 collections.defaultdict 简化分组逻辑
from collections import defaultdict
by_sector = defaultdict(list)
for stock in stocks:
by_sector[stock['sector']].append(stock)
# 2. 使用 itertools 进行高效迭代
from itertools import combinations, product, chain
# 所有股票的配对组合
pairs = list(combinations(stock_list, 2))
# 3. 使用 functools.lru_cache 缓存计算结果
from functools import lru_cache
@lru_cache(maxsize=128)
def compute_black_scholes(S, K, T, r, sigma, option_type):
# 期权定价计算...
pass
# 4. 使用 dataclass 管理策略参数
from dataclasses import dataclass, field
@dataclass
class StrategyConfig:
name: str
lookback_window: int = 60
rebalance_frequency: str = 'monthly'
max_position: float = 0.10
stop_loss: float = 0.05
parameters: dict = field(default_factory=dict)
# 5. finally 在资源管理中的应用
def process_market_data(filepath: str):
connection = None
try:
connection = Database.connect(filepath)
data = connection.query("SELECT * FROM trades")
return process(data)
except DatabaseError as e:
logger.error(f"数据库错误: {e}")
raise
finally:
if connection:
connection.close()点击展开可浏览运行结果
📘 本段为代码片段（依赖上文变量或外部输入，如 df/data/参数等），无法独立运行
常见陷阱
1. 浮点数精度问题
此处有展示代码展开 ▼
python
# 陷阱
0.1 + 0.2 == 0.3 # False!
# 解决方案
import math
math.isclose(0.1 + 0.2, 0.3) # True
# 或在金融计算中使用 Decimal
from decimal import Decimal
Decimal('0.1') + Decimal('0.2') == Decimal('0.3') # True点击展开可浏览运行结果
（脚本执行完成，未产生可展示的模块级变量或函数定义）
2. 可变默认参数
此处有展示代码展开 ▼
python
# 陷阱：默认参数只计算一次
def bad_append(item, lst=[]):
lst.append(item)
return lst
# 解决方案
def good_append(item, lst=None):
if lst is None:
lst = []
lst.append(item)
return lst点击展开可浏览运行结果
📘 本段代码定义了 2 个函数/类：函数 `bad_append`（解决方案），以及 函数 `good_append`。该片段为教学展示（未包含独立运行的输入数据），可在实战练习中结合真实数据调用。
3. 闭包中的变量绑定
此处有展示代码展开 ▼
python
# 陷阱：lambda 中的 i 是引用，循环结束后的值
funcs = [lambda x: x + i for i in range(10)]
# 所有 funcs 都用 i=9
# 解决方案
funcs = [lambda x, i=i: x + i for i in range(10)]点击展开可浏览运行结果
--- 结果 --- funcs = [<function <lambda> at 0x000001AF647D13A0>, <function <lambda> at 0x000001AF647D1440>, <function <lambda> at 0x000001AF647D14E0>, <function <lambda> at 0x000001AF647D1580>, <function <lambda> at 0x000001AF647D1620>, <function <lambda> at 0x000001AF647D16C0>, <function <lambda> at 0x000001AF647D1760>, <function <lambda> at 0x000001AF647D1800>, <function <lambda> at 0x000001AF647D18A0>, <function <lambda> at 0x000001AF647D1940>]
时间复杂度速查
| 数据结构/操作 | Python 实现 | 时间效率 |
|---|---|---|
| list.append | 动态数组 | |
| list.insert(0) | 需要移动所有元素 | |
| set.add / dict key lookup | 哈希表 | |
| list in (线性扫描) | - | |
| sorted() | Timsort | |
| heapq.heappush/heappop | 二叉堆 | |
| collections.deque.popleft | 双向链表 | |
| bisect (有序列表插入) | 二分查找 |
面试中的编程题不仅考察你能不能写出正确的代码，更考察你是否理解这些代码在量化交易场景中的适用边界。代码的效率、可读性和鲁棒性，三个维度同样重要。