The previously written intertemporal arbitrage strategy requires manual input of the hedging spread for opening and closing positions. Judging the price difference is more subjective. In this article, we will change the previous hedging strategy to the strategy of using the BOLL indicator to open and close positions.
class Hedge:
'Hedging control class'
def __init__(self, q, e, initAccount, symbolA, symbolB, maPeriod, atrRatio, opAmount):
self.q = q
self.initAccount = initAccount
self.status = 0
self.symbolA = symbolA
self.symbolB = symbolB
self.e = e
self.isBusy = False
self.maPeriod = maPeriod
self.atrRatio = atrRatio
self.opAmount = opAmount
self.records = []
self.preBarTime = 0
def poll(self):
if (self.isBusy or not exchange.IO("status")) or not ext.IsTrading(self.symbolA):
Sleep(1000)
return
insDetailA = exchange.SetContractType(self.symbolA)
if not insDetailA:
return
recordsA = exchange.GetRecords()
if not recordsA:
return
insDetailB = exchange.SetContractType(self.symbolB)
if not insDetailB:
return
recordsB = exchange.GetRecords()
if not recordsB:
return
# Calculate the spread price K line
if recordsA[-1]["Time"] != recordsB[-1]["Time"]:
return
minL = min(len(recordsA), len(recordsB))
rA = recordsA.copy()
rB = recordsB.copy()
rA.reverse()
rB.reverse()
count = 0
arrDiff = []
for i in range(minL):
arrDiff.append(rB[i]["Close"] - rA[i]["Close"])
arrDiff.reverse()
if len(arrDiff) < self.maPeriod:
return
# Calculate Bollinger Bands indicator
boll = TA.BOLL(arrDiff, self.maPeriod, self.atrRatio)
ext.PlotLine("upper trail", boll[0][-2], recordsA[-2]["Time"])
ext.PlotLine("middle trail", boll[1][-2], recordsA[-2]["Time"])
ext.PlotLine("lower trail", boll[2][-2], recordsA[-2]["Time"])
ext.PlotLine("Closing price spread", arrDiff[-2], recordsA[-2]["Time"])
LogStatus(_D(), "upper trail:", boll[0][-1], "\n", "middle trail:", boll[1][-1], "\n", "lower trail:", boll[2][-1], "\n", "Current closing price spread:", arrDiff[-1])
action = 0
# Signal trigger
if self.status == 0:
if arrDiff[-1] > boll[0][-1]:
Log("Open position A buy B sell", ", A latest price:", recordsA[-1]["Close"], ", B latest price:", recordsB[-1]["Close"], "#FF0000")
action = 2
# Add chart markers
ext.PlotFlag(recordsA[-1]["Time"], "A buy B sell", "Positive")
elif arrDiff[-1] < boll[2][-1]:
Log("Open position A sell B buy", ", A latest price:", recordsA[-1]["Close"], ", B latest price:", recordsB[-1]["Close"], "#FF0000")
action = 1
# Add chart markers
ext.PlotFlag(recordsA[-1]["Time"], "A sell B buy", "Negative")
elif self.status == 1 and arrDiff[-1] > boll[1][-1]:
Log("Close position A buy B sell", ", A latest price:", recordsA[-1]["Close"], ", B latest price:", recordsB[-1]["Close"], "#FF0000")
action = 2
# Add chart markers
ext.PlotFlag(recordsA[-1]["Time"], "A buy B sell", "Close Negative")
elif self.status == 2 and arrDiff[-1] < boll[1][-1]:
Log("Close position A sell B buy", ", A latest price:", recordsA[-1]["Close"], ", B latest price:", recordsB[-1]["Close"], "#FF0000")
action = 1
# Add chart markers
ext.PlotFlag(recordsA[-1]["Time"], "A sell B buy", "Close Positive")
# Execute specific instructions
if action == 0:
return
self.isBusy = True
tasks = []
if action == 1:
tasks.append([self.symbolA, "sell" if self.status == 0 else "closebuy"])
tasks.append([self.symbolB, "buy" if self.status == 0 else "closesell"])
elif action == 2:
tasks.append([self.symbolA, "buy" if self.status == 0 else "closesell"])
tasks.append([self.symbolB, "sell" if self.status == 0 else "closebuy"])
def callBack(task, ret):
def callBack(task, ret):
self.isBusy = False
if task["action"] == "sell":
self.status = 2
elif task["action"] == "buy":
self.status = 1
else:
self.status = 0
account = _C(exchange.GetAccount)
LogProfit(account["Balance"] - self.initAccount["Balance"], account)
self.q.pushTask(self.e, tasks[1][0], tasks[1][1], self.opAmount, callBack)
self.q.pushTask(self.e, tasks[0][0], tasks[0][1], self.opAmount, callBack)
def main():
SetErrorFilter("ready|login|timeout")
Log("Connecting to the trading server...")
while not exchange.IO("status"):
Sleep(1000)
Log("Successfully connected to the trading server")
initAccount = _C(exchange.GetAccount)
Log(initAccount)
def callBack(task, ret):
Log(task["desc"], "success" if ret else "failure")
q = ext.NewTaskQueue(callBack)
p = ext.NewPositionManager()
if CoverAll:
Log("Start closing all remaining positions...")
p.CoverAll()
Log("Operation complete")
t = Hedge(q, exchange, initAccount, SA, SB, MAPeriod, ATRRatio, OpAmount)
while True:
q.poll()
t.poll()
Strategy parameter setting:
The overall strategy framework is basically the same as the Python version of commodity futures intertemporal hedging strategy, except that the corresponding BOLL indicator parameters are added. When the strategy is running, the K-line data of the two contracts is obtained, and then the price difference is calculated to calculate the spread. The array is used as data of the TA.BOLL function to calculate the Bollinger Bands. When the spread exceeds the Bollinger Band's upper rail, it will be hedged, and when it touches the lower rail, it will be opposed operating. When holding a position, touch the middle rail to close the positions.
Backtest:
This article is mainly used for study purpose only.
Complete strategy: https://www.fmz.com/strategy/213826