如何将多路复用套接字返回的嵌套字典转换为pandas DataFrame

报错原因 #

你触发TypeError: string indices must be integers报错的核心原因是：BinanceSocketManager的recv()方法已经自动将接口返回的JSON字符串解析为Python原生字典结构，无需额外调用json.loads()处理。如果强行对字典执行json.loads()，方法会先将字典转为不符合JSON格式的字符串，后续用键索引字符串时就会触发该类型错误。

完整实现代码 #

代码包含K线字段提取、DataFrame格式化、HDF5存储、PostgreSQL写入全流程：

import asyncio
import pandas as pd
from binance import AsyncClient, BinanceSocketManager
from sqlalchemy import create_engine

# 提前配置PostgreSQL连接（替换为你自己的数据库信息）
pg_engine = create_engine('postgresql://用户名:密码@主机地址:端口/数据库名')
# 提前定义K线字段映射，可根据你的OHLCV表结构调整
KLINE_COLUMNS = [
    'start_time', 'end_time', 'symbol', 'interval', 'first_trade_id', 'last_trade_id',
    'open', 'close', 'high', 'low', 'volume', 'trade_count', 'is_closed',
    'quote_asset_volume', 'taker_buy_base_volume', 'taker_buy_quote_volume', 'ignore'
]

async def main():
    client = await AsyncClient.create()
    bm = BinanceSocketManager(client)
    ms = bm.multiplex_socket(['ltcusdt@kline_1m', 'btcusdt@kline_1m'])
    
    async with ms as mscm:
        while True:
            res = await mscm.recv()
            # 直接提取k字段数据，不需要json.loads
            kline_raw = res['data']['k']
            # 转成单条数据的列表，用于构建DataFrame
            kline_list = [list(kline_raw.values())]
            # 构建DataFrame并转换数值类型
            df = pd.DataFrame(kline_list, columns=KLINE_COLUMNS)
            # 价格、成交量类字段从字符串转浮点型，可根据需求转Decimal提高精度
            df[['open', 'close', 'high', 'low', 'volume', 'quote_asset_volume', 'taker_buy_base_volume', 'taker_buy_quote_volume']] = df[['open', 'close', 'high', 'low', 'volume', 'quote_asset_volume', 'taker_buy_base_volume', 'taker_buy_quote_volume']].astype(float)
            # 时间戳转日期时间格式可选，根据你的存储需求决定
            df['start_time'] = pd.to_datetime(df['start_time'], unit='ms')
            df['end_time'] = pd.to_datetime(df['end_time'], unit='ms')
            
            # 存储到HDF5文件，append模式追加数据
            df.to_hdf('kline_data.h5', key='1m_kline', mode='a', append=True, index=False)
            
            # 写入PostgreSQL的OHLCV表，自动追加
            df.to_sql('ohlcv', pg_engine, if_exists='append', index=False, method='multi')
            
            # 可选：如果只需要存储已收盘的K线，加if判断即可
            # if kline_raw['x']:
            #     执行存储逻辑
            

if __name__ == "__main__":
    loop = asyncio.get_event_loop()
    loop.run_until_complete(main())

注意事项 #

• 运行前需要安装额外依赖：pip install sqlalchemy psycopg2-binary
• PostgreSQL连接字符串需要替换为你自己的数据库配置
• 如果你的OHLCV表字段和上述映射不一致，调整KLINE_COLUMNS列表和DataFrame字段顺序即可
• HDF5存储的key参数可以按交易对拆分，也可以统一用一个key存储全量数据
• 高频率写入时可以攒多条数据批量存储，降低IO开销

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