How is IObservable<double>.Average supposed to work?
- by Dan Tao
Update
Looks like Jon Skeet was right (big surprise!) and the issue was with my assumption about the Average extension providing a continuous average (it doesn't).
For the behavior I'm after, I wrote a simple ContinuousAverage extension method, the implementation of which I am including here for the benefit of others who may want something similar:
public static class ObservableExtensions {
private class ContinuousAverager {
private double _mean;
private long _count;
public ContinuousAverager() {
_mean = 0.0;
_count = 0L;
}
// undecided whether this method needs to be made thread-safe or not
// seems that ought to be the responsibility of the IObservable (?)
public double Add(double value) {
double delta = value - _mean;
_mean += (delta / (double)(++_count));
return _mean;
}
}
public static IObservable<double> ContinousAverage(this IObservable<double> source) {
var averager = new ContinuousAverager();
return source.Select(x => averager.Add(x));
}
}
I'm thinking of going ahead and doing something like the above for the other obvious candidates as well -- so, ContinuousCount, ContinuousSum, ContinuousMin, ContinuousMax ... perhaps ContinuousVariance and ContinuousStandardDeviation as well? Any thoughts on that?
Original Question
I use Rx Extensions a little bit here and there, and feel I've got the basic ideas down.
Now here's something odd: I was under the impression that if I wrote this:
var ticks = Observable.FromEvent<QuoteEventArgs>(MarketDataProvider, "MarketTick");
var bids = ticks
.Where(e => e.EventArgs.Quote.HasBid)
.Select(e => e.EventArgs.Quote.Bid);
var bidsSubscription = bids.Subscribe(
b => Console.WriteLine("Bid: {0}", b)
);
var avgOfBids = bids.Average();
var avgOfBidsSubscription = avgOfBids.Subscribe(
b => Console.WriteLine("Avg Bid: {0}", b)
);
I would get two IObservable<double> objects (bids and avgOfBids); one would basically be a stream of all the market bids from my MarketDataProvider, the other would be a stream of the average of these bids.
So something like this:
Bid Avg Bid
1 1
2 1.5
1 1.33
2 1.5
It seems that my avgOfBids object isn't doing anything. What am I missing? I think I've probably misunderstood what Average is actually supposed to do. (This also seems to be the case for all of the aggregate-like extension methods on IObservable<T> -- e.g., Max, Count, etc.)
