3
关注
225
关注者
教你使用python实现一个止盈止损类库
接上篇文章，使用Pine语言实现了跟踪止盈止损。有很多用户也希望给出一个python语言设计类似止盈止损功能的例子。那么本篇我们就来动手实现一个简单的「python版跟踪止盈止损类库」。类库的主要功能和Pine语言的strategy.exit函数基本类似，为了便于理解没有做过多、复杂的设计。非常方便商品期货python量化入门的同学了解编写思路。
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：交易所对象，就是在优宽实盘或者回测上配置的交易所对象，exchange即exchanges[0]。不太明白的同学可以看下优宽文档。
2、entrustId：可以自己定义的名称，这个可以用来后续的设计，比如要标识一个计划止盈止损用于后续判断之类的。
3、contractType：合约代码，即你要对于哪个合约的持仓进行计划止损止盈。
4、direction：监控的仓位方向，即你要进行止损止盈的仓位的方向。写PD_LONG即监控多头的持仓。
5、amount：计划止盈止损的仓位数量。
6、trail_price：触发跟踪止盈的价格。
7、trail_offset： 触发跟踪止盈后距离参考价格（参考价格就是触发到目前为止最高价或者最低价）的偏移量。
8、loss：止损距离，设置20意思就是，距离当前持仓均价亏损20元的距离，进行止损平仓。
整个代码设计非常简单，首先代码里面我们编写了一个类Entrust，根据传入exit函数的参数初始化一个Entrust类的实例，也就是用来执行某个计划委托操作的对象。然后我们使用这个对象一直调用其监控函数，根据传入的参数，执行止盈止损计划。
测试例子
我们来看一个测试例子（假设这个代码就是我们的策略，策略想使用类库进行止损、跟踪止盈的功能）：
python
'''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版跟踪止盈止损类库
勾选引用后记得点击保存按钮。
策略在n==10的时候，使用python版CTP商品期货交易类库创建的对象，进行开仓操作:
python
q.pushTask(exchange, "rb2301", "sell", 3, lambda task, ret: Log(task["desc"], ret))
开仓操作之后，我们就可以使用我们编写设计的「跟踪止损止盈」类库来进行计划委托、监控了。
python
obj = ext.exit(exchange, "rb2301_short", "rb2301", PD_SHORT, 2, 3748, 20, 70)
使用ext.exit函数，我们创建了一个监控对象，设置了监控的合约为rb2301，也就是螺纹钢合约。监控的持仓为空头持仓，执行的止损、跟踪止盈手数为2手，启动跟踪止盈功能的触发价格为3748，跟踪止盈偏移量为20元，止损距离为70元（以持仓价为基准亏70元止损）。
然后开始监控Entrust类的实例对象。
python
for obj in arrEntrust:
ret = obj.monitor()
ret = obj.monitor()，当监控函数给出信号时：
python
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"]去执行离场平仓操作。
回测测试
可以看到回测结果中，程序在开出螺纹钢合约的空头仓位之后，价格一直持续下跌，触发跟踪止盈，当价格上涨超过触发跟踪止盈以来的最低点向上偏移20元之后，进行跟踪止盈平仓操作。
当然，回测测试的时候也可以改动一下止损、止盈的设置，看看会不会触发止损，例如把exit参数中的70改为10，就很容易触发止损了。
完整类库：https://www.youquant.com/strategy/373025
该模版类库仅用于交流、分享、学习，实盘请根据需求自行修改，优化。
相关推荐
评论
全部评论 (0)
暂无数据
- 1