Rx.Net - 获取股票价格变化并处理它们

Question

我要解决的问题

1. 获取股票报价
2. 始终考虑最新股票价格
3. 每 x 秒拍摄一个快照的刻度并发送处理

所以我有一个Observable 的股票报价来源。它只发送我感兴趣的股票的报价。我需要做的是接收这些股票价格，并在每 x 秒后（例如，假设每 3 秒）发送处理价格的快照。如果在 3 秒内我收到同一股票的 2 个报价，我只需要最新的报价。这个处理计算量很大，所以如果可能的话，我希望避免发送相同的股票价格进行两次处理。

举个例子。

假设在序列开始时我收到了 2 个滴答声 -> MSFT:1$, GOOG:2$。

在接下来的 3 秒内，我什么也没有收到，所以应该发送 MSFT 和 GOOG 报价以进行处理。

现在下一秒我收到新的报价 -> MSFT:1$, GOOG:3$, INTL:3$

再次假设在接下来的 3 秒内没有任何东西进入。

这里，由于 MSFT 价格没有变化（仍然是 1 美元），因此应该只发送 GOOG & INTL 进行处理。

这会在一天中重复。

现在我认为 Rx 有助于以简单而优雅的方式解决这类问题。但是我在进行适当的查询时遇到了问题。 这是我到目前为止所拥有的，将尝试解释它的作用以及它的问题所在

var finalQuery =
               from priceUpdate in **Observable<StockTick>**
               group priceUpdate by priceUpdate.Stock into grouped
               from combined in Observable.Interval(TimeSpan.FromSeconds(3))
                      .CombineLatest(grouped, (t, pei) => new { PEI = pei, Interval = t })
               group combined by new { combined.Interval } into combined
               select new
               {
                   Interval = combined.Key.Interval,
                   PEI = combined.Select(c => new StockTick(c.PEI.Stock, c.PEI.Price))
               };

            finalQuery
                .SelectMany(combined => combined.PEI)
                .Distinct(pu => new { pu.Stock, pu.Price })
                .Subscribe(priceUpdate =>
                {
                    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;}
}

所以这会获取股票价格，按股票分组，然后将此分组序列中的最新数据与Observable.Interval 组合。通过这种方式，我试图确保只处理股票的最新报价，并且每 3 秒触发一次。

然后它再次按间隔将其分组，因此我每经过 3 秒的间隔就有一组序列。

最后一步，我使用SelectMany 将此序列扁平化为股票价格更新序列，并且我正在应用Distinct 以确保同一股票的相同价格不会被处理两次。

我不喜欢这个查询有 2 个问题。首先是我不太喜欢双重分组 - 有什么办法可以避免吗？其次 - 使用这种方法，我必须一个一个地处理价格，我真正想要的是 snapshots - 即在 3 秒内我将捆绑并发送处理的任何内容，但不能t 弄清楚如何捆绑。

我很乐意以其他方式解决此问题的建议，但我更愿意留在 Rx 中，除非真的有更好的东西。

Answer 1

几件事：

1. 您需要利用 Sample 运算符：
2. 您可能需要DistinctUntilChanged 而不是Distinct。如果您使用Distinct，那么如果 MSFT 从 1 美元涨到 2 美元，然后又回到 1 美元，您将不会在第三次交易时收到事件。

我想您的解决方案将如下所示：

IObservable<StockTick> source;
source
    .GroupBy(st => st.Stock)
    .Select(stockObservable => stockObservable
        .Sample(TimeSpan.FromSeconds(3))
        .DistinctUntilChanged(st => st.Price)
    )
    .Merge()
    .Subscribe(st => Process(st));

EDIT（Distinct 性能问题）：

每个Distinct 操作员都必须在其中维护完整的不同历史记录。如果你有一只高价股票，例如 AMZN，到目前为止它的价格从 958 美元到 974 美元不等，那么你最终可能会得到很多数据。在您取消订阅Distinct 之前，大约有 1600 个可能的数据点必须保存在内存中。它最终也会降低性能，因为每个 AMZN 滴答都必须在通过之前与 1600 个现有数据点进行比较。如果这是一个长期运行的过程（跨越多个交易日），那么您最终会得到更多的数据点。

给定 N 个股票，您有 N 个需要相应操作的 Distinct 操作员。将这种行为乘以 N 个股票，就会遇到越来越多的问题。