Comtrade API调用超时求助：Python代码执行反复报错

问题背景 #

编写Python代码调用Comtrade API，逻辑为按年度遍历1-12月数据，每次请求传入5个HS编码，但运行时频繁触发Request failed due to timeout错误，无法正常获取并存储数据。

解决方案 #

1. 增加超时配置与重试机制 #

针对API请求超时，为请求添加明确超时参数，并实现指数退避重试逻辑，处理临时网络波动或API服务繁忙场景。

2. 拆分请求粒度 #

当前单次请求包含12个月+5个HS编码，数据负载过大导致超时。将请求拆分为每月单独请求，同时将每组HS编码数量从5个调整为3个，进一步降低单请求的数据量。

3. 优化数据库批量写入 #

原代码逐条插入数据库，效率极低且占用大量时间。改为批量插入方式，减少数据库连接和IO开销。

4. 添加请求间隔 #

避免短时间内发送大量请求触发API限流，在每次请求后添加固定延迟。

修改后的完整代码 #

import comtradeapicall
import pandas as pd
import mysql.connector
from time import sleep
from tenacity import retry, stop_after_attempt, wait_exponential, retry_if_exception_type
import requests

# 数据库配置
db_config = {
    "host": "localhost",
    "user": "root",
    "password": "",
    "database": "datacomtrade"
}

# 批量写入数据库（优化版）
def save_to_database_and_clean(data):
    if data.empty:
        print("无数据可存储")
        return
    
    conn = mysql.connector.connect(**db_config)
    cursor = conn.cursor()
    table_name = "trade_data"
    
    # 预处理数据：替换NaN、清理特殊字符
    data = data.fillna(0)
    data['reporterDesc'] = data['reporterDesc'].str.replace("'", "", regex=False)
    data['partnerDesc'] = data['partnerDesc'].str.replace("'", "", regex=False)
    data['partner2Desc'] = data['partner2Desc'].str.replace("'", "", regex=False)
    
    # 构造批量插入的占位符
    columns = ','.join(data.columns)
    placeholders = ','.join(['%s'] * len(data.columns))
    insert_query = f"INSERT INTO {table_name} ({columns}) VALUES ({placeholders})"
    
    # 将DataFrame转为元组列表
    data_tuples = [tuple(row) for _, row in data.iterrows()]
    
    try:
        cursor.executemany(insert_query, data_tuples)
        conn.commit()
        print(f"成功存储 {cursor.rowcount} 条数据到数据库")
    except Exception as e:
        conn.rollback()
        print(f"存储数据失败: {str(e)}")
    finally:
        cursor.close()
        conn.close()

subscription_key = 'your_subscription_key'

# 固定参数
type_code = 'C'
freq_code = 'M'
cl_code = 'HS'
flow_code = None
format_output = 'JSON'
aggregate_by = None
breakdown_mode = 'plus'
count_only = None
include_desc = True
max_records = '100000'

# 年份范围
years = range(2022, 2023) 

# 读取HS编码
file_path = r'F:\07 Analisa Pasar\Data\database\hscode6.xlsx'
df = pd.read_excel(file_path, header=None, usecols=[1, 2], skiprows=1)
filtered_df = df[df[1] == 'HS22']
hs_codes = [str(code).zfill(6) for code in filtered_df[2].astype(int).astype(str)]

# 调整每组HS编码数量为3个（减少单请求负载）
grouped_hs_codes = [hs_codes[i:i+3] for i in range(0, len(hs_codes), 3)]

# 带重试和超时的API请求函数
@retry(
    stop=stop_after_attempt(3),  # 最多重试3次
    wait=wait_exponential(multiplier=1, min=2, max=10),  # 指数退避：2s,4s,8s...
    retry=retry_if_exception_type((requests.exceptions.Timeout, requests.exceptions.ConnectionError))
)
def call_api(type_code, freq_code, cl_code, year, month, group):
    period = f'{year}{month:02d}'
    cmd_code = ','.join(group)
    
    # 添加超时参数（根据实际情况调整）
    result = comtradeapicall.getFinalData(
        subscription_key,
        typeCode=type_code,
        freqCode=freq_code,
        clCode=cl_code,
        period=period,
        reporterCode=None,
        cmdCode=cmd_code,
        flowCode=flow_code,
        partnerCode=None,
        partner2Code=None,
        customsCode=None,
        motCode=None,
        maxRecords=max_records,
        format_output=format_output,
        aggregateBy=aggregate_by,
        breakdownMode=breakdown_mode,
        countOnly=count_only,
        includeDesc=include_desc,
        timeout=30  # 设置超时时间
    )
    
    return result

# 执行API请求
def perform_api_call(type_code, freq_code, cl_code, year, month, group, api_call_number):
    print(f"API Call #{api_call_number}: 处理 {year}年{month}月，HS组: {group}")
    try:
        api_result = call_api(type_code, freq_code, cl_code, year, month, group)
        # 过滤world数据
        api_result = api_result[api_result['partnerCode'] != 0]
        save_to_database_and_clean(api_result)
        print(f"API Call #{api_call_number}: 成功获取 {len(api_result)} 条数据")
    except Exception as e:
        print(f"API Call #{api_call_number}: 失败 - {str(e)}")
    # 添加请求间隔，避免限流
    sleep(2)

# 主执行逻辑
api_call_number = 0  
for year in years:
    for month in range(1, 13):
        for group in grouped_hs_codes:
            api_call_number += 1
            perform_api_call(type_code, freq_code, cl_code, year, month, group, api_call_number)

关键优化说明 #

• 重试机制：使用tenacity库实现指数退避重试，针对超时和连接错误自动重试，提升请求成功率。
• 请求拆分：将原12个月的聚合请求拆分为每月单独请求，同时减少每组HS编码数量，降低单请求的数据负载。
• 批量写入：使用executemany批量插入数据库，相比逐条插入效率提升数倍。
• 超时配置：为API请求添加明确的超时时间，避免无限等待。
• 请求间隔：每次请求后延迟2秒，避免触发API的速率限制。

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