做实时行情接入这事儿我最早是在一个凌晨被电话叫醒的。客户说策略跑得好好的突然不更新了我一查日志行情连接断了已经四个多小时代码里硬编码的重试逻辑根本扛不住推送型数据的断线重连。那会儿我意识到行情接入不是“调一个API拿到数据”那么简单从数据API到策略层中间那一大段路才是一个量化工程师真正的日常。这篇文章主要讲我在实战里怎么设计实时行情和五档行情接入方案的。内容适合几类人看一是刚接触量化、想把聚宽/米筐/Tushare这类数据源接到自己策略里的Python开发者二是已经在跑策略但受够了“数据断线、字段对不上、回测实盘不一致”这些问题的人三是想了解L2五档行情到底比普通快照多出什么价值的从业者。我尽量把从数据源选型、工程结构设计、代码实现到排障经验一次讲透不绕弯全是我实际踩过坑后的整理。需要说明的是文中的代码和方案是基于我多年实盘项目的常规实践总结出来的行业里靠谱的做法基本大同小异你可以直接照着搭也可以按自己环境调整。1. 实时行情与五档行情的本质差异1.1 行情数据到底分几层很多人一开始接触的数据接口比如Tushare、AkShare这类拿到的其实是日线数据和分钟线数据。这类数据的特点是非实时、历史归档、低频。它们适合做回测、做盘后分析但如果你要做日内短线、做盘口博弈、做高频择时日线和分钟线根本不够用因为价格在日内是怎么走出来的订单簿是怎么变化的这些信息都已经丢掉了。再往上一个层级就是实时快照。它通常是每3秒推送一次沪深两市的普通行情快照频率包含最新价、涨跌幅、成交量、成交额、买卖五档有时只有一档等字段。这是绝大多数个人量化策略用得最多的数据。它解决了“实时性”的问题但你看到的盘口是一个时间点的切片看不到两次快照之间发生了什么。再往上就是五档行情和逐笔成交。这里需要区分两个概念普通的Level-1行情也提供五档报价但它是3秒一次的快照而真正的Level-2五档行情是实时推送的每当盘口任何一个档位的挂单量或挂单价发生变化就会推一条数据过来频率可以达到毫秒级。L2还附带逐笔委托、逐笔成交这些更细的数据。这里说的五档指的是买一至买五、卖一至卖五共10个价位每个价位上有对应的委托量和委托笔数。把这三个层级放在一起看你会发现越往上数据量越大、延迟越低、信息粒度越细同时对工程能力的要求也越高。很多策略设计之初没有想清楚自己到底需要哪一层数据结果要么用高频数据跑低频逻辑白白浪费性能要么用低频数据硬撑高频策略结果频繁失效。所以动手前先量化一下自己的需求策略最短持仓周期是多少如果我们做的是分钟级以上的交易3秒快照完全够用如果是秒级或毫秒级博弈那L2逐笔是硬门槛。这个判断直接决定了后面所有设计。1.2 五档行情的独特价值与适用场景我做策略编程这十几年经常被问到一个问题普通快照也有五档数据为什么要单独强调“五档行情接进策略”这里面的差距主要体现在三个维度上。第一是频次。3秒一次的快照五档和实时变化的五档信息量差了几个数量级。拿一只成交活跃的股票来说3秒内盘口可能已经变了十几次。你看到的快照五档很可能已经是“过去式”了如果你依据它判断买卖压力天然就滞后。L2五档能看到盘口动态变化的连续性比如买一挂单一瞬间被吃掉又立刻补上这种信息只在逐笔级数据里才有。第二是挂单撤单行为。L2里最值钱的信息之一就是挂单和撤单的明细。主力资金经常用“挂大单吸引跟风、成交前撤单”的手法这在3秒快照里几乎不可能被发现——等方式你看到那笔大单它已经消失了。而L2数据能记录委托进入订单簿的完整轨迹你可以统计挂撤比、大单活跃度等特征。但这有个前提你得有能力分析这些高频数据流否则拿到L2也只是多了几行数字。第三是盘口重构能力。有了L2的逐笔委托和成交数据你可以精确重建交易时段内任意一个时刻的订单簿状态。这意味着你可以做非常精细的微观结构研究比如计算买卖压力不平衡度、识别大单方向、评估冲击成本等。单纯靠快照是不具备这个能力的。所以什么场景适合用五档我的经验是高频做市、高频统计套利、日内T0策略以及需要精确计算冲击成本和订单簿状态的算法交易这些场景离不开L2。而对于日频选股、分钟级趋势跟踪策略用L2反而是性能浪费——你要处理的微结构特征在分钟线里已经被抹平了。这篇文章后面要讲的工程设计其实同时覆盖了快照接入和L2接入因为底层的数据分发框架是相通的。2. 数据API到策略层的整体架构设计2.1 从API到策略中间到底发生了什么很多人第一次写实时行情策略代码是这样的import time from some_api import get_quote while True: quote get_quote(600519) if quote[last_price] my_ma20: buy() time.sleep(3)这段代码如果拿来做演示没问题但拿到实盘环境里跑几周内一定会出问题。问题不在逻辑而在工程。API返回的数据格式可能变更网络随时可能断开行情推送的节奏和自己策略的计算节奏不匹配策略里某个函数抛异常导致整个循环退出等等。这就是为什么要强调“从数据API到策略层的工程设计”。一个能稳定运行的行情接入系统中间至少要经过这么几个环节连接管理层维护和行情服务器之间的连接状态处理鉴权、心跳、断线重连。数据接收层接收原始推送或轮询返回的数据做初步的格式解析。数据标准化层把不同数据源的数据统一成内部格式解决字段命名不一致、单位不一致、时间时区不一致的问题。缓存与快照层维护当前最新的行情快照供策略按需查询同时维护必要的近期历史数据比如最近N根K线、最近N笔成交。事件分发层把新到的行情数据包装成事件按订阅关系分发到不同的策略模块。策略层策略逻辑本身接收到行情事件后更新自身状态、产生信号、下单。如果少了其中任何一层短期看不出问题长期必然崩。我有一次接手别人的策略代码发现他的策略模块里直接塞了API连接代码策略重启就得重建连接盘中连接断了还会把策略状态搞乱。这个架构说白了是在用单机进程写一个微型“消息总线”我后来把所有行情接入统一到一个中间层策略只订阅事件不再关心数据从哪来。2.2 拉取模式与推送模式两种接入模型对比行情接入方案从数据获取方式上看本质上只有两种模式Pull拉取和Push推送。搞清楚它们的区别很多架构选型问题就迎刃而解。拉取模式你的程序主动向数据服务器发起请求拿到某一时刻的数据。比如你每隔3秒调一次RESTful API获取最新快照。这种方式实现简单、调试方便不需要维护长连接但它有两个天然问题一是延迟取决于你的轮询间隔永远比实时慢二是有请求频率限制不可能做到毫秒级。它适合数据频率要求不高、策略逻辑简单、或者只做盘后分析的场景。推送模式你建立一条长连接通常是WebSocket或自定义TCP协议服务器有新的行情数据就主动推给你。这种方式延迟低、实时性好数据一到就能触发策略逻辑是实盘行情接入的主流方式。缺点是需要处理连接生命周期、心跳保活、断线重连、消息顺序、积压处理等一堆问题。L2五档行情几乎都是推送模式因为数据量太大拉取根本跟不上。我的建议是如果只是做日线和分钟级策略拉取模式够用只要涉及秒级以下的实时决策就必须上推送模式。这两个模式对应的工程复杂度是数量级差异所以在项目启动前就要定下来。有些免费数据接口同时提供REST和WebSocket两种方式的我一般都建议实盘用WebSocketREST可以用来做启动时的补数据。2.3 分层设计的核心思想让策略层不感知数据来源我个人在做这套系统时最核心的一条设计原则就是策略层不应该关心数据是从哪里来的也不应该关心数据是REST轮询拿到的还是WebSocket推送过来的。为了实现这一点我定义了一个统一的数据接入接口class MarketDataProvider(ABC): 行情数据提供者抽象接口 abstractmethod def subscribe(self, symbols: list[str], callback: Callable[[dict], None]) - None: 订阅标的数据注册回调函数 ... abstractmethod def get_snapshot(self, symbol: str) - dict | None: 获取指定标的最新快照 ... abstractmethod def close(self) - None: 关闭连接、释放资源 ...任何实现这个接口的类不管底层是聚宽、米筐、还是自己搭的行情网关对策略来说都是一样的。策略只需要在初始化时拿到一个MarketDataProvider实例然后调用subscribe注册自己的回调函数剩下的细节全部被隔离在实现类里。这个抽象的收益在换数据源的时候体现得最明显。我在一个项目里客户一开始用的免费数据源测试后来数据质量达不到要求换成了商业L2数据源策略代码一行没改只换了provider的实现类。如果你把API调用直接写进策略里光是替换数据源就得重构一遍整个策略模块。3. 数据源选型与API接入细节3.1 常见行情API类型与选型建议在写代码之前先解决数据源的问题。市面上可选的行情数据源按数据质量和费用可以分为这么几类类型代表方案实时性五档L2费用适合场景免费数据接口Tushare Pro、AkShare日线/分钟线为主不提供免费学习、回测、低频策略券商/第三方量化平台聚宽JoinQuant、米筐RiceQuant实时快照部分提供按策略收费平台内策略开发商业数据API恒生、Wind、迅投QMT等实时/逐笔完整提供费用较高专业实盘、高频自建行情网关对接交易所行情源或期货公司柜台实时/逐笔完整提供硬件通道成本专业机构、高频团队对个人开发者和中小团队我最常用的组合是学习阶段用Tushare Pro或AkShare做历史回测实盘阶段如果做A股直接用券商提供的QMT或Ptrade接口做加密货币就接交易所的WebSocket接口。没有必要一上来就花大价钱买商业行情源。选型时我通常关注几个指标数据延迟快照延迟多少毫秒/秒、字段完整性有没有五档、逐笔、买卖队列、历史数据深度能不能拿到分钟级历史、接口稳定性会不会经常断流、限流、费用结构按量计费还是包年。免费接口虽然省成本但经常有请求频率限制而且节假日、盘中和盘后的数据一致性偶尔会出问题这些都要在接入前评估好。这里顺便说一句很多人搜“股票数据接口api 免费”然后找到一些别人打包好的接口直接用。我的建议是能一线对接就一线对接尽量不要用不明来历的第三方代理接口。一是不安全你的代码和交易数据可能被人截获二是不稳定代理接口随时可能停止服务。免费接口中Tushare Pro和AkShare在社区里口碑相对靠谱可以作为起步选择。3.2 一个完整的WebSocket行情接入示例下面给出一个我实际用过的、基于WebSocket的快照行情接入代码示例。这个示例用通用结构演示不绑定某一家厂商方便你迁移到自己的数据源上import json import threading import time import websocket class WebSocketQuoteProvider: 基于WebSocket的实时行情提供者推送模式 def __init__(self, ws_url, api_key, symbols, callback): self.ws_url ws_url self.api_key api_key self.symbols symbols self.callback callback # 收到行情后回调策略层 self.ws None self.connected False self._running False self._heartbeat_interval 15 # 心跳间隔秒数 def _on_message(self, ws, message): 收到服务端推送的消息 try: data json.loads(message) # 根据你的数据源协议判断是行情快照还是心跳响应 if data.get(type) quote: quote self._normalize_quote(data[data]) self.callback(quote) except Exception as e: # 消息解析失败不能影响主流程 print(f消息解析异常: {e}) def _normalize_quote(self, raw): 标准化行情字段统一内部命名 return { symbol: raw[sec_code], time: raw[timestamp], price: raw[last_px], volume: raw[business_amount], bid_prices: [raw[fbid_px_{i}] for i in range(1, 6)], bid_volumes: [raw[fbid_volume_{i}] for i in range(1, 6)], ask_prices: [raw[fask_px_{i}] for i in range(1, 6)], ask_volumes: [raw[fask_volume_{i}] for i in range(1, 6)], } def _on_open(self, ws): print(行情连接已建立) self.connected True # 连接建立后发送订阅请求 subscribe_msg { op: subscribe, symbols: self.symbols, api_key: self.api_key } ws.send(json.dumps(subscribe_msg)) def _on_error(self, ws, error): print(f连接错误: {error}) self.connected False def _on_close(self, ws, code, msg): print(连接关闭) self.connected False def _heartbeat(self): 心跳保活线程 while self._running: if self.connected: try: self.ws.send(json.dumps({op: ping})) except Exception as e: print(f心跳发送失败: {e}) time.sleep(self._heartbeat_interval) def connect(self): 建立长连接并启动心跳线程 self.ws websocket.WebSocketApp( self.ws_url, on_messageself._on_message, on_errorself._on_error, on_closeself._on_close, on_openself._on_open ) self._running True # WebSocketApp的run_forever是阻塞的放到独立线程运行 ws_thread threading.Thread(targetself.ws.run_forever, daemonTrue) ws_thread.start() heartbeat_thread threading.Thread(targetself._heartbeat, daemonTrue) heartbeat_thread.start() def close(self): self._running False if self.ws: self.ws.close()这个代码有个关键点叫_normalize_quote。不同数据源返回的字段名千奇百怪比如有的叫last_px有的叫latest_price有的叫newPrice。如果不想办法把字段统一你的策略逻辑就会被数据源的字段命名绑架换个数据源就得改策略。所以我把它单独拆成一个方法所有原始字段在这个方法里被翻译成内部统一的字典格式。后面所有模块都只用这个标准化后的格式这就是工程上的“防腐层”。我在代码里还开了心跳线程。推送型连接最怕的不是数据报错而是连接已经断了但你的程序不知道。很多行情服务端会定期检查客户端心跳超时没收到就断开连接。客户端这边也要定时发送心跳保活同时监控最后一次收到数据的时间如果超过阈值没收到任何数据主动判定连接异常并触发重连。这个机制在长连接类应用里是必须的不是可有可无的优化。3.3 五档行情的订阅与数据处理五档行情和普通快照行情的接入流程大体一致差别主要在数据模型和订阅协议上。沪深L2行情的推送频率非常高所以代码实现上有几个地方需要特别处理。先看数据模型。一个典型的五档行情推送事件除了最新价、成交量这些基础字段外还包含买一至买五、卖一至卖五的价格和数量。有些数据源还会带每档的委托笔数这通常只在L2里提供。我的标准五档数据字典长这样l2_quote { symbol: 600519, time: 1699999999.123, # 事件时间毫秒时间戳 last_price: 1750.0, last_volume: 100, # 最后一笔成交量 accum_volume: 1234567, # 累计成交量 accum_amount: 987654321.0, # 累计成交额 bids: [ {price: 1749.99, volume: 1000, orders: 12}, # 买一 {price: 1749.98, volume: 800, orders: 8}, # 买二 # ... 买三、买四、买五 ], asks: [ {price: 1750.01, volume: 600, orders: 5}, # 卖一 # ... ] }这个结构里bids和asks是列表按价格从优到劣排列。这个排列顺序是有讲究的买盘第一个是最高的买入价卖盘第一个是最低的卖出价。你后续计算买卖价差、订单簿不平衡度时直接取第0个元素就是最优档位省去排序逻辑性能更好。处理L2数据时有一个容易踩的坑五档数组长度可能不满5个。比如股票涨停时卖一可能直接没有挂单刚开盘集合竞价结束时盘口可能只有一档两档。所以解析代码里必须做长度判断不能假设每次都能读到完整的五个档位def calc_spread(quote: dict) - float | None: 计算买卖价差盘口不完整时返回None bids quote.get(bids) or [] asks quote.get(asks) or [] if not bids or not asks: return None best_bid bids[0][price] best_ask asks[0][price] if best_bid 0 or best_ask 0: return None return best_ask - best_bid另一个需要注意的点是涨跌停状态。当股票涨停时整个卖盘可能为空盘口严重倾斜跌停时则买盘为空。如果不做状态判断直接拿空盘口去算特征会算出荒谬的结果导致策略产生假信号。标准做法是维护一个trading_status字段明确标记标的当前处于“正常/涨停/跌停/停牌”中的哪种状态策略层拿到异常状态时直接跳过计算或走风控逻辑。有了前面这些基础做五档行情接入策略的工程框架就逐步清晰了下面我按第二部分的分层思路把每一层具体怎么写在代码里展开来说。4. 行情数据与策略层之间的数据流转设计4.1 从回调函数到策略模块事件驱动的核心写法推送型行情的天然编程模型是事件驱动。数据源一有消息就回调我们注册的函数我们的策略逻辑就运行在这个回调函数里。听起来简单细节却很容易失控。最典型的问题如果策略处理逻辑比较复杂比如要做复杂的特征计算、调用外部模型推理那么回调函数里的一次处理就可能耗时几百毫秒甚至几秒。而在这段时间里行情服务端又推过来了几十条新数据你的程序根本处理不过来数据越积越多延迟越来越大最终策略看到的行情越来越“旧”策略表现自然越来越差。解决这个问题的标准方案是回调函数只做“收数据”和“缓存”这两件最轻的事真正的策略计算放到另一个线程或进程里处理。看代码import queue import threading import time class QuoteEngine: 行情引擎接收行情推送异步送给策略处理器 def __init__(self, provider, strategy_handler, buffer_size10000): self.provider provider self.strategy_handler strategy_handler self.buffer queue.Queue(maxsizebuffer_size) self._running False def _on_quote(self, quote: dict): 行情回调只做入队操作快速返回 try: self.buffer.put_nowait(quote) except queue.Full: # 队列已满说明消费速度跟不上丢弃最旧数据或记录告警 # 这里选择丢弃并报警防止内存无限增长 print(行情队列已满丢弃最新行情) def _worker(self): 消费线程从队列取数据交给策略处理器 while self._running: try: quote self.buffer.get(timeout1) self.strategy_handler.on_quote(quote) except queue.Empty: continue except Exception as e: print(f策略处理异常: {e}) def start(self): self._running True thread threading.Thread(targetself._worker, daemonTrue) thread.start() self.provider.subscribe(self._on_quote) def stop(self): self._running False self.provider.close()这个QuoteEngine类做的事情很简单provider拿到行情推送到_on_quote_on_quote只把数据放进队里后台一个worker线程不停地从队里取数据交给策略处理器。这样做的好处是生产者和消费者的速度被解耦了哪怕策略某一次计算超时也不会阻塞行情数据的接收。队列长度要设置上限。我在代码里设了maxsize10000超过就丢弃并打印告警。这是个权衡如果无限积压系统内存一会儿就爆了如果无脑丢弃策略可能漏掉关键行情。实际项目中我会监控队列积压长度如果经常出现队列满的情况说明策略处理能力远低于行情速度需要优化策略逻辑或者分批处理而不是单纯把队列调大。有人可能想问为什么不直接在回调函数里处理省得队列中转我刚开始也是这么干的直到有一次策略里调用了一个第三方指标库碰到极端行情时计算特别慢行情连接直接因为处理超时被服务端断开了。从那以后我再也不在回调里做重活了。回调函数要像接线员一样接电话、记下来、挂电话别在电话里跟人聊天。4.2 快照缓存让策略能随时拿到最新状态事件驱动模式有个特点策略只有在收到新的行情事件时才会被唤醒。但很多时候你的策略需要主动查询某个标的的最新状态。比如策略主循环运行时发现某个持仓股需要重新评估它希望拿到“当前”的盘口状态而不是被动等待下一条行情。要支持这种主动查询就需要一个快照缓存层。它保存所有订阅标的最新的一份标准化行情快照随时提供按标的查询的能力。这个缓存的实现很简单核心就是一个字典专门用个锁保护并发访问import threading class SnapshotStore: 最新行情快照缓存支持并发读写 def __init__(self): self._store {} self._lock threading.RLock() def update(self, quote: dict): 收到新行情时更新缓存 symbol quote[symbol] with self._lock: self._store[symbol] quote def get(self, symbol: str) - dict | None: 获取指定标的最新快照没有则返回None with self._lock: return self._store.get(symbol) def get_all(self) - list[dict]: 获取全部标的最新快照用于策略遍历 with self._lock: return list(self._store.values())别小看这个简单的类。有了它你的策略就可以随时“看一眼”市场而不是只能“等通知”。我在实际项目里还会在快照里额外存一个local_timestamp字段记录本地收到这条数据的时间。为什么因为行情服务端的时间戳是交易所撮合时间但数据到达你的程序可能已经有网络延迟。排查问题时比如要判断是不是网络链路有延迟这两个时间戳的差就是关键证据。这个是实战里很有用的细节一般教程不会提。更进一步快照缓存还可以保存近期数据的滚动窗口。比如虽然你的策略事件周期是逐笔级但你想计算过去5分钟的成交量变化那你需要保存最近5分钟的成交数据。这个滚动窗口用Python的collections.deque实现很合适它支持从两端快速 append 和 pop。我通常会为这个窗口设定最大长度比如最多存10万条防止内存膨胀。4.3 Tick数据到Bar数据的聚合几行代码实现K线构建很多策略并不需要逐笔处理行情而是基于1分钟、5分钟K线做判断。这种情况下你不需要把每条tick都送进策略而是先把tick聚合成bar再把bar事件发给策略。这里的聚合逻辑就是所谓的tick-to-bar。最普通的K线聚合是基于时间窗口的比如每个自然分钟内的所有tick聚合成一根1分钟K线。核心逻辑是记录这一分钟的开盘价、最高价、最低价、收盘价和总成交量分钟结束时触发一次bar事件。看实现import threading import time class BarAggregator: 将tick行情聚合为K线bar def __init__(self, timeframe_seconds: int 60, on_barNone): self.timeframe_seconds timeframe_seconds self.on_bar on_bar # 回调函数K线生成后调用 self._bars {} # symbol - 当前积累中的bar self._lock threading.RLock() def update(self, quote: dict): 每来一条tick更新对应的bar symbol quote[symbol] price quote[price] volume quote.get(last_volume, 0) ts quote[time] bar_time (ts // self.timeframe_seconds) * self.timeframe_seconds with self._lock: bar self._bars.get(symbol) if bar is None or bar[time] ! bar_time: # 新的一根K线开始先触发上一根完成事件 if bar is not None and self.on_bar: self.on_bar(bar) # 初始化新K线 bar { symbol: symbol, time: bar_time, open: price, high: price, low: price, close: price, volume: 0 } self._bars[symbol] bar # 更新K线数据 bar[high] max(bar[high], price) bar[low] min(bar[low], price) bar[close] price bar[volume] volume这个聚合逻辑有个细节判断“新K线开始”用的是当前tick的时间戳除以时间窗口然后取整。也就是整分钟的边界一旦跨过就会自动开启新K线。这避免了用实时钟判断时的各种时区、延迟导致的错位问题。我这里用的是“时间驱动”的简单聚合依赖行情自带的时间戳。如果数据源的tick时间戳本身就是乱的那聚合结果也会乱这时候就需要额外的tick排序逻辑这也是L2数据源考验工程能力的一个点。行情服务器的多路推送之间的先后顺序偶尔会错乱必须先按时间和序号排序再做聚合否则K线的最高最低价都会被算错。5. 策略层对接实战从收到行情到产生交易信号5.1 策略状态机的设计一切决策基于最新快照策略层是整个系统的大脑。行情数据经过前面的一系列加工到这里已经是标准化的、带完整盘口信息的快照或bar了。策略要做的是根据这些数据维护自身状态并在合适的时机产生交易信号或风控指令。策略的自身状态通常包括当前持仓、当前可用资金、每个标的的信号状态、是否已有未完成的订单等。这些状态和行情数据不一样行情数据是外部实时变化的而状态是策略内部维护的、跟随交易行为更新的。我的经验是把这两种数据彻底分开管理行情状态放快照缓存策略状态放策略对象内部。如果不分开很容易出现“策略状态被行情回调改乱了”的问题。一个简单的策略状态机示例class SimpleIntradayStrategy: 一个简单的日内策略示例突破策略 def __init__(self, snapshot_store, broker, break_threshold0.02): self.snapshot_store snapshot_store self.broker broker self.break_threshold break_threshold self.position {} # symbol - 持仓数量 self.day_high {} # symbol - 当日最高价 self.day_low {} # symbol - 当日最低价 def on_bar(self, bar: dict): K线回调更新日内高低点判断突破信号 symbol bar[symbol] # 更新当日高低点 if symbol not in self.day_high or bar[high] self.day_high[symbol]: self.day_high[symbol] bar[high] if symbol not in self.day_low or bar[low] self.day_low[symbol]: self.day_low[symbol] bar[low] # 突破逻辑当前价突破昨日高点一定比例则买入 # 这里省略昨日高点的获取逻辑假设存在self.prev_high字典里 prev_high self.prev_high.get(symbol) if prev_high is None: return if bar[close] prev_high * (1 self.break_threshold): if self.position.get(symbol, 0) 0: self.broker.buy(symbol, volume100) print(f{symbol} 触发突破买入价格 {bar[close]})这段代码体现了一个重要的设计选择策略从bar事件开始计算但决策依据是它自己维护的状态而不是被动的“看到什么就是什么”。比如突破策略要求当日最高点它不能直接拿当前快照的high字段而要自己把每个bar的high累加更新。这样做的好处是你可以随时回放历史数据重建策略状态而且不受行情乱序影响。策略层还有一个常被忽略的点幂等性设计。实时行情推送偶尔会发生重复或者因为断线重连导致同一根K线被重复推送。如果策略的每次信号计算都不是幂等的比如买了信号来一次就买一次那么数据重复就会导致重复开仓。我处理这类问题的手段是在策略里维护已处理的bar时间戳集合或者用“当前signal状态成交回报”双重确认机制确保一个信号只触发一次交易。5.2 策略与回测逻辑一致性别再让实盘和回测“两套逻辑”聊策略层就不能不提一个我见过无数次的坑回测和实盘用的是两套完全不同的行情处理逻辑。回测时你从历史数据里一根根K线遍历过去认为“上一根K线收盘时我就知道这个数据了”实盘时你在实时收到K线的瞬间才拿到数据还要处理盘中突变、断线重连造成的缺口。两套逻辑如果处理方式不一致回测表现再好的策略实盘也大概率拉胯。要解决这个问题核心是让“数据喂给策略”的方式在回测和实盘中保持一致。具体做法是回测时也走同一套事件驱动框架把历史数据“重放”给策略模块。也就是说平时实盘用的BarAggregator、QuoteEngine、Strategy这些模块在回测中原封不动地复用只是把数据来源从“WebSocket推送”换成“文件批量读取”。这里展示一个简单的历史数据回放引擎骨架class BacktestReplayer: 历史数据回放引擎用同一套事件框架回测 def __init__(self, quotes, strategy): self.quotes sorted(quotes, keylambda q: q[time]) self.strategy strategy self.aggregator BarAggregator( timeframe_seconds60, on_barstrategy.on_bar ) def run(self): for quote in self.quotes: # 和实盘一样先更新快照缓存再喂给聚合器 # 这里为了演示简化直接喂给聚合器 self.aggregator.update(quote)理解这段代码的关键是实盘和回测只是数据来源不同数据处理和策略执行的代码路径完全一致。这就是所谓“一次编写两处复用”。如果没有这样一个统一框架你迟早会在回测里发现一个策略表现优异实盘却不知所措然后花整整一周去排查到底哪里不一致。我用这套思路后回测和实盘之间的一致性显著提高再也没出现过因为“数据延迟假设不同”导致的严重失真。5.3 风控模块订单前的最后一道防线策略产生了信号不代表可以直接执行。在信号和交易执行之间还要放一道风控模块。它的职责是检查这个信号是否符合当前风控规则比如单笔下单量是否超过限制、当日累计交易次数是否超限、当前持仓比例是否超过上限、标的是否在禁买名单里等。风控模块的输入是“交易信号”和“当前策略/账户状态”输出是“允许/拒绝”或“允许但调整数量”。这样一来具体的风控规则就可以灵活配置策略层不需要关心这些细节。看一个极简风控实现class RiskManager: 风控模块所有交易信号经过这里过滤 def __init__(self, max_position_ratio0.2, max_trade_per_day10): self.max_position_ratio max_position_ratio self.max_trade_per_day max_trade_per_day self.trade_count {} def check_order(self, symbol: str, volume: float, price: float, account_total: float, current_position: float) - bool: 检查一笔订单是否通过风控 # 规则1单只标的持仓市值不能超过总资产20% order_value volume * price if (current_position order_value) account_total * self.max_position_ratio: return False # 规则2当日交易次数限制 if self.trade_count.get(symbol, 0) self.max_trade_per_day: return False return True这个风控模块虽然简单但位置很重要。我习惯把它放在策略信号产生之后、真正下单之前。信号产生后先过风控风控通过再交给broker执行。这个顺序千万别颠倒否则一旦风控逻辑有bug策略就可能在非预期条件下下单风险很大。风控规则的判定还要考虑极端行情比如涨跌停附近流动性枯竭时即使信号明确也可能无法按期望价格成交。这不是风控模块能解决的需要交易执行模块配合处理比如设置滑点保护、超时撤单等。后面第6节会讲到。6. 工程化落地的关键细节与避坑指南6.1 日志、监控与告警跑起来只是起点行情接入系统能跑起来只是万里长征第一步。真正决定它在生产环境稳定性的是日志、监控与告警这三件套。日志方面我的核心原则是“关键时刻必有日志高频日志必须采样”。连接断开、重连成功、订阅失败、队列满、风控拒绝这些关键事件必须完整记录而且要带上时间戳和上下文信息。但像“收到一条行情”这种每秒几十次的事件不能每条都打日志否则日志文件几小时就能撑爆磁盘。一般做法是抽样打日志比如每1000条打一条或者仅在数据异常时去打。监控方面最实用的几个指标是连接状态是否在线、最近一次行情时间距现在的秒数衡量行情新鲜度、队列积压长度、每秒处理行情数、策略信号产生数。这些指标可以用简单的计数器维护定时打印或上报到监控系统。我用过PrometheusGrafana这样的专业方案也用过只用日志文件加定时任务扫描的极简方案。对个人项目和中小团队来说后者往往已经够用。告警方面最核心的告警规则就三条连接断开超过阈值比如30秒、行情数据超时未更新比如15秒没有任何新行情、策略异常连续出现比如连续5次报错。告警方式可以用邮件、企业微信机器人、或者直接打日志然后由外部监控系统触发选项很多。关键是告警必须能被看见、被处理不能发了告警没人管那还不如不发。6.2 断线重连与数据补偿稳定性的最后一块拼图网络不可能永远稳定。不管是个人宽带的抖动还是云服务器的网络波动连接断开都不可避免。所以断线重连机制必须从一开始就纳入设计。一个健壮的重连逻辑需要考虑这么几件事退避策略断线后不要立即疯狂重连应该依次等待1秒、2秒、4秒、……到达上限比如30秒后固定频率重试。这叫指数退避防止服务端把你当成攻击性连接封掉。数据补偿重连成功后断线期间缺失的行情数据怎么补L2高频数据一般无法完整补回但至少可以拉取断线期间的最新快照和K线把策略状态恢复到当前时刻。这个补偿逻辑越简单越好别在盘中做复杂的历史数据回填。状态重置重连后之前维护的滚动窗口、当日高低点等数据可能已经不准确了。策略必须能从“数据缺失”状态安全恢复到“正常计算”状态必要时要清空部分缓存重新开始积累。我见过最典型的错误是重连成功后程序继续用断线前的旧数据跑策略结果策略按照过时信号下单。正确做法是重连成功后先拉取一次全量快照把SnapshotStore更新到最新再恢复订阅。在确认数据恢复新鲜之前暂停策略交易信号产生或者直接进入“观察模式”只更新状态不下单。6.3 实盘运行中我踩过的几个经典错误按真实经历整理几个我踩过、也是周围朋友经常踩的坑希望能帮你省去几周的排查时间。第一个坑时区与时间戳的错位。历史数据的时间戳可能是UTC时间实时行情则是本地时间北京时间如果直接混用K线的切分时间就会错乱导致策略在错误的边界触发信号。解决办法是在数据标准化阶段统一把所有时间转成“毫秒级epoch时间戳”任何策略逻辑都不要直接依赖字符串形式的时间。我因为这个坑曾经在回测里发现K线“穿越”排查了两天才发现是时区问题。第二个坑把“收到行情的时间”当成“行情发生的时间”。推送模式下数据从交易所到你的程序之间存在延迟。当你用行情自带的时间戳做bar切分时这是没问题的但如果你用本地收到消息的时间去切分遇到网络抖动就会让整段K线时间轴错乱。所以我在代码里坚持使用行情自带的时间戳作为业务时间本地时间只用于监控和数据到达延迟分析。第三个坑L2数据高频推送导致程序内存持续增长。如果你的程序在逐笔行情下运行几分钟后内存飙高多半是哪个容器的数据只进不出。我会在开发阶段给每个核心容器加一个“最大长度”或“过期时间”的上限并且每天收盘后做一次内存泄漏检查。这个检查其实很简单收盘后记录程序内存占用第二天开盘前再看一次如果比前一日高很多基本可以确定有泄漏。第四个坑把策略逻辑和行情接入混在一个文件里。哪怕你的项目再小我也建议至少拆成provider.py、quote_engine.py、strategy.py、risk.py这几个模块。这样做的好处是数据源要换、策略要改、风控要调各改各的互不影响。我知道很多人喜欢把所有代码写在一个notebook或一个py文件里方便是方便但等代码量超过1000行维护成本就直线上升了。6.4 常见问题速查表下面这个表整理了我被问到最多、也是最常出问题的场景可以直接当排查手册参考现象可能原因排查/解决办法策略收不到任何行情连接未建立/订阅失败检查连接状态、订阅请求是否被拒、API Key是否有效行情时断时续网络不稳定/心跳超时被杀增加心跳保活间隔、检查本地网络丢包率行情延迟越来越大回调函数处理耗时过长队列积压把重计算移出回调用异步队列解耦回测和实盘结果差异大回测与实盘数据/逻辑不一致统一事件框架回测也走同一套聚合与策略代码策略重复下单行情重复推送/信号非幂等用已处理时间戳去重或信号与订单状态双重确认五档盘口数据不完整涨跌停/集合竞价阶段盘口不全增加行情状态判断异常盘口直接跳过或告警程序内存持续增长容器或缓存无上限检查所有deque/dict/list是否设了最大长度重连后策略乱下单断线期间策略状态过期重连后先拉快照更新缓存再恢复事件流必要时暂停信号这张表的核心思路是先判断问题出在数据链路还是策略逻辑再层层定位。数据链路的问题连接、延迟、重复通常通过日志和监控能立刻发现策略逻辑的问题非幂等、状态错乱则需要结合交易记录和信号日志来复盘。我自己排查问题时的顺序永远是连接状态→数据新鲜度→队列积压→策略信号→执行回报按照这条链路逐层检查绝大多数问题都能在几分钟内定位。7. 一个完整的最小可运行示例前面讲了不少设计原则和代码片段这一节我把一个完整的最小示例串起来从主程序入口到策略执行让你能直接看到各个模块是怎么协作的。这个示例模拟接入一个WebSocket行情源收到快照后聚合为1分钟K线K线完成后送入一个简单的策略模块策略产生信号后经过风控检查再打印交易指令。为了不引入太多项目外的依赖行情源部分我用一个“仿真数据源”代替真实API主流程和真实场景完全一致。import time import threading # 假设下面这些类都在各自模块里定义好了 # from quote_engine import QuoteEngine # from bar_aggregator import BarAggregator # from snapshot_store import SnapshotStore # from risk_manager import RiskManager class FakeMarketProvider: 仿真行情源模拟WebSocket推送每秒推送一次快照 def __init__(self, symbols): self.symbols symbols self.price_map {s: 100.0 for s in symbols} self.callback None self._running False def subscribe(self, callback): self.callback callback self._running True thread threading.Thread(targetself._run, daemonTrue) thread.start() def _run(self): i 0 while self._running: for symbol in self.symbols: # 模拟价格随机波动 self.price_map[symbol] ( (i * 7 len(symbol)) % 11 - 5 ) / 100.0 quote { symbol: symbol, time: int(time.time() * 1000), # 毫秒时间戳 price: round(self.price_map[symbol], 2), last_volume: (i % 5 1) * 100, bids: [{price: self.price_map[symbol] - 0.01, volume: 1000, orders: 3}], asks: [{price: self.price_map[symbol] 0.01, volume: 800, orders: 2}], } if self.callback: self.callback(quote) time.sleep(0.5) def close(self): self._running False class DummyStrategy: 演示策略连续上涨N次后生成买入信号 def __init__(self, risk_manager): self.risk_manager risk_manager self.rising_count 0 def on_bar(self, bar: dict): symbol bar[symbol] change (bar[close] - bar[open]) / bar[open] if bar[open] else 0 if change 0: self.rising_count 1 else: self.rising_count 0 if self.rising_count 3: volume 100 if self.risk_manager.check_order( symbol, volume, bar[close], account_total100000, current_position0 ): print(f产生买入信号: {symbol} 数量{volume} 价格{bar[close]}) self.rising_count 0 def main(): # 初始化各个模块 snapshot_store SnapshotStore() risk_manager RiskManager() strategy DummyStrategy(risk_manager) # K线聚合器K线完成后回调策略 aggregator BarAggregator( timeframe_seconds60, on_barstrategy.on_bar ) # 行情回调先更新快照缓存再喂给K线聚合器 def on_quote(quote): snapshot_store.update(quote) aggregator.update(quote) # 行情源仿真和行情引擎 provider FakeMarketProvider(symbols[600519, 000001]) engine QuoteEngine(provider, strategy_handlertype( Handler, (), {on_quote: staticmethod(on_quote)} )) engine.start() try: while True: time.sleep(1) except KeyboardInterrupt: print(程序退出) engine.stop() if __name__ __main__: main()运行这个示例你会看到程序每收到一条快照就更新快照缓存聚合器在整分钟切换时生成一根新K线策略模块计算连续上涨次数达到阈值后尝试通过风控并打印信号。整个流程和真实系统的模块划分几乎一模一样只是行情源和交易执行被替换成了仿真实现。你可以在这个骨架上替换成真实的WebSocket数据源、真实的broker接口就基本具备实盘雏形了。我的经验是先用仿真数据源把整套事件流打通再对接到真实行情源。这样可以排除“数据源问题”和“工程框架问题”这两个变量。等框架稳定了接真实数据源时即使有问题也容易定位到数据源这一层而不是在层层叠叠的代码里大海捞针。最后再分享一点我个人的体会。我从最早用while循环加sleep轮询行情到现在用事件驱动加层次化解耦技术方案变了很多但最值钱的经验反而是在开始写代码之前先把“数据从哪来、到哪去、中间经过谁”这条链路画清楚。很多项目起初看着简单越到后面越复杂如果没有清晰的分层代码就会变成一团互相纠缠的线。各位在实战里不管用什么数据源、写什么策略只要守住“数据层不管策略策略层不碰网络”这条边界系统就垮不到哪里去。希望这套设计思路能帮你在行情接入的工程路上少走几步弯路。
实时行情与五档行情接入实战：从数据API到策略层的完整架构设计
做实时行情接入这事儿我最早是在一个凌晨被电话叫醒的。客户说策略跑得好好的突然不更新了我一查日志行情连接断了已经四个多小时代码里硬编码的重试逻辑根本扛不住推送型数据的断线重连。那会儿我意识到行情接入不是“调一个API拿到数据”那么简单从数据API到策略层中间那一大段路才是一个量化工程师真正的日常。这篇文章主要讲我在实战里怎么设计实时行情和五档行情接入方案的。内容适合几类人看一是刚接触量化、想把聚宽/米筐/Tushare这类数据源接到自己策略里的Python开发者二是已经在跑策略但受够了“数据断线、字段对不上、回测实盘不一致”这些问题的人三是想了解L2五档行情到底比普通快照多出什么价值的从业者。我尽量把从数据源选型、工程结构设计、代码实现到排障经验一次讲透不绕弯全是我实际踩过坑后的整理。需要说明的是文中的代码和方案是基于我多年实盘项目的常规实践总结出来的行业里靠谱的做法基本大同小异你可以直接照着搭也可以按自己环境调整。1. 实时行情与五档行情的本质差异1.1 行情数据到底分几层很多人一开始接触的数据接口比如Tushare、AkShare这类拿到的其实是日线数据和分钟线数据。这类数据的特点是非实时、历史归档、低频。它们适合做回测、做盘后分析但如果你要做日内短线、做盘口博弈、做高频择时日线和分钟线根本不够用因为价格在日内是怎么走出来的订单簿是怎么变化的这些信息都已经丢掉了。再往上一个层级就是实时快照。它通常是每3秒推送一次沪深两市的普通行情快照频率包含最新价、涨跌幅、成交量、成交额、买卖五档有时只有一档等字段。这是绝大多数个人量化策略用得最多的数据。它解决了“实时性”的问题但你看到的盘口是一个时间点的切片看不到两次快照之间发生了什么。再往上就是五档行情和逐笔成交。这里需要区分两个概念普通的Level-1行情也提供五档报价但它是3秒一次的快照而真正的Level-2五档行情是实时推送的每当盘口任何一个档位的挂单量或挂单价发生变化就会推一条数据过来频率可以达到毫秒级。L2还附带逐笔委托、逐笔成交这些更细的数据。这里说的五档指的是买一至买五、卖一至卖五共10个价位每个价位上有对应的委托量和委托笔数。把这三个层级放在一起看你会发现越往上数据量越大、延迟越低、信息粒度越细同时对工程能力的要求也越高。很多策略设计之初没有想清楚自己到底需要哪一层数据结果要么用高频数据跑低频逻辑白白浪费性能要么用低频数据硬撑高频策略结果频繁失效。所以动手前先量化一下自己的需求策略最短持仓周期是多少如果我们做的是分钟级以上的交易3秒快照完全够用如果是秒级或毫秒级博弈那L2逐笔是硬门槛。这个判断直接决定了后面所有设计。1.2 五档行情的独特价值与适用场景我做策略编程这十几年经常被问到一个问题普通快照也有五档数据为什么要单独强调“五档行情接进策略”这里面的差距主要体现在三个维度上。第一是频次。3秒一次的快照五档和实时变化的五档信息量差了几个数量级。拿一只成交活跃的股票来说3秒内盘口可能已经变了十几次。你看到的快照五档很可能已经是“过去式”了如果你依据它判断买卖压力天然就滞后。L2五档能看到盘口动态变化的连续性比如买一挂单一瞬间被吃掉又立刻补上这种信息只在逐笔级数据里才有。第二是挂单撤单行为。L2里最值钱的信息之一就是挂单和撤单的明细。主力资金经常用“挂大单吸引跟风、成交前撤单”的手法这在3秒快照里几乎不可能被发现——等方式你看到那笔大单它已经消失了。而L2数据能记录委托进入订单簿的完整轨迹你可以统计挂撤比、大单活跃度等特征。但这有个前提你得有能力分析这些高频数据流否则拿到L2也只是多了几行数字。第三是盘口重构能力。有了L2的逐笔委托和成交数据你可以精确重建交易时段内任意一个时刻的订单簿状态。这意味着你可以做非常精细的微观结构研究比如计算买卖压力不平衡度、识别大单方向、评估冲击成本等。单纯靠快照是不具备这个能力的。所以什么场景适合用五档我的经验是高频做市、高频统计套利、日内T0策略以及需要精确计算冲击成本和订单簿状态的算法交易这些场景离不开L2。而对于日频选股、分钟级趋势跟踪策略用L2反而是性能浪费——你要处理的微结构特征在分钟线里已经被抹平了。这篇文章后面要讲的工程设计其实同时覆盖了快照接入和L2接入因为底层的数据分发框架是相通的。2. 数据API到策略层的整体架构设计2.1 从API到策略中间到底发生了什么很多人第一次写实时行情策略代码是这样的import time from some_api import get_quote while True: quote get_quote(600519) if quote[last_price] my_ma20: buy() time.sleep(3)这段代码如果拿来做演示没问题但拿到实盘环境里跑几周内一定会出问题。问题不在逻辑而在工程。API返回的数据格式可能变更网络随时可能断开行情推送的节奏和自己策略的计算节奏不匹配策略里某个函数抛异常导致整个循环退出等等。这就是为什么要强调“从数据API到策略层的工程设计”。一个能稳定运行的行情接入系统中间至少要经过这么几个环节连接管理层维护和行情服务器之间的连接状态处理鉴权、心跳、断线重连。数据接收层接收原始推送或轮询返回的数据做初步的格式解析。数据标准化层把不同数据源的数据统一成内部格式解决字段命名不一致、单位不一致、时间时区不一致的问题。缓存与快照层维护当前最新的行情快照供策略按需查询同时维护必要的近期历史数据比如最近N根K线、最近N笔成交。事件分发层把新到的行情数据包装成事件按订阅关系分发到不同的策略模块。策略层策略逻辑本身接收到行情事件后更新自身状态、产生信号、下单。如果少了其中任何一层短期看不出问题长期必然崩。我有一次接手别人的策略代码发现他的策略模块里直接塞了API连接代码策略重启就得重建连接盘中连接断了还会把策略状态搞乱。这个架构说白了是在用单机进程写一个微型“消息总线”我后来把所有行情接入统一到一个中间层策略只订阅事件不再关心数据从哪来。2.2 拉取模式与推送模式两种接入模型对比行情接入方案从数据获取方式上看本质上只有两种模式Pull拉取和Push推送。搞清楚它们的区别很多架构选型问题就迎刃而解。拉取模式你的程序主动向数据服务器发起请求拿到某一时刻的数据。比如你每隔3秒调一次RESTful API获取最新快照。这种方式实现简单、调试方便不需要维护长连接但它有两个天然问题一是延迟取决于你的轮询间隔永远比实时慢二是有请求频率限制不可能做到毫秒级。它适合数据频率要求不高、策略逻辑简单、或者只做盘后分析的场景。推送模式你建立一条长连接通常是WebSocket或自定义TCP协议服务器有新的行情数据就主动推给你。这种方式延迟低、实时性好数据一到就能触发策略逻辑是实盘行情接入的主流方式。缺点是需要处理连接生命周期、心跳保活、断线重连、消息顺序、积压处理等一堆问题。L2五档行情几乎都是推送模式因为数据量太大拉取根本跟不上。我的建议是如果只是做日线和分钟级策略拉取模式够用只要涉及秒级以下的实时决策就必须上推送模式。这两个模式对应的工程复杂度是数量级差异所以在项目启动前就要定下来。有些免费数据接口同时提供REST和WebSocket两种方式的我一般都建议实盘用WebSocketREST可以用来做启动时的补数据。2.3 分层设计的核心思想让策略层不感知数据来源我个人在做这套系统时最核心的一条设计原则就是策略层不应该关心数据是从哪里来的也不应该关心数据是REST轮询拿到的还是WebSocket推送过来的。为了实现这一点我定义了一个统一的数据接入接口class MarketDataProvider(ABC): 行情数据提供者抽象接口 abstractmethod def subscribe(self, symbols: list[str], callback: Callable[[dict], None]) - None: 订阅标的数据注册回调函数 ... abstractmethod def get_snapshot(self, symbol: str) - dict | None: 获取指定标的最新快照 ... abstractmethod def close(self) - None: 关闭连接、释放资源 ...任何实现这个接口的类不管底层是聚宽、米筐、还是自己搭的行情网关对策略来说都是一样的。策略只需要在初始化时拿到一个MarketDataProvider实例然后调用subscribe注册自己的回调函数剩下的细节全部被隔离在实现类里。这个抽象的收益在换数据源的时候体现得最明显。我在一个项目里客户一开始用的免费数据源测试后来数据质量达不到要求换成了商业L2数据源策略代码一行没改只换了provider的实现类。如果你把API调用直接写进策略里光是替换数据源就得重构一遍整个策略模块。3. 数据源选型与API接入细节3.1 常见行情API类型与选型建议在写代码之前先解决数据源的问题。市面上可选的行情数据源按数据质量和费用可以分为这么几类类型代表方案实时性五档L2费用适合场景免费数据接口Tushare Pro、AkShare日线/分钟线为主不提供免费学习、回测、低频策略券商/第三方量化平台聚宽JoinQuant、米筐RiceQuant实时快照部分提供按策略收费平台内策略开发商业数据API恒生、Wind、迅投QMT等实时/逐笔完整提供费用较高专业实盘、高频自建行情网关对接交易所行情源或期货公司柜台实时/逐笔完整提供硬件通道成本专业机构、高频团队对个人开发者和中小团队我最常用的组合是学习阶段用Tushare Pro或AkShare做历史回测实盘阶段如果做A股直接用券商提供的QMT或Ptrade接口做加密货币就接交易所的WebSocket接口。没有必要一上来就花大价钱买商业行情源。选型时我通常关注几个指标数据延迟快照延迟多少毫秒/秒、字段完整性有没有五档、逐笔、买卖队列、历史数据深度能不能拿到分钟级历史、接口稳定性会不会经常断流、限流、费用结构按量计费还是包年。免费接口虽然省成本但经常有请求频率限制而且节假日、盘中和盘后的数据一致性偶尔会出问题这些都要在接入前评估好。这里顺便说一句很多人搜“股票数据接口api 免费”然后找到一些别人打包好的接口直接用。我的建议是能一线对接就一线对接尽量不要用不明来历的第三方代理接口。一是不安全你的代码和交易数据可能被人截获二是不稳定代理接口随时可能停止服务。免费接口中Tushare Pro和AkShare在社区里口碑相对靠谱可以作为起步选择。3.2 一个完整的WebSocket行情接入示例下面给出一个我实际用过的、基于WebSocket的快照行情接入代码示例。这个示例用通用结构演示不绑定某一家厂商方便你迁移到自己的数据源上import json import threading import time import websocket class WebSocketQuoteProvider: 基于WebSocket的实时行情提供者推送模式 def __init__(self, ws_url, api_key, symbols, callback): self.ws_url ws_url self.api_key api_key self.symbols symbols self.callback callback # 收到行情后回调策略层 self.ws None self.connected False self._running False self._heartbeat_interval 15 # 心跳间隔秒数 def _on_message(self, ws, message): 收到服务端推送的消息 try: data json.loads(message) # 根据你的数据源协议判断是行情快照还是心跳响应 if data.get(type) quote: quote self._normalize_quote(data[data]) self.callback(quote) except Exception as e: # 消息解析失败不能影响主流程 print(f消息解析异常: {e}) def _normalize_quote(self, raw): 标准化行情字段统一内部命名 return { symbol: raw[sec_code], time: raw[timestamp], price: raw[last_px], volume: raw[business_amount], bid_prices: [raw[fbid_px_{i}] for i in range(1, 6)], bid_volumes: [raw[fbid_volume_{i}] for i in range(1, 6)], ask_prices: [raw[fask_px_{i}] for i in range(1, 6)], ask_volumes: [raw[fask_volume_{i}] for i in range(1, 6)], } def _on_open(self, ws): print(行情连接已建立) self.connected True # 连接建立后发送订阅请求 subscribe_msg { op: subscribe, symbols: self.symbols, api_key: self.api_key } ws.send(json.dumps(subscribe_msg)) def _on_error(self, ws, error): print(f连接错误: {error}) self.connected False def _on_close(self, ws, code, msg): print(连接关闭) self.connected False def _heartbeat(self): 心跳保活线程 while self._running: if self.connected: try: self.ws.send(json.dumps({op: ping})) except Exception as e: print(f心跳发送失败: {e}) time.sleep(self._heartbeat_interval) def connect(self): 建立长连接并启动心跳线程 self.ws websocket.WebSocketApp( self.ws_url, on_messageself._on_message, on_errorself._on_error, on_closeself._on_close, on_openself._on_open ) self._running True # WebSocketApp的run_forever是阻塞的放到独立线程运行 ws_thread threading.Thread(targetself.ws.run_forever, daemonTrue) ws_thread.start() heartbeat_thread threading.Thread(targetself._heartbeat, daemonTrue) heartbeat_thread.start() def close(self): self._running False if self.ws: self.ws.close()这个代码有个关键点叫_normalize_quote。不同数据源返回的字段名千奇百怪比如有的叫last_px有的叫latest_price有的叫newPrice。如果不想办法把字段统一你的策略逻辑就会被数据源的字段命名绑架换个数据源就得改策略。所以我把它单独拆成一个方法所有原始字段在这个方法里被翻译成内部统一的字典格式。后面所有模块都只用这个标准化后的格式这就是工程上的“防腐层”。我在代码里还开了心跳线程。推送型连接最怕的不是数据报错而是连接已经断了但你的程序不知道。很多行情服务端会定期检查客户端心跳超时没收到就断开连接。客户端这边也要定时发送心跳保活同时监控最后一次收到数据的时间如果超过阈值没收到任何数据主动判定连接异常并触发重连。这个机制在长连接类应用里是必须的不是可有可无的优化。3.3 五档行情的订阅与数据处理五档行情和普通快照行情的接入流程大体一致差别主要在数据模型和订阅协议上。沪深L2行情的推送频率非常高所以代码实现上有几个地方需要特别处理。先看数据模型。一个典型的五档行情推送事件除了最新价、成交量这些基础字段外还包含买一至买五、卖一至卖五的价格和数量。有些数据源还会带每档的委托笔数这通常只在L2里提供。我的标准五档数据字典长这样l2_quote { symbol: 600519, time: 1699999999.123, # 事件时间毫秒时间戳 last_price: 1750.0, last_volume: 100, # 最后一笔成交量 accum_volume: 1234567, # 累计成交量 accum_amount: 987654321.0, # 累计成交额 bids: [ {price: 1749.99, volume: 1000, orders: 12}, # 买一 {price: 1749.98, volume: 800, orders: 8}, # 买二 # ... 买三、买四、买五 ], asks: [ {price: 1750.01, volume: 600, orders: 5}, # 卖一 # ... ] }这个结构里bids和asks是列表按价格从优到劣排列。这个排列顺序是有讲究的买盘第一个是最高的买入价卖盘第一个是最低的卖出价。你后续计算买卖价差、订单簿不平衡度时直接取第0个元素就是最优档位省去排序逻辑性能更好。处理L2数据时有一个容易踩的坑五档数组长度可能不满5个。比如股票涨停时卖一可能直接没有挂单刚开盘集合竞价结束时盘口可能只有一档两档。所以解析代码里必须做长度判断不能假设每次都能读到完整的五个档位def calc_spread(quote: dict) - float | None: 计算买卖价差盘口不完整时返回None bids quote.get(bids) or [] asks quote.get(asks) or [] if not bids or not asks: return None best_bid bids[0][price] best_ask asks[0][price] if best_bid 0 or best_ask 0: return None return best_ask - best_bid另一个需要注意的点是涨跌停状态。当股票涨停时整个卖盘可能为空盘口严重倾斜跌停时则买盘为空。如果不做状态判断直接拿空盘口去算特征会算出荒谬的结果导致策略产生假信号。标准做法是维护一个trading_status字段明确标记标的当前处于“正常/涨停/跌停/停牌”中的哪种状态策略层拿到异常状态时直接跳过计算或走风控逻辑。有了前面这些基础做五档行情接入策略的工程框架就逐步清晰了下面我按第二部分的分层思路把每一层具体怎么写在代码里展开来说。4. 行情数据与策略层之间的数据流转设计4.1 从回调函数到策略模块事件驱动的核心写法推送型行情的天然编程模型是事件驱动。数据源一有消息就回调我们注册的函数我们的策略逻辑就运行在这个回调函数里。听起来简单细节却很容易失控。最典型的问题如果策略处理逻辑比较复杂比如要做复杂的特征计算、调用外部模型推理那么回调函数里的一次处理就可能耗时几百毫秒甚至几秒。而在这段时间里行情服务端又推过来了几十条新数据你的程序根本处理不过来数据越积越多延迟越来越大最终策略看到的行情越来越“旧”策略表现自然越来越差。解决这个问题的标准方案是回调函数只做“收数据”和“缓存”这两件最轻的事真正的策略计算放到另一个线程或进程里处理。看代码import queue import threading import time class QuoteEngine: 行情引擎接收行情推送异步送给策略处理器 def __init__(self, provider, strategy_handler, buffer_size10000): self.provider provider self.strategy_handler strategy_handler self.buffer queue.Queue(maxsizebuffer_size) self._running False def _on_quote(self, quote: dict): 行情回调只做入队操作快速返回 try: self.buffer.put_nowait(quote) except queue.Full: # 队列已满说明消费速度跟不上丢弃最旧数据或记录告警 # 这里选择丢弃并报警防止内存无限增长 print(行情队列已满丢弃最新行情) def _worker(self): 消费线程从队列取数据交给策略处理器 while self._running: try: quote self.buffer.get(timeout1) self.strategy_handler.on_quote(quote) except queue.Empty: continue except Exception as e: print(f策略处理异常: {e}) def start(self): self._running True thread threading.Thread(targetself._worker, daemonTrue) thread.start() self.provider.subscribe(self._on_quote) def stop(self): self._running False self.provider.close()这个QuoteEngine类做的事情很简单provider拿到行情推送到_on_quote_on_quote只把数据放进队里后台一个worker线程不停地从队里取数据交给策略处理器。这样做的好处是生产者和消费者的速度被解耦了哪怕策略某一次计算超时也不会阻塞行情数据的接收。队列长度要设置上限。我在代码里设了maxsize10000超过就丢弃并打印告警。这是个权衡如果无限积压系统内存一会儿就爆了如果无脑丢弃策略可能漏掉关键行情。实际项目中我会监控队列积压长度如果经常出现队列满的情况说明策略处理能力远低于行情速度需要优化策略逻辑或者分批处理而不是单纯把队列调大。有人可能想问为什么不直接在回调函数里处理省得队列中转我刚开始也是这么干的直到有一次策略里调用了一个第三方指标库碰到极端行情时计算特别慢行情连接直接因为处理超时被服务端断开了。从那以后我再也不在回调里做重活了。回调函数要像接线员一样接电话、记下来、挂电话别在电话里跟人聊天。4.2 快照缓存让策略能随时拿到最新状态事件驱动模式有个特点策略只有在收到新的行情事件时才会被唤醒。但很多时候你的策略需要主动查询某个标的的最新状态。比如策略主循环运行时发现某个持仓股需要重新评估它希望拿到“当前”的盘口状态而不是被动等待下一条行情。要支持这种主动查询就需要一个快照缓存层。它保存所有订阅标的最新的一份标准化行情快照随时提供按标的查询的能力。这个缓存的实现很简单核心就是一个字典专门用个锁保护并发访问import threading class SnapshotStore: 最新行情快照缓存支持并发读写 def __init__(self): self._store {} self._lock threading.RLock() def update(self, quote: dict): 收到新行情时更新缓存 symbol quote[symbol] with self._lock: self._store[symbol] quote def get(self, symbol: str) - dict | None: 获取指定标的最新快照没有则返回None with self._lock: return self._store.get(symbol) def get_all(self) - list[dict]: 获取全部标的最新快照用于策略遍历 with self._lock: return list(self._store.values())别小看这个简单的类。有了它你的策略就可以随时“看一眼”市场而不是只能“等通知”。我在实际项目里还会在快照里额外存一个local_timestamp字段记录本地收到这条数据的时间。为什么因为行情服务端的时间戳是交易所撮合时间但数据到达你的程序可能已经有网络延迟。排查问题时比如要判断是不是网络链路有延迟这两个时间戳的差就是关键证据。这个是实战里很有用的细节一般教程不会提。更进一步快照缓存还可以保存近期数据的滚动窗口。比如虽然你的策略事件周期是逐笔级但你想计算过去5分钟的成交量变化那你需要保存最近5分钟的成交数据。这个滚动窗口用Python的collections.deque实现很合适它支持从两端快速 append 和 pop。我通常会为这个窗口设定最大长度比如最多存10万条防止内存膨胀。4.3 Tick数据到Bar数据的聚合几行代码实现K线构建很多策略并不需要逐笔处理行情而是基于1分钟、5分钟K线做判断。这种情况下你不需要把每条tick都送进策略而是先把tick聚合成bar再把bar事件发给策略。这里的聚合逻辑就是所谓的tick-to-bar。最普通的K线聚合是基于时间窗口的比如每个自然分钟内的所有tick聚合成一根1分钟K线。核心逻辑是记录这一分钟的开盘价、最高价、最低价、收盘价和总成交量分钟结束时触发一次bar事件。看实现import threading import time class BarAggregator: 将tick行情聚合为K线bar def __init__(self, timeframe_seconds: int 60, on_barNone): self.timeframe_seconds timeframe_seconds self.on_bar on_bar # 回调函数K线生成后调用 self._bars {} # symbol - 当前积累中的bar self._lock threading.RLock() def update(self, quote: dict): 每来一条tick更新对应的bar symbol quote[symbol] price quote[price] volume quote.get(last_volume, 0) ts quote[time] bar_time (ts // self.timeframe_seconds) * self.timeframe_seconds with self._lock: bar self._bars.get(symbol) if bar is None or bar[time] ! bar_time: # 新的一根K线开始先触发上一根完成事件 if bar is not None and self.on_bar: self.on_bar(bar) # 初始化新K线 bar { symbol: symbol, time: bar_time, open: price, high: price, low: price, close: price, volume: 0 } self._bars[symbol] bar # 更新K线数据 bar[high] max(bar[high], price) bar[low] min(bar[low], price) bar[close] price bar[volume] volume这个聚合逻辑有个细节判断“新K线开始”用的是当前tick的时间戳除以时间窗口然后取整。也就是整分钟的边界一旦跨过就会自动开启新K线。这避免了用实时钟判断时的各种时区、延迟导致的错位问题。我这里用的是“时间驱动”的简单聚合依赖行情自带的时间戳。如果数据源的tick时间戳本身就是乱的那聚合结果也会乱这时候就需要额外的tick排序逻辑这也是L2数据源考验工程能力的一个点。行情服务器的多路推送之间的先后顺序偶尔会错乱必须先按时间和序号排序再做聚合否则K线的最高最低价都会被算错。5. 策略层对接实战从收到行情到产生交易信号5.1 策略状态机的设计一切决策基于最新快照策略层是整个系统的大脑。行情数据经过前面的一系列加工到这里已经是标准化的、带完整盘口信息的快照或bar了。策略要做的是根据这些数据维护自身状态并在合适的时机产生交易信号或风控指令。策略的自身状态通常包括当前持仓、当前可用资金、每个标的的信号状态、是否已有未完成的订单等。这些状态和行情数据不一样行情数据是外部实时变化的而状态是策略内部维护的、跟随交易行为更新的。我的经验是把这两种数据彻底分开管理行情状态放快照缓存策略状态放策略对象内部。如果不分开很容易出现“策略状态被行情回调改乱了”的问题。一个简单的策略状态机示例class SimpleIntradayStrategy: 一个简单的日内策略示例突破策略 def __init__(self, snapshot_store, broker, break_threshold0.02): self.snapshot_store snapshot_store self.broker broker self.break_threshold break_threshold self.position {} # symbol - 持仓数量 self.day_high {} # symbol - 当日最高价 self.day_low {} # symbol - 当日最低价 def on_bar(self, bar: dict): K线回调更新日内高低点判断突破信号 symbol bar[symbol] # 更新当日高低点 if symbol not in self.day_high or bar[high] self.day_high[symbol]: self.day_high[symbol] bar[high] if symbol not in self.day_low or bar[low] self.day_low[symbol]: self.day_low[symbol] bar[low] # 突破逻辑当前价突破昨日高点一定比例则买入 # 这里省略昨日高点的获取逻辑假设存在self.prev_high字典里 prev_high self.prev_high.get(symbol) if prev_high is None: return if bar[close] prev_high * (1 self.break_threshold): if self.position.get(symbol, 0) 0: self.broker.buy(symbol, volume100) print(f{symbol} 触发突破买入价格 {bar[close]})这段代码体现了一个重要的设计选择策略从bar事件开始计算但决策依据是它自己维护的状态而不是被动的“看到什么就是什么”。比如突破策略要求当日最高点它不能直接拿当前快照的high字段而要自己把每个bar的high累加更新。这样做的好处是你可以随时回放历史数据重建策略状态而且不受行情乱序影响。策略层还有一个常被忽略的点幂等性设计。实时行情推送偶尔会发生重复或者因为断线重连导致同一根K线被重复推送。如果策略的每次信号计算都不是幂等的比如买了信号来一次就买一次那么数据重复就会导致重复开仓。我处理这类问题的手段是在策略里维护已处理的bar时间戳集合或者用“当前signal状态成交回报”双重确认机制确保一个信号只触发一次交易。5.2 策略与回测逻辑一致性别再让实盘和回测“两套逻辑”聊策略层就不能不提一个我见过无数次的坑回测和实盘用的是两套完全不同的行情处理逻辑。回测时你从历史数据里一根根K线遍历过去认为“上一根K线收盘时我就知道这个数据了”实盘时你在实时收到K线的瞬间才拿到数据还要处理盘中突变、断线重连造成的缺口。两套逻辑如果处理方式不一致回测表现再好的策略实盘也大概率拉胯。要解决这个问题核心是让“数据喂给策略”的方式在回测和实盘中保持一致。具体做法是回测时也走同一套事件驱动框架把历史数据“重放”给策略模块。也就是说平时实盘用的BarAggregator、QuoteEngine、Strategy这些模块在回测中原封不动地复用只是把数据来源从“WebSocket推送”换成“文件批量读取”。这里展示一个简单的历史数据回放引擎骨架class BacktestReplayer: 历史数据回放引擎用同一套事件框架回测 def __init__(self, quotes, strategy): self.quotes sorted(quotes, keylambda q: q[time]) self.strategy strategy self.aggregator BarAggregator( timeframe_seconds60, on_barstrategy.on_bar ) def run(self): for quote in self.quotes: # 和实盘一样先更新快照缓存再喂给聚合器 # 这里为了演示简化直接喂给聚合器 self.aggregator.update(quote)理解这段代码的关键是实盘和回测只是数据来源不同数据处理和策略执行的代码路径完全一致。这就是所谓“一次编写两处复用”。如果没有这样一个统一框架你迟早会在回测里发现一个策略表现优异实盘却不知所措然后花整整一周去排查到底哪里不一致。我用这套思路后回测和实盘之间的一致性显著提高再也没出现过因为“数据延迟假设不同”导致的严重失真。5.3 风控模块订单前的最后一道防线策略产生了信号不代表可以直接执行。在信号和交易执行之间还要放一道风控模块。它的职责是检查这个信号是否符合当前风控规则比如单笔下单量是否超过限制、当日累计交易次数是否超限、当前持仓比例是否超过上限、标的是否在禁买名单里等。风控模块的输入是“交易信号”和“当前策略/账户状态”输出是“允许/拒绝”或“允许但调整数量”。这样一来具体的风控规则就可以灵活配置策略层不需要关心这些细节。看一个极简风控实现class RiskManager: 风控模块所有交易信号经过这里过滤 def __init__(self, max_position_ratio0.2, max_trade_per_day10): self.max_position_ratio max_position_ratio self.max_trade_per_day max_trade_per_day self.trade_count {} def check_order(self, symbol: str, volume: float, price: float, account_total: float, current_position: float) - bool: 检查一笔订单是否通过风控 # 规则1单只标的持仓市值不能超过总资产20% order_value volume * price if (current_position order_value) account_total * self.max_position_ratio: return False # 规则2当日交易次数限制 if self.trade_count.get(symbol, 0) self.max_trade_per_day: return False return True这个风控模块虽然简单但位置很重要。我习惯把它放在策略信号产生之后、真正下单之前。信号产生后先过风控风控通过再交给broker执行。这个顺序千万别颠倒否则一旦风控逻辑有bug策略就可能在非预期条件下下单风险很大。风控规则的判定还要考虑极端行情比如涨跌停附近流动性枯竭时即使信号明确也可能无法按期望价格成交。这不是风控模块能解决的需要交易执行模块配合处理比如设置滑点保护、超时撤单等。后面第6节会讲到。6. 工程化落地的关键细节与避坑指南6.1 日志、监控与告警跑起来只是起点行情接入系统能跑起来只是万里长征第一步。真正决定它在生产环境稳定性的是日志、监控与告警这三件套。日志方面我的核心原则是“关键时刻必有日志高频日志必须采样”。连接断开、重连成功、订阅失败、队列满、风控拒绝这些关键事件必须完整记录而且要带上时间戳和上下文信息。但像“收到一条行情”这种每秒几十次的事件不能每条都打日志否则日志文件几小时就能撑爆磁盘。一般做法是抽样打日志比如每1000条打一条或者仅在数据异常时去打。监控方面最实用的几个指标是连接状态是否在线、最近一次行情时间距现在的秒数衡量行情新鲜度、队列积压长度、每秒处理行情数、策略信号产生数。这些指标可以用简单的计数器维护定时打印或上报到监控系统。我用过PrometheusGrafana这样的专业方案也用过只用日志文件加定时任务扫描的极简方案。对个人项目和中小团队来说后者往往已经够用。告警方面最核心的告警规则就三条连接断开超过阈值比如30秒、行情数据超时未更新比如15秒没有任何新行情、策略异常连续出现比如连续5次报错。告警方式可以用邮件、企业微信机器人、或者直接打日志然后由外部监控系统触发选项很多。关键是告警必须能被看见、被处理不能发了告警没人管那还不如不发。6.2 断线重连与数据补偿稳定性的最后一块拼图网络不可能永远稳定。不管是个人宽带的抖动还是云服务器的网络波动连接断开都不可避免。所以断线重连机制必须从一开始就纳入设计。一个健壮的重连逻辑需要考虑这么几件事退避策略断线后不要立即疯狂重连应该依次等待1秒、2秒、4秒、……到达上限比如30秒后固定频率重试。这叫指数退避防止服务端把你当成攻击性连接封掉。数据补偿重连成功后断线期间缺失的行情数据怎么补L2高频数据一般无法完整补回但至少可以拉取断线期间的最新快照和K线把策略状态恢复到当前时刻。这个补偿逻辑越简单越好别在盘中做复杂的历史数据回填。状态重置重连后之前维护的滚动窗口、当日高低点等数据可能已经不准确了。策略必须能从“数据缺失”状态安全恢复到“正常计算”状态必要时要清空部分缓存重新开始积累。我见过最典型的错误是重连成功后程序继续用断线前的旧数据跑策略结果策略按照过时信号下单。正确做法是重连成功后先拉取一次全量快照把SnapshotStore更新到最新再恢复订阅。在确认数据恢复新鲜之前暂停策略交易信号产生或者直接进入“观察模式”只更新状态不下单。6.3 实盘运行中我踩过的几个经典错误按真实经历整理几个我踩过、也是周围朋友经常踩的坑希望能帮你省去几周的排查时间。第一个坑时区与时间戳的错位。历史数据的时间戳可能是UTC时间实时行情则是本地时间北京时间如果直接混用K线的切分时间就会错乱导致策略在错误的边界触发信号。解决办法是在数据标准化阶段统一把所有时间转成“毫秒级epoch时间戳”任何策略逻辑都不要直接依赖字符串形式的时间。我因为这个坑曾经在回测里发现K线“穿越”排查了两天才发现是时区问题。第二个坑把“收到行情的时间”当成“行情发生的时间”。推送模式下数据从交易所到你的程序之间存在延迟。当你用行情自带的时间戳做bar切分时这是没问题的但如果你用本地收到消息的时间去切分遇到网络抖动就会让整段K线时间轴错乱。所以我在代码里坚持使用行情自带的时间戳作为业务时间本地时间只用于监控和数据到达延迟分析。第三个坑L2数据高频推送导致程序内存持续增长。如果你的程序在逐笔行情下运行几分钟后内存飙高多半是哪个容器的数据只进不出。我会在开发阶段给每个核心容器加一个“最大长度”或“过期时间”的上限并且每天收盘后做一次内存泄漏检查。这个检查其实很简单收盘后记录程序内存占用第二天开盘前再看一次如果比前一日高很多基本可以确定有泄漏。第四个坑把策略逻辑和行情接入混在一个文件里。哪怕你的项目再小我也建议至少拆成provider.py、quote_engine.py、strategy.py、risk.py这几个模块。这样做的好处是数据源要换、策略要改、风控要调各改各的互不影响。我知道很多人喜欢把所有代码写在一个notebook或一个py文件里方便是方便但等代码量超过1000行维护成本就直线上升了。6.4 常见问题速查表下面这个表整理了我被问到最多、也是最常出问题的场景可以直接当排查手册参考现象可能原因排查/解决办法策略收不到任何行情连接未建立/订阅失败检查连接状态、订阅请求是否被拒、API Key是否有效行情时断时续网络不稳定/心跳超时被杀增加心跳保活间隔、检查本地网络丢包率行情延迟越来越大回调函数处理耗时过长队列积压把重计算移出回调用异步队列解耦回测和实盘结果差异大回测与实盘数据/逻辑不一致统一事件框架回测也走同一套聚合与策略代码策略重复下单行情重复推送/信号非幂等用已处理时间戳去重或信号与订单状态双重确认五档盘口数据不完整涨跌停/集合竞价阶段盘口不全增加行情状态判断异常盘口直接跳过或告警程序内存持续增长容器或缓存无上限检查所有deque/dict/list是否设了最大长度重连后策略乱下单断线期间策略状态过期重连后先拉快照更新缓存再恢复事件流必要时暂停信号这张表的核心思路是先判断问题出在数据链路还是策略逻辑再层层定位。数据链路的问题连接、延迟、重复通常通过日志和监控能立刻发现策略逻辑的问题非幂等、状态错乱则需要结合交易记录和信号日志来复盘。我自己排查问题时的顺序永远是连接状态→数据新鲜度→队列积压→策略信号→执行回报按照这条链路逐层检查绝大多数问题都能在几分钟内定位。7. 一个完整的最小可运行示例前面讲了不少设计原则和代码片段这一节我把一个完整的最小示例串起来从主程序入口到策略执行让你能直接看到各个模块是怎么协作的。这个示例模拟接入一个WebSocket行情源收到快照后聚合为1分钟K线K线完成后送入一个简单的策略模块策略产生信号后经过风控检查再打印交易指令。为了不引入太多项目外的依赖行情源部分我用一个“仿真数据源”代替真实API主流程和真实场景完全一致。import time import threading # 假设下面这些类都在各自模块里定义好了 # from quote_engine import QuoteEngine # from bar_aggregator import BarAggregator # from snapshot_store import SnapshotStore # from risk_manager import RiskManager class FakeMarketProvider: 仿真行情源模拟WebSocket推送每秒推送一次快照 def __init__(self, symbols): self.symbols symbols self.price_map {s: 100.0 for s in symbols} self.callback None self._running False def subscribe(self, callback): self.callback callback self._running True thread threading.Thread(targetself._run, daemonTrue) thread.start() def _run(self): i 0 while self._running: for symbol in self.symbols: # 模拟价格随机波动 self.price_map[symbol] ( (i * 7 len(symbol)) % 11 - 5 ) / 100.0 quote { symbol: symbol, time: int(time.time() * 1000), # 毫秒时间戳 price: round(self.price_map[symbol], 2), last_volume: (i % 5 1) * 100, bids: [{price: self.price_map[symbol] - 0.01, volume: 1000, orders: 3}], asks: [{price: self.price_map[symbol] 0.01, volume: 800, orders: 2}], } if self.callback: self.callback(quote) time.sleep(0.5) def close(self): self._running False class DummyStrategy: 演示策略连续上涨N次后生成买入信号 def __init__(self, risk_manager): self.risk_manager risk_manager self.rising_count 0 def on_bar(self, bar: dict): symbol bar[symbol] change (bar[close] - bar[open]) / bar[open] if bar[open] else 0 if change 0: self.rising_count 1 else: self.rising_count 0 if self.rising_count 3: volume 100 if self.risk_manager.check_order( symbol, volume, bar[close], account_total100000, current_position0 ): print(f产生买入信号: {symbol} 数量{volume} 价格{bar[close]}) self.rising_count 0 def main(): # 初始化各个模块 snapshot_store SnapshotStore() risk_manager RiskManager() strategy DummyStrategy(risk_manager) # K线聚合器K线完成后回调策略 aggregator BarAggregator( timeframe_seconds60, on_barstrategy.on_bar ) # 行情回调先更新快照缓存再喂给K线聚合器 def on_quote(quote): snapshot_store.update(quote) aggregator.update(quote) # 行情源仿真和行情引擎 provider FakeMarketProvider(symbols[600519, 000001]) engine QuoteEngine(provider, strategy_handlertype( Handler, (), {on_quote: staticmethod(on_quote)} )) engine.start() try: while True: time.sleep(1) except KeyboardInterrupt: print(程序退出) engine.stop() if __name__ __main__: main()运行这个示例你会看到程序每收到一条快照就更新快照缓存聚合器在整分钟切换时生成一根新K线策略模块计算连续上涨次数达到阈值后尝试通过风控并打印信号。整个流程和真实系统的模块划分几乎一模一样只是行情源和交易执行被替换成了仿真实现。你可以在这个骨架上替换成真实的WebSocket数据源、真实的broker接口就基本具备实盘雏形了。我的经验是先用仿真数据源把整套事件流打通再对接到真实行情源。这样可以排除“数据源问题”和“工程框架问题”这两个变量。等框架稳定了接真实数据源时即使有问题也容易定位到数据源这一层而不是在层层叠叠的代码里大海捞针。最后再分享一点我个人的体会。我从最早用while循环加sleep轮询行情到现在用事件驱动加层次化解耦技术方案变了很多但最值钱的经验反而是在开始写代码之前先把“数据从哪来、到哪去、中间经过谁”这条链路画清楚。很多项目起初看着简单越到后面越复杂如果没有清晰的分层代码就会变成一团互相纠缠的线。各位在实战里不管用什么数据源、写什么策略只要守住“数据层不管策略策略层不碰网络”这条边界系统就垮不到哪里去。希望这套设计思路能帮你在行情接入的工程路上少走几步弯路。