RedisTimeSeries执行CREATE RULE聚合时如何填充缺失数据？

RedisTimeSeries 的 CREATE RULE 本身不会自动为无原始数据的聚合区间生成空值并填充，但可以通过以下两种方案实现你需要的类似 pandas fillna 的逻辑，适配股票 OHLC 数据的补全需求。

方案一：On-Demand 聚合 + 客户端填充 #

这种方式利用 RedisTimeSeries 的 On-Demand 聚合规则（无需实时生成聚合数据），查询后在客户端用 pandas 补全缺失区间，和你之前的处理逻辑最接近。

步骤与代码示例 #

1. 创建原始股票 Tick 时间序列（存储成交价 Close）：

from redis import Redis
from redistimeseries.client import Client

r = Redis(host='localhost', port=6379)
ts = Client(r)

# 创建原始Tick序列（以股票代码SYMBOL为例）
ts.create('SYMBOL:Close', labels={'type': 'tick', 'symbol': 'SYMBOL'})

2. 创建 On-Demand 聚合规则（生成指定时间维度的 OHLC 数据）：

# 创建1分钟维度的OHLC聚合规则，On-Demand模式无需指定DESTINATION
ts.create_rule(
    source='SYMBOL:Close',
    aggregation_type='ohlc',
    bucket_size_msec=60000,  # 1分钟=60*1000毫秒
    on_demand=True
)

3. 查询聚合数据并在客户端填充缺失值：

import pandas as pd

# 查询指定时间范围的1分钟OHLC数据
start_time = 1690000000000  # 起始时间（毫秒）
end_time = 1690003600000    # 结束时间（毫秒）
data = ts.range('SYMBOL:Close', start_time, end_time, aggregation_type='ohlc', bucket_size_msec=60000)

# 转换为DataFrame
df = pd.DataFrame(data, columns=['timestamp', 'Open', 'High', 'Low', 'Close'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('timestamp', inplace=True)

# 按1分钟频率重采样，填充缺失值（和你之前的逻辑一致）
resampled_df = df.resample('1T').asfreq()
resampled_df['Close'] = resampled_df['Close'].ffill()
resampled_df['Open'] = resampled_df['Open'].fillna(resampled_df['Close'])
resampled_df['High'] = resampled_df['High'].fillna(resampled_df['Close'])
resampled_df['Low'] = resampled_df['Low'].fillna(resampled_df['Close'])

# 处理后的数据就是补全空值的OHLC
print(resampled_df)

方案二：主动补全缺失聚合区间 #

如果需要在 Redis 端直接存储补全后的聚合数据，可以通过定时任务主动检查并填充缺失的时间区间。

步骤与代码示例 #

1. 创建实时聚合规则（生成到目标序列）：

# 创建目标聚合序列（1分钟OHLC）
ts.create('SYMBOL:Close:1min:OHLC', labels={'type': 'aggregated', 'symbol': 'SYMBOL', 'interval': '1min'})

# 创建实时聚合规则
ts.create_rule(
    source='SYMBOL:Close',
    destination='SYMBOL:Close:1min:OHLC',
    aggregation_type='ohlc',
    bucket_size_msec=60000
)

2. 定时补全缺失区间的脚本（Python示例，可配合APScheduler等定时工具）：

import time
from datetime import datetime, timedelta

def fill_missing_ohlc():
    # 获取聚合序列的最后一条数据
    last_data = ts.get('SYMBOL:Close:1min:OHLC')
    if not last_data:
        return  # 无数据时跳过
    
    last_ts, last_open, last_high, last_low, last_close = last_data
    last_dt = datetime.fromtimestamp(last_ts / 1000)
    
    # 计算当前时间对应的1分钟区间的结束时间
    now = datetime.now()
    current_interval_end = now.replace(second=0, microsecond=0) + timedelta(minutes=1)
    current_ts = int(current_interval_end.timestamp() * 1000)
    
    # 遍历缺失的每个1分钟区间，填充前值
    current_check_dt = last_dt.replace(second=0, microsecond=0) + timedelta(minutes=1)
    while current_check_dt < current_interval_end:
        check_ts = int(current_check_dt.timestamp() * 1000)
        # 检查该区间是否已有数据
        exists = ts.range('SYMBOL:Close:1min:OHLC', check_ts, check_ts)
        if not exists:
            # 插入填充后的数据（Open/High/Low/Close都用最后一次的Close值）
            ts.add('SYMBOL:Close:1min:OHLC', check_ts, [last_close, last_close, last_close, last_close])
        
        current_check_dt += timedelta(minutes=1)

# 每30秒执行一次补全（可根据需求调整频率）
while True:
    fill_missing_ohlc()
    time.sleep(30)

注意事项 #

• 方案一适合查询时才需要补全的场景，不会增加Redis存储压力；
• 方案二适合需要持久化补全后数据的场景，但会增加Redis的写入操作，需根据数据量调整定时频率；
• 填充逻辑可根据需求修改，比如用0填充或其他策略，这里沿用了你之前的ffill和同值填充逻辑。

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