《量化交易入门》第 12 章：从零到一的行动清单——量化之路怎么走#

1
# -*- coding: utf-8 -*-
2
"""最小量化流水线：数据 → 信号 → 回测（含手续费）→ 绩效"""
3
import random
4

5
random.seed(7)
6

7
# ① 数据：没有 API 时先用模拟数据练手（真实项目换成行情接口）
8
prices = [100.0]
9
for _ in range(240):
10
    prices.append(max(1.0, prices[-1] * (1 + random.gauss(0.0004, 0.02))))
11

12
def ma(data, n):
13
    out = [None] * len(data)
14
    for i in range(n - 1, len(data)):
15
        out[i] = sum(data[i - n + 1:i + 1]) / n
16
    return out
17

18
# ② 信号：双均线金叉买入、死叉卖出
19
fast, slow = ma(prices, 5), ma(prices, 20)
20
signals = [0] * len(prices)
21
for i in range(1, len(prices)):
22
    if fast[i] and slow[i] and fast[i - 1] and slow[i - 1]:
23
        if fast[i - 1] <= slow[i - 1] and fast[i] > slow[i]:
24
            signals[i] = 1
25
        elif fast[i - 1] >= slow[i - 1] and fast[i] < slow[i]:
26
            signals[i] = -1
27

28
# ③ 回测：全仓买卖，扣万三手续费
29
cash, position, equity_curve = 100000.0, 0.0, []
30
fee_rate = 0.0003
31
trades = []
32
for i, sig in enumerate(signals):
33
    if sig == 1 and position == 0:
34
        position = cash * (1 - fee_rate) / prices[i]
35
        cash = 0.0
36
        trades.append(("买", i, prices[i]))
37
    elif sig == -1 and position > 0:
38
        cash = position * prices[i] * (1 - fee_rate)
39
        position = 0.0
40
        trades.append(("卖", i, prices[i]))
41
    equity_curve.append(cash + position * prices[i])
42

43
# ④ 绩效：总收益、交易次数、最大回撤
44
total_ret = equity_curve[-1] / 100000.0 - 1
45
peak, max_dd = equity_curve[0], 0.0
46
for e in equity_curve:
47
    peak = max(peak, e)
48
    max_dd = max(max_dd, (peak - e) / peak)
49

50
print(f"交易次数: {len(trades)}")
51
print(f"总收益率: {total_ret:.2%}")
52
print(f"最大回撤: {max_dd:.2%}")
53
print("前五笔交易:", [(a, b, round(c, 2)) for a, b, c in trades[:5]])

1
交易次数: 16
2
总收益率: -4.91%
3
最大回撤: 15.52%
4
前五笔交易: [('买', 31, 101.63), ('卖', 41, 98.3), ('买', 59, 99.91), ('卖', 66, 91.45), ('买', 85, 91.25)]