Pandas处理夏令时重复小时的Excel/CSV兼容方案问询

核心问题分析 #

夏令时切换至冬令时时，本地时间会重复出现同一小时（如示例中的30.10.2022 02:00出现两次），Pandas默认将这些时间视为重复索引，去重操作会丢失数据，排序也会打乱原本的先后顺序；同时CSV格式需兼容Excel，无法存储时区信息，因此需要通过本地时间的辅助标记来解决重复项识别问题。

分步解决方案 #

1. CSV导入：保留原始顺序，标记重复小时 #

导入CSV时不直接将时间设为索引，先读取为普通列，通过辅助列标记重复时间的先后顺序，再用复合索引（时间+重复计数）保证唯一性：

修改load_old_data函数：

def load_old_data():
    try:
        if USE_PKL: 
            df = pd.read_pickle(filepath_or_buffer=pricing_pkl, compression='infer')
        else: 
            # 读取时不设置索引，保留时间列为普通字段
            df = pd.read_csv(filepath_or_buffer=pricing_file, delimiter=';', header=0, decimal=",")
            # 解析时间列
            df['from'] = pd.to_datetime(df['from'], format=DATE_FORMAT)
            # 添加辅助列：对相同时间分组，标记是第几次出现（1=第一次，2=第二次）
            df['dup_count'] = df.groupby('from').cumcount() + 1
            # 设置复合索引，确保唯一标识每一条记录
            df.set_index(['from', 'dup_count'], inplace=True)
    except:
        # 初始化空DataFrame，保持结构一致
        df = pd.DataFrame(columns=['Price'])
        df = df.set_index(['from', 'dup_count'])
    return df

2. API拉取数据：同步标记重复项 #

从EXAA API获取的数据是按交割时段顺序返回的，直接对时间分组计数，生成和老数据一致的复合索引：

修改load_api_data函数：

def load_api_data():
    s = requests.Session()
    data_list = []

    for offset in range(0, 2):
        onDate = (date.today() - timedelta(days=offset))
        print(onDate)
        
        url = f'https://www.exaa.at/data/trading-results?delivery_day={onDate.strftime("%Y/%m/%d")}&market=de&auction=grey'
        resp = s.get(url)
        quarterly_data = resp.json()["data"]["q"]

        for i in quarterly_data:
            timestamp = datetime.strptime(i["AuctionDay"],'%Y-%m-%d')
            parsed_regex = re.search(r"^Q(\d\d)_(\d)$",i["Product"])
            timestamp_hour = int(parsed_regex.group(1)) - 1 
            timestamp_minute = (int(parsed_regex.group(2)) - 1) * 15
            timestamp = timestamp.replace(hour=timestamp_hour, minute=timestamp_minute)
            # 用列表存储，方便后续处理重复项
            data_list.append({'from': timestamp, 'Price': float(i["Price"])})
    
    # 转换为DataFrame并添加重复计数
    appending_df = pd.DataFrame(data_list)
    appending_df['dup_count'] = appending_df.groupby('from').cumcount() + 1
    appending_df.set_index(['from', 'dup_count'], inplace=True)
    return appending_df

3. 数据追加与去重：基于复合索引处理 #

复合索引(from, dup_count)是唯一的，追加后去重仅针对完全重复的复合索引（即同一时间的同一次出现），排序时按时间+计数排序，保证重复小时的原始顺序：

修改主流程代码：

df = load_old_data()
appending_df = load_api_data()

# 合并新旧数据
df = pd.concat([df, appending_df])
# 基于复合索引去重，保留最新的记录
df = df[~df.index.duplicated(keep='last')]
# 按时间+重复计数排序，确保重复小时的顺序正确
df.sort_index(inplace=True)
save_data(df)

4. 保存CSV：兼容Excel格式 #

保存CSV时拆分复合索引，仅保留原始时间列和价格列，去掉辅助计数列，保证Excel可以正常解析：

修改save_data函数：

def save_data(df):
    # PKL文件保留复合索引，方便下次快速加载
    df.to_pickle(path=pricing_pkl, compression='infer')
    # 重置索引，拆分复合索引为普通列
    df_reset = df.reset_index()
    # 保存CSV时仅保留时间和价格列，格式符合要求
    df_reset[['from', 'Price']].to_csv(
        path_or_buf=pricing_file, 
        sep=";", 
        index=False, 
        encoding="ANSI", 
        decimal=",", 
        date_format=DATE_FORMAT
    )

关键逻辑说明 #

• 复合索引：通过(时间, 重复计数)作为唯一标识，既保留了原始本地时间，又能区分夏令时切换时的重复小时，解决了Pandas无法识别本地时间重复项的问题。
• 顺序保留：groupby('from').cumcount()按数据原始顺序标记重复项，排序时按时间+计数排序，确保第一次出现的重复小时在前，第二次在后。
• CSV兼容性：保存时自动去掉辅助计数列，仅输出Excel可识别的时间和价格字段，同时PKL文件保留完整索引信息，避免每次加载都重新处理重复项。

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