从练习到一个库
你已经写了上百个小函数。库就是把这些函数组织起来，让别人（包括半年后的你自己）能放心使用：清晰的公共接口、每个部分单独测试、各部分组合成流水线，整体可安装、有文档、并被自动检查。这个毕业项目就是这种模式的缩影。策略库是练习它最容易的地方：信号把价格变成仓位，回测把仓位变成扣除成本后的收益，指标总结收益，运行器把它们串起来。每个阶段接收朴素的数组、返回朴素的数组，因此每一个都能用手算可验证的数字单独测试。
# The shape of a small library: a package, its tests, its metadata and its automated checks
tree = """\
quantlib/
├── pyproject.toml name, version, dependencies (numpy)
├── src/quantlib/
│ ├── __init__.py the public API: sma_positions, strategy_returns, performance_summary, run_strategy
│ ├── signals.py prices -> positions
│ ├── backtest.py positions -> returns after costs
│ └── metrics.py returns -> total return, volatility, Sharpe, drawdown
├── tests/ one test file per module, fixed seeds, hand-checked cases
└── .github/workflows/ci.yml lint, types, tests on every push
"""
print(tree.splitlines()[0]) # quantlib/
print(sum(1 for line in tree.splitlines() if ".py" in line)) # 4 the four module files信号与回测：无前视、含成本
这里的信号是均线交叉：价格的快均线高于慢均线时做多，低于时做空，二者相等或慢均线尚不存在时空仓。决定回测是否诚实的两条规则就体现在代码里。第一，无前视：在第 t 日收盘时决定的仓位只能使用到第 t 日的价格，并赚取第 t + 1 日的收益，所以第 t 步的收益使用 positions[t - 1]。可以用测试证明：去掉最后几个价格不应改变之前的仓位。第二，成本：每次改变仓位都要付 cost * |变化量|，所以不断翻转的策略可能在扣费前盈利、扣费后亏损。
# The signal and the honest backtest: positions decided at the close are earned on the NEXT step, and trading costs money
import numpy as np
def sma_positions(prices, fast, slow):
prices = np.asarray(prices, dtype=float)
positions = np.zeros(len(prices), dtype=int)
for t in range(slow - 1, len(prices)):
f = prices[t - fast + 1 : t + 1].mean()
s = prices[t - slow + 1 : t + 1].mean()
positions[t] = 1 if f > s else -1 if f < s else 0
return positions
def strategy_returns(prices, positions, cost):
returns = []
for t in range(1, len(prices)):
before = positions[t - 2] if t >= 2 else 0
returns.append(positions[t - 1] * (prices[t] / prices[t - 1] - 1) - cost * abs(positions[t - 1] - before))
return np.array(returns)
prices = np.array([100, 101, 103, 102, 105, 108, 107, 110, 112, 111, 115, 118], dtype=float)
positions = sma_positions(prices, 2, 4)
print(positions) # [0 0 0 1 1 1 1 1 1 1 1 1] flat until the 4-day average exists, then long
print(np.allclose(sma_positions(prices[:8], 2, 4), positions[:8])) # True later prices never change earlier positions
print(np.round(strategy_returns(prices, positions, 0.0).sum() - strategy_returns(prices, positions, 0.001).sum(), 3)) # 0.001 one purchase at 0.1%绩效数字与运行器
收益序列用几个数字来总结。总收益把各期复合起来，prod(1 + r) - 1；年化收益把它换算成一年，(1 + 总收益)^(每年期数 / n) - 1；年化波动率是样本标准差乘以 sqrt(每年期数)；夏普比率是平均收益除以该标准差并年化，是衡量单位风险收益的主要指标（平坦序列为 0，而不是除法错误）；最大回撤是权益曲线从峰值起的最大跌幅，它告诉你这个策略要你承受多少痛苦。运行器函数把一切串起来：先仓位，再收益，最后汇总。它应当只组合其他函数，这样出现 bug 总能追溯到某一个阶段。
# Performance statistics of a return series, and the runner that wires the three stages together
import math
import numpy as np
def performance_summary(returns, periods_per_year):
returns = np.asarray(returns, dtype=float)
n = len(returns)
equity = np.concatenate(([1.0], np.cumprod(1 + returns)))
total = float(equity[-1] - 1)
deviation = float(np.std(returns, ddof=1))
peak = np.maximum.accumulate(equity)
return {
"total_return": total,
"annual_return": float((1 + total) ** (periods_per_year / n) - 1),
"annual_vol": deviation * math.sqrt(periods_per_year),
"sharpe": float(returns.mean() / deviation * math.sqrt(periods_per_year)) if deviation > 0 else 0.0,
"max_drawdown": float(np.max((peak - equity) / peak)),
}
summary = performance_summary([0.10, -0.20, 0.05], 252)
print(round(summary["total_return"], 4), round(summary["max_drawdown"], 4)) # -0.076 0.2 equity 1 -> 1.1 -> 0.88 -> 0.924: a 20% fall from the peak
print(performance_summary([0.0, 0.0, 0.0], 252)["sharpe"]) # 0.0 no variation: a Sharpe of 0, not an error# The runner only composes: if a number looks wrong you can test each stage on its own
def run_strategy(prices, fast, slow, cost, periods_per_year=252):
positions = sma_positions(prices, fast, slow)
returns = strategy_returns(prices, positions, cost)
return performance_summary(returns, periods_per_year)
# (sma_positions, strategy_returns and performance_summary are the functions you wrote in this lab)
steps = ["sma_positions", "strategy_returns", "performance_summary"]
print(" -> ".join(steps)) # sma_positions -> strategy_returns -> performance_summary
print(len(steps), "stages, one result: a dict of five numbers") # 3 stages, one result: a dict of five numbers你的毕业项目
现在在你自己的仓库里构建你自己的项目。从两个方向中选一个。定价库：Black-Scholes 与希腊字母，带对偶变量的蒙特卡洛定价器，隐含波动率求解器，以及把一个模型校准到一组报价。或者策略库：多个信号，带成本和仓位限制的回测，前向滚动评估和一份报告。无论哪个，都要交付同样的东西：公共 API 清晰、带类型提示的包；使用固定种子和手工验证用例的测试；一个 pyproject.toml 和展示如何安装并运行示例的 README；运行 lint、类型检查和测试的 CI 工作流；一份包含一个已分析瓶颈的性能说明；以及一篇关于假设和什么会使模型失效的简短说明。安排好时间：先设计接口，建出最小的可运行流水线，再逐步扩展。
- 库是在清晰接口后面、由小而经过测试的函数组成的流水线：信号→仓位→扣费后收益→指标，由运行器串联。
- 诚实的回测没有前视（t 时刻的仓位只用到 t 为止的价格并赚取下一期收益），并对每次仓位变化收费。
- 用总收益、年化收益、波动率、夏普和最大回撤来总结；然后交付带测试、CI、README 和坦诚说明的包。
路线图到此完成。你现在具备了一名在职量化开发者所需的语言、数值工具、行情数据与执行组件、模型以及工程习惯。请保留你的毕业项目仓库：它是你在这里所学技能的最佳证明。
实验：评分内容
- 用
sma_positions把趋势变成仓位25 分 - 在
strategy_returns中带成本回测25 分 - 用
performance_summary评判结果25 分 - 在
run_strategy中把一切串起来25 分