Celso Axelrud: HIGH PERFORMANCE REAL-TIME STOCKS COMPUTATION ENGINE

HIGH PERFORMANCE REAL-TIME STOCKS COMPUTATION ENGINE - part 1

This blog explores a High Performance Real-Time Stocks Computation Engine for data collection and processing. It can be the basis of a Real-time Trading System.

High Performance is obtained by using Reactive and Non-blocking code.

The code has the following properties:

Developed for F# REPL script.
Uses .NET Reactive Extensions and FSharp.Control.Reactive for the PUSH pattern style coding.
Uses F# Asynchonous workflow for PULL pattern style coding.
Collects Stock data from "https://code.google.com/p/yahoo-finance-managed/wiki/YahooFinanceAPIs"

As an overall pattern, the PUSH pattern is the preferred style, eventhough, the example, presented here, starts by, periodicly, pull data from the web.

The main reason is related to the fact that the communication with other processes/computers is expected to use Publish/Subscribe, and this is a PUSH pattern.

Example steps

Define references to the libraries.

For Reactive Extensions.

1: 
2: 
3: 
4: 
5: 
6:

#r @"C:\Project(comp)\Dev_2015\Stocks_1\Appl1\lib\System.Reactive.Core.dll"
#r @"C:\Project(comp)\Dev_2015\Stocks_1\Appl1\lib\System.Reactive.Interfaces.dll"
#r @"C:\Project(comp)\Dev_2015\Stocks_1\Appl1\lib\System.Reactive.Linq.dll"
#r @"C:\Project(comp)\Dev_2015\Stocks_1\Appl1\lib\System.Reactive.PlatformServices.dll"
#r @"C:\Project(comp)\Dev_2015\Stocks_1\Appl1\lib\System.Reactive.Providers.dll"
#r @"C:\Project(comp)\Dev_2015\Stocks_1\Appl1\lib\Microsoft.Reactive.Testing.dll"

For FSharp.Control.Reactive.

1:	`#r @"C:\Project(comp)\Dev_2015\Stocks_1\Appl1\lib\FSharp.Control.Reactive.dll" //Need to be the last reference`

Open the libraries.

1: 
2: 
3: 
4:

open System
open System.Collections.Generic
open System.Threading
open System.Net

Open Reactive Extensions libraries.

1: 
2: 
3: 
4: 
5: 
6:

open System.Reactive
open System.Reactive.Linq
open System.Reactive.Disposables
open System.Reactive.Concurrency
open System.Reactive.Subjects
open Microsoft.Reactive.Testing

Open FSharp.Control.Reactive libraries.

1: 
2: 
3:

open FSharp.Control.Reactive
open FSharp.Control.Reactive.Builders
open FSharp.Control.Reactive.Observable

Define a Helper module.

It includes several observers to help display results.

 1: 
 2: 
 3: 
 4: 
 5: 
 6: 
 7: 
 8: 
 9: 
10: 
11: 
12: 
13: 
14: 
15: 
16: 
17: 
18:

module Helper=
    let name=String.Empty
    let PrintThread (location:string)=
            printfn "Loc:%s Thread:%d" location Thread.CurrentThread.ManagedThreadId
    let f1=fun (i:'T)->PrintThread "OnNext"
                       printfn "%s-->%A" name i
    let f2=fun (ex:Exception)->printfn "%s-->%A" name ex.Message
    let f3=fun ()-> PrintThread "OnCompleted"
                    printfn "%s completed" name
    let Dump1<'T>  (name:string) (source:IObservable<'T>) =
        Observable.subscribeWithCallbacks f1 f2 f3 source              
    let Dump2<'T>  (name:string) (source:IObservable<'T>) =
        source |> (Observable.subscribeOn Scheduler.Default) 
               |>  Observable.subscribeWithCallbacks f1 f2 f3              
    let Dump3<'T>  (name:string) (source:IObservable<'T>) =
        source |> (Observable.subscribeOn Scheduler.Default)   
               |> (Observable.observeOn Scheduler.Default) 
               |>  Observable.subscribeWithCallbacks f1 f2 f3

The following fields are being collected. They are combined to form the field f=nsl1opin http://download.finance.yahoo.com/d/quotes.csv?s=MSFT,GOOG&f=nsl1op&e=.csv URL:

name(n),symbol(s),latestValue(l1),open(o),closeValue(p)

Define the URL.

1:	`let url = "http://download.finance.yahoo.com/d/quotes.csv?s="`

Define a Stock type to hold Stocks information.

1:	`type Stocks={Name:string;Symbol:string;LatestValue:float;OpenValue:float;CloseValue:float}`

Define an Asynchonous Workflow to get Stock data.

 1: 
 2: 
 3: 
 4: 
 5: 
 6: 
 7: 
 8: 
 9: 
10: 
11: 
12: 
13: 
14: 
15: 
16: 
17: 
18: 
19: 
20:

let getStockData stock fields =
    async {
        //try 
            let wc = new WebClient()
            Helper.PrintThread "Async before Download"
            let! data = wc.AsyncDownloadString(Uri(url + stock + "&f="+ fields + "&e=.csv"))
            Helper.PrintThread "Async after Download"
            //let data = wc.DownloadString(url + stock + "&f="+ fields + "&e=.csv")
            //printfn "%s" data
            let dataLines = data.Split([| '\n' |], StringSplitOptions.RemoveEmptyEntries) 
            let s=seq { for line in dataLines do
                            let infos = line.Split(',')
                            yield {Name=infos.[0]; Symbol=infos.[1];
                                    LatestValue=float infos.[2]; OpenValue=float infos.[3]; 
                                    CloseValue=float infos.[4]}
                       }
                |> Array.ofSeq 
            return s //Some s
        //with _ -> return None
      }

You can test getStockInfo using:

1: 
2:

let r=(getStockData "GOOG,MSFT" "nsl1op") |>Async.RunSynchronously
r.[0],r.[1];;

Create Observables Sequences.

Create an "event" (observable) every 5 seconds (fast for testing pursose)
Take 3 elements
Get Stocks information asynchoronous
Split into individual Stocks

1: 
2: 
3: 
4: 
5:

let t1=Observable.interval (TimeSpan.FromSeconds(5.0))
let o1=t1 |> Observable.take(3) 
let o2= Observable.SelectMany(o1,fun _-> Observable.ofAsync (getStockData "GOOG,MSFT" "nsl1op"))
let oStock1=o2 |> Observable.map (fun (i:Stocks[])->i.[0])
let oStock2=o2 |> Observable.map (fun (i:Stocks[])->i.[1])

Observe with different Observers

Comments:

Case 1:

Main thread (1) is not blocked.
Observable Sequence is blocking Thread 12.
Obsever (OnNext) is called in the same thread as the Observable Sequence

Case 2:

Obsever (OnNext) is called in the same thread as the Observable Sequence

Case 3:

No blocking of Threads.

Case 1

(using Dump1 observer)

1: 
2: 
3:

Helper.PrintThread "REPL example1-1-begin"
let d1=Helper.Dump1 "Stock1" oStock1
Helper.PrintThread "REPL example1-1-end"

Use d1.Dispose() to stop it.

Case 2

(using Dump2 observer)

1: 
2: 
3:

Helper.PrintThread "REPL example1-2-begin"
let d2=Helper.Dump2 "Stock2" oStock2
Helper.PrintThread "REPL example1-2-end"

Use d2.Dispose() to stop it.

Case 3

(using Dump3 observer)

1: 
2: 
3:

Helper.PrintThread "REPL example1-3-begin"
let d3=Helper.Dump3 "Stock3" oStock2
Helper.PrintThread "REPL example1-3-end"

Use d3.Dispose() to stop it.

The output of the Case 1 is the following:

 1: 
 2: 
 3: 
 4: 
 5: 
 6: 
 7: 
 8: 
 9: 
10: 
11: 
12: 
13: 
14: 
15: 
16: 
17: 
18: 
19:

//Output:
(*
Loc:REPL example1-1-begin Thread:1
Loc:REPL example1-1-end Thread:1
    Loc:Async before Download Thread:12 //*
    Loc:Async after Download Thread:16
    Loc:OnNext Thread:16
        Stock1-->{Name = ""Google Inc."";Symbol = ""GOOG"";LatestValue = 523.4;OpenValue = 521.08;CloseValue = 521.84;}
    Loc:Async before Download Thread:12 //*
    Loc:Async after Download Thread:15
    Loc:OnNext Thread:15
        Stock1-->{Name = ""Google Inc."";Symbol = ""GOOG"";LatestValue = 523.4;OpenValue = 521.08;CloseValue = 521.84;}
    Loc:Async before Download Thread:12 //*
    Loc:Async after Download Thread:20
    Loc:OnNext Thread:20
        Stock1-->{Name = ""Google Inc."";Symbol = ""GOOG"";LatestValue = 523.4;OpenValue = 521.08;CloseValue = 521.84;}
    Loc:OnCompleted Thread:20
Stock1 completed
*)

The output of the Case 2 is the following:

 1: 
 2: 
 3: 
 4: 
 5: 
 6: 
 7: 
 8: 
 9: 
10: 
11: 
12: 
13: 
14: 
15: 
16: 
17: 
18: 
19:

//Output:
(*
Loc:REPL example1-2-begin Thread:1
Loc:REPL example1-2-end Thread:1
    Loc:Async before Download Thread:7
    Loc:Async after Download Thread:7 //*
    Loc:OnNext Thread:7 //*
        Stock2-->{Name = ""Microsoft Corporation"";Symbol = ""MSFT"";LatestValue = 44.4;OpenValue = 44.48;CloseValue = 44.44;}
    Loc:Async before Download Thread:20
    Loc:Async after Download Thread:8 //*
    Loc:OnNext Thread:8 //*
        Stock2-->{Name = ""Microsoft Corporation"";Symbol = ""MSFT"";LatestValue = 44.4;OpenValue = 44.48;CloseValue = 44.44;}
    Loc:Async before Download Thread:20
    Loc:Async after Download Thread:7 //*
    Loc:OnNext Thread:7 //*
        Stock2-->{Name = ""Microsoft Corporation"";Symbol = ""MSFT"";LatestValue = 44.4;OpenValue = 44.48;CloseValue = 44.44;}
    Loc:OnCompleted Thread:7 //*
Stock2 completed
*)

The output of the Case 3 is the following:

 1: 
 2: 
 3: 
 4: 
 5: 
 6: 
 7: 
 8: 
 9: 
10: 
11: 
12: 
13: 
14: 
15: 
16: 
17: 
18: 
19:

//Output:
(*
Loc:REPL example1-3-begin Thread:1
Loc:REPL example1-3-end Thread:1
    Loc:Async before Download Thread:6
    Loc:Async after Download Thread:6
    Loc:OnNext Thread:15
        Stock3-->{Name = ""Microsoft Corporation"";Symbol = ""MSFT"";LatestValue = 44.4;OpenValue = 44.48;CloseValue = 44.44;}
    Loc:Async before Download Thread:16
    Loc:Async after Download Thread:6
    Loc:OnNext Thread:15
        Stock3-->{Name = ""Microsoft Corporation"";Symbol = ""MSFT"";LatestValue = 44.4;OpenValue = 44.48;CloseValue = 44.44;}
    Loc:Async before Download Thread:5
    Loc:Async after Download Thread:6
    Loc:OnNext Thread:15
        Stock3-->{Name = ""Microsoft Corporation"";Symbol = ""MSFT"";LatestValue = 44.4;OpenValue = 44.48;CloseValue = 44.44;}
    Loc:OnCompleted Thread:15
Stock3 completed
*)

Query Style

Repeat the same example using Query style programming.

Create an "event" (observable) every 5 seconds (fast for testing pursose)
Use observe to create a Observable Sequence that gets Stocks data asynchoronous.
Use rxquery to take 5 elements of the Stock Observable Sequence.
Use observe to split into individual Stocks Observable Sequence.

 1: 
 2: 
 3: 
 4: 
 5: 
 6: 
 7: 
 8: 
 9: 
10: 
11: 
12: 
13: 
14: 
15: 
16: 
17: 
18: 
19: 
20: 
21: 
22:

let t1a=Observable.interval (TimeSpan.FromSeconds(5.0))
let QStocksA= observe {
        let! r=Observable.ofAsync (getStockData "GOOG,MSFT" "nsl1op") //Is it blocking?
        yield r
    }
let QStocksB= rxquery {
        for i in t1a do
        take 5
        for j in QStocksA do
        select j
}
let QStocksC1= observe {
        let! r=QStocksB
        yield r.[0].LatestValue
    }
let QStocksC2= observe {
        let! r=QStocksB
        yield r.[1].LatestValue
    }

let d3a=Helper.Dump3 "Stock1" QStocksC1
let d3b=Helper.Dump3 "Stock1" QStocksC2

Use d3a.Dispose() and d3b.Dispose() to stop observation.

To do:

Change the Query style solution to make it efficient.* To keep it simple, this Query solution creates one Observable Sequence for each Stock. This makes one Stock data internet collection per each Stock.

namespace System

namespace System.Collections

namespace System.Collections.Generic

namespace System.Threading

namespace System.Net

namespace System.Reactive

namespace System.Reactive.Linq

namespace System.Reactive.Disposables

namespace System.Reactive.Concurrency

namespace System.Reactive.Subjects

namespace Microsoft

namespace Microsoft.Reactive

namespace Microsoft.Reactive.Testing

namespace FSharp

namespace FSharp.Control

namespace FSharp.Control.Reactive

module Builders

from FSharp.Control.Reactive

module Observable

from FSharp.Control.Reactive

val name : string

Full name: Test_1.Helper.name

Multiple items
type String =
  new : value:char -> string + 7 overloads
  member Chars : int -> char
  member Clone : unit -> obj
  member CompareTo : value:obj -> int + 1 overload
  member Contains : value:string -> bool
  member CopyTo : sourceIndex:int * destination:char[] * destinationIndex:int * count:int -> unit
  member EndsWith : value:string -> bool + 2 overloads
  member Equals : obj:obj -> bool + 2 overloads
  member GetEnumerator : unit -> CharEnumerator
  member GetHashCode : unit -> int
  ...

Full name: System.String

--------------------
String(value: nativeptr<char>) : unit
String(value: nativeptr<sbyte>) : unit
String(value: char []) : unit
String(c: char, count: int) : unit
String(value: nativeptr<char>, startIndex: int, length: int) : unit
String(value: nativeptr<sbyte>, startIndex: int, length: int) : unit
String(value: char [], startIndex: int, length: int) : unit
String(value: nativeptr<sbyte>, startIndex: int, length: int, enc: Text.Encoding) : unit

field string.Empty

val PrintThread : location:string -> unit

Full name: Test_1.Helper.PrintThread

val location : string

Multiple items
val string : value:'T -> string

Full name: Microsoft.FSharp.Core.Operators.string

--------------------
type string = String

Full name: Microsoft.FSharp.Core.string

val printfn : format:Printf.TextWriterFormat<'T> -> 'T

Full name: Microsoft.FSharp.Core.ExtraTopLevelOperators.printfn

Multiple items
type Thread =
  inherit CriticalFinalizerObject
  new : start:ThreadStart -> Thread + 3 overloads
  member Abort : unit -> unit + 1 overload
  member ApartmentState : ApartmentState with get, set
  member CurrentCulture : CultureInfo with get, set
  member CurrentUICulture : CultureInfo with get, set
  member DisableComObjectEagerCleanup : unit -> unit
  member ExecutionContext : ExecutionContext
  member GetApartmentState : unit -> ApartmentState
  member GetCompressedStack : unit -> CompressedStack
  member GetHashCode : unit -> int
  ...

Full name: System.Threading.Thread

--------------------
Thread(start: ThreadStart) : unit
Thread(start: ParameterizedThreadStart) : unit
Thread(start: ThreadStart, maxStackSize: int) : unit
Thread(start: ParameterizedThreadStart, maxStackSize: int) : unit

property Thread.CurrentThread: Thread

property Thread.ManagedThreadId: int

val f1 : i:'T -> unit

Full name: Test_1.Helper.f1

val i : 'T

val f2 : ex:Exception -> unit

Full name: Test_1.Helper.f2

val ex : Exception

Multiple items
type Exception =
  new : unit -> Exception + 2 overloads
  member Data : IDictionary
  member GetBaseException : unit -> Exception
  member GetObjectData : info:SerializationInfo * context:StreamingContext -> unit
  member GetType : unit -> Type
  member HelpLink : string with get, set
  member InnerException : Exception
  member Message : string
  member Source : string with get, set
  member StackTrace : string
  ...

Full name: System.Exception

--------------------
Exception() : unit
Exception(message: string) : unit
Exception(message: string, innerException: exn) : unit

property Exception.Message: string

val f3 : unit -> unit

Full name: Test_1.Helper.f3

val Dump1 : name:string -> source:IObservable<'T> -> IDisposable

Full name: Test_1.Helper.Dump1

val name : string

val source : IObservable<'T>

type IObservable<'T> =
member Subscribe : observer:IObserver<'T> -> IDisposable

Full name: System.IObservable<_>

type Observable =
  static member Aggregate<'TSource> : source:IObservable<'TSource> * accumulator:Func<'TSource, 'TSource, 'TSource> -> IObservable<'TSource> + 2 overloads
  static member All<'TSource> : source:IObservable<'TSource> * predicate:Func<'TSource, bool> -> IObservable<bool>
  static member Amb<'TSource> : params sources:IObservable<'TSource>[] -> IObservable<'TSource> + 2 overloads
  static member And<'TLeft, 'TRight> : left:IObservable<'TLeft> * right:IObservable<'TRight> -> Pattern<'TLeft, 'TRight>
  static member Any<'TSource> : source:IObservable<'TSource> -> IObservable<bool> + 1 overload
  static member AsObservable<'TSource> : source:IObservable<'TSource> -> IObservable<'TSource>
  static member Average : source:IObservable<float> -> IObservable<float> + 19 overloads
  static member Buffer<'TSource, 'TBufferClosing> : source:IObservable<'TSource> * bufferClosingSelector:Func<IObservable<'TBufferClosing>> -> IObservable<IList<'TSource>> + 10 overloads
  static member Case<'TValue, 'TResult> : selector:Func<'TValue> * sources:IDictionary<'TValue, IObservable<'TResult>> -> IObservable<'TResult> + 2 overloads
  static member Cast<'TResult> : source:IObservable<obj> -> IObservable<'TResult>
  ...

Full name: System.Reactive.Linq.Observable

val subscribeWithCallbacks : onNext:('T -> unit) -> onError:(exn -> unit) -> onCompleted:(unit -> unit) -> observable:IObservable<'T> -> IDisposable

Full name: FSharp.Control.Reactive.Observable.subscribeWithCallbacks

val Dump2 : name:string -> source:IObservable<'T> -> IDisposable

Full name: Test_1.Helper.Dump2

val subscribeOn : scheduler:IScheduler -> source:IObservable<'Source> -> IObservable<'Source>

Full name: FSharp.Control.Reactive.Observable.subscribeOn

type Scheduler =
  static member AsLongRunning : scheduler:IScheduler -> ISchedulerLongRunning
  static member AsPeriodic : scheduler:IScheduler -> ISchedulerPeriodic
  static member AsStopwatchProvider : scheduler:IScheduler -> IStopwatchProvider
  static member Catch<'TException> : scheduler:IScheduler * handler:Func<'TException, bool> -> IScheduler
  static member CurrentThread : CurrentThreadScheduler
  static member Default : DefaultScheduler
  static member DisableOptimizations : scheduler:IScheduler -> IScheduler + 1 overload
  static member Immediate : ImmediateScheduler
  static member NewThread : IScheduler
  static member Normalize : timeSpan:TimeSpan -> TimeSpan
  ...

Full name: System.Reactive.Concurrency.Scheduler

property Scheduler.Default: DefaultScheduler

val Dump3 : name:string -> source:IObservable<'T> -> IDisposable

Full name: Test_1.Helper.Dump3

val observeOn : scheduler:IScheduler -> source:IObservable<'a> -> IObservable<'a>

Full name: FSharp.Control.Reactive.Observable.observeOn

val url : string

Full name: Test_1.url

type Stocks =
  {Name: string;
   Symbol: string;
   LatestValue: float;
   OpenValue: float;
   CloseValue: float;}

Full name: Test_1.Stocks

Stocks.Name: string

Stocks.Symbol: string

Stocks.LatestValue: float

Multiple items
val float : value:'T -> float (requires member op_Explicit)

Full name: Microsoft.FSharp.Core.Operators.float

--------------------
type float = Double

Full name: Microsoft.FSharp.Core.float

--------------------
type float<'Measure> = float

Full name: Microsoft.FSharp.Core.float<_>

Stocks.OpenValue: float

Stocks.CloseValue: float

val getStockData : stock:string -> fields:string -> Async<Stocks []>

Full name: Test_1.getStockData

val stock : string

val fields : string

val async : AsyncBuilder

Full name: Microsoft.FSharp.Core.ExtraTopLevelOperators.async

val wc : WebClient

Multiple items
type WebClient =
  inherit Component
  new : unit -> WebClient
  member BaseAddress : string with get, set
  member CachePolicy : RequestCachePolicy with get, set
  member CancelAsync : unit -> unit
  member Credentials : ICredentials with get, set
  member DownloadData : address:string -> byte[] + 1 overload
  member DownloadDataAsync : address:Uri -> unit + 1 overload
  member DownloadFile : address:string * fileName:string -> unit + 1 overload
  member DownloadFileAsync : address:Uri * fileName:string -> unit + 1 overload
  member DownloadString : address:string -> string + 1 overload
  ...

Full name: System.Net.WebClient

--------------------
WebClient() : unit

module Helper

from Test_1

val data : string

member WebClient.AsyncDownloadString : address:Uri -> Async<string>

Multiple items
type Uri =
  new : uriString:string -> Uri + 5 overloads
  member AbsolutePath : string
  member AbsoluteUri : string
  member Authority : string
  member DnsSafeHost : string
  member Equals : comparand:obj -> bool
  member Fragment : string
  member GetComponents : components:UriComponents * format:UriFormat -> string
  member GetHashCode : unit -> int
  member GetLeftPart : part:UriPartial -> string
  ...

Full name: System.Uri

--------------------
Uri(uriString: string) : unit
Uri(uriString: string, uriKind: UriKind) : unit
Uri(baseUri: Uri, relativeUri: string) : unit
Uri(baseUri: Uri, relativeUri: Uri) : unit

val dataLines : string []

String.Split(params separator: char []) : string []
String.Split(separator: string [], options: StringSplitOptions) : string []
String.Split(separator: char [], options: StringSplitOptions) : string []
String.Split(separator: char [], count: int) : string []
String.Split(separator: string [], count: int, options: StringSplitOptions) : string []
String.Split(separator: char [], count: int, options: StringSplitOptions) : string []

type StringSplitOptions =
| None = 0
| RemoveEmptyEntries = 1

Full name: System.StringSplitOptions

field StringSplitOptions.RemoveEmptyEntries = 1

val s : Stocks []

Multiple items
val seq : sequence:seq<'T> -> seq<'T>

Full name: Microsoft.FSharp.Core.Operators.seq

--------------------
type seq<'T> = IEnumerable<'T>

Full name: Microsoft.FSharp.Collections.seq<_>

val line : string

val infos : string []

type Array =
  member Clone : unit -> obj
  member CopyTo : array:Array * index:int -> unit + 1 overload
  member GetEnumerator : unit -> IEnumerator
  member GetLength : dimension:int -> int
  member GetLongLength : dimension:int -> int64
  member GetLowerBound : dimension:int -> int
  member GetUpperBound : dimension:int -> int
  member GetValue : params indices:int[] -> obj + 7 overloads
  member Initialize : unit -> unit
  member IsFixedSize : bool
  ...

Full name: System.Array

val ofSeq : source:seq<'T> -> 'T []

Full name: Microsoft.FSharp.Collections.Array.ofSeq

val r : Stocks []

Full name: Test_1.r

Multiple items
type Async
static member AsBeginEnd : computation:('Arg -> Async<'T>) -> ('Arg * AsyncCallback * obj -> IAsyncResult) * (IAsyncResult -> 'T) * (IAsyncResult -> unit)
static member AwaitEvent : event:IEvent<'Del,'T> * ?cancelAction:(unit -> unit) -> Async<'T> (requires delegate and 'Del :> Delegate)
static member AwaitIAsyncResult : iar:IAsyncResult * ?millisecondsTimeout:int -> Async<bool>
static member AwaitTask : task:Task -> Async<unit>
static member AwaitTask : task:Task<'T> -> Async<'T>
static member AwaitWaitHandle : waitHandle:WaitHandle * ?millisecondsTimeout:int -> Async<bool>
static member CancelDefaultToken : unit -> unit
static member Catch : computation:Async<'T> -> Async<Choice<'T,exn>>
static member FromBeginEnd : beginAction:(AsyncCallback * obj -> IAsyncResult) * endAction:(IAsyncResult -> 'T) * ?cancelAction:(unit -> unit) -> Async<'T>
static member FromBeginEnd : arg:'Arg1 * beginAction:('Arg1 * AsyncCallback * obj -> IAsyncResult) * endAction:(IAsyncResult -> 'T) * ?cancelAction:(unit -> unit) -> Async<'T>
static member FromBeginEnd : arg1:'Arg1 * arg2:'Arg2 * beginAction:('Arg1 * 'Arg2 * AsyncCallback * obj -> IAsyncResult) * endAction:(IAsyncResult -> 'T) * ?cancelAction:(unit -> unit) -> Async<'T>
static member FromBeginEnd : arg1:'Arg1 * arg2:'Arg2 * arg3:'Arg3 * beginAction:('Arg1 * 'Arg2 * 'Arg3 * AsyncCallback * obj -> IAsyncResult) * endAction:(IAsyncResult -> 'T) * ?cancelAction:(unit -> unit) -> Async<'T>
static member FromContinuations : callback:(('T -> unit) * (exn -> unit) * (OperationCanceledException -> unit) -> unit) -> Async<'T>
static member Ignore : computation:Async<'T> -> Async<unit>
static member OnCancel : interruption:(unit -> unit) -> Async<IDisposable>
static member Parallel : computations:seq<Async<'T>> -> Async<'T []>
static member RunSynchronously : computation:Async<'T> * ?timeout:int * ?cancellationToken:CancellationToken -> 'T
static member Sleep : millisecondsDueTime:int -> Async<unit>
static member Start : computation:Async<unit> * ?cancellationToken:CancellationToken -> unit
static member StartAsTask : computation:Async<'T> * ?taskCreationOptions:TaskCreationOptions * ?cancellationToken:CancellationToken -> Task<'T>
static member StartChild : computation:Async<'T> * ?millisecondsTimeout:int -> Async<Async<'T>>
static member StartChildAsTask : computation:Async<'T> * ?taskCreationOptions:TaskCreationOptions -> Async<Task<'T>>
static member StartImmediate : computation:Async<unit> * ?cancellationToken:CancellationToken -> unit
static member StartWithContinuations : computation:Async<'T> * continuation:('T -> unit) * exceptionContinuation:(exn -> unit) * cancellationContinuation:(OperationCanceledException -> unit) * ?cancellationToken:CancellationToken -> unit
static member SwitchToContext : syncContext:SynchronizationContext -> Async<unit>
static member SwitchToNewThread : unit -> Async<unit>
static member SwitchToThreadPool : unit -> Async<unit>
static member TryCancelled : computation:Async<'T> * compensation:(OperationCanceledException -> unit) -> Async<'T>
static member CancellationToken : Async<CancellationToken>
static member DefaultCancellationToken : CancellationToken

Full name: Microsoft.FSharp.Control.Async

--------------------
type Async<'T>

Full name: Microsoft.FSharp.Control.Async<_>

static member Async.RunSynchronously : computation:Async<'T> * ?timeout:int * ?cancellationToken:CancellationToken -> 'T

val t1 : IObservable<int64>

Full name: Test_1.t1

val interval : period:TimeSpan -> IObservable<int64>

Full name: FSharp.Control.Reactive.Observable.interval

Multiple items
type TimeSpan =
  struct
    new : ticks:int64 -> TimeSpan + 3 overloads
    member Add : ts:TimeSpan -> TimeSpan
    member CompareTo : value:obj -> int + 1 overload
    member Days : int
    member Duration : unit -> TimeSpan
    member Equals : value:obj -> bool + 1 overload
    member GetHashCode : unit -> int
    member Hours : int
    member Milliseconds : int
    member Minutes : int
    ...
  end

Full name: System.TimeSpan

--------------------
TimeSpan()
TimeSpan(ticks: int64) : unit
TimeSpan(hours: int, minutes: int, seconds: int) : unit
TimeSpan(days: int, hours: int, minutes: int, seconds: int) : unit
TimeSpan(days: int, hours: int, minutes: int, seconds: int, milliseconds: int) : unit

TimeSpan.FromSeconds(value: float) : TimeSpan

val o1 : IObservable<int64>

Full name: Test_1.o1

val take : n:int -> source:IObservable<'Source> -> IObservable<'Source>

Full name: FSharp.Control.Reactive.Observable.take

val o2 : IObservable<Stocks []>

Full name: Test_1.o2

Observable.SelectMany<'TSource,'TResult>(source: IObservable<'TSource>, selector: Func<'TSource,int,IEnumerable<'TResult>>) : IObservable<'TResult>
   (+0 other overloads)
Observable.SelectMany<'TSource,'TResult>(source: IObservable<'TSource>, selector: Func<'TSource,IEnumerable<'TResult>>) : IObservable<'TResult>
   (+0 other overloads)
Observable.SelectMany<'TSource,'TResult>(source: IObservable<'TSource>, selector: Func<'TSource,int,CancellationToken,Tasks.Task<'TResult>>) : IObservable<'TResult>
   (+0 other overloads)
Observable.SelectMany<'TSource,'TResult>(source: IObservable<'TSource>, selector: Func<'TSource,CancellationToken,Tasks.Task<'TResult>>) : IObservable<'TResult>
   (+0 other overloads)
Observable.SelectMany<'TSource,'TResult>(source: IObservable<'TSource>, selector: Func<'TSource,int,Tasks.Task<'TResult>>) : IObservable<'TResult>
   (+0 other overloads)
Observable.SelectMany<'TSource,'TResult>(source: IObservable<'TSource>, selector: Func<'TSource,Tasks.Task<'TResult>>) : IObservable<'TResult>
   (+0 other overloads)
Observable.SelectMany<'TSource,'TResult>(source: IObservable<'TSource>, selector: Func<'TSource,int,IObservable<'TResult>>) : IObservable<'TResult>
   (+0 other overloads)
Observable.SelectMany<'TSource,'TResult>(source: IObservable<'TSource>, selector: Func<'TSource,IObservable<'TResult>>) : IObservable<'TResult>
   (+0 other overloads)
Observable.SelectMany<'TSource,'TOther>(source: IObservable<'TSource>, other: IObservable<'TOther>) : IObservable<'TOther>
   (+0 other overloads)
Observable.SelectMany<'TSource,'TCollection,'TResult>(source: IObservable<'TSource>, collectionSelector: Func<'TSource,int,IEnumerable<'TCollection>>, resultSelector: Func<'TSource,int,'TCollection,int,'TResult>) : IObservable<'TResult>
   (+0 other overloads)

val ofAsync : asyncOperation:Async<'a> -> IObservable<'a>

Full name: FSharp.Control.Reactive.Observable.ofAsync

val oStock1 : IObservable<Stocks>

Full name: Test_1.oStock1

Multiple items
val map : f:('a -> 'b) -> source:IObservable<'a> -> IObservable<'b>

Full name: FSharp.Control.Reactive.Observable.map

--------------------
val map : mapping:('T -> 'U) -> source:IObservable<'T> -> IObservable<'U>

Full name: Microsoft.FSharp.Control.Observable.map

val i : Stocks []

val oStock2 : IObservable<Stocks>

Full name: Test_1.oStock2

val d1 : IDisposable

Full name: Test_1.d1

val d2 : IDisposable

Full name: Test_1.d2

val d3 : IDisposable

Full name: Test_1.d3

val t1a : IObservable<int64>

Full name: Test_1.t1a

val QStocksA : IObservable<Stocks []>

Full name: Test_1.QStocksA

val observe : ObservableBuilder

Full name: FSharp.Control.Reactive.Builders.observe

val r : Stocks []

val QStocksB : IObservable<Stocks []>

Full name: Test_1.QStocksB

val rxquery : RxQueryBuilder

Full name: FSharp.Control.Reactive.Builders.rxquery

val i : int64

custom operation: take (int)

Calls RxQueryBuilder.Take

val j : Stocks []

custom operation: select ('a)

Calls RxQueryBuilder.Select

val QStocksC1 : IObservable<float>

Full name: Test_1.QStocksC1

val QStocksC2 : IObservable<float>

Full name: Test_1.QStocksC2

val d3a : IDisposable

Full name: Test_1.d3a

val d3b : IDisposable

Full name: Test_1.d3b

Celso Axelrud

Wednesday, July 15, 2015

HIGH PERFORMANCE REAL-TIME STOCKS COMPUTATION ENGINE - part 1

HIGH PERFORMANCE REAL-TIME STOCKS COMPUTATION ENGINE - part 1

Example steps

Define references to the libraries.

Define a Helper module.

Define an Asynchonous Workflow to get Stock data.

Create Observables Sequences.

Observe with different Observers

Comments:

Case 1

Case 2

Case 3

Query Style

To do:

No comments:

Post a Comment

Followers

Blog Archive