期魔方期魔方
注册
KB / ARTICLE

策略案例 · 风控模型 - 固定止盈止损

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

本文给出 4 种固定止盈止损风控模型：价格 / 点数 / 价格比例 / 盈亏比例，达到设定的止盈止损条件立即平仓，可直接挂载到任意已有策略上作为风控层。

概述

固定止盈止损：达到设定的止盈止损条件立即平仓。

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

函数 作用
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）；其他交易所合并为昨仓处理。

一、固定止盈止损_价格

以绝对价格作为止盈止损触发条件：多头价格涨破 tp_price 止盈、跌破 sl_price 止损；空头反之。

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

from pydantic import BaseModel, Field


class Params(BaseModel, validate_assignment=True):
   """策略参数模型：定义止盈止损价格及报单方式"""
    tp_price: float = Field(default=0.0, title="止盈价格")         # 触发止盈的固定价格
    sl_price: float = Field(default=0.0, title="止损价格")         # 触发止损的固定价格
    send_order_mode: dict = Field(
        default={"options": ["市价", "现价", "对价"], "value": "市价"},
        title="报单方式"
    )  # 平仓报单方式：市价=涨停/跌停价，现价=最新价，对价=对手价


def on_init(context):
    """策略初始化"""
    put_log("固定价格止盈止损风控模块(版本2025_09_04 09:00)启动", level="USER_LOG")
    context.counter = 0
    context.processing = False


def on_tick(context):
    """每个tick触发"""
    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, exchangid, direction, volume, comboffsetflag):
    """
    发送平仓订单：根据报单方式确定价格并下单
    symbol: 合约代码
    exchangid: 交易所代码
    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

        # 默认用涨跌停价格（市价）
        upper_price = float(tick.get("UpperLimitPrice"))
        lower_price = float(tick.get("LowerLimitPrice"))

        # 现价模式：用最新价
        if context.send_order_mode.get("value") == "现价":
            upper_price = float(tick.get("LastPrice"))
            lower_price = float(tick.get("LastPrice"))

        # 对价模式：买平用卖一，卖平用买一
        if context.send_order_mode.get("value") == "对价":
            upper_price = float(tick.get("BidPrice1"))
            lower_price = float(tick.get("AskPrice1"))

        # 根据平仓方向选择价格
        price = upper_price if direction == "0" else lower_price

        order = {
            "symbol": symbol,
            "exchangeid": exchangid,
            "limitprice": price,
            "direction": direction,
            "orderpricetype": "2",          # 限价单
            "comboffsetflag": comboffsetflag,
            "volumn": volume,
        }
        send_order(order)
    except Exception as e:
        print(f"开单出现错误=>请查看GridTrade.close_order[{e}]")
        return 0
    return 1


def auto_tpsl(context):
    """
    止盈止损主逻辑：
    1. 只处理挂载合约的持仓
    2. 未订阅合约先订阅
    3. 根据多头/空头分别判断止盈止损
    4. 区分平今/平昨
    """
    position = get_position()
    if position is None:
        put_log("持仓获取失败", level="USER_LOG")
        return

    CLOSE_DIRECTION_MAP = {"2": "1", "3": "0"}  # 2=多头→卖平=1，3=空头→买平=0

    for item in position:
        InstrumentID = item.get("InstrumentID")

        # 只处理挂载合约
        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")
            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.get("symbolinfo")
        VolumeMultiple = info.get("VolumeMultiple")
        last_price = float(tick.get("LastPrice"))
        open_cost = item.get("OpenCost")
        Position = item.get("Position")
        position_cost = Position * last_price * VolumeMultiple
        profit = position_cost - open_cost   # 多头：正=盈利；空头：负=盈利

        PosiDirection = item.get("PosiDirection")  # 2=多头，3=空头
        should_close = False

        # 多头止盈止损判断
        if PosiDirection == "2":
            tp_price = max(context.tp_price, context.sl_price)  # 高为止盈
            sl_price = min(context.tp_price, context.sl_price)    # 低为止损
            if last_price >= tp_price:
                should_close = True
                put_log(f"{InstrumentID}触发止盈, 当前价格{last_price},止盈价格{tp_price}", level="USER_LOG")
            elif last_price <= sl_price:
                should_close = True
                put_log(f"{InstrumentID}触发止损, 当前价格{last_price},止损价格{sl_price}", level="USER_LOG")

        # 空头止盈止损判断
        if PosiDirection == "3":
            tp_price = min(context.tp_price, context.sl_price)  # 低为止盈
            sl_price = max(context.tp_price, context.sl_price)  # 高为止损
            if last_price <= tp_price:
                should_close = True
                put_log(f"{InstrumentID}触发止盈, 当前价格{last_price},止盈价格{tp_price}", level="USER_LOG")
            elif last_price >= sl_price:
                should_close = True
                put_log(f"{InstrumentID}触发止损, 当前价格{last_price},止损价格{sl_price}", level="USER_LOG")

        if not should_close:
            continue

        # 计算可平今/平昨数量
        direction = CLOSE_DIRECTION_MAP[PosiDirection]
        LongFrozen = item.get("LongFrozen")
        ShortFrozen = item.get("ShortFrozen")
        YdStrikeFrozen = item.get("YdStrikeFrozen")
        Frozen = LongFrozen if direction == "1" else ShortFrozen
        todayFrozen = Frozen - YdStrikeFrozen
        today_volume = item.get("TodayPosition") - todayFrozen
        yestoday_volume = item.get("YdPosition") - YdStrikeFrozen
        exchangeid = item.get("ExchangeID")

        # 上期所/能源中心区分平今平昨
        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")  # 平昨


# 策略停止：打印日志
def on_stop(context):
    print("固定价格止盈止损风控模块")

二、固定止盈止损_点数

以点数作为触发条件：盈利点数达 tp_points 止盈，亏损点数达 sl_points 止损。

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

from pydantic import BaseModel, Field


class Params(BaseModel, validate_assignment=True):
    tp_points: int = Field(default=5, title="止盈点数")  # 止盈触发点数
    sl_points: int = 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


def on_tick(context):
    """每个tick触发"""
    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, exchangid, direction, volume, comboffsetflag):
    """
    发送平仓订单：根据报单方式确定价格并下单
    symbol: 合约代码
    exchangid: 交易所代码
    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

        # 根据报单方式确定委托价格
        upper_price = float(tick.get("UpperLimitPrice"))
        lower_price = float(tick.get("LowerLimitPrice"))

        if context.send_order_mode.get("value") == "现价":
            upper_price = float(tick.get("LastPrice"))
            lower_price = float(tick.get("LastPrice"))

        if context.send_order_mode.get("value") == "对价":
            upper_price = float(tick.get("BidPrice1"))
            lower_price = float(tick.get("AskPrice1"))

        price = upper_price if direction == "0" else lower_price
        order = {
            "symbol": symbol,
            "exchangeid": exchangid,
            "limitprice": price,
            "direction": direction,
            "orderpricetype": "2",
            "comboffsetflag": comboffsetflag,
            "volumn": volume,
        }
        send_order(order)
    except Exception as e:
        print(f"开单出现错误=>请查看GridTrade.close_order[{e}]")
        return 0
    return 1


def auto_tpsl(context):
    """
    固定点数止盈止损主逻辑
    遍历持仓，计算盈亏点数，满足条件即平仓
    """
    position = get_position()
    if position is None:
        put_log("持仓获取失败", level="USER_LOG")
        return

    # 多头平仓方向映射：2-多头->1-卖平
    CLOSE_DIRECTION_MAP = {"2": "1", "3": "0"}

    for item in position:
        InstrumentID = item.get("InstrumentID")
        # 仅处理挂载合约
        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")
            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.get("symbolinfo")
        VolumeMultiple = info.get("VolumeMultiple")
        last_price = float(tick.get("LastPrice"))
        Position = item.get("Position")
        open_cost = item.get("OpenCost")
        avg_open_price = round(open_cost / VolumeMultiple / Position, 2)  # 开仓均价

        fixed_take_profit_points = context.tp_points
        fixed_stop_loss_points = context.sl_points
        PosiDirection = item.get("PosiDirection")

        should_close = False
        # 多头止盈止损判断
        if PosiDirection == "2":  # 多头
            profit_points = last_price - avg_open_price
            loss_points = avg_open_price - last_price
            if fixed_take_profit_points > 0 and profit_points >= fixed_take_profit_points:
                should_close = True
                put_log(
                    f"触发固定点数止盈: 开仓价{avg_open_price}, 当前价{last_price}, 盈利点数{profit_points:.1f}",
                    level="USER_LOG",
                )
            elif fixed_stop_loss_points > 0 and loss_points >= fixed_stop_loss_points:
                should_close = True
                put_log(
                    f"触发固定点数止损: 开仓价{avg_open_price}, 当前价{last_price}, 亏损点数{loss_points:.1f}",
                    level="USER_LOG",
                )
        # 空头止盈止损判断
        elif PosiDirection == "3":  # 空头
            profit_points = avg_open_price - last_price
            loss_points = last_price - avg_open_price
            if fixed_take_profit_points > 0 and profit_points >= fixed_take_profit_points:
                should_close = True
                put_log(
                    f"触发固定点数止盈: 开仓价{avg_open_price}, 当前价{last_price}, 盈利点数{profit_points:.1f}",
                    level="USER_LOG",
                )
            elif fixed_stop_loss_points > 0 and loss_points >= fixed_stop_loss_points:
                should_close = True
                put_log(
                    f"触发固定点数止损: 开仓价{avg_open_price}, 当前价{last_price}, 亏损点数{loss_points:.1f}",
                    level="USER_LOG",
                )

        if not should_close:
            continue

        # 计算可平今/平昨数量
        direction = CLOSE_DIRECTION_MAP[PosiDirection]
        LongFrozen = item.get("LongFrozen")
        ShortFrozen = item.get("ShortFrozen")
        Frozen = LongFrozen if direction == "1" else ShortFrozen
        YdStrikeFrozen = item.get("YdStrikeFrozen")
        todayFrozen = Frozen - YdStrikeFrozen
        today_volume = item.get("TodayPosition") - todayFrozen
        yestoday_volume = item.get("YdPosition") - YdStrikeFrozen
        exchangeid = item.get("ExchangeID")

        # 上期所/能源中心区分平今平昨
        if exchangeid in ("SHFE", "INE"):
            today_position = today_volume
            yestoday_position = yestoday_volume
        else:
            yestoday_position = today_volume + yestoday_volume
            today_position = 0

        # 分别发送平今与平昨订单
        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")


def on_stop(context):
    """策略停止时日志输出"""
    print("固定点数止盈止损风控模块")

三、固定止盈止损_价格比例

以开仓成本比例作为触发条件：盈亏占开仓成本的百分比达 tp_rate / sl_rate 即平仓。

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

from pydantic import BaseModel, Field


class Params(BaseModel, validate_assignment=True):
    tp_rate: float = Field(default=5, 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.counter = 0
    context.processing = False


def on_tick(context):
    """每个tick触发"""
    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, exchangid, direction, volume, comboffsetflag):
    """
    发送平仓订单：根据报单方式确定价格并下单
    symbol: 合约代码
    exchangid: 交易所代码
    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")
            # tick获取失败 返回等待下一次运行
            return 0

        upper_price = float(tick.get("UpperLimitPrice"))
        lower_price = float(tick.get("LowerLimitPrice"))

        if context.send_order_mode.get("value") == "现价":
            upper_price = float(tick.get("LastPrice"))
            lower_price = float(tick.get("LastPrice"))

        if context.send_order_mode.get("value") == "对价":
            upper_price = float(tick.get("BidPrice1"))
            lower_price = float(tick.get("AskPrice1"))

        price = upper_price if direction == "0" else lower_price
        order = {
            "symbol": symbol,
            "exchangeid": exchangid,
            "limitprice": price,
            "direction": direction,
            "orderpricetype": "2",
            "comboffsetflag": comboffsetflag,
            "volumn": volume,
        }
        send_order(order)
    except Exception as e:
        print(f"开单出现错误=>请查看GridTrade.close_order[{e}]")
        return 0
    return 1


def auto_tpsl(context):
    """
    止盈止损主逻辑：
    1. 只处理挂载合约的持仓
    2. 未订阅合约先订阅
    3. 根据多头/空头分别判断止盈止损
    4. 区分平今/平昨
    """
    position = get_position()
    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
            # 检查是否订阅了 'InstrumentID'
            if InstrumentID not in BASE_SETTING["sub_Instruments"]:
                put_log(f"订阅{InstrumentID}", level="USER_LOG")
                register_futures_for_quotes(InstrumentID, "M1")
            tick = get_tick(InstrumentID)
            if not tick.get("ActionDay"):
                # tick获取失败 返回等待下一次运行
                put_log(f"{InstrumentID}的tick获取失败,等待下一次运行", level="USER_LOG")
                continue

            # 持仓信息处理
            info = item.get("symbolinfo")
            VolumeMultiple = info.get("VolumeMultiple")
            last_price = float(tick.get("LastPrice"))
            LongFrozen = item.get("LongFrozen")
            ShortFrozen = item.get("ShortFrozen")
            Position = item.get("Position")
            open_cost = item.get("OpenCost")
            position_cost = Position * last_price * VolumeMultiple
            profit = position_cost - open_cost

        	# 止盈止损判断，不满足则跳过
            if profit > 0:
                if profit / open_cost * 100 < context.tp_rate:
                    continue
            else:
                if abs(profit / open_cost * 100) < context.sl_rate:
                    continue

            put_log(
                f"{round(profit / VolumeMultiple / Position, 2)}/{round(open_cost / VolumeMultiple / Position, 2)} = {round(profit / open_cost * 100, 2)}%",
                level="USER_LOG",
            )

            PosiDirection = item.get("PosiDirection")
            direction = CLOSE_DIRECTION_MAP[PosiDirection]

            YdStrikeFrozen = item.get("YdStrikeFrozen")
            Frozen = LongFrozen if direction == "1" else ShortFrozen
            todayFrozen = Frozen - YdStrikeFrozen
            today_volume = item.get("TodayPosition") - todayFrozen
            yestoday_volume = item.get("YdPosition") - YdStrikeFrozen
            exchangeid = item.get("ExchangeID")
            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")
    else:
        put_log("持仓获取失败", level="USER_LOG")


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

四、固定止盈止损_盈亏比例

以保证金比例作为触发条件：盈亏占保证金的百分比达 tp_rate / sl_rate 即平仓。需调用 get_symbolinfo 获取 margin_ratio。

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

from pydantic import BaseModel, Field


class Params(BaseModel, validate_assignment=True):
    tp_rate: float = Field(default=5, 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


def on_tick(context):
    """每个tick触发"""
    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, exchangid, direction, volume, comboffsetflag):
    """
    发送平仓订单：根据报单方式确定价格并下单
    symbol: 合约代码
    exchangid: 交易所代码
    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")
            # tick获取失败 返回等待下一次运行
            return 0

        upper_price = float(tick.get("UpperLimitPrice"))
        lower_price = float(tick.get("LowerLimitPrice"))

        if context.send_order_mode.get("value") == "现价":
            upper_price = float(tick.get("LastPrice"))
            lower_price = float(tick.get("LastPrice"))

        if context.send_order_mode.get("value") == "对价":
            upper_price = float(tick.get("BidPrice1"))
            lower_price = float(tick.get("AskPrice1"))

        price = upper_price if direction == "0" else lower_price
        order = {
            "symbol": symbol,
            "exchangeid": exchangid,
            "limitprice": price,
            "direction": direction,
            "orderpricetype": "2",
            "comboffsetflag": comboffsetflag,
            "volumn": volume,
        }
        send_order(order)
    except Exception as e:
        print(f"开单出现错误=>请查看GridTrade.close_order[{e}]")
        return 0
    return 1


def auto_tpsl(context):
    """
    止盈止损主逻辑：
    1. 只处理挂载合约的持仓
    2. 未订阅合约先订阅
    3. 根据多头/空头分别判断止盈止损
    4. 区分平今/平昨
    """
    position = get_position()
    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

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

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

            # 计算逐笔盈亏
            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 > 0:
                profit_ratio = (profit / margin) * 100
                loss_ratio = (-profit / margin) * 100
            else:
                profit_ratio = 0.0
                loss_ratio = 0.0
            should_close = False

            if context.tp_rate > 0 and profit_ratio >= context.tp_rate:  # 止盈
                put_log(
                    f"触发固定盈亏比例止盈: 开仓价{open_price}, 当前价{last_price}, 盈亏比例{profit_ratio:.2f}%",
                    level="USER_LOG",
                )
                should_close = True
            if context.sl_rate > 0 and loss_ratio >= context.sl_rate:  # 止损
                put_log(
                    f"触发固定盈亏比例止损: 开仓价{open_price}, 当前价{last_price}, 盈亏亏损比例{loss_ratio:.2f}%",
                    level="USER_LOG",
                )
                should_close = True

            if not should_close:
                continue

            direction = CLOSE_DIRECTION_MAP[PosiDirection]

            YdStrikeFrozen = item.get("YdStrikeFrozen")
            Frozen = LongFrozen if direction == "1" else ShortFrozen
            todayFrozen = Frozen - YdStrikeFrozen
            today_volume = item.get("TodayPosition") - todayFrozen
            yestoday_volume = item.get("YdPosition") - YdStrikeFrozen
            exchangeid = item.get("ExchangeID")
            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")
    else:
        put_log("持仓获取失败", level="USER_LOG")


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

模型对比总览

模型 触发基准 参数 适用场景
固定_价格 绝对价格 tp_price / sl_price 已知关键价位的区间交易
固定_点数 盈亏点数 tp_points / sl_points 按跳动数设止损的短线
固定_价格比例 开仓成本比例 tp_rate / sl_rate 按资金占比控制风险
固定_盈亏比例 保证金比例 tp_rate / sl_rate 按保证金杠杆控制风险

关键要点

  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. 价格 vs 点数 vs 比例：价格模型直接比较绝对价位；点数模型需计算 avg_open_price；比例模型（价格比例）以 open_cost 为基准；（盈亏比例）以 margin 为基准，需额外调用 get_symbolinfo 获取 margin_ratio

相关阅读

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

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

评论