OHLCV data compaction in redis timeseries

Question

I am using a websocket connection which streams stock tick data and stores them into redis. I want to compact this into 1 min candlestick.

Below is how i setup redis. Here I have 'NIFTY' for raw tick data and 'NIFTYV' for tick volume data. I created O, H, L, C, V timeseries to store the compacted data.

When running the code, i can see tick price and volume is being added to NIFTY and NIFTYV timeseries. But the compaction is not happening.

r = redis.Redis(db=0)
ts = r.ts()
inst = 'NIFTY'

ts.create(inst, labels={'name':inst})
ts.create(inst + 'V', labels={'name':inst})

ts.create('O', labels={'name':inst, 'type':'O'})
ts.create('H', labels={'name':inst, 'type':'H'})
ts.create('L', labels={'name':inst, 'type':'L'})
ts.create('C', labels={'name':inst, 'type':'C'})
ts.create('V', labels={'name':inst, 'type':'V'})

ts.createrule(inst, 'O', 'first', bucket_size_msec=60000)
ts.createrule(inst, 'H', 'max', bucket_size_msec=60000)
ts.createrule(inst, 'L', 'min', bucket_size_msec=60000)
ts.createrule(inst, 'C', 'last', bucket_size_msec=60000)
ts.createrule(inst + 'V', 'V', 'sum', bucket_size_msec=60000)

This is how i add the tick and volume data. Timestamp is one provided by the websocket.

According to documentation setting timestamp to * will use the redis server time instead.

timestamp = msg['timestamp']

ts.add(inst, timestamp, tick['price'])
ts.add(inst + 'V', timestamp, tick['qty'])

I am assuming redis automatically compacts the data at 1 min intervals or do i need to manually trigger this?

Answer 1

RedisTimeSeries has, by design, no concept of time progression. It does not maintain an internal clock. Compaction buckets are 'closed' based solely on the difference between their timestamp and the most recent timestamp passed to TS.ADD, TS.MADD, TS.INCRBY, and TS.DECRBY.

A bucket is 'closed' and compacted only upon arrival of a new sample that 'opens' a 'new latest' bucket.

Starting with RedisTimeSeries 1.8, you can use the LATEST flag with TS.GET, TS.MGET, TS.RANGE, TS.REVRANGE, TS.MRANGE and TS.MREVRANGE to also retrieve the latest 'open' bucket of compacted time series.