问题描述
解决方案
在PTrade中,排除上市不足90天的次新股主要依赖于 get_stock_info 接口获取股票的上市日期(listed_date),然后与当前回测或交易日期进行比较。
以下是一个完整的策略示例,展示了如何在每日盘前(before_trading_start)获取股票池并剔除上市不满90天的股票。
策略逻辑说明
- 获取股票池:使用
get_index_stocks获取某个指数(如沪深300)的成分股,或者使用get_Ashares获取全市场股票。 - 获取上市日期:调用
get_stock_info获取股票的listed_date。 - 计算天数差:将当前日期与上市日期进行相减,判断天数是否大于等于90天。
- 设置股票池:将筛选后的股票列表通过
set_universe设置为当天的交易股票池。
策略代码实现
import datetime
def initialize(context):
# 初始化时设置一个默认的股票池,防止报错,实际逻辑在before_trading_start中更新
g.security = ['600570.SS']
set_universe(g.security)
# 设定上市时间阈值(天)
g.listed_days_threshold = 90
def before_trading_start(context, data):
# 1. 获取初始股票池,这里以沪深300为例
# 如果需要全市场,可以使用 get_Ashares()
initial_stocks = get_index_stocks('000300.XBHS')
# 2. 调用过滤函数,剔除次新股
filtered_stocks = filter_new_stocks(context, initial_stocks, g.listed_days_threshold)
# 3. 更新全局变量和股票池
g.security = filtered_stocks
set_universe(g.security)
log.info("初始股票数量: %d, 剔除上市不满%d天股票后数量: %d" % (len(initial_stocks), g.listed_days_threshold, len(filtered_stocks)))
def handle_data(context, data):
# 这里编写具体的买卖逻辑
# 示例:买入股票池中的第一只股票
if len(g.security) > 0:
target_stock = g.security[0]
# 简单的买入示例
order_target_value(target_stock, 10000)
def filter_new_stocks(context, stock_list, days):
"""
过滤上市时间不足指定天数的股票
:param context: 上下文对象
:param stock_list: 待过滤的股票代码列表
:param days: 最小上市天数要求
:return: 过滤后的股票列表
"""
filtered_list = []
# 获取当前回测/交易日期的 datetime.date 对象
current_date = context.blotter.current_dt.date()
# 批量获取股票的基础信息,只获取 'listed_date' 字段以提高效率
# 注意:get_stock_info 返回的是字典 {code: {field: value}}
infos = get_stock_info(stock_list, field=['listed_date'])
for stock in stock_list:
# 获取该股票的信息
info = infos.get(stock)
# 确保获取到了信息且包含上市日期
if info and 'listed_date' in info:
listed_date_str = info['listed_date']
# 某些异常情况下上市日期可能为空,需做判断
if not listed_date_str or listed_date_str == '0':
continue
try:
# 将字符串转换为 date 对象
listed_date = datetime.datetime.strptime(listed_date_str, "%Y-%m-%d").date()
# 计算上市天数
delta = (current_date - listed_date).days
# 如果上市天数大于等于阈值,则保留
if delta >= days:
filtered_list.append(stock)
except Exception as e:
# 防止日期格式解析错误导致策略中断
log.error("解析股票 %s 上市日期失败: %s" % (stock, e))
continue
return filtered_list
代码关键点解析
-
get_stock_info的使用:- 我们显式传入
field=['listed_date'],这样可以减少数据传输量,提高回测速度。 - 返回的数据结构是嵌套字典,例如:
{'600570.SS': {'listed_date': '2003-12-16'}}。
- 我们显式传入
-
日期处理:
context.blotter.current_dt是datetime.datetime类型,包含时分秒。listed_date是字符串类型(如'2023-01-01')。- 代码中使用
strptime将字符串转为日期对象,并统一使用.date()获取纯日期部分进行减法运算,得出相差的天数。
-
异常处理:
- 增加了
try-except块和空值检查,防止因为个别股票数据缺失导致整个策略报错停止。
- 增加了
-
兼容性:
- 代码遵循 Python 3.5 语法规范(不使用 f-string),确保在 PTrade 所有环境中均可运行。
Q&A
Q: 为什么在 before_trading_start 中进行过滤而不是 initialize?
A: initialize 只在策略启动时运行一次。如果在 initialize 中过滤,随着回测时间的推移,原本不满90天的股票可能会满足条件,但因为没有重新过滤,它们永远不会被加入股票池。在 before_trading_start 中过滤可以确保每天都根据当天的日期动态更新股票池。
Q: get_stock_info 获取不到数据怎么办?
A: 如果 get_stock_info 返回 None 或者字典中没有 listed_date,代码中的 if info and 'listed_date' in info: 判断会将其跳过,这意味着数据缺失的股票会被默认剔除,这是一种安全的处理方式。
Q: 如何将此逻辑应用到全市场股票?
A: 将代码中的 get_index_stocks('000300.XBHS') 替换为 get_Ashares() 即可获取当天所有A股代码进行过滤。但请注意,全市场股票数量较多,循环处理可能会增加每日盘前的耗时。