Python调用币安API脚本偶发IndexError空DataFrame索引越界报错
问题现象
脚本可正常运行10秒左右后突然终止,抛出如下错误:
IndexError: index 0 is out of bounds for axis 0 with size 0
运行前期可正常返回计算结果、打印Check Pass,多次循环后会返回空DataFrame:
Empty DataFrame Columns: [Close, ST] Index: []
报错栈定位到check函数内访问historicals['ST'].values[0]的代码行,因当前轴长度为0,访问索引0触发越界。
问题复现代码
from binance.client import Client client = Client() coin = "PYRBUSD" ST = 20 LT = 60 def gethistoricals(symbol, LT): df = pd.DataFrame(client.get_historical_klines(symbol, '1m', str(LT) + 'minutes ago UTC')) closes = pd.DataFrame(df[4]) closes.columns = ['Close'] closes['ST'] = closes.Close.rolling(ST).sum() closes.dropna(inplace=True) return closes def check(coin): ma1 = round((float(historicals['ST'].values[0]) / ST), 4) print(ma1) print(type(ma1)) return True while True: historicals = gethistoricals(coin, 20) print(historicals) if check(coin): print("Check Pass")
问题诱因
- 高频请求触发接口限频:代码里是无间隔无限循环请求币安K线接口,短时间请求量超过接口阈值后,部分请求会直接返回空结果,没有任何K线数据。
- 数据量不足触发全空结果:代码里调用
gethistoricals时硬编码传入参数20,只拉取最近20根1分钟K线,做20周期滚动求和后,dropna()会把不足20周期的行全部删掉,只要接口少返回1根K线,最终结果就是空DataFrame;如果接口直接返回空,结果也必然为空。 - 缺少有效性校验:
check函数直接取DataFrame的第0个元素,完全没判断数据是否为空,拿到空数据时直接触发索引越界。 - 隐藏依赖问题:代码里用了pandas但没写导入,只是运行环境里提前装了才没第一时间报导入错误。
- 全局变量滥用:
check函数传了根本没用到的coin参数,直接读取全局的historicals变量,逻辑耦合度高。
修复方案
- 补全缺失的pandas导入,避免依赖问题。
- 给循环加上合理的请求间隔,每次请求后休眠1-2秒,避免触发接口限频。
- 修正参数传递错误,拉取K线时传入预定义的
LT=60参数,多取一部分历史数据,避免数据量刚好卡滚动窗口阈值导致空结果。 - 给接口请求加异常捕获,同时增加空数据判断,拿到空结果或者请求出错时直接跳过本轮计算,等下一轮请求拿到有效数据再继续逻辑,不要强行访问空数据。
- 调整
check函数参数,直接把需要处理的historicals传入,去掉无意义的全局变量依赖。
修复后可稳定运行的完整代码:
import time import pandas as pd from binance.client import Client client = Client() coin = "PYRBUSD" ST = 20 LT = 60 def gethistoricals(symbol, LT): try: klines = client.get_historical_klines(symbol, '1m', f'{LT}minutes ago UTC') except Exception as e: print(f"请求K线接口异常: {str(e)}") return pd.DataFrame(columns=['Close', 'ST']) if not klines: return pd.DataFrame(columns=['Close', 'ST']) df = pd.DataFrame(klines) closes = pd.DataFrame(df[4], dtype=float) closes.columns = ['Close'] closes['ST'] = closes.Close.rolling(ST).sum() closes.dropna(inplace=True) return closes def check(historicals): if len(historicals) == 0: return False ma1 = round(float(historicals['ST'].iloc[0]) / ST, 4) print(ma1) print(type(ma1)) return True while True: historicals = gethistoricals(coin, LT) print(historicals) if check(historicals): print("Check Pass") time.sleep(1.5)
内容的提问来源于stack exchange,提问作者zaster
相关产品推荐
相关产品推荐

