Order Book

Suppose there is a stream called Bid of bids of stocks (the example is kept simple by not considering the offer side), with records of the type:

[integer Id; | string Symbol; float Price; integer Shares; ]

where Id is the key field, the field that uniquely identifies a bid. Bids can be changed, so not only might the stream insert a new bid, but also update or delete a previous bid.

The goal is to output the top three highest bids any time a bid is inserted or changed for a particular stock. The type of the output where Position ranges from 1 to 3 is:

 [integer Position; | string Symbol; float Price; integer Shares; ]

For example, suppose the Bids have been:

[Id=1; | Symbol='IBM'; Price=43.11; Shares=1000; ]
[Id=2; | Symbol='IBM'; Price=43.17; Shares=900]
[Id=3; | Symbol='IBM'; Price=42.66; Shares=800]
[Id=4; | Symbol='IBM'; Price=45.81; Shares=50]

With the next event:

[Id=5; | Symbol='IBM'; Price=46.41; Shares=75]

The stream should output the records

[Position=1; Symbol='IBM'; | Price=46.41; Shares=75]
[Position=2; Symbol='IBM'; | Price=45.81; Shares=50]
[Position=3; Symbol='IBM'; | Price=43.17; Shares=900]

One way to solve this problem is with an event cache that groups by stock and orders the events by price:

eventCache(Bids[Symbol], coalesce, Price desc) previous;

The following code outputs the current block of the order book, down to the level specified by the depth variable.

{
  integer i := 0;
  string symbol := Bids.Symbol;
  while ((i < count(previous.Id)) and (i < depth) ) {
    output setOpcode([ Position=i; Symbol = symbol; |
                       Price=nth(i,previous.Price); 
                       Shares=nth(i,previous.Shares); 
                       ], upsert); 
    i++;
  }
  while (i < depth) {
    output setOpcode([ Position=i; Symbol=symbol ], safedelete);
    i++;
  }
}