Versión Python de la estrategia de cobertura intertemporal de futuros de materias primas

Portado de la versión JavaScript deEl valor de la inversión en el mercado de divisas se calculará en función de la cantidad de divisas de divisas de divisas de divisas de divisas., esta estrategia es una estrategia de enseñanza simple, destinada a mostrar el diseño de estrategias de futuros de productos básicos en lenguaje Python.

class Hedge:
    'Hedging control class'
    def __init__(self, q, e, initAccount, symbolA, symbolB, hedgeSpread, coverSpread):
        self.q = q 
        self.initAccount = initAccount
        self.status = 0
        self.symbolA = symbolA
        self.symbolB = symbolB
        self.e = e
        self.isBusy = False 
        self.hedgeSpread = hedgeSpread
        self.coverSpread = coverSpread
        self.opAmount = OpAmount 
        
    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 

        tickerA = exchange.GetTicker()
        if not tickerA:
            return 

        insDetailB = exchange.SetContractType(self.symbolB)
        if not insDetailB:
            return 

        tickerB = exchange.GetTicker()
        if not tickerB:
            return 

        LogStatus(_D(), "A sell B buy", _N(tickerA["Buy"] - tickerB["Sell"]), "A buy B sell", _N(tickerA["Sell"] - tickerB["Buy"]))
        action = 0

        if self.status == 0:
            if (tickerA["Buy"] - tickerB["Sell"]) > self.hedgeSpread:
                Log("open position A sell B buy", tickerA["Buy"], tickerB["Sell"], "#FF0000")
                action = 1
            elif (tickerB["Buy"] - tickerA["Sell"]) > self.hedgeSpread:
                Log("open position B sell A buy", tickerB["Buy"], tickerA["Sell"], "#FF0000")
                action = 2
        elif self.status == 1 and (tickerA["Sell"] - tickerB["Buy"]) <= self.coverSpread:
            Log("close position A buy B sell", tickerA["Sell"], tickerB["Buy"], "#FF0000")
            action = 2
        elif self.status == 2 and (tickerB["Sell"] - tickerA["Buy"]) <= self.coverSpread:
            Log("close position B buy A sell", tickerB["Sell"] - tickerA["Buy"], "#FF0000")
            action = 1 

        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)
    n = 0 

    def callBack(task, ret):
        Log(task["desc"], "success" if ret else "fail")

    q = ext.NewTaskQueue(callBack)

    if CoverAll:
        Log("Start closing all remaining positions...")
        ext.NewPositionManager().CoverAll()
        Log("Operation complete")

    t = Hedge(q, exchange, initAccount, SA, SB, HedgeSpread, CoverSpread)
    while True:
        q.poll()
        t.poll()

Sólo trasplantar el código, se siente un poco demasiado simple, seguimos haciendo algunas transformaciones, añadir gráficos a esta estrategia comercial.

Se añade el siguiente código antes de la posición en la que elLogStatusLa función se llama para hacer la diferencia de precio en tiempo real en una estadística de línea K.self.preBarTimees un miembro añadido por elHedgePara dibujar, usamos Drawing Class library, llamamos directamente a la interfaz de dibujo, puedes dibujar gráficos fácilmente.

# Calculate the spread K line
        r = exchange.GetRecords()
        if not r:
            return 
        diff = tickerB["Last"] - tickerA["Last"]
        if r[-1]["Time"] != self.preBarTime:
            # Update
            self.records.append({"Time": r[-1]["Time"], "High": diff, "Low": diff, "Open": diff, "Close": diff, "Volume": 0})
            self.preBarTime = r[-1]["Time"]
        if diff > self.records[-1]["High"]:
            self.records[-1]["High"] = diff
        if diff < self.records[-1]["Low"]:
            self.records[-1]["Low"] = diff
        self.records[-1]["Close"] = diff
        ext.PlotRecords(self.records, "diff:B-A")
        ext.PlotHLine(self.hedgeSpread if diff > 0 else -self.hedgeSpread, "hedgeSpread")
        ext.PlotHLine(self.coverSpread if diff > 0 else -self.coverSpread, "coverSpread")

Efecto de retroevaluación:

A continuación, añadiremos funciones interactivas, de modo que la estrategia pueda modificar elHedgeSpreadyCoverSpreadLos parámetros en tiempo de ejecución para controlar el hedging spread y el closing spread. También necesita un botón para cerrar la posición con un clic. Añadimos estos controles en la página de edición de estrategias.

En el circuito principal de la estrategia, después de laq.poll(), t.poll()Llama, añade el código de control interactivo.

while True:
        q.poll()
        t.poll()
        # The following interactive control code
        cmd = GetCommand()
        if cmd:
            arr = cmd.split(":")
            if arr[0] == "AllCover":
                p.CoverAll()
            elif arr[0] == "SetHedgeSpread":
                t.SetHedgeSpread(float(arr[1]))
            elif arr[0] == "SetCoverSpread":
                t.SetCoverSpread(float(arr[1]))

Puede copiar toda la estrategia de negociación aquí:https://www.fmz.com/strategy/211504