simple moving average of "live" stream - fast implementation

Question

In my trading application I have "live ticks" of stock prices. I need to maintain SMA. Let's assume I want SMA of 20 candles, where duration of each candle is 10 seconds. This means that

Every 10 seconds I have "checkpoint" where:

1. I "close" current candle and store average price for the last 10 seconds. Average is (max - min) / 2
2. I "start" new candle and store last price.
3. I clean-up "outdated" candle.

Every tick:

1. I update "last" price of current "forming" candle and recalculate SMA.

So on any tick I need to "recalculate" SMA. In most cases only price of the last candle is changed (cause we using last price). Once per 10 seconds I need a little bit more extra work - I need to "forget" average of the outdated candle, and "store" average of "just created" candle.

Can you suggest how to implement this with lowest latency? Low latency is primary requirement.

Answer 1

I'm not sure whether this is the approach you're looking for but here is the pseudocode for very quick SMAs.

Simple Moving Average:

I assume that your data is coming in the form of some stream and stored in continuous memory location (at least with continuously mappable addresses)

x[1] to x[2000] contain the first 2000 data points

// they don't have to be a real array, could just be a function which manages
// a 'circular' array and maps the 'locations' to real locations in memory.
// Either case, it's small enough to be fully in the cache hopefully
//
// Subsequent prices will be accessible in 'locations x[2001], etc.
// Here we are looking to calculate the MA over the latest 2000 ticks

MA2000(1,2000) = (x[1] + x[2] + ... + x[2000]) / 2000 // Usual average
                                                      // Done only once

MA2000(2,2001) = MA2000(1,2000) * 2000 + x[2001] - x[1]
MA2000(2,2001) /= 2000

That way with two additions and one multiplication ( with 1/2000 ) you can generate subsequent moving averages for the new ticks.

Exponential moving average: That is a decent alternative, as mentioned above:

// For an N-day EMA
Alpha = 2 / (N+1)      // one time calculation

EMA(new) = Alpha * NewTickPrice + (1-Alpha) * EMA(old)

Here it's not really an N-day moving average. It's just a weighted moving average with ~87% weightage to the last N-days, so almost N-days is more like it.

Note on compiler optimizations:

Do note that turning on SSE or AVX options if available will enable massive speedup of these algorithms as multiple calculations can be churned out in a single CPU cycle.

Answer 2

So you need a queue, of pretty much fixed size, where you can efficiently add new items and remove the oldest item (to remove it from your running total). Why not std::queue?

This can sit on top of various containers, but if you really only have 20 elements I suspect a vector would perform well. (Removing an item requires moving all the other items down one - but moving contiguous blocks of memory is fast.) You might want to compare the performance against a deque or list though.

(The answer may depend on what you store for each "candle" - just a single float/double/int, or a more complex structure?)

Answer 3

My implementation. .h:

#pragma once

#include <deque>

class MovingAverage
{
public:
    MovingAverage(int length);
    ~MovingAverage(void);
    void Add(double val);
    void Modify(double value);
    double Value;
    std::deque<double> _candlesExceptNewest;
private:
    MovingAverage(MovingAverage& rhs):
        _length(rhs._length)
    {
        printf("MovingAverage copy-constructor mustn't be executed, exiting.");
        exit(0);
    }

    const int _length;

    int addCounter;
    static const int RECALCULATE_VALUE_MASK;

    double _sumExceptNewest;

    double NewestCandleMedian() {
        return (_newestCandleMin + _newestCandleMax) / 2;
    }
    void RecalculateValue();
    double _newestCandleMin;
    double _newestCandleMax;
};

.cpp:

#include "MovingAverage.h"
#include "CommonsNative.h"

const int MovingAverage::RECALCULATE_VALUE_MASK = 1024 - 1;

MovingAverage::MovingAverage(int length):
    Value(quiet_NaN),
    _length(length),
    addCounter(0)
{
    if (length < 20) {
        std::cout << "Warning, MA is very short, less than 20! length = " 
            << length << std::endl;
    }
}

MovingAverage::~MovingAverage(void)
{
}

void MovingAverage::Add(double val)
{
    ++addCounter;
    if (addCounter & RECALCULATE_VALUE_MASK) {
        _sumExceptNewest = 0;
        for (double val : _candlesExceptNewest)
        {
            _sumExceptNewest += val;
        }
    }

    auto curQueueSize = _candlesExceptNewest.size();
    if (curQueueSize == 0) {
        _newestCandleMax = _newestCandleMin = val;
    }
    _sumExceptNewest += NewestCandleMedian();
    _candlesExceptNewest.push_back(NewestCandleMedian());
    if (curQueueSize == _length) {
        _sumExceptNewest -= _candlesExceptNewest.front();
        _candlesExceptNewest.pop_front();
    }
    _newestCandleMax = _newestCandleMin = val;
    RecalculateValue();
}

void MovingAverage::RecalculateValue()
{
    Value = (_sumExceptNewest + NewestCandleMedian())/(_candlesExceptNewest.size() + 1);
}

void MovingAverage::Modify(double val)
{
    if (_candlesExceptNewest.size() == 0) {
        Add(val);
    } else {
        if (val > _newestCandleMax) {
            _newestCandleMax = val;
        } 
        if (val < _newestCandleMin) {
            _newestCandleMin = val;
        }
    }
}