嘿，作为Python新手能想到用列表来统一管理多个交易所实例，这个思路非常赞！下面我会一步步教你怎么实现，同时解决批量运行函数生成数据列表的问题。

一、批量创建交易所实例列表 #

不用逐个手动实例化每个交易所类，我们可以通过循环+反射的方式批量生成实例，只需要维护一个交易所名称的列表就行，扩展性极强：

import ccxt

# 第一步：定义你需要连接的交易所名称（对应ccxt中的类名，比如ccxt.binance()对应名称'binance'）
target_exchanges = ['binance', 'bitfinex', 'coinbasepro', 'kraken']

# 第二步：循环创建实例并存入列表
exchanges = []
for exchange_name in target_exchanges:
    # 先检查ccxt是否支持该交易所，避免报错
    if hasattr(ccxt, exchange_name):
        # 通过getattr获取对应的交易所类，然后实例化
        exchange_instance = getattr(ccxt, exchange_name)()
        exchanges.append(exchange_instance)
        print(f"成功初始化交易所：{exchange_instance.name}")
    else:
        print(f"⚠️ CCXT暂不支持交易所：{exchange_name}")

# 现在就可以通过索引访问了
print("\n第一个交易所名称：", exchanges[0].name)
print("第二个交易所名称：", exchanges[1].name)

补充：需要API密钥的情况 #

如果某些交易所需要API密钥才能访问更多数据（比如私有账户数据），实例化时可以传入配置参数：

exchange_instance = getattr(ccxt, exchange_name)({
    'apiKey': '你的API_KEY',
    'secret': '你的API_SECRET',
    'enableRateLimit': True,  # 建议开启，避免触发交易所限流
})

二、批量运行函数，生成OHLC数据列表 #

有了交易所实例列表后，我们可以通过循环或者列表推导式批量调用fetch_ohlcv方法，同时处理可能的异常（比如网络问题、交易所不支持该交易对等）：

# 定义要获取的交易对、时间周期和K线数量
symbol = 'BTC/USDT'
timeframe = '1h'  # 可选值：'1m', '5m', '1h', '1d'等，具体看交易所支持
limit = 100  # 获取最近100条K线

# 批量获取OHLC数据并存入列表
ohlc_results = []
for exchange in exchanges:
    try:
        # 有些交易所需要先加载市场数据才能正常获取K线
        exchange.load_markets()
        # 调用fetch_ohlcv获取数据
        ohlc_data = exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
        # 将结果整理成字典存入列表，方便后续处理
        ohlc_results.append({
            'exchange': exchange.name,
            'symbol': symbol,
            'timeframe': timeframe,
            'ohlc': ohlc_data
        })
        print(f"✅ 已获取{exchange.name}的{symbol} {timeframe} K线数据")
    except Exception as e:
        # 捕获异常并记录错误信息，避免程序中断
        error_info = {
            'exchange': exchange.name,
            'error': str(e)
        }
        ohlc_results.append(error_info)
        print(f"❌ {exchange.name}获取数据失败：{str(e)}")

# 查看第一个交易所的K线数据（前2条）
print("\n", ohlc_results[0]['exchange'], "的K线数据示例：")
for candle in ohlc_results[0]['ohlc'][:2]:
    print(f"时间戳：{candle[0]}, 开盘价：{candle[1]}, 最高价：{candle[2]}, 最低价：{candle[3]}, 收盘价：{candle[4]}, 成交量：{candle[5]}")

进阶优化：加速获取数据 #

如果交易所数量较多，同步获取会比较慢，你可以用多线程来并行请求，比如使用concurrent.futures.ThreadPoolExecutor：

from concurrent.futures import ThreadPoolExecutor

def fetch_ohlc(exchange):
    try:
        exchange.load_markets()
        return {
            'exchange': exchange.name,
            'symbol': symbol,
            'ohlc': exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
        }
    except Exception as e:
        return {'exchange': exchange.name, 'error': str(e)}

# 用线程池并行执行
with ThreadPoolExecutor(max_workers=5) as executor:
    ohlc_results = list(executor.map(fetch_ohlc, exchanges))

# 结果和之前的格式一致
print(ohlc_results)

小贴士 #

• 记得开启enableRateLimit: True，这是CCXT推荐的设置，可以自动处理交易所的请求频率限制，避免被封禁。
• 不同交易所支持的交易对、时间周期可能不同，建议先通过exchange.markets查看该交易所支持的市场信息。
• 如果需要处理更多交易所，可以直接修改target_exchanges列表，无需修改大量实例化代码。

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