SQLAlchemy按30分钟间隔分组计算能耗的问题求助

当前查询用max(kwh)-min(kwh)计算区间能耗的方式，无法覆盖跨区间的读数差（比如xx:59到xx:01的消耗），因为这部分差值会被拆分到两个区间。以下是两种针对性的解决方案：

方案1：基于窗口函数分配跨区间能耗 #

利用LEAD()窗口函数获取每条记录的下一条读数，计算相邻读数的实际消耗，再将消耗分配到当前记录所属的30分钟区间，最终按区间汇总。这种方式自动将跨区间的消耗归到前一个区间。

# 1. 为每条记录关联下一条记录的读数和时间
with_next_readings = select(
    models.EnergyLogTable.created_at,
    models.EnergyLogTable.kwh,
    func.lead(models.EnergyLogTable.kwh).over(order_by=models.EnergyLogTable.created_at).label('next_kwh'),
    func.lead(models.EnergyLogTable.created_at).over(order_by=models.EnergyLogTable.created_at).label('next_created_at')
).where(
    models.EnergyLogTable.created_at >= from_date,
    models.EnergyLogTable.created_at < to_date,
    models.EnergyLogTable.kwh > 0
).cte('with_next_readings')

# 2. 计算单条记录的能耗，并确定所属30分钟区间
interval_consumption = select(
    func.to_timestamp(
        cast(
            cast(func.extract('epoch', func.date_trunc('minute', with_next_readings.c.created_at)), Integer) / 1800,
            Integer
        ) * 1800
    ).label('date_label'),
    (with_next_readings.c.next_kwh - with_next_readings.c.kwh).label('energy_delta')
).where(
    with_next_readings.c.next_kwh.is_not(None)  # 排除无后续读数的最后一条记录
).cte('interval_consumption')

# 3. 按区间汇总总能耗
final_query = select(
    interval_consumption.c.date_label,
    func.sum(interval_consumption.c.energy_delta).label('energy_spent')
).group_by(interval_consumption.c.date_label).order_by(interval_consumption.c.date_label)

逻辑说明 #

• 每条记录的energy_delta对应两个时间点之间的实际消耗，将其分配到当前记录所属的区间，跨区间的消耗会自然归属前一个区间
• 汇总后每个区间的能耗是该时间段内所有连续消耗的总和，包含跨区间的部分

方案2：生成完整时间区间+关联首尾读数 #

如果需要统计所有30分钟区间（包括无数据的区间），可以先生成完整的时间区间序列，再为每个区间匹配起始和结束时刻的电表读数，计算差值得到能耗。

# 1. 生成所有30分钟区间的起始时间（PostgreSQL语法，其他数据库需调整）
interval_starts = select(
    func.generate_series(from_date, to_date, '30 minutes').label('interval_start')
).cte('interval_starts')

# 2. 计算每个区间的结束时间（左闭右开：[interval_start, interval_end)）
intervals = select(
    interval_starts.c.interval_start,
    (interval_starts.c.interval_start + func.interval('30 minutes')).label('interval_end')
).cte('intervals')

# 3. 获取区间开始前的最后一次电表读数
start_kwh = select(
    intervals.c.interval_start,
    func.coalesce(func.max(models.EnergyLogTable.kwh), 0).label('start_reading')
).outerjoin(
    models.EnergyLogTable,
    models.EnergyLogTable.created_at < intervals.c.interval_start
).group_by(intervals.c.interval_start).cte('start_kwh')

# 4. 获取区间结束前的最后一次电表读数
end_kwh = select(
    intervals.c.interval_start,
    func.coalesce(func.max(models.EnergyLogTable.kwh), start_kwh.c.start_reading).label('end_reading')
).outerjoin(
    models.EnergyLogTable,
    models.EnergyLogTable.created_at < intervals.c.interval_end
).outerjoin(
    start_kwh,
    intervals.c.interval_start == start_kwh.c.interval_start
).group_by(intervals.c.interval_start, start_kwh.c.start_reading).cte('end_kwh')

# 5. 计算每个区间的能耗
final_query = select(
    intervals.c.interval_start.label('date_label'),
    (end_kwh.c.end_reading - start_kwh.c.start_reading).label('energy_spent')
).select_from(
    intervals.join(start_kwh, intervals.c.interval_start == start_kwh.c.interval_start)
             .join(end_kwh, intervals.c.interval_start == end_kwh.c.interval_start)
).order_by(intervals.c.interval_start)

逻辑说明 #

• 生成完整的30分钟区间序列，确保无数据的区间也能显示（能耗为0）
• start_reading是区间起始时的累计读数，end_reading是区间结束时的累计读数，差值即为该区间的总能耗
• 跨区间的读数差会被包含在前一个区间的end_reading中，符合你的需求

示例验证 #

以你提供的示例数据为例：

• 06:30-07:00区间的start_reading=559.40，end_reading=560.11，能耗为560.11-559.40=0.71，完全覆盖跨区间的消耗
• 07:00-07:30区间的start_reading=560.11，end_reading=565.27，能耗为5.16，无重复计算

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