K-line 합성 함수의 Python 버전을 단계별로 작성하는 방법을 알려드립니다.

전략을 작성하고 사용할 때 흔하지 않은 K-라인 사이클 데이터가 종종 사용됩니다. 하지만 거래소와 데이터 출처에서는 이 기간에 대한 데이터를 제공하지 않습니다. 기존 사이클의 데이터를 사용해서만 합성이 가능합니다. 합성 알고리즘에는 이미 JavaScript 버전이 있습니다(링크), 실제로 JavaScript 코드를 Python 버전으로 이식하는 것은 매우 간단합니다. 다음으로, K-라인 합성 알고리즘의 Python 버전을 작성해 보겠습니다.

자바스크립트 버전


            
            
            
  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의 시간 라이브러리를 사용하세요.n = time.altzone바꾸다. 그 밖의 차이점은 언어 구문에만 존재합니다(for 루프의 사용, 부울 값의 차이, 논리적 and, 논리적 not, 논리적 or 등의 사용의 차이 등).

이식된 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)