title: Rx.Net - Get stock price changes and process them
body: <p>The problem I'm trying to solve</p>
  <ol>
  <li>Get stock ticks</li>
  <li>Always consider <em>latest</em> stock price</li>
  <li>Each <em><strong>x</strong></em> second take a <em>snapshot</em> of ticks and send for processing</li>
  </ol>
  <p>So I have an <code>Observable</code> source of stock ticks. It sends only the ticks for stocks I'm interested in. What I need to do is to receive these stock prices, and after each <strong>x</strong> seconds (for the sake of example let's say every 3 seconds) send a <em>snapshot</em> of prices for processing. If within 3 seconds I receive 2 ticks for the same stock, I only need the latest tick. This processing is compute heavy, so if possible I would like to avoid sending same stock price for processing twice.</p>
  <p>To bring an example.</p>
  <p>Let's say at the beginning of sequence I received 2 ticks -&gt; <strong>MSFT:1$, GOOG:2$</strong>.</p>
  <p>In the next 3 seconds I receive nothing, so <strong>MSFT &amp; GOOG</strong> ticks should be sent for processing.</p>
  <p>Now the next second I receive new ticks -&gt; <strong>MSFT:1$, GOOG:3$, INTL:3$</strong></p>
  <p>Again let's assume within next 3 seconds nothing comes in.</p>
  <p>Here, since <strong>MSFT</strong> price didn't change (it's still 1$), only <strong>GOOG &amp; INTL</strong> should be sent for processing.</p>
  <p>And this repeats throughout a day.</p>
  <p>Now I think Rx helps to solve this kind of problems in easy &amp; elegant way. But I'm having a problem to have the proper queries.
  This is what I have so far, will try to explain what it does and what's the issue with it</p>
  <pre><code>var finalQuery =
                 from priceUpdate in **Observable&lt;StockTick&gt;**
                 group priceUpdate by priceUpdate.Stock into grouped
                 from combined in Observable.Interval(TimeSpan.FromSeconds(3))
                        .CombineLatest(grouped, (t, pei) =&gt; new { PEI = pei, Interval = t })
                 group combined by new { combined.Interval } into combined
                 select new
                 {
                     Interval = combined.Key.Interval,
                     PEI = combined.Select(c =&gt; new StockTick(c.PEI.Stock, c.PEI.Price))
                 };
  
              finalQuery
                  .SelectMany(combined =&gt; combined.PEI)
                  .Distinct(pu =&gt; new { pu.Stock, pu.Price })
                  .Subscribe(priceUpdate =&gt;
                  {
                      Process(priceUpdate);
                  });
  
  public class StockTick
  {
     public StockTick(string stock, decimal price)
     {    
        Stock = stock;
        Price = price;
     }
     public string Stock {get;set;}
     public decimal Price {get;set;}
  }
  </code></pre>
  <p>So this gets the stock price, groups it by stock, then combines latest from this grouped sequence with <code>Observable.Interval</code>. This way I'm trying to ensure only latest ticks for a stock are processed and it fires up every 3 seconds.</p>
  <p>Then again it groups it up by interval this time, as a result I have group of sequences for each 3 second intervals that passed.</p>
  <p>And as a last step, I flatten this sequence to sequence of stock price updates using <code>SelectMany</code> and also I'm applying <code>Distinct</code> to ensure same price for the same stock is not processed twice.</p>
  <p>There are 2 issues with this query I don't like. First is I don't really like double group by's - is there any way to avoid it ? Second - with this approach I have to process prices one by one, what I really would like to have is <em>snapshots</em> - that is within 3 seconds whatever I have I will bundle and send for processing, but can't figure out how to <em>bundle</em>.</p>
  <p>I'll be happy for suggestions to solve this problem other way, but I would prefer to stay within Rx, unless there is really something much much better.</p>