3 分钟快速生成代码
输入想法,AI 即刻生成可运行代码
在量化交易中,**ARBR(人气意愿指标)**是衡量市场买卖双方人气与意愿的重要技术因子。在聚宽(JoinQuant)平台上构建 ARBR 选股策略时,为了贴近实盘,我们需要解决三个核心痛点:
avoid_future_data 保护。run_weekly 定时器降低交易频率。initialize 中调用 set_option('avoid_future_data', True)。开启后,若策略试图获取当前时间之后的未来数据,系统将直接抛出 FutureDataError 异常。run_weekly(my_rebalance, weekday=1, time='09:30'),设定在每周第一个交易日开盘时执行调仓逻辑。get_factor_values 获取全市场股票的 ARBR 因子值(ARBR 为因子 AR 与因子 BR 的差)。get_current_data() 获取实时盘口数据:
current_data[security].last_price >= current_data[security].high_limithandle_data(分钟级)中监控,若持仓股曾触及涨停但最新价回落,则执行卖出。# 导入聚宽函数库
from jqdata import *
from jqfactor import get_factor_values
import numpy as np
import pandas as pd
def initialize(context):
# 1. 开启避免未来数据模式(强烈推荐)
set_option("avoid_future_data", True)
# 2. 设置真实价格回测
set_option('use_real_price', True)
# 3. 设定沪深300作为基准
set_benchmark('000300.XSHG')
# 设定交易税费
set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
# 定义全局变量
g.stock_num = 10 # 每次持股数量
g.factor_name = 'ARBR' # 使用ARBR因子
# 4. 注册周度调仓函数:每周一开盘运行
run_weekly(my_rebalance, weekday=1, time='09:30')
def my_rebalance(context):
"""周度调仓主逻辑"""
# 获取全市场可交易股票池(过滤退市、ST、新股)
all_stocks = get_all_securities(['stock'], context.current_dt.date())
all_stocks = all_stocks[all_stocks['end_date'] > context.current_dt.date()].index.tolist()
# 过滤掉ST股与上市未满一年的新股
current_data = get_current_data()
all_stocks = [s for s in all_stocks if not current_data[s].is_st
and (context.current_dt.date() - get_security_info(s).start_date).days > 365]
# 获取前一交易日的 ARBR 因子值
prev_date = context.previous_date
factor_data = get_factor_values(
securities=all_stocks,
factors=[g.factor_name],
start_date=prev_date,
end_date=prev_date
)[g.factor_name]
# 提取因子值并排序(假设ARBR值越低,反弹意愿越强,选择因子值最小的前N只)
factor_series = factor_data.loc[pd.to_datetime(prev_date)].dropna()
target_stocks = factor_series.nsmallest(g.stock_num).index.tolist()
# 获取当前持仓
current_holdings = list(context.portfolio.positions.keys())
# 调仓卖出逻辑(引入涨停保护)
for stock in current_holdings:
if stock not in target_stocks:
# 获取当前股票的实时盘口数据
stock_data = current_data[stock]
# 涨停保护:如果当前股票依然封死涨停,则不卖出,保留持仓
if stock_data.last_price >= stock_data.high_limit and stock_data.last_price > 0:
log.info(f"持仓股 {stock} 当前涨停,触发涨停保护,暂不卖出。")
continue
else:
order_target(stock, 0)
log.info(f"调仓卖出:{stock}")
# 调仓买入逻辑
current_holdings = list(context.portfolio.positions.keys()) # 更新卖出后的持仓
buy_list = [s for s in target_stocks if s not in current_holdings]
if len(buy_list) > 0:
# 均分可用资金买入新标的
available_cash = context.portfolio.available_cash
value_per_stock = available_cash / len(buy_list)
for stock in buy_list:
# 过滤停牌与开盘即涨停无法买入的股票
stock_data = current_data[stock]
if not stock_data.paused and stock_data.last_price < stock_data.high_limit:
order_target_value(stock, value_per_stock)
log.info(f"调仓买入:{stock},分配金额:{value_per_stock:.2f}")
def handle_data(context, data):
"""分钟级运行,处理炸板卖出逻辑"""
current_data = get_current_data()
current_holdings = list(context.portfolio.positions.keys())
for stock in current_holdings:
stock_data = current_data[stock]
# 炸板卖出逻辑:
# 如果股票盘中曾触及涨停(high_limit),但当前价格跌破了涨停价(例如跌幅超过涨停价的1%),则判定为炸板,立即市价平仓
if stock_data.last_price > 0:
# 检查今日最高价是否曾达到涨停价
# 注意:由于 avoid_future_data 开启,我们只能通过 data[stock].high 获取今日截止当前的最高价
today_high = data[stock].high
if today_high >= stock_data.high_limit:
# 如果当前价格相比涨停价回落超过 1.5%,则执行炸板卖出
if stock_data.last_price < stock_data.high_limit * 0.985:
order_target(stock, 0)
log.warn(f"警报:持仓股 {stock} 盘中炸板!当前价 {stock_data.last_price} 跌破涨停价 {stock_data.high_limit},立即止盈卖出。")
avoid_future_data 的安全边界:my_rebalance 中,我们获取因子值时使用的是 context.previous_date(前一交易日),这完全符合实盘逻辑。在 handle_data 中,我们使用 data[stock].high 获取今日截止到当前分钟的最高价,而不是使用日频的未来数据,确保回测结果真实可靠。handle_data 中实时监控,一旦发现个股曾触及涨停但最新价回落超过 1.5%,便会触发市价单强制平仓,落袋为安。