如何使用Django Queryset聚合时序数据生成30分钟OHLC价格数据

你开发加密货币追踪工具时，需要将每5分钟采集的汇率数据压缩为30分钟粒度的OHLC（开盘/最高/最低/收盘）数据用于图表绘制，核心逻辑是按30分钟时间窗口分组统计，具体实现可参考以下两种方案：

核心统计规则 #

• 时间窗口划分：将所有数据按时间归属到30分钟区间，例如10:00:00~10:29:59的所有数据归属到2021-10-23 10:00窗口，10:30:00~10:59:59归属到2021-10-23 10:30窗口，以此类推
• 指标计算规则：
  • Open（开盘）：窗口内时间最早的一条数据的汇率值
  • High（最高）：窗口内所有汇率值的最大值
  • Low（最低）：窗口内所有汇率值的最小值
  • Close（收盘）：窗口内时间最晚的一条数据的汇率值

注：你提供的示例数据存在两处笔误：1. 第二组最后一条数据时间应为2021-10-23 10:55:00而非11:55:00，否则会跨到11:00的窗口；2. 第一组最后一条汇率为70，对应正确Close值为70，你给出的期望输出中57属于输入错误。

方案1：Django ORM直接聚合（适配你给出的Crypto模型） #

适合直接从数据库查询并返回结果的场景，依赖Django内置的时间截断和聚合函数：

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

# 先定义子查询获取每个窗口最早和最晚的价格
# 1. 获取每个窗口的开盘价（最早一条的价格）
open_subq = Crypto.objects.filter(
    window_start=OuterRef('window_start')
).order_by('timestamp').values('price')[:1]

# 2. 获取每个窗口的收盘价（最晚一条的价格）
close_subq = Crypto.objects.filter(
    window_start=OuterRef('window_start')
).order_by('-timestamp').values('price')[:1]

# 主查询分组聚合
ohlc_data = Crypto.objects.annotate(
    window_start=Trunc('timestamp', '30 minute')
).values('window_start').annotate(
    open=Subquery(open_subq),
    high=Max('price'),
    low=Min('price'),
    close=Subquery(close_subq)
).order_by('window_start').distinct()

方案2：Pandas处理（兼容性更高，适合灵活计算） #

如果需要后续做更多数据处理或者适配不同数据库，推荐先拉取原始数据用Pandas重采样计算，代码更简洁：

import pandas as pd

# 从数据库查询指定币种的所有时间序列数据
raw_data = Crypto.objects.filter(ticker="你的币种代码").order_by('timestamp').values('timestamp', 'price')
df = pd.DataFrame(list(raw_data))

# 转换时间格式并设置为索引
df['timestamp'] = pd.to_datetime(df['timestamp'])
df = df.set_index('timestamp')

# 按30分钟重采样直接生成OHLC数据
ohlc_result = df['price'].resample('30T').ohlc()

处理完成的ohlc_result索引为30分钟窗口起始时间，自带open、high、low、close四列数据，可直接用于图表渲染。

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