{"cells":[{"metadata":{"trusted":true},"cell_type":"code","source":"from functools import reduce\nimport gc\nimport numpy as np\nimport pandas as pd\nfrom pathlib import Path\nimport sys\nimport time","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"## Summary\nInspired by [Simulating Pipeline for Test Data](https://www.kaggle.com/abdurrafae/simulating-pipeline-for-test-data), this notebook creates a validation set generator that serves batches similarly to the way the competition testing api serves them:\n* Each batch includes interactions from only one `task_container_id` for each `user_id`\n* For each `user_id`, `timestamps` increase monotonically and are greater than the latest `timestamp` in the training set\n* A maximum of 1,000 interactions per batch\n\n\nThe code that ensures that only one `task_container_id` per `user_id` is included in each batch is slow (averaging ~5 seconds per batch with a total validation set size of approximately 2.5 million) and may not have any discernible benefit with respect to cross validation. I noted in the [discussion formus ](https://www.kaggle.com/c/riiid-test-answer-prediction/discussion/192919#1058744) that some competitors at the top of the leaderboard are utlizing a simpler approach. The [competition api](https://www.kaggle.com/c/riiid-test-answer-prediction/data) says it gets through 2.5 million records in roughly 15 minutes, so I'm obviously way off the mark. It would be much faster if the `task_container_id` logic was elimniated and reworked to instead include only one `row_id` per `user_id`, maintaining the other constraints, but it would be nice to have something fast that replicated the test api as closely as possible.\n\nIt could be faster using [cudf](https://docs.rapids.ai/api/cudf/stable/), but it may be better just to fade back to the simpler `user_id` logic. I'm sure there are some obvious ways to speed things up that I'm missing - I'd be appreciative of any suggestions."},{"metadata":{},"cell_type":"markdown","source":"## Read train.csv"},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\n\ndtypes = {'row_id': 'int64',\n          'timestamp': 'int64',\n          'user_id': 'int32',\n          'content_id': 'int16',\n          'content_type_id': 'int8',\n          'task_container_id': 'int16',\n          'user_answer': 'int8',\n          'answered_correctly': 'int8',\n          'prior_question_elapsed_time': 'float32', \n          'prior_question_had_explanation': 'boolean'}\n\ndf_train = pd.read_feather('../input/riiid-train-data-multiple-formats/riiid_train.feather')\n\nfor c,t in dtypes.items():\n    df_train[c] = df_train[c].astype(t)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\n# ensure task_container_id increases monotonically per \n# https://www.kaggle.com/c/riiid-test-answer-prediction/discussion/189465\n\nif True:\n    df_train.task_container_id = (df_train.groupby('user_id')['task_container_id']\n                                  .transform(lambda x: pd.factorize(x)[0])\n                                  .astype('int16'))","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"## Create Validation Set"},{"metadata":{"trusted":true},"cell_type":"code","source":"np.random.seed(42)\n\n# get unique user_ids\nuser_ids = df_train.user_id.unique()\n\n# choose random set of user ids\nuser_ids_valid = np.random.choice(user_ids, int(0.1 * len(user_ids)))\n\n# filter training set to include only records with chosen user_ids\ndf_user_valid = df_train[df_train.user_id.isin(user_ids_valid)][['user_id', 'task_container_id', 'row_id']]\n\n# get unique user_id-task_container_id combinations from records of chosen user_ids\ndf_user_task_valid = df_user_valid.groupby(['user_id', 'task_container_id']).head(1).reset_index()\n\n# get index of trailing number of unique user_id-task_container_id combinations\nindex_valid = df_user_task_valid.groupby('user_id').tail(100).set_index(['user_id', 'task_container_id']).index\n\n# use index to get ids of all rows in the chosen set of user_id-task_container combinations\nrow_valid = df_train.set_index(['user_id', 'task_container_id'])['row_id'].loc[index_valid].values\n\n# get train row_ids using inverse of valid row_ids\nrow_train = df_train.row_id[np.isin(df_train.row_id, row_valid, invert=True)].values\n\n# create train index of unique user_id-task_container_id combinations\nindex_train = (df_train[['user_id', 'task_container_id']].iloc[row_train]\n               .set_index(['user_id', 'task_container_id']).index)\n\n# compare valid and train indices to make sure they are mutually exclusive \n# for index_train_split in np.array_split(index_train, 20):\n#     assert index_valid.isin(index_train_split).sum() == 0\n\n# del index_train_split\n# gc.collect()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"print(f'Validation set selected from {len(df_user_valid):,d} interactions \\\nby {len(user_ids_valid):,d} users in \\n {len(df_user_task_valid):,d} \\\nunique user_id-task_container_id combinations.\\n')\n\nprint(f'Validation set includes {len(row_valid):,d} interactions \\\nin \\n {len(index_valid):,d} unique user_id-task_container_id combinations.')","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"df_valid = df_train.loc[row_valid, ['user_id', 'task_container_id', 'row_id']].set_index(['user_id', 'task_container_id'])","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# quick memory check to see what the big variables are\n\nif True:\n    local_vars = list(locals().items())\n    for var, obj in local_vars:\n        size = sys.getsizeof(obj)\n        if size > 1e7:\n            print(f'{var:<18}{size/1e6:>10,.1f} MB')","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"## Create Validation Batch Generator"},{"metadata":{"trusted":true},"cell_type":"code","source":"def valid_batches(df_valid=df_valid, verbose=True, tests=True):\n \n    while len(df_valid):\n        if verbose:\n            print(f'Count of row_ids in validation set before selecting batch: {len(df_valid):,d}')\n\n        # set number of users to include in batch\n        n_users = np.random.randint(1000)\n\n        # get remaining user_ids\n        users_unique = df_valid.index.unique(level='user_id').to_numpy()\n\n        # shuffle user_ids\n        np.random.shuffle(users_unique)\n\n        # get index of first user_id-task_container_id for each user\n        df_valid_g = df_valid.loc[users_unique[:n_users]].reset_index().groupby(['user_id'])\n        index_valid_batch = df_valid_g.head(1).set_index(['user_id', 'task_container_id']).index.sort_values()\n        if verbose:\n            print(f'Count of unique user_id-task_container combinations to include in batch: {len(index_valid_batch):,d}')\n\n        # select row_ids to include in batch\n        df_valid_batch = df_valid.loc[index_valid_batch]\n        if verbose:\n            print(f'Count of row_ids included in batch: {len(df_valid_batch):,d}')\n\n        # drop selected row_ids from validation set\n        df_valid = df_valid.drop(index_valid_batch)\n        if verbose:\n            print(f'Count of row_ids in validation set after selecting batch: {len(df_valid):,d}')\n\n        if tests:\n            # ensure only one task_container_id for each user_id\n            assert (df_valid_batch.groupby(['user_id', 'task_container_id'])\n            .head(1).groupby('user_id').count() == 1).all().values[0]\n\n            # ensure task_container_ids in remaining valid set are greater than\n            # task_container_ids in batch for each user\n            batch_tids = df_valid_batch.groupby('user_id').head(1)\n            valid_tids_batch_userids = (df_valid.loc[df_valid_batch.index\n                                                     .get_level_values('user_id')\n                                                     .unique()].groupby('user_id')\n                                                     .head(1))\n            assert (valid_tids_batch_userids.reset_index().task_container_id > \n                    batch_tids.reset_index().task_container_id).all()\n        \n        gc.collect()\n        \n        yield df_valid_batch","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"valid_gen = valid_batches(tests=False)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\ndf_valid_batch = next(valid_gen)","execution_count":null,"outputs":[]}],"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"pygments_lexer":"ipython3","nbconvert_exporter":"python","version":"3.6.4","file_extension":".py","codemirror_mode":{"name":"ipython","version":3},"name":"python","mimetype":"text/x-python"}},"nbformat":4,"nbformat_minor":4}