Generating Market Data Snapshot from a Tick Stream using Rx

One of my colleges (http://weblogs.asp.net/sweinstein/) brought up an interesting problem: there is a common theme in finance applications to subscribe to a market data source to receive real time updates on the instruments that this source covers. A late subscriber (the one that has not been subscribed since the start of the trading day, for example) usually expects to receive a snapshot per instrument before real time updates. So what is the best way to support snapshot guarantee, per instrument, from a “hot” underlying source using Rx? There are multiple ways to do it, of course, and here is my version of it:

public static IObservable<T> WithCachedSnapshot<T, TKey>(
    this IObservable<T> source,
    Func<T, TKey> keySelector)
{
    var cache = new Dictionary<TKey, T>();
    return Observable.Defer(() => {
            lock (cache)
                return cache.Values.ToList().ToObservable();
        })
        .Concat(source.Do(s => {
            lock (cache) cache[keySelector(s)] = s;
        }));
}

If you can’t take my word for it, here is the proof that it works:

[TestMethod()]
public void CacheByKeyTestHelper()
{
    // Arrange
    TestScheduler scheduler = new TestScheduler();
    var source = scheduler.CreateHotObservable(
        OnNext(100, new { Ticker = "MSFT", Price = 30.5 }),
        OnNext(101, new { Ticker = "IBM", Price = 130.2 }),
        OnNext(102, new { Ticker = "MSFT", Price = 31.5 }),
        OnNext(103, new { Ticker = "IBM", Price = 130.1 }),
        OnNext(104, new { Ticker = "AAPL", Price = 150.5 }));

    // Act
    var target = RxExtensions.WithCachedSnapshot(
        source,
        quote => quote.Ticker);
    var results = scheduler.Run(() => target, 0, 0, 105);
    var cachedResults = scheduler.Run(() => target, 0, 0, 200);

    // Assert
    results.AssertEqual(
        OnNext(100, new { Ticker = "MSFT", Price = 30.5 }),
        OnNext(101, new { Ticker = "IBM", Price = 130.2 }),
        OnNext(102, new { Ticker = "MSFT", Price = 31.5 }),
        OnNext(103, new { Ticker = "IBM", Price = 130.1 }),
        OnNext(104, new { Ticker = "AAPL", Price = 150.5 }));

    cachedResults.AssertEqual(
        OnNext(105, new { Ticker = "MSFT", Price = 31.5 }),
        OnNext(105, new { Ticker = "IBM", Price = 130.1 }),
        OnNext(105, new { Ticker = "AAPL", Price = 150.5 }));
}

private Recorded<Notification<T>> OnNext<T>(
    long ticks, T value)
{
    return new Recorded<Notification<T>>(
        ticks,
        new Notification<T>.OnNext(value));
}

Detecting Running High/Low Prices Using Reactive Framework – Follow-up

This article shows some improvements to the original code presented in my previous post. Improvements are based on the feedback I received from Erik Meijer and his team and mainly focus on some built-in opportunities in Rx framework that I missed. Here is an extract from the original implementation of the running low feed:

// Daily low price feed
double min = double.MaxValue;
var feedLo = feed
    .Where(p => p < min)
    .Do(p => min = Math.Min(min, p))
    .Select(p => "New LO: " + p);

This implementation depends on the external (to the framework) state (current low value stored in “min” local variable), “Do” combinator to continuously update this state with every price tick, and a “Where” combinator to suppress the ticks that do not represent running low value. Turns out, there is a better, more elegant and framework friendly way to implement this. It relies on the built-in “Scan” and “HoldUntilChanged” combinators:

// Daily low price feed
var feedLo = feed
    .Scan(Math.Min)
    .HoldUntilChanged()
    .Select(p => "New LO: " + p);

The code above is functionally equivalent to the original one but is cleaner, more compact, and easier to read. This is how it works:

• “Scan” operator is used to accumulate the running min value. “Scan” is similar to “Aggregate” in the LINQ-to-Objects world but, instead of collapsing the sequence to a single accumulated value, it keeps outputting the current accumulated value with every tick. As a result “Math.Min” function is continuously applied with every new tick with accumulated min value and new price as arguments and its result is “ticked out”. This produces a stream where every successive number is either less than or equal to the previous one. Close to what we need except we want to suppress duplicates.
• “HoldUntilChanged” is another built-in combinator that wraps an underlying stream and only repeats a value from it if this value differs from the previous one. Perfect for removing successive duplicates (please, note, it will NOT remove ALL the duplicates, only successive ones, so it is not a proper substitute to Distinct, when such a combinator exists) . We use “HoldUntilChanged” instead of original “Where” clause.

Running VWAP Calculation Using Reactive Extensions for .NET

Reactive Framework Labs

This article is one from the “Processing Financial Data with Microsoft Reactive Framework (Reactive Extensions for .NET)” series of articles that demonstrate some basic applications of Rx framework to financial data. This article shows how to calculate a running VWAP (volume weighted average price over a defined window of time). Other articles in the series are:

• Detecting daily Hi/Lo prices
• Throttling high-frequency data feed
• Data feed failover

Visual Studio 2008 solution containing full source code for all examples can be downloaded here.

Calculating VWAP/Sliding Window Implementation

We will use this simple “Trade” type to represent a trade event. To simplify things we will only consider Price and Volume:

class Trade
{
    public double Price { get; set; }
    public int Volume { get; set; }
}

The following “TradeWithVWAP” type wraps underlying trade and attaches running VWAP value to it. It is used for calculation output:

class TradeWithVWAP
{
    public Trade Trade { get; set; }
    public double VWAP { get; set; }
}

Our main module starts with a simple market data generator that produces infinite stream of random prices ($30 to $33 range) at random volumes (1,000 to 100,000 shares) at a random frequency (0 to 3 seconds between trades). BuildVwapFeed functions takes generated trade stream and converts it to another stream with running VWAP attached to every trade. And finally we simply subscribe to VWAP-wrapped feed and output the results.

Data Feed Failover Using Reactive Extensions for .NET

Reactive Framework Labs

This article is one from the “Processing Financial Data with Microsoft Reactive Framework (Reactive Extensions for .NET)” series of articles that demonstrate some basic applications of Rx framework to financial data streams. This article demonstrates automatic and transparent failover from primary data feed to a backup one implemented in terms of Microsoft Reactive Framework. Other articles in the series are:

• Detecting daily Hi/Lo prices
• Throttling high-frequency data feed
• Running VWAP calculation

Visual Studio 2008 solution containing full source code for all examples in this article can be downloaded here.

Failover Implementation

The effect of automatic failover is implemented using the “Catch” operator built into the Rx framework.  The “Catch” operator simply repeats the first stream until it fails. When the first stream fails, exception is swallowed and the second stream is subscribed to and repeated. You can chain any number of streams this way. Here is a code sample that uses “Catch” to switch from some primary feed to a backup:

Throttling High-Frequency Data Feed Using Reactive Extensions for .NET

Reactive Framework Labs

This article is one from the “Processing Financial Data with Microsoft Reactive Framework (Reactive Extensions for .NET)” series of articles that demonstrate some basic applications of Rx framework to financial data. This article shows how to throttle your incoming feed to a desired frequency. Other articles in the series are:

• Detecting daily Hi/Lo prices
• Data feed failover
• Running VWAP calculation

Visual Studio 2008 solution containing full source code for all examples can be downloaded here.

Throttling Implementation

The key operator here is “Sample”. Sampling at a desired rate effectively returns your a throttled stream. My intuition was that if underlying feed does not produce a value during the sampling interval, “Sample” will repeat the most recent available value when sampling is due. But it doesn’t and this makes it a perfect fit for throttling. Note that there is also an operator called “Throttle” but its name is misleading and its behavior most likely will not match your definition of throttling. The way implemented it will not emit any values at all if underlying feed produces faster than throttling interval, so it looks more like “Choke” to me.

Detecting Running High/Low Prices Using Reactive Extensions for .NET

Reactive Framework Labs

This article is one from the “Processing Financial Data with Microsoft Reactive Framework (Reactive Extensions for .NET)” series of articles that demonstrate some basic applications of Rx framework to financial data. This article focuses on detecting running high/low values (like daily max and min) from a market data feed. Other articles in the series are:

• Throttling high-frequency data feed
• Data feed failover
• Running VWAP calculation

Visual Studio 2008 solution containing full source code for all examples can be downloaded here.

Detecting Running High/Low Prices

The key operators in this implementation are “Do” and “Where”, both built into the Reactive Framework. The first one is used to produce side effect of recording the running High or Low value, and the second is used to filter incoming values that do not produce newest High or Low:

Processing Financial Data with Microsoft Reactive Framework (Reactive Extensions for .NET)

Reactive Framework Labs

I have recently discovered Microsoft Reactive Framework (thanks to spontaneous introduction at New York .Net Meetup Group) and have been playing with it a lot trying to apply it to many financial problems I’ve coded over the last years. I am truly amazed with the technology and the power it unleashes for processing and analyzing streaming financial data (like quotes, market data feeds, trade executions, etc.). Seamless integration with LINQ combined with very powerful support for composition makes self coded CEP in .Net a low hanging fruit. In the following series of articles I am going to demonstrate sample Rx-based implementations for some commonly occurring tasks in finance, including:

• Detecting daily Hi/Lo prices
• Throttling high-frequency data feed
• Data feed failover
• Running VWAP calculation

Visual Studio 2008 solution containing full source code for all examples can be downloaded here.

Advance Your Composite Application MVVM to the Next Level – Use Attributes to Automatically Populate Your View Model with Service Calls

Abstract

MVVM is a proven design pattern in WFP/Silverlight world. Its use helps achieve separation of concerns, frees UI from any logic to enhance testing, and generally makes the code easier to understand and maintain. In many LOB multi-tier applications MVVM is extended to include a presenter class. Presenter usually controls the view model, receives necessary services from dependency injection container and has some logic to populate the view model with business entities and/or reference data and send any changes back. The resulting code is usually pretty clean, easy to maintain and test. This article argues though that most of the code in such presenters to populate the view models is also very mechanical, routine, boring and can be avoided all together by introducing small attribute-based view model framework to declaratively populate view models from service calls and avoid extra coding, and, thus, avoid writing extra mechanical, routine, and boring tests.

If you prefer the “code first” approach VS 2008 solution can be downloaded here: MVVMAttributes.zip

LINQ to SQL Tricks: Building Efficient Queries that Include Reference Data or Child Entities

Problem Summary

Some times you need to read entities from a table along with some properties that are reference data or child entities. This relationship is usually represented by foreign keys in your database. LINQ to SQL makes this job extremely easy but, by default, also very inefficient. There are multiple ways this inefficiency can be addressed. Some of these alternatives are standard facilities provided by LINQ to SQL and some are custom coded patterns and tricks. This article walks your through these solutions, explaining their pros and cons and helping you decide which one is right for you.

Default Behavior

Consider the following simple subset from the AdventureWorksLT database:

Lets say your job is to select and display all the customer addresses in the database (we will limit results to first 3 to simplify output), with the following fields: First Name, Last Name, City, State, and Modified Date. You have your Data Access Layer (DAL) to get the data and your driver code to print the results. Something like this:

Running multiple WPF applications in the same process using AppDomains

Why?

There are various reasons you may want to run several full-blown independent WPF applications side-by-side in the same process. In my particular case this was a client project where we were building a composite WPF trading desk. One of the requirements was to be able to run different versions of the same component (montage, P&L, blotter, etc.) side-by-side but within seemingly integrated environment. The best solution turned out to be running different versions in completely independent environment (application) and only communicate across domains using a thin protocol to provide “integrated” look and feel. Running components in separate domains let us avoid all the problems that come with signing, versioning, evolving dependencies, and many more.

This post is a step by step guide to demonstrate the techniques involved.