Phiên bản Python của Commodity Futures Intertemporal Bollinger Hedge Strategy (Chỉ dành cho mục đích nghiên cứu)

Trong bài viết này, chúng ta sẽ thay đổi chiến lược phòng ngừa trước đó thành chiến lược sử dụng chỉ số BOLL để mở và đóng các vị trí.

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

Cài đặt tham số chiến lược:

Khung chiến lược tổng thể về cơ bản giống nhưPhiên bản Python của chiến lược phòng ngừa rủi ro giữa thời gian và tương lai hàng hóa, ngoại trừ việc các thông số chỉ số BOLL tương ứng được thêm vào. Khi chiến lược đang chạy, dữ liệu đường K của hai hợp đồng được thu được, và sau đó sự khác biệt giá được tính toán để tính chênh lệch.TA.BOLLKhi chênh lệch vượt quá đường ray trên của Bollinger Band, nó sẽ được bảo hiểm, và khi nó chạm vào đường ray dưới, nó sẽ chống lại hoạt động. Khi giữ vị trí, chạm vào đường ray giữa để đóng các vị trí.

Kiểm tra hậu quả:

Bài này chủ yếu chỉ dùng cho mục đích nghiên cứu. Chiến lược hoàn chỉnh:https://www.fmz.com/strategy/213826