Django使用TruncHour统计时补全零计数小时的问题求助

你当前的查询仅返回有事件的小时，是因为GROUP BY聚合只会处理表中存在的数据，不会自动生成不存在的时间区间。另外你原有查询的写法存在分组逻辑错误：annotate(event_count=Count('*')) 放在values之前不会按小时分组统计，需要调整顺序先指定分组字段再执行聚合，否则会得到错误的计数结果。

以下是两种可行的实现方案： #

方案1：Python层补全（实现成本最低，24小时场景性能无损耗） #

该方案无需修改数据库查询逻辑，仅对查询结果做后处理，代码简单易维护：

import datetime
from django.db.models import Count
from django.db.models.functions import TruncHour

# 修正后的原有统计查询，按小时分组统计
event_volume = IngestModels.EventData.objects.filter(
    product=my_product, created__date=selected_date
).annotate(
    hour=TruncHour('created')
).values("hour").annotate(
    event_count=Count('*')
)
# 把统计结果转成 小时:计数 的映射字典
event_count_map = {item['hour']: item['event_count'] for item in event_volume}

# 生成全天24小时的完整结果，缺失小时补0
full_result = []
start_dt = datetime.datetime.combine(selected_date, datetime.time(0,0,0))
for hour_offset in range(24):
    current_hour = start_dt + datetime.timedelta(hours=hour_offset)
    # 若项目开启时区支持，需给current_hour加上对应时区避免匹配错误
    full_result.append({
        "hour": current_hour,
        "event_count": event_count_map.get(current_hour, 0)
    })

方案2：PostgreSQL数据库层面实现（适合大数量/多时间区间聚合场景） #

利用PostgreSQL内置的generate_series函数生成完整时间序列，再和统计结果左连接，直接在数据库层面完成补0，无需Python层后处理：

原生SQL实现（性能最优） #

import datetime
from django.db import connection

start_dt = datetime.datetime.combine(selected_date, datetime.time(0,0,0))
end_dt = datetime.datetime.combine(selected_date, datetime.time(23,59,59))

with connection.cursor() as cursor:
    cursor.execute("""
        WITH all_hours AS (
            SELECT generate_series(%s, %s, INTERVAL '1 hour') AS hour
        )
        SELECT 
            all_hours.hour,
            COALESCE(COUNT(ed.id), 0) AS event_count
        FROM all_hours
        LEFT JOIN event_data ed 
            ON DATE_TRUNC('hour', ed.created) = all_hours.hour
            AND ed.product_id = %s
        GROUP BY all_hours.hour
        ORDER BY all_hours.hour
    """, [start_dt, end_dt, my_product.id])
    # 直接返回补0后的完整结果
    full_result = [{"hour": row[0], "event_count": row[1]} for row in cursor.fetchall()]

纯Django ORM实现（无需写原生SQL） #

import datetime
from django.db.models import Count, F, OuterRef, Subquery, Coalesce, Value, DateTimeField
from django.db.models.functions import TruncHour

start_dt = datetime.datetime.combine(selected_date, datetime.time(0,0,0))
end_dt = datetime.datetime.combine(selected_date, datetime.time(23,59,59))

# 自定义ORM函数调用PG的generate_series
class GenerateSeries(Func):
    function = 'generate_series'
    output_field = DateTimeField()

# 生成全天24小时的基础查询集
all_hours_qs = GenerateSeries(
    Value(start_dt),
    Value(end_dt),
    Value('1 hour', output_field=DateTimeField())
).annotate(hour=F('generate_series')).values('hour')

# 事件统计子查询
event_subq = IngestModels.EventData.objects.filter(
    product=my_product,
    hour=OuterRef('hour')
).annotate(
    hour=TruncHour('created')
).values('hour').annotate(
    cnt=Count('*')
).values('cnt')

# 关联查询补0
full_result = all_hours_qs.annotate(
    event_count=Coalesce(Subquery(event_subq), 0)
).values('hour', 'event_count')

注意事项 #

如果你的Django项目开启了时区支持，需要确保生成的时间序列和created字段的时区一致，避免出现时间匹配错误。

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