Ежедневные данные, resample каждые 3 дня, рассчитывать за последние 5 дней эффективно

Ответ 1

Ниже перечислены two три несколько основанных на NumPy решений с использованием суммирования на основе bin, охватывающего в основном три сценария.

Сценарий №1: несколько записей за день, но отсутствующих дат

Подход №1:

# For now hard-coded to use Window size of 5 and stride length of 3
def vectorized_app1(df):
    # Extract the index names and values
    vals = df.A.values
    indx = df.index.values

    # Extract IDs for bin based summing
    mask = np.append(False,indx[1:] > indx[:-1])
    date_id = mask.cumsum()
    search_id = np.hstack((0,np.arange(2,date_id[-1],3),date_id[-1]+1))
    shifts = np.searchsorted(date_id,search_id)
    reps = shifts[1:] - shifts[:-1]
    id_arr = np.repeat(np.arange(len(reps)),reps)

    # Perform bin based summing and subtract the repeated ones
    IDsums = np.bincount(id_arr,vals)
    allsums = IDsums[:-1] + IDsums[1:]
    allsums[1:] -= np.bincount(date_id,vals)[search_id[1:-2]]

    # Convert to pandas dataframe if needed
    out_index = indx[np.nonzero(mask)[0][3::3]] # Use last date of group
    return pd.DataFrame(allsums,index=out_index,columns=['A'])

Подход № 2:

# For now hard-coded to use Window size of 5 and stride length of 3
def vectorized_app2(df):
    # Extract the index names and values
    indx = df.index.values

    # Extract IDs for bin based summing
    mask = np.append(False,indx[1:] > indx[:-1])
    date_id = mask.cumsum()

    # Generate IDs at which shifts are to happen for a (2,3,5,8..) patttern    
    # Pad with 0 and length of array at either ends as we use diff later on
    shiftIDs = (np.arange(2,date_id[-1],3)[:,None] + np.arange(2)).ravel()
    search_id = np.hstack((0,shiftIDs,date_id[-1]+1))

    # Find the start of those shifting indices    
    # Generate ID based on shifts and do bin based summing of dataframe
    shifts = np.searchsorted(date_id,search_id)
    reps = shifts[1:] - shifts[:-1]
    id_arr = np.repeat(np.arange(len(reps)),reps)    
    IDsums = np.bincount(id_arr,df.A.values)

    # Sum each group of 3 elems with a stride of 2, make dataframe if needed
    allsums = IDsums[:-1:2] + IDsums[1::2] + IDsums[2::2]    

    # Convert to pandas dataframe if needed
    out_index = indx[np.nonzero(mask)[0][3::3]] # Use last date of group
    return pd.DataFrame(allsums,index=out_index,columns=['A'])

Подход № 3:

def vectorized_app3(df, S=3, W=5):
    dt = df.index.values
    shifts = np.append(False,dt[1:] > dt[:-1])
    c = np.bincount(shifts.cumsum(),df.A.values)
    out = np.convolve(c,np.ones(W,dtype=int),'valid')[::S]
    out_index = dt[np.nonzero(shifts)[0][W-2::S]]
    return pd.DataFrame(out,index=out_index,columns=['A'])

Мы могли бы заменить часть свертки на прямое нарезанное суммирование для модифицированной версии ее -

def vectorized_app3_v2(df, S=3, W=5):  
    dt = df.index.values
    shifts = np.append(False,dt[1:] > dt[:-1])
    c = np.bincount(shifts.cumsum(),df.A.values)
    f = c.size+S-W
    out = c[:f:S].copy()
    for i in range(1,W):
        out += c[i:f+i:S]
    out_index = dt[np.nonzero(shifts)[0][W-2::S]]
    return pd.DataFrame(out,index=out_index,columns=['A'])

Сценарий №2: несколько записей за дату и отсутствие дат

Подход № 4:

def vectorized_app4(df, S=3, W=5):
    dt = df.index.values
    indx = np.append(0,((dt[1:] - dt[:-1])//86400000000000).astype(int)).cumsum()
    WL = ((indx[-1]+1)//S)
    c = np.bincount(indx,df.A.values,minlength=S*WL+(W-S))
    out = np.convolve(c,np.ones(W,dtype=int),'valid')[::S]
    grp0_lastdate = dt[0] + np.timedelta64(W-1,'D')
    freq_str = str(S)+'D'
    grp_last_dt = pd.date_range(grp0_lastdate, periods=WL, freq=freq_str).values
    out_index = dt[dt.searchsorted(grp_last_dt,'right')-1]
    return pd.DataFrame(out,index=out_index,columns=['A'])

Сценарий № 3: последовательные даты и ровно одна запись за день

Подход № 5:

def vectorized_app5(df, S=3, W=5):
    vals = df.A.values
    N = (df.shape[0]-W+2*S-1)//S
    n = vals.strides[0]
    out = np.lib.stride_tricks.as_strided(vals,shape=(N,W),\
                                        strides=(S*n,n)).sum(1)
    index_idx = (W-1)+S*np.arange(N)
    out_index = df.index[index_idx]
    return pd.DataFrame(out,index=out_index,columns=['A'])

Предложения для создания тестовых данных

Сценарий №1:

# Setup input for multiple dates, but no missing dates
S = 4 # Stride length (Could be edited)
W = 7 # Window length (Could be edited)
datasize = 3  # Decides datasize
tidx = pd.date_range('2012-12-31', periods=datasize*S + W-S, freq='D')
start_df = pd.DataFrame(dict(A=np.arange(len(tidx))), tidx)
reps = np.random.randint(1,4,(len(start_df)))
idx0 = np.repeat(start_df.index,reps)
df_data = np.random.randint(0,9,(len(idx0)))
df = pd.DataFrame(df_data,index=idx0,columns=['A'])

Сценарий № 2:

Чтобы создать установку для нескольких дат и с отсутствующими датами, мы могли бы просто отредактировать шаг создания df_data, например:

df_data = np.random.randint(0,9,(len(idx0)))

Сценарий № 3:

# Setup input for exactly one entry per date
S = 4 # Could be edited
W = 7
datasize = 3  # Decides datasize
tidx = pd.date_range('2012-12-31', periods=datasize*S + W-S, freq='D')
df = pd.DataFrame(dict(A=np.arange(len(tidx))), tidx)

Ответ 2

вы дали нам:

             A
2012-12-31   0
2013-01-01   1
2013-01-02   2
2013-01-03   3
2013-01-04   4
2013-01-05   5
2013-01-06   6
2013-01-07   7
2013-01-08   8
2013-01-09   9
2013-01-10  10

вы можете создать свою текущую 5-дневную серию сумм, а затем перепрограммировать ее. Я не могу придумать более эффективный способ, чем это. В целом это должно быть относительно эффективным по времени.

df.rolling(5,min_periods=5).sum().dropna().resample('3D').first()
Out[36]: 
                 A
2013-01-04 10.0000
2013-01-07 25.0000
2013-01-10 40.0000

Ответ 3

Только для регулярных интервалов

Вот два метода: сначала pandas и вторая функция numpy.

>>> n=5   # trailing periods for rolling sum
>>> k=3   # frequency of rolling sum calc

>>> df.rolling(n).sum()[-1::-k][::-1]

               A
2013-01-01   NaN
2013-01-04  10.0
2013-01-07  25.0
2013-01-10  40.0

И здесь функция numpy (адаптирована из Jaime numpy moving_average):

def rolling_sum(a, n=5, k=3):
    ret = np.cumsum(a.values)
    ret[n:] = ret[n:] - ret[:-n]
    return pd.DataFrame( ret[n-1:][-1::-k][::-1], 
                         index=a[n-1:][-1::-k][::-1].index )

rolling_sum(df,n=6,k=4)   # default n=5, k=3

Для нерегулярно расположенных интервалов (или регулярных интервалов)

Просто представьте:

df.resample('D').sum().fillna(0)

Например, вышеприведенные методы становятся:

df.resample('D').sum().fillna(0).rolling(n).sum()[-1::-k][::-1]

и

rolling_sum( df.resample('D').sum().fillna(0) )

Обратите внимание, что работа с нерегулярными интервалами может быть выполнена просто и элегантно в pandas, поскольку это сила pandas почти во всем остальном. Но вы, скорее всего, найдете метод numpy (или numba или cython), который будет компрометировать некоторую простоту для увеличения скорости. Разумеется, это хороший компромисс, будет зависеть от ваших размеров данных и требований к производительности.

Для нерегулярно разнесенных дат я тестировал следующие данные примера и, похоже, работал правильно. Это приведет к смешиванию отсутствующих, одиночных и нескольких записей за дату:

np.random.seed(12345)
per = 11
tidx = np.random.choice( pd.date_range('2012-12-31', periods=per, freq='D'), per )
df = pd.DataFrame(dict(A=np.arange(len(tidx))), tidx).sort_index()

Ответ 4

это еще не совсем идеально, но сегодня я должен сделать фальшивую кровь для вечеринки с галоином... вы должны уметь видеть, что я получаю от комментариев. Одним из самых больших ускорений является поиск краев окна с помощью np.searchsorted. он еще не совсем работает, но я бы поставил на него только некоторые смещения индекса, которые нуждаются в настройке

import pandas as pd
import numpy as np

tidx = pd.date_range('2012-12-31', periods=11, freq='D')
df = pd.DataFrame(dict(A=np.arange(len(tidx))), tidx)

sample_freq = 3 #days
sample_width = 5 #days

sample_freq *= 86400 #seconds per day
sample_width *= 86400 #seconds per day

times = df.index.astype(np.int64)//10**9  #array of timestamps (unix time)
cumsum = np.cumsum(df.A).as_matrix()  #array of cumulative sums (could eliminate extra summation with large overlap)
mat = np.array([times, cumsum]) #could eliminate temporary times and cumsum vars

def yieldstep(mat, freq):
    normtime = ((mat[0] - mat[0,0]) / freq).astype(int) #integer numbers indicating sample number
    for i in range(max(normtime)+1):
        yield np.searchsorted(normtime, i) #yield beginning of window index

def sumwindow(mat,i , width): #i is the start of the window returned by yieldstep
    normtime  = ((mat[0,i:] - mat[0,i])/ width).astype(int) #same as before, but we norm to window width
    j = np.searchsorted(normtime, i, side='right')-1 #find the right side of the window
    #return rightmost timestamp of window in seconds from unix epoch and sum of window
    return mat[0,j], mat[1,j] - mat[1,i] #sum of window is just end - start because we did a cumsum earlier

windowed_sums = np.array([sumwindow(mat, i, sample_width) for i in yieldstep(mat, sample_freq)])

Ответ 5

Похоже на развернутое центрированное окно, где вы берете данные каждые n дней:

def rolleach(df, ndays, window):
    return df.rolling(window, center=True).sum()[ndays-1::ndays]

rolleach(df, 3, 5)
Out[95]: 
               A
2013-01-02  10.0
2013-01-05  25.0
2013-01-08  40.0

Ответ 6

Если фреймворк данных сортируется по дате, то на самом деле мы имеем итерацию по массиву при вычислении чего-то.

Здесь алгоритм, который вычисляет суммы всего на одной итерации по массиву. Чтобы понять это, просмотрите мои заметки ниже. Это базовая неоптимизированная версия, предназначенная для демонстрации алгоритма _{(оптимизированные для Python и Cython)}, а list(<call>) принимает ~500 ms для массива в 100k в моей системе (P4). Так как диапазоны Python и диапазоны относительно медленные, это должно принести огромную пользу от перехода на уровень C.

from __future__ import division
import numpy as np

#The date column is unimportant for calculations.
# I leave extracting the numbers' column from the dataframe
# and adding a corresponding element from data column to each result
# as an exercise for the reader
data = np.random.randint(100,size=100000)

def calc_trailing_data_with_interval(data,n,k):
    """Iterate over `data', computing sums of `n' trailing elements
    for each `k'th element.
    @type data: ndarray
    @param n: number of trailing elements to sum up
    @param k: interval with which to calculate sums
    """
    lim_index=len(data)-k+1

    nsums = int(np.ceil(n/k))
    sums = np.zeros(nsums,dtype=data.dtype)
    M=n%k
    Mp=k-M

    index=0
    currentsum=0

    while index<lim_index:
        for _ in range(Mp):
            #np.take is awkward, requiring a full list of indices to take
            for i in range(currentsum,currentsum+nsums-1):
                sums[i%nsums]+=data[index]
            index+=1
        for _ in range(M):
            sums+=data[index]
            index+=1
        yield sums[currentsum]
        currentsum=(currentsum+1)%nsums

• Обратите внимание, что он производит первую сумму в k -ном элементе, а не n th (это можно изменить, но, пожертвовав элегантностью - количество фиктивных итераций перед основным циклом - и более элегантно сделано путем добавления data с дополнительными нулями и отбрасыванием ряда первых сумм)
• Его можно легко обобщить на любую операцию, заменив sums[slice]+=data[index] на operation(sums[slice],data[index]), где operation является параметром и должен быть мутирующей операцией (например, ndarray.__iadd__).
• распараллеливание между любым числом или рабочими путем разделения данных так же просто (если n>k, куски после первого должны быть добавлены дополнительные элементы в начале)

Чтобы вывести алгоритм, я написал образец для случая, когда приличное количество сумм вычисляется одновременно, чтобы увидеть шаблоны _{(щелкните изображение, чтобы увидеть его полноразмерным)}.

Оптимизировано: чистый Python

Кэширование range приводит к сокращению времени до ~300ms. Удивительно, но функциональность numpy не нужна: np.take непригодна для использования, а замена логики currentsum статическими срезами и np.roll является регрессией. Еще более удивительно, что преимущество сохранения вывода на np.empty в отличие от yield несуществует.

def calc_trailing_data_with_interval(data,n,k):
    """Iterate over `data', computing sums of `n' trailing elements
    for each `k'th element.
    @type data: ndarray
    @param n: number of trailing elements to sum up
    @param k: interval with which to calculate sums
    """
    lim_index=len(data)-k+1

    nsums = int(np.ceil(n/k))
    sums = np.zeros(nsums,dtype=data.dtype)
    M=n%k
    Mp=k-M
    RM=range(M)     #cache for efficiency
    RMp=range(Mp)   #cache for efficiency

    index=0
    currentsum=0
    currentsum_ranges=[range(currentsum,currentsum+nsums-1)
            for currentsum in range(nsums)]     #cache for efficiency

    while index<lim_index:
        for _ in RMp:
            #np.take is unusable as it allocates another array rather than view
            for i in currentsum_ranges[currentsum]:
                sums[i%nsums]+=data[index]
            index+=1
        for _ in RM:
            sums+=data[index]
            index+=1
        yield sums[currentsum]
        currentsum=(currentsum+1)%nsums

Оптимизировано: Cython

Статическая типизация всего в Cython мгновенно ускоряет работу до 150ms. И (необязательно), предполагая np.int как dtype, чтобы иметь возможность работать с данными на уровне C, сокращает время до ~11ms. На данный момент сохранение в np.empty действительно имеет значение, сохраняя невероятное ~6.5ms, суммируя ~5.5ms.

def calc_trailing_data_with_interval(np.ndarray data,int n,int k):
    """Iterate over `data', computing sums of `n' trailing elements
    for each `k'th element.
    @type data: 1-d ndarray
    @param n: number of trailing elements to sum up
    @param k: interval with which to calculate sums
    """
    if not data.ndim==1: raise TypeError("One-dimensional array required")
    cdef int lim_index=data.size-k+1

    cdef np.ndarray result = np.empty(data.size//k,dtype=data.dtype)
    cdef int rindex = 0

    cdef int nsums = int(np.ceil(float(n)/k))
    cdef np.ndarray sums = np.zeros(nsums,dtype=data.dtype)

    #optional speedup for dtype=np.int
    cdef bint use_int_buffer = data.dtype==np.int and data.flags.c_contiguous
    cdef int[:] cdata = data
    cdef int[:] csums = sums
    cdef int[:] cresult = result

    cdef int M=n%k
    cdef int Mp=k-M

    cdef int index=0
    cdef int currentsum=0

    cdef int _,i
    while index<lim_index:
        for _ in range(Mp):
            #np.take is unusable as it allocates another array rather than view
            for i in range(currentsum,currentsum+nsums-1):
                if use_int_buffer:  csums[i%nsums]+=cdata[index]    #optional speedup
                else:               sums[i%nsums]+=data[index]
            index+=1
        for _ in range(M):
            if use_int_buffer:
                for i in range(nsums): csums[i]+=cdata[index]   #optional speedup
            else:               sums+=data[index]
            index+=1

        if use_int_buffer:  cresult[rindex]=csums[currentsum]     #optional speedup
        else:               result[rindex]=sums[currentsum]
        currentsum=(currentsum+1)%nsums
        rindex+=1
    return result