按833天周期循环调用API下载历史数据并合并为DataFrame/CSV的实现咨询

核心逻辑说明 #

• 用datetime模块的strftime方法统一格式化日期字符串，避免手动拼接出现单月/单日无前置零的格式错误
• 循环过程中自动截断超过当前日期的时间区间，避免请求无效日期
• 每轮请求结束后自动更新下一轮的起始日期，保证时间区间无重叠、无遗漏
• 每次拿到分段数据后直接追加到总DataFrame，最后统一去重导出

完整代码实现 #

import pandas as pd
from datetime import datetime, timedelta

def fetch_api_data(start_str, end_str):
    # 此处替换为你自己的API调用逻辑，返回对应时间段的DataFrame即可
    # 示例：print(f"正在请求时间段：{start_str} 至 {end_str}")
    # return 调用API后返回的分段DataFrame
    pass

def time_machine():
    # 初始化基础参数
    global_start = datetime(2002, 5, 1)
    global_end = datetime.today()
    max_period = timedelta(days=833)
    df_total = pd.DataFrame()
    current_start = global_start

    while current_start < global_end:
        # 计算当前轮次的结束日期，超过今日则取今日作为结束
        current_end = min(current_start + max_period, global_end)
        # 格式化日期为API要求的标准字符串格式
        start_str = current_start.strftime("%Y-%m-%d")
        end_str = current_end.strftime("%Y-%m-%d")
        
        # 调用API拉取当前分段数据
        df_segment = fetch_api_data(start_str, end_str)
        
        # 将分段数据追加到总表，可根据实际需求调整拼接逻辑
        if not df_segment.empty:
            df_total = pd.concat([df_total, df_segment], ignore_index=True)
        
        # 更新下一轮起始日期，加1天避免重复拉取当前结束日的数据
        current_start = current_end + timedelta(days=1)
    
    # 可选：按日期字段去重，避免接口返回重复数据
    # df_total = df_total.drop_duplicates(subset=["你的日期列名"], keep="last")
    
    # 导出为CSV文件
    df_total.to_csv("完整历史数据.csv", index=False, encoding="utf-8-sig")
    return df_total

# 执行拉取任务
final_df = time_machine()

优化建议 #

• 可在API调用部分增加重试机制，避免网络波动导致任务中断
• 数据量极大的场景下，可每完成几轮拉取就存一次临时CSV，避免程序崩溃丢失已拉取的数据
• 可增加进度提示，比如打印当前拉取的时间区间、总进度百分比，方便监控运行状态

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