期魔方期魔方
注册
KB / ARTICLE

策略案例 · 风控模型 - 移动止盈止损

策略文档 · 期货版修订 R1发布于 2026-10-10

本文给出 3 种移动止盈止损风控模型：点数 / 价格比例 / 盈亏比例，包括固定止损 + 移动止盈，移动止盈指达到激活条件后启动，发生回撤设定条件后平仓，可直接挂载到任意已有策略上作为风控层。

概述

移动止盈止损：包括固定止损 + 移动止盈。移动止盈指达到设置的激活条件后启动，发生回撤设定条件后进行平仓。

3 个模型共享相同的架构：

函数 作用
on_init 初始化计数器 counter、处理标志 processing，额外初始化极值跟踪变量
on_tick 每 4 个 tick 触发一次 auto_tpsl，避免过于频繁
close_order 根据 send_order_mode（市价/现价/对价）确定平仓价格并发单，区分平今/平昨
auto_tpsl 核心逻辑：遍历持仓 → 过滤挂载合约 → 计算盈亏 → 更新极值 → 判断激活/回撤 → 平今平昨分别下单

报单方式

send_order_mode 支持三种平仓报价方式：

方式 多头卖平取价 空头买平取价
市价 UpperLimitPrice（涨停价） LowerLimitPrice（跌停价）
现价 LastPrice（最新价） LastPrice（最新价）
对价 BidPrice1（买一） AskPrice1（卖一）

平今平昨处理

上期所（SHFE）/ 能源中心（INE）需要区分平今（comboffsetflag=3）与平昨（comboffsetflag=1）；其他交易所合并为昨仓处理。

移动止盈三要素

  1. 激活条件：盈利达到启动阈值（tp_start_points / tp_start_rate）后激活移动止盈
  2. 极值跟踪：记录持仓以来的最高价（多头）/ 最低价（空头）/ 最高盈亏比例
  3. 回撤触发：从极值回撤达到设定阈值（tp_points / tp_rate）后平仓

一、移动止盈止损_点数

固定止损 + 移动止盈。盈利达 tp_start_points 激活移动止盈，价格从极值回撤 tp_points 后平仓。

# ------------------------------------------#
# 文件类型:量化策略
# 帮助文档:https://jukequant.com/knowledge/category/strategy
# 期魔方，为您提供专业的量化服务
# ------------------------------------------#

from pydantic import BaseModel, Field


class Params(BaseModel, validate_assignment=True):
    tp_start_points: float = Field(default=5, title="移动止盈启动点数")
    tp_points: float = Field(default=2, title="移动止盈回撤点数")
    sl_points: float = Field(default=2, title="固定止损点数")
    send_order_mode: dict = Field(default={"options": ["市价", "现价", "对价"], "value": "市价"}, title="报单方式")


def on_init(context):
    put_log("移动止盈止损(点数)风控模块(版本2025_09_04 17:30)启动", level="USER_LOG")
    context.counter = 0
    context.processing = False
    context.highest_price = 0
    context.lowest_price = float("inf")
    context.trailing_activated_points = False


def on_tick(context):
    if not context.processing:
        context.processing = True
        context.counter += 1
        if context.counter >= 4:
            auto_tpsl(context)
            context.counter = 0
        context.processing = False


def close_order(context, symbol, exchangeid, direction, volume, comboffsetflag):
    """
    发送平仓订单：根据报单方式确定价格并下单
    symbol: 合约代码
    exchangeid: 交易所代码
    direction: 平仓方向 0=买平，1=卖平
    volume: 平仓手数
    comboffsetflag: 平仓标志 1=平昨，3=平今
    """
    try:
        tick = get_tick(symbol)
        if not tick.get("ActionDay"):
            put_log(f"{symbol}的tick获取失败,等待下一次运行", level="USER_LOG")
            return 0

        price_field = ""

        if context.send_order_mode["value"] == "市价":
            price_field = "UpperLimitPrice" if direction == "0" else "LowerLimitPrice"

        if context.send_order_mode["value"] == "现价":
            price_field = "LastPrice" if direction == "0" else "LastPrice"

        if context.send_order_mode["value"] == "对价":
            price_field = "BidPrice1" if direction == "0" else "AskPrice1"

        price = float(tick.get(price_field))
        order = {
            "symbol": symbol,
            "exchangeid": exchangeid,
            "limitprice": price,
            "direction": direction,
            "orderpricetype": "2",
            "comboffsetflag": comboffsetflag,
            "volumn": volume,
        }
        send_order(order)
    except Exception as e:
        print(f"平仓出错: {e}")
        return 0
    return 1


def auto_tpsl(context):
    """核心止盈止损逻辑：固定止损 + 移动止盈"""
    position = get_position()
    put_log("检测当前持仓", level="USER_LOG")
    if position is not None:
        CLOSE_DIRECTION_MAP = {"2": "1", "3": "0"}

        for item in position:
            InstrumentID = item.get("InstrumentID")
            # 获取挂载合约ID，并进行过滤
            mounted_instrument = BASE_SETTING.get("Instrument")
            # 如果当前持仓不是挂载合约，则跳过这个持仓
            if not mounted_instrument or InstrumentID != mounted_instrument:
                continue

            # 订阅行情
            if InstrumentID not in BASE_SETTING["sub_Instruments"]:
                put_log(f"订阅{InstrumentID}", level="USER_LOG")
                put_log(BASE_SETTING["sub_Instruments"], level="USER_LOG")
                register_futures_for_quotes(InstrumentID, "M1")
            # 获取行情
            tick = get_tick(InstrumentID)
            if not tick.get("ActionDay"):
                put_log(f"{InstrumentID}的tick获取失败,等待下一次运行", level="USER_LOG")
                continue

            # 持仓信息处理
            info = item["symbolinfo"]
            VolumeMultiple = info["VolumeMultiple"]
            last_price = float(tick["LastPrice"])
            PosiDirection = item["PosiDirection"]
            open_price = item["OpenCost"] / VolumeMultiple / item["Position"]

            # 更新价格极值
            if VolumeMultiple > 0:
                if PosiDirection == "2":
                    if last_price > context.highest_price:
                        context.highest_price = last_price
                        put_log(f"{InstrumentID}多方向最高价更新为{last_price}", level="USER_LOG")
                else:
                    if last_price < context.lowest_price:
                        context.lowest_price = last_price
                        put_log(f"{InstrumentID}空方向最低价更新为{last_price}", level="USER_LOG")

            # 计算盈亏
            if PosiDirection == "2":
                profit_points = last_price - open_price
                loss_points = open_price - last_price
            else:
                profit_points = open_price - last_price
                loss_points = last_price - open_price

            # 计算点数
            fixed_stop_loss_points = context.sl_points  # 固定止损
            trailing_take_profit_points = context.tp_points  # 移动止盈
            trailing_activation_points = context.tp_start_points  # 移动止盈激活

            # 检查是否激活移动止盈
            if (
                not context.trailing_activated_points
                and profit_points >= trailing_activation_points
                and not context.trailing_activated_points
            ):
                context.trailing_activated_points = True
                put_log(f"{InstrumentID}激活移动止盈:盈利点数{profit_points:.1f}", level="USER_LOG")

            # 止损条件计算
            loss_condition = fixed_stop_loss_points > 0 and loss_points >= fixed_stop_loss_points

            # 止盈条件计算
            profit_condition = False
            if context.trailing_activated_points:
                if PosiDirection == "2":  # 多头
                    trailing_price = context.highest_price - trailing_take_profit_points
                    current_high = context.highest_price
                    profit_condition = last_price <= trailing_price
                else:  # 空头
                    trailing_price = context.lowest_price + trailing_take_profit_points
                    current_low = context.lowest_price
                    profit_condition = last_price >= trailing_price

            # 触发平仓逻辑
            if profit_condition or loss_condition:
                direction = CLOSE_DIRECTION_MAP[PosiDirection]
                exchangeid = item["ExchangeID"]

                YdStrikeFrozen = item.get("YdStrikeFrozen")  # 执行昨仓冻结
                LongFrozen = item.get("LongFrozen")  # 多头冻结
                ShortFrozen = item.get("ShortFrozen")  # 空头冻结
                Frozen = LongFrozen if direction == "1" else ShortFrozen  # 选择冻结方向
                todayFrozen = Frozen - YdStrikeFrozen  # 今日冻结 = 方向冻结 - 昨仓冻结
                today_volume = item.get("TodayPosition") - todayFrozen  # 今日可平仓 = 今日持仓 - 今日冻结
                yestoday_volume = item.get("YdPosition") - YdStrikeFrozen  # 昨仓可平仓 = 昨仓持仓 - 昨仓冻结
                today_position = 0
                context.highest_price = 0
                context.lowest_price = float("inf")
                context.trailing_activated_points = False

                # 计算可平仓位
                if exchangeid in ("SHFE", "INE"):
                    today_position = today_volume
                    yestoday_position = yestoday_volume
                else:
                    yestoday_position = today_volume + yestoday_volume

                # 发送平仓指令
                if today_position > 0:
                    close_order(context, InstrumentID, exchangeid, direction, today_position, "3")
                if yestoday_position > 0:
                    close_order(context, InstrumentID, exchangeid, direction, yestoday_position, "1")

                # 记录日志
                log_msg = f"{InstrumentID}触发{'止盈' if profit_condition else '止损'} "
                if profit_condition:
                    if PosiDirection == "2":
                        drawdown = current_high - last_price
                        log_msg += f"当前价格:{last_price:.2f} 最高价:{current_high:.2f} 回撤点数:{drawdown:.2f}"
                    else:
                        drawdown = last_price - current_low
                        log_msg += f"当前价格:{last_price:.2f} 最低价:{current_low:.2f} 回撤点数:{drawdown:.2f}"
                else:
                    fixed_stop_loss_points = loss_points

                    log_msg += f"现价:{last_price:.2f} 开仓价:{open_price} 止损点数:{fixed_stop_loss_points:.2f}"
                put_log(log_msg, level="USER_LOG")
    else:
        put_log("持仓获取失败", level="USER_LOG")


def on_stop(context):
    print("移动止盈止损(点数)风控模块")

二、移动止盈止损_价格比例

以价格比例作为触发条件。使用 trailing_highs 字典按 合约_方向 键记录每个持仓的极值，支持多持仓独立跟踪。

# ------------------------------------------#
# 文件类型:量化策略
# 帮助文档:https://jukequant.com/knowledge/category/strategy
# 期魔方，为您提供专业的量化服务
# ------------------------------------------#

from pydantic import BaseModel, Field


class Params(BaseModel, validate_assignment=True):
    tp_start_rate: float = Field(default=5, title="移动止盈启动比例(%)")
    tp_rate: float = Field(default=2, title="移动止盈回撤比例(%)")
    sl_rate: float = Field(default=2, title="固定止损比例(%)")
    send_order_mode: dict = Field(default={"options": ["市价", "现价", "对价"], "value": "市价"}, title="报单方式")


def on_init(context):
    put_log("移动止盈止损(价格比例)风控模块(版本2025_09_10 10:00)启动", level="USER_LOG")
    context.trailing_highs = {}  # 记录持仓最高价/最低价
    context.counter = 0
    context.processing = False


def on_tick(context):
    if not context.processing:
        context.processing = True
        context.counter += 1
        if context.counter >= 4:
            auto_tpsl(context)
            context.counter = 0
        context.processing = False


def close_order(context, symbol, exchangeid, direction, volume, comboffsetflag):
    """
    发送平仓订单：根据报单方式确定价格并下单
    symbol: 合约代码
    exchangeid: 交易所代码
    direction: 平仓方向 0=买平，1=卖平
    volume: 平仓手数
    comboffsetflag: 平仓标志 1=平昨，3=平今
    """
    try:
        tick = get_tick(symbol)
        if not tick.get("ActionDay"):
            put_log(f"{symbol}的tick获取失败,等待下一次运行", level="USER_LOG")
            return 0

        price_field = ""

        if context.send_order_mode["value"] == "市价":
            price_field = "UpperLimitPrice" if direction == "0" else "LowerLimitPrice"

        if context.send_order_mode["value"] == "现价":
            price_field = "LastPrice" if direction == "0" else "LastPrice"

        if context.send_order_mode["value"] == "对价":
            price_field = "BidPrice1" if direction == "0" else "AskPrice1"

        price = float(tick.get(price_field))
        order = {
            "symbol": symbol,
            "exchangeid": exchangeid,
            "limitprice": price,
            "direction": direction,
            "orderpricetype": "2",
            "comboffsetflag": comboffsetflag,
            "volumn": volume,
        }
        send_order(order)
    except Exception as e:
        print(f"平仓出错: {e}")
        return 0
    return 1


def auto_tpsl(context):
    """核心止盈止损逻辑：固定止损 + 移动止盈"""
    position = get_position()
    put_log("检测当前持仓", level="USER_LOG")
    if position is not None:
        CLOSE_DIRECTION_MAP = {"2": "1", "3": "0"}
        # 清理已平仓的跟踪记录
        current_keys = set()
        for item in position:
            key = f"{item['InstrumentID']}_{item['PosiDirection']}"
            current_keys.add(key)
        for key in list(context.trailing_highs.keys()):
            if key not in current_keys:
                del context.trailing_highs[key]

        for item in position:
            InstrumentID = item.get("InstrumentID")
            # 获取挂载合约ID，并进行过滤
            mounted_instrument = BASE_SETTING.get("Instrument")
            # 如果当前持仓不是挂载合约，则跳过这个持仓
            if not mounted_instrument or InstrumentID != mounted_instrument:
                continue

            # 订阅行情
            if InstrumentID not in BASE_SETTING["sub_Instruments"]:
                put_log(f"订阅{InstrumentID}", level="USER_LOG")
                put_log(BASE_SETTING["sub_Instruments"], level="USER_LOG")
                register_futures_for_quotes(InstrumentID, "M1")
            # 获取行情
            tick = get_tick(InstrumentID)
            if not tick.get("ActionDay"):
                put_log(f"{InstrumentID}的tick获取失败,等待下一次运行", level="USER_LOG")
                continue

            # 持仓信息处理
            info = item["symbolinfo"]
            VolumeMultiple = info["VolumeMultiple"]
            last_price = float(tick["LastPrice"])
            PosiDirection = item["PosiDirection"]
            open_price = item["OpenCost"] / VolumeMultiple / item["Position"]
            key = f"{InstrumentID}_{PosiDirection}"

            # 止损条件计算
            if PosiDirection == "2":  # 多头
                sl_price = open_price * (1 - context.sl_rate / 100)
                loss_condition = last_price <= sl_price
            else:  # 空头
                sl_price = open_price * (1 + context.sl_rate / 100)
                loss_condition = last_price >= sl_price

            # 止盈条件计算
            profit_condition = False
            if PosiDirection == "2":  # 多头
                if key in context.trailing_highs:
                    context.trailing_highs[key] = max(context.trailing_highs[key], last_price)
                    current_high = context.trailing_highs[key]
                    profit_condition = last_price <= current_high * (1 - context.tp_rate / 100)
                else:
                    if last_price >= open_price * (1 + context.tp_start_rate / 100):
                        context.trailing_highs[key] = last_price
                        put_log(f"{InstrumentID}多方向盈利部分超过启动比例,开始记录回撤比例", level="USER_LOG")
            else:  # 空头
                if key in context.trailing_highs:
                    context.trailing_highs[key] = min(context.trailing_highs[key], last_price)
                    current_low = context.trailing_highs[key]
                    profit_condition = last_price >= current_low * (1 + context.tp_rate / 100)
                else:
                    if last_price <= open_price * (1 - context.tp_start_rate / 100):
                        context.trailing_highs[key] = last_price
                        put_log(f"{InstrumentID}空方向盈利部分超过启动比例,开始记录回撤比例", level="USER_LOG")

            # 触发平仓逻辑
            if profit_condition or loss_condition:
                direction = CLOSE_DIRECTION_MAP[PosiDirection]
                exchangeid = item["ExchangeID"]

                YdStrikeFrozen = item.get("YdStrikeFrozen")  # 执行昨仓冻结
                LongFrozen = item.get("LongFrozen")  # 多头冻结
                ShortFrozen = item.get("ShortFrozen")  # 空头冻结
                Frozen = LongFrozen if direction == "1" else ShortFrozen  # 选择冻结方向
                todayFrozen = Frozen - YdStrikeFrozen  # 今日冻结 = 方向冻结 - 昨仓冻结
                today_volume = item.get("TodayPosition") - todayFrozen  # 今日可平仓 = 今日持仓 - 今日冻结
                yestoday_volume = item.get("YdPosition") - YdStrikeFrozen  # 昨仓可平仓 = 昨仓持仓 - 昨仓冻结
                today_position = 0

                # 计算可平仓位
                if exchangeid in ("SHFE", "INE"):
                    today_position = today_volume
                    yestoday_position = yestoday_volume
                else:
                    yestoday_position = today_volume + yestoday_volume

                # 发送平仓指令
                if today_position > 0:
                    close_order(context, InstrumentID, exchangeid, direction, today_position, "3")
                if yestoday_position > 0:
                    close_order(context, InstrumentID, exchangeid, direction, yestoday_position, "1")

                # 记录日志
                log_msg = f"{InstrumentID}触发{'止盈' if profit_condition else '止损'} "
                if profit_condition:
                    if PosiDirection == "2":
                        current_high = context.trailing_highs.get(key, last_price)
                        drawdown = (current_high - last_price) / current_high * 100
                        log_msg += f"最高价:{current_high:.2f} 回撤:{drawdown:.2f}%"
                    else:
                        current_low = context.trailing_highs.get(key, last_price)
                        drawdown = (last_price - current_low) / current_low * 100
                        log_msg += f"最低价:{current_low:.2f} 回撤:{drawdown:.2f}%"
                else:
                    log_msg += f"现价:{last_price:.2f} 止损价格:{sl_price:.2f}"
                put_log(log_msg, level="USER_LOG")
    else:
        put_log("持仓获取失败", level="USER_LOG")


def on_stop(context):
    print("移动止盈止损(价格比例)停止")

三、移动止盈止损_盈亏比例

以保证金比例作为触发条件。跟踪最高盈亏比例，回撤达 tp_rate 后平仓。需调用 get_symbolinfo 获取 margin_ratio。

# ------------------------------------------#
# 文件类型:量化策略
# 帮助文档:https://jukequant.com/knowledge/category/strategy
# 期魔方，为您提供专业的量化服务
# ------------------------------------------#

from pydantic import BaseModel, Field


class Params(BaseModel, validate_assignment=True):
    tp_start_rate: float = Field(default=5, title="移动止盈启动保证金比例(%)")
    tp_rate: float = Field(default=2, title="移动止盈回撤保证金比例(%)")
    sl_rate: float = Field(default=2, title="固定止损保证金比例(%)")
    send_order_mode: dict = Field(default={"options": ["市价", "现价", "对价"], "value": "市价"}, title="报单方式")


def on_init(context):
    put_log("移动止盈止损(盈亏比例)风控模块(版本2025_09_04 9:00)启动", level="USER_LOG")
    context.counter = 0
    context.processing = False
    context.trailing_activated = False
    context.highest_profit_ratio = 0


def on_tick(context):
    if not context.processing:
        context.processing = True
        context.counter += 1
        if context.counter >= 4:
            auto_tpsl(context)
            context.counter = 0
        context.processing = False


def close_order(context, symbol, exchangeid, direction, volume, comboffsetflag):
    """
    发送平仓订单：根据报单方式确定价格并下单
    symbol: 合约代码
    exchangeid: 交易所代码
    direction: 平仓方向 0=买平，1=卖平
    volume: 平仓手数
    comboffsetflag: 平仓标志 1=平昨，3=平今
    """
    try:
        tick = get_tick(symbol)
        if not tick.get("ActionDay"):
            put_log(f"{symbol}的tick获取失败,等待下一次运行", level="USER_LOG")
            return 0

        price_field = ""

        if context.send_order_mode["value"] == "市价":
            price_field = "UpperLimitPrice" if direction == "0" else "LowerLimitPrice"

        if context.send_order_mode["value"] == "现价":
            price_field = "LastPrice" if direction == "0" else "LastPrice"

        if context.send_order_mode["value"] == "对价":
            price_field = "BidPrice1" if direction == "0" else "AskPrice1"

        price = float(tick.get(price_field))
        order = {
            "symbol": symbol,
            "exchangeid": exchangeid,
            "limitprice": price,
            "direction": direction,
            "orderpricetype": "2",
            "comboffsetflag": comboffsetflag,
            "volumn": volume,
        }
        send_order(order)
    except Exception as e:
        print(f"平仓出错: {e}")
        return 0
    return 1


def auto_tpsl(context):
    """核心止盈止损逻辑：固定止损 + 移动止盈"""
    position = get_position()
    put_log("检测当前持仓", level="USER_LOG")
    if position is not None:
        CLOSE_DIRECTION_MAP = {"2": "1", "3": "0"}

        for item in position:
            InstrumentID = item.get("InstrumentID")
            # 获取挂载合约ID，并进行过滤
            mounted_instrument = BASE_SETTING.get("Instrument")
            # 如果当前持仓不是挂载合约，则跳过这个持仓
            if not mounted_instrument or InstrumentID != mounted_instrument:
                continue

            # 订阅行情
            if InstrumentID not in BASE_SETTING["sub_Instruments"]:
                put_log(f"订阅{InstrumentID}", level="USER_LOG")
                put_log(BASE_SETTING["sub_Instruments"], level="USER_LOG")
                register_futures_for_quotes(InstrumentID, "M1")
            # 获取行情
            tick = get_tick(InstrumentID)
            symbolinfo = get_symbolinfo(InstrumentID)
            if not tick.get("ActionDay"):
                put_log(f"{InstrumentID}的tick获取失败,等待下一次运行", level="USER_LOG")
                continue

            # 持仓信息处理
            info = item["symbolinfo"]
            VolumeMultiple = info["VolumeMultiple"]
            last_price = float(tick["LastPrice"])
            Position = item.get("Position")
            PosiDirection = item["PosiDirection"]
            open_price = item["OpenCost"] / VolumeMultiple / item["Position"]
            margin_ratio = symbolinfo.get("margin_ratio")  # 保证金

            # 计算逐笔盈亏
            if PosiDirection == "2":
                profit = (last_price - open_price) * Position * VolumeMultiple
            else:
                profit = (open_price - last_price) * Position * VolumeMultiple

            # 计算保证金
            margin = open_price * VolumeMultiple * Position * margin_ratio
            # 计算盈亏比例(基于保证金)
            if margin_ratio > 0:
                profit_ratio = (profit / margin) * 100
                loss_ratio = (-profit / margin) * 100
            else:
                profit_ratio = 0.0
                loss_ratio = 0.0

            # 更新最高盈亏比例
            if profit_ratio > context.highest_profit_ratio:
                context.highest_profit_ratio = profit_ratio
                put_log(f"{InstrumentID}最高盈利比例更新为{context.highest_profit_ratio:.2f}", level="USER_LOG")

            # 止损条件计算
            loss_condition = loss_ratio >= context.sl_rate

            # 止盈条件计算
            profit_condition = False
            # 检查是否激活移动止盈
            if profit_ratio >= context.tp_start_rate and not context.trailing_activated:
                context.trailing_activated = True
                put_log(f"{InstrumentID}激活移动止盈(盈亏比例):盈亏比例{profit_ratio:.2f}%", level="USER_LOG")
            if context.trailing_activated:
                current_drawdown = context.highest_profit_ratio - profit_ratio
                profit_condition = current_drawdown >= context.tp_rate

            # 触发平仓逻辑
            if profit_condition or loss_condition:
                direction = CLOSE_DIRECTION_MAP[PosiDirection]
                exchangeid = item["ExchangeID"]

                YdStrikeFrozen = item.get("YdStrikeFrozen")  # 执行昨仓冻结
                LongFrozen = item.get("LongFrozen")  # 多头冻结
                ShortFrozen = item.get("ShortFrozen")  # 空头冻结
                Frozen = LongFrozen if direction == "1" else ShortFrozen  # 选择冻结方向
                todayFrozen = Frozen - YdStrikeFrozen  # 今日冻结 = 方向冻结 - 昨仓冻结
                today_volume = item.get("TodayPosition") - todayFrozen  # 今日可平仓 = 今日持仓 - 今日冻结
                yestoday_volume = item.get("YdPosition") - YdStrikeFrozen  # 昨仓可平仓 = 昨仓持仓 - 昨仓冻结
                today_position = 0
                context.trailing_activated = False
                context.highest_profit_ratio = 0.0

                # 计算可平仓位
                if exchangeid in ("SHFE", "INE"):
                    today_position = today_volume
                    yestoday_position = yestoday_volume
                else:
                    yestoday_position = today_volume + yestoday_volume

                # 发送平仓指令
                if today_position > 0:
                    close_order(context, InstrumentID, exchangeid, direction, today_position, "3")
                if yestoday_position > 0:
                    close_order(context, InstrumentID, exchangeid, direction, yestoday_position, "1")

                # 记录日志
                log_msg = f"{InstrumentID}触发{'止盈' if profit_condition else '止损'} "
                if profit_condition:
                    log_msg += f"最高盈亏比例:{context.highest_profit_ratio:.2f}  当前盈亏比例:{profit_ratio:.2f} 回撤盈亏比例:{current_drawdown:.2f}%"
                else:
                    log_msg += f"当前价格:{last_price:.2f} 开仓价格:{open_price:.2f} 盈亏亏损比例:{loss_ratio:.2f}"
                put_log(log_msg, level="USER_LOG")
    else:
        put_log("持仓获取失败", level="USER_LOG")


def on_stop(context):
    print("移动止盈止损(盈亏比例)风控模块")

模型对比总览

模型 触发基准 参数 极值跟踪方式 适用场景
移动_点数 盈亏点数 tp_start_points / tp_points / sl_points 单一变量 highest_price / lowest_price 趋势跟踪 + 锁浮盈（单持仓）
移动_价格比例 价格比例 tp_start_rate / tp_rate / sl_rate 字典 trailing_highs 按合约_方向 多持仓独立跟踪
移动_盈亏比例 保证金比例 tp_start_rate / tp_rate / sl_rate 单一变量 highest_profit_ratio 按保证金杠杆跟踪趋势（单持仓）

关键要点

  1. 挂载即用：所有模型只处理 BASE_SETTING["Instrument"] 挂载合约的持仓，可直接作为风控模块附加到任意策略上，不影响原策略的开仓逻辑
  2. tick 降频：on_tick 中用 counter >= 4 每 4 个 tick 才执行一次 auto_tpsl，降低风控判断频率，平衡实时性与性能
  3. 处理防重入：context.processing 标志防止 auto_tpsl 在一次执行未完成时被重复触发
  4. 平今平昨：上期所（SHFE）/ 能源中心（INE）必须区分平今（"3"）/ 平昨（"1"），其他交易所合并处理；通过 TodayPosition / YdPosition / 冻结量计算可平手数
  5. 移动止盈三要素：① 激活条件（盈利达启动阈值）→ ② 极值跟踪（最高价/最低价/最高盈亏比）→ ③ 回撤触发（从极值回撤达阈值）
  6. 多持仓支持：移动_价格比例用 trailing_highs 字典按 合约_方向 键独立跟踪每个持仓；移动_点数和移动_盈亏比例用单一变量（highest_price / highest_profit_ratio），适合单持仓场景
  7. 状态重置：触发平仓后将极值/激活标志重置为初始值，确保下次开仓后重新跟踪
  8. 盈亏比例需保证金：移动_盈亏比例需调用 get_symbolinfo 获取 margin_ratio，以保证金为基准计算盈亏比例

相关阅读

  • [核心API 实时行情数据]
  • [核心API 账户持仓查询]
  • [核心API 合约信息查询]
  • [案例：网格策略]
  • [风控模型 - 固定止盈止损]
已复制到剪贴板
策略文档 · 期货版修订 R1

返回策略文档 · 期货版板块

评论