Timeseries forecast with irregular input streams

[dberardo-com]

Forecasting time series with regularly spaced input vectors (timedelta between samples is always a constant, say 1 hour) is nice and good, but what happens when one has very irregular samples? E.g. first 10 points come in within the first 2 minutes, then 10 minutes gap, then another 3 samples, and so on.

Linking related comments:

• Problem with ThresholdFilter and GaussianScorer #948 (comment)
• Improvements on GaussianScorer #949 (comment)

Link related discussions:

• How to use ensembles with anomalies detectors #978

[dberardo-com]

@emreguenduez you might be interested in watching this

[bahia14]

@dberardo-com I did try to use river for time series, like stock prices, however; I could not find any example.
Could you please help me, show me any example ? I usually get stock quotes from yahoo finance.

Thank you so much

[dberardo-com]

bom dia @bahia14

although i am not a maintainer of this project i would be glad to help. I am currently doing some infrastructure work related to streaming processing. Once that work is finalized i will get my hands back to river. My current concern is to figure out a way to implement the thumbling window discussed above in this thread. Once thats over, i would spend some time improving the forecast accuracy of SNARIMAX.

What is on your roadmap @bahia14 ?

[MaxHalford]

The short term answer is we won't support this for a long time. I think it's already a hefty challenge us to support to regular spaced time series. If you really want support for this, then I would suggest looking at other libraries which cater to time series. There are many good ones available that will likely satisfy your needs :)

As for what we could do, maybe the simplest approach would be to implement an aggregation mechanism. It takes as input a stream, and in return yields one element per day/hour/whatever. But this is a big topic and not something I'm inclined to personally dive into right now.

[MaxHalford]

Right! Tumbling windows sounds exactly like what is needed.

[dberardo-com]

@MaxHalford i get back after some time ... i had to implement a version of thumbling window for another use case i have at hand and i thought of sharing the implementation here to collect some feedback. This window function is currently being tested with the running Mean statistic and a time period (DeltaT) of 3600 seconds (1 hour mean values):

from datetime import datetime, timedelta
from river import stats
from .global_vars import DEBUG_FUNC

AGGREGATORS = {"mean": stats.Mean}

def createObj(diction=dict()):
    return type('obj', (object,), diction)

 """Thumbling Window
    Attributes
    ----------
    id: string
        This string will be used as the chache-key to store the state of this window between calls. It is a mandatory field, and cannot be defaulted, as only the user knows which id can be used as unique identifier for this window
    v: number
        the new value to be stored in the aggregator / window. The current implementation of the window is based on process-time and not event-time, so v needs to be a number only (not (t,v) tuples)
    dt: int
        Window's period, in seconds. New window will be opened at t_opening time. t_opening will always be a multiple of dt (e.g. 00:00,00:10,00:20 for dt=10).  Window's aggregator will be updated until (t_opening + dt). If a value is received between [(t_opening + dt);(t_opening + dt + grace)] a new window is open and the last value of the aggregator is returned
    f: string
        The name of the aggregator to be used. f should exist in AGGREGATORS.keys()
    grace: int
        grace period of the window. The window will only accept values until it is closed, or until (dt + grace) seconds have elapsed. An open window will close as soon as the one value between [dt;dt+grace] seconds is received. (dt+grace) is to be seen as the de-facto TTL value of the underlying cache object (see below).
        By default the grace period is equal dt, so that the window will expire after 2*dt seconds.
    cache: object
        cache can be any object capable of storing keys and values. In theory also a dictionary could do the job, but we recommend using a TLRUCache object from the cachetools package
    
    Examples
    ---------
    dt=10
    assert window("uid", 200, dt) == None # open new 10s window.
    time.sleep(dt/2)
    assert window("uid", 100, dt) == None # update the window with a new value
    time.sleep(dt/2 + 1)
    assert window("uid", 200, dt) == 150 # closes the old window and stores the new value in a new window
    time.sleep(2*dt + 1) # wait until window expires
    assert window("uid", 200, dt) == None # because the previous window has silently expired, this will open a new window without returning any value
    """
def window(id, v, dt, f="mean", grace = 0, cache=None):
    dt = int(dt)
    if not grace: grace = dt
    now = int(datetime.now().timestamp() * 1000)
    if DEBUG_FUNC>2: print("[DEBUG]: Window ",id," received params: ",v, dt, f, id)
    if not v: return None
    try: v=int(v)
    except ValueError: v=float(v)
    
    old_dict = cache.get(id) if id in cache.keys() else None
    expiration_interval = (dt + grace) * 1000
    
    remainder = now % (dt * 1000)
    
    noneValue = None
    
    if not old_dict:
        # open window using process time. including nnew running stat
        aggr = AGGREGATORS[f]()
        aggr.update(v)
        new_opening = now - remainder # this is to make sure that t_opening is always a multiple of dt
        if DEBUG_FUNC: print("[DEBUG]: Window ",id," opening with value", v, "on time", datetime.fromtimestamp(new_opening/1000).strftime('%Y-%m-%d %H:%M:%S.%f'), "expiring in", dt*1000)
        cache.__setitem__(id,createObj(dict(ttl=expiration_interval / 1000, aggr=aggr, t_opening = new_opening)))
        return noneValue
    elif (now - old_dict.t_opening) <  timedelta(seconds=dt).seconds * 1000 :
        # update window
        elapsed_time = now - old_dict.t_opening
        new_value = old_dict.aggr.update(v).get()
        old_dict.ttl = dt + grace -  elapsed_time / 1000 
        cache.__setitem__(id, old_dict)
        if DEBUG_FUNC>1: print("[DEBUG]: Window ",id," updating: ", new_value, "elapsed", elapsed_time , "expiring in", old_dict.ttl - grace, "listening until", old_dict.ttl)
        return noneValue
    else:
        # close window and open new one
        aggr_value = old_dict.aggr.get()
        catchup = (now - old_dict.t_opening) % (dt * 1000)
        ttl = expiration_interval - catchup - remainder # this is to make sure that t_opening is always a multiple of dt
        new_opening = now - catchup
        if DEBUG_FUNC: print("[DEBUG]: Closing window ",id,"on value",old_dict.aggr.get(),"expired since", catchup, "@" , datetime.fromtimestamp(new_opening/1000).strftime('%Y-%m-%d %H:%M:%S.%f'))
        if DEBUG_FUNC: print("[DEBUG]: Opening new window ",id," with value", v, "on time", datetime.fromtimestamp(new_opening/1000).strftime('%Y-%m-%d %H:%M:%S.%f'), "expiring in", ttl - grace * 1000)
        aggr = AGGREGATORS[f]()
        aggr.update(v)
        cache.__setitem__(id,createObj(dict(ttl=ttl / 1000, aggr=aggr , t_opening = new_opening ) ) ) 
        return aggr_value

[dberardo-com]

needless to say, the whole function should be restructured to be used in river, but in the use case i had i could only use function calls and not objects, that's why i had to call the function with the same arguments all over again.

I am using cachetools as state-store for the window. i assume that this is also compliant with the libraries that river uses @MaxHalford

[MaxHalford]

Thanks a lot for sharing. It could be interesting to have this in River.

[dberardo-com]

let me know if i should try to setup a PR, in which case i would need to know at least:

• which package / directory to use
• whether we have already an object to use as base
• whether i could import directly from cachetools

also, this implementation cannot be currently extended to a sliding window, and i think it could never be. Also, it currently suffers the same limitations as explained in here: confluentinc/ksql#9760 (comment) but again, i think this is also in line with the event-driven nature of river (so no interest in implementing timers).

[MaxHalford]

By the way, I suggest you check out this article if you haven't already :)