Python：将API嵌套字典转为带Datetime索引的DataFrame

问题描述 #

我用Python的requests.get从API获取小时级价格数据：

result = (requests.get(url_prices, headers=headers, params={'SpotKey':'1','Fields':'hours','FromDate':'2016-05-05','ToDate':'2016-12-05','Currency':'eur','SortType':'ascending'}).json())

其中SpotKey='1'用于指定获取小时级价格时间序列，其他参数含义清晰。

API返回嵌套字典结构，核心列表Elements包含Date和TimeSpans字段，TimeSpans内存储各时段的价格数据。我需要将其转换为以Datetime为索引、价格为列的DataFrame，目标格式如下：

Datetime           eur/mwh
2016-05-05 00:00:00 23.69
2016-05-05 01:00:00 21.86
... ...
2016-12-05 23:00:00 33.51

目前已实现转换，但夏令时期间TimeSpan会变为'dts'字符串，导致日期解析错误。由于需要频繁请求不同粒度数据，希望找到高效、标准的转换方法，同时规避夏令时问题。

解决方案 #

1. 统一处理夏令时特殊时段 #

夏令时期间的'dts'通常对应夏令时转换的特殊时段（比如欧洲中部时间中，3月最后一个周日的2点不存在，10月最后一个周日的2点会重复）。处理逻辑如下：

• 遍历Elements中的每个日期条目，将Date解析为日期对象
• 对TimeSpans中的每个元素：
  • 若为数字（代表0-23的小时数），直接拼接日期与小时生成Datetime
  • 若为'dts'，根据业务需求选择跳过不存在的时段，或保留重复时段的价格

2. Pandas批量高效转换 #

利用Pandas的时间处理能力批量生成时间序列，避免逐行解析的低效：

import pandas as pd
import requests

# 获取API数据
result = requests.get(url_prices, headers=headers, params={'SpotKey':'1','Fields':'hours','FromDate':'2016-05-05','ToDate':'2016-12-05','Currency':'eur','SortType':'ascending'}).json()

# 提取日期与价格数据
all_dates = []
all_prices = []
for elem in result['Elements']:
    base_date = pd.to_datetime(elem['Date']).date()
    # 假设TimeSpans和Prices是一一对应的列表
    for ts, price in zip(elem['TimeSpans'], elem['Prices']):
        if ts != 'dts':
            hour = int(ts)
            dt = pd.Timestamp(base_date) + pd.Timedelta(hours=hour)
            all_dates.append(dt)
            all_prices.append(price)
        # 若需处理dts时段，可在此补充逻辑（如重复时段保留价格）

# 构建并排序DataFrame
price_df = pd.DataFrame({'eur/mwh': all_prices}, index=all_dates).sort_index()

3. 时区标准化彻底规避问题 #

如果API基于特定时区（比如欧洲中部时间CET/CEST），使用Pandas的时区本地化功能处理夏令时：

# 为索引添加时区信息，自动处理夏令时的歧义与不存在时段
price_df.index = price_df.index.tz_localize(
    'Europe/Brussels',
    ambiguous='infer',  # 自动推断重复时段的夏令时转换
    nonexistent='shift_forward'  # 不存在的时段自动跳转到下一小时
)

4. 封装为可复用函数 #

针对频繁请求的场景，封装成函数提高复用性：

def fetch_and_transform_prices(url, headers, params, tz='Europe/Brussels'):
    result = requests.get(url, headers=headers, params=params).json()
    all_dates = []
    all_prices = []
    
    for elem in result['Elements']:
        base_date = pd.to_datetime(elem['Date']).date()
        for ts, price in zip(elem['TimeSpans'], elem['Prices']):
            if ts == 'dts':
                continue  # 可根据需求修改处理逻辑
            hour = int(ts)
            dt_naive = pd.Timestamp(base_date) + pd.Timedelta(hours=hour)
            # 本地化时区，处理夏令时
            dt_tz = dt_naive.tz_localize(tz, ambiguous='infer', nonexistent='shift_forward')
            all_dates.append(dt_tz)
            all_prices.append(price)
    
    return pd.DataFrame({'eur/mwh': all_prices}, index=all_dates).sort_index()

# 使用示例
params = {'SpotKey':'1','Fields':'hours','FromDate':'2016-05-05','ToDate':'2016-12-05','Currency':'eur','SortType':'ascending'}
price_df = fetch_and_transform_prices(url_prices, headers, params)

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