使用Alpha Vantage Python包装器传入PostgreSQL列表时触发无效API调用

问题背景 #

使用Alpha Vantage Python包装器批量拉取股票数据时，直接传入显式定义的股票代码列表（如['X','Y','Z','A','B','C']）能正常运行，代码如下：

stock_symbols = ['X','Y','Z','A','B','C'] 

async def get_data(stock_symbol):
    
    # 创建Alpha Vantage时间序列对象
    ts = TimeSeries(key = config.ALPHAV_KEY, output_format = 'pandas')
    data, _ = await ts.get_intraday(stock_symbol, interval = '5min', outputsize = 'full')
    await ts.close()
    return data

loop = asyncio.get_event_loop()
tasks = [get_data(stock_symbol) for stock_symbol in stock_symbols]
group1 = asyncio.gather(*tasks)
results = loop.run_until_complete(group1)
loop.close()
print(results)

但换成从PostgreSQL查询得到的股票代码列表时，会触发「无效API调用」错误。数据库查询相关代码如下：

# 连接数据库
connection = psycopg2.connect(host = config.DB_HOST, database = config.DB_NAME, user=config.DB_USER, password = config.DB_PASS)

cursor = connection.cursor(cursor_factory=psycopg2.extras.DictCursor)

cursor.execute("SELECT * FROM data WHERE id IN (SELECT product FROM old_products)")

rows = cursor.fetchall()

stock_symbols = []  
for row in rows:
    stock_symbols.append(row['symbol'])

# 后续get_data函数与异步调用逻辑和上面一致

报错回溯信息：

Traceback (most recent call last):
  File "/Users/test/Documents/Programming/VSCode/Ark_Funds_Tracking/pop_prices_EOD.py", line 34, in <module>
    results = loop.run_until_complete(group1)
  File "/Users/test/opt/anaconda3/envs/Py310/lib/python3.10/asyncio/base_events.py", line 649, in run_until_complete
    return future.result()
  File "/Users/test/Documents/Programming/VSCode/Ark_Funds_Tracking/pop_prices_EOD.py", line 27, in get_data
    data, _ = await ts.get_intraday(stock_symbol, interval = '5min', outputsize = 'full')
  File "/Users/test/opt/anaconda3/envs/Py310/lib/python3.10/site-packages/alpha_vantage/async_support/alphavantage.py", line 164, in _format_wrapper
    call_response, data_key, meta_data_key = await func(
  File "/Users/test/opt/anaconda3/envs/Py310/lib/python3.10/site-packages/alpha_vantage/async_support/alphavantage.py", line 106, in _call_wrapper
    return await self._handle_api_call(url), data_key, meta_data_key
  File "/Users/test/opt/anaconda3/envs/Py310/lib/python3.10/site-packages/alpha_vantage/async_support/alphavantage.py", line 254, in _handle_api_call
    raise ValueError(json_response["Error Message"])
ValueError: Invalid API call. Please retry or visit the documentation for TIME_SERIES_INTRADAY.

问题原因与解决方案 #

核心原因 #

从数据库取出的股票代码大概率存在格式问题，和显式定义的干净字符串不一致，常见情况包括：

• 字符串前后包含空格（比如' AAPL '而非'AAPL'）
• 存在不可见字符（如换行符\n、制表符\t）
• 数据库中存储的代码本身无效（拼写错误、退市代码等）

排查与修复步骤 #

1. 验证数据库取出的代码格式
   在构建stock_symbols列表后，添加打印语句检查每个元素的实际内容：

   for sym in stock_symbols:
       print(f"代码内容: '{sym}', 长度: {len(sym)}")
   

   如果长度超出预期，说明存在多余字符。

2. 清洗代码字符串
   对从数据库取出的代码做标准化处理，去除前后空白字符：

   stock_symbols = []  
   for row in rows:
       # 去除前后所有空白字符（空格、换行、制表符等）
       clean_symbol = row['symbol'].strip()
       stock_symbols.append(clean_symbol)
   

3. 过滤无效代码
   添加简单校验逻辑，只保留符合规则的代码，避免无效API调用：

   import re
   stock_symbols = []  
   for row in rows:
       clean_symbol = row['symbol'].strip()
       # 仅保留由大写字母和数字组成的非空代码
       if re.match(r'^[A-Z0-9]+$', clean_symbol):
           stock_symbols.append(clean_symbol)
       else:
           print(f"跳过无效代码: {clean_symbol}")
   

4. 异步调用的额外优化
   Alpha Vantage有API调用频率限制，批量异步调用可能触发限流，建议添加延迟或控制并发数：

   async def get_data(stock_symbol):
       ts = TimeSeries(key = config.ALPHAV_KEY, output_format = 'pandas')
       try:
           data, _ = await ts.get_intraday(stock_symbol, interval = '5min', outputsize = 'full')
           return data
       finally:
           await ts.close()
       # 添加0.5秒延迟，避免触发API限流
       await asyncio.sleep(0.5)
   

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