3 分钟快速生成代码
输入想法,AI 即刻生成可运行代码
在聚宽(JoinQuant)平台进行量化回测或模拟交易时,止盈止损未成交是导致策略逻辑混乱(如错误删除持仓记录、重复下单)的常见痛点。由于 A 股存在涨跌停限制,当调用 order_target(security, 0) 尝试清仓时,订单可能因为跌停无法撮合而处于挂单状态,甚至被交易所拒绝。
为了避免在订单未成交时就错误地更新了策略内部的持仓状态,必须通过订单状态(OrderStatus)、**持仓可卖数量(closeable_amount)或成交回报(Trade对象)**进行三重确认。以下是具体的避坑指南和标准实现方案。
调用 order_target 后会返回一个 Order 对象。只有当 Order.status 变为 OrderStatus.held(全部成交)时,才代表清仓成功。如果状态是 open(未完成)或 canceled(已撤销)、rejected(已拒绝),则不能更新持仓状态。
handle_data 或收盘后进行状态检查。通过 context.portfolio.positions[security].closeable_amount 检查。若该值大于 0,说明账户中仍有可卖出的持仓,清仓并未成功。只有当 total_amount 或 closeable_amount 确实归零,才能在策略全局变量(如 g.target_positions)中移除该股票。
聚宽系统在每日收盘后(15:00 后)会自动撤销所有未完成的订单。因此,在 after_trading_end 中检查未完成订单并更新策略状态是最佳的“防错兜底”时机。
以下代码展示了如何在盘中记录订单 ID,并在后续交易时间及盘后通过双重机制确认成交,确保持仓记录的准确性。
import jqdata
def initialize(context):
set_benchmark('000300.XSHG')
set_option('use_real_price', True)
# 策略实际持仓记录(避免直接使用 context.portfolio,用 g.my_holdings 维护策略逻辑)
g.my_holdings = set(['000001.XSHE'])
# 记录处于止损清仓中的订单:{security: order_id}
g.pending_stop_orders = {}
def handle_data(context, data):
# 1. 检查之前处于清仓中的订单状态
check_pending_orders(context)
# 2. 模拟触发止损逻辑
for security in list(g.my_holdings):
# 假设触发了止损条件
if trigger_stop_loss(security):
log.info(f"{security} 触发止损,尝试清仓...")
# 如果该标的当前没有正在处理的清仓单,则下单
if security not in g.pending_stop_orders:
# order_target 会自动撤销该标的之前未完成的订单
order_obj = order_target(security, 0)
if order_obj is not None:
# 记录订单 ID,等待成交确认
g.pending_stop_orders[security] = order_obj.order_id
else:
log.warn(f"{security} 下单失败(可能停牌或已无持仓)")
def check_pending_orders(context):
"""盘中实时检查清仓订单状态"""
for security, order_id in list(g.pending_stop_orders.items()):
# 获取当天的订单信息
orders = get_orders(order_id=order_id)
if order_id in orders:
order_obj = orders[order_id]
status_str = str(order_obj.status)
# 情况 A:全部成交 (held)
if 'held' in status_str:
log.info(f"止损单已全部成交: {security},正式从持仓记录中移除。")
if security in g.my_holdings:
g.my_holdings.remove(security)
g.pending_stop_orders.pop(security, None)
# 情况 B:被拒绝 (rejected) 或 已撤销 (canceled)
elif 'rejected' in status_str or 'canceled' in status_str:
log.error(f"止损单未成功,状态为 {status_str}。保留持仓记录,准备下次重新尝试。")
g.pending_stop_orders.pop(security, None)
# 情况 C:部分成交或未成交 (open / filled)
else:
# 检查实际持仓,如果已经通过其他方式清空(例如手动干预)
if context.portfolio.positions[security].total_amount == 0:
if security in g.my_holdings:
g.my_holdings.remove(security)
g.pending_stop_orders.pop(security, None)
def after_trading_end(context):
"""收盘后兜底检查,清理未成交的订单记录"""
log.info("进入收盘后结算,进行持仓状态兜底校验...")
# 此时所有未完成订单已被系统自动撤销
for security in list(g.my_holdings):
# 检查真实持仓数量
real_amount = context.portfolio.positions[security].total_amount
if real_amount == 0:
log.info(f"兜底确认:{security} 真实持仓为0,更新策略持仓记录。")
if security in g.my_holdings:
g.my_holdings.remove(security)
else:
log.warn(f"兜底确认:{security} 仍有真实持仓 {real_amount} 股,可能因跌停未成交。保留持仓记录。")
# 清空当天的订单监控队列
g.pending_stop_orders.clear()
def trigger_stop_loss(security):
# 示例止损判断,实际策略中替换为价格判断
return True
order_target(security, 0) 的下一行代码直接执行 g.my_holdings.remove(security)。必须通过 get_orders() 异步查询订单状态。after_trading_end 兜底机制,可以确保在第二天开盘前,策略持仓状态与账户真实持仓完全一致。get_current_data():在下单前,可以通过 get_current_data()[security].last_price 和 low_limit 检查标的是否已封死跌停。如果已跌停,可选择暂缓下单,避免产生无意义的废单。