RX: trasformazione di stato della sequenza, ad es. media mobile esponenziale

Question

Come si può fare in RX una semplice trasformazione di stato di una sequenza?

Supponiamo di voler realizzare una trasformata esponenziale a media mobile di un IO disturbabile.

Ogni volta noisySequence zecche, emaSequence deve spuntare e restituire il valore (previousEmaSequenceValue * (1-lambda) + latestNoisySequenceValue * lambda)

Credo che usiamo soggetti, ma come esattamente?

public static void Main() 
    { 

     var rand = new Random(); 

     IObservable<double> sequence = Observable 
      .Interval(TimeSpan.FromMilliseconds(1000)) 
      .Select(value => value + rand.NextDouble()); 

     Func<double, double> addNoise = x => x + 10*(rand.NextDouble() - 0.5); 

     IObservable<double> noisySequence = sequence.Select(addNoise); 

     Subject<double> exponentialMovingAverage = new Subject<double>(); // ??? 


     sequence.Subscribe(value => Console.WriteLine("original sequence "+value)); 
     noisySequence.Subscribe(value => Console.WriteLine("noisy sequence " + value)); 
     exponentialMovingAverage.Subscribe(value => Console.WriteLine("ema sequence " + value)); 

     Console.ReadLine(); 
    } 

Answer 1

Per molti di questi tipi di calcoli, Buffer è il modo più semplice

var movingAverage = noisySequence.Buffer(/*last*/ 3, /*move forward/* 1 /*at a time*/) 
    .Select(x => (x[0] + x[1] + x[2])/3.0); 

Se avete bisogno di portare lo stato in giro, utilizzare l'operatore Scan, che è come Aggregate tranne che produce valori ogni iterazione.

Answer 2

Ecco come è possibile collegare lo stato a una sequenza. In questo caso calcola la media degli ultimi 10 valori.

var movingAvg = noisySequence.Scan(new List<double>(), 
(buffer, value)=> 
{ 
    buffer.Add(value); 
    if(buffer.Count>MaxSize) 
    { 
     buffer.RemoveAt(0); 
    } 
    return buffer; 
}).Select(buffer=>buffer.Average()); 

Ma si potrebbe utilizzare Window (che Buffer è una sorta di generalizzazione di) per ottenere il vostro media troppo.

noisySequence.Window(10) 
    .Select(window=>window.Average()) 
    .SelectMany(averageSequence=>averageSequence); 

Answer 3

Grazie! Ecco una soluzione usando Scan

const double lambda = 0.99; 
    IObservable<double> emaSequence = noisySequence.Scan(Double.NaN, (emaValue, value) => 
     { 
      if (Double.IsNaN(emaValue)) 
      { 
       emaValue = value; 
      } 
      else 
      { 
       emaValue = emaValue*lambda + value*(1-lambda); 
      } 
      return emaValue; 
     }).Select(emaValue => emaValue);