使用Python和Pandas将OHLC股票数据转换为不同的时间框架

Question

使用Python和Pandas将OHLC股票数据转换为不同的时间框架

34

请问有人能够指导我使用Pandas将OHLC数据转换为不同时间框架吗？我想要做的是使用低时间框架的数据构建出高时间框架的Dataframe。

例如，给定以下一分钟（M1）数据：

                       Open    High     Low   Close  Volume
Date                                                       
1999-01-04 10:22:00  1.1801  1.1819  1.1801  1.1817       4
1999-01-04 10:23:00  1.1817  1.1818  1.1804  1.1814      18
1999-01-04 10:24:00  1.1817  1.1817  1.1802  1.1806      12
1999-01-04 10:25:00  1.1807  1.1815  1.1795  1.1808      26
1999-01-04 10:26:00  1.1803  1.1806  1.1790  1.1806       4
1999-01-04 10:27:00  1.1801  1.1801  1.1779  1.1786      23
1999-01-04 10:28:00  1.1795  1.1801  1.1776  1.1788      28
1999-01-04 10:29:00  1.1793  1.1795  1.1782  1.1789      10
1999-01-04 10:31:00  1.1780  1.1792  1.1776  1.1792      12
1999-01-04 10:32:00  1.1788  1.1792  1.1788  1.1791       4

数据集包括每分钟的开盘价、最高价、最低价、收盘价（OHLC）和成交量数值。我想构建一组5分钟的读数（M5），如下所示：

                       Open    High     Low   Close  Volume
Date                                                       
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

因此工作流程如下：

Open是时间窗口中第一行的Open值
High是时间窗口内最高的High值
Low是最低的Low值
Close是最后一个Close值
Volume是各个时间间隔内Volume的总和

然而，还有一些问题：

数据存在缺口（请注意，没有10:30:00行）
5分钟的时间间隔必须从整点开始，例如M5从10:25:00开始，而不是10:22:00
首个不完整的集可以被省略（就像这个例子中一样），也可以被包含在内（因此我们可以有10:20:00的5分钟记录）

Pandas关于上下采样的文档给出了一个示例，但它们使用平均值作为上采样行的值，这在这里不起作用。我尝试使用groupby和agg，但没有成功。对于获取最高的High和最低的Low可能不是那么困难，但我不知道如何获取第一个Open和最后一个Close。

我尝试的方法是类似于：

grouped = slice.groupby( dr5minute.asof ).agg( 
    { 'Low': lambda x : x.min()[ 'Low' ], 'High': lambda x : x.max()[ 'High' ] } 
)

但是它会导致以下错误，我不理解：

In [27]: grouped = slice.groupby( dr5minute.asof ).agg( { 'Low' : lambda x : x.min()[ 'Low' ], 'High' : lambda x : x.max()[ 'High' ] } )
---------------------------------------------------------------------------
IndexError                                Traceback (most recent call last)
/work/python/fxcruncher/<ipython-input-27-df50f9522a2f> in <module>()
----> 1 grouped = slice.groupby( dr5minute.asof ).agg( { 'Low' : lambda x : x.min()[ 'Low' ], 'High' : lambda x : x.max()[ 'High' ] } )

/usr/lib/python2.7/site-packages/pandas/core/groupby.pyc in agg(self, func, *args, **kwargs)
    242         See docstring for aggregate
    243         """
--> 244         return self.aggregate(func, *args, **kwargs)
    245 
    246     def _iterate_slices(self):

/usr/lib/python2.7/site-packages/pandas/core/groupby.pyc in aggregate(self, arg, *args, **kwargs)
   1153                     colg = SeriesGroupBy(obj[col], column=col,
   1154                                          grouper=self.grouper)
-> 1155                     result[col] = colg.aggregate(func)
   1156 
   1157             result = DataFrame(result)

/usr/lib/python2.7/site-packages/pandas/core/groupby.pyc in aggregate(self, func_or_funcs, *args, **kwargs)
    906                 return self._python_agg_general(func_or_funcs, *args, **kwargs)
    907             except Exception:
--> 908                 result = self._aggregate_named(func_or_funcs, *args, **kwargs)
    909 
    910             index = Index(sorted(result), name=self.grouper.names[0])

/usr/lib/python2.7/site-packages/pandas/core/groupby.pyc in _aggregate_named(self, func, *args, **kwargs)
    976             grp = self.get_group(name)
    977             grp.name = name
--> 978             output = func(grp, *args, **kwargs)
    979             if isinstance(output, np.ndarray):
    980                 raise Exception('Must produce aggregated value')

/work/python/fxcruncher/<ipython-input-27-df50f9522a2f> in <lambda>(x)
----> 1 grouped = slice.groupby( dr5minute.asof ).agg( { 'Low' : lambda x : x.min()[ 'Low' ], 'High' : lambda x : x.max()[ 'High' ] } )

IndexError: invalid index to scalar variable.

非常感谢能给予任何帮助。如果我选择的路径不可行，请提供其他相对高效的方法（我有数百万行数据）。使用Pandas进行财务处理的一些资源也会很好。

- kgr

2

你使用的pandas版本是什么？我们正在开发重新设计的时间序列功能，这将大大简化这个过程，但不太可能在四月底之前发布。不过，这里也可能有需要修复的错误。 - Wes McKinney

1

嗨Wes，我正在使用0.7.2版本。我想等待新版本发布是一个可行的选择，因为我没有这个转换的截止日期（我需要数据进行私人研究）。顺便感谢您为开发Pandas付出的努力！ :) - kgr

至于可能存在的错误，请注意我没有为Dataframe中的所有列指定值（仅有5个中的2个），如果这是您所指的。 - kgr

1

https://www.quantinsti.com/blog/tick-tick-ohlc-data-pandas-tutorial - jaggedsoft

4个回答

16

你的方法是正确的，但会失败，因为字典中应用于agg()的每个函数都会接收到反映键值匹配的列的Series对象。因此，再次过滤列标签是不必要的。如果假设groupby保留顺序，那么你可以对Series进行切片以提取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() 功能。

- Garrett

1

首先感谢您提供的非常有用的答案 :) 您可以告诉我您使用的Pandas版本以及如何创建dr5minute吗？我在使用groupby(dr5minute.asof)时遇到了问题，它只返回一个组。 - kgr

我猜问题可能出在不正确的索引上。我认为CSV中的日期没有被正确解析成日期...但这是另一个故事，所以不需要在评论中讨论。再次感谢@crewburm！ - kgr

不客气，@kgr。我正在使用0.7.2版本。如果要解释csv中的日期，请查看“read_csv()”函数的“converters”参数。 - Garrett

5

df = df.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()
})

- Yundong Cai

4

你好！为了改进这个答案，您可能希望加入一些描述为什么/如何起作用。 - Auden Young

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

- Bicameral Mind

网页内容由stack overflow 提供, 点击上面的

可以查看英文原文，
原文链接

- Andrea · Accepted Answer

使用更新版本的Pandas，有一个resample方法。它非常快速，用于完成相同的任务非常有用：

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

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