Python Version Commodity Futures Moving Average Strategy

It is completely transplanted from the “CTP Commodity Futures Variety Moving Average Strategy”. Since the Python version of the commodity futures strategy does not yet have a multi-variety strategy, the JavaScript version of the “CTP Commodity Futures Multi-Variable Moving Average Strategy” was ported. Providing some design ideas and examples of Python commodity futures multi-variety strategy. Regardless of the JavaScript version or the Python version, the strategy architecture design originates from the Commodity futures multi-varieties turtle strategy.

As the simplest strategy, the moving average strategy is very easy to learn, because the moving average strategy does not have any advanced algorithms and complex logic. The ideas are clear and easy, allowing beginners to focus more on the study of strategy design, and even remove the coding related part, leaving a multi-variety strategy framework that can be easily expanded into ATR, MACD, BOLL and other indicator strategies.

Articles related to the JavaScript version: https://www.fmz.com/bbs-topic/5235.

Strategy source code

'''backtest
start: 2019-07-01 09:00:00
end: 2020-03-25 15:00:00
period: 1d
exchanges: [{"eid":"Futures_CTP","currency":"FUTURES"}]
'''

import json
import re
import time

_bot = ext.NewPositionManager()

class Manager:
    'Strategy logic control'

    ACT_IDLE = 0
    ACT_LONG = 1
    ACT_SHORT = 2 
    ACT_COVER = 3

    ERR_SUCCESS = 0
    ERR_SET_SYMBOL = 1
    ERR_GET_ORDERS = 2
    ERR_GET_POS = 3
    ERR_TRADE = 4
    ERR_GET_DEPTH = 5
    ERR_NOT_TRADING = 6
    errMsg = ["Success", "Failed to switch contract", "Failed to get order info", "Failed to get position info", "Placing Order failed", "Failed to get order depth info", "Not in trading hours"]
    
    def __init__(self, needRestore, symbol, keepBalance, fastPeriod, slowPeriod):
        # Get symbolDetail 
        symbolDetail = _C(exchange.SetContractType, symbol)
        if symbolDetail["VolumeMultiple"] == 0 or symbolDetail["MaxLimitOrderVolume"] == 0 or symbolDetail["MinLimitOrderVolume"] == 0 or symbolDetail["LongMarginRatio"] == 0 or symbolDetail["ShortMarginRatio"] == 0:
            Log(symbolDetail)
            raise Exception("Abnormal contract information")
        else :
            Log("contract", symbolDetail["InstrumentName"], "1 lot", symbolDetail["VolumeMultiple"], "lot, Maximum placing order quantity", symbolDetail["MaxLimitOrderVolume"], "Margin rate: ", _N(symbolDetail["LongMarginRatio"]), _N(symbolDetail["ShortMarginRatio"]), "Delivery date", symbolDetail["StartDelivDate"])

        # Initialization    
        self.symbol = symbol
        self.keepBalance = keepBalance
        self.fastPeriod = fastPeriod
        self.slowPeriod = slowPeriod

        self.marketPosition = None
        self.holdPrice = None
        self.holdAmount = None 
        self.holdProfit = None

        self.task = {
            "action" : Manager.ACT_IDLE,
            "amount" : 0,
            "dealAmount" : 0,
            "avgPrice" : 0,
            "preCost" : 0,
            "preAmount" : 0,
            "init" : False,
            "retry" : 0,
            "desc" : "idle",
            "onFinish" : None
        }
        
        self.lastPrice = 0 
        self.symbolDetail = symbolDetail

        # Position status information
        self.status = {
            "symbol" : symbol,
            "recordsLen" : 0,
            "vm" : [], 
            "open" : 0,
            "cover" : 0,
            "st" : 0,
            "marketPosition" : 0,
            "lastPrice" : 0,
            "holdPrice" : 0, 
            "holdAmount" : 0,
            "holdProfit" : 0, 
            "symbolDetail" : symbolDetail,
            "lastErr" : "",
            "lastErrTime" : "",
            "isTrading" : False   
        }
        
        # Other processing work during object construction
        vm = None
        if RMode == 0:
            vm = _G(self.symbol)
        else:
            vm = json.loads(VMStatus)[self.symbol]
        if vm:
            Log("Ready to resume progress, current contract status is", vm)
            self.reset(vm[0])
        else:
            if needRestore:
                Log("could not find" + self.symbol + "progress recovery information")
            self.reset()

    def setLastError(self, err=None):
        if err is None:
            self.status["lastErr"] = ""
            self.status["lastErrTime"] = ""
            return 
        t = _D()
        self.status["lastErr"] = err
        self.status["lastErrTime"] = t
    
    def reset(self, marketPosition=None):
        if marketPosition is not None:
            self.marketPosition = marketPosition
            pos = _bot.GetPosition(self.symbol, PD_LONG if marketPosition > 0 else PD_SHORT)
            if pos is not None:
                self.holdPrice = pos["Price"]
                self.holdAmount = pos["Amount"]
                Log(self.symbol, "Position", pos)
            else :
                raise Exception("Restore" + self.symbol + "position status is wrong, no position information found")
            Log("Restore", self.symbol, "average holding position price:", self.holdPrice, "Number of positions:", self.holdAmount)
            self.status["vm"] = [self.marketPosition]
        else :
            self.marketPosition = 0
            self.holdPrice = 0 
            self.holdAmount = 0 
            self.holdProfit = 0
        self.holdProfit = 0
        self.lastErr = ""
        self.lastErrTime = ""

    def Status(self):
        self.status["marketPosition"] = self.marketPosition
        self.status["holdPrice"] = self.holdPrice
        self.status["holdAmount"] = self.holdAmount
        self.status["lastPrice"] = self.lastPrice
        if self.lastPrice > 0 and self.holdAmount > 0 and self.marketPosition != 0:
            self.status["holdProfit"] = _N((self.lastPrice - self.holdPrice) * self.holdAmount * self.symbolDetail["VolumeMultiple"], 4) * (1 if self.marketPosition > 0 else -1)
        else :
            self.status["holdProfit"] = 0 
        return self.status

    def setTask(self, action, amount = None, onFinish = None):
        self.task["init"] = False 
        self.task["retry"] = 0
        self.task["action"] = action
        self.task["preAmount"] = 0
        self.task["preCost"] = 0
        self.task["amount"] = 0 if amount is None else amount
        self.task["onFinish"] = onFinish
        if action == Manager.ACT_IDLE:
            self.task["desc"] = "idle"
            self.task["onFinish"] = None
        else:
            if action != Manager.ACT_COVER:
                self.task["desc"] = ("Adding long position" if action == Manager.ACT_LONG else "Adding short position") + "(" + str(amount) + ")"
            else :
                self.task["desc"] = "Closing Position"
            Log("Task received", self.symbol, self.task["desc"])
            self.Poll(True)

    def processTask(self):
        insDetail = exchange.SetContractType(self.symbol)
        if not insDetail:
            return Manager.ERR_SET_SYMBOL
        SlideTick = 1
        ret = False
        if self.task["action"] == Manager.ACT_COVER:
            hasPosition = False
            while True:
                if not ext.IsTrading(self.symbol):
                    return Manager.ERR_NOT_TRADING
                hasPosition = False
                positions = exchange.GetPosition()
                if positions is None:
                    return Manager.ERR_GET_POS
                depth = exchange.GetDepth()
                if depth is None:
                    return Manager.ERR_GET_DEPTH
                orderId = None
                for i in range(len(positions)):
                    if positions[i]["ContractType"] != self.symbol:
                        continue
                    amount = min(insDetail["MaxLimitOrderVolume"], positions[i]["Amount"])
                    if positions[i]["Type"] == PD_LONG or positions[i]["Type"] == PD_LONG_YD:
                        exchange.SetDirection("closebuy_today" if positions[i].Type == PD_LONG else "closebuy")
                        orderId = exchange.Sell(_N(depth["Bids"][0]["Price"] - (insDetail["PriceTick"] * SlideTick), 2), min(amount, depth["Bids"][0]["Amount"]), self.symbol, "Close today's position" if positions[i]["Type"] == PD_LONG else "Close yesterday's position", "Bid", depth["Bids"][0])
                        hasPosition = True
                    elif positions[i]["Type"] == PD_SHORT or positions[i]["Type"] == PD_SHORT_YD:
                        exchange.SetDirection("closesell_today" if positions[i]["Type"] == PD_SHORT else "closesell")
                        orderId = exchange.Buy(_N(depth["Asks"][0]["Price"] + (insDetail["PriceTick"] * SlideTick), 2), min(amount, depth["Asks"][0]["Amount"]), self.symbol, "Close today's position" if positions[i]["Type"] == PD_SHORT else "Close yesterday's position", "Ask", depth["Asks"][0])
                        hasPosition = True
                    if hasPosition:
                        if not orderId:
                            return Manager.ERR_TRADE
                        Sleep(1000)
                        while True:
                            orders = exchange.GetOrders()
                            if orders is None:
                                return Manager.ERR_GET_ORDERS
                            if len(orders) == 0:
                                break
                            for i in range(len(orders)):
                                exchange.CancelOrder(orders[i]["Id"])
                                Sleep(500)
                if not hasPosition:
                    break
            ret = True
        elif self.task["action"] == Manager.ACT_LONG or self.task["action"] == Manager.ACT_SHORT:
            while True:
                if not ext.IsTrading(self.symbol):
                    return Manager.ERR_NOT_TRADING
                Sleep(1000)
                while True:
                    orders = exchange.GetOrders()
                    if orders is None:
                        return Manager.ERR_GET_ORDERS
                    if len(orders) == 0:
                        break
                    for i in range(len(orders)):
                        exchange.CancelOrder(orders[i]["Id"])
                        Sleep(500)
                positions = exchange.GetPosition()
                if positions is None:
                    return Manager.ERR_GET_POS
                pos = None
                for i in range(len(positions)):
                    if positions[i]["ContractType"] == self.symbol and (((positions[i]["Type"] == PD_LONG or positions[i]["Type"] == PD_LONG_YD) and self.task["action"] == Manager.ACT_LONG) or ((positions[i]["Type"] == PD_SHORT) or positions[i]["Type"] == PD_SHORT_YD) and self.task["action"] == Manager.ACT_SHORT):
                        if not pos:
                            pos = positions[i]
                            pos["Cost"] = positions[i]["Price"] * positions[i]["Amount"]
                        else :
                            pos["Amount"] += positions[i]["Amount"]
                            pos["Profit"] += positions[i]["Profit"]
                            pos["Cost"] += positions[i]["Price"] * positions[i]["Amount"]
                # records pre position 
                if not self.task["init"]:
                    self.task["init"] = True
                    if pos:
                        self.task["preAmount"] = pos["Amount"]
                        self.task["preCost"] = pos["Cost"]
                    else:
                        self.task["preAmount"] = 0
                        self.task["preCost"] = 0
                remain = self.task["amount"]
                if pos:
                    self.task["dealAmount"] = pos["Amount"] - self.task["preAmount"]
                    remain = int(self.task["amount"] - self.task["dealAmount"])
                    if remain <= 0 or self.task["retry"] >= MaxTaskRetry:
                        ret = {
                            "price" : (pos["Cost"] - self.task["preCost"]) / (pos["Amount"] - self.task["preAmount"]),
                            "amount" : (pos["Amount"] - self.task["preAmount"]),
                            "position" : pos
                        }
                        break
                elif self.task["retry"] >= MaxTaskRetry:
                    ret = None
                    break

                depth = exchange.GetDepth()
                if depth is None:
                    return Manager.ERR_GET_DEPTH
                orderId = None
                if self.task["action"] == Manager.ACT_LONG:
                    exchange.SetDirection("buy")
                    orderId = exchange.Buy(_N(depth["Asks"][0]["Price"] + (insDetail["PriceTick"] * SlideTick), 2), min(remain, depth["Asks"][0]["Amount"]), self.symbol, "Ask", depth["Asks"][0])
                else:
                    exchange.SetDirection("sell")
                    orderId = exchange.Sell(_N(depth["Bids"][0]["Price"] - (insDetail["PriceTick"] * SlideTick), 2), min(remain, depth["Bids"][0]["Amount"]), self.symbol, "Bid", depth["Bids"][0])
                if orderId is None:
                    self.task["retry"] += 1
                    return Manager.ERR_TRADE
        if self.task["onFinish"]:
            self.task["onFinish"](ret)
        self.setTask(Manager.ACT_IDLE)
        return Manager.ERR_SUCCESS

    def Poll(self, subroutine = False):
        # Judge the trading hours
        self.status["isTrading"] = ext.IsTrading(self.symbol)
        if not self.status["isTrading"]:
            return 

        # Perform order trading tasks
        if self.task["action"] != Manager.ACT_IDLE:
            retCode = self.processTask()
            if self.task["action"] != Manager.ACT_IDLE:
                self.setLastError("The task was not successfully processed:" + Manager.errMsg[retCode] + ", " + self.task["desc"] + ", Retry:" + str(self.task["retry"]))
            else :
                self.setLastError()
            return 

        if subroutine:
            return

        suffix = "@" if WXPush else ""
        # switch symbol
        _C(exchange.SetContractType, self.symbol)

        # Get K-line data
        records = exchange.GetRecords()
        if records is None:
            self.setLastError("Failed to get K line")
            return 
        self.status["recordsLen"] = len(records)
        if len(records) < self.fastPeriod + 2 or len(records) < self.slowPeriod + 2:
            self.setLastError("The length of the K line is less than the moving average period:" + str(self.fastPeriod) + "or" + str(self.slowPeriod))
            return 

        opCode = 0   # 0 : IDLE , 1 : LONG , 2 : SHORT , 3 : CoverALL 
        lastPrice = records[-1]["Close"]
        self.lastPrice = lastPrice

        fastMA = TA.EMA(records, self.fastPeriod)
        slowMA = TA.EMA(records, self.slowPeriod)

        # Strategy logic
        if self.marketPosition == 0:
            if fastMA[-3] < slowMA[-3] and fastMA[-2] > slowMA[-2]:
                opCode = 1 
            elif fastMA[-3] > slowMA[-3] and fastMA[-2] < slowMA[-2]:
                opCode = 2
        else:
            if self.marketPosition < 0 and fastMA[-3] < slowMA[-3] and fastMA[-2] > slowMA[-2]:
                opCode = 3
            elif self.marketPosition > 0 and fastMA[-3] > slowMA[-3] and fastMA[-2] < slowMA[-2]:
                opCode = 3

        # If no condition is triggered, the opcode is 0 and return
        if opCode == 0:
            return 

        # Preforming closing position action
        if opCode == 3:
            def coverCallBack(ret):
                self.reset()
                _G(self.symbol, None)
            self.setTask(Manager.ACT_COVER, 0, coverCallBack)
            return 
        
        account = _bot.GetAccount()
        canOpen = int((account["Balance"] - self.keepBalance) / (self.symbolDetail["LongMarginRatio"] if opCode == 1 else self.symbolDetail["ShortMarginRatio"]) / (lastPrice * 1.2) / self.symbolDetail["VolumeMultiple"])
        unit = min(1, canOpen)

        # Set up trading tasks
        def setTaskCallBack(ret):
            if not ret:
                self.setLastError("Placing Order failed")
                return 
            self.holdPrice = ret["position"]["Price"]
            self.holdAmount = ret["position"]["Amount"]
            self.marketPosition += 1 if opCode == 1 else -1
            self.status["vm"] = [self.marketPosition]
            _G(self.symbol, self.status["vm"])

        self.setTask(Manager.ACT_LONG if opCode == 1 else Manager.ACT_SHORT, unit, setTaskCallBack)

def onexit():
    Log("Exited strategy...")

def main():
    if exchange.GetName().find("CTP") == -1:
        raise Exception("Only support commodity futures CTP")
    SetErrorFilter("login|ready|flow control|connection failed|initial|Timeout")
    mode = exchange.IO("mode", 0)
    if mode is None:
        raise Exception("Failed to switch modes, please update to the latest docker!")
    while not exchange.IO("status"):
        Sleep(3000)
        LogStatus("Waiting for connection with the trading server," + _D())
    positions = _C(exchange.GetPosition)
    if len(positions) > 0:
        Log("Detecting the current holding position, the system will start to try to resume the progress...")
        Log("Position information:", positions)

    initAccount = _bot.GetAccount()
    initMargin = json.loads(exchange.GetRawJSON())["CurrMargin"]
    keepBalance = _N((initAccount["Balance"] + initMargin) * (KeepRatio / 100), 3)
    Log("Asset information", initAccount, "Retain funds:", keepBalance)

    tts = []
    symbolFilter = {}
    arr = Instruments.split(",")
    arrFastPeriod = FastPeriodArr.split(",")
    arrSlowPeriod = SlowPeriodArr.split(",")
    if len(arr) != len(arrFastPeriod) or len(arr) != len(arrSlowPeriod):
        raise Exception("The moving average period parameter does not match the number of added contracts, please check the parameters!")
    for i in range(len(arr)):
        symbol = re.sub(r'/\s+$/g', "", re.sub(r'/^\s+/g', "", arr[i]))
        if symbol in symbolFilter.keys():
            raise Exception(symbol + "Already exists, please check the parameters!")
        symbolFilter[symbol] = True
        hasPosition = False 
        for j in range(len(positions)):
            if positions[j]["ContractType"] == symbol:
                hasPosition = True
                break
        fastPeriod = int(arrFastPeriod[i])
        slowPeriod = int(arrSlowPeriod[i])
        obj = Manager(hasPosition, symbol, keepBalance, fastPeriod, slowPeriod)
        tts.append(obj)
    
    preTotalHold = -1
    lastStatus = ""
    while True:
        if GetCommand() == "Pause/Resume":
            Log("Suspending trading ...")
            while GetCommand() != "Pause/Resume":
                Sleep(1000)
            Log("Continue trading...")
        while not exchange.IO("status"):
            Sleep(3000)
            LogStatus("Waiting for connection with the trading server," + _D() + "\n" + lastStatus)

        tblStatus = {
            "type" : "table",
            "title" : "Position information",
            "cols" : ["Contract Name", "Direction of Position", "Average Position Price", "Number of Positions", "Position profits and Losses", "Number of Positions Added", "Current Price"],
            "rows" : [] 
        }

        tblMarket = {
            "type" : "table", 
            "title" : "Operating status", 
            "cols" : ["Contract name", "Contract multiplier", "Margin rate", "Trading time", "Bar length", "Exception description", "Time of occurrence"], 
            "rows" : []
        }

        totalHold = 0
        vmStatus = {}
        ts = time.time()
        holdSymbol = 0
        for i in range(len(tts)):
            tts[i].Poll()
            d = tts[i].Status()
            if d["holdAmount"] > 0:
                vmStatus[d["symbol"]] = d["vm"]
                holdSymbol += 1
            tblStatus["rows"].append([d["symbolDetail"]["InstrumentName"], "--" if d["holdAmount"] == 0 else ("long" if d["marketPosition"] > 0 else "short"), d["holdPrice"], d["holdAmount"], d["holdProfit"], abs(d["marketPosition"]), d["lastPrice"]])
            tblMarket["rows"].append([d["symbolDetail"]["InstrumentName"], d["symbolDetail"]["VolumeMultiple"], str(_N(d["symbolDetail"]["LongMarginRatio"], 4)) + "/" + str(_N(d["symbolDetail"]["ShortMarginRatio"], 4)), "is #0000ff" if d["isTrading"] else "not #ff0000", d["recordsLen"], d["lastErr"], d["lastErrTime"]])
            totalHold += abs(d["holdAmount"])

        now = time.time()
        elapsed = now - ts
        tblAssets = _bot.GetAccount(True)
        nowAccount = _bot.Account()

        if len(tblAssets["rows"]) > 10:
            tblAssets["rows"][0] = ["InitAccount", "Initial asset", initAccount]
        else:
            tblAssets["rows"].insert(0, ["NowAccount", "Currently available", nowAccount])
            tblAssets["rows"].insert(0, ["InitAccount", "Initial asset", initAccount])
        
        lastStatus = "`" + json.dumps([tblStatus, tblMarket, tblAssets]) + "`\nPolling time:" + str(elapsed) + " Seconds, current time:" + _D() + ", Number of varieties held:" + str(holdSymbol)
        if totalHold > 0:
            lastStatus += "\nManually restore the string:" + json.dumps(vmStatus)
        LogStatus(lastStatus)
        if preTotalHold > 0 and totalHold == 0:
            LogProfit(nowAccount.Balance - initAccount.Balance - initMargin)
        preTotalHold = totalHold
        Sleep(LoopInterval * 1000)

Strategy address: https://www.fmz.com/strategy/208512

Backtest comparison

We compared the JavaScript version and Python version of the strategy with backtest.

• Python version backtest

We use a public server for backtest, and we can see that the backtest of the Python version is slightly faster.

• JavaScript version backtest

It can be seen that the backtest results are exactly the same. Interested friends can delve into the code, and there will be no small gains.

Expand

Let’s make an extension demonstration and extend the chart function to the strategy, as shown in the figure:

Mainly increase the coding part:

• Add a member to the Manager class: objChart
• Add a method to the Manager class: PlotRecords

Some other modifications are made around these two points. You can compare the differences between the two versions and learn the ideas of extended functions.

Python version of commodity futures multi-variable moving average strategy (extended chart)