- 策略广场
- 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)
template: strategy.tpl:40:21: executing "strategy.tpl" at <.api.GetStrategyListByName>: wrong number of args for GetStrategyListByName: want 7 got 6