Trading Demo Part 5 – Aggregate Dynamic Data

This is part 5 of building a trading system series. The idea has been to take one source of rapidly changing trading data and demonstrate the range, power and simplicity of applying dynamic data operators to it. Throughout the series I have made the assumption that the reader is already familiar and comfortable with the absolutely brilliant reactive extensions framework. For those who are not I strongly recommend to check them out. There is a steep learning curve but I assure anyone that once you start thinking reactive you will never turn back. You will probably wonder how you ever programmed without it.

If you are new to dynamic data I suggest you first read this Getting Started and if you have not seen the previous parts they are

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

Also if you want to download the code which goes with this example, go to Trading Demo on GitHub and download and run in visual studio. Open the menu option labelled Trading Positions and you will see the following screen.

Aggregated Positions

This screen connects to the trade service data, groups by currency pair and aggregates the totals for each grouping of currency pairs. With dynamic data doing this becomes so simple with very little code involved as I am about to show.

The starting point is to group the trade data by currency pair. This is achieved using dynamic data’s Group operator.

//where tradeService is the trade service created in Part 1.
var mygroup = tradeService.Live.Connect()
                          .Group(trade => trade.CurrencyPair)

This does not look like much but the group operator is very powerful indeed. As with all things reactive, changes in the trade service are automatically propagated to the appropriate grouping. Dynamic Data extends the reactive framework by accounting for adds, updates and removes. So when trades are created, closed or amended the groups will automatically reflect these changes.

The grouping object creates a cache of each group and it’s signature is like this.

public interface IGroup<TObject, TKey, out TGroupKey>
{
    TGroupKey Key {get;}
    IObservableCache<TObject, TKey> Cache { get; }
}

Three generic types, a bit ugly I admit but we have a dynamic collection of caches, one for each item matching the group selector in the underlying data source. When grouping I almost always apply a transform function to do something with the group. In this example we want to aggregate totals for each group, so first I have created an object which accepts the grouping in it’s constructor and apply some dynamic aggregations on the cache. For this we need to use dynamic data’s Transform operator. This is akin to the standard reactive Select operator but as with several other operators in dynamic data I have to use an alternative semantic dues to ambiguous reference problems for extensions.

var transformedGroup = tradeService.Live.Connect()
                .Group(trade => trade.CurrencyPair)
                .Transform(group => new CurrencyPairPosition(group))

This means for each group we have a CurrencyPairPosition object which will be used to dynamically calculate the sum total of the buys and sells so that the overall position can be calculated. For the aggregation I have used another dynamic data operator QueryWhenChanged. The following exert illustrates it’s usage

//constructor only for illustration
public CurrencyPairPosition(IGroup<Trade, long, string> tradesByCurrencyPair)
{
    var aggregations = tradesByCurrencyPair.Cache.Connect()
                        .QueryWhenChanged(query => //query and return a result
}

For each change in the underlying cache the operator is invoked. The parameter exposes a querying API so the underling cache can be queried in a functional manner. The consumer must return a result which becomes an observable of whatever is returned. In this example we have used some standard linq to objects to calculate the buy and sell totals and return a new object which contains these values.

If we expand out the above code we easily create an observable of trade positions for each currency pair,

  var aggregations = tradesByCurrencyPair .Cache.Connect()
                       .QueryWhenChanged(query =>
                        {
                            var buy = query.Items
                                        .Where(trade => trade.BuyOrSell == BuyOrSell.Buy)
                                        .Sum(trade=>trade.Amount);
                            var sell = query.Items
                                        .Where(trade => trade.BuyOrSell == BuyOrSell.Sell)
                                        .Sum(trade => trade.Amount);
                            var count = query.Count;
                            return new TradesPosition(buy,sell,count);
                        });
                //subscribe to the result and set a property to the latest

We are at the point where we have a CurrencyPairPosition object which always reflects the latest position so the last things is to expand the consuming code so we can bind to the collection of these.

 //not all code show and _data is a derived Observable collection
  var subscriber = tradeService.Live.Connect()
                .Group(trade => trade.CurrencyPair)
                .Transform(group => new CurrencyPairPosition(group))
                .Sort(SortExpressionComparer<CurrencyPairPosition>.Ascending(t => t.CurrencyPair))
                .ObserveOn(schedulerProvider.MainThread)
                .Bind(_data)
                .DisposeMany()
                .Subscribe();

This will maintain the sorted order and ensure that CurrencyPairPosition objects are disposed when no longer required. Now a little xaml bound to the data property can produce the screen at the top of the page.

In summary, Dynamic Data makes the reactive management of collections of data very very easy indeed. If you look at the code list below, you will see just how little is involved. There are about 170 lines of code over three c# classes which also includes namespace and using declarations. And what’s even better is despite the underlying data rapidly moving, nowhere have I had to concern myself with thread safety and all the operators are completely thread-safe so long as all code functionally remains inside the operators.

The full code for all the objects is

1. PositionsViewer.cs (The view model)
2. CurrencyPairPosition.cs
3. TradesPosition.cs (the object which holds the result)
4. PositionsView.xaml (the screen)

Advertisements

Integration with ReactiveUI

I have released DynamicData.ReactiveUI which is a very simple adaptor layer to assist with binding dynamic data observables with reactiveui’s ReactiveList object.

Install from Nuget

Nuget

If you are not familiar with dynamic data and what it enables I suggest:

  1. Go through previous posts in the blog
  2. Download the example wpf app from github
  3. .

To cut to the chase, here’s an example. The code takes existing in-memory trade objects, transforms them into a view model proxy and updates the target reactive list object with the resulting change sets. It also pre-filters the data with live trades, applies a user entered filter, orders the resulting data and disposes the proxy when no longer required. Phew – all that in effectively one line of code.

I did the same example without using Reactive UI code see Trading Example Part 3. Integrate with UI where I have explained the code in greater detail

    public class RxUiViewer : ReactiveObject, IDisposable
    {
       //this is the target list which we will populate from the dynamic data stream
        private readonly ReactiveList<TradeProxy> _data = new ReactiveList<TradeProxy>();
        //the filter controller is used to inject filtering into a observable
        private readonly FilterController<Trade> _filter = new FilterController<Trade>();
        private readonly IDisposable _cleanUp;
        private string _searchText;

        public RxUiViewer(ITradeService tradeService)
        {
            //Change the filter when the user entered search text changes
            var filterApplier = this.WhenAnyValue(x => x.SearchText)
                .Throttle(TimeSpan.FromMilliseconds(250))
                .Subscribe(_ => ApplyFilter());
           
            ApplyFilter();

            var loader = tradeService.Trades
                .Connect(trade => trade.Status == TradeStatus.Live) //prefilter live trades only
                .Filter(_filter)    // apply user filter
                   //if targetting Net4 or Net45 platform can use parallelisation for transforms 'cause it's quicker
                .Transform(trade => new TradeProxy(trade), new ParallelisationOptions(ParallelType.Ordered, 5))
                .Sort(SortExpressionComparer<TradeProxy>.Descending(t => t.Timestamp), SortOptimisations.ComparesImmutableValuesOnly)
                .ObserveOn(RxApp.MainThreadScheduler)
                .Bind(_data)        //bind the results to the ReactiveList 
                .DisposeMany()      //since TradeProxy is disposable dispose when no longer required
                .Subscribe();

            _cleanUp = new CompositeDisposable(loader, _filter, filterApplier);
        }

        private void ApplyFilter()
        {
            if (string.IsNullOrEmpty(SearchText))
            {
                _filter.ChangeToIncludeAll();
            }
            else
            {
                _filter.Change(t => t.CurrencyPair.Contains(SearchText, StringComparison.OrdinalIgnoreCase) ||
                                    t.Customer.Contains(SearchText, StringComparison.OrdinalIgnoreCase));
            }
        }

        public string SearchText
        {
            get { return _searchText; }
            set { this.RaiseAndSetIfChanged(ref _searchText, value); }
        }

        public IReadOnlyReactiveList<TradeProxy> Data
        {
            get { return _data; }
        }

        public void Dispose()
        {
            _cleanUp.Dispose();
        }
    }

coupled with a little xaml can produce this

ReactiveUIImage

Very little code, but a very powerful example,

Gone Portable

Dynamic data is now a portable class library available on most platforms. See below.

Portable

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.

Trading Example Part 4. Filter on calculated values

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.

Near to market

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!

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