问题描述
我现在模拟盘过滤像下面这样。想在研究环境跑个横截面测试,结果研究环境没有get_current_data()。
请问研究环境怎么达成等效的过滤?
#过滤st等
def filter_stock(stock_list):
curr_data = get_current_data()
stock_list = [stock for stock in stock_list if \
(not curr_data[stock].paused) # 未停牌
and (not curr_data[stock].is_st) # 非ST
and ('ST' not in curr_data[stock].name)
and ('*' not in curr_data[stock].name)
and ('退' not in curr_data[stock].name)
and (curr_data[stock].low_limit < curr_data[stock].day_open < curr_data[stock].high_limit) # 未涨跌停
and not stock.startswith(('3', '68', '8', '4'))]
return stock_list
解决方案
在 JoinQuant 的研究环境中,get_current_data() 确实是回测和模拟盘专用的 API,因为它依赖于回测引擎的时间上下文。要在研究环境中进行横截面测试并等效地过滤股票,你需要显式地指定一个查询日期 (check_date),并组合使用 get_all_securities、get_extras 和 get_price 这几个 API。
以下是各个过滤条件的等效替代方案:
- 股票名称过滤(ST、*、退):使用
get_all_securities(date=check_date)获取当天的股票基本信息表,读取display_name字段。 - 是否 ST 过滤:使用
get_extras('is_st', stock_list, start_date=check_date, end_date=check_date)。 - 停牌与涨跌停过滤:使用
get_price(stock_list, end_date=check_date, count=1, fields=['paused', 'open', 'high_limit', 'low_limit'])获取当天的行情切片。 - 板块过滤:直接通过字符串切片判断,与环境无关。
研究环境等效过滤代码
你可以将以下代码直接复制到研究环境的 Jupyter Notebook 中运行:
from jqdata import *
import pandas as pd
def filter_stock_research(stock_list, check_date):
"""
研究环境专用的股票过滤函数
:param stock_list: 待过滤的股票代码列表
:param check_date: 截面测试的日期,例如 '2023-10-10'
:return: 过滤后的股票代码列表
"""
# 1. 过滤特定板块 (创业板3, 科创板68, 北交所8/4)
stock_list = [s for s in stock_list if not s.startswith(('3', '68', '8', '4'))]
if not stock_list:
return []
# 2. 获取股票名称,过滤名称中包含 ST、*、退 的股票
securities_info = get_all_securities(types=['stock'], date=check_date)
valid_stocks = [s for s in stock_list if s in securities_info.index]
name_filtered_stocks = []
for s in valid_stocks:
name = securities_info.loc[s, 'display_name']
if 'ST' not in name and '*' not in name and '退' not in name:
name_filtered_stocks.append(s)
if not name_filtered_stocks:
return []
# 3. 获取 is_st 数据过滤 ST 股
st_df = get_extras('is_st', name_filtered_stocks, start_date=check_date, end_date=check_date)
if not st_df.empty:
is_st_series = st_df.iloc[0]
non_st_stocks = is_st_series[is_st_series == False].index.tolist()
else:
non_st_stocks = name_filtered_stocks
if not non_st_stocks:
return []
# 4. 获取行情数据:停牌、开盘价、涨跌停价
# 注意:研究环境中获取当天数据,如果是盘中可能取不到完整的 day_open,通常横截面测试取历史某一天
price_df = get_price(non_st_stocks, end_date=check_date, count=1,
fields=['paused', 'open', 'high_limit', 'low_limit'], panel=False)
final_stocks = []
# get_price 返回的 DataFrame 索引是日期,列是股票代码(当查询单个 field 时)
# 为了方便处理,我们分别提取各个 field 的 Series
df_paused = price_df['paused'].iloc[0]
df_open = price_df['open'].iloc[0]
df_high_limit = price_df['high_limit'].iloc[0]
df_low_limit = price_df['low_limit'].iloc[0]
for s in non_st_stocks:
try:
is_paused = df_paused[s]
open_price = df_open[s]
high_limit = df_high_limit[s]
low_limit = df_low_limit[s]
# 逻辑:未停牌 且 (跌停价 < 开盘价 < 涨停价)
# 注意处理 NaN 值的情况(例如刚上市或退市无数据)
if pd.notna(is_paused) and (not is_paused):
if pd.notna(open_price) and pd.notna(high_limit) and pd.notna(low_limit):
if low_limit < open_price < high_limit:
final_stocks.append(s)
except KeyError:
continue
return final_stocks
# 测试示例
# check_date = '2023-01-05'
# raw_list = get_index_stocks('000300.XSHG', date=check_date)
# filtered_list = filter_stock_research(raw_list, check_date)
# print(f"原始数量: {len(raw_list)}, 过滤后数量: {len(filtered_list)}")
核心差异说明
- 时间上下文:回测引擎会自动维护
context.current_dt,所以get_current_data()知道“现在”是哪一天。在研究环境中,你必须手动传入check_date。 - 批量获取 vs 逐个获取:
get_current_data()返回的是一个字典,支持按单只股票查询属性。而在研究环境中,使用get_price和get_extras批量获取多只股票的数据并利用 Pandas 进行向量化或遍历处理,执行效率会高得多。