MovingAverageCross

Normalized fast-vs-slow moving average distance

MovingAverageCross computes the normalized distance between a fast and slow simple moving average. Positive when fast is above slow (uptrend), negative when fast is below slow (downtrend). It’s the standard feature for trend-following strategies.

Import

python
from horizon.features import MovingAverageCross

Signature

python
MovingAverageCross(
    fast: int = 10,
    slow: int = 30,
    market: str | None = None,
)
fastint
Fast MA window. Must be strictly less than `slow`.
slowint
Slow MA window.

Math

fast_prices = last `fast` prices
slow_prices = last `slow` prices
fast_ma     = Σ fast_prices / fast
slow_ma     = Σ slow_prices / slow
cross       = (fast_ma - slow_ma) / slow_ma

The normalization by slow_ma makes the value scale-independent: the same number means the same thing on a $200 stock and a $5 stock.

Behavior

python
def compute(self, market_id, history, feeds):
    prices = history.last_n_prices(self.slow)
    if len(prices) < self.slow:
        return float("nan")
    fast_prices = prices[-self.fast:]
    fast_ma = sum(fast_prices) / len(fast_prices)
    slow_ma = sum(prices) / len(prices)
    if slow_ma == 0:
        return 0.0
    return (fast_ma - slow_ma) / slow_ma

Interpretation

ValueMeaning
> 0.05Strong uptrend. fast MA is 5%+ above slow MA
0.01 to 0.05Moderate uptrend
-0.01 to 0.01No clear trend
-0.05 to -0.01Moderate downtrend
under -0.05Strong downtrend

The threshold magnitudes depend on asset class. equities rarely exceed |cross| > 0.10 without being in a mania or crash; crypto can easily hit |cross| > 0.20.

Use cases

Classic trend following

python
from horizon import Strategy, Signal, Direction
from horizon.asset_classes import Equity
from horizon.features import MovingAverageCross

class TrendFollower(Strategy):
    asset_classes = [Equity]
    features = {"cross": MovingAverageCross(fast=10, slow=30)}

    def evaluate(self, f, universe):
        signals = []
        for m in universe:
            c = f.cross[m.id]
            if c > 0.02:
                signals.append(Signal.increase(m, edge_bps=40))
            elif c < -0.02:
                signals.append(Signal.decrease(m, edge_bps=40))
        return signals

This is the basic MovingAverageCrossStrategy from horizon.quant.

Multiple timeframe confirmation

python
features = {
    "short": MovingAverageCross(fast=5, slow=15),
    "medium": MovingAverageCross(fast=10, slow=30),
    "long": MovingAverageCross(fast=20, slow=60),
}

def evaluate(self, f, universe):
    # Only trade when all three timeframes agree
    return [
        Signal.increase(m, edge_bps=60)    # higher conviction
        for m in universe
        if f.short[m.id] > 0 and f.medium[m.id] > 0 and f.long[m.id] > 0
    ]

As a regime filter

python
features = {
    "trend": MovingAverageCross(fast=20, slow=60),
    "z": Zscore(window=20),
}

def evaluate(self, f, universe):
    # Mean-revert only when the long-term trend is flat
    return [
        Signal.from_score(m, score=-f.z[m.id], edge_per_stdev=15)
        for m in universe
        if abs(f.trend[m.id]) < 0.02         # not trending
        and abs(f.z[m.id]) > 2.0
    ]

Classic fast/slow pairs

(5, 15) Fast, high turnover. Catches short-term swings. Noisy in chop.
(10, 30) Default. Balanced trend detection.
(20, 50) Medium-term. Smoother, catches multi-week trends.
(50, 200) The classic "golden cross / death cross". Very slow, very strong when it fires.

Pitfalls

Tests

python
def test_uptrend_positive(self) -> None:
    store = FeatureStore()
    prices = [100.0 * math.exp(0.003 * i) for i in range(60)]
    for p in prices:
        store.update_feeds({"T": FeedData(market_id="T", price=p)})
    cross = MovingAverageCross(fast=10, slow=30)
    ns = store.compute({"c": cross}, ["T"], {})
    assert ns.c["T"] > 0

def test_downtrend_negative(self) -> None:
    store = FeatureStore()
    prices = [100.0 * math.exp(-0.003 * i) for i in range(60)]
    for p in prices:
        store.update_feeds({"T": FeedData(market_id="T", price=p)})
    cross = MovingAverageCross(fast=10, slow=30)
    ns = store.compute({"c": cross}, ["T"], {})
    assert ns.c["T"] < 0

Next