코메디 선물 간시계 볼링거 헤지 전략의 파이썬 버전 (단계 연구용)

이전에 작성된 시간 간 중재 전략은 포지션을 열고 닫을 때 헤지 스프레드를 수동으로 입력해야합니다. 가격 차이를 판단하는 것은 더 주관적입니다. 이 기사에서는 이전 헤지 전략을 BOLL 지표를 사용하여 포지션을 열고 닫는 전략으로 변경합니다.

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

전략 파라미터 설정:

전체적인 전략 틀은 기본적으로재화 선물의 파이썬 버전, 그러나 대응하는 BOLL 지표 매개 변수가 추가됩니다. 전략이 실행 될 때 두 계약의 K-라인 데이터가 얻되며 가격 차이를 계산하여 스프레드를 계산합니다. 배열은TA.BOLL보링거 밴드를 계산하는 기능. 스프레드가 볼링거 밴드 상부 레일을 초과하면 헤지 될 것이고, 하부 레일을 만지면 반대 동작이 될 것입니다. 포지션을 유지 할 때 포지션을 닫기 위해 중부 레일을 만집니다.

백테스트:

이 기사 는 주로 연구 목적으로만 사용 됩니다. 전체 전략:https://www.fmz.com/strategy/213826