How to resample data with Reactive Extensions

user_4125331 0
Viewed 38

I have a large database of of 1minute OHLCV data. I'm trying to aggregate this data, into 1 hour data. The main Idea is to, in the future, replace this offline data source with some live data source, so the aggregation in the event would help us a lot.

I'm struggling to do it with Reactive Extensions. So far i have the following:

source.ListBars(coins)
    .ToObservable()
    .Select(t =>
    {
         var date = t.Date - (t.Date - timeframe);
         return new
         {
             Date = date,
             t.Open,
             t.High,
             t.Low,
             t.Close,
             t.Volume
         };
     })
    .GroupByUntil(t => t.Date, t => t.DistinctUntilChanged(u => u.Date))
    .Select(t => t.ToEnumerable().ToList())
    .Select(t => new
    {
        t[0].Date,
        t[0].Open,
        High = t.Max(x => x.High),
        Low = t.Min(x => x.Low),
        t[^1].Close,
        Volume = t.Sum(x => x.Volume)
    })
    .Subscribe(bar => Log.Information("D:{date:dd/MM/yyyy HH:mm:ss zzz}, O:{open:f4}, H:{high:f4}, L:{low:f4}, C:{close:f4}, V:{volume:f4}",
    bar.Date.ToDateTime(), bar.Open, bar.High, bar.Low, bar.Close, bar.Volume));

But, it's not working. Any of you know how to accomplish this?

thank you in advance!

reactive system.reactive c#
1 Answers

You could try replacing this:

.GroupByUntil(t => t.Date, t => t.DistinctUntilChanged(u => u.Date))
.Select(t => t.ToEnumerable().ToList())
.Select(t => new
{
    t[0].Date,
    //...

...with this:

.Publish(published => published
    .SkipLast(1)
    .Buffer(published.Select(t => TrimToHour(t.Date)).DistinctUntilChanged())
)
.Where(t => t.Count > 0)
.Select(t => new
{
    Date = TrimToHour(t[0].Date),
    //...

...where TrimToHour is a method that preserves only the Date and Hour components of a DateTime:

static DateTime TrimToHour(DateTime dateTime)
{
    return dateTime.Date.AddHours(dateTime.Hour);
}

The SkipLast(1) has the intention of creating a "lag", so that the last entry of an hour is not aggregated with the entries of the next hour.

user_111785490
Related