0

我正在尝试创建一个分析股票价格的工具。

我有不同股票的价格数据流，并且我希望有一个可观察的，以便在它收到一组新的、不同的和完整的价格时发出事件。

我的计划：将流分组为不同股票的不同子流，并重新组合它们的最新值。

假设我有这样的事件流：

from rx import Observable

stock_events = [
    {'stock': 'A', 'price': 15},
    {'stock': 'A', 'price': 16},
    {'stock': 'B', 'price': 24},
    {'stock': 'C', 'price': 37},
    {'stock': 'A', 'price': 18},
    {'stock': 'D', 'price': 42},
    {'stock': 'B', 'price': 27},
    {'stock': 'B', 'price': 27},
    {'stock': 'C', 'price': 31},
    {'stock': 'D', 'price': 44}
]

price_source = Observable.from_list(stock_events)

这是我的第一个（天真的）方法：

a_source = price_source.filter(lambda x: x['stock'] == 'A').distinct_until_changed()
b_source = price_source.filter(lambda x: x['stock'] == 'B').distinct_until_changed()
c_source = price_source.filter(lambda x: x['stock'] == 'C').distinct_until_changed()
d_source = price_source.filter(lambda x: x['stock'] == 'D').distinct_until_changed()

(Observable
    .combine_latest(a_source, b_source, c_source, d_source, lambda *x: x)
    .subscribe(print))

这正确地给了我：

({'stock': 'A', 'price': 18}, {'stock': 'B', 'price': 24}, {'stock': 'C', 'price': 37}, {'stock': 'D', 'price': 42})
({'stock': 'A', 'price': 18}, {'stock': 'B', 'price': 27}, {'stock': 'C', 'price': 37}, {'stock': 'D', 'price': 42})
({'stock': 'A', 'price': 18}, {'stock': 'B', 'price': 27}, {'stock': 'C', 'price': 31}, {'stock': 'D', 'price': 42})
({'stock': 'A', 'price': 18}, {'stock': 'B', 'price': 27}, {'stock': 'C', 'price': 31}, {'stock': 'D', 'price': 44})

然而，我觉得这应该更好地处理group_by，而不是几个过滤，所以这里是一个重写：

(price_source
 .group_by(lambda e: e['stock'])
 .map(lambda obs: obs.distinct_until_changed())
 .combine_latest(lambda *x: x)
 .subscribe(print))

但这一次，我得到：

(<rx.core.anonymousobservable.AnonymousObservable object at 0x000000000105EA20>,)
(<rx.core.anonymousobservable.AnonymousObservable object at 0x000000000776AB00>,)
(<rx.core.anonymousobservable.AnonymousObservable object at 0x000000000776A438>,)
(<rx.core.anonymousobservable.AnonymousObservable object at 0x000000000775E7F0>,)

我在这里错过了什么？我如何“解开”嵌套的 observables？

4

1 回答 1

0

如果您确实想使用 groupby，那么它在 C# 中将类似于下面的内容。但是，这不符合您对“完整”集的要求。根据评论，怀疑 CombineLatest 在这里会更好。

price_source.GroupBy(x => x.Stock)
            .Select(gp => gp.DistinctUntilChanged(x => x.Price))
            .SelectMany(x => x)
            .Subscribe(s => Console.WriteLine($"{s.Stock} : {s.Price}"));
于 2017-05-08T15:27:18.133 回答