量化交易系统开发实录（四）：测试驱动敏捷开发（AI Agent 辅助）

读者可以把这一篇当作量化交易系统的缺陷导向测试防线。Part2/Part3 的 100 个真实 pitfalls 不应该停留在问题清单里，它们需要被转化为可执行测试、回归用例、测试数据血缘和 CI Gate。测试不是为了堆数量，而是为了让真实缺陷无法再次穿过系统边界。

系列阅读顺序

推荐阅读路径是 Part1 -> Part2 -> Part3 -> Part4 -> Part5 -> Part6 -> Part7。Part4 位于性能优化之前，因为交易系统必须先证明结果可靠，再讨论速度、吞吐和渲染性能。

这一篇聚焦五个问题：

• 为什么量化系统的测试应该从真实缺陷和风险族群反推，而不是从模板化覆盖率反推。
• RED-GREEN-REFACTOR 在交易系统中如何保护 API 语义、边界时间和重构安全。
• AI 生成测试时，哪些责任可以交给 AI，哪些验收标准必须由人签收。
• 边界时间、属性测试、模糊测试、回测一致性和测试数据血缘如何组合成交易系统测试防线。
• AI 如何辅助集成测试和 E2E 测试，同时通过需求契约追踪识别实现偏移。
• CI Gate 如何把单元测试、集成测试、属性测试、边界时间测试和覆盖率报告变成合并门槛。

引子：一个未被测试覆盖的边界情况

上线前最危险的缺陷，通常不是每天都会出现的错误，而是只在特定日期、特定交易时段、特定数据组合下触发的错误。一个典型场景是：同一套策略在周一多出一根 K 线，导致指标窗口错位。普通测试用例全部通过，但真实交易日语义已经被污染。

根因来自香港期货的跨夜交易机制。周五夜盘和周一早盘在交易语义上可能属于同一个交易日，但物理时间跨过了周五、周六、周日和周一。若聚合逻辑只按自然日切分，周五夜盘、跨午夜数据和周一早盘就可能被错误拆分，日线、指标和回测结果都会被连带污染。

这个缺陷暴露的不是“测试数量不够”，而是测试设计没有对准真实风险。有效测试至少要回答三个问题：风险属于语法、状态、并发、时间、数据、GUI、网络还是安全边界；最小复现数据是什么；应该由单元测试、属性测试、集成测试、端到端测试还是 CI Gate 拦截。

第一部分：量化系统为什么难测试

量化系统难测试，不是因为测试框架复杂，而是因为被测对象同时叠加了数据、时间、状态、随机性和外部依赖。

第一类挑战是数据依赖。策略逻辑依赖历史行情，但完整历史数据体量大，不能全部提交到版本控制；真实数据可能包含敏感信息，不能公开分享；交易所也可能调整历史数据，导致旧样本失效。因此测试数据不能只依赖“某一份本地 CSV”，而需要合成数据、脱敏快照、契约测试和数据血缘共同支撑。

第二类挑战是随机性。网络延迟不确定，行情推送存在 jitter，浮点数计算存在精度误差，回测和模拟盘之间还可能因为事件顺序不同产生差异。测试不能只断言“某个值完全等于另一个值”，还需要明确容忍误差、事件顺序和可重复随机种子。

第三类挑战是状态复杂。持仓状态、订单状态、资金状态、分层数据归一处理状态和策略内部指标状态都可能跨事件延续。一个测试如果只看单个函数返回值，就很容易漏掉状态生命周期问题，例如部分成交后取消、夜盘跨日后指标窗口恢复、重连后订阅状态重复等。

第四类挑战是边界情况密集。开盘、收盘、午休、夜盘、半日市、节假日、跨周、跨日 K 线、数据缺失、网络中断、行情延迟和风控冻结都属于高风险边界。真正有效的测试金字塔，应该从这些缺陷入口反推测试层级。

这张图的重点是“缺陷导向”。底层测试覆盖纯函数、指标计算和时间语义，因为这些错误会被上层策略反复放大；中层测试覆盖数据源、聚合器、策略事件和测试替身，因为这里连接多个模块；上层测试覆盖回测、模拟盘和实盘一致性，因为这里验证系统是否真的按业务语义运行；最顶层 CI Gate 负责阻断合并，而不是替代底层测试。

第二部分：TDD 基础：RED-GREEN-REFACTOR

TDD 的核心循环是 RED-GREEN-REFACTOR。对交易系统来说，它的价值不是形式化地“先写测试”，而是用失败测试定义行为边界，用最小实现验证测试有效，再在测试保护下重构。

RED 阶段的目标是用测试定义正确性。以简单移动平均为例，测试先定义 [1, 2, 3, 4, 5] 在 period=3 时的结果是 [None, None, 2.0, 3.0, 4.0]。这个测试应该先失败，原因可以是类不存在、方法不存在或行为不满足预期。失败本身是有效信号：它证明测试确实能抓住当前缺口。

def test_sma_calculates_correctly():
    prices = [1, 2, 3, 4, 5]
    sma = SimpleMovingAverage(period=3)

    result = sma.calculate(prices)

    assert result == [None, None, 2.0, 3.0, 4.0]

GREEN 阶段的目标是用最小实现让测试通过。最小实现甚至可以先硬编码，只要它证明测试本身有效。随后再进入真正实现：前 period - 1 个结果为 None，第一个窗口使用初始和，后续通过滑动窗口增量更新。

class SimpleMovingAverage:
    def __init__(self, period: int):
        self.period = period

    def calculate(self, prices: list[float]) -> list[float | None]:
        if len(prices) < self.period:
            return [None] * len(prices)

        result: list[float | None] = [None] * (self.period - 1)
        window_sum = sum(prices[: self.period])
        result.append(window_sum / self.period)

        for index in range(self.period, len(prices)):
            window_sum += prices[index] - prices[index - self.period]
            result.append(window_sum / self.period)

        return result

REFACTOR 阶段的目标是在不改变外部行为的前提下改进结构。交易系统里的重构尤其需要测试保护，因为指标计算、数据归属和策略信号都可能被局部改动影响。测试持续绿色，才说明重构没有改变已定义行为。

TDD 对交易系统的直接收益可以压缩成一张表：

维度	没有 TDD	有 TDD
代码覆盖率	常见停留在 30% 左右	可稳定推进到 80% 以上
Bug 逃逸率	边界缺陷容易穿透	真实缺陷可沉淀为回归用例
重构信心	不敢改核心模块	可以分阶段重构并持续验证
API 设计	由实现细节反推	由测试场景驱动
文档价值	文档滞后于代码	测试成为可运行文档

第三部分：测试模式与结构

测试结构决定读者能否快速理解一个用例在验证什么。量化系统里最常用的三类结构是 AAA、Given-When-Then 和表格驱动测试。它们解决的是不同层级的问题：AAA 让函数行为清楚，Given-When-Then 让业务场景清楚，表格驱动让边界集合清楚。

AAA 适合函数和服务级测试。Arrange 准备数据和依赖，Act 执行被测操作，Assert 验证结果。下面的骨架虽然简单，但它能防止测试把准备过程、动作和断言混在一起。

# 示意代码，非实际生产代码
def test_xxx():
    """测试描述"""
    # Arrange: 准备测试数据和环境
    input_data = ...
    expected_output = ...

    # Act: 执行被测操作
    actual_output = function_under_test(input_data)

    # Assert: 验证结果
    assert actual_output == expected_output

读者看 AAA 测试时，应该先检查三个问题：输入是否最小，动作是否只有一个，断言是否直接验证业务行为。盈亏计算是最适合 AAA 的例子，因为它的输入、动作和输出边界非常清楚。

# 示意代码，非实际生产代码
def test_calculate_profit():
    """测试盈亏计算"""
    # Arrange
    entry_price = 25000
    exit_price = 25100
    position_size = 10
    expected_profit = 1000  # (25100 - 25000) * 10

    # Act
    actual_profit = calculate_profit(
        entry=entry_price,
        exit=exit_price,
        size=position_size,
    )

    # Assert
    assert actual_profit == expected_profit

这段代码的价值不在于公式复杂，而在于它把交易系统里最容易被放大的基础语义固定下来。盈亏计算如果没有独立测试，后面的持仓、风控、回测统计和绩效归因都会建立在不可靠基础上。

Given-When-Then 更适合策略行为和验收场景。它让测试读起来像业务规格：给定某个市场状态，当策略处理事件时，应该产生某个交易行为。

# 示意代码，非实际生产代码
def test_user_story_scenario():
    """场景描述"""
    # Given: 初始状态
    setup_conditions()

    # When: 触发事件
    perform_action()

    # Then: 期望结果
    verify_expected_outcomes()

策略金叉买入就是典型场景。测试不应该只验证某个内部变量变化，而要验证策略面对最后一根 Bar 时，是否生成了可执行的业务信号。

# 示意代码，非实际生产代码
def test_strategy_generates_buy_signal():
    """策略在金叉时生成买入信号"""
    # Given: MA5 上穿 MA10
    strategy = MovingAverageCrossStrategy(fast=5, slow=10)
    historical_data = load_data_with_golden_cross()

    # When: 策略处理最后一根 Bar
    signal = strategy.on_bar(historical_data[-1])

    # Then: 生成买入信号
    assert signal.action == Action.BUY
    assert signal.price == historical_data[-1].close

这段代码把“金叉”从口头描述变成了可运行规格。读者应关注的是 Given 部分是否真的构造出金叉，而不是只靠函数名暗示场景；Then 部分是否验证了动作和价格，而不是只验证返回对象不为空。

表格驱动测试适合大量边界值和等价类。新增场景只需要增加一行数据，pytest 参数化能保证每个用例独立运行，并在失败时显示具体场景。

# 示意代码，非实际生产代码
import pytest

TEST_CASES = [
    # input, expected, description
    ([1, 2, 3], 2.0, "正整数"),
    ([-1, -2, -3], -2.0, "负整数"),
    ([1.5, 2.5, 3.5], 2.5, "浮点数"),
    ([5, 5, 5], 5.0, "相同值"),
    ([1], 1.0, "单元素"),
]

@pytest.mark.parametrize("input_data,expected,desc", TEST_CASES)
def test_average_calculation(input_data, expected, desc):
    """测试平均值计算"""
    result = calculate_average(input_data)
    assert result == expected, f"Failed on {desc}"

表格驱动测试的关键是把“有哪些输入类别”显式写出来。量化系统里可以把价格序列、交易时段、订单状态、数据缺失模式和异常输入都放进表格，而不是复制多段几乎相同的测试。

模式	适用场景	风险拦截点
AAA	纯函数、指标、盈亏计算	输入、动作、输出清晰分离
Given-When-Then	策略信号、用户故事、验收场景	业务语义和行为结果绑定
表格驱动	多组边界值、等价类、异常输入	防止只测一个 happy path

第四部分：测试替身：隔离外部依赖

量化系统大量依赖外部数据源、订单管理器、数据库、日志和交易网关。测试替身的目的不是“mock 一切”，而是隔离不可控依赖，同时保留要验证的行为。Dummy、Stub、Spy、Mock 和 Fake 解决的问题并不相同，混用会让测试变脆。

Dummy 是占位对象，只用于满足参数要求。当前测试不验证日志行为时，logger 可以是一个不会被使用的对象。

# 示意代码，非实际生产代码
def test_dummy_logger_does_not_affect_fetch():
    """Dummy 只填充参数，不参与断言"""
    dummy_logger = object()
    service = DataService(logger=dummy_logger)

    result = service.fetch_data("HSI")

    assert result is not None

这段代码只证明 DataService.fetch_data 不依赖 logger 行为。读者不应该在 Dummy 测试里断言日志调用，否则 Dummy 就被错误地升级成 Mock。

Stub 是固定返回值对象，适合替代真实行情源。策略只需要一个确定价格时，Stub 可以保证测试不依赖真实 API、网络延迟和行情权限。

# 示意代码，非实际生产代码
class StubDataFeed:
    """返回固定价格的行情源"""
    def __init__(self, fixed_price: float):
        self.fixed_price = fixed_price

    def get_price(self, symbol: str) -> float:
        return self.fixed_price

def test_strategy_with_stub_price():
    """使用 Stub 测试策略在固定价格下的行为"""
    stub_feed = StubDataFeed(fixed_price=25000)
    strategy = SimpleStrategy(data_feed=stub_feed)

    signal = strategy.check_signal()

    assert signal is not None

Stub 的读者价值是稳定输入。只要测试目标是策略逻辑，就不应该让真实行情服务成为失败来源。

Spy 记录被测对象与协作者的交互，适合验证下单次数、方向和数量。它不预设复杂期望，只把交互记录下来供断言使用。

# 示意代码，非实际生产代码
class SpyOrderManager:
    """记录所有下单调用的 Spy"""
    def __init__(self):
        self.orders_placed = []

    def place_order(self, symbol: str, side: str, quantity: int):
        self.orders_placed.append({
            "symbol": symbol,
            "side": side,
            "quantity": quantity,
        })

def test_strategy_places_buy_order_once():
    """验证策略只下达一笔 BUY 订单"""
    spy = SpyOrderManager()
    strategy = SignalStrategy(order_manager=spy)

    strategy.on_signal(Signal.BUY)

    assert len(spy.orders_placed) == 1
    assert spy.orders_placed[0]["side"] == "BUY"

这段代码保护的是订单副作用边界。读者应关注“只下一笔”和“方向正确”两个断言，因为重复下单和方向错误都可能进入真实风险。

Mock 预设期望并验证调用行为，适合检查某个方法是否按指定参数调用。Mock 过度使用会让测试绑定实现细节，所以只在交互本身就是行为的一部分时使用。

# 示意代码，非实际生产代码
from unittest.mock import Mock

def test_strategy_calls_data_feed_with_symbol():
    """验证策略按约定请求 HSI 行情"""
    mock_feed = Mock()
    mock_feed.get_price.return_value = 25000

    strategy = Strategy(data_feed=mock_feed)
    strategy.update()

    mock_feed.get_price.assert_called_once_with("HSI")
    assert mock_feed.get_price.call_count == 1

这段代码的关键断言是 assert_called_once_with("HSI")。如果策略请求了错误合约、重复请求或遗漏请求，Mock 能立刻暴露接口交互错误。

Fake 是简化但真实可用的实现，例如内存数据库。Repository 测试使用 Fake 可以保留读写语义，同时避免真实数据库连接、事务和清理成本。

# 示意代码，非实际生产代码
class FakeDatabase:
    """内存数据库 Fake"""
    def __init__(self):
        self.data = {}

    def save(self, key: str, value: dict):
        self.data[key] = value

    def get(self, key: str) -> dict:
        return self.data.get(key)

    def delete(self, key: str):
        if key in self.data:
            del self.data[key]

def test_repository_with_fake_db():
    """使用 Fake 数据库测试 Repository"""
    fake_db = FakeDatabase()
    repo = BarRepository(database=fake_db)

    bar = BarData(symbol="HSI", close=25000)
    repo.save(bar)

    retrieved = repo.get("HSI")
    assert retrieved.close == 25000

Fake 的价值是保留真实语义。读者可以把它理解为“轻量但可运行的替代实现”，它比 Mock 更适合测试读写流程和 Repository 边界。

替身类型	用途	量化系统示例
Dummy	填充不会被使用的参数	不参与断言的 logger
Stub	返回固定响应	固定价格行情源
Spy	记录交互	记录策略下单次数
Mock	验证调用行为	验证行情接口调用参数
Fake	简化真实实现	内存数据库或本地订单簿

第五部分：AI 辅助 TDD 流程

AI 辅助 TDD 不应该让 AI 接管验收标准。更稳妥的分工是：人定义缺陷、业务语义和验收边界；AI 扩展测试框架、生成候选实现、提出重构建议；最终由人审查测试是否真的覆盖风险。

传统 TDD 的三步是人工写测试、人工写实现、人工重构。AI 辅助 TDD 可以改成：人工写规格和关键边界，AI 生成测试框架；AI 生成候选实现，人工 Review；AI 提供重构建议，人工判断是否接受。这样做能提升速度，但不会把业务语义签收权交给 AI。

K 线周期聚合是一个完整样例。规格先定义：给定 1 分钟 K 线数据，生成 5 分钟 K 线；输入包含 open/high/low/close/volume/turnover；输出为 5 分钟 K 线列表。核心规则包括：边界归一到 09:00、09:05、09:10 等周期边界；open 取第一根，high 取最大值，low 取最小值，close 取最后一根，volume 和 turnover 求和。边界情况包括：不足 5 根不生成、跨越 5 分钟边界正确分割、时间戳不落在边界时向下取整。

规格应该先用自然语言锁定业务规则，再交给 AI 扩展测试。下面这段规格不是装饰性文档，它限定了 AI 不能随意解释聚合语义。

## 功能：K 线周期聚合

### 需求
给定 1 分钟 K 线数据，生成 5 分钟 K 线数据。

### 输入
- 1 分钟 K 线列表，包含 open, high, low, close, volume, turnover

### 输出
- 5 分钟 K 线列表

### 规则
1. 边界归一：5 分钟 K 线起始时间为 09:00, 09:05, 09:10 ...
2. 价格计算：
   - open = 第一根 1 分钟线的 open
   - high = max(1 分钟线 high)
   - low = min(1 分钟线 low)
   - close = 最后一根 1 分钟线的 close
3. 成交量累加：volume = sum(1 分钟线 volume)

### 边界情况
- 数据不足 5 根：不生成 K 线
- 数据跨越 5 分钟边界：正确分割
- 时间戳不落在边界：向下取整到 5 分钟边界

读者可以把这段规格当成 AI 生成测试的输入契约。没有它，AI 很容易生成“看起来合理但语义不确定”的测试，例如忽略边界归一或把剩余不足周期的数据也聚合出来。

给 AI 的测试生成 prompt 应该明确测试策略、样本工厂和断言目标。

基于以上规格，生成 pytest 测试代码。

要求：
1. 使用 AAA 模式。
2. 包含正常、边界、异常三种情况。
3. 使用参数化测试覆盖多种输入。
4. 测试数据使用工厂函数生成。
5. 每个测试有清晰的 docstring。

下面的测试框架保留了原版中的关键细节：工厂函数负责生成时间连续的 Bar，正常用例验证 OHLCV 聚合，边界用例验证不足周期不生成，参数化用例验证输入数量和输出数量的关系。

# 示意代码，非实际生产代码
import pytest
from datetime import datetime, timedelta
from typing import List

from core.aggregation import aggregate_bars
from core.data import BarData


class TestAggregateBars:
    """测试 K 线周期聚合"""

    @staticmethod
    def create_bars(count: int, start_time: datetime) -> List[BarData]:
        """工厂函数：创建测试数据"""
        bars = []
        for i in range(count):
            bars.append(
                BarData(
                    symbol="HSI",
                    timestamp=start_time + timedelta(minutes=i),
                    open_price=100 + i,
                    high_price=105 + i,
                    low_price=95 + i,
                    close_price=100 + i + 0.5,
                    volume=1000,
                    turnover=100000,
                )
            )
        return bars

    def test_normal_case(self):
        """正常情况：5 根 1 分钟线合成 1 根 5 分钟线"""
        start_time = datetime(2024, 1, 8, 9, 0)
        bars = self.create_bars(5, start_time)

        result = aggregate_bars(bars, period=5)

        assert len(result) == 1
        assert result[0].open_price == 100
        assert result[0].high_price == 109
        assert result[0].low_price == 95
        assert result[0].close_price == 104.5
        assert result[0].volume == 5000

    def test_insufficient_data(self):
        """边界情况：不足 5 根不生成"""
        start_time = datetime(2024, 1, 8, 9, 0)
        bars = self.create_bars(3, start_time)

        result = aggregate_bars(bars, period=5)

        assert len(result) == 0

    @pytest.mark.parametrize("input_count,expected_count", [
        (5, 1),
        (10, 2),
        (12, 2),  # 剩余 2 根不足
        (0, 0),
    ])
    def test_various_counts(self, input_count, expected_count):
        """参数化测试：不同输入数量"""
        start_time = datetime(2024, 1, 8, 9, 0)
        bars = self.create_bars(input_count, start_time)

        result = aggregate_bars(bars, period=5)

        assert len(result) == expected_count

这段测试的架构价值在于它先定义输出语义，再允许实现变化。只要这些测试存在，后续无论使用循环、NumPy、向量化还是增量状态，都必须保留同样的业务结果。

RED 阶段应该先失败。失败不是坏事，它证明测试确实能捕获当前缺口。

$ pytest tests/test_aggregation.py -v

tests/test_aggregation.py::TestAggregateBars::test_normal_case FAILED
tests/test_aggregation.py::TestAggregateBars::test_insufficient_data FAILED
tests/test_aggregation.py::TestAggregateBars::test_various_counts FAILED

# 失败原因：aggregate_bars 函数不存在

这段输出让读者确认测试不是“事后补绿”。如果测试一开始就通过，很可能是测试没有覆盖真实缺口，或者实现早已存在但未被约束到业务语义。

GREEN 阶段的实现只需要满足已定义测试，不要提前加入过多抽象。下面的实现保留了核心聚合逻辑：按完整周期切片，分别计算 open、high、low、close、volume 和 turnover。

# 示意代码，非实际生产代码
from typing import List

def aggregate_bars(bars: List[BarData], period: int) -> List[BarData]:
    """将分钟 K 线聚合为更大的周期"""
    if len(bars) < period:
        return []

    result = []
    for i in range(0, len(bars) // period * period, period):
        group = bars[i:i + period]
        aggregated = BarData(
            symbol=group[0].symbol,
            timestamp=group[0].timestamp,
            open_price=group[0].open_price,
            high_price=max(b.high_price for b in group),
            low_price=min(b.low_price for b in group),
            close_price=group[-1].close_price,
            volume=sum(b.volume for b in group),
            turnover=sum(b.turnover for b in group),
        )
        result.append(aggregated)

    return result

这段实现只解决当前测试覆盖的行为。读者应注意它仍然有改进空间，例如时间戳边界归一逻辑还没有独立出来，剩余不足周期的处理策略还依赖测试约束。

REFACTOR 阶段再抽出边界归一和聚合构造函数。重构的前提是测试持续通过，目标是让边界语义更容易被下一位维护者理解。

# 示意代码，非实际生产代码
def aggregate_bars(bars: List[BarData], period: int) -> List[BarData]:
    """将分钟 K 线聚合为更大的周期"""
    if len(bars) < period:
        return []

    result = []
    end = len(bars) - len(bars) % period
    for i in range(0, end, period):
        group = bars[i:i + period]
        aligned_time = _align_timestamp(group[0].timestamp, period)
        result.append(_create_aggregated_bar(group, aligned_time))

    return result


def _align_timestamp(timestamp: datetime, period: int) -> datetime:
    """将时间戳归一到周期边界"""
    minute = (timestamp.minute // period) * period
    return timestamp.replace(minute=minute, second=0, microsecond=0)


def _create_aggregated_bar(group: List[BarData], timestamp: datetime) -> BarData:
    """从一组 K 线创建聚合 K 线"""
    return BarData(
        symbol=group[0].symbol,
        timestamp=timestamp,
        open_price=group[0].open_price,
        high_price=max(b.high_price for b in group),
        low_price=min(b.low_price for b in group),
        close_price=group[-1].close_price,
        volume=sum(b.volume for b in group),
        turnover=sum(b.turnover for b in group),
    )

重构后的代码让边界归一成为显式边界。读者应关注 _normalize_boundary_timestamp 是否覆盖交易所真实交易时段；如果后续要支持夜盘、半日市或跨周归属，这个函数会成为重点测试入口。

第六部分：量化系统特有的测试策略

量化测试不能只依赖手写样例。手写样例适合说明业务意图，但很难穷尽输入空间。交易系统至少需要四类专属策略：属性基于测试、模糊测试、回测一致性验证和边界时间测试。

属性基于测试适合验证数学性质。移动平均的核心属性包括：结果长度与输入长度和周期相关；每个均值应落在对应窗口的最小值和最大值之间；当输入序列单调不降时，移动平均也应单调不降。Hypothesis 可以自动生成大量价格序列和周期，帮助发现人工样本漏掉的边界。

# 示意代码，非实际生产代码
from hypothesis import given, strategies as st

@given(
    prices=st.lists(
        st.floats(min_value=1, max_value=100000),
        min_size=10,
        max_size=100,
    ),
    period=st.integers(min_value=2, max_value=20),
)
def test_ma_properties(prices, period):
    """测试移动平均线的数学属性"""
    result = calculate_ma(prices, period)

    assert len(result) == len(prices) - period + 1

    for value in result:
        assert min(prices) <= value <= max(prices)

    if all(prices[i] <= prices[i + 1] for i in range(len(prices) - 1)):
        assert all(result[i] <= result[i + 1] for i in range(len(result) - 1))

这段测试的价值是从“举几个例子”升级为“验证数学性质”。读者应注意第二个断言写成了全局范围约束；如果要更严格，可以把每个均值限定在对应滑动窗口的 min/max 之间。

模糊测试适合验证解析逻辑的健壮性。行情解析器面对任意字节输入时，允许抛出明确的 ParseError，但不应该出现未预期异常；如果解析成功，还要满足 high >= low、high >= open、high >= close 等基本约束。

# 示意代码，非实际生产代码
import atheris
import sys

@atheris.instrument_func
def test_parse_bar(input_bytes):
    """测试行情解析的健壮性"""
    fdp = atheris.FuzzedDataProvider(input_bytes)
    data = fdp.ConsumeBytes(len(input_bytes))

    try:
        bar = parse_bar_data(data)
        if bar:
            assert bar.high >= bar.low
            assert bar.high >= bar.open
            assert bar.high >= bar.close
    except ParseError:
        pass
    except Exception:
        raise

atheris.Setup(sys.argv, test_parse_bar)
atheris.Fuzz()

这段代码让脏数据持续冲击解析器。读者应关注异常边界：业务上可接受的解析失败应该进入 ParseError，未知异常才是需要修复的缺陷。

回测一致性验证用于保证策略在回测和模拟盘中表现一致。这个测试保护的是“回测和实盘走同一套业务语义”，不是单个函数。

# 示意代码，非实际生产代码
def test_strategy_consistency():
    """验证策略在回测和模拟盘中表现一致"""
    historical_data = load_data("HSI", "2024-01-01", "2024-01-31")

    backtest_result = run_backtest(
        strategy=MyStrategy(),
        data=historical_data,
        mode="backtest",
    )

    simulated_result = run_backtest(
        strategy=MyStrategy(),
        data=historical_data,
        mode="simulation",
    )

    assert len(backtest_result.signals) == len(simulated_result.signals)
    for b_sig, s_sig in zip(backtest_result.signals, simulated_result.signals):
        assert b_sig.timestamp == s_sig.timestamp
        assert b_sig.action == s_sig.action
        assert abs(b_sig.price - s_sig.price) < 0.01

这段代码不应该被理解为“回测收益和模拟盘收益完全一样”。更准确的目标是：同一份事件输入下，信号数量、时间戳、动作和价格语义应保持一致，允许的误差必须显式写入断言。

边界时间测试是量化系统最重要的回归防线之一。周五夜盘到周一早盘、半日市、午休、开盘第一根、收盘最后一根、节假日前夕夜盘都应该有固定 fixture。

# 示意代码，非实际生产代码
class TestHKFEBoundaryCases:
    """测试港股期货的边界时间情况"""

    def test_friday_night_to_monday(self):
        """周五夜盘到周一早盘的跨日交易"""
        friday_night = datetime(2024, 1, 5, 23, 0)
        monday_morning = datetime(2024, 1, 8, 9, 15)

        bars = [
            BarData(timestamp=friday_night, close=25000),
            BarData(timestamp=monday_morning, close=25100),
        ]

        result = aggregate_daily(bars)

        assert len(result) == 1
        assert result[0].date == date(2024, 1, 5)

    def test_half_day_holiday(self):
        """半日市提前收盘"""
        half_day_close = datetime(2024, 2, 9, 12, 0)
        bars = generate_bars_until(half_day_close)

        assert all(bar.timestamp <= half_day_close for bar in bars)

这段代码把跨夜、跨周和半日市变成固定回归样本。读者应关注的是交易日归属规则，而不是物理日期是否连续。

图 3 把边界时间看成状态迁移，而不是日期列表。测试需要覆盖从日盘进入午休、从午休恢复、从日盘进入夜盘、跨午夜、跨周归属、半日市提前收盘和假期休市。只要这些状态没有显式测试，聚合器和策略就会继续依赖隐含假设。

第七部分：AI 生成测试的最佳实践

让 AI 生成更好的测试，关键是给它测试策略，而不是只说“补测试”。

第一，明确测试方法。可以要求 AI 使用边界值分析法，分别生成正常范围、边界值和异常输入；也可以要求使用等价类划分，把价格、时间、交易日、订单状态拆成不同类别。

使用边界值分析法，为以下函数生成测试：
- 正常情况：输入在有效范围内
- 边界情况：输入在边界上
- 异常情况：输入超出范围

这段 prompt 的价值在于强制 AI 先组织测试空间。读者可以继续补充“每个用例必须说明风险来源和断言目标”，防止 AI 只给出表面覆盖。

第二，要求 AI 解释测试意图。每个测试都应该说明验证的业务假设，例如“周五 23:30 的夜盘 Bar 应归属于下周一交易日，而不是自然日周五”。

请为以下测试添加注释，解释每个测试的目的和验证的假设。

如果 AI 解释不清楚，它生成的测试通常也不可靠。解释测试意图不是为了增加注释，而是为了让业务假设可审查。

第三，让 AI 识别遗漏。把实现、规格和已有测试交给 AI，让它列出可能遗漏的场景，通常能发现空输入、极大值、极小值、并发访问、资源耗尽和浮点精度等问题。

基于以下规格和实现，列出可能遗漏的测试场景。
请按正常、边界、异常、并发、性能、回归六类输出。
每个场景必须说明风险来源、最小复现数据和断言目标。

这段 prompt 把“补测试”变成了风险分类任务。读者应把 AI 输出当作候选清单，再结合真实缺陷和系统边界决定是否采纳。

第四，用变异测试检查测试质量。mutmut 这类工具会修改源代码，例如把 > 改成 >=，再运行测试。如果测试仍然通过，说明测试没有真正保护行为。

# 示意代码，非实际生产代码
# 安装：pip install mutmut
# 运行：mutmut run

# mutmut 会修改源代码，例如把 > 改成 >=。
# 如果测试仍然通过，说明测试没有真正约束行为。

变异测试不适合每次提交都跑，但适合在关键模块重构前后使用。它能暴露“覆盖率很高但断言很弱”的假象。

第五，让 AI 生成集成测试和 E2E 测试时，必须先限定系统边界。最常见的问题不是 AI 不会写测试代码，而是它把 E2E 写成了“能跑通页面或命令”的 happy path，却没有验证业务链路是否真的闭合。量化系统的集成测试至少要穿过数据源、交易日归属、周期聚合、指标计算、策略信号和订单适配器；E2E 测试还要验证读者真正关心的结果，例如信号时间戳、下单方向、数量、价格容差、回测报告和异常降级路径。

基于以下需求契约生成集成测试和 E2E 测试。

要求：
1. 每个测试必须标注 REQ/INV/E2E ID。
2. 集成测试必须覆盖 DataFeed -> Calendar -> Aggregator -> Strategy -> OrderAdapter。
3. E2E 测试必须验证业务可见结果，不允许只断言页面可打开或命令退出码为 0。
4. 外部行情和交易网关必须使用 Fake 或 Stub，但断言必须覆盖真实业务语义。
5. 每个测试必须输出 evidence 路径、断言对象和失败时的定位线索。

这段 prompt 的关键是把“生成 E2E”改成“按契约生成可追踪 E2E”。读者应避免让 AI 只写浏览器点击、CLI smoke test 或 mock-only 断言。一个合格的 E2E 至少要证明某个需求从输入数据进入系统后，经过真实集成边界，最终产生可审查的业务结果。

第六，需求契约执行完成后，需要有独立的追踪工件，而不是只看测试是否通过。Trace Matrix 应该把需求、约束、不变量、测试、证据和状态放在同一张表里。这样才能识别实现偏移：代码写了但没有对应需求，测试通过但没有证据，E2E 只覆盖 happy path，或者实现路径绕过了契约中定义的边界。

trace:
  - id: TRACE-HKFE-001
    requirement: REQ-TRADING-DAY-OWNERSHIP
    invariant: INV-NATURAL-DAY-MUST-NOT-LEAK
    integration_test: tests/integration/test_hkfe_daily_aggregation.py
    e2e_test: tests/e2e/test_strategy_signal_calendar_boundary.py
    evidence:
      - reports/integration/hkfe_daily_aggregation.xml
      - reports/e2e/strategy_signal_calendar_boundary.json
    drift_signals:
      - implementation_without_requirement
      - passing_test_without_business_assertion
      - mock_only_without_integration_boundary
      - missing_evidence_artifact
    status: PASS

这段追踪工件的价值是让“完成”变成可审查状态。读者可以沿着 TRACE-HKFE-001 看到需求来自哪里、不变量是什么、集成测试和 E2E 测试在哪里、证据文件是什么、哪些偏移信号需要拦截。如果某一项缺失，状态就不应该被改成 PASS。

实现偏移通常有五类信号。第一，代码改动找不到对应 REQ 或 TRACE，说明实现可能跑出了需求边界。第二，测试只验证函数返回或页面存在，没有验证业务语义。第三，E2E 依赖 mock-only 路径，没有穿过关键集成边界。第四，AI 生成的修复让测试变绿，但删除、放宽或绕开了原有断言。第五，最终报告只有“通过”结论，没有可复查的 evidence artifact。

快速纠偏不应该从“继续让 AI 改代码”开始，而应该从失败证据包开始。最小闭环是：定位失败的 TRACE 行，读取对应 REQ/INV/E2E，确认偏移类型，补最小复现或补断言，再让 AI 只修改与该 TRACE 相关的代码和测试。修复后重新运行对应 gate，并更新 evidence 路径和状态。这个过程能把一次大范围返工压缩成围绕单条需求契约的迭代。

偏移信号	风险	纠偏动作
有代码无 REQ/TRACE	实现跑出需求边界	补需求映射或撤回无依据实现
测试无业务断言	happy path 伪通过	增加业务可见结果断言
E2E 只走 mock	集成边界未验证	接入 Fake gateway 和真实数据路径
PASS 无 evidence	完成状态不可复查	生成并记录报告、截图或 JSON 证据
AI 放宽断言	测试质量下降	恢复原断言并补失败用例

这张表的使用方式很直接：每次 AI 声称完成时，先按偏移信号逐项检查，而不是先读代码 diff。交易系统尤其需要这种纪律，因为很多错误不会在单元测试里暴露，而会在数据路径、交易日语义、策略事件和订单适配器的组合处出现。

第八部分：测试数据管理

测试数据管理决定测试是否可复现。量化系统不能把真实行情直接塞进所有测试，也不能只靠随机生成数据。更稳妥的组合是：合成数据覆盖结构性场景，数据快照保存真实缺陷样本，契约测试保护数据源边界。

合成数据适合覆盖趋势、波动率、缺口、跳空和极值。生成器可以控制起止时间、趋势方向、波动率和成交量范围。

# 示意代码，非实际生产代码
def generate_synthetic_bars(
    symbol: str,
    start: datetime,
    end: datetime,
    trend: str = "random",
    volatility: float = 0.02,
) -> List[BarData]:
    """生成合成 K 线数据"""
    bars = []
    current_time = start
    price = 25000

    while current_time <= end:
        if trend == "up":
            change = abs(random.gauss(0.001, volatility))
        elif trend == "down":
            change = -abs(random.gauss(0.001, volatility))
        else:
            change = random.gauss(0, volatility)

        price *= (1 + change)
        bar = BarData(
            symbol=symbol,
            timestamp=current_time,
            open_price=price * (1 + random.gauss(0, 0.001)),
            high_price=price * (1 + abs(random.gauss(0, 0.002))),
            low_price=price * (1 - abs(random.gauss(0, 0.002))),
            close_price=price,
            volume=random.randint(1000, 10000),
        )
        bars.append(bar)
        current_time += timedelta(minutes=1)

    return bars

这段代码适合构造趋势、波动率和成交量范围，但不应该替代真实缺陷快照。读者应固定随机种子或记录生成参数，否则合成数据本身会成为不稳定来源。

数据快照适合保存真实缺陷样本。脱敏后的 fixtures/hsi_2024_01.json 可以用来复现周五夜盘归属错误或半日市收盘异常。

# 示意代码，非实际生产代码
@pytest.fixture
def real_market_data():
    """使用脱敏后的真实市场数据快照"""
    return load_test_data("fixtures/hsi_2024_01.json")

快照要小而稳定，重点是保留复现缺陷的最小数据。读者应记录来源、脱敏方式、交易日归属规则和对应缺陷编号。

契约测试适合验证数据源边界。例如 get_bars("HSI", "1m", count=100) 返回的数据必须按时间排序；每根 Bar 必须满足 high >= open/close/low；时间戳必须有时区。

# 示意代码，非实际生产代码
def test_datafeed_contract():
    """验证数据源满足基本契约"""
    feed = DataFeed()

    bars = feed.get_bars("HSI", "1m", count=100)
    timestamps = [bar.timestamp for bar in bars]
    assert timestamps == sorted(timestamps)

    for bar in bars:
        assert bar.high >= bar.open
        assert bar.high >= bar.close
        assert bar.high >= bar.low

契约测试能防止上游数据变化悄悄破坏下游策略。读者应把它放在数据入口附近，而不是等策略测试失败后再排查数据问题。

图 4 的重点是血缘。测试数据从原始行情进入清洗、交易日归属、聚合、fixture、断言和回归记录，每一步都应该可追踪。否则测试失败时，读者无法判断是原始数据变化、清洗逻辑变化、交易日归属变化，还是断言本身过期。

第九部分：CI/CD 中的自动化测试

CI Gate 的职责不是跑所有测试，而是在合适的阶段阻断错误合并。不同测试层承担不同反馈速度和风险拦截职责。

一次典型 CI 可以分层执行：单元测试最快，覆盖核心算法；集成测试验证数据源、聚合器、策略和订单模块协作；属性测试用固定 seed 保证可复现；边界时间测试覆盖 HKFE 特殊交易日；覆盖率报告用于防止核心模块失去测试保护；失败测试必须阻止合并。

name: Tests

on:
  push:
    branches: [main, develop]
  pull_request:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v3
      - uses: actions/setup-python@v4
        with:
          python-version: "3.10"
      - run: pip install -r requirements.txt -r requirements-test.txt
      - run: pytest tests/unit -v --cov=core --cov-report=xml
      - run: pytest tests/integration -v
      - run: pytest tests/property -v --hypothesis-seed=0
      - uses: codecov/codecov-action@v3
        with:
          files: ./coverage.xml
          fail_ci_if_error: true

这段配置把快速反馈、模块协作、属性测试和覆盖率报告拆成独立步骤。读者应根据项目规模决定哪些测试进入每次 PR，哪些测试进入夜间任务，但失败测试必须能够阻止高风险合并。

测试报告需要能回答“哪些层级已经被验证”。下面的输出示例展示的是报告应该承载的信息，而不是要求读者复制具体数字。

tests/unit/test_aggregation.py PASSED
tests/unit/test_indicators.py PASSED
tests/unit/test_datafeed.py PASSED
tests/unit/test_strategy.py PASSED
tests/integration/test_end_to_end.py PASSED
tests/property/test_properties.py PASSED
tests/boundary/test_hkfe_cases.py PASSED
tests/regression/test_issue_*.py PASSED

core/aggregation.py        97%
core/indicators.py         94%
core/datafeed.py           91%
core/strategy.py           88%
total                      95%

这些数字本身不是目标，但它们让读者知道风险拦截在哪里发生。更重要的是：边界时间、属性测试和真实缺陷回归是否被纳入报告，而不是总覆盖率是否漂亮。

第十部分：TDD 反模式与陷阱

第一个反模式是测试实现而不是行为。测试 internal_cache.size() == 5 会把测试绑死在内部结构上；更稳妥的是断言外部行为。

# 示意代码，非实际生产代码
def test_implementation():
    """错误示例：测试内部实现细节"""
    result = calculate()
    assert result.internal_cache.size() == 5

这段代码的问题是测试依赖内部缓存结构。只要实现换成生成器、数组或增量状态，测试就会失败，即使外部行为没有变化。

# 示意代码，非实际生产代码
def test_behavior():
    """正确示例：测试外部行为"""
    result = calculate([1, 2, 3, 4, 5])
    assert result == 3.0

行为测试给重构保留空间。交易系统需要这种测试，因为性能优化、数据结构替换和执行域拆分都会改变内部实现。

第二个反模式是一个测试验证太多东西。一个测试同时调用 func1、func2、func3，失败时很难定位根因。

# 示意代码，非实际生产代码
def test_everything():
    """错误示例：一个测试覆盖所有功能"""
    result1 = func1()
    result2 = func2()
    result3 = func3()
    assert result1 == expected1
    assert result2 == expected2
    assert result3 == expected3

这段代码的问题是失败定位困难。读者无法快速判断是输入准备、某个函数行为，还是前置状态污染了后续断言。

# 示意代码，非实际生产代码
def test_func1():
    """正确示例：只测试 func1"""
    assert func1() == expected1

def test_func2():
    """正确示例：只测试 func2"""
    assert func2() == expected2

拆分后的测试让失败信号更清楚。量化系统中尤其要避免一个测试同时验证数据加载、指标计算、策略信号和订单执行。

第三个反模式是忽略边界情况。只测试 divide(10, 2) == 5 不足以证明函数可靠；还需要测试除零、小数、极小数和浮点近似。

# 示意代码，非实际生产代码
def test_normal_case():
    """错误示例：只测试正常情况"""
    assert divide(10, 2) == 5

这段测试只覆盖 happy path。交易系统里的边界更复杂，必须覆盖开盘、收盘、午休、夜盘、半日市和跨周。

# 示意代码，非实际生产代码
def test_normal_case():
    assert divide(10, 2) == 5

def test_divide_by_zero():
    """测试除零边界"""
    with pytest.raises(ZeroDivisionError):
        divide(10, 0)

def test_small_numbers():
    """测试浮点近似"""
    assert divide(1, 3) == pytest.approx(0.333, rel=1e-3)

边界测试的价值是把“不常发生但一旦发生就影响巨大”的情况固定下来。量化系统里，很多真实事故都出现在低频边界，而不是日常路径。

第四个反模式是测试数据没有血缘。测试失败时如果无法知道 fixture 来自哪里、经过哪些清洗和归属逻辑，就很难判断失败是否有效。所有真实缺陷样本都应该记录来源、脱敏方式、归属规则和断言目的。

第五个反模式是把 AI 生成测试当成验收结果。AI 可以扩展场景，但不能证明场景代表真实业务风险。验收权仍然来自规格、缺陷复现、人工 Review 和 CI Gate。

总结：测试 checklist

测试设计需要确认：

• 是否从真实缺陷和风险族群反推测试，而不是只追求覆盖率。
• 是否覆盖正常、边界、异常三类场景。
• 是否使用属性测试发现隐藏假设。
• 是否有针对量化场景的边界时间测试。

测试实现需要确认：

• 测试是否独立，不依赖执行顺序。
• 测试是否足够快，常规单测应保持秒级反馈。
• 测试是否可读，一眼能看出验证什么。
• 是否有清晰的 Arrange-Act-Assert 或 Given-When-Then 结构。

测试替身需要确认：

• 外部依赖是否使用 Stub、Mock、Spy 或 Fake 隔离。
• 复杂对象是否有简化但真实可用的 Fake。
• 交互验证是否只在确实需要时使用 Mock。

测试数据需要确认：

• 是否使用合成数据保证可重复。
• 是否保留真实缺陷的脱敏快照。
• 是否有数据契约测试保护上游边界。
• 敏感数据是否脱敏。

CI Gate 需要确认：

• 每次提交是否运行关键测试。
• 核心模块覆盖率是否高于约定阈值。
• 失败测试是否阻止合并。
• 回归缺陷是否转化为固定测试。

下一篇预告

下一篇进入 Python 性能调优实战。读者会看到 profiler 如何定位瓶颈，增量计算和虚拟化渲染如何降低延迟，以及为什么性能优化必须保留正确性证据。

参考资源

• pytest 文档：https://docs.pytest.org/
• Hypothesis 属性测试：https://hypothesis.readthedocs.io/
• mutmut 变异测试：https://mutmut.readthedocs.io/
• 《Test-Driven Development: By Example》（Kent Beck）