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模拟,重点测逻辑
- 配置类代码:测几个关键路径就行
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}")
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分钟再全量
- 回滚能力:每次部署都要能一键回滚
14.5 知识体系总览
下面这张图,是我对本章内容的总结。你可以把它当作一个检查清单,看看自己的系统在哪些环节还需要加强。
这张图展示了测试与部署的完整链路。从左到右,从单元测试到混沌工程,再到CI/CD流水线,每一环都不可或缺。
最后说一句:测试不是为了证明系统没问题,而是为了发现我们没想到的问题。做市商系统每天处理真金白银,多花点时间在测试上,比出事后补救要划算得多。
无相订单流研究社 微信Lucian808555