Chuyển đổi dữ liệu chứng khoán OHLC thành một khung thời gian khác với trăn và gấu trúc

Question

Ai đó có thể vui lòng chỉ cho tôi đúng hướng về chuyển đổi khung thời gian dữ liệu OHLC với Pandas? Những gì tôi đang cố gắng làm là xây dựng một khung dữ liệu với dữ liệu cho khung thời gian cao hơn, được cung cấp dữ liệu với khung thời gian thấp hơn.

Ví dụ, cho tôi có (M1) dữ liệu sau một phút:

     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 

trong đó có Open, High, Low, Close (OHLC) và các giá trị khối lượng cho mỗi phút tôi muốn xây dựng một tập hợp các bài đọc 5 phút (M5) trong đó sẽ trông giống như vậy:

     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 

vì vậy, các quy trình làm việc là:

• mở là mở của t đầu tiên anh chèo trong timewindow
• cao là cao cao nhất trong timewindow
• thấp là thấp nhất Low
• Đóng là Close
• Khối lượng cuối cùng chỉ đơn giản là một khoản Volumes

Có có một vài vấn đề mặc dù:

• dữ liệu có khoảng trống (lưu ý không có hàng 10:30:00)
• khoảng thời gian 5 phút phải bắt đầu vào thời gian vòng, ví dụ: M5 bắt đầu lúc 10:25:00 không phải 10:22:00
• trước tiên, bộ không đầy đủ có thể được bỏ qua như trong ví dụ này hoặc được bao gồm (để chúng tôi có thể có mục nhập 10:20:00 5 phút)

Ví dụ Pandas documentation on up-down sampling đưa ra một ví dụ, nhưng chúng sử dụng giá trị trung bình làm giá trị của hàng được lấy mẫu, sẽ không hoạt động ở đây. Tôi đã thử sử dụng groupby và agg nhưng không có kết quả. Đối với một người nhận được cao nhất cao nhất và thấp nhất thấp có thể không phải là khó khăn như vậy, nhưng tôi không có ý tưởng làm thế nào để có được đầu tiên mở và cuối cùng gần.

Những gì tôi cố gắng là một cái gì đó dọc theo dòng:

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

nhưng nó kết quả trong báo lỗi sau, mà tôi không hiểu:

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. 

Vì vậy, bất kỳ giúp đỡ về làm đó sẽ là đánh giá cao. Nếu con đường tôi chọn sẽ không hoạt động, vui lòng đề xuất phương pháp tiếp cận tương đối hiệu quả khác (tôi có hàng triệu hàng). Một số tài nguyên sử dụng Pandas để xử lý tài chính cũng sẽ tốt đẹp.

Answer 1

Cách tiếp cận của bạn là âm thanh, nhưng không thành công vì mỗi hàm trong hàm dict được áp dụng cho agg() nhận đối tượng Chuỗi phản ánh cột khớp với giá trị khóa. Do đó, không cần thiết phải bộ lọc trên nhãn cột một lần nữa. Với điều này, và giả sử nhóm bảo toàn trật tự, bạn có thể cắt Series để trích xuất phần tử đầu tiên/cuối cùng của các cột Mở/Đóng (lưu ý: tài liệu nhóm không yêu cầu duy trì thứ tự dữ liệu gốc series, nhưng dường như thực hành.)

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 

Để tham khảo, đây là một bảng tóm tắt dự kiến ​​ loại đầu vào và đầu ra của một chức năng tập hợp dựa trên các loại đối tượng groupby và làm thế nào các hàm gộp (s) được/được truyền cho 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) 

Từ bảng trên, nếu tập hợp đòi hỏi quyền truy cập vào nhiều hơn một cột, lựa chọn duy nhất là phải vượt qua một chức năng duy nhất để một đối tượng DataFrameGroupBy. Vì vậy, một cách khác để hoàn thành nhiệm vụ ban đầu là để xác định một chức năng như sau:

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

và áp dụng agg() với nó:

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 

Mặc dù gấu trúc có thể cung cấp một số bụi built- trong ma thuật trong tương lai, hy vọng điều này giải thích cách làm việc với các khả năng agg() ngày nay.

Answer 2

chỉ để thể hữu ích cho những người dùng khác với một phiên bản mới hơn của gấu trúc, có một resample phương pháp rất nhanh chóng và hữu ích để thực hiện các nhiệm vụ tương tự:

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

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

Answer 3

Trong chính của tôi() chức năng Tôi đang nhận dữ liệu giá thầu/yêu cầu phát trực tuyến. sau đó tôi làm như sau:

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) 

Tôi tạo ra một lớp StreamingData() mà mất đầu vào cung cấp (cũng tạo ra một số chức năng để phá vỡ các thầu/hỏi dữ liệu vào thành phần riêng lẻ (giá thầu, hỏi, giữa, công cụ, vv).

vẻ đẹp của việc này là tất cả các bạn phải làm là thay đổi 's' và '5min' để bất cứ điều gì khung thời gian mà bạn muốn. Đặt nó vào 'm' và 'D' để nhận giá hàng ngày theo phút.

Đây là những gì StreamingData của tôi() trông giống như:

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 

Vì vậy, phải mất trong các dữ liệu từ StreamingData(), tạo ra một thời gian lập chỉ mục dataframe trong df(), gắn nó, sau đó gửi thông qua mẫu lại(). Giá tôi tính dựa trên: giữa = (giá thầu + yêu cầu)/2