Truncated EMA

import sys; print(sys.version)
import numpy as np; print(np.__version__)
import pandas as pd; print(pd.__version__)
import talib; print(talib.__version__)

3.12.1 (main, May  6 2025, 20:30:25) [GCC 9.4.0]
2.2.5
2.2.3
0.6.7

EMA

지수 이동 평균(Exponential Moving Average)을 정의하는 것으로 시작하자.

x_t 는 시간 t 에서의 데이터 값이고, y_t 는 시간 t 에서의 지수 이동 평균이다. 최근 값의 기여도를 결정하는 파라미터 \alpha 는 0 \lt \alpha \le 1 을 만족한다.

일반항으로 표현하면 이렇게 된다.

계수의 합은 항상 1이 된다. 따라서 식 (2) 는 가중 평균이다.

Span

여러 문헌에서 EMA(N) 이라는 표기법이 자주 사용된다. 비슷한 표기법은 단순 이동 평균에 대해서도 사용되고 있다. 가령 SMA(3) 은 최근 3개의 데이터를 평균한다는 뜻이다. 즉 N 은 이동 평균에 사용하는 데이터의 범위(span)다. 하지만 지수 이동 평균은 무한한¹ 항을 사용하는 가중 평균이기 때문에, 범위가 고정되지 않는다.

EMA(N) 의 N 은 일종의 "유효(effective) 범위"다. EMA(N) 과 SMA(N) 이 가중치 분포의 어떤 특성을 공유하도록 만들어서, SMA(N) 과 EMA(N) 을 바꿔 사용하기 편리하도록 만들려는 것이다. 이 "어떤 특성" 이 "무게 중심(Center of Mass)"이다.² 무게 중심이란 대략 가중치가 어디를 중심으로 분포하는지에 대한 측정이다. 가중 평균에 사용되는 과거 데이터들과의 거리를 같은 가중치로 평균한 것이다.

가령 SMA(N) 의 가중치는 모두 \frac{1}{N} 이고, 거리는 N-1, N-2, ..., 1, 0 이다. 따라서 무게 중심은 이렇게 구할 수 있다.

마찬가지로, EMA 의 무게 중심은 이렇게 된다.

흔히 사용되는 값은 C_t 의 극한값이다.

C_\infty 가 SMA(N) 의 무게 중심과 일치하도록 만드는 \alpha 를 구하면

따라서 EMA(N) 을 상기한 \alpha 값을 갖는 EMA 로 정의하면 (어느 정도는) SMA(N) 과 교환적으로 사용할 수 있다.

TA-Lib

TA-Lib 를 사용하면 간단하게 계산할 수 있다.³

close = pd.Series([0, 1, 2, 3, 4])

talib.set_compatibility(1)  # EMA 종류 선택
talib.EMA(close, timeperiod=3)

0       NaN
1       NaN
2    1.2500
3    2.1250
4    3.0625
dtype: float64

timeperiod 는 앞 절에서 소개한 span(N) 이다. 따라서 이 계산에 사용된 \alpha 는 0.5 다.

계산에는 최소 timeperiod 개의 데이터가 필요하므로, 그만큼을 채우지 못하는 앞쪽 timeperiod - 1 개의 초깃값들은 NaN 으로 채워졌다.⁴

Pandas

Pandas 역시 함수를 제공한다.⁵

close.ewm(span=3, adjust=False).mean()

0    0.0000
1    0.5000
2    1.2500
3    2.1250
4    3.0625
dtype: float64

ewm() 함수는 다양한 옵션을 제공하는데, 상기한 호출 방식이 식 (1) 및 TA-Lib 와 유사하다. timeperiod 와 span 은 정확히 같은 역할을 수행하는 값이다.

TA-Lib 와는 달리 초깃값들을 NaN 으로 채우지 않고 식 (1) 이 정의하는 모든 값들을 결과로 제공한다.⁶

데이터 길이 의존성

앞에서도 언급했듯이 EMA 는 무한항을 사용하기 때문에, 계산에 사용할 데이터를 얼마나 로드하는지에 따라 결과가 달라진다.

데이터의 길이를 두 배로 늘려보자.

close = pd.Series([0, 1, 2, 3, 4, 0, 1, 2, 3, 4])

df = pd.DataFrame(
    {
        "x": close,
        "SMA(3)": talib.SMA(close, timeperiod=3),
        "EMA(3)": talib.EMA(close, timeperiod=3),
    }
)
df

EMA 에서는 SMA 에서와 달리 인덱스 값이 2,3,4 일 때와 7,8,9 일 때의 이동 평균 값이 달라진다. 식 (1) 에서 x_0 는 데이터의 처음을 뜻한다. 이는 우리가 어느 시점부터의 데이터를 로드해서 계산할 것인지에 따라 y_t 가 달라진다는 뜻이다. 가령 최근 한 달간의 주식 가격 데이터로 계산한 어제 EMA 와 최근 일 년간의 주식 가격 데이터로 계산한 어제 EMA 가 다르다는 뜻이다. 큰 차이는 나지 않겠지만, 최근 10년 데이터로 백테스팅할 때와 라이브 데이터로 알고리즘 트레이딩할 때 달라질 수 있다는 것은 숙면을 방해하는 성질이다.

Truncated EMA

식 (2) 를 변형하여, n 개의 항만 포함하도록 만들자.

y_{n-1} 은 n 개의 항으로 이루어진다. 각 항의 계수를 w_k 라고 하면

이제 Truncated EMA 는 이 가중치를 사용한 가중 이동 평균으로 정의한다.

가중치를 사용하면 Pandas 에서 쉽게 Truncated EMA 를 계산할 수 있다.

가중치를 계산하는 함수는 이런 식으로 구성할 수 있다.

def truncated_ema_weights(n: int, alpha: float | None = None) -> list[float]:
    if alpha is None:
        alpha = 2 / (n + 1)
    b = 1 - alpha
    c = b ** (n - 1)
    w = [c]
    c *= alpha
    for i in range(1, n):
        c /= b
        w.append(c)
    return w


truncated_ema_weights(3)

[0.25, 0.25, 0.5]

\alpha 를 \frac{2}{n+1} 로 고정해야 할 필요는 없다. 하지만 지정하지 않으면 EMA 의 관례를 따르도록 한다.

이제 Pandas 의 Rolling apply 에 전달할 수 있는 함수 팩토리를 만든다.⁷

def truncated_ema(n: int, alpha: float = None):
    weights = np.array(truncated_ema_weights(n, alpha))

    return weights.dot


df["Truncated EMA(3)"] = close.rolling(3).apply(truncated_ema(3), raw=True)
df

Truncated EMA(3) 열을 보면, SMA 와 마찬가지로 인덱스 값이 2,3,4 일 때와 7,8,9 일 때의 이동 평균 값이 같음을 확인할 수 있다.

맺음말

시작할 때 생각했던 것과는 조금 다른 곳에 도착했다.

원래는 EMA 를 먼저 계산한 다음, n 시간 이전 값과의 재귀적 관계를 활용하여 후처리하는 방식을 사용해 왔다. 이 방법에 대한 기록을 남겨, 미래의 기억을 보조하려는 것이 애초의 계획이었다. 하지만 결정적으로 방향이 바뀌게 된 것은, Polars 의 rolling_mean() 에 weights 인자가 있음을 발견했을 때다. 그럼에도 Pandas 를 위주로 작업한 것은, 인터페이스 불안정성 때문이다. 2023년 후반에 한 지자체를 위해 Polars 로 작업한 적이 있다. 한 10개월쯤 후에 AS 요청이 있어 들여다보니, Polars 의 최신 버전과 호환되지 않는 부분이 상당함을 발견했다. 프로젝트 당시의 버전이 1.0 이전이라, 예상할 수 있는 문제이기는 하지만, 그 후로 선뜻 선택하지 못하고 있다. 내년에는 트라우마를 극복할 수 있지 않을까?