Luca Bolognese's WebLog

Luca Bolognese leaves Microsoft

lucabol — Fri, 19 Feb 2010 16:58:02 GMT

This is my last post on this blog. My new blog is here: http://lucabolognese.wordpress.com/

I accepted a role as Director for Credit Suisse in London. I’m excited by the opportunity to work in the financial industry, a long-standing desire of mine. I’m also excited to write more F# code and to be closer to Italy, where my extended family is.

The past ten years in Microsoft have been a wild ride. I’m proud to have been part of ObjectSpaces, Generics, LINQ and F# (and much more …). I’ve been lucky to be able to post on this blog and present at conferences about such innovative technologies. It has all been a lot of fun. I’m sure the next ten years will be as good.

I’ll see you guys on my new blog.

LChart: displaying charts in F# – Part I

lucabol — Wed, 17 Feb 2010 19:20:00 GMT

I want to use F# as a exploratory data analysis language (like R). But I don’t know how to get the same nice graphic capabilities. So I decided to create them. Here is a library to draw charts in F#. It steals ideas from this book and this R package. It is nothing more than a wrapper on top of the Microsoft Chart Controls to give it a more ‘exploratory’ one line calling syntax. It is also rough work in progress: I don’t wrap all the chart types and there are bugs in the ones I wrap. Also the architecture is all wrong (more on this in another post). But it’s a start and it kind of works. Attached the full code.

I will continue this series in my new blog at wordpress: http://lucabolognese.wordpress.com/. The reason I need a new blog will be explained in an upcoming post.

Part II is now here.

Ok, let’s start. How do I draw a chart?

let x = [1.;2.5;3.1;4.;4.8;6.0;7.5;8.;9.1;15.]
let y = [1.6;2.1;1.4;4.;2.3;1.9;2.4;1.4;5.;2.9]

lc.scatter(x, y) |> display

X and Y are just some make up data. lc is the name of a class (????) and scatter is a static method on it. scatter doesn’t display the chart, it just produces a an object that represents the chart. Display displays the chart. The reason for using the bizarre lc static class is that I want it to be short so that it is easy to type in the fsi.exe. At the same time it needs to support optional parameters (which are not supported on top level functions in F#).

You get a window with this chart on it. You can press CTRL+C to copy it (as I did to post it here).

You might want to customize the chart a bit by passing some of these famous optional parameters:

lc.scatter(x = x, y = y, markerSize = 10, markerStyle = MarkerStyle.Diamond,
    xname = "Players", yname = "Ratings", title = "Players' Ratings")  |> display

Or you might want to print different types of charts:

lc.line(y = y, markerSize = 10, markerStyle = MarkerStyle.Diamond, xname = "Players", yname = "Ratings", title = "Players' Ratings", isValueShownAsLabel = true,
    color = Color.Red) |> display

lc.spline(x = x, y = y, markerSize = 10, markerStyle = MarkerStyle.Diamond, xname = "Players", yname = "Ratings",
    title = "Players' Ratings", isValueShownAsLabel = true, color = Color.Red) |> display

lc.stepline(x = x, y = y, markerSize = 10, markerStyle = MarkerStyle.Diamond, xname = "Players", yname = "Ratings",
    title = "Players' Ratings", isValueShownAsLabel = true, color = Color.Red) |> display

lc.bar(y = y, xname = "Players", yname = "Ratings", title = "Players' Ratings", isValueShownAsLabel = true,
    drawingStyle = "Emboss") |> display

lc.column(y = y, xname = "Players", yname = "Ratings", title = "Players' Ratings",
    isValueShownAsLabel = true, drawingStyle = "Cylinder") |> display

lc.boxplot(y = y, xname = "Players", yname = "Ratings", title = "Players' Ratings", color = Color.Blue, whiskerPercentile = 5, percentile = 30,
    showAverage = false, showMedian = false, showUnusualValues = true) |> display

Ok, the last one is weird. You probably want more than one boxplot in a chart. I'll show you how to do that in the next post.

The next post will be on how to have more than one series on the same chart and more than one chart in the same windows. Something like the below:

Excel Financial Functions upgraded to Visual Studio 2010 RC

lucabol — Tue, 16 Feb 2010 22:55:00 GMT

Here is a link to the home page: http://code.msdn.microsoft.com/FinancialFunctions/Wiki/View.aspx?title=Home

New version of LAgent compatible with VS 2010 RC online

lucabol — Mon, 15 Feb 2010 17:36:54 GMT

My crazy agent framework now works with RC: http://code.msdn.microsoft.com/LAgent

A simpler F# MailboxProcessor

lucabol — Fri, 12 Feb 2010 23:29:46 GMT

I always forget the pattern to use to create a functioning MailboxProcessor in F#. I mean, which piece has to be async and how to structure the recursive loop. When I find myself in that kind of a situation situation, my instincts scream at me: “Wrap it and make it work how your mind expects it to work”. So here is a simplification of the paradigm.

Let’s see how some standard MailboxProcessor code looks like:

let counter0 =
    MailboxProcessor.Start(fun inbox ->
        let rec loop n =
            async { 
                    let! msg = inbox.Receive()
                    return! loop(n+msg) }
        loop 0)

This keeps a running sum of the messages it receives. The only part that is really unique to this guy is “n + msg”. All the rest is infrastructure.

You’d probably prefer to write code like the following:

let counter1 = MailboxProcessor.SpawnAgent( (fun msg n -> msg + n), 0)

Yep, just one line of code. But, is it possible? Let’s look at one way of doing it:

type AfterError<'state> =
| ContinueProcessing of 'state
| StopProcessing
| RestartProcessing
    
type MailboxProcessor<'a> with

    static member public SpawnAgent<'b>(messageHandler :'a->'b->'b,
                                        initialState : 'b, ?timeout:'b -> int,
                                        ?timeoutHandler:'b -> AfterError<'b>,
                                        ?errorHandler:
                                            Exception -> 'a option -> 'b -> AfterError<'b>)
                                        : MailboxProcessor<'a> =
        let timeout = defaultArg timeout (fun _ -> -1)
        let timeoutHandler = defaultArg timeoutHandler (fun state –>
                                                                  ContinueProcessing(state))
        let errorHandler = defaultArg errorHandler (fun _ _ state –>
                                                                  ContinueProcessing(state))
        MailboxProcessor.Start(fun inbox ->
            let rec loop(state) = async {
                let! msg = inbox.TryReceive(timeout(state))
                try
                    match msg with
                    | None      -> match timeoutHandler state with
                                    | ContinueProcessing(newState)    ->
                                                                   return! loop(newState)
                                    | StopProcessing        -> return ()
                                    | RestartProcessing     -> return! loop(initialState)
                    | Some(m)   -> return! loop(messageHandler m state)
                with
                | ex -> match errorHandler ex msg state with
                        | ContinueProcessing(newState)    -> return! loop(newState)
                        | StopProcessing        -> return ()
                        | RestartProcessing     -> return! loop(initialState)
                }
            loop(initialState))

The funny formatting is because I have to fit it in the small horizontal space of this blog. In any case, this is just a simple (?) wrapper for the MailboxProcessor pattern. The function takes two necessary parameters and two optional ones:

messageHandler: a function to execute when a message comes in, it takes the message and the current state as parameters and returns the new state.
initialState: the initial state for the MailboxProcessor
timeoutHandler: a function that is executed whenever a timeout occurs. It takes as a parameter the current state and returns one of ContinueProcessing(newState), StopProcessing or RestartProcessing
errorHandler: a function that gets call if an exception is generated inside the messageHandler function. It takes the exception, the message, the current state and returns ContinueProcessing(newState), StopProcessing or RestartProcessing

An example of how to use errorHandler to implement the CountingAgent in the Expert F# book follows:

type msg = Increment of int | Fetch of AsyncReplyChannel<int> | Stop

exception StopException

type CountingAgent() =
    let counter = MailboxProcessor.SpawnAgent((fun msg n ->
                    match msg with
                    | Increment m ->  n + m
                    | Stop -> raise(StopException)
                    | Fetch replyChannel ->
                        do replyChannel.Reply(n)
                        n
                  ), 0, errorHandler = (fun _ _ _ -> StopProcessing))
    member a.Increment(n) = counter.Post(Increment(n))
    member a.Stop() = counter.Post(Stop)
    member a.Fetch() = counter.PostAndReply(fun replyChannel -> Fetch(replyChannel))    
        
let counter2 = CountingAgent()
counter2.Increment(1)
counter2.Fetch()
counter2.Increment(2)
counter2.Fetch()
counter2.Stop()

Sometimes your agent doesn’t need a state, it is purely stateless. Something as simple as the following:

let echo = MailboxProcessor<_>.SpawnWorker(fun msg -> printfn "%s" msg)

You can easily make that happen by using this toned down version of an agent, called worker:

static member public SpawnWorker(messageHandler,  ?timeout, ?timeoutHandler,?errorHandler) =
    let timeout = defaultArg timeout (fun () -> -1)
    let timeoutHandler = defaultArg timeoutHandler (fun _ -> ContinueProcessing(()))
    let errorHandler = defaultArg errorHandler (fun _ _ -> ContinueProcessing(()))
    MailboxProcessor.SpawnAgent((fun msg _ -> messageHandler msg; ()),
                                 (), timeout, timeoutHandler,
                                 (fun ex msg _ -> errorHandler ex msg))

Given that they are parallel, you might want to run a whole bunch of them at the same time. You might want something that looks like a worker, but that, under the cover, execute each messageHandler in parallel. Something like:

type msg1 = Message1 | Message2 of int | Message3 of string
            
let a = MailboxProcessor.SpawnParallelWorker(function
                | Message1 -> printfn "Message1";
                | Message2 n -> printfn "Message2 %i" n;
                | Message3 _ -> failwith "I failed"
                , 10
                , errorHandler = (fun ex _ -> printfn "%A" ex; ContinueProcessing()))


a.Post(Message1)
a.Post(Message2(100))
a.Post(Message3("abc"))
a.Post(Message2(100))

In this example, the different messages, are likely to cause things to print out of order. Notice the number 10 above which is how many workers you want to process your messages. This is implemented by round-robin messages to the various workers:

static member public SpawnParallelWorker(messageHandler, howMany, ?timeout,
                                                        ?timeoutHandler,?errorHandler) =
    let timeout = defaultArg timeout (fun () -> -1)
    let timeoutHandler = defaultArg timeoutHandler (fun _ -> ContinueProcessing(()))
    let errorHandler = defaultArg errorHandler (fun _ _ -> ContinueProcessing(()))
    MailboxProcessor<'a>.SpawnAgent((fun msg (workers:MailboxProcessor<'a> array, index) ->
                                        workers.[index].Post msg
                                        (workers, (index + 1) % howMany))  
                                    , (Array.init howMany
                                      (fun _ -> MailboxProcessor<'a>.SpawnWorker(
                                                 messageHandler, timeout, timeoutHandler,
                                                 errorHandler)), 0))

One drawback with the current code is that it doesn’t supports cancellations. It should be possible to wrap that too, but I haven’t done it (yet). If you don’t want to cut and paste the code, it is inside the AgentSystem.fs file here.

The F#unctional Londoners Meetup Group

lucabol — Thu, 11 Feb 2010 17:24:05 GMT

If you live in London and work with F#, you should join http://www.meetup.com/FSharpLondon/

Also, if you don’t work with F#, but you’d like too, and you live in London you should join it. Also if you don’t live in London, but you’d like to and you’d like not to work on F# … Ok, you get the point.

LAgent: an agent framework in F# – Part X – ActiveObject

lucabol — Thu, 10 Dec 2009 18:09:14 GMT

Download framework here.

All posts are here:

If you stare long enough at agents, you start to realize that they are just ‘glorified locks’. They are a convenient programming model to protect a resource from concurrent access. The programming model is convenient because both the client and the server can write their code without worrying about concurrency problems, and yet the program runs in parallel. Protecting a resource sounds a lot like state encapsulation and the concept of state encapsulation is what object orientation is all about.

So you start thinking if there is a way to enhance vanilla objects to make them agents. You want to reuse all the concepts that you are familiar with (i.e. inheritance, visibility rules, etc…) and you want your clients to call agents as if they were calling normal objects. Obviously, under the cover, the method calls won’t execute immediately, but they would be queued. Let’s look at an example.

This is our simple counter agent:

type CounterMessage =
| Add of int
| Print

let counterF = fun msg count ->
    match msg with
    | Add(i)    -> count + i
    | Print     -> printfn "The value is %i" count; count
    
let c1 = spawnAgent counterF 0
c1 <-- Add(3)
c1 <—Print

As nice as this looks, there are unfamiliar things in this model:

The communication is through messages. This requires packing and unpacking which, albeit easy in F#, is unfamiliar and feels like machinery that we’d like to get rid off.
The management of state is bizarre, it gets passed into the lambda and returned from it instead of being represented as fields and properties on the agent

My best attempt at creating an object-like syntax follows:

type Counter() =
    let w = new WorkQueue()
    let mutable count = 0
    member c.Add x = w.Queue (fun () -> 
        count <- count + x
        )
    member c.Print () = w.Queue (fun () -> 
        printfn "The value is %i" count
        )

let c = new Counter()
c.Add 3
c.Print

With this syntax, you write your agents like you write your vanilla classes except:

You need a private field of type WorkQueue
You need to write your methods as lambdas passed to the WorkQueue.Queue function
Your methods cannot return values

The most worrisome of these constraints is 2. because you can easily forget about it. If you do forget, then everything compiles just fine, but it doesn’t do what you expect. That’s pure badness. I haven’t found a way to enforce it. This is a place where the language could help me. Other than that, the whole model works rather nicely.

Regarding the third point, you can concoct a programming model that allows you to return values from your methods. Here it is:

member c.CountTask = w.QueueWithTask(fun () ->
    count
    )
member c.CountAsync = w.QueueWithAsync(fun () ->
    count
    )

printfn "The count using Task is %i" (c.CountTask.Result)

The first method returns a Task; the second method returns an AsyncResultCell. Both are ways to represent a promise. The latter allows a natural integration with the async block in F# as in the following code:

Async.RunSynchronously (
            async {
                let! count = c.CountAsync
                printfn "The countusing Async is %i" count
            })

As for myself, I don’t like methods returning values. Every time I use them, I end up going back and thinking about my problem in a traditional way, aka as method calls that return results, instead of thinking about it in a more actor oriented fashion. I end up waiting for these promises to be materialized and, by doing so, I limit the amount of parallelism that I unleash. As a matter of fact, the whole business of hiding the message passing nature of the programming model is dubious. It makes for a nicer syntax, but you need to make an extra effort in your mind to translate it to what it really is: just message passing with a nice syntactical veneer. I haven’t decided yet which model I like the most.

You should have a sense of what WorkQueue is. In essence, it is a Mailbox of lambdas (look at the red bold code below).

type WorkQueue() =
    let workQueue = spawnWorker (fun f -> f())
    member w.Queue (f) = workQueue <-- f
    member w.QueueWithTask f : Task<'T> =
        let source = new TaskCompletionSource<_>()
        workQueue <-- (fun () -> f() |> source.SetResult)
        source.Task
    member w.QueueWithAsync (f:unit -> 'T) : Async<'T> =
        let result = new AsyncResultCell<'T>()
        workQueue <-- (fun () -> f() |> result.RegisterResult )
        result.AsyncWaitResult
    member w.Restart () = workQueue <-! Restart
    member w.Stop () = workQueue <-! Stop
    member w.SetErrorHandler(h) =
        let managerF = fun (_, name:string, ex:Exception, _, _, _) -> h name ex                             
        let manager = spawnWorker managerF
        workQueue <-! SetManager manager
    member w.SetName(name) = workQueue <-! SetName(name)
    member w.SetQueueHandler(g) = workQueue <-! SetWorkerHandler(g)
    member w.SetTimeoutHandler(timeout, f) = workQueue <-! SetTimeoutHandler(timeout, f)

I implemented all the services that are in the message passing model. The two are equivalent as expressing power goes. In case you wonder how a real piece of code looks like using this model, here is an ActiveObject version of the map reduce algorithm. One of these days, I will gather the strength to go trough this code and explain what it does, but not today :-)

#load "AgentSystem.fs"
open AgentSystem.LAgent
open System
open System.Collections
open System.Collections.Generic
open System.Threading

type IOutput<'out_key, 'out_value> =
    abstract Reduced: 'out_key -> seq<'out_value> -> unit
    abstract MapReduceDone: unit -> unit
    
type Mapper<'in_key, 'in_value, 'my_out_key, 'out_value when 'my_out_key : comparison>
                                                      (map:'in_key -> 'in_value -> seq<'my_out_key * 'out_value>, i, partitionF) =
    let w = new WorkQueue()
    let mutable reducerTracker: BitArray = null
    let mutable controller = Unchecked.defaultof<Controller<'in_key, 'in_value, 'my_out_key, 'out_value>>
    let mutable reducers = Unchecked.defaultof<Reducer<'in_key, 'in_value, 'my_out_key, 'out_value> array>
    member m.Init c reds =
        w.Queue (fun () ->
            controller <- c
            reducers <- reds
            reducerTracker <- new BitArray(reducers.Length, false))
    member m.Process inKey inValue =
        w.Queue (fun () ->
            let outKeyValues = map inKey inValue
            outKeyValues |> Seq.iter (fun (outKey, outValue) ->
                                        let reducerUsed = partitionF outKey (reducers.Length)
                                        reducerTracker.Set(reducerUsed, true)
                                        reducers.[reducerUsed].Add(outKey, outValue)))
    member m.Done () =
        w.Queue (fun () ->
            controller.MapDone i reducerTracker)
    member m.Stop () = w.Stop ()            
    
and Reducer<'in_key, 'in_value, 'out_key, 'out_value when 'out_key :
                     comparison>(reduce:'out_key -> seq<'out_value> -> seq<'out_value>, i, output:IOutput<'out_key, 'out_value>) =
    let w = new WorkQueue()
    let mutable workItems = new List<'out_key * 'out_value>()
    let mutable controller = Unchecked.defaultof<Controller<'in_key, 'in_value, 'out_key, 'out_value>>
    member r.Init c =
        w.Queue (fun () ->
            controller <- c)
    member r.StartReduction () =
        w.Queue (fun () ->
            workItems
            |> Seq.groupBy fst
            |> Seq.sortBy fst
            |> Seq.map (fun (key, values) -> (key, reduce key (values |> Seq.map snd)))
            |> Seq.iter (fun (key, value) -> output.Reduced key value)
            controller.ReductionDone i) 
    member r.Add (outKey:'out_key, outValue:'out_value) : unit =
        w.Queue (fun () ->
            workItems.Add((outKey, outValue)))
    member m.Stop () = w.Stop ()            
                
and Controller<'in_key, 'in_value, 'out_key, 'out_value when 'out_key : comparison>(output:IOutput<'out_key, 'out_value>) =
    let w = new WorkQueue()
    let mutable mapperTracker: BitArray = null
    let mutable reducerUsedByMappers: BitArray = null
    let mutable reducerDone: BitArray = null
    let mutable mappers = Unchecked.defaultof<Mapper<'in_key, 'in_value, 'out_key, 'out_value> array>
    let mutable reducers = Unchecked.defaultof<Reducer<'in_key, 'in_value, 'out_key, 'out_value> array>
    let BAtoSeq (b:BitArray) = [for x in b do yield x]
    member c.Init maps reds =
        w.Queue (fun () ->
            mappers <- maps
            reducers <- reds
            mapperTracker <- new BitArray(mappers.Length, false)
            reducerUsedByMappers <- new BitArray(reducers.Length, false)
            reducerDone <- new BitArray(reducers.Length, false))
    member c.MapDone (i : int) (reducerTracker : BitArray) : unit =
        w.Queue (fun () ->
            mapperTracker.Set(i, true)
            let reducerUsedByMappers = reducerUsedByMappers.Or(reducerTracker)
            if not( BAtoSeq mapperTracker |> Seq.exists(fun bit -> bit = false)) then
                BAtoSeq reducerUsedByMappers |> Seq.iteri (fun i r -> if r = true then reducers.[i].StartReduction ())
                mappers |> Seq.iter (fun m -> m.Stop ())
              )
    member c.ReductionDone (i: int) : unit =
        w.Queue (fun () ->
            reducerDone.Set(i, true)
            if BAtoSeq reducerDone |> Seq.forall2 (fun x y -> x = y) (BAtoSeq reducerUsedByMappers) then
                output.MapReduceDone ()
                reducers |> Seq.iter (fun r -> r.Stop ())
                c.Stop()
             )         
    member m.Stop () = w.Stop ()
                
let mapReduce   (inputs:seq<'in_key * 'in_value>)
                (map:'in_key -> 'in_value -> seq<'out_key * 'out_value>)
                (reduce:'out_key -> seq<'out_value> -> seq<'out_value>)
                (output:IOutput<'out_key, 'out_value>)
                M R partitionF =
                    
    let len = inputs |> Seq.length
    let M = if len < M then len else M
    
    let mappers = Array.init M (fun i -> new Mapper<'in_key, 'in_value, 'out_key, 'out_value>(map, i, partitionF))
    let reducers = Array.init R (fun i -> new Reducer<'in_key, 'in_value, 'out_key, 'out_value>(reduce, i, output))
    let controller = new Controller<'in_key, 'in_value, 'out_key, 'out_value>(output)
    
    mappers |> Array.iter (fun m -> m.Init controller reducers)
    reducers |> Array.iter (fun r -> r. Init controller )
    controller.Init mappers reducers
    
    inputs |> Seq.iteri (fun i (inKey, inValue) -> mappers.[i % M].Process inKey inValue)
    mappers |> Seq.iter (fun m -> m.Done ())    

let partitionF = fun key M -> abs(key.GetHashCode()) % M 

let map = fun (fileName:string) (fileContent:string) ->
            let l = new List<string * int>()
            let wordDelims = [|' ';',';';';'.';':';'?';'!';'(';')';'\n';'\t';'\f';'\r';'\b'|]
            fileContent.Split(wordDelims) |> Seq.iter (fun word -> l.Add((word, 1)))
            l :> seq<string * int>
                                  
let reduce = fun key (values:seq<int>) -> [values |> Seq.sum] |> seq<int>

let printer () =
  { new IOutput<string, int> with
        member o.Reduced key values = printfn "%A %A" key values
        member o.MapReduceDone () = printfn "All done!!"}
    
let testInput =
     ["File1", "I was going to the airport when I saw someone crossing"; "File2", "I was going home when I saw you coming toward me"]   
mapReduce testInput map reduce (printer ()) 2 2 partitionF

open System.IO
open System.Text

let gatherer(step) =
  let w = new WorkQueue()
  let data = new List<string * int>()
  let counter = ref 0 
  { new IOutput<string, int> with
        member o.Reduced key values =
            w.Queue (fun () ->
                if !counter % step = 0 then
                    printfn "Processed %i words. Now processing %s" !counter key 
                data.Add((key, values |> Seq.hd))
                counter := !counter + 1)
        member o.MapReduceDone () =
            w.Queue (fun () ->
                data
                |> Seq.distinctBy (fun (key, _) -> key.ToLower())
                |> Seq.filter (fun (key, _) -> not(key = "" || key = "\"" || (fst (Double.TryParse(key)))))
                |> Seq.to_array
                |> Array.sortBy snd
                |> Array.rev
                |> Seq.take 20
                |> Seq.iter (fun (key, value) -> printfn "%A\t\t%A" key value)
                printfn "All done!!")           
        }
                        
let splitBook howManyBlocks fileName =
    let buffers = Array.init howManyBlocks (fun _ -> new StringBuilder())
    fileName
    |> File.ReadAllLines
    |> Array.iteri (fun i line -> buffers.[i % (howManyBlocks)].Append(line) |> ignore)
    buffers

let blocks1 = __SOURCE_DIRECTORY__ + "\kjv10.txt" |> splitBook 100
let blocks2 = __SOURCE_DIRECTORY__ + "\warandpeace.txt" |> splitBook 100
let input =
    blocks1
    |> Array.append blocks2
    |> Array.mapi (fun i b -> i.ToString(), b.ToString())
        
//mapReduce input map reduce (gatherer(1000)) 20 20 partitionF

type BookSplitter () =
    let blocks = new List<string * string>()
    member b.Split howManyBlocks fileName =
            let b =
                fileName
                |> splitBook howManyBlocks
                |> Array.mapi (fun i b -> i.ToString(), b.ToString())
            blocks.AddRange(b)
    member b.Blocks () =
            blocks.ToArray() :> seq<string * string>

type WordCounter () =
    let w = new WorkQueue()
    let words = new Dictionary<string,int>()
    let worker(wordCounter:WordCounter, ev:EventWaitHandle) =
          let w1 = new WorkQueue()
          { new IOutput<string, int> with
                member o.Reduced key values =
                    w1.Queue (fun() ->
                        wordCounter.AddWord key (values |> Seq.hd))
                member o.MapReduceDone () =
                    w1.Queue(fun () ->
                        ev.Set() |> ignore)
           }
    member c.AddWord word count =
            let exist, value = words.TryGetValue(word)
            if exist then
                words.[word] <- value + count
            else
                words.Add(word, count)
    member c.Add fileName =
        w.Queue (fun () ->
            let s = new BookSplitter()
            fileName |> s.Split 100
            let ev = new EventWaitHandle(false, EventResetMode.AutoReset)
            let blocks = s.Blocks ()
            mapReduce blocks map reduce (worker(c, ev)) 20 20 partitionF
            ev.WaitOne() |> ignore
            )  
    member c.Words =
        w.QueueWithAsync (fun () ->
            words |> Seq.to_array |> Array.map (fun kv -> kv.Key, kv.Value)
        )    

let wc = new WordCounter()
wc.Add (__SOURCE_DIRECTORY__ + "\kjv10.txt")
wc.Add (__SOURCE_DIRECTORY__ + "\warandpeace.txt")

let wordsToPrint = async {
                    let! words = wc.Words
                    return words
                        |> Seq.distinctBy (fun (key, _) -> key.ToLower())
                        |> Seq.filter (fun (key, _) -> not(key = "" || key = "\"" || (fst (Double.TryParse(key)))))
                        |> Seq.to_array
                        |> Array.sortBy snd
                        |> Array.rev
                        |> Seq.take 20
                        |> Seq.iter (fun (key, value) -> printfn "%A\t\t%A" key value)}


Async.RunSynchronously wordsToPrint
                        
Thread.Sleep(15000)
printfn "Closed session"

My PDC session is online - “Future directions for C# and Visual Basic”

lucabol — Mon, 23 Nov 2009 21:17:44 GMT

In case you are training to play the part of a first generation Italian immigrant in a Broadway show or if you want to understand why I never short msft stock: http://microsoftpdc.com/Sessions/FT11

Luca at PDC 2009 next week

lucabol — Fri, 13 Nov 2009 18:49:18 GMT

I’ll be in Los Angeles next week for PDC 2009. My session is called “Future Directions for C# and Visual Basic” and it comes on Tuesday immediately after the first keynote.

I’m planning on spending the first half of the session talking about the biggest trends influencing C# and VB. The second part will be about more future looking features we are playing with these days.

My deck has just two slides, as always. It’s all about watching me typing code on stage, hoping that I make crazy errors …

There are several other interesting language related sessions at PDC. Here is a list of them (please notice that the times might change). Also feel free to drop by the languages booth and chat. My team and I will be there most of the time.

Day	Time	Title	Presenter	Room
Tues 11/17	11:00 - 12:00	Future Directions for C# and Visual Basic	Luca Bolognese	Hall F
Tues 11/17	11:00 - 12:00	Accelerated Windows Application Development with Microsoft Visual C++ 2010	Boris Jabes	408B
Tues 11/17	12:30 - 13:15	Dynamic Binding in C# 4	Mads Torgersen	Hall F
Tues 11/17	12:30 - 13:15	Using Dynamic Languages to Build Scriptable Applications	Dino Viehland	403AB
Tues 11/17	13:30 - 14:30	C++ Forever: Interactive Applications in the Age of Manycore	Rick Molloy	515B
Tues 11/17	16:30 - 17:30	Manycore and the Microsoft .NET Framework 4: A Match Made in Microsoft Visual Studio 2010	Stephen Toub	502A
Tues 11/17	16:30 - 17:30	Code Contracts and Pex: Power Charge Your Assertions and Unit Tests	Mike Barnett, Nikolai Tillmann	408A
Wed 11/18	11:30 - 12:30	Microsoft Perspectives on the Future of Programming	Panel	Petree Hall C
Wed 11/18	12:00 - 13:00	F# for Architects – Hitting the Sweet Spot	Chris Smith	VS and .NET Pavillion
Wed 11/18	13:00 - 13:45	Code Like the Wind with Microsoft Visual Basic 2010	Lucian Wischik	Petree Hall D
Wed 11/18	13:00 - 13:45	Future of Garbage Collection	Patrick Dussud	Petree Hall C
Wed 11/18	13:00 - 13:45	Microsoft Project Code Name “M”: The Data and Modeling Language	Don Box, Jeff Pinkston	408A
Thu 11/19	08:30 - 09:30	PLINQ: LINQ, but Faster!	Ed Essey, Igor Ostrovsky	515A
Thu 11/19	10:00 - 11:00	A Lap around Microsoft Visual Studio 2010 for the Visual Basic Developer	Lisa Feigenbaum	VS and .NET Pavillion
Thu 11/19	10:00 - 11:00	Axum: A .NET Language for Safe and Scalable Concurrency	Niklas Gustafsson	515A
Thu 11/19	11:30 - 12:30	F# for Parallel and Asynchronous Programming	Luke Hoban	515A
Thu 11/19	12:45 - 13:30	Microsoft Visual C# IDE Tips and Tricks	DJ Park	Petree Hall D
Thu 11/19	12:45 - 13:30	Microsoft Visual Basic IDE Tips and Tricks	Dustin Campbell	Petree Hall C

Becoming really rich with C#

lucabol — Wed, 23 Sep 2009 02:40:00 GMT

Or maybe not, please do not hold me responsible if you lose money following this system. Having said that, it is my opinion that there are very few concepts that are important in investing. Three big ones are value, diversification and momentum. This post is about the latter two and how to use C# to create a simple trading system that uses both.

Diversification is ‘not put all your eggs in one basket’ (contrary to ‘put all of them in one basket and watch that basket’). I don’t believe you can ‘watch’ very much in financial markets, so I tend to prefer diversification.

Momentum is a mysterious tendency of financial prices that have risen the most in the recent past, to continue outperforming in the close future. In essence, buying the top stocks/sectors/asset classes tends to outperform buying the bottom ones over horizons from three months to one year.

The idea then is to rank some assets (i.e. ETFs) by how fast they have risen in the past, go long the top ones and short the bottom ones. There are hundreds of variations of this basic strategy, we’ll add the rule that we won’t buy assets that are below their 200 days moving average or sell short assets that are above it.

I’m writing this code with VS 2010 Beta 2 (which hasn’t shipped yet). It should be trivial to modify it to run on B1 (or maybe it does run on it already). I attach the code and data files to this post.

struct Event {
    internal Event(DateTime date, double price) { Date = date; Price = price; }
    internal readonly DateTime Date;
    internal readonly double Price;
}

We’ll use this simple structure to load the closing price for a particular date. My use of internal is kind of bizarre. Actually the whole code might look strange. It is an interesting (maybe un-elegant) mix of object orientation and functional programming.

class Summary {
    internal Summary(string ticker, string name, string assetClass,
                    string assetSubClass, double? weekly, double? fourWeeks,
                    double? threeMonths, double? sixMonths, double? oneYear,
                    double? stdDev, double price, double? mav200) {
        Ticker = ticker;
        Name = name;
        AssetClass = assetClass;
        AssetSubClass = assetSubClass;
        // Abracadabra ...
        LRS = (fourWeeks + threeMonths + sixMonths + oneYear) / 4;
        Weekly = weekly;
        FourWeeks = fourWeeks;
        ThreeMonths = threeMonths;
        SixMonths = sixMonths;
        OneYear = oneYear;
        StdDev = stdDev;
        Mav200 = mav200;
        Price = price;
    }
    internal readonly string Ticker;
    internal readonly string Name;
    internal readonly string AssetClass;
    internal readonly string AssetSubClass;
    internal readonly double? LRS;
    internal readonly double? Weekly;
    internal readonly double? FourWeeks;
    internal readonly double? ThreeMonths;
    internal readonly double? SixMonths;
    internal readonly double? OneYear;
    internal readonly double? StdDev;
    internal readonly double? Mav200;
    internal double Price;

    internal static void Banner() {
        Console.Write("{0,-6}", "Ticker");
        Console.Write("{0,-50}", "Name");
        Console.Write("{0,-12}", "Asset Class");
        //Console.Write("{0,-30}\t", "Asset SubClass";
        Console.Write("{0,4}", "RS");
        Console.Write("{0,4}", "1Wk");
        Console.Write("{0,4}", "4Wk");
        Console.Write("{0,4}", "3Ms");
        Console.Write("{0,4}", "6Ms");
        Console.Write("{0,4}", "1Yr");
        Console.Write("{0,6}", "Vol");
        Console.WriteLine("{0,2}", "Mv");
        //Console.Write("{0,6}", "Pr");
        //Console.WriteLine("{0,6}", "M200");
    }

    internal void Print() {

        Console.Write("{0,-6}", Ticker);
        Console.Write("{0,-50}", new String(Name.Take(48).ToArray()));
        Console.Write("{0,-12}", new String(AssetClass.Take(10).ToArray()));
        //Console.Write("{0,-30}\t", new String(AssetSubClass.Take(28).ToArray()));
        Console.Write("{0,4:N0}", LRS * 100);
        Console.Write("{0,4:N0}", Weekly * 100);
        Console.Write("{0,4:N0}", FourWeeks * 100);
        Console.Write("{0,4:N0}", ThreeMonths * 100);
        Console.Write("{0,4:N0}", SixMonths * 100);
        Console.Write("{0,4:N0}", OneYear * 100);
        Console.Write("{0,6:N0}", StdDev * 100);
        if (Price <= Mav200)
            Console.WriteLine("{0,2}", "X");
        else
            Console.WriteLine();
        //Console.Write("{0,6:N2}", Price);
        //Console.WriteLine("{0,6:N2}", Mav200);
    }
}

The class Summary above is how I want to present my results. A few comments on the code. I use Nullable<T> because some of this values can be null (i.e. not enough history), but I still don’t want to worry about it. It ends up working rather neatly.

I also print the results out to Console, which is crazy. I really should be using WPF/Silverlight as the presentation layer. Also the {0,4:N0} notation might be unfamiliar to some of you, but this is how mad Console guys like myself avoid using real UI frameworks. Sometimes we print things in color too.

The real meat is in the following line:

LRS = (fourWeeks + threeMonths + sixMonths + oneYear) / 4;

That is our highway to richness. It’s a very elaborated quant formula, never before shown, that calculate a magick relative strength (aka momentum) factor as the average of the performance of four weeks, three months, six months and one year.

class TimeSeries {
    internal readonly string Ticker;
    readonly DateTime _start;
    readonly Dictionary<DateTime, double> _adjDictionary;
    readonly string _name;
    readonly string _assetClass;
    readonly string _assetSubClass;

    internal TimeSeries(string ticker, string name, string assetClass, string assetSubClass, 
                                                                IEnumerable<Event> events) {
        Ticker = ticker;
        _name = name;
        _assetClass = assetClass;
        _assetSubClass = assetSubClass;
        _start = events.Last().Date;
        _adjDictionary = events.ToDictionary(e => e.Date, e => e.Price);
    }

I then built myself a little TimeSeries class that represents a series of (date, price). I choose a dictionary to store it because of my assumption that I will be accessing it by date a lot. In retrospect, I was kind of right and kind of wrong. It doesn’t really matter much.

bool GetPrice(DateTime when, out double price, out double shift) {
    // To nullify the effect of hours/min/sec/millisec being different from 0
    when = new DateTime(when.Year, when.Month, when.Day);
    var found = false;
    shift = 1;
    double aPrice = 0;
    while (when >= _start && !found) {
        if (_adjDictionary.TryGetValue(when, out aPrice)) {
            found = true;
        }
        when = when.AddDays(-1);
        shift -= 1;
    }
    price = aPrice;
    return found;
}

A TimeSeries can give you back the price at a particular date. This looks bizarre and complex, but there is a reason for it. I might ask for a date that doesn’t have a price associated with it (i.e. holidays, week-ends). In such cases I want to return the previous price which could be N days in the past.

I also want to return how many days in the past I had to go, so that other calculations (i.e. Return) can modify their end date by the same amount. Also I might not find such a price at all, in which case I don’t want to throw an exception, but instead notify the caller. In retrospect, I should have used double? to signify ‘price not found’.

double? GetReturn(DateTime start, DateTime end) {
    var startPrice = 0.0;
    var endPrice = 0.0;
    var shift = 0.0;
    var foundEnd = GetPrice(end, out endPrice, out shift);
    var foundStart = GetPrice(start.AddDays(shift), out startPrice, out shift);
    if (!foundStart || !foundEnd)
        return null;
    else
        return endPrice / startPrice - 1;
}

We can now go and calculate the return between two dates. Also the TimeSeries object needs to perform a little more calculations.

    internal double? LastWeekReturn() {
        return GetReturn(DateTime.Now.AddDays(-7), DateTime.Now);
    }
    internal double? Last4WeeksReturn() {
        return GetReturn(DateTime.Now.AddDays(-28), DateTime.Now);
    }
    internal double? Last3MonthsReturn() {
        return GetReturn(DateTime.Now.AddMonths(-3), DateTime.Now);
    }
    internal double? Last6MonthsReturn() {
        return GetReturn(DateTime.Now.AddMonths(-6), DateTime.Now);
    }
    internal double? LastYearReturn() {
        return GetReturn(DateTime.Now.AddYears(-1), DateTime.Now);
    }
    internal double? StdDev() {
        var now = DateTime.Now;
        now = new DateTime(now.Year, now.Month, now.Day);
        var limit = now.AddYears(-3);
        var rets = new List<double>();
        while (now >= _start.AddDays(12) && now >= limit) {
            var ret = GetReturn(now.AddDays(-7), now);
            rets.Add(ret.Value);
            now = now.AddDays(-7);
        }
        var mean = rets.Average();
        var variance = rets.Select(r => Math.Pow(r - mean, 2)).Sum();
        var weeklyStdDev = Math.Sqrt(variance / rets.Count);
        return weeklyStdDev * Math.Sqrt(40);
    }
    internal double? MAV200() {
        return _adjDictionary
               .ToList()
               .OrderByDescending(k => k.Key)
               .Take(200)
               .Average(k => k.Value);
    }
    internal double TodayPrice() {
        var price = 0.0;
        var shift = 0.0;
        GetPrice(DateTime.Now, out price, out shift);
        return price;
    }
    internal Summary GetSummary() {
        return new Summary(Ticker, _name, _assetClass, _assetSubClass, 
                           LastWeekReturn(), Last4WeeksReturn(), Last3MonthsReturn(),
                           Last6MonthsReturn(), LastYearReturn(), StdDev(), TodayPrice(),
                           MAV200());
    }
}

Nothing particularly interesting in this code. Just a bunch of calculations. The MAV200 is the 200 days moving average of closing prices. It shows a more functional way of doing things. The StdDev function is instead very imperative.

We now can work on downloading the prices. This is how you construct the right URL:

static string CreateUrl(string ticker, DateTime start, DateTime end) {
    return @"http://ichart.finance.yahoo.com/table.csv?s=" + ticker + "&a="
            + (start.Month - 1).ToString() + "&b=" + start.Day.ToString() + "&c="
            + start.Year.ToString() + "&d=" + (end.Month - 1).ToString() + "&e="
            + end.Day.ToString() + "&f=" + end.Year.ToString() + "&g=d&ignore=.csv";
}

And let’s set how many concurrent connections we are going to use …

ServicePointManager.DefaultConnectionLimit = 10;

On my machine, setting this number too high causes errors to be returned. I’m not sure on which side of the connection the problem lies.

We can then load all the tickers we want to load from a file. One of the files has Leveraged ETFs, which I want to filter out because they tend to pop up always at the top.

var tickers =
    //File.ReadAllLines("ETFs.csv")
    //File.ReadAllLines("ETFTest.csv")
    File.ReadAllLines("AssetClasses.csv")
    .Skip(1)
    .Select(l => l.Split(new[] { ',' }))
    .Where(v => v[2] != "Leveraged")
    .Select(values => Tuple.Create(values[0], values[1], values[2], values[3]))
    .ToArray();
var len = tickers.Length;

var start = DateTime.Now.AddYears(-2);
var end = DateTime.Now;
var cevent = new CountdownEvent(len);
var summaries = new Summary[len];

And then load all of them, making sure to make an asynchronous call so not to keep the thread busy.

for(var i = 0; i < len; i++)  {
    var t = tickers[i];
    var url = CreateUrl(t.Item1, start, end);
    using (var webClient = new WebClient()) {
        webClient.DownloadStringCompleted +=
                          new DownloadStringCompletedEventHandler(downloadStringCompleted);
        webClient.DownloadStringAsync(new Uri(url), Tuple.Create(t, cevent, summaries, i));
    }
}

cevent.Wait();

Notice the use of a Countdown event to wait for all the thread to complete before printing out the results. Also notice the new Tuple<T> class used to package things to send around.

We can then print out the top and bottom 15%:

var top15perc =
        summaries
        .Where(s => s.LRS.HasValue)
        .OrderByDescending(s => s.LRS)
        .Take((int)(len * 0.15));
var bottom15perc =
        summaries
        .Where(s => s.LRS.HasValue)
        .OrderBy(s => s.LRS)
        .Take((int)(len * 0.15));

Console.WriteLine();
Summary.Banner();
Console.WriteLine("TOP 15%");
foreach(var s in top15perc)
    s.Print();

Console.WriteLine();
Console.WriteLine("Bottom 15%");
foreach (var s in bottom15perc)
    s.Print();

Here is what we do when a request comes back with data:

static void downloadStringCompleted(object sender, DownloadStringCompletedEventArgs e) {
    var bigTuple =
             (Tuple<Tuple<string, string, string, string>, CountdownEvent, Summary[], int>)
              e.UserState;
    var tuple = bigTuple.Item1;
    var cevent = bigTuple.Item2;
    var summaries = bigTuple.Item3;
    var i = bigTuple.Item4;
    var ticker = tuple.Item1;
    var name = tuple.Item2;
    var asset = tuple.Item3;
    var subAsset = tuple.Item4;

    if (e.Error == null) {
        var adjustedPrices =
                e.Result
                .Split(new[] { '\n' })
                .Skip(1)
                .Select(l => l.Split(new[] { ',' }))
                .Where(l => l.Length == 7)
                .Select(v => new Event(DateTime.Parse(v[0]), Double.Parse(v[6])));

        var timeSeries = new TimeSeries(ticker, name, asset, subAsset, adjustedPrices);
        summaries[i] = timeSeries.GetSummary();
        cevent.Signal();
        Console.Write("{0} ", ticker);
    }
    else {
        Console.WriteLine("[{0} ERROR] ", ticker);
        //Console.WriteLine(e.Error);
        summaries[i] = new Summary(ticker, name, "ERROR", "ERROR", 0, 0, 0, 0, 0, 0,0,0); 
        cevent.Signal();
    }
}

We first unpack the Tuple we sent out originally, we then extract the Date and Price, create a Summary object and store it in the summaries array. It’s important to remember to Signal to the cevent in the error case as well because we want to print out the results even if some downloading failed.

And here is what you get for your effort: