如何对嵌套字典列表进行排序、分组与聚合计算？

我有一组嵌套字典列表，需要完成以下操作：

1. 按顶层key「name」对数据排序
2. 按「items」下的嵌套key「name」排序
3. 按聚合间隔（比如「1d」）对items下的values分组
4. 对分组后的结果重新计算min、max和avg值

目前我用遍历取值结合pandas做分组聚合，操作繁琐且性能差，求更优方案。

示例数据 #

import datetime

sample_data = [
    {
        "_id": 2,
        "name": "b",
        "device": "b",
        "items": [
            {
                "item_id": "item_id_2", "name": "item_2", "unit": "b/s",
                "values": [
                    {"time": datetime.datetime(2022, 9, 5, 15, 0), "min": 0.0, "max": 1.0, "avg": 0.5},
                    {"time": datetime.datetime(2022, 9, 5, 16, 0), "min": 0.0, "max": 1.0, "avg": 0.5},
                    {"time": datetime.datetime(2022, 9, 5, 17, 0), "min": 0.0, "max": 1.0, "avg": 0.5},
                    {"time": datetime.datetime(2022, 9, 5, 18, 0), "min": 0.0, "max": 1.0, "avg": 0.5},
                    {"time": datetime.datetime(2022, 9, 5, 19, 0), "min": 0.0, "max": 1.0, "avg": 0.5},
                    {"time": datetime.datetime(2022, 9, 5, 20, 0), "min": 0.0, "max": 1.0, "avg": 0.5},
                ]
            }
        ]
    },
    {
        "_id": 1,
        "name": "a",
        "device": "a",
        "items": [
            {
                "item_id": "item_id_1", "name": "item_1", "unit": "b/s",
                "values": [
                    {"time": datetime.datetime(2022, 9, 5, 15, 0), "min": 0.0, "max": 1.0, "avg": 0.5},
                    {"time": datetime.datetime(2022, 9, 5, 16, 0), "min": 0.0, "max": 1.0, "avg": 0.5},
                    {"time": datetime.datetime(2022, 9, 5, 17, 0), "min": 0.0, "max": 1.0, "avg": 0.5},
                    {"time": datetime.datetime(2022, 9, 5, 18, 0), "min": 0.0, "max": 1.0, "avg": 0.5},
                    {"time": datetime.datetime(2022, 9, 5, 19, 0), "min": 0.0, "max": 1.0, "avg": 0.5},
                    {"time": datetime.datetime(2022, 9, 5, 20, 0), "min": 0.0, "max": 1.0, "avg": 0.5},
                ]
            }
        ]
    }
]

预期结果 #

import datetime

expected_result = [
    {
        "_id": 1,
        "name": "a",
        "device": "a",
        "items": [
            {
                "item_id": "item_id_1", "name": "item_1", "unit": "b/s",
                "values": [
                    {"time": datetime.datetime(2022, 9, 5, 0, 0), "min": 0.0, "max": 1.0, "avg": 0.5},
                ]
            }
        ]
    },
    {
        "_id": 2,
        "name": "b",
        "device": "b",
        "items": [
            {
                "item_id": "item_id_2", "name": "item_2", "unit": "b/s",
                "values": [
                    {"time": datetime.datetime(2022, 9, 5, 0, 0), "min": 0.0, "max": 1.0, "avg": 0.5},
                ]
            }
        ]
    }
]

可以用纯Python结合itertools.groupby实现，避免pandas的额外开销，同时保持代码简洁：

步骤说明 #

1. 顶层排序：用sorted按name字段排序顶层列表
2. items排序：对每个顶层字典的items列表按name排序
3. values分组聚合：用groupby按日期（天）分组，计算每组的min、max、avg

实现代码 #

import datetime
from itertools import groupby

def aggregate_values(values, interval="1d"):
    # 按指定间隔生成分组key，这里先实现天级聚合
    def get_group_key(value):
        if interval == "1d":
            return value["time"].date()
        # 可扩展其他间隔，比如小时级
        # elif interval == "1h":
        #     return (value["time"].date(), value["time"].hour)
        else:
            raise ValueError(f"不支持的聚合间隔：{interval}")
    
    # groupby要求序列有序，先按分组key排序
    sorted_values = sorted(values, key=get_group_key)
    aggregated = []
    
    for group_key, group in groupby(sorted_values, key=get_group_key):
        group_list = list(group)
        # 计算聚合值：min取所有min的最小值，max取所有max的最大值，avg取所有avg的平均值
        agg_min = min(v["min"] for v in group_list)
        agg_max = max(v["max"] for v in group_list)
        agg_avg = sum(v["avg"] for v in group_list) / len(group_list)
        # 生成聚合后的基准时间（天级取当天0点）
        agg_time = datetime.datetime.combine(group_key, datetime.time.min)
        aggregated.append({
            "time": agg_time,
            "min": agg_min,
            "max": agg_max,
            "avg": agg_avg
        })
    return aggregated

def process_data(data):
    # 1. 按顶层name排序
    sorted_top = sorted(data, key=lambda x: x["name"])
    
    for item_group in sorted_top:
        # 2. 按items下的name排序
        item_group["items"] = sorted(item_group["items"], key=lambda x: x["name"])
        
        # 3. 对每个item的values进行聚合
        for item in item_group["items"]:
            item["values"] = aggregate_values(item["values"], interval="1d")
    
    return sorted_top

# 测试运行
result = process_data(sample_data)
print(result)

优势 #

• 性能更优：避免pandas的数据转换开销，纯Python操作更轻量
• 代码简洁：用内置函数实现，无需额外依赖
• 可扩展性：aggregate_values函数可轻松扩展其他聚合间隔（如小时、周）

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