使用TD Ameritrade API批量获取股票价格历史出现JSONDecodeError如何解决

错误根源 #

TD Ameritrade不同接口的传参规则存在差异：

• marketdata/quotes、instruments接口支持传入逗号分隔的多标的代码，可单次批量查询
• marketdata/pricehistory接口仅支持单标的查询，标的代码需要拼接在请求路径中，不支持在query参数中传入多标的列表。将多标的列表直接传入参数后，接口会返回错误格式的响应（非JSON），因此触发JSONDecodeError报错。

修复方案 #

保留原有的列表分块逻辑，对每个分块内的标的逐个发起单标的请求，再将所有返回结果合并为统一的DataFrame。 #

完整实现代码 #

import pandas as pd
import requests
import time

# 保留原有的列表分块函数
def chunks(l, n):
    """
    :param l: 待拆分的标的列表
    :param n: 每个分块的长度
    """
    n = max(1, n)
    return(l[i:i+n] for i in range(0, len(l), n))

# 单标的5分钟K线查询函数
def get_single_five_min_ph(stock, API_KEY, BASE):
    """
    :param stock: 单个股票代码
    :param API_KEY: TD API密钥
    :param BASE: TD API基础路径
    :returns: 单标的5分钟K线DataFrame
    """
    # 标的代码拼接至请求路径
    url = f"{BASE}marketdata/{stock}/pricehistory"
    params = {
        'apikey': API_KEY,
        'period': '10',
        'periodType': 'day',
        'frequency': '5',
        'frequencyType': 'minute'
    }
    try:
        resp = requests.get(url=url, params=params)
        # 先校验HTTP请求状态，避免解析错误响应
        resp.raise_for_status()
        res_json = resp.json()
        if res_json.get('empty', False):
            return pd.DataFrame()
        # 转换K线数据为DataFrame，新增symbol字段区分标的
        df = pd.DataFrame(res_json['candles'])
        df['symbol'] = stock
        return df
    except Exception as e:
        print(f"查询{stock}失败: {str(e)}")
        return pd.DataFrame()

# 分块批量查询函数
def get_five_minute_ph(stocks_chunk, API_KEY, BASE):
    """
    :param stocks_chunk: 分块后的标的列表
    :param API_KEY: TD API密钥
    :param BASE: TD API基础路径
    :returns: 分块内所有标的5分钟K线合并DataFrame
    """
    all_df = []
    for stock in stocks_chunk:
        single_df = get_single_five_min_ph(stock, API_KEY, BASE)
        if not single_df.empty:
            all_df.append(single_df)
        time.sleep(1)
    return pd.concat(all_df, ignore_index=True) if all_df else pd.DataFrame()

# 调用示例
if __name__ == "__main__":
    BASE = "https://api.tdameritrade.com/v1/"
    API_KEY = "替换为你的TD API密钥"
    # 待查询的标的列表
    target_symbols = ["MSFT", "FB", "AAPL", "GOOG"]
    # 按原逻辑拆分200个为一组
    symbols_chunked = list(chunks(list(set(target_symbols)), 200))
    # 遍历所有分块查询合并结果
    total_df = pd.DataFrame()
    for chunk in symbols_chunked:
        chunk_df = get_five_minute_ph(chunk, API_KEY, BASE)
        total_df = pd.concat([total_df, chunk_df], ignore_index=True)
    print(total_df)

注意事项 #

• 新增HTTP状态校验逻辑，提前捕获权限不足、速率超限、参数错误等问题，避免无效JSON解析报错
• 每条K线新增symbol字段，方便后续多标的数据筛选处理
• 保留1秒请求间隔，避免触发TD API的速率限制规则

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