유튜브 neurotrader의 3 Must-Know Algorithms for Automating Chart Pattern Trading in Python를 참고한 글이다.

Intro

기술적 분석을 위한 차트 패턴 분석에 가장 기초적인 것은 지역 고점과 지역 저점을 판단하는 것이다. 많은 차트 패턴이 지역 고점과 지역 저점을 통해 만들어진다.

Rolling Window, Directional Change, Perceptually Important Points 알고리즘에 대해 알아본다.

Rolling Window

붉은 점 주변을 살펴보면 다른 지점은 붉은 점보다 낮다.
붉은 점이 지역 고점이 된다.

좁은 범위를 이동시키면서 지역 고점들을 찾는다.
이를 1차 지역 고점이라 부른다.

범위를 조금 더 넓혀보자. 붉은 점이 줄어든다. 이를 2차 지역고점이라 한다.

범위를 충분히 넓히고 나면 지역 고점의 수가 감소하고, 선별적이 되는 것을 볼 수 있다.

지역 저점 역시 같은 방식으로 찾는다.

이때 유의할 점은, 붉은 지역 고점이 확정되는 시점은 초록색, 즉 약간의 지연이 있은 후이다.

코드 분석

import matplotlib.pyplot as plt

# 현재 인덱스에서 지역 고점이 탐지되었는지 확인한다.
def rw_top(data: np.array, curr_index: int, order: int) -> bool:
    if curr_index < order * 2 + 1:
        return False

    top = True
    k = curr_index - order
    v = data[k]
    for i in range(1, order + 1):
        if data[k + i] > v or data[k - i] > v:
            top = False
            break
    
    return top

# 현재 인덱스에서 지역 저점이 탐지되었는지 확인한다.
def rw_bottom(data: np.array, curr_index: int, order: int) -> bool:
    if curr_index < order * 2 + 1:
        return False

    bottom = True
    k = curr_index - order
    v = data[k]
    for i in range(1, order + 1):
        if data[k + i] < v or data[k - i] < v:
            bottom = False
            break
    
    return bottom

def rw_extremes(data: np.array, order:int):
    # 지역 고점과 지역 저점을 찾는 Window를 Rolling한다.
    tops = []
    bottoms = []
    for i in range(len(data)):
        if rw_top(data, i, order):
            # top[0] = 감지가 확정된 인덱스
            # top[1] = 고점의 인덱스
            # top[2] = 고점의 가격
            top = [i, i - order, data[i - order]]
            tops.append(top)
        
        if rw_bottom(data, i, order):
            # bottom[0] = 감지가 확정된 인덱스
            # bottom[1] = 저점의 인덱스
            # bottom[2] = 저점의 가격
            bottom = [i, i - order, data[i - order]]
            bottoms.append(bottom)
    
    return tops, bottoms



if __name__ == "__main__":
    data = pd.read_csv('BTCUSDT86400.csv')
    data['date'] = data['date'].astype('datetime64[s]')
    data = data.set_index('date')

    tops, bottoms = rw_extremes(data['close'].to_numpy(), 10)
    data['close'].plot()
    idx = data.index
    for top in tops:
        plt.plot(idx[top[1]], top[2], marker='o', color='green')

    for bottom in bottoms:
        plt.plot(idx[bottom[1]], bottom[2], marker='o', color='red')


    plt.show()  

위에서 Window의 크기는 Order가 결정한다.
가령 k번째 인덱스에서 탐지하는 경우 k-order부터 k+order까지를 탐색한다.

Directional Change

Directional Change 알고리즘은 지그재그 알고리즘으로 불리기도 한다. Directional Change 알고리즘은 최근의 저점 혹은 고점에서 반대로 향할 때(Retraced) 저점 및 고점을 확정한다.

표시한 지점은 현재의 지역 저점이다.

현재의 지역 저점으로부터 가격이 상승하고 있다. 현재의 고점이 포함된 캔들을 주황색으로 표시했다.

현재의 고점보다 3% 낮은 지점을 기준선(Confirmation Line)으로 지정한다.

가격이 지속적으로 상승함에 따라 고점이 달라지며, 기준선 역시 높아진다.

가격이 하락하면서 노란색 기준선을 돌파하였고, 이에 따라 마지막 고점을 지역 고점으로 확정했다.
새로운 캔들은 저점이 되며, 지역 저점보다 3% 높은 곳을 파란색 기준선으로 한다.

가격이 낮아짐에 따라 파란색 기준선도 낮아지고,

해당 기준선을 돌파하였을 때, 저점을 확정하고 새로운 노란색 기준선을 지정한다.

코드 분석

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import mplfinance as mpf

def directional_change(close: np.array, high: np.array, low: np.array, sigma: float):
    # up_zig : 이전 극한값이 저점이었는지를 나타내는 flag이다. 기본값을 True로 하되, 초기의 극한값 탐지가 부정확할 수 있다.
    # tmp_max : 이전 최대값을 담은 값이다. 초기값은 최초 캔들의 최고값이다.
    # tmp_max : 이전 최솟값을 담은 값이다. 초기값은 최초 캔들의 최솟값이다.
    up_zig = True # Last extreme is a bottom. Next is a top. 
    tmp_max = high[0]
    tmp_min = low[0]
    tmp_max_i = 0
    tmp_min_i = 0

    tops = []
    bottoms = []

    for i in range(len(close)):
        if up_zig: # Last extreme is a bottom
            if high[i] > tmp_max:
                # New high, update 
                tmp_max = high[i]
                tmp_max_i = i
            elif close[i] < tmp_max - tmp_max * sigma: 
                # 주가가 하락하고 있을 때, 현재의 가격(close)이 기존 고점보다 sigma%만큼 벗어났는지를 판별한다.
                # Price retraced by sigma %. Top confirmed, record it
                # top[0] = confirmation index
                # top[1] = index of top
                # top[2] = price of top
                top = [i, tmp_max_i, tmp_max]
                tops.append(top)

                # Setup for next bottom
                up_zig = False
                tmp_min = low[i]
                tmp_min_i = i
        else: # Last extreme is a top
            if low[i] < tmp_min:
                # New low, update 
                tmp_min = low[i]
                tmp_min_i = i
            elif close[i] > tmp_min + tmp_min * sigma: 
                # Price retraced by sigma %. Bottom confirmed, record it
                # bottom[0] = confirmation index
                # bottom[1] = index of bottom
                # bottom[2] = price of bottom
                bottom = [i, tmp_min_i, tmp_min]
                bottoms.append(bottom)

                # Setup for next top
                up_zig = True
                tmp_max = high[i]
                tmp_max_i = i

    return tops, bottoms

def get_extremes(ohlc: pd.DataFrame, sigma: float):
    tops, bottoms = directional_change(ohlc['close'], ohlc['high'], ohlc['low'], sigma)
    tops = pd.DataFrame(tops, columns=['conf_i', 'ext_i', 'ext_p'])
    bottoms = pd.DataFrame(bottoms, columns=['conf_i', 'ext_i', 'ext_p'])
    tops['type'] = 1
    bottoms['type'] = -1
    extremes = pd.concat([tops, bottoms])
    extremes = extremes.set_index('conf_i')
    extremes = extremes.sort_index()
    return extremes

PIP

가장 시작점과 끝점은 항상 지각적으로 중요한 지점이 된다. 이를 점으로 찍고 잇는다.

두 점을 선과 점들 사이의 거리를 파악하며 새로운 지각적으로 중요한 지점을 찾는다. 가장 높은 고점이 선과의 거리가 가장 멀기에 해당 지점이 지각적으로 중요한 지점으로 선택된다.

3개의 점을 이은 선과 점들 사이의 거리를 파악하면 저점이 지각적으로 중요한 지점으로 선택된다.

5개의 지각적으로 중요한 지점을 찾고 이은 점 그래프는 위와 같이 나온다.

코드 분석

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import mplfinance as mpf

def find_pips(data: np.array, n_pips: int, dist_measure: int):
    # dist_measure
    # 1 = Euclidean Distance
    # 2 = Perpindicular Distance
    # 3 = Vertical Distance

    pips_x = [0, len(data) - 1]  # Index
    pips_y = [data[0], data[-1]] # Price

    for curr_point in range(2, n_pips):

        md = 0.0 # Max distance
        md_i = -1 # Max distance index
        insert_index = -1

        for k in range(0, curr_point - 1):

            # Left adjacent, right adjacent indices
            left_adj = k
            right_adj = k + 1

            time_diff = pips_x[right_adj] - pips_x[left_adj]
            price_diff = pips_y[right_adj] - pips_y[left_adj]
            slope = price_diff / time_diff
            intercept = pips_y[left_adj] - pips_x[left_adj] * slope;

            for i in range(pips_x[left_adj] + 1, pips_x[right_adj]):
                
                d = 0.0 # Distance
                if dist_measure == 1: # Euclidean distance
                    d =  ( (pips_x[left_adj] - i) ** 2 + (pips_y[left_adj] - data[i]) ** 2 ) ** 0.5
                    d += ( (pips_x[right_adj] - i) ** 2 + (pips_y[right_adj] - data[i]) ** 2 ) ** 0.5
                elif dist_measure == 2: # Perpindicular distance
                    d = abs( (slope * i + intercept) - data[i] ) / (slope ** 2 + 1) ** 0.5
                else: # Vertical distance    
                    d = abs( (slope * i + intercept) - data[i] )

                if d > md:
                    md = d
                    md_i = i
                    insert_index = right_adj

        pips_x.insert(insert_index, md_i)
        pips_y.insert(insert_index, data[md_i])

    return pips_x, pips_y


if __name__ == "__main__":
    data = pd.read_csv('BTCUSDT86400.csv')
    data['date'] = data['date'].astype('datetime64[s]')
    data = data.set_index('date')

    i = 1198
    x = data['close'].iloc[i-40:i].to_numpy()
    pips_x, pips_y = find_pips(x, 5, 2)

    pd.Series(x).plot()
    for i in range(5):
        plt.plot(pips_x[i], pips_y[i], marker='o', color='red')

    plt.show()

선과의 거리를 측정하기 위해 유클리드 거리, 직각거리, 수직거리 등을 사용할 수 있다.

셋 모두 결과는 비슷하게 나오지만, 경우에 따라 성능이 달라질 수 있다.

결론

롤링 윈도우 알고리즘은 시간에 따라 지역 고점과 지역 저점을 찾아낸다. 하지만 추세로부터 얼마나 높고, 낮은지는 판별하지 않는다.
디렉셔널 체인지 알고리즘은 시간에 따라 지역 고점과 지역 저점을 찾아냄과 더불어, 가격 변동의 정도를 파라미터로 두는 차이가 있다.
PIP 알고리즘은 시계열의 영향을 크게 받지 않고 시작과 끝 지점 사이의 중간에 위치한 점들을 지역 고점과 지역 저점으로 찾아낸다. 또한 데이터의 일부만으로도 실행할 수 있는 알고리즘이다.

셋 모두 기술적 분석에 유용하며, 경우에 따라서는 특정한 알고리즘이 최선의 선택일 수 있다. 가령 디렉셔널 체인지 알고리즘은 지지 및 저항선을 식별하기에 유용하고, PIP 알고리즘은 데이터 마이닝에 특히 유용하다.