Dynamic data is now a portable class library available on most platforms. See below.
Additionally there is a separate dotnet 4.0 library because I know there are enterprises out there stuck in the old days (investment banks maybe?).
Now you can do some cool rx for collections stuff of WP8, IOS and Android as well as windows desktop and server.
I will explain what the plinq assemblies enable in a post in the near future.
Intrinsic to collection change notifications, items can be notified by add, update and remove events. If ordering is supported a collection will support a move notification. Of course Dynamic Data supports these. Yet something which is often overlooked is data or functions of data are sometimes by necessity mutable. How can a collection which notifies respond to mutable changes?
To deal with this scenario, dynamic data introduces the concept of an Evaluate notification. This notification forms part of a change set and tells a consumer an item needs to be re-assessed or re-evaluated. This may effect things like filtering or sorting of an item. I feel a concrete example best illustrates this concept. Suppose we have the following function:
Func<Trade,bool> isNearToMarketPrice = trade => return Math.Abs(trade.PercentFromMarket) <= 1 %
Where PercentFromMarket is a calculation which is recalculated with each and every market data tick. Using standard linq a list of near to market trades can be retrieved as follows.
var nearToMaketTrades = myListOfTrades.Where(isNearToMarketPrice);
The manifest problem with this query is it pull based only and has no means of observing when the market price has been recalculated, or alternatively how can re-evaluation be forced. Based on practical experience of dealing with this kind problem I introduced multiple means of injecting the evaluate command into a dynamic data stream. Here I will use a filter controller illustrated in a previous blog and we will apply it to the trade service here.
In summary any of the controllers in dynamic data are used to inject commands and meta data into a stream. For this example we need a dynamic filter which is applied to a stream of trades.
//create a filter controller and set it's filter
var filter = new FilterController<Trade>();
filter.Change(trade => Math.Abs(trade.PercentFromMarket) <= percentFromMarket());
//create a stream of live trades where the trade matches the above filter
var filteredByPercent = myTradeService.Trades
.Connect(trade=>trade.Status==TradeStatus.Live)
.Synchronize(locker)
.Filter(filter)
The filter controller has an overload to force re-evaluation.
filter.Revalue() // to reevaluate all // or filter.Reevaluate(Func<T,bool> itemSelector) //to re-evaluate selected items
The only missing element is when do we invoke re-evalulation. We have 2 choices. Either the service which calculates market prices provides a notification of trades which have been re-calculated or we poll on a period. Option 1 would be suitable for algo trading where everything must happen preferably with zero latency but for simplicity in the example we will poll as follows.
//re-evaluate filter periodically
var reevaluator = Observable.Interval(TimeSpan.FromMilliseconds(250))
.Subscribe(_ => filter.Reevaluate());
And that is that. We have a live stream of trades, where closed trades are automatically filtered out from the source and the filter controller constantly re-applies to ensure only trades near to the market are included in the result. And as ever what is beginning to become my catch phrase – that is easy!
After wrapping the function into a cold observable, here’s the final code.
public IObservable<IChangeSet<Trade, long>> Query(Func<uint> percentFromMarket)
{
if (percentFromMarket == null) throw new ArgumentNullException("percentFromMarket");
return Observable.Create<IChangeSet<Trade, long>>
(observer =>
{
var locker = new object();
Func<Trade,bool> nearToMaketTrades = trade => Math.Abs(trade.PercentFromMarket) <= percentFromMarket();
var filter = new FilterController<Trade>(nearToMaketTrades);
//re-evaluate filter periodically
var reevaluator = Observable.Interval(TimeSpan.FromMilliseconds(250))
.Synchronize(locker)
.Subscribe(_ => filter.Reevaluate());
//filter on live trades matching % specified
var subscriber = _tradeService.Trades
.Connect(trade=>trade.Status==TradeStatus.Live)
.Synchronize(locker)
.Filter(filter)
.SubscribeSafe(observer);
return new CompositeDisposable(subscriber, filter, reevaluator);
});
}
This observable will form part of a future post where I want to build the foundation of an auto trading system.
But for now, in a few lines of code (see NearToMarketViewer.cs) we can put the data onto the screen.
This is the third part of the trading example series. If you are not already familiar with the previous parts they are, part 1 is here and part 2 here. Also a fully working demo project is here. Since publishing part 2 I have embellished the sample code to make it look more production like so if you previously downloaded it I suggest you download it again.
So far we have created a trade service and a job which updates the market prices. Now I am going to demonstrate how dynamic data can help to present the data onto a WPF screen.
There are few ingredients we need to throw into the mix:
The filter controller which is used to reapply filters within a dynamic stream is constructed this
private readonly FilterController<Trade> _filter = new FilterController<Trade>();
I recommend using the dynamic data version of observable collection
private readonly IObservableCollection<TradeProxy> _data = new ObservableCollectionExtended<TradeProxy>();
which is optimised for binding dynamic streams. If however you choose to use the standard observable collection or the one provided by ReactiveUI, you can but will have to write you own operator to update the bindings. I will discuss how to in a future post.
Finally we need a simple proxy of the trade object.
public class TradeProxy:AbstractNotifyPropertyChanged, IDisposable, IEquatable<TradeProxy>;
{
private readonly Trade _trade;
private readonly IDisposable _cleanUp;
public TradeProxy(Trade trade)
{
_trade = trade;
//market price changed is a observable on the trade object
_cleanUp = trade.MarketPriceChanged
.Subscribe(_ =>; OnPropertyChanged("MarketPrice"));
}
public decimal MarketPrice
{
get { return _trade.MarketPrice; }
}
// additional members below (not show)
With these elements in place we can now easily get data from the trade service, filter it, convert it to a proxy, bind to an observable collection and dispose the proxy when it is no longer required.
var loader = tradeService.Trades
.Connect(trade => trade.Status == TradeStatus.Live) //prefilter live trades only
.Filter(_filter) // apply user filter
.Transform(trade => new TradeProxy(trade))
.Sort(SortExpressionComparer<TradeProxy>.Descending(t => t.Timestamp),SortOptimisations.ComparesImmutableValuesOnly)
.ObserveOnDispatcher()
.Bind(_data) // update observable collection bindings
.DisposeMany() //since TradeProxy is disposable dispose when no longer required
.Subscribe();
The only missing code is to apply a user entered filter. It looks something like this:
var filterApplier =//..watch for changes to the search text bindings
.Sample(TimeSpan.FromMilliseconds(250))
.Subscribe(_ => {
Func<Trade,bool> predicate= //build search predicate;
_filter.Change(predicate);
}
);
I will not explain the xaml required as this is beyond dynamic data. But a few lines of xaml bound to the result of observable collection can give this.
Was that easy? Take a look at the source code LiveTradesViewer.cs. 80 lines of code including white space.
All I say now is don’t tell your boss that you can do all this in a few lines of code otherwise you may have a pay cut!
Dynamic Data 