统一止盈止损封装方案：OK、火币、币安三所通用类库

Posted by JZW 加密货币资讯站 on September 5, 2025

金融市场瞬息万变，想在合约交易中主动止盈、及时止损，却苦于各交易所接口不同？这份缺省模块专为 FMZ 量化平台打造，通过 Python 类库一次性解决 OK（欧易）、火币、币安 三大所的止盈止损差异，降低风险、节省手续费，开发者即可专注策略本身。

为什么需要专门的止盈止损封装？

  • 节省手续费：限价止盈止损避免市价滑点。
  • 统一接口：逐仓/全仓、U 本位/币本位全部支持。
  • 防止爆仓：极端行情中提前触发止损，提早上锁亏损。

👉 立即一键体验实时风控模块，避免错过关键行情！

核心关键词

止盈止损、合约交易、逐仓、全仓、币本位、U 本位、Python 封装、OKX、火币、币安

方法一：使用场景拆解

场景 原生接口差异 封装后的优势
逐仓模式小额试仓 OK 需独立字段指定止盈止损 同一行参数即可调用
全仓模式寻求杠杆效率 火币 URL、参数多变 统一成 cangType=1
做多/做空双侧挂单 币安须分开 STOP、TAKE_PROFIT 一行代码批量下发

方法二：完整源码（开源）

import json
import time

# === 1. 主入口 ===
def zhiyingzhisun(ex, amount, directionStr, zhiying_price, zhisun_price, cangType=0):
    """
    三大所通用止盈止损
    :param ex: FMZ exchange
    :param amount: 委托数量
    :param directionStr: 'buy' or 'sell'
    :param zhiying_price: 止盈触发价（限价）
    :param zhisun_price: 止损触发价（限价）
    :param cangType: 0 逐仓 1 全仓
    :return: True/False
    """
    name = ex.GetName()
    if 'OK' in name:
        return okexSwap(ex, amount, directionStr, zhiying_price, zhisun_price)
    if 'Huobi' in name:
        return huobiSwap(ex, amount, directionStr, zhiying_price, zhisun_price, cangType)
    if 'Binance' in name:
        return binanceSwap(ex, amount, directionStr, zhiying_price, zhisun_price)
    return False

# === 2. 统一异步请求 ===
def AsynIo(ex, paramList):
    """
    FMZ 的 Go 语法封装
    """
    arrRoutine = ex.Go("IO", *paramList)
    data, ok = arrRoutine.wait()
    return data

# === 3. 火币专属 ===
def huobiSwap(ex, amount, directionStr, zhiying_price, zhisun_price, cangType):
    instrument_id = ex.GetCurrency().replace('_', '-')
    usdt_type = 'USDT' in instrument_id
    
    # 选择 URL
    if usdt_type and cangType == 0:
        url = "/linear-swap-api/v1/swap_tpsl_order"
    elif usdt_type and cangType == 1:
        url = "/linear-swap-api/v1/swap_cross_tpsl_order"
    elif not usdt_type:
        url = "/swap-api/v1/swap_tpsl_order"
    else:
        return False

    payload = {
        "contract_code": instrument_id,
        "direction": directionStr,
        "volume": str(amount),
        "tp_order_price": str(zhiying_price),
        "tp_trigger_price": str(zhiying_price),
        "sl_trigger_price": str(zhisun_price),
        "sl_order_price": str(zhisun_price),
    }
    res = AsynIo(ex, ['api', 'POST', url, '', json.dumps(payload)])
    return res.get("status") == "ok"

# === 4. 币安专属 ===
def binanceSwap(ex, amount, directionStr, zhiying_price, zhisun_price):
    instrument_id = ex.GetCurrency().replace('_', '')
    url = "/fapi/v1/order" if 'USDT' in instrument_id else "/dapi/v1/order"
    ts = str(int(time.time()*1000))
    
    # 止损
    sl_param = {
        "symbol": instrument_id,
        "side": directionStr.upper(),
        "type": "STOP",
        "quantity": str(amount),
        "price": str(zhisun_price),
        "stopPrice": str(zhisun_price),
        "timestamp": ts
    }
    sl_data = AsynIo(ex, ["api", "POST", url, "", json.dumps(sl_param)])
    if int(sl_data.get("stopPrice", 0)) != int(zhisun_price):
        return False

    # 止盈
    tp_param = {
        "symbol": instrument_id,
        "side": directionStr.upper(),
        "type": "TAKE_PROFIT",
        "quantity": str(amount),
        "price": str(zhiying_price),
        "stopPrice": str(zhiying_price),
        "timestamp": ts
    }
    tp_data = AsynIo(ex, ["api", "POST", url, "", json.dumps(tp_param)])
    return int(tp_data.get("stopPrice", 0)) == int(zhiying_price)

# === 5. OK 专属 ===
def okexSwap(ex, amount, directionStr, zhiying_price, zhisun_price):
    instrument_id = ex.GetCurrency().replace('_', '-') + "-SWAP"
    direction_code = "4" if directionStr == "buy" else "3"
    payload = {
        "instrument_id": instrument_id,
        "type": direction_code,
        "order_type": "5",  # 5 = 止盈止损委托
        "size": str(amount),
        "tp_trigger_price": str(zhiying_price),
        "tp_price": str(zhiying_price),
        "sl_trigger_price": str(zhisun_price),
        "sl_price": str(zhisun_price)
    }
    res = AsynIo(ex, ["api", "POST", "/v1/order/orders/place", "", json.dumps(payload)])
    return res.get("error_code") == "0"

# === FMZ 插件导出 ===
ext.zhiyingzhisun = zhiyingzhisun

方法三：集成到现有策略（示例）

假设你已有一条跟随趋势的策略机器人，可在关键入场处添加两行代码：

# 先下单
exchange.SetDirection("buy")
id = exchange.Buy(price, amount)

# 立即挂止盈止损
ext.zhiyingzhisun(exchange, amount, "buy", tp_price, sl_price, cangType=1)

👉 快速提升策略稳定性，防止黑天鹅？一分钟试通止盈止损实战！

FAQ：常见疑问一次讲清

Q1：如何确认止盈止损是否真的挂到交易所？
A：你可以在回调中再次查询订单列表；火币与 OK 也支持获取“止盈止损委托列表”，用统一接口即可核实成交状态。

Q2：币安返回 -1022 签名错误怎么办？
A：检查 API key 的权限，合约需开启 Delivery 或 Futures，另外确保已经完成白名单绑定。

Q3：逐仓模式下是否可以单独给入金账户加保证金？
A：不行，该类库仅负责委托逻辑，如需追加保证金需再用 POST /fapi/v1/positionMargin（币安）等接口。

Q4：在回测环境里能否使用？
A：回测环境并不对接真实交易所，止盈止损先隐藏为虚拟逻辑，上线前记得转真实环境验证。

Q5：如何批量撤销止盈止损？
A：根据各自交易所的规则，先调用查询接口得到相应的 orderId，再批量发送取消即可。

Q6：止盈价和止损价可否设置成市价触发？
A：为了兼顾手续费，本类库统一使用限价；需要市价触发，可直接在请求体内把 price 改成交易所允许的极端值，但请谨慎评估滑点风险。

结语

通过本文的完整封装，任何 Python 量化爱好者都能在 分钟级 内为自己的策略增效：

  • 摆脱接口差异，安全统一止盈止损
  • 降低成本，提高风险场景下的存活率
  • 代码开源，可随业务灵活改写

赶紧把这段公共模块嵌入你的交易机器人，让每一次开仓背后都拥有专业级风控！

CATALOG