Calculate Exponential Moving Average in Python (Correctly)

Why is my EMA wrong? A four-point checklist

The EMA formula is EMA_t = alpha*price_t + (1-alpha)*EMA_{t-1}, with alpha = 2/(period+1). Nearly every 'my EMA does not match' report comes down to one of four things: data order, alpha, seed, or the pandas adjust flag.

Work through the table top to bottom. The first row is by far the most common, especially with CSV or API exports that list the latest bar first.

How do I calculate exponential moving average in Python with pandas?

The one-liner is below. ewm() builds an exponentially weighted window, and mean() evaluates it. The key is choosing adjust=False so the result follows the recursive formula.

Pandas offers four ways to specify the decay, and they are equivalent ways of setting alpha. Specify exactly one of them.

import pandas as pd

df = pd.read_csv("prices.csv", parse_dates=["date"], index_col="date")
df = df.sort_index()                      # oldest -> newest
assert df.index.is_monotonic_increasing

N = 10
df["ema"] = df["close"].ewm(span=N, adjust=False).mean()

# Wilder smoothing (RSI, ATR, ADX use this): alpha = 1/N
df["wilder"] = df["close"].ewm(alpha=1/N, adjust=False).mean()

# Mask the warm-up rows
df["ema_masked"] = df["close"].ewm(span=N, adjust=False, min_periods=N).mean()

What does adjust=True vs adjust=False change?

With adjust=False, pandas computes y_0 = x_0 and y_t = (1-alpha)*y_{t-1} + alpha*x_t. This is the textbook recursion and what TradingView, most charting platforms and trading libraries use.

With adjust=True (the default), pandas divides a weighted sum by the sum of weights over the finite history: y_t = sum((1-alpha)^i * x_{t-i}) / sum((1-alpha)^i). Early on, the weights do not yet sum to 1, so the denominator corrects for it. The result is an unbiased weighted average, but it is not the recursive EMA.

Because the missing weight (1-alpha)^t shrinks geometrically, the two modes converge. For a 10-period EMA, the weight on the missing pre-history is 0.818^t, which is about 0.25% after 30 bars. For long series, the difference is invisible, but in the first rows it is not.

import pandas as pd
s = pd.Series([22, 24, 23, 25, 27], dtype=float)
print(s.ewm(span=3, adjust=True).mean().round(4).tolist())
print(s.ewm(span=3, adjust=False).mean().round(4).tolist())
# adjust=True : first values are weighted averages, e.g. row 1 = (24 + 0.5*22)/1.5 = 23.3333
# adjust=False: [22.0, 23.0, 23.0, 24.0, 25.5]

How do I calculate EMA in plain Python from a list of prices?

A loop is enough, and it is also the best way to understand the seed. The textbook version seeds with the SMA of the first N prices, so the first valid value sits at index N-1. Everything before it is undefined, not zero.

Worked example: prices [22, 24, 23, 25, 27], period 3, so alpha = 2/(3+1) = 0.5. The SMA seed is (22+24+23)/3 = 23 at index 2. Then index 3 is 0.5*25 + 0.5*23 = 24, and index 4 is 0.5*27 + 0.5*24 = 25.5. Expected output: [nan, nan, 23, 24, 25.5].

import math

def ema(prices, period, seed="sma"):
    """prices must be ordered oldest -> newest."""
    n = len(prices)
    out = [math.nan] * n
    if period < 1 or n < period:
        return out
    alpha = 2.0 / (period + 1)          # 2.0, not 2: avoids integer division in Python 2
    if seed == "sma":
        e = sum(prices[:period]) / period
        start = period - 1               # off-by-one trap: slice is [:period], index is period-1
    else:                                # seed == "first" matches pandas adjust=False
        e = prices[0]
        start = 0
    out[start] = e
    for i in range(start + 1, n):
        e = alpha * prices[i] + (1 - alpha) * e
        out[i] = e
    return out

print(ema([22, 24, 23, 25, 27], 3))             # [nan, nan, 23.0, 24.0, 25.5]
print(ema([22, 24, 23, 25, 27], 3, "first"))    # [22, 23.0, 23.0, 24.0, 25.5]

Why do TA-Lib, Excel and TradingView give different EMA values?

The formula is the same everywhere. The tools differ in how they start the recursion. TA-Lib and Excel-style textbook EMAs seed with an SMA of the first N values. Pandas adjust=False seeds with the first price. Platforms such as TradingView compute from the first bar of the loaded history, so their numbers also depend on how much history is loaded.

The seed's influence on later values is (1-alpha)^t times the initial gap. After 3N bars it is small, which is why you compare after warm-up and not from row 0.

import numpy as np, pandas as pd

def ema_sma_seed(s: pd.Series, n: int) -> pd.Series:
    """Pandas EMA with the textbook SMA seed (matches talib.EMA)."""
    seeded = s.astype(float).copy()
    seeded.iloc[: n - 1] = np.nan           # leading NaNs stay NaN in ewm
    seeded.iloc[n - 1] = s.iloc[:n].mean()  # replace price n-1 with the SMA
    return seeded.ewm(span=n, adjust=False).mean()

rng = np.random.default_rng(0)
close = pd.Series(100 + rng.normal(0, 1, 500).cumsum())
N = 10

a = ema_sma_seed(close, N).to_numpy()
b = close.ewm(span=N, adjust=False).mean().to_numpy()
warm = 3 * N
print(np.allclose(a[warm:], b[warm:], rtol=1e-2))   # True after warm-up

# With TA-Lib installed:
# import talib
# t = talib.EMA(close.to_numpy(dtype=float), timeperiod=N)
# assert np.allclose(t[N-1:], a[N-1:], atol=1e-9)    # exact match with the SMA-seeded version

How do I update EMA live, and handle gaps and NaNs?

Recomputing the whole series on every bar costs O(n) per update and adds latency for no benefit. The EMA needs only its previous value, so each new bar is an O(1) update. Persist the last EMA with the last bar timestamp so a restart or reconnect can resume without a full replay.

Update on closed bars only. Applying the formula to every tick of a forming bar and then again at the close double-counts the bar.

Missing data needs a decision. In ewm, a NaN input row carries the previous EMA forward. With ignore_na=False (default), a gap still counts as elapsed time, so older weights decay as if the bar existed. With ignore_na=True, the missing bar is skipped entirely. For irregular timestamps, use a time-based decay: ewm(halflife='1h', times=df.index).mean(), where the index must be a datetime index in ascending order.

class StreamingEMA:
    def __init__(self, period):
        self.alpha = 2.0 / (period + 1)
        self.period = period
        self.value = None
        self._warm = []                     # collects first N closes for the SMA seed

    def update(self, price):
        if self.value is None:
            self._warm.append(price)
            if len(self._warm) == self.period:
                self.value = sum(self._warm) / self.period
                self._warm = None
            return self.value               # None until seeded
        self.value = self.alpha * price + (1 - self.alpha) * self.value
        return self.value

e = StreamingEMA(3)
print([e.update(p) for p in [22, 24, 23, 25, 27]])   # [None, None, 23.0, 24.0, 25.5]

How do I test that my EMA implementation is correct?

Three properties catch almost every bug. A constant series must return that constant, a step change must converge to the new level, and the pure-Python, pandas and streaming versions must agree. Add a data-order assertion so reversed input fails loudly.

The step test also checks convergence speed. After a step from 100 to 110, the remaining gap is 10*(1-alpha)^k after k bars, so for N=10 it falls below 1% of the step after about 23 bars (0.818^23 is roughly 0.0099).

1. Constant series: ema([50.0]*100, N) must equal 50.0 everywhere after the seed, with tolerance 1e-12.
2. Step change: 50 bars at 100, then 100 bars at 110; the last value must be within 1e-6 of 110 for N=10.
3. Cross-implementation: the pure-Python SMA-seeded output must match ema_sma_seed (pandas) and StreamingEMA with np.allclose(atol=1e-9).
4. Cross-tool: compare against talib.EMA from index N-1, and against pandas adjust=False only after 3N bars.
5. Order guard: assert the index is strictly increasing before every computation.

import numpy as np

def test_ema():
    N = 10
    const = [50.0] * 100
    out = np.array(ema(const, N))
    assert np.allclose(out[N-1:], 50.0, atol=1e-12)

    step = [100.0] * 50 + [110.0] * 100
    out = np.array(ema(step, N))
    assert abs(out[-1] - 110.0) < 1e-6

    rng = np.random.default_rng(1)
    px = (100 + rng.normal(0, 1, 300).cumsum()).tolist()
    py = np.array(ema(px, N))
    pd_ = ema_sma_seed(pd.Series(px), N).to_numpy()
    assert np.allclose(py[N-1:], pd_[N-1:], atol=1e-9)

    s = StreamingEMA(N)
    live = np.array([s.update(p) for p in px], dtype=float)
    assert np.allclose(live[N-1:], py[N-1:], atol=1e-9)

test_ema()