def window_accumulator(acc, new, diff=None, window=None, agg=None):
    """ An accumulation binary operator for windowed aggregations

    This is the function that is actually given to the ``Stream.accumulate``
    function.  It performs all of the work given old state, new data, a diff
    function, window value, and aggregation object.

    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()``

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

    See Also
    --------
    accumulator
    windowed_groupby_accumulator
    """
    if acc is None:
        acc = {'dfs': [], 'state': agg.initial(new)}
    dfs = acc['dfs']
    state = acc['state']
    dfs, old = diff(dfs, new, window=window)
    if new is not None:
        state, result = agg.on_new(state, new)
    for o in old:
        if len(o):
            state, result = agg.on_old(state, o)
    acc2 = {'dfs': dfs, 'state': state}
    return acc2, result
