{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.10.14","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"gpu","dataSources":[{"sourceId":84493,"databundleVersionId":9871156,"sourceType":"competition"}],"dockerImageVersionId":30805,"isInternetEnabled":false,"language":"python","sourceType":"notebook","isGpuEnabled":true}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"markdown","source":"# Introduction \n\nThis is the final notebook in my odyssey to create a LGBM Online Learning approach.\n\nRelated previous work:\n- [JRTSMDF - ARIMA online Learning](https://www.kaggle.com/code/serjhenrique/jrtsmdf-arima-online-learning)\n- [Blog post explaining ARIMA Solution in detail](https://serjhenrique.com/arima-and-online-learning-in-financial-forecasting/)\n- [JSRTMDF - LGBM Online Learning Evaluation](https://www.kaggle.com/code/serjhenrique/jsrtmdf-lgbm-online-learning-evaluation-pb)\n\nI find it easy to experiment with this notebook, and additionally, I created a section to \"debug\" the submission process by simulating it with training data.\n\n**If you find it useful, please, upvote. I hope you enjoy!!**","metadata":{}},{"cell_type":"code","source":"import pandas as pd\nimport polars as pl\nimport polars.selectors as cs\nimport numpy as np\nimport lightgbm as lgb\nimport time\nimport os\n\n\nfrom typing import List, Union\nimport gc\n\nimport kaggle_evaluation.jane_street_inference_server","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true,"execution":{"iopub.status.busy":"2024-12-23T11:02:25.211090Z","iopub.execute_input":"2024-12-23T11:02:25.211420Z","iopub.status.idle":"2024-12-23T11:02:29.266202Z","shell.execute_reply.started":"2024-12-23T11:02:25.211379Z","shell.execute_reply":"2024-12-23T11:02:29.265463Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"class FeatureEngineering:\n\n    def __init__(\n        self, \n        data: Union[pl.DataFrame, pl.LazyFrame], \n        lags: Union[pl.DataFrame, pl.LazyFrame, None] = None\n    ):\n        # Check if data is LazyFrame or DataFrame\n        if isinstance(data, pl.DataFrame):\n            self.data = data.lazy()  # Convert to LazyFrame if it's a DataFrame\n        else:\n            self.data = data  # Keep as LazyFrame\n\n        # Check if lags is LazyFrame or DataFrame\n        if isinstance(lags, pl.DataFrame):\n            self.lags = lags.lazy()  # Convert to LazyFrame if it's a DataFrame\n        else:\n            self.lags = lags  # Keep as LazyFrame\n\n    def add_lag_responder(self, data: pl.LazyFrame, columns: List[str]) -> pl.LazyFrame:\n        if self.lags is None:\n            last_responder_df = data.group_by(['symbol_id','date_id']).agg(\n                (pl.col(c).last().cast(pl.Float32).alias(f'{c}_lag_1') for c in columns)\n            )\n            last_responder_df = last_responder_df.with_columns(\n                pl.col('date_id') + 1\n            )\n        else:\n            last_responder_df = self.lags.group_by(['symbol_id','date_id']).agg(\n                (pl.col(f'{c}_lag_1').last().cast(pl.Float32).alias(f'{c}_lag_1') for c in columns)\n            )\n\n        data = data.join(\n            last_responder_df,\n            on=['symbol_id','date_id'],\n            how='left'\n        ) \n\n        return data\n\n    def drop_responders(self, data: pl.LazyFrame) -> pl.LazyFrame:\n        columns_to_drop = [f'responder_{i}' for i in range(9) if i != 6]\n        return data.drop([c for c in columns_to_drop if c in data.collect_schema().names()])\n\n    def drop_partition_id(self, data: pl.LazyFrame) -> pl.LazyFrame:\n        return data.drop(['partition_id']) if 'partition_id' in data.collect_schema().names() else data\n\n    def differencing_transform(self, data: pl.LazyFrame, cols: List[str]) -> pl.LazyFrame:\n\n        return (\n            data.sort([\"symbol_id\", \"date_id\", \"time_id\"])\n            .with_columns(\n                [\n                    (pl.col(col).diff(1))\n                    .over(\"symbol_id\")\n                    .alias(col)\n                    for col in cols\n                ]\n            )\n        )\n\n    def create_rolling_features(self, data: pl.LazyFrame, period: int, group_col: Union[str, List[str]], agg_col: Union[str, List[str]]) -> pl.LazyFrame:\n\n        rolling_df = data.with_columns(\n            [\n                pl.col(c)\n                .rolling_mean(window_size=period, min_periods=1)\n                .over(group_col)\n                .cast(pl.Float32)\n                .alias(f\"{c}_rolling_{str(period)}_mean\")\n                for c in (agg_col if isinstance(agg_col, list) else [agg_col])\n            ]\n        )\n\n        return rolling_df\n\n    def create_daily_rolling_features(self, data: pl.LazyFrame, period: int , group_col: Union[str, List[str]], agg_col: Union[str, List[str]]) -> pl.LazyFrame:\n\n        daily_df = data.select(group_col+agg_col).unique().sort(group_col)\n\n        rolling_df = daily_df.with_columns(\n            [\n                pl.col(c)\n                .rolling_mean(window_size=period, min_periods=0)\n                .over('symbol_id')\n                .cast(pl.Float32)\n                .alias(f\"{c}_rolling_{str(period)}_mean\")\n                for c in (agg_col if isinstance(agg_col, list) else [agg_col])\n            ]\n        ).select(group_col + [ f\"{c}_rolling_{str(period)}_mean\" for c in (agg_col if isinstance(agg_col, list) else [agg_col]) ] )\n\n        data = data.join(\n            rolling_df,\n            on=group_col,\n            how='left'\n        ) \n\n        return data\n\n    def create_sin_features(self, data: pl.LazyFrame) -> pl.LazyFrame:\n        return data.with_columns(\n            (pl.col('date_id') * (2 * np.pi / 5)).sin().cast(pl.Float32).alias('date_id_5_sin_feature'),\n            (pl.col('date_id') * (2 * np.pi / 21)).sin().cast(pl.Float32).alias('date_id_21_sin_feature'),\n             pl.when(pl.col('date_id') < 677)\n            .then((pl.col('time_id') * (2 * np.pi / 849)).sin().cast(pl.Float32))\n            .otherwise((pl.col('time_id') * (2 * np.pi / 968)).sin().cast(pl.Float32))\n            .alias('time_id_sin_feature')\n        )\n\n    def fill_null(self, data: pl.LazyFrame) -> pl.LazyFrame:\n        return data.fill_null(strategy=\"zero\") \n\n    def filter_from_date(self, data: pl.LazyFrame, day: int) -> pl.LazyFrame:\n        return data.filter(\n            (pl.col('date_id') >= day)\n        )\n    \n\n    def run(self, is_train=True):\n        \n        if is_train:\n            result = self.data.pipe(\n                self.fill_null\n            ).pipe(\n                self.add_lag_responder, [f'responder_{i}' for i in range(9)]\n            ).pipe(\n                self.drop_responders\n            ).pipe(\n                self.drop_partition_id\n            ).pipe(\n                self.create_sin_features\n            )\n        else:\n            result = self.data.pipe(\n                self.fill_null \n            ).pipe(\n                self.add_lag_responder, [f'responder_{i}' for i in range(9)]\n            ).pipe(\n                self.drop_partition_id\n            ).pipe(\n                self.create_sin_features\n            )\n        \n        return result if isinstance(result, pl.DataFrame) else result.collect()","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-23T11:02:29.267878Z","iopub.execute_input":"2024-12-23T11:02:29.268317Z","iopub.status.idle":"2024-12-23T11:02:29.286058Z","shell.execute_reply.started":"2024-12-23T11:02:29.268291Z","shell.execute_reply":"2024-12-23T11:02:29.285237Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"class LGBMTrainer():\n    def __init__(\n        self, \n        data: Union[pl.LazyFrame, pl.DataFrame],\n        y: str,\n        train_window: int,\n        forecast_window: int,\n        lgbm_params: dict\n    ):\n        # Check if data is LazyFrame or DataFrame\n        if isinstance(data, pl.DataFrame):\n            self.data = data.lazy()  # Convert to LazyFrame if it's a DataFrame\n        else:\n            self.data = data  # Keep as LazyFrame\n        self.exog = []\n        self.y = y\n        self.train_window = train_window\n        self.forecast_window = forecast_window\n        self.lgbm_params = lgbm_params\n\n    def get_last_n_dates_per_symbol(self, data: pl.LazyFrame, train_size: int) -> pl.LazyFrame:\n        # Get unique dates per symbol\n        unique_dates = (\n            data\n            .select(['symbol_id', 'date_id'])\n            .unique()\n        )\n        \n        # Sort and get top N dates per symbol\n        top_dates = (\n            unique_dates\n            .sort(['symbol_id', 'date_id'], descending=[False, True])\n            .group_by('symbol_id')\n            .head(train_size)\n        )\n        \n        # Filter original data using these dates\n        final_result = (\n            data\n            .join(\n                top_dates,\n                on=['symbol_id', 'date_id'],\n                how='inner'\n            )\n            .sort(['symbol_id', 'date_id', 'time_id'])\n        )\n        \n        return final_result\n\n    def _calculate_r2(self, y_true, y_pred, weights):\n        \"\"\"\n        Calculate the sample weighted zero-mean R-squared score (R2).\n    \n        Parameters:\n        - y_true (pd.Series or np.array): Ground truth values.\n        - y_pred (pd.Series or np.array): Predicted values.\n        - weights (pd.Series or np.array): Sample weights.\n    \n        Returns:\n        - float: R2 score.\n        \"\"\"\n        numerator = np.sum(weights * (y_true - y_pred) ** 2)\n        denominator = np.sum(weights * (y_true ** 2))\n        r2_score = 1 - (numerator / denominator)\n        return r2_score\n\n    def train(self, train_df: pl.DataFrame, forecast_df: Union[pl.DataFrame,None]=None) -> lgb.Booster:\n        \n        self.exog = [c for c in train_df.columns if c not in ['date_id','row_id','weight','responder_6']]\n\n        train_ds = lgb.Dataset(\n            data = train_df[self.exog],\n            label = train_df[self.y].to_numpy()\n        )\n\n        if isinstance(forecast_df, pl.DataFrame):\n\n            forecast_ds = lgb.Dataset(\n                data = forecast_df[self.exog],\n                label = forecast_df[self.y].to_numpy()\n            )\n    \n            booster = lgb.train(\n                params = self.lgbm_params,\n                train_set = train_ds,\n                valid_sets = [forecast_ds],\n                callbacks = [\n                    #lgb.log_evaluation(20), \n                    lgb.early_stopping(20),\n                ]\n            )\n        else:\n\n            booster = lgb.train(\n                params = self.lgbm_params,\n                train_set = train_ds\n            )\n\n        return booster\n        \n\n    def evaluate(self):\n        unique_date_ids = data.select('date_id').unique().sort('date_id').collect().to_numpy().flatten()\n\n        step = self.train_window + self.forecast_window\n        metric_list = []\n        r2_score_list = []\n        time_list = []\n        for i in range(0, len(unique_date_ids) - step):\n            start_time = time.perf_counter()\n            \n            forecast_dates = unique_date_ids[i+self.train_window:i+step]\n\n            forecast_df = self.data.filter(pl.col('date_id').is_in(forecast_dates))\n            train_df = self.data.pipe(\n                lambda x: x.filter(pl.col('date_id') < forecast_dates[0])\n            ).pipe(\n                self.get_last_n_dates_per_symbol, self.train_window\n            )\n\n            transform = FeatureEngineering(\n                train_df.collect()\n            )\n            train_df = transform.run()\n\n            transform = FeatureEngineering(\n                forecast_df.collect()\n            )\n            forecast_df = transform.run()\n            \n            booster = self.train(train_df, forecast_df)\n\n            metric = booster.best_score['valid_0']['rmse']\n\n            y_valid_pred = booster.predict(forecast_df[self.exog])\n            r2_score = self._calculate_r2(forecast_df[self.y].to_numpy(), y_valid_pred, forecast_df['weight'].to_numpy()) \n\n            metric_list.append(metric)\n            r2_score_list.append(r2_score)\n\n            end_time = time.perf_counter()\n            delta_time = end_time - start_time\n            time_list.append(delta_time)\n            print(f\"Elapsed time: {delta_time:.6f} seconds. Metric: {metric}. R2 Score: {r2_score}\")\n        \n\n        return metric_list, r2_score_list, time_list\n\n    def run(self):\n\n        train_df = self.get_last_n_dates_per_symbol(self.data, self.train_window).collect()\n        booster = self.train(train_df)\n\n        return booster\n        ","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-23T11:02:29.287379Z","iopub.execute_input":"2024-12-23T11:02:29.287704Z","iopub.status.idle":"2024-12-23T11:02:29.302864Z","shell.execute_reply.started":"2024-12-23T11:02:29.287667Z","shell.execute_reply":"2024-12-23T11:02:29.301969Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"class JanePredictor():\n    def __init__(\n        self,\n        initial_data: pl.DataFrame,\n        y: str,\n        train_window: int,\n        forecast_window: int,\n        lgbm_params: dict\n    ):\n        \"\"\"\n        Initializes the JanePredictor class with initial data and training size.\n\n        Parameters:\n        initial_data (pl.DataFrame): The initial dataset to be used for predictions.\n        train_size (int): The size of the training dataset.\n\n        Attributes:\n        lags_ (None): Placeholder for lag data.\n        cached_test_data (pl.DataFrame): Stores the initial data for caching purposes.\n        train_size (int): Stores the size of the training data.\n        \"\"\"\n        self.lags = None\n        self.exog = []\n        self.booster = None\n        self.cached_test_data = initial_data\n        self.train_size = train_size\n\n        self.y = y\n        self.train_window = train_window\n        self.forecast_window = forecast_window\n        self.lgbm_params = lgbm_params\n\n        transform = FeatureEngineering(\n            data = self.cached_test_data,\n            lags = self.lags\n        )\n        self.cached_test_data = transform.run()\n        self.exog = [c for c in self.cached_test_data.columns if c not in ['date_id','row_id','weight','responder_6']]\n\n    \n    def predict(self, test: pl.DataFrame, lags: pl.DataFrame | None) -> pl.DataFrame | pd.DataFrame:\n        \"\"\"\n        Predicts future values based on the test data and optional lag data.\n\n        Parameters:\n        test (pl.DataFrame): The test dataset containing new observations.\n        lags (pl.DataFrame | None): Optional lagged data to enhance prediction accuracy.\n\n        Returns:\n        pl.DataFrame | pd.DataFrame: A DataFrame containing predictions with 'row_id' and 'responder_6' columns.\n\n        Raises:\n        TypeError: If the returned predictions are not a DataFrame.\n        \n        Notes:\n        - Aligns column types between cached data and lag data.\n        - Joins lag data with cached test data to fill missing values.\n        - Uses an ARIMA model to generate predictions.\n        - Ensures that the output DataFrame has the same number of rows as the input test data.\n        \"\"\"\n        self.cached_test_data = self.cached_test_data.with_columns([\n            pl.col(c).cast(pl.Float32) for c in self.cached_test_data.columns if c not in ['date_id','time_id','symbol_id','row_id']\n        ])\n\n        test = test.with_columns([\n            pl.col(c).cast(pl.Float32) for c in test.columns if c not in ['date_id','time_id','symbol_id','row_id', 'is_score']\n        ])\n        test = test.with_columns([\n            pl.col(\"date_id\").cast(pl.Int16),\n            pl.col(\"time_id\").cast(pl.Int16),\n            pl.col(\"symbol_id\").cast(pl.Int8)\n        ])\n\n        if lags is not None:\n            self.lags = lags\n\n            '''\n            convert data types\n            '''\n            self.cached_test_data = self.cached_test_data.with_columns([\n                pl.col(\"date_id\").cast(pl.Int16),\n                pl.col(\"time_id\").cast(pl.Int16),\n                pl.col(\"symbol_id\").cast(pl.Int8)\n            ])\n            \n            self.lags = self.lags.with_columns([\n                pl.col(\"date_id\").cast(pl.Int16),\n                pl.col(\"time_id\").cast(pl.Int16),\n                pl.col(\"symbol_id\").cast(pl.Int8)\n            ])\n\n            '''\n            fill responders from previous cached data\n            '''\n            \n            last_responder_df = self.lags.group_by(['symbol_id','date_id']).agg(\n                (pl.col(f'{c}_lag_1').last().cast(pl.Float32).alias(f'{c}_lag_1') for c in [f'responder_{i}' for i in range(9)])\n            )\n\n            self.cached_test_data = self.cached_test_data.join(\n                last_responder_df,\n                how='left',\n                on=['date_id','symbol_id'],\n                suffix= '_new'\n            )\n\n            for i in range(9):\n                c = f\"responder_{i}_lag_1\"\n                self.cached_test_data = self.cached_test_data.with_columns(\n                    pl.when(pl.col(f\"{c}_new\").is_null())\n                    .then(pl.col(c))\n                    .otherwise(pl.col(f\"{c}_new\"))\n                    .alias(c)\n                )\n\n            self.cached_test_data = self.cached_test_data.drop([f'responder_{i}_lag_1_new' for i in range(9)])\n\n            # update responder_6 with ground truth\n            responder_6_fix_df = self.lags.select(\n                [\"date_id\", \"symbol_id\", \"time_id\", \"responder_6_lag_1\"]\n            ).rename({\"responder_6_lag_1\": \"responder_6_new\"})\n            \n            # Subtract 1 from the 'date_id' column\n            responder_6_fix_df = responder_6_fix_df.with_columns(\n                (pl.col(\"date_id\") - 1).alias(\"date_id\")\n            )\n        \n            self.cached_test_data = self.cached_test_data.join(\n                responder_6_fix_df,\n                on=[\"date_id\", \"symbol_id\", \"time_id\"],\n                how=\"left\"\n            )\n\n            self.cached_test_data = self.cached_test_data.with_columns(\n                pl.when(pl.col(\"responder_6_new\").is_null())\n                .then(pl.col(\"responder_6\"))\n                .otherwise(pl.col(\"responder_6_new\"))\n                .alias(\"responder_6\")\n            )\n            self.cached_test_data = self.cached_test_data.drop(\"responder_6_new\")\n            \n            trainer = LGBMTrainer(\n                data = self.cached_test_data,\n                y = self.y,\n                train_window = self.train_window,\n                forecast_window = self.forecast_window,\n                lgbm_params = self.lgbm_params\n            )\n\n            self.booster = trainer.run()\n        \n        transform = FeatureEngineering(\n            data = test,\n            lags = self.lags\n        ) \n        test = transform.run(is_train=False)\n\n        self.cached_test_data = pl.concat([\n            self.cached_test_data,\n            test.with_columns(pl.lit(0.0).cast(pl.Float32).alias('responder_6')).select(self.cached_test_data.columns)\n        ], how='vertical_relaxed')\n\n        pred = self.booster.predict(test[self.exog])\n\n        predictions = test.select(\n            'row_id',\n        ).with_columns(\n            pl.Series(name=\"responder_6\", values=pred).cast(pl.Float32)\n        )\n    \n        if isinstance(predictions, pl.DataFrame):\n            assert predictions.columns == ['row_id', 'responder_6']\n        elif isinstance(predictions, pd.DataFrame):\n            assert (predictions.columns == ['row_id', 'responder_6']).all()\n        else:\n            raise TypeError('The predict function must return a DataFrame')\n        \n        # Confirm has as many rows as the test data.\n        assert len(predictions) == len(test)\n\n        print(predictions.head())\n    \n        return predictions","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-23T11:02:29.304372Z","iopub.execute_input":"2024-12-23T11:02:29.304768Z","iopub.status.idle":"2024-12-23T11:02:29.323119Z","shell.execute_reply.started":"2024-12-23T11:02:29.304715Z","shell.execute_reply":"2024-12-23T11:02:29.322308Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"def get_last_n_dates_per_symbol(file_path: str, train_size: int) -> pl.DataFrame:\n    \"\"\"\n    Load data from parquet file and return the last n dates for each symbol_id\n    \n    Args:\n        file_path: Path to the parquet file\n        train_size: Number of distinct dates to keep per symbol\n        \n    Returns:\n        pl.DataFrame: Filtered dataframe with last n dates per symbol\n    \"\"\"\n    # Load data\n    data = pl.scan_parquet(file_path)\n    data = data.filter(pl.col('date_id') > 1000)\n    \n    # Get unique dates per symbol\n    unique_dates = (\n        data\n        .select(['symbol_id', 'date_id'])\n        .unique()\n    )\n    \n    # Sort and get top N dates per symbol\n    top_dates = (\n        unique_dates\n        .sort(['symbol_id', 'date_id'], descending=[False, True])\n        .group_by('symbol_id')\n        .head(train_size)\n    )\n    \n    # Filter original data using these dates\n    final_result = (\n        data\n        .join(\n            top_dates,\n            on=['symbol_id', 'date_id'],\n            how='inner'\n        )\n        .sort(['symbol_id', 'date_id', 'time_id'])\n        .collect()\n    )\n    \n    # Validate that each symbol has exactly train_size distinct dates\n    date_counts = final_result.group_by(\"symbol_id\").agg(\n        pl.col(\"date_id\").n_unique().alias(\"distinct_date_count\")\n    )\n    \n    assert (date_counts['distinct_date_count'] == train_size).all(), (\n        f\"Not all symbols have exactly {train_size} distinct dates. \"\n        f\"Found counts: {date_counts.filter(pl.col('distinct_date_count') != train_size)}\"\n    )\n    \n    return final_result","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-23T11:02:29.324209Z","iopub.execute_input":"2024-12-23T11:02:29.324508Z","iopub.status.idle":"2024-12-23T11:02:29.335526Z","shell.execute_reply.started":"2024-12-23T11:02:29.324483Z","shell.execute_reply":"2024-12-23T11:02:29.334641Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"%%time \nfile_path = '/kaggle/input/jane-street-real-time-market-data-forecasting/train.parquet'\ntrain_size = 84\ndata = get_last_n_dates_per_symbol(file_path, train_size)","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-23T11:02:29.337118Z","iopub.execute_input":"2024-12-23T11:02:29.337359Z","iopub.status.idle":"2024-12-23T11:03:01.068927Z","shell.execute_reply.started":"2024-12-23T11:02:29.337336Z","shell.execute_reply":"2024-12-23T11:03:01.068032Z"}},"outputs":[],"execution_count":null},{"cell_type":"code","source":"lgbm_params = {  \n    'verbose':-1,\n    'boosting_type': 'gbdt',\n    'objective': 'regression_l2',\n    'metric': 'rmse',\n    'n_estimators': 150,  \n    'learning_rate': 0.022167654221948718, \n    'max_depth': 5, \n    'num_leaves': 42, \n    'min_child_samples': 15, \n    'subsample': 0.8658662737679551, \n    'colsample_bytree': 0.6316624163538186, \n    'reg_alpha': 7.313049180654483, \n    'reg_lambda': 8.113493177057705,\n    'random_state': 42,\n    'device': 'gpu'\n}\n\njane_predictor = JanePredictor(\n    initial_data = data,\n    y = 'responder_6',\n    train_window = train_size,\n    forecast_window = 1,\n    lgbm_params = lgbm_params\n)\n\ninference_server = kaggle_evaluation.jane_street_inference_server.JSInferenceServer(jane_predictor.predict)\n\nif os.getenv('KAGGLE_IS_COMPETITION_RERUN'):\n    inference_server.serve()\nelse:\n    inference_server.run_local_gateway(\n        (\n            '/kaggle/input/jane-street-real-time-market-data-forecasting/test.parquet',\n            '/kaggle/input/jane-street-real-time-market-data-forecasting/lags.parquet',\n        )\n    )","metadata":{"trusted":true,"execution":{"iopub.status.busy":"2024-12-23T11:03:01.070162Z","iopub.execute_input":"2024-12-23T11:03:01.070430Z","iopub.status.idle":"2024-12-23T11:03:41.248985Z","shell.execute_reply.started":"2024-12-23T11:03:01.070404Z","shell.execute_reply":"2024-12-23T11:03:41.247519Z"}},"outputs":[],"execution_count":null},{"cell_type":"markdown","source":"# Debug","metadata":{}},{"cell_type":"markdown","source":"import time\n\ntest_data = pl.scan_parquet('/kaggle/input/jane-street-real-time-market-data-forecasting/train.parquet')\ntest_data = test_data.filter(\n    pl.col('date_id') > 1572\n).collect()\nlag_data = pl.read_parquet('/kaggle/input/jane-street-real-time-market-data-forecasting/lags.parquet')\n\nlag_data_columns = {\n    'date_id':'date_id',\n    'time_id':'time_id',\n    'symbol_id':'symbol_id',\n    'responder_0':'responder_0_lag_1',\n    'responder_1':'responder_1_lag_1',\n    'responder_2':'responder_2_lag_1',\n    'responder_3':'responder_3_lag_1',\n    'responder_4':'responder_4_lag_1',\n    'responder_5':'responder_5_lag_1',\n    'responder_6':'responder_6_lag_1',\n    'responder_7':'responder_7_lag_1',\n    'responder_8':'responder_8_lag_1'\n}\n\njane_predictor = JanePredictor(\n    initial_data = data,\n    y = 'responder_6',\n    train_window = train_size,\n    forecast_window = 1,\n    lgbm_params = lgbm_params\n)\n\ni = 0\ntime_list = []\nfor row in test_data.select(['date_id','time_id']).unique().sort(['date_id','time_id']).iter_rows(named=True):\n    \n    print(row['date_id'], row['time_id'])\n    \n    test = test_data.filter(\n        ( pl.col('date_id') == row['date_id'] ) &\n        ( pl.col('time_id') == row['time_id'] )\n    ).with_row_count(\"row_id\")\n    \n    if row['time_id'] == 0:\n        lag = test.select(\n            list( lag_data_columns.keys() )\n        ).rename(lag_data_columns)\n        \n        start_time = time.perf_counter()\n        jane_predictor.predict(test, lag)\n        end_time = time.perf_counter()\n        delta_time = delta_time = end_time - start_time\n        time_list.append(delta_time)\n        print(f\"Elapsed time: {delta_time:.6f} seconds\")\n\n    else:\n        start_time = time.perf_counter()\n        jane_predictor.predict(test, None)\n        end_time = time.perf_counter()\n        delta_time = delta_time = end_time - start_time\n        time_list.append(delta_time)\n        print(f\"Elapsed time: {delta_time:.6f} seconds\")","metadata":{"execution":{"iopub.status.busy":"2024-12-06T10:38:31.965174Z","iopub.execute_input":"2024-12-06T10:38:31.965602Z","iopub.status.idle":"2024-12-06T10:38:47.604156Z","shell.execute_reply.started":"2024-12-06T10:38:31.965565Z","shell.execute_reply":"2024-12-06T10:38:47.602913Z"}}},{"cell_type":"code","source":"","metadata":{"trusted":true},"outputs":[],"execution_count":null}]}