def cov_corr_agg(data, meta, min_periods=2, corr=False, scalar=False):
    """Aggregation part of a covariance or correlation computation"""
    data = np.concatenate(data).reshape((len(data),) + data[0].shape)
    sums = np.nan_to_num(data['sum'])
    counts = data['count']

    cum_sums = np.cumsum(sums, 0)
    cum_counts = np.cumsum(counts, 0)

    s1 = cum_sums[:-1]
    s2 = sums[1:]
    n1 = cum_counts[:-1]
    n2 = counts[1:]
    d = (s2/n2) - (s1/n1)
    C = (np.nansum((n1 * n2)/(n1 + n2) * (d * d.transpose((0, 2, 1))), 0) +
         np.nansum(data['cov'], 0))

    C[cum_counts[-1] < min_periods] = np.nan
    nobs = np.where(cum_counts[-1], cum_counts[-1], np.nan)
    if corr:
        mu = cum_sums[-1] / nobs
        counts_na = np.where(counts, counts, np.nan)
        m2 = np.nansum(data['m'] + counts*(sums/counts_na - mu)**2, axis=0)
        den = np.sqrt(m2 * m2.T)
    else:
        den = nobs - 1
    mat = C/den
    if scalar:
        return mat[0, 1]
    return pd.DataFrame(mat, columns=meta.columns, index=meta.columns)
dataset/ETHPy150Open dask/dask/dask/dataframe/core.py/cov_corr_agg/original