教你使用python实现一个止盈止损类库

Author: 雨幕, Created: 2022-10-18 17:38:43, Updated: 2023-11-20 20:37:26

img

教你使用python实现一个止盈止损类库

接上篇文章,使用Pine语言实现了跟踪止盈止损。有很多用户也希望给出一个python语言设计类似止盈止损功能的例子。那么本篇我们就来动手实现一个简单的「python版跟踪止盈止损类库」。类库的主要功能和Pine语言的strategy.exit函数基本类似,为了便于理解没有做过多、复杂的设计。非常方便商品期货python量化入门的同学了解编写思路。

python版跟踪止盈止损类库源码

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

代码不多,除去测试用的main函数,剩余的代码只有100行。

只有一个接口函数即:ext.exit(...)。参数分别为:

1、exchange:交易所对象,就是在FMZ实盘或者回测上配置的交易所对象,exchange即exchanges[0]。不太明白的同学可以看下FMZ文档。 2、entrustId:可以自己定义的名称,这个可以用来后续的设计,比如要标识一个计划止盈止损用于后续判断之类的。 3、contractType:合约代码,即你要对于哪个合约的持仓进行计划止损止盈。 4、direction:监控的仓位方向,即你要进行止损止盈的仓位的方向。写PD_LONG即监控多头的持仓。 5、amount:计划止盈止损的仓位数量。 6、trail_price:触发跟踪止盈的价格。 7、trail_offset: 触发跟踪止盈后距离参考价格(参考价格就是触发到目前为止最高价或者最低价)的偏移量。 8、loss:止损距离,设置20意思就是,距离当前持仓均价亏损20元的距离,进行止损平仓。

整个代码设计非常简单,首先代码里面我们编写了一个类Entrust,根据传入exit函数的参数初始化一个Entrust类的实例,也就是用来执行某个计划委托操作的对象。然后我们使用这个对象一直调用其监控函数,根据传入的参数,执行止盈止损计划。

测试例子

我们来看一个测试例子(假设这个代码就是我们的策略,策略想使用类库进行止损、跟踪止盈的功能):

'''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, 2, 3748, 20, 70)
                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)

这段测试代码需要新建一个策略,用来测试,并且要引用2个模版(在策略广场复制这两个模版后,勾选引用模版类库):

1、python版CTP商品期货交易类库

2、python版跟踪止盈止损类库

img

勾选引用后记得点击保存按钮。

策略在n==10的时候,使用python版CTP商品期货交易类库创建的对象,进行开仓操作:

q.pushTask(exchange, "rb2301", "sell", 3, lambda task, ret: Log(task["desc"], ret))

开仓操作之后,我们就可以使用我们编写设计的「跟踪止损止盈」类库来进行计划委托、监控了。

obj = ext.exit(exchange, "rb2301_short", "rb2301", PD_SHORT, 2, 3748, 20, 70)

使用ext.exit函数,我们创建了一个监控对象,设置了监控的合约为rb2301,也就是螺纹钢合约。监控的持仓为空头持仓,执行的止损、跟踪止盈手数为2手,启动跟踪止盈功能的触发价格为3748,跟踪止盈偏移量为20元,止损距离为70元(以持仓价为基准亏70元止损)。

然后开始监控Entrust类的实例对象。

for obj in arrEntrust:
    ret = obj.monitor()

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))

根据给出的信号中的信息ret["contractType"], ret["active"], ret["amount"]去执行离场平仓操作。

回测测试

img

img

可以看到回测结果中,程序在开出螺纹钢合约的空头仓位之后,价格一直持续下跌,触发跟踪止盈,当价格上涨超过触发跟踪止盈以来的最低点向上偏移20元之后,进行跟踪止盈平仓操作。

当然,回测测试的时候也可以改动一下止损、止盈的设置,看看会不会触发止损,例如把exit参数中的70改为10,就很容易触发止损了。

完整类库:https://www.fmz.cn/strategy/373025 该模版类库仅用于交流、分享、学习,实盘请根据需求自行修改,优化。


更多内容