파이썬 코딩으로 말하는 데이터 분석 - 10. DTW (Dynamic time wrapping)

데이터 분석에 대한 기본적인 감과 덤으로 파이썬 코딩에 대한 감을 익히기 위한 강좌입니다. 
개발자는 코드로 이해하는게 가장 빠른 길이며, 오래 기억하는 길이라 믿습니다.

순서 

1. 통계 - 카운팅,min,max,평균,중앙값,산포도,분산,편차,공분산,상관관계 

2. 가설과 추론 (베이지언 - 사후확률,우도) 

3. 군집화 (K-Means)

4. 연관 (Apriori)

5. 함수형으로 데이터 다루기 

6.경사하강법

7. 회귀분석

8. 은닉 마코프법 (HMM) 

9. k-NN

10. DTW 

DTW (Dynamic time wrapping) 란?

두개의 시계열 데이터가 있다고 할 때 그 둘간의 유사도를 알아내기 위한 알고리즘 중 하나 이다.

예를들어 IoT 센서에서 시간마다 생성되는 데이터를 비교하여, 현재 데이터흐름은 과거의 어떤 흐름과 매칭되는지 체크하는데 사용된다. 즉 에어컨을 켰을 때 전력패턴을 기억해서, 후에 내 스마트폰을 통해 외부에서 전력사용을 감지해서 누가 내 집에서 에어컨을 켠다던가 하는?? 도둑놈?? 혹은 냉장고 문이 열렸을 때의 패턴을 감지하여  혹시 아들내미가 냉장고 문을 열어두고 가진 않았는지? (물론 DTW 만 가지고서 저런 NILM 문제를 해결할 수 있는것은 아니다. ) 가장 많이 떠올릴 수 있는 데이터인 주식데이터에도 사용 될 수 있다.

다음 그림을 보자. 

보통 떠올리기에는 0~1분 사이의 1초간의 데이터가 있다고 할 때, 0초,1초,2초 각각의 데이터의 차의 제곱의 평균  를 이용해서 차이를 구하는게 가장 먼저 떠오를 것이다. 하지만 그렇게 되면 위 그림의 오른쪽과 같은 좀 비틀어진 상황에서는 영 맥을 못추는 패턴매칭이 될 것이다.DTW 는 이럴때 사용되는 시계열 패턴 매칭 알고리즘이다. 어떻게 해결 할 수 있을지 잠시 생각을 해보자.

자신과 같은 시간 index 를 가진 요소와의 비교 뿐 만 아니라, 그 주변의 다른 요소와도 비교를 해서 더 비슷한 요소를 자신의 비교쌍으로 놓는다면 어느정도 극복되지 않을까? 자세한것은 코드를 보고 이해해보자.

사실 쉬운 알고리즘이라도 글로 보면 이해하기가 쉽지 않다. 

코드로 말해요  

# -*- coding: utf-8 -*-

from math import *
import numpy as np
import sys


def DTW(A, B, window=sys.maxint, d=lambda x, y: abs(x - y)):
    # 비용 행렬 초기화 
    A, B = np.array(A), np.array(B)
    M, N = len(A), len(B)
    cost = sys.maxint * np.ones((M, N))

    # 첫번째 로우,컬럼 채우기
    cost[0, 0] = d(A[0], B[0])
    for i in range(1, M):
        cost[i, 0] = cost[i - 1, 0] + d(A[i], B[0])

    for j in range(1, N):
        cost[0, j] = cost[0, j - 1] + d(A[0], B[j])
    # 나머지 행렬 채우기 
    for i in range(1, M):
        for j in range(max(1, i - window), min(N, i + window)):
            choices = cost[i - 1, j - 1], cost[i, j - 1], cost[i - 1, j]
            cost[i, j] = min(choices) + d(A[i], B[j])

    # 최적 경로 구하기 
    n, m = N - 1, M - 1
    path = []

    while (m, n) != (0, 0):
        path.append((m, n))
        m, n = min((m - 1, n), (m, n - 1), (m - 1, n - 1), key=lambda x: cost[x[0], x[1]])

    path.append((0, 0))
    return cost[-1, -1], path


def main():
    A = np.array([1,2,3,4,2,3])
    B = np.array([7,8,5,9,11,9])
 
    cost, path = DTW(A, B, window = 6)
    print 'Total Distance is ', cost
    import matplotlib.pyplot as plt
    offset = 5
    plt.xlim([-1, max(len(A), len(B)) + 1])
    plt.plot(A)
    plt.plot(B + offset)
    for (x1, x2) in path:
        plt.plot([x1, x2], [A[x1], B[x2] + offset])
    plt.show()

if __name__ == '__main__':
    main()

소스참고: https://gist.github.com/bistaumanga/6023705

설명

아래와 같은 2개의 타임시리즈가 있다고 하자.

1. 비용 행렬 초기화 (6x6의 행렬을 만들고, max_int 값으로 채운다)

2 첫번째 로우와 컬럼을 각 타임스탬프(A,B)의 값을 이용하여 채운다.

 -> (0,0) 행렬은 A , B 요소의 첫번째 값의 차로 
 -> A 의 요소와 B의 첫번째 요소간의 차이의 절대값에 이전 값을 더해서 로우를, 그 반대로 컬럼을

3.행렬의 나머지 부분을 채운다.

 -> choices : cost[i - 1, j - 1], cost[i, j - 1], cost[i - 1, j]  이 3요소를 선택한다. (자신을 둘러싼 3방향요소)
 -> choices 에서 선택된 3가지 요소중 작은것과  A[i], B[j] 의 차의 절대값을 더해준다. 

여기서 34라는 DTW 라는 값이 구해졌다.
위에서 B를 np.array([1,3,3,4,3,3]) 이런것으로 한다면 DTW 가 2밖에 되지 않을 것이다.

4. 가장 최적의 경로를 구한다.

 -> 마지막부터 시작해서 주위의 가장 작은 숫자의 칸으로 이동한다.

위의 경우처럼 대각선으로 이동한다는 것은 A,B 두 시계열 데이터가 일정하게 변화했다는 의미이다.

경로를 차트로 그리면 다음과 같다.

B를 np.array([1,3,3,4,3,3]) 이런것으로 한다면 차트는 다음과 같을 것이다.