(acc, new, diff=None, window=None, agg=None, grouper=None, with_state=False)
return acc2, result
An accumulation binary operator for windowed groupb-aggregations

    This is the function that is actually given to the ``Stream.accumulate``
    function.

    Parameters
    ----------
    acc: state
    new: DataFrame, Series
        The new data to add to the window.
    diff: callable
        One of ``diff_iloc`` or ``diff_loc``
    window: int, value
        Either an integer for ``n=...`` for a value like ``value='2h'``
    agg: Aggregation
        The aggregation object to apply, like ``Sum()``
    grouper: key or Frame
        Either a column like ``'x'`` or a Pandas Series if the groupby was
        given a streaming frame.

    Returns
    -------
    acc: state
    result: newly emitted result

    See Also
    --------
    accumulator
    windowed_accumulator
An accumulation binary operator for windowed groupb-aggregations
An accumulation binary operator for windowed groupb - aggregations
def windowed_groupby_accumulator(acc, new, diff=None, window=None, agg=None, grouper=None, with_state=False):
    """ An accumulation binary operator for windowed groupb-aggregations

    This is the function that is actually given to the ``Stream.accumulate``
    function.

    Parameters
    ----------
    acc: state
    new: DataFrame, Series
        The new data to add to the window.
    diff: callable
        One of ``diff_iloc`` or ``diff_loc``
    window: int, value
        Either an integer for ``n=...`` for a value like ``value='2h'``
    agg: Aggregation
        The aggregation object to apply, like ``Sum()``
    grouper: key or Frame
        Either a column like ``'x'`` or a Pandas Series if the groupby was
        given a streaming frame.

    Returns
    -------
    acc: state
    result: newly emitted result

    See Also
    --------
    accumulator
    windowed_accumulator
    """
    if agg.grouper is None and isinstance(new, tuple):
        new, grouper = new
    else:
        grouper = None

    size = GroupbySize(agg.columns, agg.grouper)

    if acc is None:
        acc = {'dfs': [],
               'state': agg.initial(new, grouper=grouper),
               'size-state': size.initial(new, grouper=grouper)}
        if isinstance(grouper, np.ndarray) or is_series_like(grouper) or is_index_like(grouper):
            acc['groupers'] = deque([])

    dfs = acc['dfs']
    state = acc['state']
    size_state = acc['size-state']

    dfs, old = diff(dfs, new, window=window)

    if 'groupers' in acc:
        groupers = deque(acc['groupers'])
        if len(grouper) > 0:
            groupers.append(grouper)
        old_groupers, groupers = diff_align(dfs, groupers)
    else:
        old_groupers = [grouper] * len(old)

    if new is not None:
        state, result = agg.on_new(state, new, grouper=grouper)
        size_state, _ = size.on_new(size_state, new, grouper=grouper)
    for o, og in zip(old, old_groupers):
        if 'groupers' in acc:
            assert len(o) == len(og)
        if len(o):
            state, result = agg.on_old(state, o, grouper=og)
            size_state, _ = size.on_old(size_state, o, grouper=og)

    nonzero = size_state != 0
    if not nonzero.all():
        size_state = size_state[nonzero]
        result = result[nonzero]
        if isinstance(state, tuple):
            state = tuple(s[nonzero] for s in state)
        else:
            state = state[nonzero]

    acc2 = {'dfs': dfs, 'state': state, 'size-state': size_state}
    if 'groupers' in acc:
        acc2['groupers'] = groupers
    return acc2, result
def windowed_groupby_accumulator ( acc , new , diff = None , window = None , agg = None , grouper = None , with_state = False ) : if agg . grouper is None and isinstance ( new , tuple ) : new , grouper = new else : grouper = None size = GroupbySize ( agg . columns , agg . grouper ) if acc is None : acc = { 'dfs' : [ ] , 'state' : agg . initial ( new , grouper = grouper ) , 'size-state' : size . initial ( new , grouper = grouper ) } if isinstance ( grouper , np . ndarray ) or is_series_like ( grouper ) or is_index_like ( grouper ) : acc [ 'groupers' ] = deque ( [ ] ) dfs = acc [ 'dfs' ] state = acc [ 'state' ] size_state = acc [ 'size-state' ] dfs , old = diff ( dfs , new , window = window ) if 'groupers' in acc : groupers = deque ( acc [ 'groupers' ] ) if len ( grouper ) > 0 : groupers . append ( grouper ) old_groupers , groupers = diff_align ( dfs , groupers ) else : old_groupers = [ grouper ] * len ( old ) if new is not None : state , result = agg . on_new ( state , new , grouper = grouper ) size_state , _ = size . on_new ( size_state , new , grouper = grouper ) for o , og in zip ( old , old_groupers ) : if 'groupers' in acc : assert len ( o ) == len ( og ) if len ( o ) : state , result = agg . on_old ( state , o , grouper = og ) size_state , _ = size . on_old ( size_state , o , grouper = og ) nonzero = size_state != 0 if not nonzero . all ( ) : size_state = size_state [ nonzero ] result = result [ nonzero ] if isinstance ( state , tuple ) : state = tuple ( s [ nonzero ] for s in state ) else : state = state [ nonzero ] acc2 = { 'dfs' : dfs , 'state' : state , 'size-state' : size_state } if 'groupers' in acc : acc2 [ 'groupers' ] = groupers return acc2 , result