前文介绍了PTrade提供的财务数据获取函数get_fundamentals，以及该函数在使用中的问题。本文尝试对该函数进行包装改造。
需求描述
获取上市公司最新发布的财务数据。
源码
def get_new_fundamentals(security, table, fields, num_per_query=500, attempts=5):
"""
查询最新财务数据
ptrade原有get_fundamentals函数，当上一季季报未公布时会返回NaN，此函数遇到这类情况时，会返回上上季度的数据
:param security: 待查询标的列表
:param table: 财务数据表名
:param fields: 数据结果集中所需输出业务字段
:param num_per_query: 单次查询的标的数目，get_fundamentals有流量限制，每秒不得调用超过100次，单次最大调用量是500条数据
:parma attempts: 重复尝试次数，用于处理因网络拥堵等原因导致应答失败的情况
:return: 查询到的财务数据，查询成功返回类型为DataFrame，查询失败则返回None
"""
ret_data = None
# 分批次查询财务数据
for i in range(0, len(security), num_per_query):
data = None
# 待查询标的列表
query_list = security[i: i+num_per_query]
# 分三次查询财务数据，查询日期依次为当前一日、上一季度的最后一日、上上季度的最后一日
for date_index in range(3):
# 先查当日的财务数据
if 0 == date_index:
date = get_trading_day(-1).strftime('%Y%m%d')
# 查询上、上上季度的财务数据
else:
date = last_day_of_quarter_before_n_quarters(date_index)
# 多次尝试获取财务数据
for attempt in range(attempts):
# 调用ptrade接口查询财务数据
data = get_fundamentals(security=query_list, table=table, fields=fields, date=date)
# 成功获取数据则跳出
if data is not None:
break
# 获取数据失败则跳出，继续查询下一批次数据。
if data is None:
log.info('获取财务数据失败')
continue
# 合并已查询成功的数据
try:
ret_data = pd.concat([ret_data, data[data[fields].notnull().all(axis=1)]])
except Exception as e:
log.info('合并财务数据异常:{}'.format(e))
# 获取未查询成功的标的
try:
query_list = data[data[fields].isnull().any(axis=1)].index.tolist()
except Exception as e:
log.info('获取未查询成功标的异常:{}'.format(e))
# 所有数据都有查询结果则进入下一批标的的查询
if not query_list:
break
return ret_data
def last_day_of_quarter_before_n_quarters(n):
"""
获取当日n个季度前的日期所处季度的最后一天
:return: 当日n个季度前的日期所处季度的最后一天，格式为字符串
"""
# 当前日期
current_date = get_trading_day()
# 当前季度
current_quarter = (current_date.month - 1) // 3 + 1
# 目标日期总季度数，用于计算目标日期
total_quarters = current_date.year * 4 + current_quarter - 1 - (n - 1)
# 当前季度的n-1个季度前的日期季度
quarter = total_quarters % 4 + 1
# 当前季度的n-1个季度前的日期年份
year = total_quarters // 4
# 当前季度的n-1个季度前的季度的第一个月的月份
month = (quarter - 1) * 3 + 1
# 当前季度的n-1个季度前的第一个月份减去一天，得到n个季度前的日期所处季度的最后一天，并转换为字符串格式
return (dt.datetime(year, month, 1) - dt.timedelta(days=1)).strftime('%Y%m%d')
源码分析
源码部分有两个函数，第1个函数为改造后的财务数据获取函数get_new_fundamentals，第2个函数为日期获取函数，在前文中进行了介绍。本文重点介绍get_new_fundamentals函数。
def get_new_fundamentals(security, table, fields, num_per_query=500, attempts=5):
get_new_fundamentals函数的参数：
– security：多只股票代码组成的list
– table：财务数据表名，表名可为以下内容：
| 表名 | 包含内容 |
|---|---|
| valuation | 估值数据 |
| balance_statement | 资产负债表 |
| income_statement | 利润表 |
| cashflow_statement | 现金流量表 |
| growth_ability | 成长能力指标 |
| profit_ability | 盈利能力指标 |
| eps | 每股指标 |
| operating_ability | 营运能力指标 |
| debt_paying_ability | 偿债能力指标 |
- fields: 指明数据结果集中所需输出业务字段，输出具体字段请参考PTrade。
- param num_per_query: 单次查询的标的数目，get_fundamentals有流量限制，每秒不得调用超过100次，单次最大调用量是500条数据。
- attempts: 重复尝试次数，用于处理因网络拥堵等原因导致应答失败的情况。
ret_data = None
首先定义返回值为None，如果后面函数查询成功将返回类型为DataFrame的财务数据，查询失败则返回None。
for i in range(0, len(security), num_per_query):
PTrade的get_fundamentals函数有流量限制，每秒不得调用超过100次，单次最大调用量是500条数据，每一条数据的定义为：一个股票对应一个表的一个字段，相当于最大不超过5万条。
因此这里我们分批次查询数据，单次查询num_per_query条数据，默认为500。
data = None
定义一个临时变量，用于临时存储查询到的财务数据。
query_list = security[i: i+num_per_query]
将待查询的标的列表保存在query_list中。
for date_index in range(3):
根据前文的分析，get_fundamentals返回的是上一季度的财务数据，如果上一季度财报未发布，则返回空。
为了确保查询出有效数据，这里分三次查询财务数据，查询日期依次为前一日、上一季度的最后一日、上上季度的最后一日。
# 先查当日的财务数据
if 0 == date_index:
date = get_trading_day(-1).strftime('%Y%m%d')
# 查询上、上上季度的财务数据
循环内第1次查询，查询日期设置为前一日。这里不使用当日而使用前一日的原因是，在实盘过程中，发现如果使用当日去查询valuation表时，会反馈空数据，而在回测过程则是返回前一日的数据。所以这里要使用前一日去查询，避免出现反馈空数据后，直接去查询上一季度的最后一日的财务数据，并且与回测保持一致。
else:
date = last_day_of_quarter_before_n_quarters(date_index)
循环内第2、3次查询，查询日期分别设置为上一季度的最后一日、上上季度的最后一日。
# 多次尝试获取财务数据
for attempt in range(attempts):
由于get_fundamentals函数为http在线获取，会存在因网络拥堵等原因导致应答失败的情况，如果返回数据结果为空请多次尝试，策略中需增加保护机制。这里采取多次请求的方式，来降低访问失败的概率。
data = get_fundamentals(security=query_list, table=table, fields=fields, date=date)
调用ptrade接口查询财务数据。
if data is not None:
break
如果成功获取数据则跳出。
if data is None:
log.info('获取财务数据失败')
continue
如果获取数据失败则跳出，继续查询下一批次数据。
# 合并已查询成功的数据
try:
ret_data = pd.concat([ret_data, data[data[fields].notnull().all(axis=1)]])
except Exception as e:
log.info('合并财务数据异常:{}'.format(e))
如果查询数据成功，则把分批次查询到的数据进行合并。
try:
query_list = data[data[fields].isnull().any(axis=1)].index.tolist()
except Exception as e:
log.info('获取未查询成功标的异常:{}'.format(e))
获取未查询成功的标的，然后用下一个日期（上一季度的最后一日、上上季度的最后一日）再次查询。
if not query_list:
break
如果所有标的都有查询结果，则进入下一批标的的查询。
return ret_data
最后返回所有查询到的结果。
示例
import datetime as dt
import pandas as pd
def before_trading_start(context, data):
share_list = get_Ashares()
print(len(share_list))
ret = get_new_fundamentals(share_list, 'growth_ability', ['operating_revenue_grow_rate', 'net_profit_grow_rate'])
print(len(ret))
print(ret)
def initialize(context):
pass
def handle_data(context, data):
pass
这里在before_trading_start里调用get_new_fundamentals，以回测日期2023-09-28为例，系统输出：
2023-10-07 19:02:21 开始运行回测, 策略名称: get_new_fundamentals
2023-09-28 08:30:00 - INFO - 5065
2023-09-28 08:30:00 - WARNING - 获取GTN数据为空!
2023-09-28 08:30:00 - WARNING - 获取GTN数据为空!
2023-09-28 08:30:00 - INFO - 5061
2023-09-28 08:30:00 - INFO - secu_abbr publ_date net_profit_grow_rate \
secu_code
000001.SZ 平安银行 2023-08-24 14.9357
000002.SZ 万科A 2023-08-31 -16.3406
000004.SZ 国华网安 2023-08-24 21.3409
000005.SZ ST星源 2023-08-30 -195.5156
000006.SZ 深振业A 2023-08-25 -97.1310
... ... ... ...
688798.SS 艾为电子 2023-08-22 -153.5513
688799.SS 华纳药厂 2023-08-28 26.5001
688800.SS 瑞可达 2023-08-31 -47.1334
688819.SS 天能股份 2023-08-29 42.4311
688981.SS 中芯国际 2023-08-26 -52.7844
operating_revenue_grow_rate end_date
secu_code
000001.SZ -3.7078 2023-06-30
000002.SZ -2.9112 2023-06-30
000004.SZ 41.0661 2023-06-30
000005.SZ -11.6666 2023-06-30
000006.SZ -57.4572 2023-06-30
... ... ...
688798.SS -22.3381 2023-06-30
688799.SS 21.7366 2023-06-30
688800.SS -11.7285 2023-06-30
688819.SS 25.1396 2023-06-30
688981.SS -13.3150 2023-06-30
[5061 rows x 5 columns]
2023-10-07 19:02:40 策略回测结束
这里有几个地方值得说明一下：
- 当日所有A股的数量为5065，最后查询到财务数据的只数为5061。我对差额的4只进行了单独的查询，确认是PTrade没有这4只股票近期这几个查询字段的财务数据。
- 分别用国盛的PTrade和湘财的PTrade运行了上面的代码，国盛显示的A股数量为5065，湘财显示的是5066，略有不同。两者都是有4只股票没有查询到财务数据，且4只股票相同。
- 回测过程中会出现“WARNING – 获取GTN数据为空!”，券商反馈是接口访问量的限制，实盘时可以考虑先把数据离线下到本地，实盘交易时再从本地读取。
小结
用改造后的get_new_fundamentals函数，就可以获取上市公司最新发布的财务数据，该函数可应用于基于财务数据的策略。PTrade用于回测的财务数据会有少量缺失，对一般策略的回测影响不大。
博客内容只用于交流学习，不构成投资建议，盈亏自负！
个人博客：http://coderx.com.cn/（优先更新）
项目最新代码：https://gitee.com/sl/quant_from_scratch
欢迎大家转发、留言。有微信群用于学习交流，感兴趣的读者请扫码加微信！
如果认为博客对您有帮助，可以扫码进行捐赠，感谢！
| 微信二维码 | 微信捐赠二维码 |
|---|---|
 |
 |
