事件驱动回测器

一次一根K线

事件驱动回测器的行为像一个实盘系统：时间逐根K线推进，每一根K线上程序只能看到过去，做出决策、发出订单并更新账户。它比向量化回测慢，但它携带状态（现金、股数），可以模拟订单和成交，同一份策略代码以后还能跑在实时行情上。循环是关键：对K线做 for，账户从一次迭代传到下一次。

def crossover_signal(history, fast, slow):
    if len(history) < slow:
        return 0                                           # not enough history yet
    fast_mean = sum(history[-fast:]) / fast
    slow_mean = sum(history[-slow:]) / slow
    return 1 if fast_mean > slow_mean else 0

prices = [100, 101, 103, 102, 105, 107, 104, 103, 99, 98]
print(crossover_signal(prices[:3], 2, 4))                  # 0: only 3 prices
print(crossover_signal(prices[:6], 2, 4))                  # 1: the 2-bar mean is above the 4-bar mean
print(crossover_signal(prices[:9], 2, 4))                  # 0: the trend has turned

订单会改变账户

订单把现金换成股票，或反过来。股数为整数时，买入 int(cash // price) 股会剩下一点现金，账户价值始终是 cash + shares * price。把订单逻辑放进一个小函数 execute_order，循环就会很短，之后还能在一处加入手续费和滑点。全部卖出就是现金变为 cash + shares * price，股数为零。

def execute_order(cash, shares, price, target):
    if target == 1:
        bought = int(cash // price)                        # whole shares only
        return cash - bought * price, shares + bought
    return cash + shares * price, 0

cash, shares = execute_order(1000.0, 0, 103.0, 1)
print(cash, shares, cash + shares * 103.0)                 # 73.0 9 1000.0
cash, shares = execute_order(cash, shares, 99.0, 0)
print(cash, shares)                                        # 964.0 0

循环，以及它能知道什么

在第 t 根K线上，策略可以使用到上一次收盘为止的 prices[:t]，并以 prices[t] 成交。对 prices[:t] 切片就把未来挡在了外面：下一个价格根本不在列表里。由此得到检验事件驱动引擎的好办法：只对前 k 根K线运行，结果必须等于完整运行结果的前 k 个。若不同，说明策略偷看了。

def crossover_signal(history, fast, slow):
    if len(history) < slow:
        return 0
    return 1 if sum(history[-fast:]) / fast > sum(history[-slow:]) / slow else 0

def execute_order(cash, shares, price, target):
    if target == 1:
        bought = int(cash // price)
        return cash - bought * price, shares + bought
    return cash + shares * price, 0

def run_backtest(prices, fast, slow, cash):
    shares = 0
    values = []
    for t, price in enumerate(prices):
        if t > 0:
            target = crossover_signal(prices[:t], fast, slow)       # only what was known at the last close
            cash, shares = execute_order(cash, shares, price, target)
        values.append(cash + shares * price)
    return values

prices = [100, 101, 103, 102, 105, 107, 104, 103, 99, 98]
full = run_backtest(prices, 2, 4, 1000.0)
print([round(v, 2) for v in full])
print(run_backtest(prices[:6], 2, 4, 1000.0) == full[:6])             # True: the future never leaks

# The benchmark: buy once at the start and hold
import random

def buy_and_hold(prices, cash):
    shares = int(cash // prices[0])
    left = cash - shares * prices[0]
    return [left + shares * price for price in prices]

random.seed(3)
prices = [100.0]
for _ in range(99):
    prices.append(round(prices[-1] * (1 + random.gauss(0.0004, 0.012)), 2))
held = buy_and_hold(prices, 10000.0)
print(round(held[0], 2), round(held[-1], 2))

• 事件驱动回测器带着状态（现金、股数）逐根循环；在第 t 根只能使用 prices[:t]。
• 订单在现金和整数股之间转换：账户价值是 cash + shares * price；把订单逻辑放进一个函数。
• 检验前视偏差：只跑前 k 根的结果必须等于完整运行的前 k 个结果；并与买入持有作比较。

下一节：前视偏差，让回测好得不真实的那个隐蔽错误。

实验：评分内容

• 用 crossover_signal 比较两条均线25 分
• 用 execute_order 执行订单25 分
• 用 run_backtest 逐根K线推进25 分
• 用 buy_and_hold 设定基准25 分