本文字数：6400；估计阅读时间：17分钟

作者：Lionel Palacin

编者按：本文译自 ClickHouse 原博客（链接点文末-查看原文）。 原文围绕「用 ClickHouse 承载高频 tick 数据的实时写入、存储与查询。」展开。文章从报价与成交的基本概念切入，逐步搭建数据链路，对实时分析场景的选型有参考价值。技术栈组合清晰，适合作为同类应用的落地范例。

实时 tick 数据应用是实时分析的典型场景。追踪 Web 应用中的用户行为、监控物联网设备的指标，都属于这一类：面对高频事件流，需要以低延迟完成写入、存储和查询。

放到金融市场，区别就在于紧迫性。哪怕只延迟几秒，一笔本来盈利的交易也可能变成亏损。每一笔成交、每一次报价更新都会产生一个新的 tick，多个标的加在一起，每秒可能多达数千个。

ClickHouse 很适合这类负载。它能处理高频写入、基于时间的查询和低延迟查询。内置压缩能降低存储开销，哪怕每个标的有数十亿行也不成问题。物化视图可以在数据写入时就完成预聚合或重新组织，从而优化查询性能，无需额外增加处理层。

下面一步步演示如何构建一个实时 tick 数据应用：用 Massive（原 Polygon.io）获取市场数据，用 ClickHouse 实时存储和查询 tick，后端用 Node.js，实时可视化用 React。下面正式开始。

什么是 tick？

开始之前，有必要先弄清楚报价（quote）和成交（trade）分别是什么。报价代表市场参与者当前愿意买入或卖出某只证券的价格，具体包含最优买价（best bid，即有人愿意支付的最高价）和最优卖价（best ask，即有人愿意接受的最低卖出价）。随着新订单进入或退出市场，报价会不断更新。

成交则是买卖双方之间真实发生的交易。当有人接受当前的卖价或买价、订单被撮合成交时，就产生了一笔成交。成交记录包含成交价、成交量和时间戳。

逐笔数据（tick data）通常来自两类数据流。一类是报价更新，另一类是成交记录。两者对理解市场行为都不可或缺，但在分析和策略开发中各有各的用处。

获取实时市场数据

清楚了要采集的数据类型，接下来看怎么获取。首先要找到并订阅一个股票市场 API。可选的服务商很多，本次演示选用的是 Massive。所选套餐必须明确包含通过 WebSocket 提供的股票 成交与报价。只有 REST API 密钥，或者只含聚合数据的套餐，都不够用。订阅前先确认当前的频道权限。

WebSockets 对行情数据的流式传输至关重要，因为它免去了轮询 REST API 的延迟与开销。每次请求数据都新建连接，还可能漏掉两次调用之间的逐笔更新；WebSocket 则维持一条长连接，数据一就绪就推送过来。对于毫秒必争的高频行情数据，这一点尤为关键。

用 Massive API 开始采集数据并不复杂：与 /stocks 端点建立连接，用你的 Massive API 密钥完成认证，然后开始处理消息。

下面这个独立的 Node.js 诊断脚本会先完成认证，认证成功后再订阅，并记录收到的事件。装好依赖、配置好.env之后，在示例目录下运行它。启动完整应用之前要先把它停掉；账户的连接数上限会导致多个客户端无法共用同一路数据。

require("dotenv").config(); const WebSocket = require("ws"); if (!process.env.MASSIVE_API_KEY) throw new Error("Set MASSIVE_API_KEY in .env"); const symbols = (process.env.MASSIVE_SYMBOLS || "AAPL,MSFT,NVDA") .split(",").map((symbol) => symbol.trim()).filter(Boolean); const ws = new WebSocket("wss://socket.massive.com/stocks"); ws.on("open", () => { ws.send(JSON.stringify({ action: "auth", params: process.env.MASSIVE_API_KEY })); }); ws.on("message", (data) => { const payload = JSON.parse(data.toString()); for (const row of payload) { if (row.ev === "status") { console.log(row.status, row.message); if (row.status === "auth_success") { ws.send(JSON.stringify({ action: "subscribe", params: symbols.flatMap((symbol) => [`T.${symbol}`, `Q.${symbol}`]).join(","), })); } else if (row.status === "auth_failed") { ws.close(); } } else if (row.ev === "T" || row.ev === "Q") { console.log(row); } } }); ws.on("error", (error) => console.error(error.message)); process.on("SIGINT", () => ws.close());

将数据写入 ClickHouse

在 ClickHouse 中为逐笔数据建模

Tick 数据的建模相对直接：它只有两类事件，一类是 trade，一类是 quote，每类事件的字段都不多，而且大多是数值型。下面是创建两张独立表的 DDL。t是 SIP 时间戳，单位为毫秒；q是每个标的的序列号。condition 和 indicator 数组用UInt32，因为其取值可能超过 255。trade 表结构还保留了可选的ds（小数数量，文本形式）和pt（参与者时间戳）字段。可选字段缺失时使用默认值。当前 quote 的 size 以股数表示。示例中价格用Float64，并对整数 trade 数量s求和；如果你的应用有需要，应改用合适的 decimal 类型，并对小数成交量单独处理。

CREATE TABLE IF NOT EXISTS quotes ( `sym` LowCardinality(String), `bx` UInt8, `bp` Float64, `bs` UInt64, `ax` UInt8, `ap` Float64, `as` UInt64, `c` UInt8, `i` Array(UInt32), `t` UInt64, `q` UInt64, `z` Enum8('NYSE' = 1, 'AMEX' = 2, 'Nasdaq' = 3), `inserted_at` UInt64 DEFAULT toUnixTimestamp64Milli(now64()) ) ENGINE = MergeTree ORDER BY (sym, t - (t % 60000)); CREATE TABLE IF NOT EXISTS trades ( `sym` LowCardinality(String), `i` String, `x` UInt8, `p` Float64, `s` UInt64, `ds` String DEFAULT '', `pt` UInt64 DEFAULT 0, `c` Array(UInt32), `t` UInt64, `q` UInt64, `z` Enum8('NYSE' = 1, 'AMEX' = 2, 'Nasdaq' = 3), `trfi` UInt64, `trft` UInt64, `inserted_at` UInt64 DEFAULT toUnixTimestamp64Milli(now64()) ) ENGINE = MergeTree ORDER BY (sym, t - (t % 60000));

数据量会增长得很快。

开发阶段先订阅少量标的，测好吞吐量之后，再订阅整个市场。选择合适的 order key 对性能至关重要。在本例中，行首先按 sym（股票代码）排序，把同一标的的所有事件归到一起。在每个标的分组内，行按t - (t % 60000)排序，从而形成 1 分钟的时间桶。这种做法在这里很合适，因为我们是按标的聚合数据来生成可视化的。分钟桶把时间相近的事件聚在一起，而查询仍然用原始时间戳和序列号来确定事件顺序。排序键和过滤条件要结合实际负载做基准测试；单靠时间桶并不能保证每个查询都能高效裁剪。

数据写入策略

为这类应用设计数据写入管道有多种方式，比如借助 Kafka 这样的消息队列。不过，为了尽量降低延迟，通常更好的做法是保持系统简单：只要条件允许，就直接把数据从 WebSocket 连接推入 ClickHouse。

这套方案就绪之后，下一步是选择合适的写入方式。ClickHouse 同时支持同步插入和异步插入。

对于同步写入，数据先在客户端攒批，再发送出去。批大小（batch size）要在内存占用、延迟和系统开销之间取得平衡。批次越大，插入请求越少，吞吐越高，但内存占用会增加，单条记录的延迟也会变大。批次越小，内存压力越小，但会产生过多小数据分片，反而给 ClickHouse 带来更多负载。

异步写入时，数据持续发往 ClickHouse，攒批由内部完成。到达的记录先写入内存缓冲区，再按可配置的阈值刷写到存储。当客户端不方便攒批时——比如数据来自大量小客户端——这种方式很合适。

在这个场景里，同步写入更合适。数据只由一个客户端从 WebSocket API 推送过来，批处理可以放在客户端做，从而更好地控制性能和资源占用。

下面这个精简的攒批类展示了写入路径。创建一个TickBatcher，把收到的 WebSocket 帧连同上面的认证处理程序一起传给它的handleMessage方法。在 awaitclose()之前先停止接收消息。完整示例另外加上了有界并发插入、重连处理、关闭超时和指标。两者都用同步插入，因此成功就意味着 ClickHouse 已确认写入。

const { createClient } = require("@clickhouse/client"); class TickBatcher { constructor() { this.client = createClient({ url: process.env.CLICKHOUSE_URL || "http://localhost:8123", username: process.env.CLICKHOUSE_USERNAME || "default", password: process.env.CLICKHOUSE_PASSWORD || "", database: process.env.CLICKHOUSE_DATABASE || "default", compression: { request: true, response: true }, clickhouse_settings: { async_insert: 0 }, }); this.batches = { trades: [], quotes: [] }; this.flushing = { trades: false, quotes: false }; this.failedRecords = 0; this.droppedRecords = 0; this.timer = setInterval(() => { void this.flushBatch("trades"); void this.flushBatch("quotes"); }, 2000); } handleMessage(data) { const payload = JSON.parse(data.toString()); for (const { ev, ...fields } of payload) { const table = ev === "T" ? "trades" : ev === "Q" ? "quotes" : null; if (!table) continue; if (this.batches[table].length >= 10000) { this.droppedRecords++; continue; } this.batches[table].push(fields); if (this.batches[table].length >= 1000) void this.flushBatch(table); } } async flushBatch(table) { if (this.flushing[table]) return; this.flushing[table] = true; try { while (this.batches[table].length) { const values = this.batches[table].splice(0, 1000); try { await this.client.insert({ table, values, format: "JSONEachRow" }); } catch (error) { this.failedRecords += values.length; console.error(`Insert failed for ${table}:`, error.message); } } } finally { this.flushing[table] = false; } } async close() { clearInterval(this.timer); while (Object.values(this.flushing).some(Boolean)) { await new Promise((resolve) => setTimeout(resolve, 10)); } await Promise.all([this.flushBatch("trades"), this.flushBatch("quotes")]); await this.client.close(); } }

这个精简类把每张待写入表的缓冲区限制在 10,000 条记录，并统计被丢弃或失败的记录；失败的写入不会重试。要做到不丢数据，还需要持久化缓冲、重放和去重。完整示例在 1,000 行或每两秒触发一次刷写，最多同时执行四个插入。

可视化实时市场数据

数据存进 ClickHouse 之后，搭建可视化层就很直接了。真正的难点在于写出正确的 SQL 查询。下面看看具体怎么做。我们聚焦支撑两个关键可视化所需的查询。第一个是实时表格，持续更新，展示某只股票的最新交易数据。

要做这个可视化，一个查询就够了，数据可以用 ClickHouse 强大的 SQL 查询语言和自定义函数来格式化。

with {syms: Array(String)} as symbols, toDate(now('America/New_York')) as curr_day, trades_info as ( select sym, argMax(p, tuple(t, q)) as last_price, round(((last_price - (argMin(p, tuple(t, q)))) / nullIf(argMin(p, tuple(t, q)), 0)) * 100, 2) as change_pct, sum(s) as total_volume, max(t) as latest_t from trades where toDate(fromUnixTimestamp64Milli(toInt64(t), 'America/New_York')) = curr_day and sym in symbols group by sym order by sym asc ), quotes_info as ( select sym, argMax(bp, tuple(t, q)) as bid, argMax(ap, tuple(t, q)) as ask, max(t) as latest_t from quotes where toDate(fromUnixTimestamp64Milli(toInt64(t), 'America/New_York')) = curr_day and sym in symbols group by sym order by sym asc ) select t.sym as ticker, t.last_price as last, q.bid as bid, q.ask as ask, t.change_pct as change, t.total_volume as volume, toUnixTimestamp64Milli(now64()) - toInt64(greatest(t.latest_t, q.latest_t)) as latency from trades_info as t left join quotes_info as q on t.sym = q.sym;

下面拆解一下这个查询做了什么。

首先定义两个变量：symbols保存待分析的股票代码列表，curr_day记录纽约时区的当前日期。

查询随后获取交易数据，包括：

• last_price ：最新成交价，使用 argMax(p, tuple(t, q)) 在毫秒时间戳相同时按序列号打破平局

• change_pct ：相对于当前纽约日历日第一笔已摄取交易的百分比变化，带有零分母保护。这不是官方开盘价或前收盘价。

• total_volume ：该日历日已摄取的总整数股成交量

它还获取报价数据：

• bid ：最新买价，使用 argMax(bp, tuple(t, q))

• ask ：最新卖价，使用 argMax(ap, tuple(t, q))

最后，将交易和报价结果合并。额外的latency列报告最新收到事件距今的毫秒数，而不是只测量数据库插入时间。下列行是示例历史值。该演示不筛选交易条件，不处理取消/更正，也不重建官方交易所的 OHLCV 序列。

我们要分析的第二个可视化是蜡烛图，展示某只股票的价格走势和成交量。

我们来看看驱动这一可视化的 SQL 查询。

select toUnixTimestamp64Milli(toDateTime64(toStartOfInterval(fromUnixTimestamp64Milli(toInt64(t)), interval 2 minute), 3)) as x, argMin(p, tuple(t, q)) as o, max(p) as h, min(p) as l, argMax(p, tuple(t, q)) as c, sum(s) as v from trades where x > toUnixTimestamp64Milli(now64() - interval 1 hour) and sym = {sym: String} group by x order by x asc;

该查询将最近一小时按两分钟分桶，并计算开盘价、最高价、最低价、收盘价和成交量。开盘价和收盘价使用(t, q)来处理时间戳相同的交易。桶过滤器排除了窗口起始处部分重叠的桶。

浏览器调用一个命名查询的 Express 端点；只接受服务器定义、参数化的 SQL。ClickHouse 凭据保留在服务器上，绝不放入NEXT_PUBLIC_变量。为了可视化查询结果，我们使用 click-ui 组件显示表格，使用 Chart.js 绘制蜡烛图。

扩展与实用技巧

在生产环境中处理高频市场数据，不仅需要快速数据库。以下技巧和方法有助于确保系统在数据量增长时保持高性能和可靠。

扩展数据摄取

处理多个交易标的的 tick 级数据时，持续吞吐量很容易超过每秒数万条记录。

要应对这种情况，可以从以下几个方面入手：

• 使用 客户端批量写入 ，批量大小按系统的内存和延迟限制调到最优。

• 使用 压缩 ：压缩插入数据能缩小网络传输的载荷体积，降低带宽占用，加快传输速度。

• 监控 ClickHouse 中创建的 part 数量，避免合并过于频繁。这篇 博客 讲的是异步插入，但其中关于 part 创建的部分同样适用于同步写入。你也可以借助 高级仪表盘 来 监控 数据 part 的数量。

• Massive 也提供了处理高吞吐数据消费的 性能建议 。

监控写入延迟

前面提到，金融应用对数据的实时性要求很高，所以监控延迟很有必要。

你可以很方便地计算并跟踪事件时间戳（tick 发生的时刻）与写入时间戳（数据被存储的时刻）之间的差值。这个「事件到写入」的延迟涵盖行情源传输、客户端批量处理、网络耗时和数据库写入，并非单纯的数据库延迟。下面的查询统计最近一小时内每个交易标的最新成交的该延迟。使用有符号运算，可以避免时间戳出现意外差异时发生无符号下溢。

SELECT sym, count() AS trade_count, argMax(toInt64(inserted_at) - toInt64(t), tuple(t, q)) AS ingest_latency_ms FROM trades WHERE t >= toUnixTimestamp64Milli(now64() - INTERVAL 1 HOUR) GROUP BY sym ORDER BY trade_count DESC LIMIT 100;

把这个指标可视化到仪表盘上，有助于及早发现变慢的迹象。

善用物化视图

Materialized views 适合在数据到达时就预先聚合。对于依赖按时间汇总的特定查询模式，这能带来明显的优化。典型例子是按固定间隔（比如 1 分钟窗口）计算金融数据的 OHLCV（开盘价、最高价、最低价、收盘价、成交量）指标。在写入时生成这些聚合结果，查询就能快速返回，不必每次重新计算。

先创建一张目标表，用来存放 1 分钟的 OHLCV 聚合结果。用AggregatingMergeTree合并聚合状态。排序键要包含所有分组维度：这里是 symbol、tape 和 minute。开盘价和收盘价状态沿用与原始查询相同的(t, q)排序。

CREATE TABLE trades_1min_ohlcv ( `sym` LowCardinality(String), `z` Enum8('NYSE' = 1, 'AMEX' = 2, 'Nasdaq' = 3), `minute_bucket_ms` UInt64, `open_price_state` AggregateFunction(argMin, Float64, Tuple(UInt64, UInt64)), `high_price_state` AggregateFunction(max, Float64), `low_price_state` AggregateFunction(min, Float64), `close_price_state` AggregateFunction(argMax, Float64, Tuple(UInt64, UInt64)), `volume_state` AggregateFunction(sum, UInt64), `trade_count_state` AggregateFunction(count) ) ENGINE = AggregatingMergeTree ORDER BY (sym, z, minute_bucket_ms);

下一步是创建物化视图。

CREATE MATERIALIZED VIEW trades_1min_ohlcv_mv TO trades_1min_ohlcv AS SELECT sym, z, intDiv(t, 60000) * 60000 AS minute_bucket_ms, argMinState(p, tuple(t, q)) AS open_price_state, maxState(p) AS high_price_state, minState(p) AS low_price_state, argMaxState(p, tuple(t, q)) AS close_price_state, sumState(s) AS volume_state, countState() AS trade_count_state FROM trades GROUP BY sym, z, minute_bucket_ms;

目标表和视图都要在开始写入前建好。视图只处理新插入的数据块，不会自动回填已有数据行。如果需要，可以用一条安排好时机的INSERT ... SELECT回填已有数据，注意别把重叠的实时数据重复计入。查询里用了-Merge系列函数，即使后台合并尚未完成也能合并状态。

要查看数据，执行下面这个查询。

-- Query the table SELECT sym, z, minute_bucket_ms, fromUnixTimestamp64Milli(toInt64(minute_bucket_ms)) as minute_timestamp, argMinMerge(open_price_state) AS open_price, maxMerge(high_price_state) AS high_price, minMerge(low_price_state) AS low_price, argMaxMerge(close_price_state) AS close_price, sumMerge(volume_state) AS volume, countMerge(trade_count_state) AS trade_count FROM trades_1min_ohlcv GROUP BY sym, z, minute_bucket_ms ORDER BY sym, z, minute_bucket_ms;

运行完整示例

示例源码 现在位于blog-examples/stock-data-demo。需要 Node.js 22 或更高版本，以及一个已有的本地或 Cloud ClickHouse 数据库。

git clone https://github.com/ClickHouse/examples.git cd examples/blog-examples/stock-data-demo npm ci cp env.example .env # Set CLICKHOUSE_URL, CLICKHOUSE_USERNAME, CLICKHOUSE_PASSWORD, # CLICKHOUSE_DATABASE, and MASSIVE_API_KEY in .env. npm run setup npm run build npm start

打开http://localhost:34567/stocks/查看图表，或打开http://localhost:34567/admin控制数据写入。默认订阅 AAPL、MSFT 和 NVDA。想换其他代码，修改MASSIVE_SYMBOLS即可；仪表盘上的自选列表不会改变上游订阅。服务只绑定回环地址，API 也没有鉴权，公开暴露之前请先加上访问控制。

如果你的密钥缺少所需的数据通道，管理页面会提示认证失败。周末和节假日市场也可能冷清或休市。想在没有行情权限的情况下试用仪表盘，可以另用一个 ClickHouse 数据库，把 API key 留空，并在npm start之前先运行npm run seed。这会插入明确标注为合成的价格数据，时间戳落在最近三分钟内，不会获取真实行情。当这些时间戳滑出图表窗口后，重新运行 seed 即可。

README 涵盖了本地 ClickHouse 的搭建、开发模式、配置和验证方法。

结论

这篇文章里，我们用 Massive 提供市场数据、用 ClickHouse 做高速写入和查询，搭建了一个实时逐笔数据（tick data）应用。内容涉及逐笔数据的流式传输与结构设计、写入性能的管理，以及高效查询和可视化的实现。

在这个 GitHub 仓库 里，有一个可运行的示例，可视化层用 React 实现。你可以借助它了解写入和查询的常见模式，再针对生产应用补上可靠投递、回放、身份认证，以及按具体负载做的调优。

关于我们

ClickHouse（clickhouse.com) 是面向 AI 时代打造的高性能实时分析数据库，能够以极致性能处理海量数据分析任务。凭借高并发、低延迟和云原生架构，ClickHouse 广泛应用于可观测性、数据仓库、实时分析及 AI 数据基础设施等场景。我们致力于帮助企业在公有云平台上构建安全、弹性且高性价比的实时分析与 AI 数据平台，加速释放数据价值，推动智能化创新与数字化转型。目前，Trip.com、DiDi、Meta、Sony、Netflix、Deutsche Bank、Sierra、Cloudflare 等全球领先企业均在使用 ClickHouse 支撑其关键业务和数据分析平台。