手把手教你寫一個Python版的K線合成函數

在編寫、使用策略時，經常會使用一些不常用的K線周期數據。然而交易所、數據源又沒有提供這些周期的數據。只能通過使用已有周期的數據進行合成。合成算法已經有一個JavaScript版本了（鏈接），其實移植一段JavaScript代碼為Python版本很簡單。接下來我們一起寫一個Python版本的K線合成算法。

 

JavaScript版本

function GetNewCycleRecords (sourceRecords, targetCycle) {    // K線合成函數
    var ret = []
    
    // 首先獲取源K線數據的周期
    if (!sourceRecords || sourceRecords.length < 2) {
        return null
    }
    var sourceLen = sourceRecords.length
    var sourceCycle = sourceRecords[sourceLen - 1].Time - sourceRecords[sourceLen - 2].Time

    if (targetCycle % sourceCycle != 0) {
        Log("targetCycle:", targetCycle)
        Log("sourceCycle:", sourceCycle)
        throw "targetCycle is not an integral multiple of sourceCycle."
    }

    if ((1000 * 60 * 60) % targetCycle != 0 && (1000 * 60 * 60 * 24) % targetCycle != 0) {
        Log("targetCycle:", targetCycle)
        Log("sourceCycle:", sourceCycle)
        Log((1000 * 60 * 60) % targetCycle, (1000 * 60 * 60 * 24) % targetCycle)
        throw "targetCycle cannot complete the cycle."
    }

    var multiple = targetCycle / sourceCycle


    var isBegin = false 
    var count = 0
    var high = 0 
    var low = 0 
    var open = 0
    var close = 0 
    var time = 0
    var vol = 0
    for (var i = 0 ; i < sourceLen ; i++) {
        // 獲取 時區偏移數值
        var d = new Date()
        var n = d.getTimezoneOffset()

        if (((1000 * 60 * 60 * 24) - sourceRecords[i].Time % (1000 * 60 * 60 * 24) + (n * 1000 * 60)) % targetCycle == 0) {
            isBegin = true
        }

        if (isBegin) {
            if (count == 0) {
                high = sourceRecords[i].High
                low = sourceRecords[i].Low
                open = sourceRecords[i].Open
                close = sourceRecords[i].Close
                time = sourceRecords[i].Time
                vol = sourceRecords[i].Volume

                count++
            } else if (count < multiple) {
                high = Math.max(high, sourceRecords[i].High)
                low = Math.min(low, sourceRecords[i].Low)
                close = sourceRecords[i].Close
                vol += sourceRecords[i].Volume

                count++
            }

            if (count == multiple || i == sourceLen - 1) {
                ret.push({
                    High : high,
                    Low : low,
                    Open : open,
                    Close : close,
                    Time : time,
                    Volume : vol,
                })
                count = 0
            }
        }
    }

    return ret 
}

 

有JavaScript算法，對於Python其實逐行翻譯移植就可以了，遇到JavaScript的內置函數，或者固有方法，對應的去Python中查找對應的方法即可，所以移植還是比較容易的。
算法邏輯完全一模一樣，只是JavaScript的函數調用var n = d.getTimezoneOffset()，移植到Python時，使用Python的time庫中的n = time.altzone代替。其它差異僅僅是語言語法方面的了（例如for循環的使用，布爾值的差別，邏輯與、邏輯非、邏輯或的使用差別等..）。

 

移植后的Python代碼：

import time

def GetNewCycleRecords(sourceRecords, targetCycle):
    ret = []

    # 首先獲取源K線數據的周期
    if not sourceRecords or len(sourceRecords) < 2 : 
        return None

    sourceLen = len(sourceRecords)
    sourceCycle = sourceRecords[-1]["Time"] - sourceRecords[-2]["Time"]

    if targetCycle % sourceCycle != 0 :
        Log("targetCycle:", targetCycle)
        Log("sourceCycle:", sourceCycle)
        raise "targetCycle is not an integral multiple of sourceCycle."

    if (1000 * 60 * 60) % targetCycle != 0 and (1000 * 60 * 60 * 24) % targetCycle != 0 : 
        Log("targetCycle:", targetCycle)
        Log("sourceCycle:", sourceCycle)
        Log((1000 * 60 * 60) % targetCycle, (1000 * 60 * 60 * 24) % targetCycle)
        raise "targetCycle cannot complete the cycle."
    
    multiple = targetCycle / sourceCycle

    isBegin = False
    count = 0 
    barHigh = 0 
    barLow = 0 
    barOpen = 0
    barClose = 0 
    barTime = 0 
    barVol = 0 

    for i in range(sourceLen) : 
        # 獲取時區偏移數值
        n = time.altzone        

        if ((1000 * 60 * 60 * 24) - (sourceRecords[i]["Time"] * 1000) % (1000 * 60 * 60 * 24) + (n * 1000)) % targetCycle == 0 :
            isBegin = True

        if isBegin : 
            if count == 0 : 
                barHigh = sourceRecords[i]["High"]
                barLow = sourceRecords[i]["Low"]
                barOpen = sourceRecords[i]["Open"]
                barClose = sourceRecords[i]["Close"]
                barTime = sourceRecords[i]["Time"]
                barVol = sourceRecords[i]["Volume"]
                count += 1
            elif count < multiple : 
                barHigh = max(barHigh, sourceRecords[i]["High"])
                barLow = min(barLow, sourceRecords[i]["Low"])
                barClose = sourceRecords[i]["Close"]
                barVol += sourceRecords[i]["Volume"]
                count += 1

            if count == multiple or i == sourceLen - 1 :
                ret.append({
                    "High" : barHigh,
                    "Low" : barLow,
                    "Open" : barOpen,
                    "Close" : barClose,
                    "Time" : barTime,
                    "Volume" : barVol,
                })
                count = 0
    
    return ret 

# 測試
def main():
    while True:
        r = exchange.GetRecords()
        r2 = GetNewCycleRecords(r, 1000 * 60 * 60 * 4)      

        ext.PlotRecords(r2, "r2")                                 
        Sleep(1000)     

 

測試

火幣行情圖表

回測合成4小時圖表

以上代碼僅作為學習參考使用，如果用於具體策略中，請根據需求修改、測試。
如有BUG或者改進建議，歡迎留言，十分感謝 o^_^o