1. 策略广场
2. python版跟踪止盈止损类库（参考学习）

python版跟踪止盈止损类库（参考学习）

Author: 雨幕（youquant）, Date: 2022-10-13 14:16:41
Tags:

---

---

#!/usr/bin/python3

class Entrust:
    def __init__(self, exchange, entrustId, contractType, direction, amount, trail_price, trail_offset, loss):
        self.entrustId = entrustId
        self.contractType = contractType
        self.direction = direction
        self.amount = amount
        self.trail_price = trail_price
        self.trail_offset = trail_offset
        self.loss = loss 
        self.isFinished = False
        self.refPrice = -1 
        self.e = exchange

    def getPosition(self, e, contractType, direction, positions = None):
        allCost = 0
        allAmount = 0
        allProfit = 0
        allFrozen = 0
        posMargin = 0
        if not positions:
            positions = _C(e.GetPosition)
        for i in range(len(positions)):
            if (positions[i]['ContractType'] == contractType and (((positions[i]['Type'] == PD_LONG or positions[i]['Type'] == PD_LONG_YD) and direction == PD_LONG) or ((positions[i]['Type'] == PD_SHORT or positions[i]['Type'] == PD_SHORT_YD) and direction == PD_SHORT))):
                posMargin = positions[i]['MarginLevel']
                allCost += positions[i]['Price'] * positions[i]['Amount']
                allAmount += positions[i]['Amount']
                allProfit += positions[i]['Profit']
                allFrozen += positions[i]['FrozenAmount']
        if allAmount == 0:
            return 
        return {
            "MarginLevel": posMargin,
            "FrozenAmount": allFrozen,
            "Price": _N(allCost / allAmount),
            "Amount": allAmount,
            "Profit": allProfit,
            "Type": direction,
            "ContractType": contractType
        }

    def monitor(self):
        e = self.e
        if e.IO("status") and not self.isFinished:
            info = e.SetContractType(self.contractType)
            if not info:
                return False
            ticker = e.GetTicker()
            if not ticker:
                return False
            pos = e.GetPosition()
            if not pos:
                return False
            pos = self.getPosition(e, self.contractType, self.direction, pos)
            if not pos:
                return False 
            if pos["Amount"] < self.amount:
                Log("持仓量小于计划量，调整计划量为持仓量。", "#FF0000")
                self.amount = pos["Amount"]
            if self.refPrice == -1:
                # 止损
                if (pos["Type"] == PD_LONG or pos["Type"] == PD_LONG_YD) and ticker.Last < pos["Price"] - self.loss:
                    Log("跟踪止盈止损模版触发：多头止损")
                    self.isFinished = True
                    return {"exchange": self.e, "contractType": self.contractType, "active": "closebuy", "amount": self.amount}
                elif (pos["Type"] == PD_SHORT or pos["Type"] == PD_SHORT_YD) and ticker.Last > pos["Price"] + self.loss:
                    Log("跟踪止盈止损模版触发：空头止损")
                    self.isFinished = True
                    return {"exchange": self.e, "contractType": self.contractType, "active": "closesell", "amount": self.amount}
                
                # 跟踪止盈
                if (pos["Type"] == PD_LONG or pos["Type"] == PD_LONG_YD) and ticker.Last > self.trail_price:
                    Log("多头触发跟踪止盈，触发价格：", self.trail_price, "，当前价格：", ticker.Last)
                    self.refPrice = ticker.Last
                elif (pos["Type"] == PD_SHORT or pos["Type"] == PD_SHORT_YD) and ticker.Last < self.trail_price:
                    Log("空头触发跟踪止盈，触发价格：", self.trail_price, "，当前价格：", ticker.Last)
                    self.refPrice = ticker.Last
            else:
                if (pos["Type"] == PD_LONG or pos["Type"] == PD_LONG_YD):
                    self.refPrice = ticker.Last if ticker.Last > self.refPrice else self.refPrice
                    if ticker.Last < self.refPrice - self.trail_offset:
                        Log("多头跟踪止盈，参考最高价格：", self.refPrice, "，偏移距离：", self.trail_offset)
                        self.isFinished = True
                        return {"exchange": self.e, "contractType": self.contractType, "active": "closebuy", "amount": self.amount}
                elif (pos["Type"] == PD_SHORT or pos["Type"] == PD_SHORT_YD):
                    self.refPrice = ticker.Last if ticker.Last < self.refPrice else self.refPrice
                    if ticker.Last > self.refPrice + self.trail_offset:
                        Log("空头跟踪止盈，参考最低价格：", self.refPrice, "，偏移距离：", self.trail_offset)
                        self.isFinished = True
                        return {"exchange": self.e, "contractType": self.contractType, "active": "closesell", "amount": self.amount}
            return False 
        else:
            return False

# exchange 交易所对象, entrustId 自定义的ID, contractType 合约代码, direction 要监控的持仓方向, amount 要处理的持仓数量, trail_price 触发跟踪止盈的价格, trail_offset 跟踪止盈偏移价格, loss 止损距离
def exit(exchange, entrustId, contractType, direction, amount, trail_price, trail_offset, loss):
    return Entrust(exchange, entrustId, contractType, direction, amount, trail_price, trail_offset, loss)

# 导出函数
ext.exit = exit

# 测试函数和回测设置，需要在另一个策略中测试，并且要引用当前模版、python版CTP商品期货交易类库（支持CTA函数测试版）模版
'''backtest
start: 2022-10-13 09:00:00
end: 2022-10-15 15:00:00
period: 1m
basePeriod: 1m
exchanges: [{"eid":"Futures_CTP","currency":"FUTURES"}]
'''

# 测试函数
def main():
    q = ext.NewTaskQueue() # python版CTP商品期货交易类库（支持CTA函数测试版）的导出函数
    n = 0
    arrEntrust = []
    while True:
        if exchange.IO("status"):
            LogStatus("已经连接")
            if n == 10:
                # 模拟
                q.pushTask(exchange, "rb2301", "sell", 3, lambda task, ret: Log(task["desc"], ret))
                '''
                调用ext.exit进行跟踪止盈止损，参数：exchange 为要操作的交易所对象，"rb2301_long" 为自定义起的名字，"rb2301" 为要操作的合约，PD_LONG 为方向，即监控多头仓位，
                2的意思是操作的仓位数量，3766为触发跟踪止盈的价格，20为跟踪止盈的偏移量，70为止损距离（价格）
                '''
                obj = ext.exit(exchange, "rb2301_short", "rb2301", PD_SHORT, 3, 3748, 20, 20)
                arrEntrust.append(obj)
                
            # 遍历委托    
            for obj in arrEntrust:
                ret = obj.monitor()
                if ret:
                    Log("ret:", ret["contractType"], ret["active"], ret["amount"])
                    q.pushTask(ret["exchange"], ret["contractType"], ret["active"], ret["amount"], lambda task, ret: Log(task["desc"], ret))
        
            # 增加计数n
            n += 1
            
            # 执行python版CTP商品期货交易类库（支持CTA函数测试版）的处理函数，处理具体交易
            q.poll()
        else :
            LogStatus("未连接")
        
        Sleep(1000)

---

• 移植Pine语言单品种成为多品种策略
• Pine语言顶/底背离画图函数
• 多品种海龟交易策略
• 多品种跨期布林对冲策略
• 商品期货跨品种套利
• 跨期布林线对冲
• 简易波动EMV策略
• 唐奇安通道
• 商品期货MACD策略
• 商品期货跨品种套利
• 半自动跟踪止损止盈委托工具
• 超级趋势跟踪止盈策略
• 均线策略-赢减输加
• 双重范围过滤器策略
• TUE ADX/MACD汇集策略
• SAR-高低点
• 线性趋势
• 精确的摆动交易系统
• 范围过滤器5分钟买卖策略
• 倒锤策略（扩展选项）