экспоненциальная скользящая средняя

Question

У меня есть экспоненциальная скользящая средняя, ​​которая вызывается миллионы раз, и, следовательно, является самой дорогой частью моего кода:

double _exponential(double price[ ], double smoothingValue, int dataSetSize)
{
    int i;
    double cXAvg;
    cXAvg = price[ dataSetSize - 2 ] ;  

    for (i= dataSetSize - 2; i > -1; --i)   
        cXAvg += (smoothingValue * (price[ i ] - cXAvg)) ;

     return ( cXAvg) ;
}

Есть ли более эффективный способ кодирования, чтобы ускорить процесс? У меня многопоточное приложение, и я использую Visual C ++.

Спасибо.

Answer 1

Ой!

Конечно, многопоточность может помочь. Но вы почти наверняка сможете улучшить производительность однопоточного станка.

Во-первых, вы рассчитываете это в неправильном направлении. Только самые современные машины могут делать предварительную выборку с отрицательным шагом. Почти все машины быстрее для юнитов. То есть изменение направления массива так, чтобы вы сканировали с низкого на высокое, а не с высокого на низкое, почти всегда лучше.

Далее, немного переписав - пожалуйста, позвольте мне сократить имена переменных, чтобы было легче набирать:

avg = price[0]

for i
    avg = s * (price[i] - avg)

Кстати, я начну использовать сокращения p для цены и s для сглаживания, чтобы сохранить типизацию. Я ленивый ...

avg0 = p0
avg1 = s*(p1-p0)
avg2 = s*(p2-s*(p1-p0)) = s*(p2-s*(p1-avg0))
avg3 = s*(p3-s*(p2-s*(p1-p0))) = s*p3 - s*s*p2 + s*s*avg1

и вообще

avg[i] = s*p[i] - s*s*p[i-1] + s*s*avg[i-2]

предварительный расчет s * s

вы могли бы сделать

avg[i] = s*p[i] - s*s*(p[i-1] + s*s*avg[i-2])

но это, вероятно, быстрее сделать

avg[i] = (s*p[i] - s*s*p[i-1]) + s*s*avg[i-2])

Задержка между avg [i] и avg [i-2] составляет 1, умножение и сложение, а не вычитание и умножение между avg [i] и avg [i-1]. То есть более чем в два раза быстрее.

Как правило, вы хотите переписать повторение так, чтобы avg [i] вычислялось в терминах avg [j] для j настолько далеко, насколько это возможно, без заполнения машины, ни исполнительных блоков, ни регистров.
В целом вы делаете больше умножений, чтобы получить меньше цепочек умножений (и вычитаний) на критическом пути. Переход от avg [i-2] к avg [i [легко, вы можете сделать три и четыре. Точно, как далеко зависит от того, какая у вас машина и сколько у вас регистров.

И задержка сумматора с плавающей запятой и множителя. Или, что еще лучше, у вас есть вкус комбинированных команд умножения и сложения - они есть у всех современных машин. Например. если MADD или MSUB имеет длину 7 циклов, вы можете выполнить до 6 других вычислений в его тени, даже если у вас есть только одна единица с плавающей запятой. Полностью конвейерный. И так далее. Меньше, если конвейеризовать каждый второй цикл, как обычно для двойной точности на старых чипах и графических процессорах. Код на ассемблере должен быть конвейерным, чтобы разные итерации цикла перекрывались. Хороший компилятор должен сделать это за вас, но вам, возможно, придется переписать код C, чтобы получить лучшую производительность.

Кстати: я НЕ имею в виду, что вы должны создавать массив avg []. Вместо этого вам понадобятся два средних значения, если avg [i] рассчитывается в терминах avg [i-2] и так далее. Вы можете использовать массив avg [i], если хотите, но я думаю, что вам нужно иметь только 2 или 4 avgs, call, creative, avg0 и avg1 (2, 3 ...) и «вращать» их.

avg0 = p0
avg1 = s*(p1-p0)
/*avg2=reuses*/avg0 = s*(p2-s*(p1-avg0))
/*avg3=reusing*/avg3 = s*p3 - s*s*p2 + s*s*avg1
for i from 2 to N by 2 do
    avg0 = s*p3 - s*s*p2 + s*s*avg0
    avg1 = s*p3 - s*s*p2 + s*s*avg1

Такая хитрость, разбивая аккумулятор или среднее на два или более, объединение нескольких этапов повторения, характерное для высокопроизводительного кода.

О, да: предварительные расчеты * и т. Д.

Если бы я сделал это правильно, с бесконечной точностью это было бы идентично. (Проверьте меня дважды, пожалуйста.)

Однако в FP с конечной точностью ваши результаты могут отличаться, надеюсь, лишь незначительно, из-за различных округлений. Если развертывание правильное и ответы существенно отличаются, возможно, у вас нестабильный алгоритм. Ты тот, кто будет знать.

Примечание: ошибки округления с плавающей запятой изменят младшие биты вашего ответа. Как из-за перестановки кода, так и с помощью MADD. Я думаю, что это нормально, но вы должны решить.

Примечание: вычисления для avg [i] и avg [i-1] теперь независимы. Таким образом, вы можете использовать SIMD набор команд, как Intel SSE2, который позволяет работать одновременно с двумя 64-битными значениями в 128-битном регистре. Это будет хорошо почти в 2 раза, на машине с достаточным количеством ALU.

Если у вас достаточно регистров, чтобы переписать avg [i] в ​​терминах avg [i-4] (и я уверен, что вы делаете на iA64), тогда вы можете пойти в 4 раза больше, если у вас есть доступ к машине, как 256-битный AVX.

На графическом процессоре ... вы можете пойти на более глубокие повторения, переписать avg [i] в ​​терминах avg [i-8] и т. Д.

В некоторых графических процессорах есть инструкции, которые рассчитывают AX + B или даже AX + BY как одну инструкцию. Хотя это более распространено для 32-битной, чем для 64-битной точности.

В какой-то момент я, вероятно, начал бы спрашивать: хотите ли вы сделать это по нескольким ценам одновременно? Это не только поможет вам с многопоточностью, но и подойдет для работы на GPU. И с использованием широкого SIMD.

Незначительное позднее добавление

Я немного смущен тем, что не применил правило Хорнера к таким выражениям, как

avg1 = s*p3 - s*s*p2 + s*s*avg1

1061 * дает *

avg1 = s*(p3 - s*(p2 + avg1))

немного эффективнее. немного другие результаты с округлением.

В мою защиту любой достойный компилятор должен сделать это для вас.

Но правило Хрнера делает цепочку зависимостей глубже с точки зрения умножения. Возможно, вам придется развернуть и конвейеризовать цикл еще несколько раз. Или вы можете сделать

avg1 = s*p3 - s2*(*p2 + avg1)

где вы рассчитываете

s2 = s*s

Answer 2

Вот наиболее эффективный способ, хотя в C # вам потребуется перенести его на C ++, который должен быть очень простым. Он вычисляет EMA и Slope на лету наиболее эффективным способом.

public class ExponentialMovingAverageIndicator
{
    private bool _isInitialized;
    private readonly int _lookback;
    private readonly double _weightingMultiplier;
    private double _previousAverage;

    public double Average { get; private set; }
    public double Slope { get; private set; }

    public ExponentialMovingAverageIndicator(int lookback)
    {
        _lookback = lookback;
        _weightingMultiplier = 2.0/(lookback + 1);
    }

    public void AddDataPoint(double dataPoint)
    {
        if (!_isInitialized)
        {
            Average = dataPoint;
            Slope = 0;
            _previousAverage = Average;
            _isInitialized = true;
            return;
        }

        Average = ((dataPoint - _previousAverage)*_weightingMultiplier) + _previousAverage;
        Slope = Average - _previousAverage;

        //update previous average
        _previousAverage = Average;
    }
}