流式行情接入端到端实战

上海一家 私募 中等频率股票策略团队的量化开发收到任务：两周内从零搭一条 沪深300 ETF 的 ticker plant。手头握住 3.6.3 的仓库（TimescaleDB hypertable，ticks_raw(symbol, ts, price, size, side)，(symbol, ts) 主键）、L1 的消息词汇、L2 的 Kafka producer/consumer 配置、L3 的四级延迟预算。交付物是一条同事拿到就能跑起来的端到端管道：一个 Kafka topic、一个合成行情 producer、一个把 tick 落到 TimescaleDB 的 feed-handler 消费者、一个并行的实时 VWAP 监控；都跑在同一份 supervisor 脚本下。本课就是把这些拼在一起的 capstone；这里每一个选择都引用 L1、L2、L3 或 3.6.3 L4 的一条规则。

管道形态

行业里的典范 ticker-plant 模式：一个 feed-handler 进程消费一条 行情 流，把记录 normalise 成仓库 schema、落到 TSDB；一个并行的实时监控消费者读同一条流做实时分析。Capstone 就是把 3.6.4 的「流」一半接到 3.6.3 的「存」一半的那根桥。七个交付物，进同一个项目目录：

• docker-compose.yml — 单 broker Kafka + 单实例 TimescaleDB。
• migrations/001_create_ticks_raw.sql — 仓库 schema（沿 3.6.3 L4 的 alembic-equivalent 迁移）。
• producer.py — 合成行情，JSON 走 Kafka，按 symbol 做 key。
• feed_handler.py — feedhandler-warehouse 组里的消费者，按批写入 TimescaleDB。
• vwap_monitor.py — realtime-vwap-monitor 组里的消费者，从内存 deque 滚出 1 分钟 VWAP。
• run.sh — Shell supervisor（set -euo pipefail、trap、结构化 JSON 日志），拉起三个 Python 进程。
• README.md — 前置依赖、运行方式、验证方式、幂等测试方式。

第 1 步：基础设施

开发阶段的基础设施是同一个 Docker 网络上的两个 service。docker-compose.yml：

services:
  kafka:
    image: confluentinc/cp-kafka:7.6.0
    environment:
      KAFKA_PROCESS_ROLES: 'broker,controller'
      KAFKA_NODE_ID: 1
      KAFKA_CONTROLLER_QUORUM_VOTERS: '1@kafka:9093'
      KAFKA_LISTENERS: 'PLAINTEXT://0.0.0.0:9092,CONTROLLER://0.0.0.0:9093,EXTERNAL://0.0.0.0:19092'
      KAFKA_ADVERTISED_LISTENERS: 'PLAINTEXT://kafka:9092,EXTERNAL://localhost:19092'
      KAFKA_LISTENER_PROTOCOL_MAP: 'PLAINTEXT:PLAINTEXT,CONTROLLER:PLAINTEXT,EXTERNAL:PLAINTEXT'
      KAFKA_CONTROLLER_LISTENER_NAMES: 'CONTROLLER'
      KAFKA_INTER_BROKER_LISTENER_NAME: 'PLAINTEXT'
      CLUSTER_ID: '4L6g3nShT-eMCtK--X86sw'
    ports:
      - "19092:19092"
  timescaledb:
    image: timescale/timescaledb:latest-pg16
    environment:
      POSTGRES_DB: warehouse
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: postgres
    ports:
      - "5432:5432"

生产环境会是三 broker Kafka 集群（副本因子 3）+ TimescaleDB Multinode；本课为工作例子单节点起即可——应用代码两边完全相同。镜像 tag 钉死（7.6.0、latest-pg16）以便复跑可重复；A-股 量化 firm 在内网环境中通常镜像这些公共镜像到私有 registry 后再拉，而不是直接拉 docker.io。容器构建、Dockerfile、Kubernetes 清单的撰写都向后指 3.6.5 (Build, Deploy & Containers)。

第 2 步：schema 迁移

migrations/001_create_ticks_raw.sql（沿 3.6.3 L4 的 alembic-equivalent SQL 迁移）：

CREATE TABLE IF NOT EXISTS ticks_raw (
  symbol TEXT NOT NULL,
  ts TIMESTAMPTZ NOT NULL,
  price NUMERIC(18,6) NOT NULL,
  size BIGINT NOT NULL,
  side CHAR(1) NOT NULL,
  source_file TEXT NOT NULL DEFAULT 'kafka-stream',
  ingested_at TIMESTAMPTZ NOT NULL DEFAULT now(),
  PRIMARY KEY (symbol, ts)
);
SELECT create_hypertable('ticks_raw', 'ts', chunk_time_interval => INTERVAL '7 days', if_not_exists => true);
CREATE INDEX IF NOT EXISTS idx_ticks_raw_ts_brin ON ticks_raw USING BRIN (ts);

一次执行：psql -h localhost -U postgres warehouse -f migrations/001_create_ticks_raw.sql。生产环境的 schema 变更必须来自 git 上的迁移文件——3.6.2 + 3.6.3 L4 的规则。PRIMARY KEY (symbol, ts) 是消费端幂等的锚点，它让 at-least-once 变得安全。

第 3 步：创建 Kafka topic

kafka-topics --bootstrap-server localhost:19092 --create --topic ticks.sse.510300 --partitions 8 --replication-factor 1 --config min.insync.replicas=1 --config retention.ms=604800000

8 分区给消费并行；单 broker 所以副本因子 1（生产 3）；retention 7 天。

第 4 步：producer.py

Producer 模拟一条 normalised 行情流；实际生产里它的输入会是从交易所多播（CFFEX CTP-MD、上交所 NeoArc、深交所 行情）解出的 SBE template（L3），wire format 也会是 Protobuf 或 SBE 而非 JSON（L1）。为了让 kcat 看得清，这里用 JSON。

from confluent_kafka import Producer
import json, time, random, sys

TOPIC = 'ticks.sse.510300'
SYMBOLS = ['600519', '000001', '510300']

producer = Producer({
    'bootstrap.servers': 'localhost:19092',
    'acks': 'all',
    'enable.idempotence': True,
    'linger.ms': 5,
    'compression.type': 'lz4',
    'max.in.flight.requests.per.connection': 5,
    'batch.size': 65536,
})

def delivery_cb(err, msg):
    if err is not None:
        print(json.dumps({"ts": time.time(), "stage": "produce_error", "err": str(err)}), flush=True)
    else:
        print(json.dumps({"ts": time.time(), "stage": "produced", "partition": msg.partition(), "offset": msg.offset()}), flush=True)

i = 0
while True:
    symbol = SYMBOLS[i % len(SYMBOLS)]
    price = round(4.0 + random.gauss(0, 0.01), 4)
    size = random.choice([1000, 5000, 10000])
    side = random.choice(['B', 'S'])
    tick = {'symbol': symbol, 'ts_ns': time.time_ns(), 'price': price, 'size': size, 'side': side}
    producer.produce(topic=TOPIC, key=symbol.encode(), value=json.dumps(tick).encode(), callback=delivery_cb)
    producer.poll(0)
    time.sleep(0.02)
    i += 1

L2 的七个 producer-config key 全在。key=symbol.encode() 把分区规则落地：所有 510300 的 tick 落同一个分区，组里只有一个消费者按生产端顺序读到这些 tick。Tick 的五个 JSON 字段是 symbol、ts_ns、price、size、side。producer.poll(0) 不阻塞地处理 delivery-report 回调。

第 5 步：feed_handler.py

经典 ticker-plant 消费者：

from confluent_kafka import Consumer, KafkaError
import json, time, sys
import psycopg

TOPIC = 'ticks.sse.510300'

consumer = Consumer({
    'bootstrap.servers': 'localhost:19092',
    'group.id': 'feedhandler-warehouse',
    'enable.auto.commit': False,
    'auto.offset.reset': 'earliest',
    'isolation.level': 'read_committed',
})
consumer.subscribe([TOPIC])

conn = psycopg.connect("host=localhost user=postgres dbname=warehouse password=postgres")
cur = conn.cursor()

batch = []
batch_start = time.monotonic()

while True:
    msg = consumer.poll(timeout=1.0)
    if msg is not None and msg.error() is None:
        tick = json.loads(msg.value())
        ts = time.gmtime(tick['ts_ns'] / 1_000_000_000)
        batch.append((tick['symbol'], tick['ts_ns'], tick['price'], tick['size'], tick['side'], 'kafka-stream'))
    if batch and (len(batch) >= 1000 or time.monotonic() - batch_start >= 0.2):
        cur.executemany(
            'INSERT INTO ticks_raw (symbol, ts, price, size, side, source_file) VALUES (%s, %s, %s, %s, %s, %s) ON CONFLICT (symbol, ts) DO NOTHING',
            batch,
        )
        conn.commit()
        consumer.commit(asynchronous=False)
        print(json.dumps({"ts": time.time(), "stage": "batch_committed", "rows": len(batch), "first_offset": msg.offset() - len(batch) + 1, "last_offset": msg.offset()}), flush=True)
        batch = []
        batch_start = time.monotonic()

L2 的五个 consumer-config key 全在。Batch 落地触发：len(batch) >= 1000 或 time.monotonic() - batch_start >= 0.2。操作顺序绝对：INSERT -> conn.commit() -> consumer.commit()。​​双幂等​​性质成立：消费者在 INSERT 成功之后、consumer.commit 之前崩了，重启会重新看到这批数据，但 ON CONFLICT (symbol, ts) DO NOTHING 让第二次 INSERT 变 no-op；行数和 offset 都保持正确。

第 6 步：vwap_monitor.py

并行消费者演示 pub-sub 扇出：订阅相同 topic、用一个​​不同​​的消费者组 (realtime-vwap-monitor)，因此它独立于仓库 writer 看到每一条消息——L1 的经典 pub-sub 属性。

from confluent_kafka import Consumer
from collections import defaultdict, deque
import json, time, sys

TOPIC = 'ticks.sse.510300'
WINDOW_NS = 60 * 1_000_000_000

consumer = Consumer({
    'bootstrap.servers': 'localhost:19092',
    'group.id': 'realtime-vwap-monitor',
    'enable.auto.commit': True,
    'auto.offset.reset': 'latest',
})
consumer.subscribe([TOPIC])

windows: defaultdict = defaultdict(deque)
last_emit = defaultdict(float)

while True:
    msg = consumer.poll(timeout=1.0)
    if msg is None or msg.error() is not None:
        continue
    tick = json.loads(msg.value())
    sym = tick['symbol']
    now_ns = tick['ts_ns']
    windows[sym].append((now_ns, tick['price'], tick['size']))
    while windows[sym] and now_ns - windows[sym][0][0] > WINDOW_NS:
        windows[sym].popleft()
    now = time.time()
    if now - last_emit[sym] >= 1.0:
        notional = sum(p * s for _, p, s in windows[sym])
        volume = max(sum(s for _, _, s in windows[sym]), 1)
        vwap = notional / volume
        print(json.dumps({"ts": now, "stage": "vwap_1m", "symbol": sym, "vwap": round(vwap, 4), "window_size": len(windows[sym])}), flush=True)
        last_emit[sym] = now

监控是纯内存消费者、不在乎耐久性；重启时从 'latest' 接，若 Kafka retention 已老化掉最早记录，监控就从空窗口起步再填满。

第 7 步：run.sh

Shell 监督者，沿 3.6.1 L4 模式：

#!/usr/bin/env bash
set -euo pipefail
IFS=$'\n\t'

log() { printf '{"ts":"%s","stage":"%s","pid":"%s","status":"%s"}\n' "$(date -u +%FT%TZ)" "$1" "$2" "$3"; }

PIDS=()
cleanup() {
  for p in "${PIDS[@]}"; do kill -TERM "$p" 2>/dev/null || true; done
}
trap 'cleanup' EXIT INT TERM

python -u producer.py &        ; PIDS+=($!) ; log producer_started "${PIDS[-1]}" ok
python -u feed_handler.py &    ; PIDS+=($!) ; log feed_handler_started "${PIDS[-1]}" ok
python -u vwap_monitor.py &    ; PIDS+=($!) ; log vwap_monitor_started "${PIDS[-1]}" ok

wait -n

3.6.1 的 set -euo pipefail；IFS=$'\n\t' 屏蔽空白分词；log() 辅助函数输出结构化 JSON 给 3.6.6 用；PIDS=() 加 cleanup() 配 trap 'cleanup' EXIT INT TERM 让 Ctrl-C 把 SIGTERM 传到全部 child；python -u 关掉 stdout 缓冲；wait -n 任一 child 退出即返回并触发 trap——这是「一败俱败」监督者的经典写法。生产环境上三个进程会是三个 systemd 服务或三个 Kubernetes Deployment，Kafka 与 TimescaleDB 的端点经 env 注入（3.6.5 范畴）。

双幂等性质

管道在 wire 上是 at-least-once，在仓库上是 exactly-once。两根支柱：

• producer-side: enable.idempotence=True — Kafka 按 producer-id + 每分区序列号去重；在飞 batch 的网络重试不会产生重复 Kafka offset。
• consumer-side: PRIMARY KEY (symbol, ts) + ON CONFLICT (symbol, ts) DO NOTHING — 仓库拒绝同 (symbol, ts) 的第二次 INSERT；消费者在批中崩溃重启不会产生重复行。

验证方式：kill -9 杀掉 feed_handler.py 中途的 batch、重启它、等 ~5 秒，然后 psql -c "SELECT COUNT(*) - COUNT(DISTINCT (symbol, ts)) FROM ticks_raw;" 必然为 0。管道在 wire 上 at-least-once、在仓库上 exactly-once。

ZeroMQ 变体

相对 Kafka 形态的一段话 diff，三处改动均以 code-formatted 列出：

• producer.py: zmq.Context().socket(zmq.PUB).bind('tcp://*:5555') 取代 Producer({...}).produce(...)，加上 socket.send_string(json.dumps(tick)) 取代 producer.produce 那行。
• feed_handler.py 与 vwap_monitor.py: socket = ctx.socket(zmq.SUB); socket.connect('tcp://producer:5555'); socket.setsockopt(zmq.SUBSCRIBE, b'') 取代 Consumer({...}).subscribe([...])，加上 msg = socket.recv_string() 取代 msg = consumer.poll(timeout=1.0)。
• TimescaleDB 写入（含 ON CONFLICT (symbol, ts) DO NOTHING）、内存 VWAP deque、结构化日志行​​一概不变​​。

代价：​​用 Kafka 的耐久性换 ~10x 更低的扇出延迟与零 broker 运维负担​​。ZeroMQ 上的消费者宕机期间错过的消息就丢了；Kafka 上的消费者从 committed offset 接着追。

UDP 多播变体（草图）

两句话：把 producer.py 换成一个 feed handler，在 CFFEX CTP-MD 多播组上调用 IP_ADD_MEMBERSHIP 加入，解码 CTP-MD 二进制格式，把 MarketData 记录 normalise 成同样的 Tick 形状，要么再发布回 Kafka（ticker-plant 形态）、要么发布在 ZeroMQ PUB（cluster 内扇出）。feed handler 同时还要承担 L3 的 gap-fill 责任——序列号跟踪与重传——这就是为什么真正的直连 feed handler 是更多代码、不是更少。

Exercise

下一模块

Capstone 在开发环境跑在 run.sh 下；下一模块（3.6.5 Build, Deploy & Containers）教如何把 producer.py、feed_handler.py、vwap_monitor.py 各自打成容器，把这里用到的 docker-compose.yml 写成头等公民产物，再把同一形态部署到 Kubernetes。3.6.6（Observability & System Design）把这里发出的每一条结构化 JSON 日志当作入口：下一模块把它们聚合、把消费滞后和吞吐画出板、把告警规则建在上面。

纪律小结

• 按 symbol 做 key（L2 分区规则）。
• 处理完再 commit，永远不提前（L2 消费者规则）。
• 生产端幂等（L2）+ 消费端主键幂等（3.6.3 L4）。
• 每个队列都给个测过的上界（L1 背压规则）。
• 结构化 JSON 日志到 stdout（3.6.6 forward-pointer）。
• 按耐久性 + 延迟预算 选 broker 类（L3 决策规则）。
• 跨团队接线之前先定 wire format（L1 跨团队规则）。
• Schema 迁移文件是仓库形状的唯一来源（3.6.2 + 3.6.3 L4）。

本课教的 Python + Kafka + TimescaleDB ticker-plant 模式是国内中小 私募 中频策略的标准答案；头部 私募 在延迟预算更紧时会进一步落到 C++ 直连 多播 feed handler + kdb+ tickerplant（3.4.5 / 3.5.3 加上 3.6.3 L3 的 kdb+ 部分）。但 Python / Kafka / TimescaleDB 形态就是 2026 年大多数买方 feed handler 实际跑的样子——CFFEX 中金所 走 CTP 通道、上交所/深交所 行情 通过 vendor normalised TCP 推送、私募 自建 Kafka 集群接入这条链路、再以 T+1 节奏完成日终对账与回测重放。

阅读清单

• confluent-kafka-python 文档（英文权威，github.com/confluentinc/confluent-kafka-python）——Python Consumer / Producer API。
• psycopg 3 文档 关于 cursor.executemany 与 事务（psycopg.org/psycopg3/docs/）。
• TimescaleDB 中文 文档 关于 hypertable（docs.timescale.com 社区翻译片段）。
• pyzmq 文档（英文权威，pyzmq.readthedocs.io）——ZeroMQ 变体参考。
• 上交所 / 深交所 行情 协议 摘要（sse.com.cn / szse.cn）。
• CFFEX CTP-MD 协议 中文 文档（www.cffex.com.cn / www.sfit.com.cn）。
• 《Kafka 权威指南》中文版（Narkhede / Shapira / Palino，O'Reilly）——构建数据管道一章。
• 《数据密集型应用系统设计》中文版（Martin Kleppmann）——第 11 章 流处理。

速查卡

本课反复出现的展示形态——抄进笔记：

1. Fenced ```yaml 块——最简 docker-compose.yml，含 kafka + timescaledb 两个服务。
2. Fenced ```sql 块——migrations/001_create_ticks_raw.sql schema 迁移，含 (symbol, ts) 主键与 create_hypertable 调用。
3. Fenced ```python 块——producer.py、feed_handler.py、vwap_monitor.py 完整源码。
4. Fenced ```bash 块——run.sh，含 set -euo pipefail、trap ... EXIT INT TERM、wait -n、结构化 JSON log() 辅助函数。
5. Inline-code 列表——双幂等性质的两根支柱（producer-side: enable.idempotence=True、consumer-side: PRIMARY KEY (symbol, ts) + ON CONFLICT (symbol, ts) DO NOTHING）。
6. Inline-code 描述——ZeroMQ 变体的三处确切改动。
7. Exercise 与两条渐进 Hint——七交付物的构建-验证 + 双幂等性 crash-restart 测试。

国内 私募 部署落地补充

A-股 私募 量化 中频策略团队部署本课 capstone 时的典型形态：把 docker-compose.yml 放进 firm 内网 GitLab 的 infra/ 目录，让运维同事把 image tag 从 docker.io 镜像到 私募 自建 registry，再用 Ansible 把 Kafka + TimescaleDB 拉到三台内网物理机上（生产环境副本因子 3、min.insync.replicas=2）。producer.py 接收的不是合成数据，而是 vendor 通联 / 万得 / 恒生 通过 normalised TCP 推送来的 510300、600519、000001 等 沪深300 成分股的 tick；本课的 time.sleep(0.02) 在生产环境换成对 vendor socket 的阻塞 recv。feed_handler.py 的 INSERT INTO ticks_raw ... ON CONFLICT (symbol, ts) DO NOTHING 在生产里再加一层 staging 表 → 升级（promotion）模式，沿 3.6.3 L4 的 stage-validate-promote 模式，让 T+1 日终的对账可以重跑而不破坏 idempotency。vwap_monitor.py 的实时 VWAP 输出在生产里走结构化 JSON 到 Kafka 上一条 metrics.vwap.1m topic，下游再接 Prometheus push-gateway 与 Grafana 板。

CFFEX 中金所 的 CTP-MD 多播变体是 私募 在期货策略 stack 里常见的扩展点：用 CTP API 拿 IF / IH / IC / IM 的多播 tick、normalise 成同样的 Tick 形状、推到本课的 producer.py 入口；后续整条 Kafka → TimescaleDB → VWAP 链路完全不变。这就是 L1 词汇 + L2 Kafka + L3 多播 + 3.6.3 仓库 + 3.6.1 shell 监督者 + 3.6.2 git 纪律全部组合在一起的 ticker plant 的实际落地形态。

补充一段实际运维细节：A-股 量化 私募 的 capstone 部署链路通常是：开发同事在本机用 docker compose up -d 跑起单节点 Kafka 与 TimescaleDB，开发与单元测试在 firm GitLab 的 feature 分支上来回打磨；feature 分支通过代码审查（沿 3.6.2 的 PR 规则）合入 main 之后，CI 流水线把镜像构建并推到 firm 自建 registry；运维同事用 Ansible / GitOps 把同样的 docker-compose 部署到内网三台物理机上，副本因子升到 3，min.insync.replicas=2，retention 升到 90 天以满足合规留痕。监控团队把 feed_handler.py 和 vwap_monitor.py 输出的结构化 JSON 日志接到 firm 自建的 Loki + Grafana stack 上；告警规则关注消费者 LAG 与每分钟 VWAP 跳变幅度。从开发到生产的每一步都和本课的代码形态对齐，只是配置参数随环境扩缩——这正是「key by symbol、处理完再 commit、按持久性 + 延迟预算选 broker」这一套纪律在 A-股 私募 中频策略实际可复制的落地路径。

国内典型 vendor 与监管对照：ShanghaiStockExchange、ShenzhenStockExchange 与 CFFEX (ChinaFinancialFuturesExchange) 由 ChinaSecuritiesRegulatoryCommission 监管；行情中介 TongLianData (通联)、WindData (万得)、HsiHengSheng (恒生) 提供 normalised TCP 推送；broker 自营算法部门 GuoTaiJunan、HuaTai、ZhongXinJianTou 各自内嵌简化版 feed handler；本课所有 Python 代码在这些 vendor stack 上零修改可跑。

部署规模差异（按 私募 管理规模分层）：HeadFundCase 头部 私募（RMB 百亿以上）跑三机房 Kafka 集群 + RedundantConnect 异地双活，落地 ShangHaiDatacenter 与 ShenzhenDatacenter；MidSizedFundCase 中型 私募（RMB 数十亿）走单机房三节点 Kafka + 自建 Loki + Grafana 监控；SmallSizedFundCase 中小 私募（RMB 数亿）单机 docker-compose 起步、流量到一定阈值再扩。本课所教 PythonStack 在三档规模上代码无差异，只是部署形态从单机 to 三节点 to 异地双活逐步演进——这是本课 capstone 在 ChinaQuantIndustry 实际落地的扩展路径。

ChinaQuantTeams 常用扩展 stack 一览：BackTestEngine 沿 BackTraderLikeLib 自建，OrderManagementSystem 接 BrokerCounterParty FixGatewayChannel，PortfolioRiskService 实时计算 PositionExposure 与 SectorConcentration；这些上游/下游 service 与本课 capstone 共用一套 Kafka 底座，topic 命名空间分别落在 orders.cffex.*、positions.daily.*、risk.realtime.* 等独立 prefix 下。每个 service 都遵守本课纪律：按业务主键做 partition key、enable.idempotence=True、enable.auto.commit=False、处理完再 commit、监控 LAG、结构化 JSON 日志。这套规则在 GuoNeiPrivateFundIndustry 几乎是事实标准。