4. 实时行情订阅与推送

任务目标

本案例实现一个最小可用的实时行情订阅与推送系统：从行情源订阅指定交易标的的实时Tick数据，并将数据推送给下游策略引擎进行消费。系统采用发布-订阅模式，行情源与策略引擎通过一个线程安全的中间通道解耦，策略引擎无需关心行情源的具体实现。

核心实现

import threading
import queue
import time
import random
from dataclasses import dataclass
from typing import List, Callable, Dict

@dataclass
class TickData:
    symbol: str
    price: float
    volume: int
    timestamp: float

class MarketDataFeed:
    """模拟行情源：按固定频率产生随机Tick数据"""
    def __init__(self, symbols: List[str], interval: float = 0.5):
        self.symbols = symbols
        self.interval = interval
        self._running = False
        self._thread = None

    def start(self, callback: Callable[[TickData], None]):
        self._running = True
        self._thread = threading.Thread(target=self._generate, args=(callback,), daemon=True)
        self._thread.start()

    def stop(self):
        self._running = False
        if self._thread:
            self._thread.join(timeout=1)

    def _generate(self, callback: Callable[[TickData], None]):
        base_prices = {s: random.uniform(10, 500) for s in self.symbols}
        while self._running:
            for sym in self.symbols:
                # 模拟价格随机游走
                base_prices[sym] *= (1 + random.uniform(-0.005, 0.005))
                tick = TickData(
                    symbol=sym,
                    price=round(base_prices[sym], 2),
                    volume=random.randint(1, 100),
                    timestamp=time.time()
                )
                callback(tick)
            time.sleep(self.interval)

class TickDispatcher:
    """线程安全的行情分发通道：订阅-发布模式"""
    def __init__(self):
        self._subscribers: Dict[str, List[Callable[[TickData], None]]] = {}
        self._lock = threading.Lock()

    def subscribe(self, symbol: str, handler: Callable[[TickData], None]):
        with self._lock:
            if symbol not in self._subscribers:
                self._subscribers[symbol] = []
            self._subscribers[symbol].append(handler)

    def publish(self, tick: TickData):
        with self._lock:
            handlers = list(self._subscribers.get(tick.symbol, []))
        for h in handlers:
            h(tick)

class StrategyEngine:
    """策略引擎：订阅行情并处理Tick"""
    def __init__(self, dispatcher: TickDispatcher, symbols: List[str]):
        self.dispatcher = dispatcher
        self.symbols = symbols
        self.received_count = 0
        self.last_prices: Dict[str, float] = {}

    def start(self):
        for sym in self.symbols:
            self.dispatcher.subscribe(sym, self.on_tick)

    def on_tick(self, tick: TickData):
        # 策略处理逻辑：记录最新价格并计数
        self.last_prices[tick.symbol] = tick.price
        self.received_count += 1

def main():
    symbols = ["AAPL", "MSFT"]
    dispatcher = TickDispatcher()
    engine = StrategyEngine(dispatcher, symbols)
    engine.start()

    feed = MarketDataFeed(symbols, interval=0.2)
    feed.start(dispatcher.publish)

    # 运行3秒后停止
    time.sleep(3)
    feed.stop()

    print(f"策略引擎收到Tick总数: {engine.received_count}")
    for sym in symbols:
        print(f"{sym} 最新价格: {engine.last_prices.get(sym, 'N/A')}")

if __name__ == "__main__":
    main()

运行方式

python market_data_pipeline.py

验证方法

程序运行约3秒后自动退出，观察输出：

• 策略引擎收到Tick总数应大于0，说明行情数据成功从数据源推送至策略引擎；
• 每个标的的最新价格应显示为有效浮点数（非N/A），说明策略引擎正确解析并存储了对应标的的行情；
• 多次运行程序，每次输出的价格和数量不同，说明行情源在持续产生动态数据。

关键设计说明

本案例的核心解耦点在于TickDispatcher：行情源只调用publish推送数据，策略引擎通过subscribe注册回调，两者互不感知。实际生产环境中，可将MarketDataFeed替换为交易所或券商API的行情回调，将StrategyEngine.on_tick替换为真实策略逻辑，中间的分发层无需改动。若需跨进程推送，可将TickDispatcher替换为消息队列（如ZeroMQ、Kafka）的发布-订阅实现，接口保持一致即可。