如何验证 Claude 生成的 vibe coding 交易机器人代码
结论
交易机器人是高风险自动化系统。Claude 或其他 AI 生成的代码应视为未经验证的初稿。即使代码能够运行、通过了回测,或看起来逻辑完整,也不能直接连接真实账户。
上线前应建立独立的验证流程:
- 将交易规则和安全边界写成可测试的规范。
- 分别测试行情、策略、风控和下单模块。
- 默认采用 fail closed。数据缺失、过期、异常或状态不明时,禁止新增风险敞口。
- 使用静态分析、单元测试、属性测试、故障注入和依赖扫描。
- 由未参与代码生成的人或另一个 AI 按照规范审查。
- 依次进行历史回放、模拟成交、券商 paper trading 和小规模实盘。
- 在生产环境中设置限额、幂等控制、订单核对、告警和人工熔断。
“无法获得最新价格时自行假定价格”不能作为普通的容错方式。这相当于在关键输入未知时继续执行可能造成资金损失的操作。系统必须按照明确的业务规则处理,不能让模型自行补全价格。
先写规则,再审代码
代码审查需要明确的判断依据。只让另一个 AI “检查有没有 bug”,通常只能发现语法、类型或常见安全问题,很难判断实际交易行为是否符合预期。
先写出可以验证的约束,例如:
- 没有有效行情时不得开仓或加仓。
- 行情超过允许时效后不得用于决策。
price <= 0、NaN、Infinity必须被拒绝。- 单笔订单金额不得超过设定上限。
- 单个标的和整个账户的持仓不得超过上限。
- 同一个交易信号只能生成一次订单。
- 下单超时后不得直接重试,必须先查询原订单状态。
- 未确认订单成交前,不得把它记为已成交。
- 价格源、账户状态或持仓状态冲突时进入安全模式。
- 所有默认值都必须在规范中列明,不允许代码临时推断。
异常场景可以整理成验收表:
| 场景 | 预期行为 |
|---|---|
| 行情接口超时 | 阻止新增订单,记录原因并告警 |
| 行情时间戳过期 | 拒绝该行情 |
| 价格为零、负数或非有限值 | 拒绝该行情 |
| 下单请求超时 | 使用客户端订单 ID 查询,不盲目重试 |
| 订单部分成交 | 按实际成交数量更新持仓 |
| 券商拒单 | 保留拒单状态,不伪装成未触发信号 |
| 本地持仓与券商持仓不同 | 停止新增风险并执行对账 |
| 达到单日亏损上限 | 禁止继续开仓并告警 |
| 程序重启 | 从券商端恢复真实订单和持仓状态 |
这些规则既要写进测试,也要落实到运行时风控中,不能只留在设计文档里。
让系统在数据异常时安全失败
下面的 Python 示例体现了一个基本原则:遇到无效行情时阻止下单,不生成替代价格。
from dataclasses import dataclass
from datetime import datetime, timedelta
from decimal import Decimal, ROUND_DOWN
from typing import Literal
class QuoteRejected(Exception):
pass
@dataclass(frozen=True)
class Quote:
symbol: str
price: Decimal
observed_at: datetime
@dataclass(frozen=True)
class Order:
symbol: str
side: Literal["buy", "sell"]
quantity: int
reference_price: Decimal
@dataclass(frozen=True)
class Decision:
order: Order | None
reason: str
def validate_quote(
quote: Quote | None,
now: datetime,
max_age: timedelta,
) -> Decimal:
if quote is None:
raise QuoteRejected("quote_missing")
if quote.observed_at.tzinfo is None or now.tzinfo is None:
raise QuoteRejected("timezone_missing")
age = now - quote.observed_at
if age < timedelta(0):
raise QuoteRejected("quote_timestamp_in_future")
if age > max_age:
raise QuoteRejected("quote_stale")
if not quote.price.is_finite():
raise QuoteRejected("price_not_finite")
if quote.price <= Decimal("0"):
raise QuoteRejected("price_not_positive")
return quote.price
def create_order(
signal: Literal["buy", "sell", "hold"],
quote: Quote | None,
now: datetime,
max_quote_age: timedelta,
max_order_notional: Decimal,
) -> Decision:
if signal == "hold":
return Decision(order=None, reason="hold_signal")
try:
price = validate_quote(quote, now, max_quote_age)
except QuoteRejected as exc:
return Decision(order=None, reason=f"blocked:{exc}")
if max_order_notional <= Decimal("0"):
return Decision(order=None, reason="blocked:invalid_notional_limit")
quantity = int(
(max_order_notional / price).to_integral_value(
rounding=ROUND_DOWN
)
)
if quantity <= 0:
return Decision(order=None, reason="blocked:quantity_is_zero")
return Decision(
order=Order(
symbol=quote.symbol,
side=signal,
quantity=quantity,
reference_price=price,
),
reason="order_created",
)
这段代码只演示输入校验,并未覆盖完整的交易风控。真实系统还要处理交易时段、最小交易单位、费用、滑点、停牌、涨跌幅限制、拆股、币种、部分成交和券商规则。
测试也要覆盖异常路径:
from datetime import datetime, timedelta, timezone
from decimal import Decimal
NOW = datetime(2026, 1, 2, 10, 0, tzinfo=timezone.utc)
def test_missing_quote_blocks_order():
decision = create_order(
signal="buy",
quote=None,
now=NOW,
max_quote_age=timedelta(seconds=5),
max_order_notional=Decimal("1000"),
)
assert decision.order is None
assert decision.reason == "blocked:quote_missing"
def test_stale_quote_blocks_order():
quote = Quote(
symbol="EXAMPLE",
price=Decimal("100"),
observed_at=NOW - timedelta(minutes=1),
)
decision = create_order(
signal="buy",
quote=quote,
now=NOW,
max_quote_age=timedelta(seconds=5),
max_order_notional=Decimal("1000"),
)
assert decision.order is None
assert decision.reason == "blocked:quote_stale"
def test_nan_price_blocks_order():
quote = Quote(
symbol="EXAMPLE",
price=Decimal("NaN"),
observed_at=NOW,
)
decision = create_order(
signal="buy",
quote=quote,
now=NOW,
max_quote_age=timedelta(seconds=5),
max_order_notional=Decimal("1000"),
)
assert decision.order is None
assert decision.reason == "blocked:price_not_finite"
def test_valid_quote_respects_notional_limit():
quote = Quote(
symbol="EXAMPLE",
price=Decimal("120"),
observed_at=NOW,
)
decision = create_order(
signal="buy",
quote=quote,
now=NOW,
max_quote_age=timedelta(seconds=5),
max_order_notional=Decimal("1000"),
)
assert decision.order is not None
assert decision.order.quantity == 8
assert decision.order.quantity * quote.price <= Decimal("1000")
分开验证策略、风控与执行
交易机器人适合拆成边界清楚的模块:
- 行情适配器负责获取数据、标记时间并验证数据。
- 策略模块根据合法输入产生交易意图。
- 风控模块判断交易意图能否执行,并限制数量。
- 执行模块向券商提交订单。
- 对账模块以券商返回的订单、成交和持仓为准。
- 监控模块记录决策原因、异常和风险指标。
策略模块应尽量写成纯函数,使同样的输入产生同样的结果,测试时也无需访问券商 API。网络超时、重试和鉴权放在适配器中处理。
策略代码不要直接调用下单接口。流程可以设计为:
行情数据
-> 数据验证
-> 策略信号
-> 风控审批
-> 订单生成
-> 券商提交
-> 状态查询
-> 成交与持仓对账
每一步都要保留结构化日志,记录输入数据的时间戳、策略版本、拒绝原因、订单 ID 和券商状态。API secret、访问令牌等敏感信息不得写入日志。
建立自动化检查门槛
Python 项目可以根据需要组合以下工具:
ruff check .
mypy src
pytest --cov=src --cov-report=term-missing
bandit -r src
semgrep --config=auto src
pip-audit
这些工具可以检查代码质量、类型、测试覆盖情况、安全模式和依赖漏洞。具体命令应根据项目目录和配置调整,同时在 CI 中固定工具版本与规则配置。
工具本身无法证明交易逻辑正确。例如,静态分析通常无法判断“缺少价格时使用 100 元”已经违反业务规则,因此还需要以下测试:
- 单元测试:验证各项策略和风控规则。
- 集成测试:验证行情、券商沙箱、数据库和消息系统之间的交互。
- 属性测试:检查任意输入下都应成立的约束,可使用 Hypothesis。
- 状态机测试:覆盖提交、部分成交、成交、撤单和拒单等状态转换。
- 故障注入:模拟超时、断网、重复响应、乱序消息和程序重启。
- 回归测试:发现缺陷后,先加入能够复现该缺陷的测试。
覆盖率只能表明哪些代码执行过,无法说明断言是否正确。测试交易系统时,应重点检查异常分支、资金边界和订单状态。
独立审查怎么做
审查可以分成两轮。
第一轮做黑盒验证。审查者只查看需求、接口和安全约束,再根据这些材料独立编写测试,以免现有实现影响判断。
第二轮做白盒审查。审查者阅读代码和变更记录,重点检查:
- 代码中是否出现未经批准的默认值。
- 异常是否被捕获后静默忽略。
except Exception是否会让程序继续下单。- 超时重试是否可能造成重复订单。
- 金额和数量计算是否使用了浮点数。
- 时间戳、时区和交易日历的处理是否一致。
- 本地订单状态是否可能领先于券商状态。
- 日志和错误信息是否泄露凭据。
- API key 是否拥有超出需要的权限。
- 普通配置或提示词是否可能意外覆盖风控参数。
如果让另一个 AI 协助审查,应同时提供需求、失败策略、接口定义、代码差异和测试,不要只给代码。提示可以写得更具体:
请把自己当作交易系统的对抗性审查者。
以下安全约束不可更改:
1. 缺失、过期或非有限价格不得用于下单。
2. 下单结果未知时不得直接重试。
3. 单笔金额和总持仓不得超过给定上限。
4. 只有券商确认的成交才能更新实际持仓。
请逐条检查代码是否违反约束,并输出:
- 文件和代码位置
- 可触发问题的输入或事件顺序
- 可能造成的资金或安全后果
- 能复现问题的测试用例
- 最小修改建议
不要假设缺失的需求,也不要把风格问题列为安全问题。
多个 AI 得出相同结论,也不能算作独立验证。它们可能有相似的盲点,也无法代替熟悉交易业务、券商接口和生产运维的人工审查。
上线前按阶段放大权限
可以按以下顺序逐步放大权限:
- 使用固定样本进行单元测试和历史数据回放。
- 在模拟撮合器中加入手续费、滑点、延迟、部分成交和拒单。
- 使用券商提供的 paper trading 或 sandbox 环境。
- 在交易时段运行 shadow mode,只记录原本会提交的订单,不实际下单。
- 设置严格的品种白名单、订单金额和持仓上限,再进行小规模实盘。
- 稳定运行并完成人工复核后,逐步调整限额。
检查回测结果时,还要考虑前视偏差、幸存者偏差、公司行为、缺失数据、时区和不现实的成交假设。paper trading 也无法完整模拟真实市场的流动性与成交质量。
生产环境必须有最后一道保险
测试无法覆盖生产环境中的所有问题。运行时至少应设置:
- 单笔订单金额上限。
- 单个标的和账户总敞口上限。
- 单位时间内的订单数量限制。
- 单日亏损或异常波动熔断。
- 可随时触发的 kill switch。
- 标的、账户和订单类型白名单。
- 唯一的
client_order_id,用于幂等和订单查询。 - 行情源与券商状态的持续健康检查。
- 本地订单、成交、余额和持仓的定期对账。
- 异常下单、连续拒单和状态不一致告警。
- API key 最小权限、独立账户和安全存储。
处理下单请求超时时要格外谨慎。超时只说明客户端没有及时收到结果,券商仍可能已经接到订单。直接重试可能导致重复成交。应先使用 client_order_id 查询原订单,再决定如何处理。
行情失效时,停止开仓未必能保护已有持仓。如何处理现有风险需要单独设计,例如使用独立行情源、券商托管的保护订单或转为人工接管。没有可信的价格和账户状态时,不要盲目平仓。
备注:内容仅供参考。
