如何用Django TruncHour实现多商品每小时最新价格求和查询？

核心思路 #

要实现每小时取心愿单内每个商品的最新价格，再计算该小时的总价，需要分两步处理：

1. 找到每个商品在过去24小时内每小时的最新价格记录（基于creation_time的最大值）
2. 基于这些最新记录，按小时分组计算价格总和

完整代码实现 #

import datetime
from django.db.models import Max, Sum, Subquery, OuterRef, TruncHour

now = datetime.datetime.now()
last_day = now - datetime.timedelta(hours=23)

# 子查询：获取每个商品对应小时的最新价格创建时间
latest_time_subquery = FloorPrice.objects.filter(
    product=OuterRef("product"),
    creation_time__gte=last_day,
    hour=OuterRef("hour")
).annotate(
    hour=TruncHour("creation_time")
).values("product", "hour").annotate(
    latest_time=Max("creation_time")
).values("latest_time")

# 筛选出每个商品每小时的最新价格记录
filtered_prices = FloorPrice.objects.filter(
    product__in=vault_products_ids,
    creation_time__gte=last_day
).annotate(
    hour=TruncHour("creation_time")
).filter(
    creation_time=Subquery(latest_time_subquery)
)

# 按小时分组计算总价，并按时间排序
hourly_total = filtered_prices.values("hour").annotate(
    total_price=Sum("floor_price")
).order_by("hour")

关键说明 #

• OuterRef + Subquery：用来关联外层查询的商品和小时维度，精准定位每个商品在对应小时的最新价格记录，避免重复计算
• TruncHour：将creation_time截断到小时级别，保证同一小时的记录归为一组
• order_by("hour")：确保结果按时间顺序返回，直接适配图表的横轴展示需求

补充处理（可选） #

如果存在某小时部分商品无价格更新的情况，可生成完整的24小时时间序列，将缺失小时的总价填充为0，保证图表数据的连续性。

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