资源加载中... loading...

python版跟踪止盈止损类库(参考学习)

Author: 雨幕(youquant), Date: 2022-10-13 14:16:41
Tags:

参考文章:https://www.youquant.com/bbs-topic/9421


#!/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)

更多内容