如何使用Python/pandas获取带有行摘要的分钟/小时财务数据？

Question

假设我有如下几分钟的财务数据，我想编写一个用户定义的函数（下面的代码既难看又复杂），如何使用CSV中的Python/pandas获得5分钟/10分钟/30分钟/1小时/8小时/24小时的行摘要数据

    TIME     OPEN    HIGH    LOW     CLOSE   VOLUME
         ----------------------------------------------
0   1592194620 3046.00 3048.50 3046.00 3047.50 505          
1   1592194630 3047.00 3048.00 3046.00 3047.00 162          
2   1592194640 3047.50 3048.00 3047.00 3047.50 98           
3   1592194650 3047.50 3047.50 3047.00 3047.50 228          
4   1592194660 3048.00 3048.00 3047.50 3048.00 136          
5   1592194670 3048.00 3048.00 3046.50 3046.50 174          
6   1592194680 3046.50 3046.50 3045.00 3045.00 134          
7   1592194690 3045.50 3046.00 3044.00 3045.00 43           
8   1592194700 3045.00 3045.50 3045.00 3045.00 214          
9   1592194710 3045.50 3045.50 3045.50 3045.50 8            
10  1592194720 3045.50 3046.00 3044.50 3044.50 152
    .......
    .......
19999   1591594660 3048.00 3048.00 3047.50 3048.00 136

示例输出如下所示：

3048.50 2140 2020-06-13 04:34:00
3050.50 67 2020-06-13 04:35:00
3049.50 1489 2020-06-13 04:36:00
3047.50 987 2020-06-13 04:37:00
......
3099.50 2 2020-06-14 04:34:00

下面是我愚蠢的代码：

import pandas as pd
import pymysql
conn = pymysql.connect( host = "localhost",
                        user="root",
                        passwd="root",
                        db="demo")

sql = "SELECT TIME, OPEN, HIGH, LOW, CLOSE, VOLUME FROM demo_table;"
df = pd.read_sql(sql, conn)

# 12 hours for 1000 records
for i in range(1000, 20000-1000,1):
    high_price = df.loc[i,['high']][0]
    df_1000 = df.loc[i-1000:i]
    df_high = df_1000[df_1000['high']>high_price]
    high_count = df_high.shape[0]
    df_last = df_high.tail(1)
    time_dt = pd.Timestamp(df_last['TIME'], unit='s')
    print(high_price, high_count, time_dt )

Answer 1

首先，我建议读取CSV并将时间设置为索引：

import pandas as pd
import numpy as np

df = pd.read_csv(csv_file, delim_whitespace=True)
df['TIME'] = pd.to_datetime(df['TIME'], unit='s')
df.set_index('TIME', inplace=True)

如果您只是想将时间间隔缩短到另一个时间间隔（例如，从当前的1分钟缩短到5分钟），您可以使用Dataframe.resample方法轻松地对其重新采样：

# Tells what the aggregation should do for each column 
colls_agg = {'OPEN': lambda x: x.iloc[0], 
             'HIGH': 'max', 
             'LOW': 'min', 
             'CLOSE': lambda x: x.iloc[-1], 
             'VOLUME': 'sum'}

def get_summary(df, time_interval):
    # Tells what the aggregation should do for each column
    return df.resample(pd.Timedelta(time_interval)).agg(colls_agg)

如果您希望数据帧的每一行对应于最后X分钟的摘要（我相信这是您想要的），那么您需要为每一行重新计算它，如下所示

colls_agg = {'OPEN': lambda x: x.iloc[0], 
             'HIGH': 'max', 
             'LOW': 'min', 
             'CLOSE': lambda x: x.iloc[-1], 
             'VOLUME': 'sum'}

def recompute_summary_line(line, full_df, time_interval):
    """Recomputes the summary for a line of the dataframe. 
    line should be a line of the dataframe, 
    full_df is the full dataframe
    time_interval is the interval of time which will be selected"""
    
    # Selects time betwen current time - time_interval 
    # until current time (including it) 
    lines_to_select = (full_df.index > line.name - time_interval) & \
                      (full_df.index <= (line.name))
    agg_value = full_df[lines_to_select].agg(colls_agg)

    # For the first few lines, this is not possible, so it returns nan
    # Since we have included the current time, it will never happen. 
    # If you do NOT to include the current time, you might use this.
    if agg_value.empty:
        return pd.Series({'OPEN': np.nan, 'HIGH': np.nan, 
                          'LOW': np.nan, 'VOLUME': np.nan})

    return agg_value

def recompute_summary (df, time_interval):
    """Given a dataframe df, recomputes the summary for the 
    current time of each row using the information from the the previous 
    interval given in time_interval (for example '5min', '30s')"""
    
    # Use df.apply to apply it in each line of the dataframe
    return df.apply(lambda x: recompute_summary_line(
        x, df, pd.Timedelta(time_interval)), axis='columns')

recompute_summary(df, '1min')
recompute_summary(df, '12h')