如何高效获取Django带时间戳模型每日的首尾实例？

现有Django模型如下（已修正代码笔误）：

from django import models
from django.contrib.postgres.indexes import BrinIndex


class MyModel(models.Model):
    device_id = models.IntegerField()
    timestamp = models.DateTimeField(auto_now_add=True)
    my_value = models.FloatField()

    class Meta:
        indexes = (BrinIndex(fields=['timestamp']),)

系统每2分钟为多台设备创建MyModel实例，长期运行后表中会积累海量数据。

需求：获取指定设备每个有记录日期的首尾实例（包含对应my_value）。

原实现方案可行但性能极差（获取450天数据耗时约50秒），代码如下：

from django.db.models import Min, Max


results = []
device_id = 1  # 示例用1，实际可为其他设备ID

# 原代码此处存在错误：Min/Max应作用于timestamp而非timestamp__date
first_last = MyModel.objects.filter(device_id=device_id).values('timestamp__date')\
    .annotate(first=Min('timestamp__date'),last=Max('timestamp__date'))

# 循环中每次执行2次查询，导致大量额外请求
for f in first_last:
    first = f['first']
    last = f['last']

    first_value = MyModel.objects.get(device_id=device_id, timestamp=first).my_value
    last_value = MyModel.objects.get(device_id=device_id, timestamp=last).my_value

    results.append({
      'first': first,
      'last': last,
      'first_value': first_value,
      'last_value': last_value,
    })

# 对results进行后续处理

优化方案 #

原方案的核心问题是循环中产生了N×2次额外查询，以下两种方案可大幅减少查询次数，提升性能：

方案1：利用子查询批量获取数据 #

通过Django子查询功能，仅需3次查询即可完成所有数据获取：

from django.db.models import Min, Max, Subquery, OuterRef

device_id = 1

# 第一步：按日期分组，获取每天的最早、最晚时间戳
date_groups = MyModel.objects.filter(device_id=device_id)\
    .values('timestamp__date')\
    .annotate(
        earliest_ts=Min('timestamp'),
        latest_ts=Max('timestamp')
    )

# 第二步：定义子查询，获取对应时间戳的my_value
earliest_value_subq = MyModel.objects.filter(
    device_id=device_id,
    timestamp=OuterRef('earliest_ts')
).values('my_value')[:1]

latest_value_subq = MyModel.objects.filter(
    device_id=device_id,
    timestamp=OuterRef('latest_ts')
).values('my_value')[:1]

# 关联子查询，一次性获取所有结果
results = date_groups.annotate(
    first_value=Subquery(earliest_value_subq),
    last_value=Subquery(latest_value_subq)
).values(
    'timestamp__date', 'earliest_ts', 'latest_ts', 'first_value', 'last_value'
)

# 转换为目标格式
final_results = [
    {
        'first': item['earliest_ts'],
        'last': item['latest_ts'],
        'first_value': item['first_value'],
        'last_value': item['last_value']
    }
    for item in results
]

方案2：利用PostgreSQL窗口函数（性能最优） #

依托PostgreSQL的窗口函数，在数据库层面完成分组筛选，仅需1次查询：

from django.db.models import F, Window
from django.db.models.functions import RowNumber

device_id = 1

# 按日期分组，给每天的记录按时间戳升序、降序分别编号
annotated_records = MyModel.objects.filter(device_id=device_id)\
    .annotate(
        date=F('timestamp__date'),
        row_asc=Window(
            partition_by=F('timestamp__date'),
            order_by=F('timestamp').asc()
        ),
        row_desc=Window(
            partition_by=F('timestamp__date'),
            order_by=F('timestamp').desc()
        )
    )

# 筛选出每天的第一条（row_asc=1）和最后一条（row_desc=1）记录
target_records = annotated_records.filter(
    (F('row_asc') == 1) | (F('row_desc') == 1)
).values(
    'date', 'timestamp', 'my_value'
)

# 整理为目标格式
final_results = {}
for record in target_records:
    date_key = record['date']
    if date_key not in final_results:
        final_results[date_key] = {
            'first': None, 'last': None,
            'first_value': None, 'last_value': None
        }
    # 更新最早记录
    if not final_results[date_key]['first'] or record['timestamp'] < final_results[date_key]['first']:
        final_results[date_key]['first'] = record['timestamp']
        final_results[date_key]['first_value'] = record['my_value']
    # 更新最晚记录
    if not final_results[date_key]['last'] or record['timestamp'] > final_results[date_key]['last']:
        final_results[date_key]['last'] = record['timestamp']
        final_results[date_key]['last_value'] = record['my_value']

final_results = list(final_results.values())

额外优化建议 #

• 添加复合索引：原模型仅针对timestamp创建了Brin索引，建议新增复合索引models.Index(fields=['device_id', 'timestamp'])，适配按设备过滤+时间排序的查询场景，进一步提升速度。
• 分表优化：若数据量极大，可使用Django分表插件（如django-partition）按日期分表，降低单表数据量。

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