La versión Python de la estrategia de cobertura intertemporal de Bollinger de futuros de materias primas (sólo para fines de estudio)

La estrategia de arbitraje intertemporal previamente escrita requiere la entrada manual del spread de cobertura para la apertura y cierre de posiciones. Juzgar la diferencia de precio es más subjetiva. En este artículo, cambiaremos la estrategia de cobertura anterior a la estrategia de usar el indicador BOLL para abrir y cerrar posiciones.

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

Configuración de parámetros de estrategia:

El marco general de la estrategia es básicamente el mismo que el de laVersión Python de la estrategia de cobertura intertemporal de futuros de materias primasCuando la estrategia se ejecuta, se obtienen los datos de la línea K de los dos contratos, y luego se calcula la diferencia de precio para calcular el diferencial.TA.BOLLCuando el spread exceda el carril superior de la banda de Bollinger, será cubierto, y cuando toque el carril inferior, se opondrá al funcionamiento.

Prueba posterior:

Este artículo se utiliza principalmente sólo para fines de estudio. Estrategia completa:https://www.fmz.com/strategy/213826