text_plain_output_1.png
from tqdm import tqdm

import cu_utils.transform as cutran

import cudf

import cupy as cp

import glob

PATH = '/kaggle/input/optiver-realized-volatility-prediction'
def load_data(mode, path='/kaggle/input/optiver-realized-volatility-prediction'):
    file_name = f'{path}/{mode}.csv'
    return cudf.read_csv(file_name)
dev_df = load_data('train', path=PATH)

SCALE = 100
dev_df['target'] = SCALE * dev_df['target']
stock_ids = dev_df['stock_id'].unique()
len(stock_ids)

order_book_training = glob.glob(f'{PATH}/book_train.parquet/*/*')
order_book_test = glob.glob(f'{PATH}/book_test.parquet/*/*')
(len(order_book_training), len(order_book_test))

trades_training = glob.glob(f'{PATH}/trade_train.parquet/*/*')
trades_test = glob.glob(f'{PATH}/trade_test.parquet/*/*')
(len(trades_training), len(trades_test))

import cu_utils.transform as cutran
def log_diff(df, in_col, null_val):
    df['logx'] = df[in_col].log()
    df['logx_shifted'] = df[['time_id', 'logx']].groupby('time_id', method='cudf').apply_grouped(cutran.get_cu_shift_transform(shift_by=1, null_val=null_val), incols={'logx': 'x'}, outcols=dict(y_out=cp.float32), tpb=32)['y_out']
    df['keep_row'] = df[f'logx_shifted'] != null_val
    return df['logx'] - df['logx_shifted']
def realized_vol(log_return):
    return cp.sqrt((log_return * log_return).sum())
def extract_raw_book_features(df, null_val=-9999):
    df['wap1'] = (df['bid_price1'] * df['ask_size1'] + df['ask_price1'] * df['bid_size1']) / (df['bid_size1'] + df['ask_size1'])
    df['wap2'] = (df['bid_price2'] * df['ask_size2'] + df['ask_price2'] * df['bid_size2']) / (df['bid_size2'] + df['ask_size2'])
    df['wap3'] = (df['bid_price1'] * df['bid_size1'] + df['ask_price1'] * df['ask_size1']) / (df['bid_size1'] + df['ask_size1'])
    df['wap4'] = (df['bid_price2'] * df['bid_size2'] + df['ask_price2'] * df['ask_size2']) / (df['bid_size2'] + df['ask_size2'])
    for n in [1, 2, 3, 4]:
        df[f'log_return{n}'] = log_diff(df, in_col=f'wap{n}', null_val=null_val)
        df[f'realized_vol{n}'] = df[f'log_return{n}'] ** 2
    df['wap_balance'] = abs(df['wap1'] - df['wap2'])
    df['price_spread'] = (df['ask_price1'] - df['bid_price1']) / ((df['ask_price1'] + df['bid_price1']) / 2)
    df['price_spread1'] = (df['ask_price2'] - df['bid_price2']) / ((df['ask_price2'] + df['bid_price2']) / 2)
    df['bid_spread'] = df['bid_price1'] - df['bid_price2']
    df['ask_spread'] = df['ask_price1'] - df['ask_price2']
    df['bid_ask_spread'] = abs(df['bid_spread'] - df['ask_spread'])
    df['total_volume'] = df['ask_size1'] + df['ask_size2'] + (df['bid_size1'] + df['bid_size2'])
    df['volume_imbalance'] = abs(df['ask_size1'] + df['ask_size2'] - (df['bid_size1'] + df['bid_size2']))
    df = df[df['keep_row']]
    return df
def extract_raw_trade_features(df, null_val=-9999):
    df['realized_vol_trade'] = log_diff(df, in_col=f'price', null_val=null_val) ** 2
    df['amount'] = df['price'] * df['size']
    df = df[df['keep_row']]
    return df
def agg(df, feature_dict):
    agg_df = df.groupby('time_id').agg(feature_dict).reset_index()
    def f(x):
        if x[1] == '':
            return x[0]
        return x[0] + '_' + x[1]
    agg_df.columns = [f(x) for x in agg_df.columns]
    col_vol = [col for col in agg_df.columns if 'realized_vol' in col and ('mean' in col or 'sum' in col)]
    agg_df[col_vol] = agg_df[col_vol].sqrt()
    return agg_df
def extract_book_stats(df):
    feature_dict = {'wap1': ['sum', 'std'], 'wap2': ['sum', 'std'], 'wap3': ['sum', 'std'], 'wap4': ['sum', 'std'], 'realized_vol1': ['sum'], 'realized_vol2': ['sum'], 'realized_vol3': ['sum'], 'realized_vol4': ['sum'], 'price_spread': ['sum', 'max'], 'price_spread1': ['sum', 'max'], 'wap_balance': ['sum', 'max'], 'bid_spread': ['sum', 'max'], 'ask_spread': ['sum', 'max'], 'total_volume': ['sum', 'max'], 'volume_imbalance': ['sum', 'max'], 'bid_ask_spread': ['sum', 'max']}
    return agg(df, feature_dict)
def extract_book_stats_time(df):
    feature_dict = {'realized_vol1': ['sum'], 'realized_vol2': ['sum'], 'realized_vol3': ['sum'], 'realized_vol4': ['sum']}
    return agg(df, feature_dict)
def extract_trade_stats(df):
    feature_dict = {'realized_vol_trade': ['sum'], 'seconds_in_bucket': ['count'], 'size': ['sum', 'max', 'min'], 'order_count': ['sum', 'max'], 'amount': ['sum', 'max', 'min']}
    return agg(df, feature_dict)
def extract_trade_stats_time(df):
    feature_dict = {'realized_vol_trade': ['sum'], 'seconds_in_bucket': ['count'], 'size': ['sum'], 'amount': ['sum'], 'order_count': ['sum']}
    return agg(df, feature_dict)
def time_constraint_fe(df, stats_df, last_sec, fe_function, cols):
    sub_df = df[df['seconds_in_bucket'] >= last_sec].reset_index(drop=True)
    if sub_df.shape[0] > 0:
        sub_stats = fe_function(sub_df)
    else:
        sub_stats = cudf.DataFrame(columns=cols)
    return stats_df.merge(sub_stats, on='time_id', how='left', suffixes=('', f'_{last_sec}'))
def feature_engineering(book_path, trade_path):
    book_df = cudf.read_parquet(book_path)
    book_df = extract_raw_book_features(book_df)
    book_stats = extract_book_stats(book_df)
    book_cols_time = ['realized_vol1_sum', 'realized_vol2_sum', 'realized_vol3_sum', 'realized_vol4_sum'] + ['time_id']
    trade_df = cudf.read_parquet(trade_path)
    trade_df = extract_raw_trade_features(trade_df)
    trade_stats = extract_trade_stats(trade_df)
    trade_cols_time = ['realized_vol_trade_sum', 'seconds_in_bucket_count', 'size_sum', 'order_count_sum', 'amount_sum'] + ['time_id']
    for last_sec in [100, 200, 300, 400, 500]:
        book_stats = time_constraint_fe(book_df, book_stats, last_sec, extract_book_stats_time, book_cols_time)
        trade_stats = time_constraint_fe(trade_df, trade_stats, last_sec, extract_trade_stats_time, trade_cols_time)
    return book_stats.merge(trade_stats, on='time_id', how='left')
def process_data(order_book_paths, trade_paths, stock_ids):
    stock_dfs = []
    for book_path, trade_path in tqdm(list(zip(order_book_paths, trade_paths))):
        stock_id = int(book_path.split('=')[1].split('/')[0])
        df = feature_engineering(book_path, trade_path)
        df['stock_id'] = stock_id
        stock_dfs.append(df)
    return cudf.concat(stock_dfs)

train = process_data(order_book_training, trades_training, stock_ids)
test = process_data(order_book_test, trades_test, stock_ids)
(train.shape, test.shape)

def stock_time_fe(df):
    cols = ['realized_vol1_sum', 'realized_vol2_sum', 'realized_vol1_sum_200', 'realized_vol2_sum_200', 'realized_vol1_sum_300', 'realized_vol2_sum_300', 'realized_vol1_sum_400', 'realized_vol2_sum_400', 'realized_vol_trade_sum_200', 'realized_vol_trade_sum_300', 'realized_vol_trade_sum_400', 'realized_vol_trade_sum']
    tmp_df = df[~df['is_test']]
    for agg_col in ['stock_id', 'time_id']:
        for agg_func in ['mean', 'max', 'std', 'min']:
            agg_df = tmp_df.groupby(agg_col)[cols].agg(agg_func)
            agg_df.columns = [f'{agg_col}_{agg_func}_{col}' for col in agg_df.columns]
            df = df.merge(agg_df.reset_index(), on=agg_col, how='left')
    return df
train['is_test'] = False
test['is_test'] = True
all_df = train.append(test).reset_index(drop=True)
all_df = stock_time_fe(all_df)
train = all_df[~all_df['is_test']]
test = all_df[all_df['is_test']].to_pandas()

train = dev_df.merge(train, on=['stock_id', 'time_id'], how='left').to_pandas()

num_features = [col for col in list(train.columns) if col not in {'stock_id', 'time_id', 'target', 'is_test'}]
len(num_features)