Allowing end-users to perform custom computations on real-time data

Absract

This post is an introduction to a new project CalcRx that allows you to write applications to give users ability to run custom expressions on Observables. A couple of example scenarios are also discussed. To learn technical aspects about CalcRx in detail, visit the project page. This post does not assume you know about .NET and Reactive Extensions already. But in order to understand the technical aspects of CalcRx, those two are required.

You may have learned the about importance of usability, abstraction, and inference of real-time data already. In this post, I am going to jump straight in to one of the ways that we can accomplish this in .NET Applications.

Scenario 1

First consider this scenario, you have an application that displays the real-time chart of radiation levels monitored in Earth's atmosphere. The satellite is pounding your application with tons of data. The user of your application wants to see only so much, suppose your user wants to see the data where radiation levels are higher than 2 microsieverts, but your application is getting the 100 readings per seconds and your chart is animating too fast to be effectively read by the users. Now the users are telling you to show only the levels higher than 2 microsieverts, or one user wants to see only those spots where the radiation level is 2+ for at least 1 minute other wise he or she wants to ignore the reading.

Scenario 2

Another scenario is, you are the developer of the application that is showing the live chart of stock prices. Every time there is a tick, the chart has a point added. Suppose your user is monitoring Apple's stock which is changing every 2 seconds, your user does not like to be overwhelmed with a lot of unrelated information, he asks you to change your application to show you the Moving Average of last 5 ticks only. While your code change might work for the user watching Apple's stock, it will not do much good for the person who is watching a stock that does not change frequently.

CalcRx can help you in both of the scenarios described above. As long as your application is a .NET Application and the radiation levels or stock prices you are getting are received in IObservable<T>, CalcRx can make the job so much easier.

IObservable<T> is an interface for an implementation of a Publish-Subscribe model. You can convert anything coming into your .NET Application into IObservable<T>. For instance, if you have an event that is fired up every time you receive a new reading of radiation level, you could easily create an observable out of it. For more information, Checkout www.introtorx.com.

Once you are receiving your data through an Observable. You don't have to change your code anymore. You just add a Textbox to your application and ask your users to write their own custom formulas into that. CalcRx uses the expression language which is very close to what Excel uses for formulas.

For a detailed into on CalcRx, visit the Project Page

F# example: Getting live stock ticks through Google Finance API

NOTE: This article demonstrates how F# in conjunction with Reactive Extensions can be used to connect to live data streaming. The full project with example client can be found at This Github link.

Disclaimer

I do not  take any responsibility of the code and any damages caused by it. Code which is provided AS IS is only for demonstration purposes and in no way encourages anyone to use Google Finance API for business. 

Introduction

It has been a week learning F# and I have finally converted my C# code that uses Google Finance API to F#

Here is what my code does. In simple words, it gives you back the IObservable<Tick> for one security (or symbol)

Here is the screenshot of how it looks like

Here is how you use it as a library in C#

 DataMachine machine = new DataMachine();  
 IObservable<Tick> appleTicker = machine.GetStockQuote("AAPL"); // Get Apple's stock prices  
 appleTicker.Subscribe(tick => {  
        // Everytime price changes, this piece of code will be called and an   
       // An object of Tick will be supplied as tick  
       // You can either add them to your Graph series to plot live chart  
       // Or save them to CSV  
       });  

Google Finance API

The simple finance API by Google is just a URL that takes symbol in 'q' GET parameter (or query string parameter). The following will give you the current price data for Yahoo in Json format.

  http://www.google.com/finance/info?q=YHOO

How it works ?

It creates a thread using TaskFactory which queries Google Finance every 1 second, fetches the price and if there is a change in price produces the output into the Observable. Google Finance API gives back Json object. I am using Json.Net library by James-Newton King to parse that.

Here is the type definition of Tick

 type Tick() =   
     [<JsonProperty(PropertyName = "t")>]   
     member val Symbol = Unchecked.defaultof<string> with get, set  
     [<JsonProperty(PropertyName = "l")>]  
     member val Price = Unchecked.defaultof<float> with get, set  
     [<JsonProperty(PropertyName = "lt_dts")>]  
     member val LastUpdated = Unchecked.defaultof<DateTime> with 

Here is the code for DataMachine type

 type DataMachine() =   
   member this.GetLiveGoogleApiTicks(symbol : string) =  
     //google_api queries Google APi every second and produces the string output  
     let google_api (symbol : string) =  
       // When operation is cancelled by the caller, this variable will be modified  
       let cancelled = ref false  
       let operation(o : IObserver<string>) =  
         //Time to wait before fetching again through the API  
         let timeToSleep = TimeSpan.FromSeconds(1.0)  
         //Google Finance API , notice q is the query string parameter for symbol  
         let urlBuilder = new StringBuilder("""http://www.google.com/finance/info?q=""")  
         urlBuilder.Append symbol  
         let url = urlBuilder.ToString()  
         let client = new WebClient()  
         let cancelToken = Task.Factory.CancellationToken  
         client.BaseAddress = url  
         let task() =  
               try  
                 while not cancelToken.IsCancellationRequested && not !cancelled do  
                   let data = client.DownloadString url  
                   //Produce the item in observable  
                   o.OnNext data  
                   Thread.Sleep timeToSleep  
                   while client.IsBusy do  
                     Thread.Sleep 100  
               finally  
                   ignore  
         Task.Factory.StartNew task       
         let a = new Action(fun () -> (cancelled := true))  
         a |> Disposable.Create  
       Observable.Create(operation)  
     let strings = (google_api symbol)  
     // Eliminate unparseable characters for Json.Net library  
     let trimmed = strings.Select (fun (l:string) -> l.Substring(l.IndexOf("[") + 1, l.LastIndexOf("]") - 2 - l.IndexOf("[") + 1))  
     //Parse into object and return the IObservable<Tick>  
     trimmed.Select(fun (l:string) -> JsonConvert.DeserializeObject(l)).DistinctUntilChanged(fun (a:Tick) -> a.Price)
     

Remember to add Rx-Main and Json.Net from Nuget in your project and add appropriate 'open' declarations for namespaces.

Download the code

The fully functional project with sample client for this project can be found at This Github link.

Let me know your comments in the section below. If there is anything that I should have done differently, I would love to hear about that.

Happy coding!