使用 ClickHouse 和 Polygon.io 构建实时市场数据应用

本文字数：8666；估计阅读时间：22 分钟

作者：Lionel Palacin

this.authMsg = JSON.stringify({
  action: "auth",
  params: process.env.POLYGON_API_KEY,
});     


this.ws = new WebSocket("wss://socket.polygon.io/stocks");


this.ws.on("open", () => {
  console.log("WebSocket connected");
  this.isConnected = true;
  this.reconnectAttempts = 0;
  this.lastMessageTime = Date.now();
  this.connectionStartTime = Date.now();
  this.statusMessage = "Connected - Authenticating...";
  this.logConnectionEvent("connected");
  this.ws.send(this.authMsg);
});


this.ws.on("message", (data) => {
  if (!this.isPaused) {
    this.handleMessage(data);
  }
});

CREATE TABLE quotes
(
    `sym` LowCardinality(String),
    `bx` UInt8,
    `bp` Float64,
    `bs` UInt64,
    `ax` UInt8,
    `ap` Float64,
    `as` UInt64,
    `c` UInt8,
    `i` Array(UInt8),
    `t` UInt64,
    `q` UInt64,
    `z` Enum8('NYSE' = 1, 'AMEX' = 2, 'Nasdaq' = 3),
    `inserted_at` UInt64 DEFAULT toUnixTimestamp64Milli(now64())
)
ORDER BY (sym, t - (t % 60000));


CREATE TABLE trades
(
    `sym` LowCardinality(String),
    `i` String,
    `x` UInt8,
    `p` Float64,
    `s` UInt64,
    `c` Array(UInt8),
    `t` UInt64,
    `q` UInt64,
    `z` Enum8('NYSE' = 1, 'AMEX' = 2, 'Nasdaq' = 3),
    `trfi` UInt64,
    `trft` UInt64,
    `inserted_at` UInt64 DEFAULT toUnixTimestamp64Milli(now64())
)
ORDER BY (sym, t - (t % 60000));

handleMessage(data) {
    try { 
        const trades = payload.filter((row) => row.ev === "T").map(({ ev, ...fields }) => fields);
        const quotes = payload.filter((row) => row.ev === "Q").map(({ ev, ...fields }) => fields);
        this.addToBatch(trades, "trades");
        this.addToBatch(quotes, "quotes");
} catch (error) {
        console.error("Error handling message:", error);
        console.error("Message data:", data.toString().substring(0, 200));
}


addToBatch(rows, type) {
    if (rows.length === 0) return;
    const batch = type === "trades" ? this.tradesBatch : this.quotesBatch;
    batch.push(...rows);
    if (batch.length >= this.maxBatchSize) {
        this.flushBatch(type);
    }
 }


flushBatch(type) {
    const batch = type === "trades" ? this.tradesBatch : this.quotesBatch;
    if (batch.length === 0) return;
    const dataToInsert = [...batch];
    if (type === "trades") {
         this.tradesBatch = [];
    } else {
         this.quotesBatch = [];
    }
    await this.client.insert({
      table: table,
      values: data,
      format: "JSONEachRow",
    });
 }

WITH
    {syms: Array(String)} as symbols,
    toDate(now('America/New_York')) AS curr_day,
    trades_info AS
    (
        SELECT
            sym,
            argMax(p, t) AS last_price,
            round(((last_price - argMinIf(p, t, fromUnixTimestamp64Milli(t, 'America/New_York') >= curr_day))  argMinIf(p, t, fromUnixTimestamp64Milli(t, 'America/New_York') >= curr_day)) * 100, 2) AS change_pct,
            sum(s) AS total_volume,
            max(t) AS latest_t
        FROM trades
        WHERE (toDate(fromUnixTimestamp64Milli(t, 'America/New_York')) = curr_day) AND (sym IN (symbols))
        GROUP BY sym
        ORDER BY sym ASC
    ),
    quotes_info AS
    (
        SELECT
            sym,
            argMax(bp, t) AS bid,
            argMax(ap, t) AS ask,
            max(t) AS latest_t
        FROM quotes
        WHERE (toDate(fromUnixTimestamp64Milli(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
FROM trades_info AS t
LEFT JOIN quotes_info AS q ON t.sym = q.sym;

SELECT
 toUnixTimestamp64Milli(toDateTime64(toStartOfInterval(fromUnixTimestamp64Milli(t), interval 2 minute), 3)) as x,
    argMin(p, t) as o,
    max(p) as h,
    min(p) as l,
    argMax(p, t) 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;

• 在客户端使用批处理，并根据系统的内存和延迟限制合理地选择插入批次的大小。
• 启用压缩功能：压缩后的插入数据会更小，能减少网络带宽消耗并加快传输速度。
• 监控 ClickHouse 中创建的数据分片（parts），防止出现过度合并的问题。虽然本文中重点讨论异步插入，但分片管理的思路同样适用于同步摄取。你还可以通过高级的可观测性（observability）仪表盘监控分片数量。
• Polygon.io 也提供了一些高并发数据消费的性能优化建议。

SELECT
    sym,
    count() AS trade_count,
    argMax(inserted_at, t) - argMax(t, t) AS ingest_latency
FROM trades
GROUP BY sym
ORDER BY trade_count DESC
LIMIT 100;

-- Create destination table
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, UInt64),
    `high_price_state` AggregateFunction(max, Float64),
    `low_price_state` AggregateFunction(min, Float64),
    `close_price_state` AggregateFunction(argMax, Float64, UInt64),
    `volume_state` AggregateFunction(sum, UInt64),
    `trade_count_state` AggregateFunction(count)
)
ENGINE = SummingMergeTree
ORDER BY (sym, minute_bucket_ms);

-- Create view
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, t) AS open_price_state,
    maxState(p) AS high_price_state,
    minState(p) AS low_price_state,
    argMaxState(p, t) AS close_price_state,
    sumState(s) AS volume_state,
    countState() AS trade_count_state
FROM trades
GROUP BY
    sym,
    z,
    minute_bucket_ms;

-- Query the table
SELECT
    sym,
    z,
    minute_bucket_ms,
    fromUnixTimestamp64Milli(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;