基于TimescaleDB实现股票数据5秒粒度1分钟滚动窗口平均咨询

核心问题梳理 #

需要实现每5秒生成一个覆盖过去1分钟窗口的平均成交量，同时支持回溯大量历史数据，此前尝试触发器和stats_agg时遇到语法问题，纠结直接操作原始tick数据的性能。

一、性能结论：绝对不要直接操作tick_data #

原始tick数据量极大（每秒可能上百条），直接对其做1分钟滚动窗口计算会扫描海量历史数据，性能完全无法接受。必须基于已有的5秒连续聚合视图agg_5s来做计算——agg_5s已将数据降采样为每5秒一条，数据量仅为原始的1/720，查询效率会提升几个数量级。

二、滚动窗口平均成交量的实现 #

方式1：直接查询（适合临时分析） #

基于agg_5s使用窗口函数，直接计算每5秒对应的过去1分钟平均成交量：

SELECT
    bucket_timestamp,
    symbol,
    session,
    -- 计算过去1分钟窗口的平均成交量
    avg(total_volume) OVER (
        PARTITION BY symbol, session
        ORDER BY bucket_timestamp
        RANGE BETWEEN INTERVAL '1 minute' PRECEDING AND CURRENT ROW
    ) AS avg_volume_1min,
    -- 可扩展计算其他滚动统计量，比如平均成交额
    avg(total_dollars) OVER (
        PARTITION BY symbol, session
        ORDER BY bucket_timestamp
        RANGE BETWEEN INTERVAL '1 minute' PRECEDING AND CURRENT ROW
    ) AS avg_dollars_1min
FROM agg_5s
ORDER BY symbol, session, bucket_timestamp;

语法说明：

• PARTITION BY symbol, session：保证每个股票、每个交易会话单独计算滚动窗口
• ORDER BY bucket_timestamp：按时间顺序处理数据
• RANGE BETWEEN INTERVAL '1 minute' PRECEDING AND CURRENT ROW：定义窗口范围为当前5秒bucket往前推1分钟（包含当前bucket）

方式2：创建连续聚合视图（适合长期自动更新） #

如果需要滚动统计结果自动更新并持久化，可基于agg_5s创建新的连续聚合视图：

CREATE MATERIALIZED VIEW IF NOT EXISTS agg_5s_1min_rolling
WITH (timescaledb.continuous) AS
SELECT
    bucket_timestamp,
    symbol,
    session,
    avg(total_volume) OVER rolling_window AS avg_volume_1min,
    avg(total_dollars) OVER rolling_window AS avg_dollars_1min,
    -- 可添加其他需要的滚动统计量
    max(high_price) OVER rolling_window AS rolling_high_1min,
    min(low_price) OVER rolling_window AS rolling_low_1min
FROM agg_5s
WINDOW rolling_window AS (
    PARTITION BY symbol, session
    ORDER BY bucket_timestamp
    RANGE BETWEEN INTERVAL '1 minute' PRECEDING AND CURRENT ROW
)
WITH DATA;

这个视图会自动跟随agg_5s更新，无需手动维护。

三、stats_agg的正确用法 #

此前的错误在于：在同一聚合语句中，试图用聚合后的字段（total_volume）作为stats_agg的输入，这是SQL语法不允许的。stats_agg需要直接接收原始数据列，或在已聚合的视图之上使用。

1. 在5秒连续聚合中预计算统计量 #

如果需要预计算每个5秒bucket内成交量、成交额的统计信息（均值、方差等），可在agg_5s中添加stats_agg：

CREATE MATERIALIZED VIEW IF NOT EXISTS agg_5s_with_stats
WITH (timescaledb.continuous) AS
SELECT
    time_bucket('5 seconds', timestamp) AS bucket_timestamp,
    symbol,
    session,
    -- 基础OHLC聚合
    first(price, timestamp) AS open_price,
    last(price, timestamp) AS close_price,
    max(price) AS high_price,
    min(price) AS low_price,
    -- 总量聚合
    sum(size) AS total_volume,
    count(*) AS total_ticks,
    sum(dollars) AS total_dollars,
    -- 基于tick_type的聚合
    sum(CASE WHEN tick_type = 'up' THEN size ELSE 0 END) AS up_volume,
    count(CASE WHEN tick_type = 'up' THEN 1 END) AS up_ticks,
    sum(CASE WHEN tick_type = 'up' THEN dollars ELSE 0 END) AS up_dollars,
    sum(CASE WHEN tick_type = 'down' THEN size ELSE 0 END) AS down_volume,
    count(CASE WHEN tick_type = 'down' THEN 1 END) AS down_ticks,
    sum(CASE WHEN tick_type = 'down' THEN dollars ELSE 0 END) AS down_dollars,
    sum(CASE WHEN tick_type = 'side' THEN size ELSE 0 END) AS side_volume,
    count(CASE WHEN tick_type = 'side' THEN 1 END) AS side_ticks,
    sum(CASE WHEN tick_type = 'side' THEN dollars ELSE 0 END) AS side_dollars,
    -- stats_agg：直接使用原始列计算统计量
    stats_agg(size) AS volume_stats,
    stats_agg(dollars) AS dollars_stats
FROM tick_data
GROUP BY bucket_timestamp, symbol, session;

2. 提取stats_agg的统计结果 #

查询时用专门的函数提取预计算的统计值：

SELECT
    bucket_timestamp,
    symbol,
    session,
    -- 提取成交量的均值、方差、标准差
    mean(volume_stats) AS avg_volume_per_tick,
    variance(volume_stats) AS volume_variance,
    stddev(volume_stats) AS volume_stddev,
    -- 提取成交额的统计值
    mean(dollars_stats) AS avg_dollars_per_tick
FROM agg_5s_with_stats;

3. 对多字段使用stats_agg #

每个需要统计的字段单独调用stats_agg即可，比如同时对size、dollars、price做统计，就分别写stats_agg(size)、stats_agg(dollars)、stats_agg(price)，然后用对应的提取函数获取结果。

四、总结 #

1. 优先基于agg_5s做滚动窗口计算，避免直接操作原始tick数据
2. 用窗口函数OVER()实现滚动窗口逻辑，语法简洁且性能优异
3. stats_agg用于预计算单bucket内的统计量，需直接传入原始列，通过提取函数获取结果

内容的提问来源于stack exchange，提问作者Enigma