Преобразование данных запаса OHLC в другой таймфрейм с помощью python и pandas

Ответ 1

Ваш подход звучит, но терпит неудачу, потому что каждая функция в dict-of-functions применяется к agg() получает объект Series, отражающий столбец, соответствующий значению ключа. Поэтому нет необходимости снова наклейте на метку столбца. При этом и при условии, что groupby сохраняет порядок, вы можете нарезать серию для извлечения первого/последнего элемента Open/Close столбцы (примечание: документация groupby не требует сохранения порядка исходных данных серии, но, кажется, на практике.)

In [50]: df.groupby(dr5minute.asof).agg({'Low': lambda s: s.min(), 
                                         'High': lambda s: s.max(),
                                         'Open': lambda s: s[0],
                                         'Close': lambda s: s[-1],
                                         'Volume': lambda s: s.sum()})
Out[50]: 
                      Close    High     Low    Open  Volume
key_0                                                      
1999-01-04 10:20:00  1.1806  1.1819  1.1801  1.1801      34
1999-01-04 10:25:00  1.1789  1.1815  1.1776  1.1807      91
1999-01-04 10:30:00  1.1791  1.1792  1.1776  1.1780      16

Для справки, вот таблица, в которой суммируются ожидаемые входных и выходных типов функции агрегации на основе типа объекта groupby и того, как функция (аг) агрегации передана/передается в agg().

                  agg() method     agg func    agg func          agg()
                  input type       accepts     returns           result
GroupBy Object
SeriesGroupBy     function         Series      value             Series
                  dict-of-funcs    Series      value             DataFrame, columns match dict keys
                  list-of-funcs    Series      value             DataFrame, columns match func names
DataFrameGroupBy  function         DataFrame   Series/dict/ary   DataFrame, columns match original DataFrame
                  dict-of-funcs    Series      value             DataFrame, columns match dict keys, where dict keys must be columns in original DataFrame
                  list-of-funcs    Series      value             DataFrame, MultiIndex columns (original cols x func names)

Из приведенной выше таблицы, если для агрегации требуется доступ к нескольким столбец, единственный вариант - передать одну функцию в Объект DataFrameGroupBy. Поэтому альтернативный способ выполнения первоначальной задачи - определить функция, подобная следующей:

def ohlcsum(df):
    df = df.sort()
    return {
       'Open': df['Open'][0],
       'High': df['High'].max(),
       'Low': df['Low'].min(),
       'Close': df['Close'][-1],
       'Volume': df['Volume'].sum()
      }

и примените к нему agg():

In [30]: df.groupby(dr5minute.asof).agg(ohlcsum)
Out[30]: 
                       Open    High     Low   Close  Volume
key_0                                                      
1999-01-04 10:20:00  1.1801  1.1819  1.1801  1.1806      34
1999-01-04 10:25:00  1.1807  1.1815  1.1776  1.1789      91
1999-01-04 10:30:00  1.1780  1.1792  1.1776  1.1791      16

Хотя pandas может предложить более чистую встроенную магию в будущем, надеюсь, это объясняет, как работать с сегодняшними возможностями agg().

Ответ 2

просто для того, чтобы быть полезным другим пользователям с более поздней версией Pandas, метод resample очень быстрый и полезный для выполнения той же задачи:

ohlc_dict = {                                                                                                             
'Open':'first',                                                                                                    
'High':'max',                                                                                                       
'Low':'min',                                                                                                        
'Close': 'last',                                                                                                    
'Volume': 'sum'
}

df.resample('5T', how=ohlc_dict, closed='left', label='left')

Ответ 3

В моей функции main() я получаю потоковые данные о ставках/запросах. Затем я делаю следующее:

df = pd.DataFrame([])

for msg_type, msg in response.parts():
    if msg_type == "pricing.Price":
        sd = StreamingData(datetime.now(),instrument_string(msg),
                           mid_string(msg),account_api,account_id,
                           's','5min',balance)
        df = df.append(sd.df())
        sd.resample(df)

Я создал класс StreamingData(), который принимает предоставленный ввод (также создал некоторые функции, чтобы разбить данные ставки/запроса на отдельные компоненты (ставка, запрос, середина, инструмент и т.д.),.

Красота этого - все, что вам нужно сделать, это изменить 's и ' 5min ' на любые временные рамки. Установите его в "m" и "D", чтобы получить ежедневные цены за минуту.

Вот что выглядит моя StreamingData():

class StreamingData(object):
def __init__(self, time, instrument, mid, api, _id, xsec, xmin, balance):
    self.time = time
    self.instrument = instrument
    self.mid = mid
    self.api = api
    self._id = _id
    self.xsec = xsec
    self.xmin = xmin
    self.balance = balance
    self.data = self.resample(self.df())

def df(self):
    df1 = pd.DataFrame({'Time':[self.time]})
    df2 = pd.DataFrame({'Mid':[float(self.mid)]})
    df3 = pd.concat([df1,df2],axis=1,join='inner')
    df = df3.set_index(['Time'])
    df.index = pd.to_datetime(df.index,unit='s')
    return df

def resample(self, df):
    xx = df.to_period(freq=self.xsec)
    openCol = xx.resample(self.xmin).first()
    highCol = xx.resample(self.xmin).max()
    lowCol = xx.resample(self.xmin).min()
    closeCol = xx.resample(self.xmin).last()
    self.data = pd.concat([openCol,highCol,lowCol,closeCol],
                           axis=1,join='inner')
    self.data['Open'] = openCol.round(5)
    self.data['High'] = highCol.round(5)
    self.data['Low'] = lowCol.round(5)
    self.data['Close'] = closeCol.round(5)
    return self.data

Таким образом, он принимает данные из StreamingData(), создает индексированный по времени блок данных в df(), добавляет его, а затем передает через resample ( ). Цены, которые я рассчитываю, основаны на: mid = (bid + ask)/2

Ответ 4

bitfinex_klines_pd = bitfinex_klines_pd.resample('4h').agg({
    'open': lambda s: s[0],
    'high': lambda df: df.max(),
    'low': lambda df: df.min(),
    'close': lambda df: df[-1],
    'volume': lambda df: df.sum()
})