PostgreSQL补全缺失时间并插值计算能耗避免数据突刺

问题背景 #

不同工厂的电表每30分钟/小时上报读数至PostgreSQL数据库，通过相邻读数差值计算能耗。但电表存在离线情况（如周末），导致数据出现数小时甚至数天的时间间隙，传入Grafana展示时会出现数据突刺，需要生成连续时间序列并通过线性插值补全缺失值。

现有电表数据示例 #

SELECT serial_no, facility, reading, event_timestamp 
FROM public.meter_messages 
WHERE logical_address = '10DFFC0D37AD:1' 
ORDER BY event_timestamp;

查询结果：

当前能耗计算的问题 #

现有能耗计算SQL：

SELECT 
    date_trunc('hour', event_timestamp) AS event_timestamp_local,
    CASE 
        WHEN reading - lag(reading) OVER (ORDER BY event_timestamp ASC) > 0 
        THEN reading - lag(reading) OVER (ORDER BY event_timestamp ASC)
        ELSE 0 
    END AS consumption 
FROM public.meter_messages
WHERE serial_no = 'ABC/230121840267';

因数据间隙，结果会出现跳变：

需求说明 #

需要生成所选时间范围内连续的小时/30分钟级时间序列，通过线性插值补全缺失时段的读数，再计算对应能耗。示例规则：

• 原始缺失数据：14:00读数290，17:00读数410
• 插值计算：15:00读数 = 290 + (410-290)/3，16:00读数 = 320 + (410-290)/3
• 能耗为当前读数与前一读数的差值

补全后预期结果：

PostgreSQL解决方案 #

完整SQL查询（小时粒度） #

WITH time_series AS (
    -- 生成连续小时时间序列，覆盖数据的最小到最大时间范围
    SELECT generate_series(
        (SELECT date_trunc('hour', MIN(event_timestamp)) FROM public.meter_messages WHERE serial_no = 'ABC/230121840267'),
        (SELECT date_trunc('hour', MAX(event_timestamp)) FROM public.meter_messages WHERE serial_no = 'ABC/230121840267'),
        INTERVAL '1 hour'
    ) AS ts
),
meter_data AS (
    -- 按小时聚合原始数据，取每个时段最后一次有效读数（去重）
    SELECT 
        date_trunc('hour', event_timestamp) AS ts,
        LAST_VALUE(reading) OVER (PARTITION BY date_trunc('hour', event_timestamp) ORDER BY event_timestamp) AS reading
    FROM public.meter_messages
    WHERE serial_no = 'ABC/230121840267'
    GROUP BY date_trunc('hour', event_timestamp), reading, event_timestamp
),
filled_data AS (
    -- 关联时间序列与电表数据，获取前后有效读数及时间
    SELECT 
        ts,
        LAG(reading) OVER (ORDER BY ts) AS prev_reading,
        LEAD(reading) OVER (ORDER BY ts) AS next_reading,
        LAG(ts) OVER (ORDER BY ts) AS prev_ts,
        LEAD(ts) OVER (ORDER BY ts) AS next_ts,
        reading
    FROM time_series
    LEFT JOIN meter_data ON time_series.ts = meter_data.ts
),
interpolated_data AS (
    -- 计算线性插值后的读数
    SELECT 
        ts AS event_timestamp_local,
        CASE 
            WHEN reading IS NOT NULL THEN reading
            WHEN next_reading IS NULL THEN prev_reading
            WHEN prev_reading IS NULL THEN next_reading
            ELSE prev_reading + (next_reading - prev_reading) * 
                EXTRACT(EPOCH FROM (ts - prev_ts)) / 
                EXTRACT(EPOCH FROM (next_ts - prev_ts))
        END AS reading
    FROM filled_data
)
-- 计算连续序列的能耗
SELECT 
    event_timestamp_local,
    reading,
    CASE 
        WHEN LAG(reading) OVER (ORDER BY event_timestamp_local) IS NULL THEN 0
        ELSE reading - LAG(reading) OVER (ORDER BY event_timestamp_local)
    END AS consumption
FROM interpolated_data
ORDER BY event_timestamp_local;

适配30分钟粒度的修改 #

将以下两处修改为30 minutes即可：

1. generate_series的间隔参数：INTERVAL '30 minutes'
2. date_trunc的参数：date_trunc('30 minutes', event_timestamp)

关键说明 #

• 用LAST_VALUE处理重复读数，确保每个时段仅保留最后一次有效读数
• 插值逻辑覆盖序列开头、结尾及中间缺失场景，保证时间序列完全连续
• 能耗基于插值后的连续读数计算，彻底消除数据突刺问题

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