如何用Django ORM实现分组聚合，获取各证券最新价格（无需原生SQL）

没问题！我来帮你搞定这个需求——不用写原生SQL，纯靠Django ORM就能实现每种证券最新价格的查询。先假设你的模型结构大概是这样（如果和你的实际模型有出入，你可以对应调整字段名就行）：

from django.db import models

# 证券主表
class Security(models.Model):
    code = models.CharField(max_length=10, unique=True, verbose_name='证券代码')
    name = models.CharField(max_length=50, verbose_name='证券名称')

# 价格记录表（每条记录对应某一时刻的证券价格）
class PriceRecord(models.Model):
    security = models.ForeignKey(Security, on_delete=models.CASCADE, verbose_name='关联证券')
    price = models.DecimalField(max_digits=10, decimal_places=2, verbose_name='价格')
    created_at = models.DateTimeField(auto_now_add=True, verbose_name='记录时间')

下面给你两种常用且高效的实现方式：

方法一：子查询（推荐，Django 1.11+支持） #

这个方法逻辑清晰，而且性能不错，尤其是数据量较大的时候。核心思路是：先为每个证券找到它最新的价格记录时间（或记录ID），再筛选出对应这条时间/ID的价格记录。

方式1：按记录时间筛选 #

from django.db.models import Subquery, OuterRef

# 子查询：获取每个证券的最新价格记录时间
latest_time_subquery = PriceRecord.objects.filter(
    security=OuterRef('pk')  # 关联外部查询的Security主键
).order_by('-created_at').values('created_at')[:1]  # 按时间倒序，取第一条的时间

# 查询证券并关联最新价格
securities_with_latest_price = Security.objects.annotate(
    latest_price_time=Subquery(latest_time_subquery)
).filter(
    # 筛选出该证券下时间等于最新时间的价格记录
    pricerecord__created_at=models.F('latest_price_time')
).values('code', 'name', 'pricerecord__price')  # 指定要返回的字段

# 遍历结果
for item in securities_with_latest_price:
    print(f"证券：{item['code']}-{item['name']}，最新价格：{item['pricerecord__price']}")

方式2：按记录ID筛选（避免同一时间多条记录的问题） #

如果你的系统可能出现同一证券在同一时间有多条价格记录的情况，用ID筛选更准确（因为ID是自增的，最新插入的记录ID更大）：

from django.db.models import Subquery, OuterRef

# 子查询：获取每个证券的最新价格记录ID
latest_record_id_subquery = PriceRecord.objects.filter(
    security=OuterRef('pk')
).order_by('-created_at', '-id').values('id')[:1]  # 先按时间倒序，再按ID倒序

# 直接查询最新的价格记录集合
latest_price_records = PriceRecord.objects.filter(
    id__in=Subquery(latest_record_id_subquery)
).select_related('security')  # 关联证券表，避免N+1查询

# 遍历结果
for record in latest_price_records:
    print(f"证券：{record.security.code}-{record.security.name}，最新价格：{record.price}")

方法二：用annotate+Max聚合 #

这种方法先通过聚合得到每个证券的最新记录时间，再筛选出对应记录：

from django.db.models import Max

# 第一步：获取每个证券的最新价格记录时间
latest_time_map = PriceRecord.objects.values('security').annotate(
    latest_time=Max('created_at')
)

# 第二步：把(证券ID, 最新时间)转成元组列表，用来筛选
filter_tuples = [(item['security'], item['latest_time']) for item in latest_time_map]

# 第三步：筛选出符合条件的价格记录
latest_price_records = PriceRecord.objects.filter(
    # 匹配每个(证券ID, 时间)组合
    models.Q(security=models.F('security'), created_at=models.F('latest_time'))
).filter(
    security__in=[t[0] for t in filter_tuples],
    created_at__in=[t[1] for t in filter_tuples]
).distinct().select_related('security')

# 遍历结果和上面一样，这里就省略啦

如果你的模型字段名和示例不一样，比如证券代码字段叫stock_code，记录时间叫update_time，只要对应替换字段名就行，逻辑都是通用的~

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