02 数据建模

在 ClickHouse Cloud 中创建带类型的市场表、tick 表、交易表和一分钟聚合表。

起点

.env.polymarket 已 source，并且你已理解 condition ID 与 token ID 的区别。

为什么是这些表

后面的五个查询都会跨所有被监控的市场读取近期时间窗口。因此事件表的键从一个小时/时间键开始，随后是用于分组的 token 或 condition。已知字段使用原生类型： token ID 用 UInt256，事件时间用 DateTime64，价格和数量用精确 decimal，取值有限的事件字段用枚举。不透明的源数据 payload 保持为字符串，因为没有查询会读取它内部的字段。

这里没有 PARTITION BY。这项短期实训没有确定的数据保留边界；在出现生命周期需求之前就加分区只会产生大量小 part，而不带来任何收益。

第 1 步：创建数据库和原始表

把整段内容复制到 ClickHouse Cloud SQL 控制台并运行：

CREATE DATABASE IF NOT EXISTS polymarket;

CREATE TABLE IF NOT EXISTS polymarket.markets
(
    market_id UInt64,
    condition_id FixedString(66),
    token_id UInt256,
    outcome LowCardinality(String),
    question String,
    slug String,
    active Bool,
    accepting_orders Bool,
    volume_24h Decimal128(8),
    observed_at DateTime64(3, 'UTC')
)
ENGINE = ReplacingMergeTree(observed_at)
ORDER BY (condition_id, token_id);

CREATE TABLE IF NOT EXISTS polymarket.price_ticks
(
    event_id FixedString(64),
    condition_id FixedString(66),
    token_id UInt256,
    event_at DateTime64(3, 'UTC'),
    observed_at DateTime64(3, 'UTC'),
    event_kind Enum8(
        'book_snapshot' = 1,
        'price_change' = 2,
        'last_trade_price' = 3,
        'best_bid_ask' = 4,
        'rest_book' = 5
    ),
    source Enum8('WEBSOCKET' = 1, 'CLOB_REST' = 2, 'FIXTURE' = 3),
    price Decimal64(12),
    size Decimal128(8),
    side Enum8('UNKNOWN' = 0, 'BUY' = 1, 'SELL' = 2),
    best_bid Decimal64(12),
    best_ask Decimal64(12),
    midpoint Decimal64(12),
    source_hash String,
    raw_payload String
)
ENGINE = MergeTree
ORDER BY (toStartOfHour(event_at), token_id, event_at, event_id);

CREATE TABLE IF NOT EXISTS polymarket.trades
(
    trade_id FixedString(64),
    condition_id FixedString(66),
    token_id UInt256,
    event_at DateTime64(3, 'UTC'),
    observed_at DateTime64(3, 'UTC'),
    proxy_wallet FixedString(42),
    side Enum8('UNKNOWN' = 0, 'BUY' = 1, 'SELL' = 2),
    price Decimal64(12),
    size Decimal128(8),
    outcome LowCardinality(String),
    transaction_hash FixedString(66),
    title String
)
ENGINE = ReplacingMergeTree(observed_at)
ORDER BY (toStartOfHour(event_at), condition_id, event_at, trade_id);

CREATE OR REPLACE VIEW polymarket.trades_clean AS
SELECT *
FROM polymarket.trades FINAL;

采集器在插入之前就会阻止重复数据。ReplacingMergeTree 是第二道安全网。trades_clean 视图让这项小型实训中的查询在 merge 尚未完成时也能得到确定的结果。

第 2 步：创建一分钟中间价聚合

CREATE TABLE IF NOT EXISTS polymarket.market_midpoints_1m
(
    token_id UInt256,
    minute DateTime('UTC'),
    open AggregateFunction(argMin, Decimal64(12), Tuple(DateTime64(3, 'UTC'), FixedString(64))),
    high AggregateFunction(max, Decimal64(12)),
    low AggregateFunction(min, Decimal64(12)),
    close AggregateFunction(argMax, Decimal64(12), Tuple(DateTime64(3, 'UTC'), FixedString(64))),
    updates AggregateFunction(count)
)
ENGINE = AggregatingMergeTree
ORDER BY (minute, token_id);

CREATE MATERIALIZED VIEW IF NOT EXISTS polymarket.market_midpoints_1m_mv
TO polymarket.market_midpoints_1m
AS
SELECT
    token_id,
    toStartOfMinute(event_at) AS minute,
    argMinState(midpoint, tuple(event_at, event_id)) AS open,
    maxState(midpoint) AS high,
    minState(midpoint) AS low,
    argMaxState(midpoint, tuple(event_at, event_id)) AS close,
    countState() AS updates
FROM polymarket.price_ticks
WHERE midpoint > 0
  AND event_kind IN ('book_snapshot', 'price_change', 'best_bid_ask', 'rest_book')
GROUP BY token_id, minute;

这个 materialized view 只聚合报价中间价。它有意排除了变动的订单级价格和最后成交价，这样 OHLC 序列的含义才是唯一的。

第 3 步：验证每个对象

clickhouse client \
  --host "$CLICKHOUSE_HOST" \
  --port "$CLICKHOUSE_PORT" \
  --user "$CLICKHOUSE_USER" \
  --password "$CLICKHOUSE_PASSWORD" \
  --secure \
  --query "SHOW TABLES FROM polymarket"

预期输出的名称中应包含：

market_midpoints_1m
market_midpoints_1m_mv
markets
price_ticks
trades
trades_clean

完成标准

在没有运行任何本地 ClickHouse 服务器的情况下，SHOW TABLES 返回全部六个对象。

下一步：启动实时采集器。

起点

为什么是这些表

第 1 步：创建数据库和原始表

第 2 步：创建一分钟中间价聚合

第 3 步：验证每个对象

完成标准

本页内容

Track your progress?