14. 系统测试与部署:从单元测试到混沌工程

做量化交易系统,我最怕什么?不是策略亏钱,而是系统上线后突然崩了。你想想看,一个做市商系统,每天处理几百万笔订单,一旦出问题,损失可不是闹着玩的。所以这一章,我把自己这些年踩过的坑、总结的经验,全部分享给你。

14.1 单元测试:用pytest守住第一道防线

单元测试,说白了就是给每个函数、每个模块做体检。我刚开始做交易系统时,总觉得写测试浪费时间。直到有一次,一个简单的价格计算函数出了bug,导致整个报价引擎报错……嗯,那次教训让我记住了:没有测试的代码,就是定时炸弹。

14.1.1 为什么选pytest?

Python的测试框架不少,但我个人习惯用pytest。原因很简单:

  • 简洁:不需要写类,直接写函数就行
  • 强大:fixture、参数化、插件生态丰富
  • :测试发现和执行速度都很快

来看一个实际的例子。假设我们有个计算订单簿中间价的函数:

# order_book.py
def get_mid_price(bids: list, asks: list) -> float:
    """计算订单簿中间价"""
    if not bids or not asks:
        raise ValueError("订单簿不能为空")
    
    best_bid = max(bid[0] for bid in bids)
    best_ask = min(ask[0] for ask in asks)
    
    return (best_bid + best_ask) / 2.0

对应的测试代码:

# test_order_book.py
import pytest
from order_book import get_mid_price

def test_normal_case():
    """正常情况下的测试"""
    bids = [(100.5, 10), (100.0, 20)]
    asks = [(101.0, 15), (101.5, 5)]
    
    result = get_mid_price(bids, asks)
    assert result == 100.75

def test_empty_order_book():
    """空订单簿应该抛出异常"""
    with pytest.raises(ValueError):
        get_mid_price([], [(101.0, 15)])

@pytest.mark.parametrize("bids, asks, expected", [
    ([(100.0, 1)], [(101.0, 1)], 100.5),
    ([(99.5, 10)], [(100.5, 10)], 100.0),
    ([(100.0, 5), (99.0, 5)], [(101.0, 5), (102.0, 5)], 100.5),
])
def test_multiple_cases(bids, asks, expected):
    """参数化测试多个场景"""
    assert get_mid_price(bids, asks) == expected
我的小技巧:用 @pytest.mark.parametrize 可以一次性测试多种情况,省时省力。我曾经用这个装饰器,一个测试函数覆盖了50多个边界条件。

14.1.2 测试覆盖率不是越高越好

很多人追求100%的测试覆盖率。但说实话,在交易系统里,有些代码很难测,比如和交易所的WebSocket连接。我的建议是:

  • 核心业务逻辑:覆盖率要达到90%以上
  • IO相关代码:用mock模拟,重点测逻辑
  • 配置类代码:测几个关键路径就行
注意:不要为了覆盖率而写测试。我曾经见过有人把getter/setter都测一遍,纯粹是浪费生命。

14.2 集成测试:让组件协同工作

单元测试通过,不代表系统能跑起来。我记得有一次,订单管理模块和风控模块各自测试都通过了,但一联调就出问题——两个模块对“订单状态”的定义不一致。这就是集成测试要解决的问题。

14.2.1 测试策略

做市商系统的集成测试,我一般分三层:

层级 测试内容 工具
模块间 API接口、数据格式 pytest + requests
子系统 报价-风控-下单链路 docker-compose
全系统 模拟真实交易场景 自定义测试框架

来看一个模块间测试的例子:

# test_integration.py
def test_quote_to_order_flow():
    """测试从报价生成到订单提交的完整流程"""
    # 1. 模拟行情数据
    market_data = generate_mock_market_data()
    
    # 2. 调用报价引擎
    quotes = quote_engine.generate_quotes(market_data)
    assert len(quotes) > 0
    
    # 3. 风控检查
    for quote in quotes:
        risk_check = risk_controller.check(quote)
        assert risk_check.passed == True
    
    # 4. 提交订单
    orders = order_manager.submit_orders(quotes)
    assert all(order.status == "SUBMITTED" for order in orders)

核心原则:集成测试要模拟真实环境,但不要依赖真实交易所。用mock交易所或者沙箱环境。

14.3 混沌工程:主动找茬的艺术

混沌工程,说白了就是故意搞破坏。我刚开始做这个的时候,同事都觉得我疯了——好好的系统为什么要故意弄挂它?但你想啊,与其等生产环境出问题,不如我们先把它搞崩溃,看看系统能不能扛得住。

14.3.1 做市商系统的混沌实验

针对交易系统,我设计过这些实验:

  • 网络延迟注入:模拟交易所网络抖动
  • 节点宕机:随机杀掉一个服务实例
  • 数据延迟:行情数据延迟到达
  • 订单积压:模拟瞬间大量订单涌入

用Python实现一个简单的混沌实验:

# chaos_test.py
import random
import time
import threading

class NetworkChaos:
    def __init__(self, target_service):
        self.target = target_service
        self.running = False
    
    def inject_latency(self, min_ms=100, max_ms=5000):
        """注入随机网络延迟"""
        def _inject():
            while self.running:
                delay = random.randint(min_ms, max_ms)
                time.sleep(delay / 1000.0)
                # 模拟网络延迟
                self.target.add_latency(delay)
        
        thread = threading.Thread(target=_inject)
        thread.start()
    
    def kill_instance(self):
        """随机杀掉一个实例"""
        instance = random.choice(self.target.instances)
        instance.stop()
        print(f"Killed instance: {instance.id}")
经验之谈:混沌实验要在预发布环境先跑,别一上来就搞生产。我曾经在预发布环境发现,系统对网络延迟超过3秒就完全崩溃了——这个发现帮我们避免了一次生产事故。

14.4 CI/CD流水线:自动化一切

手动部署?那是石器时代的做法。在交易系统里,每一秒的延迟都意味着真金白银。所以,CI/CD流水线不是可选项,是必需品。

14.4.1 流水线设计

我设计的CI/CD流水线长这样:

# .gitlab-ci.yml 示例
stages:
  - test
  - build
  - deploy

unit_test:
  stage: test
  script:
    - pytest tests/unit/ --cov=src --cov-report=term
  only:
    - merge_requests

integration_test:
  stage: test
  script:
    - docker-compose -f docker-compose.test.yml up -d
    - pytest tests/integration/
    - docker-compose -f docker-compose.test.yml down
  only:
    - main

chaos_test:
  stage: test
  script:
    - python chaos_test.py --duration 300
  when: manual
  only:
    - main

deploy_staging:
  stage: deploy
  script:
    - ansible-playbook deploy.yml -i staging
  only:
    - main

deploy_production:
  stage: deploy
  script:
    - ansible-playbook deploy.yml -i production
  when: manual
  only:
    - tags

14.4.2 关键设计原则

  • 快速反馈:单元测试要在5分钟内跑完
  • 环境一致性:开发、测试、生产环境用同样的Docker镜像
  • 灰度发布:先部署到10%的节点,观察5分钟再全量
  • 回滚能力:每次部署都要能一键回滚
血的教训:我曾经在生产环境直接全量部署,结果有个配置项写错了,导致所有报价都变成了0。从那以后,我坚持灰度发布,再急也要分批部署。

14.5 知识体系总览

下面这张图,是我对本章内容的总结。你可以把它当作一个检查清单,看看自己的系统在哪些环节还需要加强。

系统测试与部署知识体系 单元测试 pytest框架 参数化测试 Mock模拟 覆盖率管理 集成测试 模块间接口 子系统联调 全链路测试 沙箱环境 混沌工程 网络延迟注入 节点宕机 数据延迟 订单积压 CI/CD 流水线 代码提交 自动测试 构建镜像 灰度部署 上线 从代码提交到生产部署,全流程自动化

这张图展示了测试与部署的完整链路。从左到右,从单元测试到混沌工程,再到CI/CD流水线,每一环都不可或缺。

最后说一句:测试不是为了证明系统没问题,而是为了发现我们没想到的问题。做市商系统每天处理真金白银,多花点时间在测试上,比出事后补救要划算得多。


交易系统化学习资料 微信Strategy888888