量化交易系统做到实时行情驱动自动下单这一步绕不开一个很现实的问题行情源和交易通道到底怎么接。很多人把精力全砸在策略回测上等到要落地的时候才发现数据延迟、接口限流、订单状态不同步这些工程问题比策略本身难缠得多。这篇内容围绕一个具体的接入场景展开——把实时行情和自动化交易接到某交易平台的通道上属于整个量化系统开发系列里偏工程落地的一环。适合已经写完策略逻辑、准备把信号变成真实委托的开发者也适合刚接触自动化交易、想搞清楚从行情到成交这条链路到底长什么样的人。下面不讲虚的直接从链路拆解讲到代码骨架再讲我踩过的那些坑。1. 先把实时行情到自动成交这条链路拆开看1.1 一条委托从信号到成交要经过哪些环节很多人对自动化交易的理解停留在策略出信号程序就下单实际中间隔着一整条流水线。我习惯把它拆成五段行情采集、信号计算、风控校验、委托发送、状态回执。这五段任何一段出问题最终表现都是策略看着对实盘就是亏但根因完全不同。行情采集段负责把平台推送的tick或者K线落到本地核心指标是延迟和完整性。信号计算段是策略逻辑通常跑在独立线程或者独立进程里避免被行情回调阻塞。风控校验段是最容易被新手忽略的它要在委托发出前检查仓位上限、单笔金额、当日亏损、频率限制这些硬约束。委托发送段负责和平台接口打交道处理签名、重试、幂等。状态回执段则是把平台的成交回报、拒单原因同步回本地账本。这五段之间是异步的不是顺序调用。行情来了触发信号信号进队列风控消费队列风控通过后发委托委托结果再回写。用队列解耦的好处是某一段慢了不会拖垮整条链路坏处是状态一致性要自己保证。我见过不少人图省事写成同步调用行情一密集就卡死回调线程被堵住后面的tick全丢。提示行情回调和下单动作绝对不能在同一个线程里做。平台推送回调通常有超时限制你在回调里做网络请求轻则丢包重则被平台断开连接。1.2 为什么实时性不是越快越好刚做自动化交易的人容易陷入一个误区延迟越低越好恨不得微秒级。实际在个人开发者能接触到的通道上追求极致低延迟意义不大因为瓶颈往往不在你的代码而在网络往返和平台撮合队列。真正要关注的是延迟的稳定性也就是抖动。我做过一组对比同一套策略平均延迟从80毫秒降到40毫秒收益几乎没变化但延迟抖动从正负200毫秒收敛到正负30毫秒回撤明显改善。原因很直白策略里的止损、追单逻辑依赖时间窗口抖动大意味着同样的逻辑在不同时刻执行效果差异巨大回测里根本模拟不出来。所以工程上的取舍是宁可平均延迟高一点也要把抖动压下去。具体做法包括固定心跳间隔、行情本地缓存加时间戳校验、下单前不做重计算、网络层用长连接而不是每次新建。这些细节后面章节会展开。1.3 平台通道在整个系统里的定位把交易平台通道理解成一个适配器更准确。你的策略核心不应该直接依赖某个平台的API而应该依赖一层抽象接口平台通道只是这层接口的一个实现。这样做的直接好处是换通道时策略代码不用动间接好处是测试时可以用模拟实现替换真实通道。我在项目里通常定义三个抽象行情源接口订阅、退订、回调注册、交易接口下单、撤单、查持仓、查资金、回报接口成交回报、订单状态变更。平台通道实现这三个接口策略层只认接口。这个设计在只有单一通道时看着多余但一旦要接第二个通道或者做通道切换省下的重构成本非常可观。2. 行情接入订阅、解析与本地缓存怎么设计2.1 订阅粒度与限流规避平台行情接口一般按标的订阅也有按板块或者全市场推送的。订阅粒度选错要么流量爆炸要么漏数据。我的经验是按策略实际持仓加自选池订阅不要一上来就全市场。全市场推送在行情活跃时段每秒几千条个人机器的解析和落盘根本扛不住最后积压导致延迟飙升。限流是另一个硬约束。多数平台对订阅数量、请求频率都有上限超了会返回错误码甚至临时封禁。规避办法有两个一是订阅请求做批量合并把多个标的的订阅打包成一次请求二是维护一个订阅状态表避免重复订阅同一标的。重复订阅看着无害实际会成倍消耗配额。# 订阅状态管理的最小实现思路 class SubscriptionManager: def __init__(self, max_symbols200): self.subscribed set() self.max_symbols max_symbols def subscribe(self, symbols): new_symbols [s for s in symbols if s not in self.subscribed] if len(self.subscribed) len(new_symbols) self.max_symbols: raise ValueError(订阅数量超限) if not new_symbols: return [] # 批量提交避免逐条请求触发限流 result self._batch_request(new_symbols) self.subscribed.update(new_symbols) return result这段代码的重点不在逻辑多复杂而在先算差集再批量提交这个习惯。逐条订阅在标的少的时候看不出问题标的到几十个就会开始偶发限流错误。2.2 tick解析中的字段陷阱行情推送的原始字段和你想用的字段往往不是一回事。最常见的坑是时间戳平台给的时间可能是交易所时间、可能是平台服务器时间、还可能是本地接收时间三者含义完全不同。策略里做时间窗口判断必须统一到同一个时间基准否则回测和实盘对不上。另一个坑是价格字段的精度和单位。有的平台价格是放大整数比如价格乘以10000存成整数有的是浮点。浮点直接比较相等会出问题做价格撮合判断时要用容差。成交量字段有的给累计值有的给增量搞反了会导致量能指标完全失真。我一般会在解析层做一次标准化把原始字段映射成内部统一结构时间统一转成本地毫秒时间戳并保留原始时间做对照价格统一成定点小数。这层转换看着繁琐但能避免后面策略代码里到处做单位换算减少出错面。2.3 本地缓存与断线补数实时推送一定会断这是常态不是异常。网络抖动、平台维护、心跳超时都会导致断连。断连期间错过的行情必须能补回来否则策略基于残缺数据做判断后果比不交易更糟。补数方案取决于平台能力。有的平台提供历史tick查询接口断线重连后按时间区间拉取补齐有的只提供K线那就用K线做粗粒度对齐。我的做法是本地维护一个环形缓冲区存最近N条tick同时记录最后一条的时间戳重连后先拉取该时间戳之后的增量再切回实时推送。缓存本身要注意内存。全市场tick缓存几小时就能吃掉几个G个人机器扛不住。按标的做LRU淘汰只保留策略关注的窗口长度比如最近5000条。落盘用追加写而不是全量重写避免IO尖峰。3. 交易通道对接签名、幂等与订单状态机3.1 接口签名的常见实现与易错点交易接口几乎都要求签名防止请求被篡改和重放。签名算法通常是参数排序拼接加密钥做哈希。易错点集中在三处参数排序规则字典序还是自定义顺序、空值参数是否参与签名、时间戳的时区和格式。我遇到过最隐蔽的一个问题是参数编码。某个参数值里带了特殊字符拼接前没做URL编码签名算出来和服务端不一致报的是签名错误排查了半天才发现是编码问题。所以签名前统一做一次规范化编码是必须的步骤。import hashlib import hmac import time from urllib.parse import quote def build_signature(params, secret): # 过滤空值按字典序排序 filtered {k: v for k, v in params.items() if v is not None and v ! } sorted_items sorted(filtered.items()) # 统一编码后再拼接避免特殊字符导致签名不一致 query .join(f{k}{quote(str(v), safe)} for k, v in sorted_items) timestamp str(int(time.time() * 1000)) raw query timestamp timestamp sign hmac.new(secret.encode(), raw.encode(), hashlib.sha256).hexdigest() return sign, timestamp签名还有个实践要点时间戳和服务端时间偏差不能太大多数平台允许正负5分钟。本地时钟不准会导致签名过期建议启动时做一次时间校准运行中定期检查偏差。3.2 幂等设计重复下单是怎么发生的自动化交易最怕的不是下错单是同一笔单下重复了。重复下单的来源有好几个网络超时后重试、程序崩溃重启后重放、回调重复触发。每一种都要单独防。网络超时重试的幂等靠客户端订单号。每笔委托生成一个全局唯一ID重试时带同一个ID服务端据此去重。这里的关键是ID生成要保证唯一且可追溯我一般用时间戳策略标识自增序列组合避免纯随机导致排查困难。程序重启重放要靠持久化。委托发出前先把订单意图写进本地数据库状态标记为待确认收到回执后更新状态。重启时扫描待确认的订单主动去平台查询真实状态而不是盲目重发。这一步很多人省了结果就是重启后重复下单。回调重复触发则要在消费端做去重用订单ID做幂等键处理过的ID记进一个短期集合重复的直接丢弃。3.3 订单状态机别用布尔值描述订单新手常犯的错是用一个是否成交的布尔值描述订单实际订单状态至少有已提交、已接受、部分成交、全部成交、已撤单、已拒绝、未知。用布尔值会丢失大量信息导致撤单逻辑、对账逻辑全乱。我习惯画一个明确的状态机每个状态允许的迁移路径固定死。比如已拒绝是终态不能再变部分成交可以迁到全部成交或已撤单未知状态要触发主动查询。状态机的好处是任何非法迁移都能立刻发现而不是等到对账时才发现账对不上。当前状态允许迁移到触发条件已提交已接受、已拒绝、未知平台回执已接受部分成交、全部成交、已撤单、已拒绝成交回报或撤单回执部分成交全部成交、已撤单后续成交或撤单未知任意确定状态主动查询结果这张表建议直接写进代码做校验任何不在表里的迁移都记日志告警。实盘里未知状态的处理尤其重要宁可多查几次也不能假设它成功了。4. 风控前置把亏损挡在委托发出之前4.1 为什么风控必须在客户端做一遍平台侧通常也有风控但那是最后一道防线触发时往往已经造成了实际影响比如被限制交易。客户端风控的价值在于提前拦截把明显不合理的委托挡在发出之前减少无效请求和被平台处罚的概率。客户端风控要检查的项包括单笔委托金额上限、单标的持仓上限、总仓位上限、当日累计亏损上限、单位时间委托频率、价格偏离度防止乌龙指。这些阈值不是拍脑袋定的要结合策略的历史最大回撤和资金规模反推。比如单笔金额上限我一般设成总资金的2%到5%具体看策略的持仓集中度。4.2 频率限制与熔断机制频率限制防的是策略逻辑出bug导致疯狂下单。实现上用滑动窗口计数比如任意10秒内委托不超过20笔超了直接拒绝并告警。滑动窗口比固定窗口好固定窗口在边界处会有双倍突刺。熔断是更高一级的保护。当连续出现N笔拒单或者当日亏损达到阈值直接暂停交易一段时间人工确认后再恢复。熔断状态要持久化程序重启后不能自动解除否则等于没做。我见过一个案例策略因为参数错误连续下错方向没有熔断半小时亏掉大半仓位这种教训一次就够。class RiskGuard: def __init__(self, max_orders_per_window20, window_sec10): self.order_times [] self.window_sec window_sec self.max_orders max_orders_per_window self.circuit_broken False def check_frequency(self, now): cutoff now - self.window_sec self.order_times [t for t in self.order_times if t cutoff] if len(self.order_times) self.max_orders: return False self.order_times.append(now) return True def check_and_trip(self, consecutive_rejects, daily_loss_ratio): if consecutive_rejects 5 or daily_loss_ratio 0.05: self.circuit_broken True return not self.circuit_broken4.3 价格偏离度校验的实际参数价格偏离度校验是防乌龙指的关键。逻辑很简单委托价格和当前最新价的偏离超过阈值就拒绝。阈值设多少有讲究设太小正常波动会被误拦设太大起不到保护作用。我的经验是按标的的波动率动态设。用近20个周期的平均真实波幅做基准偏离超过3到5倍ATR就拒绝。这样高波动标的阈值自动放宽低波动标的阈值收紧比固定百分比合理。另外开盘集合竞价时段和收盘前要单独处理这两个时段价格波动特性不同固定阈值容易误伤。5. 实盘跑起来之后才会暴露的那些问题5.1 回测和实盘的偏差到底出在哪策略回测漂亮实盘拉胯是自动化交易最普遍的痛点。偏差来源我归成四类成交假设、滑点、延迟、数据质量。回测里默认信号出现就能按收盘价成交实盘里信号出现到委托到达平台有时间差价格早变了。回测里滑点设个固定值实盘里滑点随流动性变化。缩小偏差的办法不是把回测调得和实盘一样而是把实盘的约束前移到回测里。具体做法回测时用下一根K线的开盘价成交而不是当前收盘价滑点按成交量的函数动态计算延迟按实测的P95值模拟。这样回测结果会难看一些但更接近真实。5.2 日志与对账出问题时靠什么定位实盘出问题第一反应是查日志。日志设计不好排查就是大海捞针。我的日志分三层行情层记录每条tick的接收时间和关键字段交易层记录每笔委托的完整生命周期风控层记录每次拦截的原因。三层用同一个订单ID或者时间戳关联方便串起来看。对账是每天收盘后的固定动作。把本地账本的持仓、资金和平台返回的做比对差异超过阈值就告警。对账能发现很多隐蔽问题比如某笔成交回报丢了导致本地持仓和实际不符。对账逻辑要能处理平台有本地无和本地有平台无两个方向不能只查一边。5.3 通道切换与降级预案单一通道总有不可用的时候维护、限流、故障都可能。成熟的系统要有降级预案主通道不可用时切备用通道或者降级为只平仓不开仓。降级判断要自动化靠心跳和错误率触发不能等人工发现。切换通道时最麻烦的是持仓同步。不同通道的持仓数据格式不同切换前要把本地账本和原通道对平再和新通道建立映射。这个过程要停交易所以切换窗口要选在非交易时段或者行情清淡时。我一般把切换做成半自动系统检测到异常自动暂停并告警人工确认后执行切换脚本避免自动切换在异常状态下越切越乱。6. 一套可复用的通道适配层代码骨架6.1 抽象接口定义前面提过策略层只认接口这里给出接口定义的具体样子。行情源、交易、回报三个接口分开各自职责单一。接口方法尽量少够用就行方法多了实现负担重还容易设计过度。from abc import ABC, abstractmethod class MarketDataFeed(ABC): abstractmethod def subscribe(self, symbols): ... abstractmethod def unsubscribe(self, symbols): ... abstractmethod def on_tick(self, callback): ... class TradeGateway(ABC): abstractmethod def place_order(self, order): ... abstractmethod def cancel_order(self, order_id): ... abstractmethod def query_position(self): ... abstractmethod def query_balance(self): ... class TradeReporter(ABC): abstractmethod def on_order_update(self, callback): ... abstractmethod def on_trade_update(self, callback): ...接口定义好之后平台通道就是这三个接口的实现类。测试时写一个模拟实现可以注入延迟、丢包、拒单用来验证策略和风控的健壮性。这个模拟实现在开发阶段的价值极高比直接连真实通道调试安全得多。6.2 事件总线与线程模型各模块之间用事件总线通信而不是直接互相调用。行情模块发tick事件策略模块订阅tick发信号事件风控模块订阅信号发委托事件交易模块订阅委托执行。事件总线用有界队列实现队列满了要有明确的丢弃策略或者背压策略不能无限堆积。线程模型上行情回调线程只做解析和入队不做任何耗时操作。策略计算跑在独立线程池风控和交易各一个线程。这样任何一段慢都不会阻塞行情接收。线程间共享状态用不可变对象或者加锁保护我倾向于传不可变消息减少锁的使用。6.3 配置化与多环境隔离通道参数、风控阈值、订阅列表这些都要配置化不能硬编码。配置分环境开发、模拟、实盘各一套用环境变量或者配置文件切换。实盘配置要格外小心密钥不能明文存代码库用环境变量或者独立的密钥管理。配置加载后要做校验比如阈值是否在合理范围、必填项是否缺失。校验失败直接启动失败不要带着错误配置跑起来。我见过因为配置文件里一个阈值写成了0导致风控形同虚设跑了一整天才发现。7. 我在这个环节踩过的几个真实坑第一个坑是心跳设计。早期我用固定间隔发心跳结果行情清淡时段平台主动断开连接因为心跳间隔比平台要求的超时时间长。后来改成动态心跳行情活跃时靠数据包自然保活清淡时缩短心跳间隔问题解决。这个细节文档里往往一笔带过实际很影响稳定性。第二个坑是浮点比较。订单价格和成交价格用浮点比较是否相等偶尔出现明明成交了却判断为未成交。根因是浮点精度改成定点小数或者带容差比较后消失。涉及金额和价格的地方能用整数就不用浮点。第三个坑是重连风暴。网络短暂抖动导致大量连接同时断开又同时重连把平台限流打满。解决办法是重连加随机退避退避时间随重试次数指数增长并加随机抖动避免所有连接同一时刻重连。第四个坑是时区。服务器用UTC平台回调用本地时间策略里做时间窗口判断时没统一导致某些时段逻辑完全错乱。后来所有时间在入口处统一转成UTC毫秒时间戳展示层再转本地问题根除。这些坑的共同点是文档里不会写只有真正跑起来才会遇到。所以我的建议是通道对接完成后先跑一段时间的模拟盘把各种异常场景人为制造一遍比如断网、重启、平台拒单观察系统行为是否符合预期再上实盘。模拟盘阶段暴露问题的成本远低于实盘。最后分享一个习惯每次上线新版本先在模拟环境跑满一个完整交易日对比新旧版本的委托数量、成交率、拒单率这些指标差异在合理范围内才切实盘。这个流程看着慢但能挡住绝大多数低级错误。自动化交易这行稳比快重要得多。
量化交易系统做到实时行情驱动自动下单这一步绕不开一个很现实的问题行情源和交易通道到底怎么接。很多人把精力全砸在策略回测上等到要落地的时候才发现数据延迟、接口限流、订单状态不同步这些工程问题比策略本身难缠得多。这篇内容围绕一个具体的接入场景展开——把实时行情和自动化交易接到某交易平台的通道上属于整个量化系统开发系列里偏工程落地的一环。适合已经写完策略逻辑、准备把信号变成真实委托的开发者也适合刚接触自动化交易、想搞清楚从行情到成交这条链路到底长什么样的人。下面不讲虚的直接从链路拆解讲到代码骨架再讲我踩过的那些坑。1. 先把实时行情到自动成交这条链路拆开看1.1 一条委托从信号到成交要经过哪些环节很多人对自动化交易的理解停留在策略出信号程序就下单实际中间隔着一整条流水线。我习惯把它拆成五段行情采集、信号计算、风控校验、委托发送、状态回执。这五段任何一段出问题最终表现都是策略看着对实盘就是亏但根因完全不同。行情采集段负责把平台推送的tick或者K线落到本地核心指标是延迟和完整性。信号计算段是策略逻辑通常跑在独立线程或者独立进程里避免被行情回调阻塞。风控校验段是最容易被新手忽略的它要在委托发出前检查仓位上限、单笔金额、当日亏损、频率限制这些硬约束。委托发送段负责和平台接口打交道处理签名、重试、幂等。状态回执段则是把平台的成交回报、拒单原因同步回本地账本。这五段之间是异步的不是顺序调用。行情来了触发信号信号进队列风控消费队列风控通过后发委托委托结果再回写。用队列解耦的好处是某一段慢了不会拖垮整条链路坏处是状态一致性要自己保证。我见过不少人图省事写成同步调用行情一密集就卡死回调线程被堵住后面的tick全丢。提示行情回调和下单动作绝对不能在同一个线程里做。平台推送回调通常有超时限制你在回调里做网络请求轻则丢包重则被平台断开连接。1.2 为什么实时性不是越快越好刚做自动化交易的人容易陷入一个误区延迟越低越好恨不得微秒级。实际在个人开发者能接触到的通道上追求极致低延迟意义不大因为瓶颈往往不在你的代码而在网络往返和平台撮合队列。真正要关注的是延迟的稳定性也就是抖动。我做过一组对比同一套策略平均延迟从80毫秒降到40毫秒收益几乎没变化但延迟抖动从正负200毫秒收敛到正负30毫秒回撤明显改善。原因很直白策略里的止损、追单逻辑依赖时间窗口抖动大意味着同样的逻辑在不同时刻执行效果差异巨大回测里根本模拟不出来。所以工程上的取舍是宁可平均延迟高一点也要把抖动压下去。具体做法包括固定心跳间隔、行情本地缓存加时间戳校验、下单前不做重计算、网络层用长连接而不是每次新建。这些细节后面章节会展开。1.3 平台通道在整个系统里的定位把交易平台通道理解成一个适配器更准确。你的策略核心不应该直接依赖某个平台的API而应该依赖一层抽象接口平台通道只是这层接口的一个实现。这样做的直接好处是换通道时策略代码不用动间接好处是测试时可以用模拟实现替换真实通道。我在项目里通常定义三个抽象行情源接口订阅、退订、回调注册、交易接口下单、撤单、查持仓、查资金、回报接口成交回报、订单状态变更。平台通道实现这三个接口策略层只认接口。这个设计在只有单一通道时看着多余但一旦要接第二个通道或者做通道切换省下的重构成本非常可观。2. 行情接入订阅、解析与本地缓存怎么设计2.1 订阅粒度与限流规避平台行情接口一般按标的订阅也有按板块或者全市场推送的。订阅粒度选错要么流量爆炸要么漏数据。我的经验是按策略实际持仓加自选池订阅不要一上来就全市场。全市场推送在行情活跃时段每秒几千条个人机器的解析和落盘根本扛不住最后积压导致延迟飙升。限流是另一个硬约束。多数平台对订阅数量、请求频率都有上限超了会返回错误码甚至临时封禁。规避办法有两个一是订阅请求做批量合并把多个标的的订阅打包成一次请求二是维护一个订阅状态表避免重复订阅同一标的。重复订阅看着无害实际会成倍消耗配额。# 订阅状态管理的最小实现思路 class SubscriptionManager: def __init__(self, max_symbols200): self.subscribed set() self.max_symbols max_symbols def subscribe(self, symbols): new_symbols [s for s in symbols if s not in self.subscribed] if len(self.subscribed) len(new_symbols) self.max_symbols: raise ValueError(订阅数量超限) if not new_symbols: return [] # 批量提交避免逐条请求触发限流 result self._batch_request(new_symbols) self.subscribed.update(new_symbols) return result这段代码的重点不在逻辑多复杂而在先算差集再批量提交这个习惯。逐条订阅在标的少的时候看不出问题标的到几十个就会开始偶发限流错误。2.2 tick解析中的字段陷阱行情推送的原始字段和你想用的字段往往不是一回事。最常见的坑是时间戳平台给的时间可能是交易所时间、可能是平台服务器时间、还可能是本地接收时间三者含义完全不同。策略里做时间窗口判断必须统一到同一个时间基准否则回测和实盘对不上。另一个坑是价格字段的精度和单位。有的平台价格是放大整数比如价格乘以10000存成整数有的是浮点。浮点直接比较相等会出问题做价格撮合判断时要用容差。成交量字段有的给累计值有的给增量搞反了会导致量能指标完全失真。我一般会在解析层做一次标准化把原始字段映射成内部统一结构时间统一转成本地毫秒时间戳并保留原始时间做对照价格统一成定点小数。这层转换看着繁琐但能避免后面策略代码里到处做单位换算减少出错面。2.3 本地缓存与断线补数实时推送一定会断这是常态不是异常。网络抖动、平台维护、心跳超时都会导致断连。断连期间错过的行情必须能补回来否则策略基于残缺数据做判断后果比不交易更糟。补数方案取决于平台能力。有的平台提供历史tick查询接口断线重连后按时间区间拉取补齐有的只提供K线那就用K线做粗粒度对齐。我的做法是本地维护一个环形缓冲区存最近N条tick同时记录最后一条的时间戳重连后先拉取该时间戳之后的增量再切回实时推送。缓存本身要注意内存。全市场tick缓存几小时就能吃掉几个G个人机器扛不住。按标的做LRU淘汰只保留策略关注的窗口长度比如最近5000条。落盘用追加写而不是全量重写避免IO尖峰。3. 交易通道对接签名、幂等与订单状态机3.1 接口签名的常见实现与易错点交易接口几乎都要求签名防止请求被篡改和重放。签名算法通常是参数排序拼接加密钥做哈希。易错点集中在三处参数排序规则字典序还是自定义顺序、空值参数是否参与签名、时间戳的时区和格式。我遇到过最隐蔽的一个问题是参数编码。某个参数值里带了特殊字符拼接前没做URL编码签名算出来和服务端不一致报的是签名错误排查了半天才发现是编码问题。所以签名前统一做一次规范化编码是必须的步骤。import hashlib import hmac import time from urllib.parse import quote def build_signature(params, secret): # 过滤空值按字典序排序 filtered {k: v for k, v in params.items() if v is not None and v ! } sorted_items sorted(filtered.items()) # 统一编码后再拼接避免特殊字符导致签名不一致 query .join(f{k}{quote(str(v), safe)} for k, v in sorted_items) timestamp str(int(time.time() * 1000)) raw query timestamp timestamp sign hmac.new(secret.encode(), raw.encode(), hashlib.sha256).hexdigest() return sign, timestamp签名还有个实践要点时间戳和服务端时间偏差不能太大多数平台允许正负5分钟。本地时钟不准会导致签名过期建议启动时做一次时间校准运行中定期检查偏差。3.2 幂等设计重复下单是怎么发生的自动化交易最怕的不是下错单是同一笔单下重复了。重复下单的来源有好几个网络超时后重试、程序崩溃重启后重放、回调重复触发。每一种都要单独防。网络超时重试的幂等靠客户端订单号。每笔委托生成一个全局唯一ID重试时带同一个ID服务端据此去重。这里的关键是ID生成要保证唯一且可追溯我一般用时间戳策略标识自增序列组合避免纯随机导致排查困难。程序重启重放要靠持久化。委托发出前先把订单意图写进本地数据库状态标记为待确认收到回执后更新状态。重启时扫描待确认的订单主动去平台查询真实状态而不是盲目重发。这一步很多人省了结果就是重启后重复下单。回调重复触发则要在消费端做去重用订单ID做幂等键处理过的ID记进一个短期集合重复的直接丢弃。3.3 订单状态机别用布尔值描述订单新手常犯的错是用一个是否成交的布尔值描述订单实际订单状态至少有已提交、已接受、部分成交、全部成交、已撤单、已拒绝、未知。用布尔值会丢失大量信息导致撤单逻辑、对账逻辑全乱。我习惯画一个明确的状态机每个状态允许的迁移路径固定死。比如已拒绝是终态不能再变部分成交可以迁到全部成交或已撤单未知状态要触发主动查询。状态机的好处是任何非法迁移都能立刻发现而不是等到对账时才发现账对不上。当前状态允许迁移到触发条件已提交已接受、已拒绝、未知平台回执已接受部分成交、全部成交、已撤单、已拒绝成交回报或撤单回执部分成交全部成交、已撤单后续成交或撤单未知任意确定状态主动查询结果这张表建议直接写进代码做校验任何不在表里的迁移都记日志告警。实盘里未知状态的处理尤其重要宁可多查几次也不能假设它成功了。4. 风控前置把亏损挡在委托发出之前4.1 为什么风控必须在客户端做一遍平台侧通常也有风控但那是最后一道防线触发时往往已经造成了实际影响比如被限制交易。客户端风控的价值在于提前拦截把明显不合理的委托挡在发出之前减少无效请求和被平台处罚的概率。客户端风控要检查的项包括单笔委托金额上限、单标的持仓上限、总仓位上限、当日累计亏损上限、单位时间委托频率、价格偏离度防止乌龙指。这些阈值不是拍脑袋定的要结合策略的历史最大回撤和资金规模反推。比如单笔金额上限我一般设成总资金的2%到5%具体看策略的持仓集中度。4.2 频率限制与熔断机制频率限制防的是策略逻辑出bug导致疯狂下单。实现上用滑动窗口计数比如任意10秒内委托不超过20笔超了直接拒绝并告警。滑动窗口比固定窗口好固定窗口在边界处会有双倍突刺。熔断是更高一级的保护。当连续出现N笔拒单或者当日亏损达到阈值直接暂停交易一段时间人工确认后再恢复。熔断状态要持久化程序重启后不能自动解除否则等于没做。我见过一个案例策略因为参数错误连续下错方向没有熔断半小时亏掉大半仓位这种教训一次就够。class RiskGuard: def __init__(self, max_orders_per_window20, window_sec10): self.order_times [] self.window_sec window_sec self.max_orders max_orders_per_window self.circuit_broken False def check_frequency(self, now): cutoff now - self.window_sec self.order_times [t for t in self.order_times if t cutoff] if len(self.order_times) self.max_orders: return False self.order_times.append(now) return True def check_and_trip(self, consecutive_rejects, daily_loss_ratio): if consecutive_rejects 5 or daily_loss_ratio 0.05: self.circuit_broken True return not self.circuit_broken4.3 价格偏离度校验的实际参数价格偏离度校验是防乌龙指的关键。逻辑很简单委托价格和当前最新价的偏离超过阈值就拒绝。阈值设多少有讲究设太小正常波动会被误拦设太大起不到保护作用。我的经验是按标的的波动率动态设。用近20个周期的平均真实波幅做基准偏离超过3到5倍ATR就拒绝。这样高波动标的阈值自动放宽低波动标的阈值收紧比固定百分比合理。另外开盘集合竞价时段和收盘前要单独处理这两个时段价格波动特性不同固定阈值容易误伤。5. 实盘跑起来之后才会暴露的那些问题5.1 回测和实盘的偏差到底出在哪策略回测漂亮实盘拉胯是自动化交易最普遍的痛点。偏差来源我归成四类成交假设、滑点、延迟、数据质量。回测里默认信号出现就能按收盘价成交实盘里信号出现到委托到达平台有时间差价格早变了。回测里滑点设个固定值实盘里滑点随流动性变化。缩小偏差的办法不是把回测调得和实盘一样而是把实盘的约束前移到回测里。具体做法回测时用下一根K线的开盘价成交而不是当前收盘价滑点按成交量的函数动态计算延迟按实测的P95值模拟。这样回测结果会难看一些但更接近真实。5.2 日志与对账出问题时靠什么定位实盘出问题第一反应是查日志。日志设计不好排查就是大海捞针。我的日志分三层行情层记录每条tick的接收时间和关键字段交易层记录每笔委托的完整生命周期风控层记录每次拦截的原因。三层用同一个订单ID或者时间戳关联方便串起来看。对账是每天收盘后的固定动作。把本地账本的持仓、资金和平台返回的做比对差异超过阈值就告警。对账能发现很多隐蔽问题比如某笔成交回报丢了导致本地持仓和实际不符。对账逻辑要能处理平台有本地无和本地有平台无两个方向不能只查一边。5.3 通道切换与降级预案单一通道总有不可用的时候维护、限流、故障都可能。成熟的系统要有降级预案主通道不可用时切备用通道或者降级为只平仓不开仓。降级判断要自动化靠心跳和错误率触发不能等人工发现。切换通道时最麻烦的是持仓同步。不同通道的持仓数据格式不同切换前要把本地账本和原通道对平再和新通道建立映射。这个过程要停交易所以切换窗口要选在非交易时段或者行情清淡时。我一般把切换做成半自动系统检测到异常自动暂停并告警人工确认后执行切换脚本避免自动切换在异常状态下越切越乱。6. 一套可复用的通道适配层代码骨架6.1 抽象接口定义前面提过策略层只认接口这里给出接口定义的具体样子。行情源、交易、回报三个接口分开各自职责单一。接口方法尽量少够用就行方法多了实现负担重还容易设计过度。from abc import ABC, abstractmethod class MarketDataFeed(ABC): abstractmethod def subscribe(self, symbols): ... abstractmethod def unsubscribe(self, symbols): ... abstractmethod def on_tick(self, callback): ... class TradeGateway(ABC): abstractmethod def place_order(self, order): ... abstractmethod def cancel_order(self, order_id): ... abstractmethod def query_position(self): ... abstractmethod def query_balance(self): ... class TradeReporter(ABC): abstractmethod def on_order_update(self, callback): ... abstractmethod def on_trade_update(self, callback): ...接口定义好之后平台通道就是这三个接口的实现类。测试时写一个模拟实现可以注入延迟、丢包、拒单用来验证策略和风控的健壮性。这个模拟实现在开发阶段的价值极高比直接连真实通道调试安全得多。6.2 事件总线与线程模型各模块之间用事件总线通信而不是直接互相调用。行情模块发tick事件策略模块订阅tick发信号事件风控模块订阅信号发委托事件交易模块订阅委托执行。事件总线用有界队列实现队列满了要有明确的丢弃策略或者背压策略不能无限堆积。线程模型上行情回调线程只做解析和入队不做任何耗时操作。策略计算跑在独立线程池风控和交易各一个线程。这样任何一段慢都不会阻塞行情接收。线程间共享状态用不可变对象或者加锁保护我倾向于传不可变消息减少锁的使用。6.3 配置化与多环境隔离通道参数、风控阈值、订阅列表这些都要配置化不能硬编码。配置分环境开发、模拟、实盘各一套用环境变量或者配置文件切换。实盘配置要格外小心密钥不能明文存代码库用环境变量或者独立的密钥管理。配置加载后要做校验比如阈值是否在合理范围、必填项是否缺失。校验失败直接启动失败不要带着错误配置跑起来。我见过因为配置文件里一个阈值写成了0导致风控形同虚设跑了一整天才发现。7. 我在这个环节踩过的几个真实坑第一个坑是心跳设计。早期我用固定间隔发心跳结果行情清淡时段平台主动断开连接因为心跳间隔比平台要求的超时时间长。后来改成动态心跳行情活跃时靠数据包自然保活清淡时缩短心跳间隔问题解决。这个细节文档里往往一笔带过实际很影响稳定性。第二个坑是浮点比较。订单价格和成交价格用浮点比较是否相等偶尔出现明明成交了却判断为未成交。根因是浮点精度改成定点小数或者带容差比较后消失。涉及金额和价格的地方能用整数就不用浮点。第三个坑是重连风暴。网络短暂抖动导致大量连接同时断开又同时重连把平台限流打满。解决办法是重连加随机退避退避时间随重试次数指数增长并加随机抖动避免所有连接同一时刻重连。第四个坑是时区。服务器用UTC平台回调用本地时间策略里做时间窗口判断时没统一导致某些时段逻辑完全错乱。后来所有时间在入口处统一转成UTC毫秒时间戳展示层再转本地问题根除。这些坑的共同点是文档里不会写只有真正跑起来才会遇到。所以我的建议是通道对接完成后先跑一段时间的模拟盘把各种异常场景人为制造一遍比如断网、重启、平台拒单观察系统行为是否符合预期再上实盘。模拟盘阶段暴露问题的成本远低于实盘。最后分享一个习惯每次上线新版本先在模拟环境跑满一个完整交易日对比新旧版本的委托数量、成交率、拒单率这些指标差异在合理范围内才切实盘。这个流程看着慢但能挡住绝大多数低级错误。自动化交易这行稳比快重要得多。
