Pandas处理含嵌套字典的DataFrame时间格式报错修复

基础信息 #

• 待处理的DataFrame数据源结构如下：

[
  {
    "symbol_id": "BITSTAMP_SPOT_BTC_USD",
    "time_exchange": "2013-09-28T22:40:50.0000000Z",
    "time_coinapi": "2017-03-18T22:42:21.3763342Z",
    "ask_price": 770.000000000,
    "ask_size": 3252,
    "bid_price": 760,
    "bid_size": 124,
    "last_trade": {
        "time_exchange": "2017-03-18T22:42:21.3763342Z",
        "time_coinapi": "2017-03-18T22:42:21.3763342Z",
        "uuid": "1EA8ADC5-6459-47CA-ADBF-0C3F8C729BB2",
        "price": 770.000000000,
        "size": 0.050000000,
        "taker_side": "SELL"
    }
  },
  {
    "symbol_id": "BITSTAMP_SPOT_BTC_USD",
    "time_exchange": "2013-09-28T22:40:50.0000000Z",
    "time_coinapi": "2017-03-18T22:42:21.3763342Z",
    "ask_price": 770.000000000,
    "ask_size": 3252,
    "bid_price": 760,
    "bid_size": 124,
    "last_trade": {
        "time_exchange": "2017-03-18T22:42:21.3763342Z",
        "time_coinapi": "2017-03-18T22:42:21.3763342Z",
        "uuid": "1EA8ADC5-6459-47CA-ADBF-0C3F8C729BB2",
        "price": 770.000000000,
        "size": 0.050000000,
        "taker_side": "SELL"
    }
  }
]

• 需求目标：
  • 移除所有顶层时间类列值中的T、Z字符，统一转换为yyyy-mm-dd hh:mm:ss格式
  • 解析嵌套字典类型的last_trade列，将列内的所有时间字段也统一转换为上述时间格式
  • 处理完成后将数据导出为CSV文件

原有代码问题 #

• 类方法嵌套调用时每次都会重新发起API请求拉取数据，运行效率低，且容易因为接口返回波动导致处理流程中断
• 时间处理逻辑零散重复，没有做通用封装，字符串处理逻辑鲁棒性差
• last_trade列处理存在语法错误：x.values([column_in_df4])属于错误的pandas调用方式，且仅处理了单个时间字段，遗漏了嵌套结构内的time_exchange字段
• 方法参数设计不合理，固定了传入列的数量，无法适配嵌套结构存在多个时间字段的场景
• CSV导出逻辑错误：to_csv方法默认返回None，原有代码将该返回值赋值回df会导致变量被覆盖为空

import requests
import json
import pandas as pd


class List_all_current_quotes_data:

    url = "https://rest.coinapi.io/v1/quotes/current" 
    headers = {"X-CoinAPI-Key": "54BE11BF-7A18-4736-A3E6-A7EAB7689DAE"} 

    @staticmethod
    def _format_time(time_str):
        """通用时间格式化工具，统一输出yyyy-mm-dd hh:mm:ss格式"""
        return str(time_str)[:19].replace("T", " ")

    def getting_response_and_df(self): 
        response = requests.get(self.url, headers=self.headers)
        response.raise_for_status()
        data = json.loads(response.text)  
        pd.set_option('display.max_columns', None) 
        pd.set_option('display.max_colwidth', None) 
        df = pd.DataFrame(data)
        return df

    def process_all_time_fields(self, df, top_time_cols, nested_col, nested_time_cols):
        # 处理DataFrame顶层时间列
        for col in top_time_cols:
            df[col] = df[col].apply(self._format_time)
        
        # 处理嵌套字典内的时间字段
        def format_nested_dict(nested_item):
            for t_col in nested_time_cols:
                if t_col in nested_item:
                    nested_item[t_col] = self._format_time(nested_item[t_col])
            return nested_item
        
        df[nested_col] = df[nested_col].apply(format_nested_dict)
        return df

    def get_csv(self, csv_file_name, top_time_cols, nested_col, nested_time_cols):
        # 整个流程仅拉取一次数据，避免重复请求
        df = self.getting_response_and_df()
        # 统一处理所有时间字段
        df = self.process_all_time_fields(df, top_time_cols, nested_col, nested_time_cols)
        # 导出CSV，关闭索引列输出
        df.to_csv(csv_file_name, index=False)
        return df


if __name__ == "__main__":
    data_processor = List_all_current_quotes_data()
    final_df = data_processor.get_csv(
        csv_file_name="List_all_current_quotes_data.csv",
        top_time_cols=["time_exchange", "time_coinapi"],
        nested_col="last_trade",
        nested_time_cols=["time_exchange", "time_coinapi"]
    )
    print("数据处理完成，前2行预览：")
    print(final_df.head(2))

• 抽出通用的_format_time静态方法，所有时间字段统一走这套格式化逻辑，减少重复代码
• 增加接口请求状态校验，接口返回异常状态码时直接抛出明确错误，避免后续JSON解析出现无意义报错
• 重构方法调用逻辑，整个数据处理流程仅发起一次API请求，提升运行效率和稳定性
• 嵌套字段处理改为逐行遍历字典修改对应值，修复原有pandas语法错误，同时支持嵌套结构下多个时间字段的批量处理
• 调整参数设计，传入列名列表代替固定数量的位置参数，适配不同的字段数量场景，代码复用性更高
• 修正CSV导出逻辑，增加index=False参数避免导出无意义的行索引，不再错误接收to_csv的空返回值

注意：代码中硬编码的API密钥属于敏感信息，正式使用时建议通过环境变量读取，不要直接写在源码中

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