3 分钟快速生成代码
输入想法,AI 即刻生成可运行代码
在 ptrade 量化平台中使用 get_fundamentals 接口获取财务数据时,偶尔出现数据为空的情况,通常是由以下两个原因导致的:
get_fundamentals 接口底层是通过 HTTP 在线获取数据的。在实盘交易或回测过程中,可能会因为网络拥堵、服务器响应延迟等原因导致应答失败,从而返回空数据(如空的 DataFrame 或 None)。
根据 ptrade 官方 API 文档说明,该接口有严格的流量限制:
如果您的策略在短时间内频繁调用该接口,或者一次性查询了大量股票的多个字段,极易触发流控,导致接口拒绝返回数据。
为了保证策略的健壮性,建议在调用 get_fundamentals 时加入重试机制和延时等待(sleep)。
您可以封装一个自定义函数,当获取数据为空或发生异常时,等待一段时间后重新请求:
import time
def get_fundamentals_with_retry(security, table, fields, date=None, max_retries=3):
"""
带重试机制的财务数据获取函数
"""
for i in range(max_retries):
try:
# 调用官方接口
data = get_fundamentals(security, table, fields=fields, date=date)
# 检查数据是否有效 (假设返回的是 DataFrame)
if data is not None and not data.empty:
return data
else:
log.warning("获取财务数据为空,正在进行第 %d 次重试..." % (i + 1))
except Exception as e:
log.error("获取财务数据发生异常: %s" % e)
# 每次重试前休眠 1 秒,避免触发流控
time.sleep(1)
log.error("达到最大重试次数,获取财务数据失败!")
return None
def initialize(context):
g.security = '600570.SS'
set_universe(g.security)
def before_trading_start(context, data):
# 使用封装好的函数获取数据
funda_data = get_fundamentals_with_retry(g.security, 'balance_statement', 'total_assets')
if funda_data is not None:
log.info("成功获取财务数据!")
log.info(funda_data)
else:
log.error("未能获取到财务数据,请检查网络或参数设置。")
def handle_data(context, data):
pass
time.sleep(1)。before_trading_start 或 run_daily 定时任务中获取,并保存到全局变量 g 中供盘中逻辑使用,切勿在分钟级或 tick 级的 handle_data 中频繁调用。