如何使用Redis TimeSeries模块聚合时间戳生成分钟级K线数据

核心结论 #

目前Redis TimeSeries原生聚合规则不支持直接通过createrule预生成时间戳类的聚合结果，但有比额外存储时间戳时序更高效的实现方案，无需存储冗余数据。

方案1：利用bucket对齐特性直接推导（最高效，推荐） #

你当前使用的是固定60000ms的分钟级对齐聚合，Redis TimeSeries的bucket默认按时间戳整除bucket大小的规则对齐，因此：

• 每个聚合结果返回的时间戳就是当前分钟bucket的起始时间，可直接作为openTimestamp
• 该起始时间 + 59999 就是当前分钟bucket的结束时间，可直接作为closeTimestamp

该方案无任何额外存储、计算开销，也是金融K线领域的通用实现方式。

代码示例 #

from redistimeseries.client import Client
from datetime import datetime
from random import randint
rts = Client()

# 原有的聚合规则完全不需要修改
rts.create('price')
rts.create('openPrice')
rts.create('closePrice')
rts.create('lowPrice')
rts.create('highPrice')
rts.createrule('price', "openPrice", 'first', bucket_size_msec=60000)
rts.createrule('price', 'closePrice', 'last', bucket_size_msec=60000)
rts.createrule('price', 'lowPrice', 'min', bucket_size_msec=60000)
rts.createrule('price', 'highPrice', 'max', bucket_size_msec=60000)

# 插入测试数据
now = datetime.utcnow()
now_int = 1000 * int(now.timestamp())
for i in range(5000):
    r_n = randint(1, 1000000)
    new_time = now_int + i*1000
    rts.add('price', new_time, r_n)

# 查询拼接K线示例
end_time = int(datetime.utcnow().timestamp() * 1000)
start_time = end_time - 3600 * 1000 # 查询最近1小时数据

# 批量查询四个预聚合指标
open_list = rts.range('openPrice', start_time, end_time)
close_list = rts.range('closePrice', start_time, end_time)
low_list = rts.range('lowPrice', start_time, end_time)
high_list = rts.range('highPrice', start_time, end_time)

kline_result = []
for idx in range(len(open_list)):
    bucket_ts, open_p = open_list[idx]
    _, close_p = close_list[idx]
    _, low_p = low_list[idx]
    _, high_p = high_list[idx]
    kline_result.append({
        "openTimestamp": bucket_ts,
        "closeTimestamp": bucket_ts + 59999,
        "openPrice": open_p,
        "closePrice": close_p,
        "highPrice": high_p,
        "lowPrice": low_p
    })

方案2：按需查询实际交易时间（适合需要严格对应第一/最后一笔交易时间的场景） #

如果你的业务要求openTimestamp必须是该分钟内第一笔交易的实际时间，closeTimestamp必须是最后一笔交易的实际时间，也无需额外存储冗余时序，可在查询时基于原始price序列快速获取：

• 获取某分钟第一笔时间：对原始序列做正向范围查询，限制返回1条即可
• 获取某分钟最后一笔时间：对原始序列做反向范围查询，限制返回1条即可

代码示例 #

# 以指定分钟bucket为例
bucket_start = 1690000000000 # 某分钟起始时间戳
bucket_end = bucket_start + 60000

# 获取实际开盘时间
first_trade = rts.range('price', bucket_start, bucket_end, count=1)
actual_open_ts = first_trade[0][0] if first_trade else None

# 获取实际收盘时间
last_trade = rts.revrange('price', bucket_start, bucket_end, count=1)
actual_close_ts = last_trade[0][0] if last_trade else None

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