- 策略广场
- 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)
更多内容