JoinQuant 策略进阶：如何用状态机模式避免重复下单与逻辑混乱？

在量化交易中，状态机（Finite State Machine, FSM） 是一种非常有效的设计模式，用于管理复杂的交易逻辑。它可以清晰地定义策略在不同阶段的行为（如空仓、持仓、止损中等），并严格控制状态之间的流转，从而从逻辑根源上避免重复下单、信号闪烁等问题。

以下是关于如何在 JoinQuant 平台设计策略状态机的详细指南和代码实现。

核心设计理念

一个基础的交易状态机通常包含以下要素：

1. 状态（States）：策略当前所处的环境。
   • EMPTY (空仓/观望)：无持仓，等待买入信号。
   • HOLDING (持仓)：持有股票，等待卖出或止损信号。
   • (进阶) STOP_LOSS (止损冷却)：刚触发止损，暂停交易一段时间。
2. 事件（Events）：触发状态变化的条件。
   • 金叉/死叉、价格突破、止损触发等。
   • 关键点：只有在特定的状态下，特定的事件才有效。例如，只有在 EMPTY 状态下，“金叉”信号才会触发买入；如果在 HOLDING 状态下再次出现金叉，状态机将忽略它，从而避免重复买入。
3. 动作（Actions）：状态流转时执行的操作。
   • 买入下单、卖出下单、记录日志等。

策略代码实现

下面的代码展示了一个基于 双均线策略 的状态机实现。它通过全局变量 g.state 来管理当前状态，确保不会在持仓时重复买入，也不会在空仓时重复卖出。

# -*- coding: utf-8 -*-
import jqdata

# 定义状态常量
class State:
    EMPTY = 'EMPTY'      # 空仓状态
    HOLDING = 'HOLDING'  # 持仓状态

def initialize(context):
    # 1. 设定基准
    set_benchmark('000300.XSHG')
    # 2. 开启动态复权模式(真实价格)
    set_option('use_real_price', True)
    # 3. 设置手续费 (股票: 买入万三，卖出万三+千一印花税)
    set_order_cost(OrderCost(close_tax=0.001, open_commission=0.0003, close_commission=0.0003, min_commission=5), type='stock')
    
    # 4. 初始化策略变量
    g.security = '000001.XSHE' # 操作标的：平安银行
    
    # --- 状态机初始化 ---
    # 在 g 对象中维护状态，确保回测和模拟重启后状态不丢失
    g.state = State.EMPTY 
    
    # 运行频率：每天开盘运行
    run_daily(market_open, time='every_bar')

def market_open(context):
    """
    核心交易逻辑：基于状态机的流转
    """
    security = g.security
    
    # 获取均线数据 (5日线和10日线)
    close_data = attribute_history(security, 11, '1d', ['close'])
    ma5 = close_data['close'][-5:].mean()
    ma10 = close_data['close'][-10:].mean()
    
    # 获取当前价格
    current_price = close_data['close'][-1]
    
    # --- 状态机逻辑开始 ---
    
    # 场景 1: 当前是 [空仓状态]
    if g.state == State.EMPTY:
        # 检查买入条件：5日线上穿10日线 (金叉)
        if ma5 > ma10:
            log.info("信号触发：金叉 -> 准备买入")
            
            # 执行动作：全仓买入
            cash = context.portfolio.available_cash
            if cash > 0:
                order_value(security, cash)
                
                # 状态流转：切换到 [持仓状态]
                g.state = State.HOLDING
                log.info(f"状态流转: EMPTY -> HOLDING, 买入价格: {current_price}")
            else:
                log.warn("资金不足，无法买入")

    # 场景 2: 当前是 [持仓状态]
    elif g.state == State.HOLDING:
        # 检查卖出条件：5日线下穿10日线 (死叉)
        # 或者 检查止损条件：亏损超过 5% (示例)
        
        # 获取持仓成本用于计算止损
        position = context.portfolio.positions[security]
        if position.total_amount == 0:
            # 容错处理：如果账户里实际没货了（可能被手动卖了），强制重置状态
            g.state = State.EMPTY
            return

        cost = position.avg_cost
        is_dead_cross = ma5 < ma10
        is_stop_loss = current_price < cost * 0.95
        
        if is_dead_cross or is_stop_loss:
            reason = "死叉" if is_dead_cross else "止损"
            log.info(f"信号触发：{reason} -> 准备卖出")
            
            # 执行动作：清仓
            order_target(security, 0)
            
            # 状态流转：切换到 [空仓状态]
            g.state = State.EMPTY
            log.info(f"状态流转: HOLDING -> EMPTY, 卖出价格: {current_price}")
            
    # --- 状态机逻辑结束 ---
    
    # 记录当前状态用于调试
    record(price=current_price, ma5=ma5, ma10=ma10)

def after_code_changed(context):
    """
    模拟交易中更新代码后执行，防止状态丢失或重置错误
    """
    # 如果 g.state 不存在（比如第一次增加状态机逻辑），则初始化
    if not hasattr(g, 'state'):
        g.state = State.EMPTY
    log.info(f"策略代码更新，当前状态维持为: {g.state}")

代码解析与优势

1. 避免重复下单 (幂等性)：

   • 在 if g.state == State.HOLDING: 分支中，我们只检查卖出逻辑。这意味着，即使市场一直处于“金叉”状态（ma5 > ma10 持续为真），只要状态已经是 HOLDING，代码根本不会进入买入判断逻辑。
   • 这比简单的 if ma5 > ma10: order(...) 要安全得多，后者在每一分钟或每一天都会尝试买入，直到现金耗尽。

2. 逻辑解耦：

   • 买入逻辑只写在 EMPTY 状态下。
   • 卖出/止损逻辑只写在 HOLDING 状态下。
   • 代码结构清晰，修改止损逻辑时不会意外影响买入逻辑。

3. 状态持久化：

   • 使用 g.state 存储状态。在 JoinQuant 的模拟交易中，g 对象会被序列化保存。即使服务器重启或第二天开盘，策略依然记得自己是处于“持仓”还是“空仓”状态，不会因为重启而错误地重新买入。

4. 容错处理：

   • 代码中包含了一个检查 position.total_amount == 0 的逻辑。这是为了防止在模拟盘中，用户手动通过券商软件卖出了股票，但策略状态机还停留在 HOLDING 的情况。如果发现实际仓位为0，强制将状态重置为 EMPTY。

进阶：如何管理多只股票？

如果策略同时操作多只股票，可以使用字典来管理状态：

def initialize(context):
    # ... 其他初始化 ...
    g.security_list = ['000001.XSHE', '000002.XSHE']
    # 使用字典存储每只股票的状态
    g.stock_states = {stock: State.EMPTY for stock in g.security_list}

def market_open(context):
    for security in g.security_list:
        # 获取当前股票的状态
        current_state = g.stock_states.get(security, State.EMPTY)
        
        if current_state == State.EMPTY:
            # 检查该股票的买入逻辑
            if buy_condition(security):
                order_value(security, 10000)
                g.stock_states[security] = State.HOLDING # 更新该股票状态
                
        elif current_state == State.HOLDING:
            # 检查该股票的卖出逻辑
            if sell_condition(security):
                order_target(security, 0)
                g.stock_states[security] = State.EMPTY # 更新该股票状态

Q&A

Q: 为什么不直接用 context.portfolio.positions[security].total_amount > 0 来判断是否持仓？
A: 虽然可以用持仓量来判断，但状态机能包含更多信息。例如，你可能持仓了，但处于“等待卖出确认”的中间状态，或者处于“止损后冷却期”（虽然仓位为0，但不允许立即买入）。使用显式的 g.state 变量可以表达比“有货/没货”更复杂的业务逻辑。

Q: 状态机能防止高频回测中的信号闪烁吗？
A: 可以。在分钟级回测中，如果某一分钟触发买入并流转为 HOLDING，即使下一分钟信号依然满足买入条件，由于状态已变，策略不会再次下单。这对于避免资金被碎片化下单非常重要。

Q: 如果模拟交易中我手动干预了账户，状态机会乱吗？
A: 会。因为 g.state 是策略内部记录的逻辑状态，与账户真实状态可能脱节。建议在 handle_data 或 market_open 的开头加入**状态同步（Reconcile）**逻辑，例如：如果 g.state == HOLDING 但真实持仓为0，则强制重置为 EMPTY。