Trading Example Part 3. Integrate with UI

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:

  1. A filter controller as I want the user to be able to search for trades by entering some text.
  2. A proxy to the Trade object as the market price changes and WPF requires an INotifyPropertyChanged invocation.
  3. An observable collection so the screen can be updated with changed items,

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.
Live trades 3

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!

Advertisements

Trading Example Part 2. Manage market data

Part 1 to this series is here. Demo code is here

In this part we will monitor trades and update them with the latest market data. A live trade will typically have market data which is constantly streaming. So the problem to solve is to create the functionality to monitor live trades and keep the market price up to date on each trade.

For this we require 2 streams of data.

  1. Live trade stream
  2. Live market data stream

and we need to join them together and keep them in sync.

The live trade stream is easy using dynamic data.

var stream = tradeService.Trades.Connect()
                .Filter(trade => trade.Status == TradeStatus.Live)

And we’ll take it as a given that we can get the market price via a reactive stream.  The method signature would look something like this.

IObservable<decimal> ObservePrice(string currencyPair)

Each trade has a currency pair so we need to subscribe to streams for each currency pair. There are many ways to crack this particular egg but for the demo, I will first of all group the trades by currency pair. Again dynamic data makes it easy. Just group by currency pair.

var stream = tradeService.Trades.Connect()
                .Filter(trade => trade.Status == TradeStatus.Live)
                .Group(trade => trade.CurrencyPair)

This will give an observable cache of trades for each distinct currency pair. How cool is that? No code at all to maintain the grouping!  When trades are closed they will be removed from the grouping. And even better if a trade is updated to use a different currency pair the trade will be moved from the old group and added to the new one.  Now the trades are grouped by currency pair, we need to subscribe to the price stream and update each trade with the latest price as it changes.  For this I will introduce a new operator ‘SubscribeMany’ which subscribes to each item in a stream.

var stream = = tradeService.Trades.Connect()
  .Filter(trade => trade.Status == TradeStatus.Live)
  .Group(trade => trade.CurrencyPair)
  .SubscribeMany(grouping =>
  {
   //do something cool and return a disposable beacuse it will be
   // disposed when removed from the stream
  })

At a glance this seems nothing special but what it enables is to apply an action when the item is added or updated in a stream and to unsubscribe when an item is removed from a stream. In this case the operation applies to each distinct currency pair in the live trades query. What does this enable? Perhaps best illustrated by the following concrete example.

//...
    .SubscribeMany(grouping =>
   {
     var locker = new object();
     decimal latestPrice = 0;

    //1. Subscribe to price and update trades with the latest price
    var priceHasChanged = MarketPriceService.ObservePrice(grouping.Key)
    .Synchronize(locker)
    .Subscribe(price =>
   {
    latestPrice = price;
    UpdateTradesWithPrice(grouping.Cache.Items, latestPrice);
   });

   //2. Connect to data changes and update with the latest price. Exclude items which are removed from the stream because they are no longer live
   var dataHasChanged = grouping.Cache.Connect()
   .WhereReasonsAre(ChangeReason.Add, ChangeReason.Update)
   .Synchronize(locker)
   .Subscribe(changes => UpdateTradesWithPrice(changes.Select(change => change.Current), latestPrice));

   return new CompositeDisposable(priceHasChanged, dataHasChanged);
 })
.Subscribe()

where UpdateTradesWithPrice is a method as follows

private void UpdateTradesWithPrice(IEnumerable<Trade> trades, decimal price)
{
  foreach (var trade in trades)
  {
     trade.Price = price;
  }
}

Now we are function where live trades will be updated with the latest market data – always. And what’s more the example is totally thread safe. And even better there is very little plumbing code required.

Believe me this is a complex as dynamic data can be.  So if you follow this you can truly run with dynamic data. Also I promise that all subsequent posts will be easy.

Trading Example Part 1. Create a trade service

One of the things I love about dynamic data is no matter what the nature of the data which you are dealing with the code will look the same. I also love the power of data transformations and just how easy it is to tame sequences of data. However a lot of the capability and power is not obvious at a glance so this is the start of a series of posts where step by step I will illustrate usage of the library. I have decided to go in at the deep end and demonstrate a trading system as these systems generally have rapidly moving data and loads of complexity. What I hope to achieve is to show that by using dynamic data your code can be simplified greatly.

Source code can be found here

We will start off with a simple trade object

    public class Trade
    {
        public long Id { get; private set; }
        public string CurrencyPair { get; private set; }
        public string Customer { get; private set; }
        public decimal Price { get; set; }
        public TradeStatus Status { get; private set; }

        public Trade(long id, string customer, string currencyPair, TradeStatus status)
        {
            Id = id;
            Customer = customer;
            CurrencyPair = currencyPair;
            Status = status;
        }
    }

I want a service which exposes a load of trades.

    public interface ITradeService
    {
        IObservableCache<Trade, long> Trades { get; }
    }

Implementing this is a trivial matter. For clarity not all code is shown.

    public class TradeService : ITradeService
    {
        private readonly IObservableCache<Trade, long> _tradesCache;

        public TradeService()
        {
            //create a cache specifying that the unique key is id
            var tradesSource = new SourceCache<Trade, long>(trade => trade.Id);

            //call AsObservableCache() to hide the update methods as we are exposing the cache
            _tradesCache = tradesSource.AsObservableCache();

           //code to emulate an external trade provider
            var tradeLoader = GenerateTradesAndMaintainCache();

            //.....
        }

        public IObservableCache<Trade, long> Trades
        {
            get { return _tradesCache; }
        }
    }

We now have a trades service which I will be using in subsequent posts to demonstrate the functionality of dynamic data.

Part 2

Binding Example

This is the first in a series of real world dynamic data examples.

Lets’s assume I have a data  which is exposed via a service.

 public interface IPersonService
 {
      IObservableCache<Person, PersonKey> Cache { get; }
 }

and I want to bind the data to a screen. In dynamic data it is done like this.

 //this is a simple extension of observable collection which comes as part of the package
 var collection = new ObservableCollectionExtended<Person>();
 
 //Dynamic data is an extension of reactive so subscribe at the end of the chain
 var loader = personService.Cache.Connect()
 .Sort(SortExpressionComparer<Person>.Ascending(p=>p.FirstName).ThenByAscending(p=>p.LastName))
 .ObserveOnDispatcher()
 .Bind(collection)
 .Subscribe(); 

The binding will always reflect the data in the cache (and in the stated sort order).  That was easy so let’s extend this so we can apply a filter. This is also easy.

var loader = personService.Cache.Connect().Filter(p=>p.Age>50) 

or use a filter controller which allows the filter to be changed any time.

 //create a filter controller 
 var filterController = new FilterController<Person>();
 
 //filter can be changed any time by applying the following method
 filterController.Change(p=>p.Age>50);

 var loader = personService.Cache.Connect().Filter(filterController)

And that is that.  We can filter, sort and bind changeable data with very little code.

On GitHub

For years I have been meaning to get this project on GitHub and finally it has arrived. it can be found https://github.com/RolandPheasant. My original repository was in mercurial but however hard I tried I could not export my repo to git.  In the end I gave up and started the git repo from scratch.

I have included the source code as well as a primitive demo project. The demo project is a bit of a mess for which I do apologise. I fully intend to strip it right back to make the examples easier to find and clearer.  As a starting point look for code which ends in ViewModel to find the code behind each screen.

Going forward I hope to make the demo a great example for the usage and to demonstrate the power of Dynamic Data.

What is Dynamic Data?

Dynamic Data is a portable class library which brings the power of Reactive Extensions (Rx) to collections.

Mutable collections frequently experience additions, updates, and removals (among other changes). Dynamic Data provides two collection implementations, an observable list and an observable cache that expose changes to the collection via an observable change set. The resulting observable change sets can be manipulated and transformed using Dynamic Data’s robust and powerful array of change set operators. These operators receive change notifications, apply some logic, and subsequently provide their own change notifications. Because of this, operators are fully composable and can be chained together to perform powerful and very complicated operations while maintaining simple, fluent code.

Using Dynamic Data’s collections and change set operators make in-memory data management extremely easy and can reduce the size and complexity of your code base by abstracting complicated and often repetitive collection based operations.

An example please

Given a Trade object create an observable list like this

var myTrades = new SourceList<Trade>();

or a cache which requires that a unique key is specified

var myTrades = new SourceCache<Trade,long>(trade => trade.Id);

Either of these collections are made observable by calling the Connect() method which produces an observable change set.

var myObservableTrades = myTrades.Connect();

This example first filters a stream of trades to select only live trades, then creates a proxy for each live trade, and finally orders the results by most recent first. The resulting trade proxies are bound on the dispatcher thread to the specified observable collection.

ReadOnlyObservableCollection<TradeProxy> data;

var loader = myObservableTrades
    .Filter(trade => trade.Status == TradeStatus.Live) //filter on live trades only
    .Transform(trade => new TradeProxy(trade))         //create a proxy
    .Sort(SortExpressionComparer<TradeProxy>.Descending(t => t.Timestamp)) 
    .ObserveOnDispatcher()          //ensure operation is on the UI thread
    .Bind(out data)         //Populate the observable collection
    .DisposeMany()          //Dispose TradeProxy when no longer required
    .Subscribe();  

Since the TradeProxy object is disposable, the DisposeMany operator ensures that the proxy objects are disposed when they are no longer part of this observable stream.

This is the tip of the iceberg. The observable cache over 60 operators and the observable list has around 35 operators and counting.

Want to find out more?

Read the following for a quick overview

1. Observable Cache at a glance
2. Observable List at a glance
3. Getting Started
4. Binding example

Download the demo and source code

1. Demo WPF trading system demo on GitHub
2. Demo WPF reactive tree demo on GitHub
3. Dynamic Data source code on GitHub

Some funky UI specific stuff

1. Dynamically filter, sort and page
2. Reactive Tree
3. Advanced search hints with logical collection operators
4. Aggregation of dynamic data

Work through the ‘Creating a trading system’ series

1. Expose the data as a service
2. Manage market data
3. Integration with the UI
4. Filter on calculated values
5. Aggregation of dynamic data

Example integration with ReactiveUI

1. Integration with ReactiveUI for a simple example.
2. Log entry viewer for an advanced example.

Install from Nuget

1. Dynamic Data
or 2. Dynamic Data Adaptors for Reactive UI if you are already using ReactiveUI