嵌套字典列表低延迟格式化优化及输出字段补全求助

问题梳理 #

你现在需要解决两个核心问题：一是让输出结果包含f_id和c_id字段，匹配期望格式；二是优化现有的处理逻辑，降低时间复杂度，实现低延迟处理。当前的实现里，时间戳转换过于冗余，而且排序+groupby的组合在数据量大的时候会拖慢速度。

优化后的代码实现 #

我直接给你写了一个更高效的版本，咱们来看看：

import datetime

def process_signals(signals):
    final_output = []
    # 用字典直接做分组，避免排序+groupby的O(n log n)开销
    groups = {}
    # 提前计算一次处理时间，别在循环里反复算
    processing_time = datetime.datetime.utcnow().replace(microsecond=0)
    
    for sig in signals:
        u_id = sig['u_id']
        # 计算分钟级时间戳：直接用整数运算，比转字符串快太多了
        minute_ts = (sig['start_ts'] // 60) * 60
        group_key = (u_id, minute_ts)
        
        if group_key not in groups:
            # 初始化分组的时候直接把f_id、c_id带上
            groups[group_key] = {
                'ui_id': u_id,
                'f_id': sig['f_id'],  # 这里如果需要固定为311，直接改成311就行
                'c_id': sig['c_id'],
                'minute_utc': datetime.datetime.fromtimestamp(minute_ts),
                'data': [],
                'processing_timestamp_utc': processing_time
            }
        
        # 格式化单条信号的数据
        temp_dict = {'timestamp_utc': sig['start_ts']}
        for s in sig['sign']:
            # 处理name里的点，取第一个部分
            key = s['name'].split('.')[0]
            temp_dict[key] = s['val']
        groups[group_key]['data'].append(temp_dict)
    
    # 把分组字典转成列表输出
    final_output = list(groups.values())
    return final_output

# 你的输入信号
signals = [
    {'c_id': '1234', 'u_id': 288, 'f_id': 331, 'sign': [{'name': 'speed', 'val': 9}, {'name': 'pwr', 'val': 1415}], 'start_ts': 1598440244, 'crt_ts': 1598440349, 'map_crt_ts': 1598440351, 'ca_id': 'AT123', 'c_n': 'demo', 'msg_cnt': 2, 'window': 'na', 'type': 'na'},
    {'c_id': '1234', 'u_id': 288, 'f_id': 331, 'sign': [{'name': 'speed', 'val': 10}, {'name': 'pwr', 'val': 1416}], 'start_ts': 1598440243, 'crt_ts': 1598440349, 'map_crt_ts': 1598440351, 'ca_id': 'AT123', 'c_n': 'demo', 'msg_cnt': 2, 'window': 'na', 'type': 'na'},
    {'c_id': '1234', 'u_id': 287, 'f_id': 331, 'sign': [{'name': 'speed', 'val': 10}, {'name': 'pwr', 'val': 1417}], 'start_ts': 1598440344, 'crt_ts': 1598440349, 'map_crt_ts': 1598440351, 'ca_id': 'AT123', 'c_n': 'demo', 'msg_cnt': 2, 'window': 'na', 'type': 'na'},
    {'c_id': '1234', 'u_id': 288, 'f_id': 331, 'sign': [{'name': 'speed.', 'val': 8.2}, {'name': 'pwr', 'val': 925}], 'start_ts': 1598440345, 'crt_ts': 1598440349, 'map_crt_ts': 1598440351, 'ca_id': 'AT172', 'c_n': 'demo', 'msg_cnt': 2, 'window': 'na', 'type': 'na'}
]

# 运行处理
result = process_signals(signals)
# 打印结果
for item in result:
    print(item)

为什么这个版本更快？ #

1. 时间复杂度从O(n log n)降到O(n)：原来用sorted+groupby需要先排序，现在直接用字典分组，遍历一次就搞定，数据量越大，提升越明显
2. 时间戳计算简化：原来反复把时间戳转字符串、转时间对象，现在直接用整数整除60再乘60，一步得到分钟级时间戳，省了超多不必要的计算
3. 字段补全一步到位：初始化分组的时候就把f_id和c_id加上，不用事后再处理
4. 减少临时变量：简化了信号数据的构建逻辑，避免了多余的字典拷贝操作

注意点 #

你期望输出里的f_id是311，但输入信号里的f_id都是331，如果需要固定输出311，直接把代码里的sig['f_id']改成311就行。

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