{"cells":[{"metadata":{},"cell_type":"markdown","source":"### Load necessary modules and data:"},{"metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","trusted":true},"cell_type":"code","source":"# Loading modules\nimport numpy as np\nimport pandas as pd\nimport matplotlib.pyplot as plt\nimport riiideducation\nimport gc\nimport tqdm\nimport time\nimport os\nimport lightgbm as lgb\n\n!pip --quiet install ../input/python-datatable/datatable-0.11.0-cp37-cp37m-manylinux2010_x86_64.whl\n!pip install --quiet -r ../input/treelite-treelite-runtime-version-093/treelite/requirements.txt --no-index --find-links ../input/treelite-treelite-runtime-version-093/treelite\n\nfrom bitarray import bitarray\nimport datatable as dt\n\ntqdm.tqdm.pandas()\n\n%matplotlib inline\n# plt.style.use(\"dark_background\")\n\n# Loading the API\nenv = riiideducation.make_env()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# check the files we have been given\n!ls ../input/riiid-test-answer-prediction/","execution_count":null,"outputs":[]},{"metadata":{"_uuid":"d629ff2d2480ee46fbb7e2d37f6b5fab8052498a","collapsed":true,"_cell_guid":"79c7e3d0-c299-4dcb-8224-4455121ee9b0","trusted":false},"cell_type":"markdown","source":"Right now we are going to subsample the dataset instead of reading it all to the memory, let's try to understand what data we have been provided with before training on all the data given.\n\nWe need to change the dtypes of certain columns and read a subset of the entire data to be able to fit it all to the memory:"},{"metadata":{"trusted":true},"cell_type":"code","source":"dtypes = {\n    \"row_id\": \"int64\",\n    \"timestamp\": \"int64\",\n    \"user_id\": \"int32\",\n    \"content_id\": \"int16\",\n    \"content_type_id\": \"boolean\",\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}\n\ndata = pd.read_csv(\"../input/riiid-test-answer-prediction/train.csv\", dtype=dtypes, nrows=int(1e6))\n\nques = pd.read_csv('../input/riiid-test-answer-prediction/questions.csv')\nlectures = pd.read_csv(\"../input/riiid-test-answer-prediction/lectures.csv\")\nex_sub = pd.read_csv('../input/riiid-test-answer-prediction/example_sample_submission.csv')\nex_test = pd.read_csv('../input/riiid-test-answer-prediction/example_test.csv')\n\ndata.shape, ques.shape, lectures.shape, ex_sub.shape, ex_test.shape","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# returns total bytes consumed\ntemp = data.memory_usage(deep=True).sum()\n\n# bytes to MB\nprint (f\"{temp / (2**20):.2f}\", 'MB')","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"### Cutting out EDA to include only neccessary parts:"},{"metadata":{"trusted":true},"cell_type":"code","source":"# combining questions and lectures together\nql = pd.concat([ques, lectures.rename({\"lecture_id\": \"question_id\"}, axis=1)], axis=0).reset_index(drop=True)\n\n# overlap the tags from both columns\nql.tags = ql.tags.fillna(ql.tag)\n\n# custom type of for questions\nql.type_of = ql.type_of.fillna(\"question\")\n\n# for distinguishing between lectures and questions\nql[\"content_type_id\"] = ql[\"type_of\"] != 'question'\n\n# bundle id and correct ans & tags is filled with -1\n# tag is missing for 1 row -> 10033\nql = ql.fillna(-1)\n\n# drop the unneeded tag feature\nql = ql.drop(\"tag\", 1)\n\n# rename the column for easy merge\nql = ql.rename({\"question_id\": \"content_id\"}, axis=1)\n\n# convert all the tags to list from string\nql.tags = ql.tags.apply(lambda x: [int(x)] if type(x) != str else list(map(int, x.split())))\n\n# some might find reading section easier and listening tougher or vice versa\nql['listening_lvl'] = ql.part.map({1: 2, 2: 2, 3: 1, 4: 1, 5: 0, 6: 0, 7: 0})\n\n# mulitple subquestions per question?\nql['multiple_q'] = ql.part.isin([3, 4, 6, 7])\n\n# Number of tag counts\nql['tag_c'] = ql.tags.apply(len)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# we create a feature that counts the number of lectures available per tag\nlec_count = ql.loc[ql.content_type_id, 'tags'].transform(lambda x: x[0]).value_counts()\n\nql['lec_available'] = (\n    ql.loc[~ql.content_type_id, 'tags'].transform(\n        lambda x: sum([lec_count.at[i] if i in lec_count.index else 0 for i in x]))\n)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# merge just the essentials, a whole merge is on the way\ndata = data.merge(ql[['content_id', 'part', 'bundle_id', 'content_type_id']], on=['content_id', 'content_type_id'])\n\nf, ax = plt.subplots(ncols=3, figsize=(18, 5))\nql.groupby(\"part\")['tags'].apply(lambda x: x.explode().nunique()).plot(\n    kind='bar', rot=0, ax=ax[0], title='Unique Tags Per Part')\n\nql[~ql.content_type_id].groupby(\"part\")['content_id'].nunique().plot(\n    kind='bar', rot=0, ax=ax[1], title='Unique Questions Per Part')\n\nql[ql.content_type_id].groupby(\"part\")['content_id'].nunique().plot(\n    kind='bar', rot=0, ax=ax[2], title='Unique Lectures Per Part')\n\nf.suptitle(\"Part Wise Count\");","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"ql['bundle_q_count'] = ql.groupby(\"bundle_id\")['content_id'].transform('count')\nql.loc[ql.content_type_id, 'bundle_q_count'] = -1","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"from mlxtend.preprocessing import TransactionEncoder\nfrom sklearn.decomposition import PCA\nfrom sklearn.cluster import KMeans\nfrom sklearn.preprocessing import StandardScaler\n\nte = TransactionEncoder()\n\ntemp = ql[~ql.content_type_id]\n\ntemp = temp.merge(\n    data[~data.content_type_id].groupby(\"content_id\")['answered_correctly'].agg(['count', 'mean']),\n    on='content_id', how='left')\n\ntemp['mean'] = temp['mean'].fillna(0.5)\ntemp['count'] = temp['count'].fillna(0)\n\ntemp = np.hstack([\n    te.fit_transform(temp.tags.to_list()).astype(int), \n    pd.get_dummies(temp[['tag_c', 'part', 'count', 'mean']], columns=['part', 'tag_c'])\n])\n\ntemp = PCA(n_components=150).fit_transform(StandardScaler().fit_transform(temp))\n\n# assign clusters to the questions\nql.loc[~ql.content_type_id, 'cluster'] = KMeans(n_clusters=30).fit_predict(temp)\nql.loc[ql.content_type_id, 'cluster'] = -1\n\n# convert to int\nql.cluster = ql.cluster.astype(int)\n\nql.sample(5)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Another important feature we could create is the tags themselves. We see in this [notebook](https://www.kaggle.com/jsylas/riiid-lgbm-starter) that tag seems to have a higher feature importance. Let's create this feature:"},{"metadata":{"trusted":true},"cell_type":"code","source":"temp = ql.tags.progress_apply(pd.Series)\n\n# we retain just the top two tags, drop the rest\nql['tagF'] = temp[0]\nql['tagS'] = temp[1]\nql['tagT'] = temp[2]\n\n# last tag is imp since we observed it determining the bundle_id\nql['tagL'] = ql.tags.apply(lambda x: x[-1])\n\n# although there will be overlap between tagL and tagF, tagS we\n# donot remove it. As this may help the model understand\n# if the number of tags there is 1, 2 or etc\n# we also impute the missing values as 0 after incrementing \n# all the tags by 1\nql[['tagF', 'tagS', 'tagL', 'tagT']] = (ql[['tagF', 'tagS', 'tagL', 'tagT']] + 1).fillna(0)\n\nql.sample(5)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"markdown","source":"We will use this cluster information for making group wise predictions for each user. Before we proceed further, let's merge `ql` with our dataframe:"},{"metadata":{"trusted":true},"cell_type":"code","source":"data = data.drop(['part', 'bundle_id'], 1).merge(ql, on=['content_id', 'content_type_id'], how='left')\ndata.shape","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# timetaken by user since her last interaction\ndata = data.sort_values(by=['user_id', 'timestamp'])\n\ndata['response_time'] = (\n    data.groupby(\"user_id\")['timestamp']\n    .transform(lambda x: x.diff().replace(0, np.nan)\n               .fillna(method='ffill').fillna(0))\n)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"There could be a still easier approach to obtaining the response time:\n\n$timestamp - (timestamp * \\frac{task\\_container\\_id - 1}{task\\_container\\_id})$\n\nThe advantage with this approach is that it doesn't require shifting or merging operations. Let's verify their efficiency:"},{"metadata":{"trusted":true},"cell_type":"code","source":"data['res_time_avg'] = (\n    data.timestamp - \n    (data.timestamp * (data.task_container_id - 1) / data.task_container_id)\n)\n\ndata['res_time_avg'] = data['res_time_avg'].replace(np.inf, np.nan)\n\nprint (\"Corelation to response_time:\\n\", \n       data.corr()['response_time'].loc[['res_time_avg']], sep='')\n\nprint (\"\\nCorrelation to ans_correctness:\\n\", \n       data.corr()['answered_correctly'].loc[['response_time', 'res_time_avg']], sep='')\n\n(data[['timestamp', 'response_time', 'res_time_avg']]\n .sample(10).fillna(0).astype(int))","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"markdown","source":"`Prior_question_had_explanation` needs to be mapped to previous bundle and it might serve as a powerful feature. We create features:\n1. `pqet_shifted`\n2. `pqhe_shifted`"},{"metadata":{"trusted":true},"cell_type":"code","source":"# sorting to ensure groupby works as intended\ndata = data.sort_values(['user_id', 'timestamp'])\n\ndata = data.merge(\n    (data[~data.content_type_id].groupby(['user_id', 'task_container_id'])\n     [['prior_question_elapsed_time', 'prior_question_had_explanation']]\n     .mean().groupby(\"user_id\").shift(-1).reset_index()\n     .rename({\"prior_question_elapsed_time\": 'pqet_shifted', \n              'prior_question_had_explanation': \"pqhe_shifted\"}, axis=1)),\n    on=['user_id', 'task_container_id'], how='left')","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# ensure timestamp and userid are sorted\ndata = data.sort_values(by=['user_id', 'timestamp'])\n\ncut_off = (1000 * 60 * 60) # one hr\ncut_off = cut_off * 1 # seperated by an hr worth of gap, tweak it!\n\ndata['sessions'] = (\n    data.groupby(\"user_id\")['timestamp'].diff() > cut_off\n).groupby(data['user_id']).cumsum()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"data['sess_event_count'] = data.groupby([\"user_id\", 'sessions']).cumcount()\ndata[['sessions', 'sess_event_count', 'answered_correctly']].corr().style.background_gradient()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"temp = data.loc[data.user_id == np.random.choice(data[data.content_type_id]['user_id'].unique()), \n         ['timestamp', 'task_container_id', 'content_id', \n          'tags', 'type_of', 'answered_correctly']]\n\nmask = list()\nstep = 10\nfor i in temp[temp.answered_correctly == -1].index:\n    \n    tag = temp.loc[i, 'tags'][0]\n    for j in np.arange(i - step, i + step):\n        if j in temp.index and tag in temp.loc[j, 'tags']:\n            mask.append(j)\n    \nmask = np.unique(mask)\n\ndef highlight_row(x, mask):\n    df = x.copy()\n    df.loc[(df.index.isin(mask)) & (df.answered_correctly != -1), :] = 'background-color: red'\n    df.loc[df.answered_correctly == -1, :] = 'background-color: green'\n    df.loc[~df.index.isin(mask), :] = 'background-color: \"\"'\n    return df\n\nprint (temp.loc[temp['answered_correctly'] == -1, 'task_container_id'].values)\n\nfilter_mask = np.unique([[j for j in range(i-step, i+step)] for i in temp[temp.answered_correctly == -1].index])\nfilter_mask = np.intersect1d(filter_mask, temp.index)\n\ntemp.loc[filter_mask].style.apply(lambda x: highlight_row(x, mask), axis=None)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"What we observe:\n- For most cases, the tags of lectures repeat afterwards or before like we assumed it would. \n- The tags to the *left* however are the tags that match with the lectures, meaning that they might be more important.\n- A student after seeing a lecture answers corresponding questions better.\n- For those cases the lecture tags match the preceding questions, these were the same questions the user answered incorrectly.\n- There were no *tests* as in formal examinations.\n- *We need some way to encode if a person has seen lecture of that particular tag.*"},{"metadata":{},"cell_type":"markdown","source":"*More EDA at a later time*\n\n#### We now try to understand `ex_sub` & `ex_test` csv files:\n\nThese files are provided as sample for how the files produced by `env.itertest` would be. At one call, it would only give us a small batch. We need to make predictions with our models on this and submit with a `env.predict` before we can call the next batch. *This is done so as to mimic real life scenarios where the future data is not available for model training.*\n\nFrom data description:\n\n- `prior_group_responses` (string) provides all of the user_answer entries for previous group in a string representation of a list in the **first row** of the group. **All other rows in each group are null**. If you are using Python, you will likely want to call eval on the non-null rows. Some rows may be null, or empty lists.\n\n- `prior_group_answers_correct` (string) provides all the answered_correctly field for previous group, with the *same format and caveats as prior_group_responses*. Some rows may be null, or empty lists.\n\nA more thorough understanding can be obtained from this post [here](https://www.kaggle.com/c/riiid-test-answer-prediction/discussion/190430) by Alex:\n\n     Once you submitted your predictions for a group, you get the next test_df off the iterator and that immediately tells you whether you were right or not. You can use this information to improve your model before continuing with going through the test set, or you can just ignore it.\n    As you can't submit predictions for the same group twice, you can't cheat with it. It's just meant to be used for improving your prediction algorithm as you get more information, as is typical for realtime applications."},{"metadata":{"trusted":true},"cell_type":"code","source":"# example input from the API looks like:\nex_test.head()","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"- Only the first row of each group would contain the answers and scores. The rest of the rows are all null.\n- We are *NOT* provided with the `user_answer` during the predictions we are to make. We are only provided that information at the next batch along with whether the user's predictions were indeed correct. If this had not been the case, we could simply compare with the `questions.csv` and be able to perfectly predict if the users were correct ;)\n- During the prediction time we only have features such as the timestamp, question meta data and info regarding prior groups response."},{"metadata":{"trusted":true},"cell_type":"code","source":"# example prediction to the API must look like:\nex_sub.head()","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"- For submission, we only pass in row_id, predictions. Group_num although present here is not required for submission. (check 'Making Our Predictions' part)\n\nSome more insights about the time series API testframes:\n- All shapes for each batch aren't identical, each batch may have differing no of samples\n- From this discussion [here](https://www.kaggle.com/c/riiid-test-answer-prediction/discussion/190698): Every test_iter will only have one group number -> The first row is a list in str format (could be an empty list too)."},{"metadata":{},"cell_type":"markdown","source":"Also from [this](https://www.kaggle.com/dwit392/riiid-challenge-time-since-last-action-for-test) notebook it is said that time elapsed between the last interaction and current one is a good predictor of the answer correctness which would make sense since that time could be used by a student to prepare before taking up the next test. However with data scattered around and us loading only a tiny fraction of the actual dataset, creating this feature would prove really difficult (for a later time).\n\nLet's now write a function that when given testframe_1 and testframe_2 returns testframe_1 with `user_answer` and `answered_correctly` columns merged to it:"},{"metadata":{"trusted":true},"cell_type":"code","source":"def post_process(fn0, fn1):\n    '''\n    fn0 is test dataframe at time t\n    fn1 is test dataframe at time t + 1\n    \n    If however no fn1 is provided, we need to simply assign all the \n    user_answer and answered_correctly as nans. For this purpose, \n    any dataframe with first row as nan will suffice.\n    '''\n    \n    fn_processed = fn0.drop(['prior_group_answers_correct', 'prior_group_responses'], 1)\n    \n    fn_processed['answered_correctly'] = eval(fn1['prior_group_answers_correct'].iloc[0])\n    fn_processed['user_answer'] = eval(fn1['prior_group_responses'].iloc[0])\n    \n    return fn_processed","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"#### Let's now load the data from the API for the purpose of understanding it, we will disable it when we wish to make a submission:\n\nWe have saved these batches as pickle to bench mark how fast our end pipeline is. We can also use this to check if our `post_process` function works good as intended.\n\nCode to save sample_batches:\n\n```\niter_test = env.iter_test()\n\nbatches = []\nwhile True:\n    try:\n        batches.append(next(iter_test))\n        env.predict(ex_sub)\n    except StopIteration:\n        break\n```"},{"metadata":{"trusted":true},"cell_type":"code","source":"import pickle\n\nwith open(\"../input/riiid-final-model-inputs/sample-batches.pkl\", 'rb') as f:\n    batches = pickle.load(f)\n\nprint (\"Batch sizes for each test sample:\", list(map(lambda x: x[0].shape[0], batches)))","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# working good? Check: prior_group_ans correct correctly mapped\npost_process(batches[0][0], batches[1][0]).sample(5)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Post_processed frame could then be used to help our model to learn about a students behaviour better in real time during submission. And since we are able to learn from the test data as well, it's best to create an *online / incremental model* for this competition.\n\n\n#### Evaluation metric:\n\nLet's understand the evaluation metric - `roc_auc_score`. It ensures that random predictions always yeild a score of 0.5. If however our score is less that 0.5 it means that we have made some mistake in our predictions (wrongly labeled the data, model does beter than random guess). A score of 1 (or 0) means that model is absolutely perfect and makes correct predictions 100% of the time. \n\nLet's verify this with some random guesses:"},{"metadata":{"trusted":true},"cell_type":"code","source":"# Predicting constant values for the entire train dataset.\nfrom sklearn.metrics import roc_auc_score\ntemp = data.loc[~data.content_type_id, \"answered_correctly\"]\nfor value in [0, 1, 0.5, temp.mean()]:\n    print (\"At {:.2f} the score is: {:.3f}\".format(value, roc_auc_score(temp, np.full_like(temp, 0))))","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"A highly robust metric indeed. So we have to be a bit more smart in making predictions to beat this. What about random predictions *per question*?"},{"metadata":{"trusted":true},"cell_type":"code","source":"roc_auc_score(temp, np.random.rand(len(temp)))","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Does marginally (very marginally) better than previous predictions. Our next naive idea is to use per user mean accuracy as predictions but befor we do that we need to *something else*.\n\nLet's now write a function to split the data to train/val as reliably as possible to mimic the test case scenario. Further it should also function as a CV generator, given some train ids:"},{"metadata":{"trusted":true},"cell_type":"code","source":"def generate_train_val(train, train_users=None, tp=.70, vp=None, put=None, dyn_p=False, return_mask=False): \n    '''\n    What we already know:\n    - Test dataset has new users but no new questions. \n    - Test follows chornologically after train\n    \n    * Parameters *  \n    train_ids  -> The ID's completely used for training (If None, generate it using tp)\n    tp         -> train users percentage (completely used for training)\n    vp         -> Validation usage percentage completely used for val \n                  (used to make val users | partial users)\n                  If None, randomly chosen\n                 \n    put        -> timestamp threshold for partial users above which \n                  timetamp the row becomes validation dataset\n                  If None value chosen is same as `vp` or randomly \n                  chosen for each user depending on dyn_p\n                  \n    dyn_p       -> Each user may have a different threshold at which they were split.\n                   If dyn_p is set to true, it is dynamically generated for each user.\n                 \n    return_mask -> If set to true, returns a mask instead of returning a modified dataframe\n    \n    * Output *\n    Returns a new dataframe with `train` column added for val/train split\n    \n    * Note *\n    1. Partial users are those users whose data is used for training and validation\n    2. Perentage of train data is always greater than tp\n    3. Exact percentage -> tp + \n    '''   \n\n    def threshold_user(arr, put=None):\n        if put is None: # each user may have a distinct threshold\n            put = np.random.choice(np.linspace(.20, .80, 13))\n        return arr < np.quantile(arr, put)\n    \n    \n    total_users = train.user_id.unique()\n    \n    if train_users is None:\n        train_users = np.random.choice(total_users, int(len(total_users) * tp), replace=False)\n        \n    if vp is None: # validation percent \n        vp = np.random.choice(np.linspace(.20, .80, 13))\n    \n    if put is None: # threshold for timestamp cutoff\n        # if dyn_p, put is left to be None\n        if not dyn_p:\n            put = vp\n            \n    # partial users percentage\n    pp = 1 - vp\n    \n    remaining_users = np.setdiff1d(total_users, train_users)\n    val_users = np.random.choice(remaining_users, int(len(remaining_users) * vp), replace=False)\n    partial_users = np.setdiff1d(remaining_users, val_users)\n    \n    # generating the train mask\n    mask = (\n        train[train.user_id.isin(partial_users)]\n        .groupby(\"user_id\")['timestamp']\n        .transform(lambda x: threshold_user(x.values, put))\n        .reindex(train.index, fill_value=False) | \n        \n        train.user_id.isin(train_users))\n    \n    # we will tinkering with it, best to copy it \n    # beware of passing in large sized DataFrames\n    if return_mask:\n        return mask\n    else:\n        Train = train.copy()\n        Train['train'] = mask\n        return Train","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"How does the above function work? Lets see the split via a pie chart:"},{"metadata":{"trusted":true},"cell_type":"code","source":"temp = generate_train_val(data, return_mask=True)\n(temp.astype(int).value_counts()\n .plot(kind='pie', autopct=lambda x: f\"{int(x)}%\", \n       title='Train/Val Split',\n       colors=['r', 'g'], explode=[.1, .15],\n       labels=[\"Train\", \"Validation\"]));","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"We are going to be using a feature to mark whether a user has already seen the same question he is currently solving. The most memory efficient way to do that would be to use a *bitarray*. Let's aquaint ourselves a bit with this module:"},{"metadata":{"trusted":true},"cell_type":"code","source":"from bitarray import bitarray\nba = bitarray(15000, endian='little')\n\n# initialize this way\nba.setall(False)\nrepeat_c = 0\n\ntemp = data[~data.content_type_id].loc[data.user_id == np.random.choice(data.user_id.unique()), ['content_id']]\n\nfor _, c in temp['content_id'].iteritems():\n    if ba[c]:\n        print (f\"{c:<5} was already viewed by the user!\")\n        repeat_c += 1\n    else:\n        ba[c] = 1\n    \nif repeat_c == 0:\n    print (\"No repeated questions for this user\")\nelse:\n    print (f\"\\n{repeat_c} question(s) were repeated\")\n    \n# simple sanity check. This way we can perform multiple indexing\nnp.array(ba.tolist())[temp.content_id.values].all() # must equal true","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"### Data Modelling (Prototype Version):\n\n##### Let's create a few numba helper functions! Its from here that we start defining useful numba functions. These functions would be useful when we generate the train data on the entire 100M training data.\n\nAs we define the numba functions, we compare them with their pandas equivalent to ensure that they are bug free. Numba functions have two limitations over pandas functions:\n1. Since they parallelize data code, we may run into OOM sometimes due to sudden increase in computations performed.\n2. Unlike pandas, a single nan value encountered by numba would be propagated until the end of the array. \n\nWe need to manually ensure that these two don't happen."},{"metadata":{"trusted":true},"cell_type":"code","source":"import numba as nb\n\n@nb.njit(parallel=True)\ndef expanding_mean(arr):\n    return np.cumsum(arr) / (np.arange(arr.size) + 1)\n\n# for compilation purposes\nexpanding_mean(np.array([1, 1, 0]))\n\ntemp = np.random.randint(0, 2, int(1e5))\n\n%timeit -n100 expanding_mean(temp)\n%timeit -n100 np.cumsum(temp) / (np.arange(int(1e5)) + 1)\n%timeit -n100 pd.Series(temp).expanding(1).mean()\n\n((expanding_mean(temp) == pd.Series(temp).expanding(1).mean()).all() and\n (np.cumsum(temp) / (np.arange(int(1e5)) + 1) == pd.Series(temp).expanding(1).mean()).all())","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"@nb.njit\ndef shifted_expanding_mean(arr, keep_last=False):\n    \n    'keep_last is set to true when we apply this func to pqet_mean. SPECIAL USE CASE!'\n    \n    temp = expanding_mean(arr)\n    if not keep_last:\n        return np.concatenate((np.array([np.nan]), temp[:-1]))\n    else:\n        return np.concatenate((np.array([np.nan]), temp))\n\n# for compilation purposes\nshifted_expanding_mean(temp)\n\n%timeit -n 20 shifted_expanding_mean(temp)\n%timeit -n 20 pd.Series(temp).expanding(1).mean().shift()\n\n# throws error if they aren't equal\nnp.testing.assert_allclose(pd.Series(temp).expanding(1).mean().shift(), shifted_expanding_mean(temp))","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Code to calculate response time:\n\nDiff operation followed by a replace of 0 with nan and forward filling the latest value in the nan positions. This is required as a mere diff operation would be erraneous for questions sharing the same bundle_id."},{"metadata":{"trusted":true},"cell_type":"code","source":"@nb.njit\ndef rt_func(arr, pred=False):\n        \n    temp = np.concatenate((np.array([np.nan]), arr[1:] - arr[:-1]))\n    temp = np.where(temp == 0, np.nan, temp)\n    \n    mask = np.isnan(temp)\n    idx = np.arange(len(mask))\n    idx = np.where(mask, 0, idx)\n    \n    rmax, i = idx[0], 0\n    for i, val in enumerate(idx):\n        if val > rmax: \n            rmax = val\n        idx[i] = rmax\n    \n    temp[mask] = temp[idx[mask]] \n    \n    if not pred:\n        temp = np.where(np.isnan(temp), 0, temp)\n    else:\n        temp = np.where(np.isnan(temp), arr, temp)\n    \n    return temp","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"%time data.groupby(\"user_id\")['timestamp'].transform(lambda x: rt_func(x.values)) \n%time data.groupby(\"user_id\")['timestamp'].transform(lambda x: x.diff().replace(0, np.nan).fillna(method='ffill').fillna(0))\n\n(\n    data.groupby(\"user_id\")['timestamp'].transform(lambda x: rt_func(x.values)) \n    == data['response_time']\n).all()","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Function to alter answered_correctly according to the question's difficulty: \n\nWe weigh the user's responsed based on the actual question difficulty. A user is given negative marks if the question when wrong and positive marks when he gets it right. Additionally a tough question answered correctly will yield better score than a easy question answered correctly. However a easy question answered wrongly must carry a higher penality than a difficult question answered wrongly."},{"metadata":{"trusted":true},"cell_type":"code","source":"@nb.njit\ndef modify_ac(arr, c_mean):\n    temp = np.where(arr == 0, -1, 1) * c_mean\n    return np.where(temp < 0, temp, 1 - temp)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time \n\n## Give more weightage to user ans acc to question difficulty?\n# penalize easy wrong questions more than tough wrong questions\n# reward tough right questions more than easy right questions\ndata['ac_modified'] = modify_ac(\n    data['answered_correctly'].values, \n    data.groupby([\"content_id\", 'content_type_id'])['answered_correctly'].transform('mean').values)\n\n# set ac_modified as 0 for questions\ndata.loc[data.content_type_id, 'ac_modified'] = 0","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Let's compute the part wise expanding mean for users. First we create a numba function that can be reused later on."},{"metadata":{"trusted":true},"cell_type":"code","source":"@nb.njit\ndef fillnshift(arr):\n    \n    '''Function to fwdfill and shift by 1 position. To replace pandas functionality.'''\n    \n    # fill logic\n    \n    mask = np.isnan(arr)\n    idx = np.arange(len(mask))\n    idx = np.where(mask, 0, idx)\n    \n    rmax, i = idx[0], 0\n    \n    for i, val in enumerate(idx):\n        if val > rmax: \n            rmax = val\n        idx[i] = rmax\n    \n    arr[mask] = arr[idx[mask]] \n    \n    # shift logic\n    \n    return np.concatenate((np.array([np.nan]), arr[:-1]))","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Check if the above function works as intended:"},{"metadata":{"trusted":true},"cell_type":"code","source":"temp = data.loc[data.user_id == np.random.choice(data.user_id.unique()), 'answered_correctly']\n\n# for compilaton purposes\nfillnshift(temp.values)\n\n# pd function timing\n%timeit -n 20 temp.fillna(method='ffill').shift()\n\n# numba timing\n%timeit -n 20 fillnshift(temp.values)\n\n# if all goes fine no error is raised\nnp.testing.assert_allclose(fillnshift(temp.values), temp.fillna(method='ffill').shift())","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"We find ourselves using calculating MA pretty frequently also we shift the values again after a groupy. This could be replaced using a numba function as follows:"},{"metadata":{"trusted":true},"cell_type":"code","source":"@nb.njit\ndef moving_average(arr, n=10, shift=True):\n    \n    mask = np.isnan(arr)\n    \n    ret = np.cumsum(np.where(mask, 0, arr))\n    \n    ret[n:] = ret[n:] - ret[:-n]\n    \n    counts = np.cumsum(~mask)\n    counts[n:] = counts[n:] - counts[:-n]\n    \n    ret[~mask] /= counts[~mask]\n    ret[mask] = np.nan\n    \n    if shift:\n        # perform shifting\n        ret = np.concatenate((np.array([np.nan]), ret[:-1]))\n\n    return ret","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"temp = data.loc[data.user_id == np.random.choice(data.user_id.unique()), 'answered_correctly']\n\nprint (\"Simple Moving Average:\")\n\n# numba timing\n%timeit -n 20 moving_average(temp.values.astype('float'), shift=False)\n\n# pd Series timing\n%timeit -n 20 pd.Series(temp).rolling(10, min_periods=1).mean()\n\n# throws error if something is off\nnp.testing.assert_allclose(\n    moving_average(temp.values.astype('float'), shift=False), \n    pd.Series(temp).rolling(10, min_periods=1).mean())\n\nprint (\"\\nNow Moving Average followed by one position shift:\")\n\n# numba timing\n%timeit -n 20 moving_average(temp.values.astype('float'))\n\n# pd Series timing\n%timeit -n 20 pd.Series(temp).rolling(10, min_periods=1).mean().shift()\n\n# throws error if something is off\nnp.testing.assert_allclose(\n    moving_average(temp.values.astype('float')), \n    pd.Series(temp).rolling(10, min_periods=1).mean().shift())","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Numba function to calculate shifted expanding cumsum:"},{"metadata":{"trusted":true},"cell_type":"code","source":"@nb.njit()\ndef shifted_expanding_sum(arr):\n    temp = np.cumsum(arr)\n    return np.concatenate((np.array([np.nan]), temp[:-1]))\n\ntemp = pd.Series(np.random.choice(2, size=100))\n\nnp.testing.assert_allclose(shifted_expanding_sum(temp.values), temp.expanding().sum().shift())\n\n%timeit -n 10 shifted_expanding_sum(temp.values)\n%timeit -n 10 temp.cumsum().shift()","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Let's create these features now:"},{"metadata":{"trusted":true},"cell_type":"code","source":"data['up_mean'] = data[~data.content_type_id].groupby(['user_id', 'part'])['ac_modified'].transform(\n    lambda x: shifted_expanding_mean(x.values))\n\n# per part count relative to all count\ndata['up_count'] = (data[~data.content_type_id].groupby(['user_id', 'part']).cumcount() \n                    / data[~data.content_type_id].groupby(\"user_id\").cumcount())\n\n# up_mean can't be zero since we ac_modified\n# up_count can be filled with 0 though\ndata['up_count'] = data['up_count'].fillna(0)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Using modified shifted_expanding_mean to compute expanding_mean for pqet:"},{"metadata":{"trusted":true},"cell_type":"code","source":"temp = np.random.randint(0, 2, size=int(1e5)).astype('float')\n\n# injecting nan like it would usually be in pqet\n# however there are few users with first question as a bundle\n# these first entries didn't have np.nan for pqet\ntemp[0] = np.nan \ntemp = pd.Series(temp)\n\nnp.testing.assert_allclose(\n    temp.expanding().mean(), \n    shifted_expanding_mean(temp.values[1:], keep_last=True)\n)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"data['content_c'] = data.groupby([\"content_id\", 'content_type_id'])['row_id'].transform(\"count\")\n\ndata['seen_ratio'] = (data.loc[~data.content_type_id, ['user_id', 'prior_question_had_explanation']]\n                      .fillna(False).astype(float).groupby(\"user_id\") # convert to float for using nb func\n                      .transform(lambda x: expanding_mean(x.values)))\n\ndata['pqet_mean'] = (\n    data[(~data.content_type_id)].groupby(\"user_id\")['prior_question_elapsed_time']\n     .transform(lambda x: shifted_expanding_mean(x.values[1:], keep_last=True))\n)\n\ndata.loc[~data.content_type_id, 'repeat_c'] = (\n    data[~data.content_type_id].groupby([\"user_id\", 'content_id'])\n    .cumcount().astype(bool))\ndata['repeat_c'] = data['repeat_c'].fillna(False)\n\n# density of user interactions\ndata['uq_per_hr'] = data.groupby(\"user_id\").cumcount() / (data['timestamp'] / 60000)\n\n# when the task_container_id occurs might be good feature\ndata['tmed'] = data[~data.content_type_id].groupby(\"content_id\")[['task_container_id']].transform('median')\n\n# create features counting the number of incorrect responses\ndata['uwrong_sum'] = (\n    data[~data.content_type_id]['answered_correctly']\n    .map({0: 1, 1: 0}).groupby(data['user_id'])\n    .transform(lambda x: shifted_expanding_sum(x.values))\n)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"data['lec_recent'] = (\n    data.loc[data.content_type_id, 'content_type_id']\n    .reindex(data.index).groupby(data['user_id'])\n    .fillna(method='ffill', limit=10)\n    .fillna(False).astype(bool)\n)\n\ndata['uf_bundle'] = data.groupby(\"user_id\")['bundle_id'].transform('first')\n\ndata['pqetmr_10'] = (data[~data.content_type_id].groupby(\"user_id\")['prior_question_elapsed_time'].transform(\n        lambda x: moving_average(x.values.astype('float32'), shift=False, n=10)))\n\ndata['che_sum'] = data[~data.content_type_id].groupby(\"content_id\")['pqhe_shifted'].transform('sum')\ndata['che_sum'] = data['che_sum'].fillna(0).astype('int')","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"data['seen_exp_when_wrong'] = (data['pqhe_shifted'].fillna(False) & (data['answered_correctly'] == 0)).astype(int)\ndata['seen_exp_when_wrong'] = (data[~data.content_type_id].groupby('user_id')['seen_exp_when_wrong']\n .transform(lambda x: shifted_expanding_sum(x.values)))\n\ndata['seen_exp_when_right'] = (data['pqhe_shifted'].fillna(False) & (data['answered_correctly'] == 1)).astype(int)\ndata['seen_exp_when_right'] = (data[~data.content_type_id].groupby('user_id')['seen_exp_when_right']\n .transform(lambda x: shifted_expanding_sum(x.values)))\n\ndata['sessions'] = (data['response_time'] > (15*60*1000)).groupby(data['user_id']).cumsum()\ndata['sess_event_count'] = data.groupby(['user_id', 'sessions']).cumcount()\n\ndata['up_recency'] = data.groupby(['user_id', 'part'])['timestamp'].transform(lambda x: rt_func(x.values))\ndata['ts_recency_10'] = data['timestamp'] - data.groupby(\"user_id\")['timestamp'].shift(10)\ndata['ts_recency_5'] = data['timestamp'] - data.groupby(\"user_id\")['timestamp'].shift(5)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"#### Training on a small subset of data (1M) to test how our features work so far:"},{"metadata":{"trusted":true},"cell_type":"code","source":"import optuna.integration.lightgbm as lgbo\nimport lightgbm as lgb\nimport time\nimport warnings\n\nwarnings.filterwarnings(\"ignore\", category=UserWarning) \n\ntrain_cols = [\n    \n    # defintely improves score besides u_mean and c_mean\n    'repeat_c',\n    'tagF', 'tagS', 'tagL', 'tagT',\n    'response_time', \n    'prior_question_elapsed_time',\n    \n    'up_mean', 'up_count', 'uq_per_hr',\n    'uwrong_sum', 'lec_recent',\n    \n    'pqet_mean', 'seen_ratio', 'tmed', \n    'up_recency', 'ts_recency_10', \n    'ts_recency_5', \n    \n#     'pqet_shifted',\n    \n    # for stability  \n    'timestamp', 'task_container_id', 'content_c',\n    \n    # nice to add features\n#     'res_time_avg', 'lec_available',\n    \n    'che_sum', 'seen_exp_when_wrong', \n    'seen_exp_when_right',\n    'sessions','sess_event_count', \n    'pqetmr_10', 'uf_bundle', \n    \n    # useless features\n#     \"cluster\", 'tag_c', 'listening_lvl', 'part',\n]\n\ncat_cols = ['tagF', 'tagS', 'tagT', 'tagL']\nother_cols = ['train', 'user_id', 'content_id', 'ac_modified', \n              'pqet_shifted', 'bundle_q_count', 'part']\n\nfolds = 7\n\nbest_params = {\n 'learning_rate':0.075,\n 'num_leaves': 200,\n 'objective': 'binary',\n 'metric': 'auc'}\n\ntotal_users = data.user_id.unique()\nnp.random.shuffle(total_users)\nval_size = len(total_users) // folds\nscores = {'TRAIN': [], 'VAL': []}\nmodels = []\n\nfor i in range(folds):\n    \n    start_time = time.time()\n    \n    train_users = np.setdiff1d(total_users, total_users[(i)*val_size:(i+1)*val_size])\n    temp = generate_train_val(data, train_users, dyn_p=True)\n    temp = temp[~temp.content_type_id]\n    temp = temp.loc[~temp.content_type_id, train_cols + other_cols + ['answered_correctly']]\n\n    temp[['u_mean']] = (\n        temp.groupby([\"user_id\"])['ac_modified']\n        .transform(lambda x: shifted_expanding_mean(x.values))\n    )\n    \n    temp['ummr_10'] = (\n        temp.groupby('user_id')['ac_modified']\n        .transform(lambda x: moving_average(x.values, n=10))\n    )\n    \n    temp['rt_per_task'] = temp['response_time'] / temp['bundle_q_count']\n    \n    # data preprocessing\n    temp[cat_cols] = temp[cat_cols].fillna(0)\n    temp[cat_cols] = temp[cat_cols].astype(int)\n    \n    # save the content wise mean score and create feature in train and val\n    temp = temp.merge(\n        temp[temp.train].groupby(\"content_id\")['answered_correctly'].agg(c_mean='mean', c_std='std'),\n        left_on=['content_id'], right_index=True, how='left')\n    \n    # impute missing c_mean\n    temp['c_mean'] = temp['c_mean'].fillna(0.5)\n    \n    # harmonic mean idea from here: https://www.kaggle.com/markwijkhuizen/riiid-training-and-prediction-using-a-state\n    temp['h_mean'] = temp.groupby(\"user_id\")['answered_correctly'].transform(\n        lambda x: shifted_expanding_mean(x.values)\n    )\n    \n    temp['h_mean'] = (2 * temp['h_mean'] * temp['c_mean']) / (temp['h_mean'] + temp['c_mean'])\n    \n    # doing the same for part wise mean\n    temp['uph_mean'] = temp.groupby(['user_id', 'part'])['answered_correctly'].transform(\n        lambda x: shifted_expanding_mean(x.values))\n    \n    temp['uph_mean'] = (2 * temp['uph_mean'] * temp['c_mean']) / (temp['uph_mean'] + temp['c_mean'])\n    \n    # add wrong and right response mean for each content\n    temp = temp.merge(\n        temp[temp.train].groupby([\"content_id\", 'answered_correctly'])['pqet_shifted'].agg('median').unstack(),\n        left_on='content_id', right_index=True, how='left')\n    temp.columns = temp.columns[:-2].tolist() + ['wrong_et_med', 'right_et_med']\n    \n    # split into train/val\n    train, val = temp[temp['train']], temp[~temp['train']]\n\n    # drop the useless columns\n    train, val = train.drop(other_cols, 1), val.drop(other_cols, 1)\n    \n    if i == 0:\n        if not best_params: # we do a hyper parameter tuning the first time\n            \n            # creating the lgb dataset for training\n            train_lgb = lgb.Dataset(\n                train.drop('answered_correctly', 1), train['answered_correctly'], \n                free_raw_data=False, categorical_feature=cat_cols)\n\n            val_lgb = lgb.Dataset(\n                val.drop('answered_correctly', 1), val['answered_correctly'], \n                free_raw_data=False, categorical_feature=cat_cols)\n\n            print (\"Searching for Optimal Hyperparameters! This may take some time to complete.\")\n            model = lgbo.train(\n                params={'objective':'binary', 'metric':'auc'}, \n                train_set=train_lgb, valid_sets=[train_lgb, val_lgb],\n                verbose_eval=0, early_stopping_rounds=10, \n                show_progress_bar=False,\n                time_budget=60*60, # ten minutes to run, increase to into deeper search space\n            )\n            params = model.params\n\n        else:\n            params = best_params\n            \n        print (f\"Starting folds: {'='*80}>\")\n        \n    # time for data preprocessing\n    d_time = time.time() - start_time \n    \n    model = lgb.LGBMClassifier(**params)\n    \n    model.fit(train.drop('answered_correctly', 1), train['answered_correctly'], \n              eval_set=[(val.drop('answered_correctly', 1), val['answered_correctly'])],\n              early_stopping_rounds=10, verbose=0)\n    \n    model = model.booster_\n    models.append(model)\n\n    scores['TRAIN'] = scores.get('TRAIN') + [roc_auc_score(train['answered_correctly'], model.predict(\n        train.drop(\"answered_correctly\", 1),  categorical_feature=cat_cols))]\n\n    scores['VAL'] = scores.get('VAL') + [roc_auc_score(val['answered_correctly'], model.predict(\n        val.drop(\"answered_correctly\", 1),  categorical_feature=cat_cols))]\n\n    print (\"AUC for Fold# {}/{}:> | T: {:.3f} | V: {:.3f} | Preprocess Time: {:5.2f}s | Total Fold Time: {:.2f}s\"\n           .format(i+1, folds, scores['TRAIN'][-1], scores['VAL'][-1], d_time, time.time() - start_time))\n\nprint (\"Summary stats : {}\\n\\nMean AUC for {} folds: T: {:.4f} @ {:.4f} std | V: {:.4f} @ {:.4f} std\"\n       .format('='*76+\">\", folds, np.mean(scores['TRAIN']), np.std(scores['TRAIN']), \n               np.mean(scores['VAL']), np.std(scores['VAL'])))\n\nprint (f\"\\nAdditional Info: {'='*80+'>'}\\nTrain Features Used By Model: {val.shape[1] - 1} Features\")\nprint (f\"Hypothetical Space Required For Entire Data: {(val.shape[1] * 101230332 * 32) / 8e+9:.2f} GB (float 32)\")\nprint (f\"Hypothetical Space Required For Entire Data: {(val.shape[1] * 101230332 * 64) / 8e+9:.2f} GB (float 64)\")","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"We plot the feature importance on all the models based on their performance. Let's define a helper functon to help us with that:"},{"metadata":{"trusted":true},"cell_type":"code","source":"def plot_ensemble_importance(models, scores=None, col_sampling=False):\n    \n    result = pd.DataFrame()\n\n    for i, m, in enumerate(models):\n        temp = pd.DataFrame(index=models[i].feature_name())\n        temp[f\"{i}_split\"] = m.feature_importance(importance_type='split')\n        temp[f\"{i}_gain\"] = m.feature_importance(importance_type='gain')\n        result = pd.concat([result, temp], axis=1)\n        \n    gain_cols = list(filter(lambda x: 'gain' in x, result.columns))\n    split_cols = list(filter(lambda x: 'split' in x, result.columns))\n\n    if col_sampling:\n        result[gain_cols] = result[gain_cols].fillna(axis=1, method='ffill').fillna(axis=1, method='bfill')\n        result[split_cols] = result[split_cols].fillna(axis=1, method='ffill').fillna(axis=1, method='bfill')\n        \n    result['gain'] = np.average(result[gain_cols], axis=1, weights=scores)\n    result['split'] = np.average(result[split_cols], axis=1, weights=scores)\n    \n    result = result[['gain', 'split']]\n    result = result.sort_values(\"split\")\n\n    f, ax = plt.subplots(figsize=(20, int(0.5*len(result))), nrows=2, sharey=False)\n    result.plot(kind='barh', subplots=True, ax=ax, legend=False)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# plot it\nplot_ensemble_importance(models, scores[\"VAL\"])","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"### Data Generation & Training Logic (Entire dataset):\nNow our prototype works well. Let's create these features on the entire dataset to be able to train our model on them:"},{"metadata":{"trusted":true},"cell_type":"code","source":"# delete data no longer needed\ndel data, temp, train, val\ndata = temp = train = val = None\ngc.collect()","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"What follows is optimized pipeline code for train data generation. I have used as `gc.collect()` liberally. We create features piece by peice and delete them once we are done with them:"},{"metadata":{"trusted":true},"cell_type":"code","source":"# monitor the time passing\nstart_time = time.time()\n\nLEC_RECENT_ROLL = 10\nROLL_WINDOW = 10\nROLL_WINDOW_PQET = 10\nSESSION_DURATION = 15 * 60 * 1000\n\n# save op as n number of chunked files\nCHUNKS = 3\n\nSAVE_LOC = f\"../input/riiid-final-model-inputs/model_train_c1.feather\"\nMODEL_LOC = f\"../input/riiid-final-model-inputs/trained_model.txt\"\n\nif not os.path.exists(SAVE_LOC):\n    \n    ########################## BASIC INFO GATHERING ##########################\n    \n    data = pd.read_feather(\"../input/riiid-train-data-multiple-formats/riiid_train.feather\", \n                       columns=['user_id', 'timestamp', 'answered_correctly'])\n\n    q_mask = data.answered_correctly != -1\n    BATCH_SIZE = q_mask.sum() // CHUNKS\n    \n    del data\n    gc.collect()\n\n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s\", end=' ')\n    print(\"| Percentage of Data used: {:.2f}% | Single chunk Size: {:9} rows\".format(\n        q_mask.sum()*100/len(q_mask), BATCH_SIZE\n    ))\n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | <= Entering Data Prep Bottleneck 1!\")\n    \n    ################ UP_RECENCY, TS_RECENCY_5, TS_RECENCY_10 ################\n    \n    data = pd.read_feather(\"../input/riiid-train-data-multiple-formats/riiid_train.feather\", \n                           columns=['user_id', 'timestamp', 'content_id', 'content_type_id'])\n    \n    data = data.merge(\n        ql[['content_id', 'content_type_id', 'part']], \n        on=['content_id', 'content_type_id'], how='left')\n    \n    data.drop(['content_id', 'content_type_id'], axis=1, inplace=True)\n    data['part'] = data['part'].astype('uint8')\n    gc.collect()\n    \n    up_recency = pd.Series(data=np.zeros(len(data), dtype='float64'), index=data.index, name='up_recency')\n    \n    for i in range(1, 8):\n        temp = (data[data.part == i].groupby(\"user_id\")['timestamp']\n                .transform(lambda x: rt_func(x.values)))\n        \n        temp = temp.reindex(data.index).fillna(0).values.astype('float64')\n        up_recency = up_recency + temp\n        \n        del temp\n        gc.collect()\n        \n        print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | * Done with up_recency_{i}!\")\n    \n    data['up_recency'] = up_recency\n    data.drop('part', axis=1, inplace=True)\n    del up_recency\n    gc.collect()\n    \n    data['ts_recency_10'] = data['timestamp'] - data.groupby('user_id')['timestamp'].shift(10)\n    data['ts_recency_5'] = data['timestamp'] - data.groupby('user_id')['timestamp'].shift(5)\n    \n    data = data.loc[q_mask]\n    data.drop(['user_id', 'timestamp'], axis=1, inplace=True)\n    data.reset_index().to_feather(\"ts_up_recency.feather\")\n    \n    del data\n    gc.collect()\n    \n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | <= Done creating up/ts recency features!\")\n    \n    ################ SEEN_EXP_WHEN_WRONG, SEEN_EXP_WHEN_RIGHT ################\n    \n    data = pd.read_feather(\"../input/riiid-train-data-multiple-formats/riiid_train.feather\", \n                           columns=['user_id', 'answered_correctly']).loc[q_mask]\n    \n    pq_shifted = pd.read_feather(\"../input/riiid-final-model-inputs/pq_shifted.feather\", \n                                 columns=['index', 'pqhe_shifted']).set_index(\"index\")\n\n    data['seen_exp_when_wrong'] = (pq_shifted['pqhe_shifted'] & (data['answered_correctly'] == 0)).astype(int)\n    data['seen_exp_when_right'] = (pq_shifted['pqhe_shifted'] & (data['answered_correctly'] == 1)).astype(int)\n    \n    del pq_shifted\n    gc.collect()\n    \n    data['seen_exp_when_wrong'] = (\n        data.groupby('user_id')['seen_exp_when_wrong']\n        .transform(lambda x: shifted_expanding_sum(x.values)))\n    \n    data['seen_exp_when_right'] = (\n        data.groupby('user_id')['seen_exp_when_right']\n        .transform(lambda x: shifted_expanding_sum(x.values)))\n    \n    data = data[['seen_exp_when_wrong', 'seen_exp_when_right']]   \n    data = data.fillna(0).astype(\"uint16\")\n    data.astype('uint16').reset_index().to_feather(\"seen_exp.feather\")\n    \n    del data\n    gc.collect()\n    \n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | Done creating seen_exp features!\")\n    \n    ############################## UF_BUNDLE ##############################\n\n    data = pd.read_feather(\"../input/riiid-train-data-multiple-formats/riiid_train.feather\", \n                           columns=['user_id', 'content_id']).loc[q_mask]\n    \n    content_df = pd.read_csv(\"../input/riiid-final-model-inputs/content-df.csv\", index_col=0, \n                             usecols=['content_id', 'bundle_id'])\n    \n    content_df['bundle_id'] = content_df['bundle_id'].astype(\"uint16\")\n    \n    data = data.merge(content_df['bundle_id'], left_on='content_id', right_index=True, how='left')\n    data.drop(['content_id'], axis=1, inplace=True)\n    \n    data = data.groupby(\"user_id\")['bundle_id'].transform(\"first\").rename(\"uf_bundle\")\n    data.astype(\"uint16\").reset_index().to_feather(\"uf_bundle.feather\")\n    \n    del data, content_df\n    gc.collect()\n    \n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | Done creating uf_bundle!\")\n    \n    ################################ PQETMR ################################\n    \n    data = pd.read_feather(\"../input/riiid-train-data-multiple-formats/riiid_train.feather\", \n                           columns=['user_id', 'prior_question_elapsed_time']).loc[q_mask]\n    \n    data['pqetmr_10'] = data.groupby(\"user_id\")['prior_question_elapsed_time'].transform(\n        lambda x: moving_average(x.values, n=10, shift=False)\n    )\n    \n    data.drop(['prior_question_elapsed_time', 'user_id'], axis=1, inplace=True)\n    data.astype('float32').reset_index().to_feather(\"pqetr.feather\")\n    \n    del data\n    gc.collect()\n    \n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | Done creating pqet rolling features!\")\n    \n    ################ HMEAN, UPHMEAN, UPMEAN, UPCOUNT UMEAN, UMMR ###########\n    \n    data = pd.read_feather(\"../input/riiid-train-data-multiple-formats/riiid_train.feather\", \n                           columns=['user_id', 'content_id', 'answered_correctly']).loc[q_mask]\n\n    content_df = pd.read_csv(\"../input/riiid-final-model-inputs/content-df.csv\", index_col=0, \n                             usecols=['part', 'c_mean', 'content_id'])\n    \n    # to save space\n    content_df['part'] = content_df['part'].astype('uint8')\n    content_df['c_mean'] = content_df['c_mean'].astype('float32')\n\n    data = data.merge(content_df, left_on='content_id', right_index=True, how='left')\n    data.drop(['content_id'], axis=1, inplace=True)\n    del content_df\n    gc.collect()\n    \n    temp = (\n        data.groupby(\"user_id\")['answered_correctly']\n        .transform(lambda x: shifted_expanding_mean(x.values))\n    )\n    \n    temp = (2 * temp * data['c_mean'].values) / (temp + data['c_mean'].values)\n    \n    (temp.astype('float32').rename(\"h_mean\")\n     .reset_index().to_feather(\"h_mean.feather\"))\n    \n    del temp\n    gc.collect()\n    \n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | Done creating h_mean!\")\n    \n    data['answered_wrongly'] = data['answered_correctly'] == 0\n    \n    (data.groupby(\"user_id\")['answered_wrongly']\n     .transform(lambda x: shifted_expanding_sum(x.values))\n     .fillna(0).astype(\"uint16\").rename(\"uwrong_sum\")\n     .reset_index().to_feather(\"uwrong_sum.feather\"))\n    \n    data.drop(['answered_wrongly'], axis=1, inplace=True)\n    gc.collect()\n    \n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | Done creating uwrong_sum!\")\n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | <= Entering Data Prep Bottleneck 2!\")\n    \n    uph_mean = pd.Series(data=np.zeros(len(data), dtype='float32'), index=data.index, name='uph_mean')\n    \n    for i in range(1, 8):\n        temp = (data[data.part == i].groupby(\"user_id\")['answered_correctly']\n                .transform(lambda x: shifted_expanding_mean(x.values)))\n        \n        temp = temp.reindex(data.index).fillna(0).values.astype('float32')\n        uph_mean = uph_mean + temp\n        \n        del temp\n        gc.collect()\n        \n        print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | * Done with uph_mean_{i}!\")\n        \n    uph_mean = (2 * uph_mean * data['c_mean'].values) / (uph_mean + data['c_mean'].values)\n    uph_mean.reset_index().to_feather(\"uph_mean.feather\")\n    \n    del uph_mean\n    gc.collect()\n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | <= Done creating uph_mean!\")\n    \n    # perform ops for converting ans_crctly to weighted values based on c_mean\n    data['answered_correctly'] = modify_ac(data['answered_correctly'].values, data['c_mean'].values)\n    data.drop(['c_mean'], 1, inplace=True)\n    gc.collect()\n    \n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | Modified labels according to Importance!\")\n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | <= Entering Data Prep Bottleneck 3!\")\n    \n    up_mean = pd.Series(data=np.zeros(len(data), dtype='float32'), index=data.index, name='up_mean')\n    \n    for i in range(1, 8):\n        temp = (data[data.part == i].groupby(\"user_id\")['answered_correctly']\n                .transform(lambda x: shifted_expanding_mean(x.values)))\n        temp = temp.reindex(data.index).fillna(0).values.astype('float32')\n        up_mean = up_mean + temp\n        \n        del temp\n        gc.collect()\n        \n        print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | * Done with up_mean_{i}!\")\n\n    up_mean.reset_index().to_feather(\"up_mean.feather\")\n    \n    del up_mean\n    gc.collect()\n    \n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | <= Done creating up_mean!\")\n    \n    data['up_count'] = data.groupby(['user_id', 'part']).cumcount().fillna(0)\n    data['up_count'] = data['up_count'] / data.groupby('user_id').cumcount()\n    data['up_count'].astype('float32').reset_index().to_feather(\"up_count.feather\")\n    \n    data.drop(['part'], axis=1, inplace=True)\n    gc.collect()\n\n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | Done creating up_count!\")\n    \n    (data.groupby([\"user_id\"])['answered_correctly']\n     .transform(lambda x: moving_average(x.values, n=ROLL_WINDOW))\n     .rename(f'ummr_{ROLL_WINDOW}').astype('float32').reset_index()\n     .to_feather(f'ummr_{ROLL_WINDOW}.feather'))\n    \n    gc.collect()\n    \n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | Done creating ummr_{ROLL_WINDOW}!\") \n\n    (data.groupby(\"user_id\")['answered_correctly']\n     .transform(lambda x: shifted_expanding_mean(x.values))\n     .rename(\"u_mean\").astype('float32').reset_index()\n     .to_feather(\"u_mean.feather\"))\n    \n    del data\n    gc.collect()\n\n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | Done creating u_mean!\")\n    \n    ############################## LEC_RECENT ##############################\n    \n    data = pd.read_feather(\"../input/riiid-train-data-multiple-formats/riiid_train.feather\",\n                          columns=['user_id', 'content_type_id'])\n    \n    data = (data['content_type_id'].replace(False, np.nan).groupby(data['user_id'])\n            .fillna(method='ffill', limit=LEC_RECENT_ROLL).fillna(False).astype(bool))\n    \n    data.loc[q_mask].rename(\"lec_recent\").reset_index().to_feather(\"lec_recent.feather\")\n    \n    del data\n    gc.collect()\n    \n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | Done creating lec_recent!\")\n    \n    ################################ REPEAT_C ################################\n\n    data = pd.read_feather(\"../input/riiid-train-data-multiple-formats/riiid_train.feather\", \n                       columns=['user_id', 'content_id'])\n\n    data = data.loc[q_mask].groupby([\"user_id\", 'content_id']).cumcount()\n    data.astype(bool).rename(\"repeat_c\").reset_index().to_feather(\"repeat_c.feather\")\n    \n    del data\n    gc.collect()\n\n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | Done creating repeat_c!\")\n    \n    ############################# SEEN_RATIO, PQET_MEAN #############################\n    \n    data = pd.read_feather(\"../input/riiid-train-data-multiple-formats/riiid_train.feather\", \n                       columns=['user_id', 'prior_question_elapsed_time', 'prior_question_had_explanation'])\n    \n    # filter out the questions\n    data = data.loc[q_mask]\n\n    data['prior_question_elapsed_time'] = (\n        data.groupby(\"user_id\")['prior_question_elapsed_time'].transform(\n            lambda x: shifted_expanding_mean(x.values[1:], keep_last=True)))\n    \n    data['prior_question_had_explanation'] = data['prior_question_had_explanation'].fillna(False).astype('float32')\n    data['prior_question_had_explanation'] = data.groupby(\"user_id\")['prior_question_had_explanation'].transform(\n        lambda x: expanding_mean(x.values))\n\n    data = data.rename({\"prior_question_had_explanation\": \"seen_ratio\", \n                        'prior_question_elapsed_time': \"pqet_mean\"}, axis=1)\n    \n    # drop the column, no longer needed\n    data.drop([\"user_id\"], axis=1, inplace=True)\n    data.astype('float32').reset_index().to_feather(\"pq_se.feather\")\n    \n    del data\n    gc.collect() \n    \n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | Done creating pqet_mean and seen_ratio!\")\n    \n    ################################# UQ_PER_HR #################################\n    \n    data = pd.read_feather(\"../input/riiid-train-data-multiple-formats/riiid_train.feather\",\n                           columns=['user_id', 'timestamp', 'task_container_id'])\n    \n    data = data.groupby(\"user_id\").cumcount() / (data['timestamp'] / 60000)\n    data.fillna(0, inplace=True)\n        \n    data.loc[q_mask].astype(\"float32\").rename('uq_per_hr').reset_index().to_feather(\"uq_per_hr.feather\")\n    \n    del data\n    gc.collect()\n    \n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | Done creating uq_per_hr!\")\n    \n    ################ INTERMEDIATE FILE CONCAT FOR SAVING SPACE ################\n    \n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | <= Beginning intermediate file concat!\")\n    \n    data = pd.DataFrame()\n    \n    for name in ['up_mean.feather', \"uph_mean.feather\", f\"ummr_{ROLL_WINDOW}.feather\"]:\n        \n        temp = pd.read_feather(name).set_index(\"index\")\n        ! rm ./{name}\n\n        # concat the dataframes together\n        data = pd.concat([data, temp], axis=1)\n\n        del temp\n        gc.collect()\n        time.sleep(2)\n        \n    data.reset_index().to_feather(\"./concat_intermediate_1.feather\")\n    del data\n    gc.collect()\n    \n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s |  * Done with Intermediate Concat 1!\")\n    \n    data = pd.DataFrame()\n    \n    for name in [\"repeat_c.feather\", \"uf_bundle.feather\", \"uwrong_sum.feather\", \n                 \"seen_exp.feather\", \"u_mean.feather\"]:\n        \n        temp = pd.read_feather(name).set_index(\"index\")\n        ! rm ./{name}\n\n        # concat the dataframes together\n        data = pd.concat([data, temp], axis=1)\n\n        del temp\n        gc.collect()\n        time.sleep(2)\n        \n    data.reset_index().to_feather(\"./concat_intermediate_2.feather\")\n    del data\n    gc.collect()\n    \n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s |  * Done with Intermediate Concat 2!\")\n    \n    data = pd.DataFrame()\n    \n    for name in [\"pq_se.feather\", \"pqetr.feather\"]:\n        \n        temp = pd.read_feather(name).set_index(\"index\")\n        ! rm ./{name}\n\n        # concat the dataframes together\n        data = pd.concat([data, temp], axis=1)\n\n        del temp\n        gc.collect()\n        time.sleep(2)\n        \n    data.reset_index().to_feather(\"./concat_intermediate_3.feather\")\n    del data\n    gc.collect()\n    \n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s |  * Done with Intermediate Concat 3!\")\n    \n    data = pd.DataFrame()\n    \n    for name in [\"uq_per_hr.feather\", \"up_count.feather\", \"lec_recent.feather\", \"h_mean.feather\"]:\n        \n        temp = pd.read_feather(name).set_index(\"index\")\n        ! rm ./{name}\n\n        # concat the dataframes together\n        data = pd.concat([data, temp], axis=1)\n\n        del temp\n        gc.collect()\n        time.sleep(2)\n        \n    data.reset_index().to_feather(\"./concat_intermediate_4.feather\")\n    del data\n    gc.collect()\n    \n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s |  * Done with Intermediate Concat 4!\")\n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | <= Intermediate file concat Complete!\")\n    \n    ############################ CONTENT_DF FEATURES ############################\n    \n    data = (pd.read_feather(\"../input/riiid-train-data-multiple-formats/riiid_train.feather\", \n                       columns=['content_id']).loc[q_mask])\n    \n    content_df = pd.read_csv(\"../input/riiid-final-model-inputs/content-df.csv\", index_col=0)    \n    content_df[cat_cols + ['bundle_q_count', 'part']] = content_df[cat_cols + ['bundle_q_count', 'part']].astype('uint8')\n    content_df[['wrong_et_med', 'right_et_med']] = content_df[['wrong_et_med', 'right_et_med']].astype('float32')\n    content_df[['c_mean', 'c_std']] = content_df[['c_mean', 'c_std']].astype('float32')\n    content_df[['content_c', 'tmed', 'che_sum']] = (\n        content_df[['content_c', 'tmed', 'che_sum']]).astype('uint16')\n    content_df.drop(['part', 'bundle_id'], axis=1, inplace=True)\n    \n    data = data.merge(content_df, how='left', left_on='content_id', right_index=True)\n    data.drop(['content_id'], axis=1, inplace=True)\n    \n    data.reset_index().to_feather(\"c_df.feather\")\n    \n    del data, content_df\n    gc.collect()\n    \n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | Done creating content_df for merge!\")\n    \n    ########### RESPONSE_TIME, SESSIONS, SESS_EVENT_COUNT, RT_PER_TASK ###########\n\n    data = pd.read_feather(\"../input/riiid-train-data-multiple-formats/riiid_train.feather\", \n                       columns=['timestamp', 'user_id'])\n    \n    data['response_time'] = ( # response time requires the lectures as well\n        data.groupby(\"user_id\")['timestamp']\n        .transform(lambda x: rt_func(x.values)))\n    \n    data.drop(['timestamp'], axis=1, inplace=True)\n    gc.collect()\n    \n    data['sessions'] = (data['response_time'] > SESSION_DURATION).astype(bool)\n    data['sessions'] = data.groupby('user_id')['sessions'].cumsum()\n    data['sess_event_count'] = data.groupby(['user_id', 'sessions']).cumcount()\n    data[['sessions', 'sess_event_count']] = data[['sessions', 'sess_event_count']].astype('uint16')\n    \n    # filter out lectures no longer needed\n    data = data.loc[q_mask]\n    \n    # creating rt_per_task\n    temp = pd.read_feather(\"./c_df.feather\", columns=['index', 'bundle_q_count'])\n    temp = temp.set_index(\"index\")\n    data['rt_per_task'] = data['response_time'] / temp['bundle_q_count'].values\n    \n    # save as file\n    data.drop(['user_id'], axis=1, inplace=True)\n    data.reset_index().to_feather(\"response_time.feather\")\n    \n    del data, temp\n    gc.collect()\n\n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | Done creating reponse_time!\")\n    print (\"\\nDone creating all required intermediate files.\\n\")\n    \n    ############################ CHUNKED CONCAT LOGIC ############################\n    \n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | <== Starting merge operation!\")\n    \n    for i in range(CHUNKS):\n        \n        print (f\"\\nTime Elapsed: {time.time() - start_time:10.2f} s |  <= Beginning chunk {i + 1} merge!\")\n        front, rear = i * BATCH_SIZE, (i + 1) * BATCH_SIZE\n        \n        data = (pd.read_feather(\"../input/riiid-train-data-multiple-formats/riiid_train.feather\", \n                                columns=['timestamp', 'user_id', 'task_container_id',\n                                         'prior_question_elapsed_time', 'answered_correctly'])\n                .loc[q_mask] # filter the lectures\n                .iloc[front: rear])\n\n        for name in ['c_df.feather', 'ts_up_recency.feather', \"response_time.feather\",\n                     \"concat_intermediate_2.feather\", \"concat_intermediate_4.feather\",\n                     \"concat_intermediate_1.feather\", \"concat_intermediate_3.feather\"]:\n\n            temp = pd.read_feather(name)\n            temp.iloc[BATCH_SIZE:].reset_index(drop=True).to_feather(name)\n            temp = temp.iloc[:BATCH_SIZE].set_index(\"index\")\n\n            if name == 'c_df.feather':\n                temp.drop(\"bundle_q_count\", axis=1, inplace=True)\n                \n            gc.collect()\n            time.sleep(3)\n\n            # concat the dataframes together\n            assert len(data) == len(temp) == BATCH_SIZE\n            data = pd.concat([data, temp], axis=1)\n\n            del temp\n            gc.collect()\n            time.sleep(2)\n\n            print (f\"Time Elapsed: {time.time() - start_time:10.2f} s |  * Done merging {name}!\")\n        \n        print (f\"Time Elapsed: {time.time() - start_time:10.2f} s |  * Saving chunk {i + 1} onto Disk!\")\n        data.reset_index(drop=True, inplace=True) \n        data.to_feather(f\"model_train_c{i + 1}.feather\")\n        print (f\"Time Elapsed: {time.time() - start_time:10.2f} s |  * Done saving chunk {i + 1} to Disk!\")\n        print (f\"Time Elapsed: {time.time() - start_time:10.2f} s |  <= Done Merging chunk {i + 1}!\")\n        \n        del data\n        gc.collect()\n        time.sleep(3)\n        \n    print (f\"\\nTime Elapsed: {time.time() - start_time:10.2f} s | <== Merge Complete!\")        \n    print (\"Deleting unneeded intermediate files.\\n\")\n    ! rm c_df.feather concat_intermediate_*.feather ts_up_recency.feather\n    ! rm response_time.feather\n    \nelse:\n    \n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | Found PreSaved File(s). Load it manually if need be!\")\n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | Copying Train File(s) back to Disk!\")\n    \n    for i in range(CHUNKS):\n        ! cp ../input/riiid-final-model-inputs/model_train_c{i + 1}.feather ./model_train_c{i + 1}.feather\n        \n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | Done copying Train File(s) back to Disk!\")","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"#### LGB Training:\nDone with data generation. Now we can't really use this entire data to fit a lgbm model since it would inevitably exhaust all available ram. So we need to load the data in chunks and train them piece by piece. This could be better accomplished using a framework such as Dask. Unfortunately I wasn't aware of the workings of Dask so I stuck with simple python logic. \n\nWe define two functions. One is to return random front and rear indices given a required batch size. Second is to map the front and rear to respective data chunks (we have three of them). Note that we each chunk have their indices reset. So we need to map them manually:"},{"metadata":{"trusted":true},"cell_type":"code","source":"def return_random_slice(batch_size, nrows):\n    '''Randomly returns a slice of `batch_size` from nrows.'''\n    \n    front = np.random.choice(nrows - batch_size)\n    rear = front + batch_size\n        \n    return front, rear","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"def return_chunk_indices(start, end, nrows, chunks=3):\n    'Maps the indices to chunk indices for data loading'\n    \n    chunk_size = (nrows//chunks)\n    indices = []\n    \n    start_chunk = start // chunk_size\n    end_chunk = (end-1) // chunk_size\n    \n    start_chunk_start = start - chunk_size * start_chunk\n    end_chunk_end = end - chunk_size * end_chunk\n    start_chunk_end = chunk_size if end_chunk != start_chunk else end_chunk_end\n    end_chunk_start = 0 if end_chunk != start_chunk else start_chunk_start \n    \n    indices.append((start_chunk+1, start_chunk_start, start_chunk_end))\n    \n    if start_chunk != end_chunk:\n\n        for i in range(end_chunk - start_chunk - 1):\n            indices.append((start_chunk + i + 2, 0, chunk_size))\n\n        indices.append((end_chunk+1, end_chunk_start, end_chunk_end))\n    \n    return indices","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\n\nALL_FEATURES = [\n    'repeat_c', 'tagF', 'tagS', 'tagL', 'tagT', 'response_time', 'prior_question_elapsed_time',\n    'up_mean', 'up_count', 'uq_per_hr', 'uwrong_sum', 'lec_recent', 'pqet_mean', 'seen_ratio',\n     'tmed', 'up_recency', 'ts_recency_10', 'ts_recency_5', 'timestamp', 'task_container_id',\n     'content_c', 'che_sum', 'seen_exp_when_wrong', 'seen_exp_when_right', 'sessions', 'sess_event_count',\n     'pqetmr_10', 'uf_bundle', 'u_mean', 'ummr_10', 'rt_per_task', 'c_mean', 'c_std', 'h_mean', 'uph_mean',\n     'wrong_et_med', 'right_et_med'\n]\n    \n# initially had the idea to randomly choose features, later I decided against it and used all the columns\nCORE = ['repeat_c', 'tagF', 'tagS', 'tagL', 'response_time', 'up_mean', 'up_count', 'pqet_mean', \n        'seen_ratio', 'tmed', 'up_recency', 'ts_recency_5', 'content_c', 'pqetmr_10', 'u_mean', \n        'rt_per_task', 'c_mean', 'c_std', 'h_mean', 'ts_recency_10', 'uph_mean']\n\nN_FEATURES = 30            # not used if col_sampling is set to false\n\nBATCH_SIZE = int(2.5e7)    # Rows taken at a time for model train/val\nSTRIDE = int(1e7)          # Stride these many elements\nHOLD_OUT = int(2.5e6)      # number of rows, no model would ever see\n\nmodels = []                # we save our models here to create a bagging ensemble\n\nCOL_SAMPLING = False       # whether to sample rows from total features, if false, n_Features is altered\nALREADY_TRAINED = 8        # reload any trained model?\nTRAIN_FOR = 0              # how many rounds to actually train for, if None, auto specified\nITERATIVE_TRAIN = False    # Do we wish to re-init models?\n\nOOF_VALIDATE = False       # Do oof validation after training/reloading models?\n\nN_ROUNDS = 3000            # number of rounds each model would be trained for\n\n# Training data location (split as three files)\nLOC1 = \"../input/riiid-final-model-inputs/model_train_c1.feather\"\nLOC2 = \"../input/riiid-final-model-inputs/model_train_c2.feather\"\nLOC3 = \"../input/riiid-final-model-inputs/model_train_c3.feather\"\n\nSIZE1 = len(pd.read_feather(LOC1, columns=['answered_correctly']))\nSIZE2 = len(pd.read_feather(LOC2, columns=['answered_correctly']))\nSIZE3 = len(pd.read_feather(LOC3, columns=['answered_correctly']))\n\n# total number of rows in training data\nN_ROWS = SIZE1 + SIZE2 + SIZE3\n\n# https://stackoverflow.com/questions/53580088/calculate-the-output-size-in-convolution-layer\n# tot. num of chunks after striding with a specified batch_size\nCHUNKS = ((N_ROWS - BATCH_SIZE - HOLD_OUT) // STRIDE) + 1\n\n# if not specified train for remaining number of chunks\nif TRAIN_FOR is None:\n    TRAIN_FOR = CHUNKS - ALREADY_TRAINED\n\n# a simple sanity check\nassert TRAIN_FOR + ALREADY_TRAINED <= CHUNKS\n\n# reload models if already trained\nif ALREADY_TRAINED:\n    \n    # read the OOF indices from memory\n    OOF = np.load(\"../input/riiid-final-model-inputs/OOF.npy\")\n    \n    for j in range(ALREADY_TRAINED):    \n        model = lgb.Booster(model_file=f\"../input/riiid-final-model-inputs/trained_model_c{j+1}.txt\")\n        model.save_model(f\"trained_model_c{j+1}.txt\")\n        models.append(model)\n        \n    print (f\"Successfully loaded {len(models)} model(s)!\")\n\nif TRAIN_FOR:\n        \n    if not ALREADY_TRAINED:\n        print (\"Beginning Model Training from Scratch!\")\n        \n        # re create OOF indices if not present\n        OOF = np.arange(N_ROWS) \n        \n    # for init_model parameter we use pre trained model\n    if ITERATIVE_TRAIN and ALREADY_TRAINED:\n        pre_model = model\n        \n    # if we wish not to iterative train or no model has been trained so far\n    else:    \n        pre_model = None\n\n    print (f\"{BATCH_SIZE} rows would be used for {CHUNKS} model(s)!\", end=' ')\n    print (f\"{TRAIN_FOR} model(s) would be trained for {N_ROUNDS} rounds!\")\n    print (f\"The models would {'BE' if ITERATIVE_TRAIN else 'NOT BE'} iteratively built!\")\n    print (f\"Col sampling has been {'ENABLED' if COL_SAMPLING else 'DISABLED'}!\")\n    print (f\"{HOLD_OUT + N_ROWS - ((CHUNKS - 1) * STRIDE + BATCH_SIZE)} rows goes straight to OOF!\")\n    print (\"Space Req. For One Chunk: {:.2f} GB ({})\".format(\n        (N_FEATURES + 1 if COL_SAMPLING else len(ALL_FEATURES)) * BATCH_SIZE * 64 / 8e+9, 'float64'))\n    \n    # train for TRAIN_FOR number of times\n    for j in range(ALREADY_TRAINED, ALREADY_TRAINED + TRAIN_FOR):\n        \n        front, rear = j * STRIDE, j * STRIDE + BATCH_SIZE\n        print (f\"\\nBeginning chunk {j + 1}/{ALREADY_TRAINED + TRAIN_FOR}! Loading the Feather File to Memory => \")\n        \n        if COL_SAMPLING:\n        \n            FEATURES = (CORE + np.random.choice(list(set(ALL_FEATURES) - set(CORE)), \n                                                size=(N_FEATURES - len(CORE)), \n                                                replace=False).tolist())\n\n            print (f\"Selected Features ({N_FEATURES}/{len(ALL_FEATURES)}): {list(set(FEATURES) - set(CORE))}\")\n            cat_cols = np.intersect1d(['tagF', 'tagS', 'tagT', 'tagL'], FEATURES).tolist()        \n            COLUMNS = ['user_id', *FEATURES, 'answered_correctly']\n\n            if 'timestamp' not in COLUMNS:\n                COLUMNS += ['timestamp']\n                \n        else:\n            \n            FEATURES = ALL_FEATURES\n            COLUMNS = ['user_id', *FEATURES, 'answered_correctly']\n            cat_cols = ['tagF', 'tagS', 'tagT', 'tagL']\n        \n        print (f\"Slicing from {front} to {rear}!\")\n        \n        data = pd.DataFrame()\n        for file, start, end in return_chunk_indices(front, rear, nrows=N_ROWS):\n            \n            data = pd.concat([\n                data,              \n                (pd.read_feather(f\"../input/riiid-final-model-inputs/model_train_c{file}.feather\", columns=COLUMNS)\n                 .iloc[start: end])\n            ])\n            \n        # reset the index to simulate the actual ordering\n        data.index = range(front, rear)\n        \n        # a simple buffer to prevent OOMs\n        gc.collect()\n        time.sleep(3)\n        \n        print (\"Generating train/val Masks for the data...\")\n        mask = generate_train_val(data, return_mask=True, tp=0.95, dyn_p=True).values.astype(bool)\n        print(\"Done generating masks! Saving to intermediate train/val files..\")\n        \n        # remove the indices from oof that are used for model train\n        OOF = np.setdiff1d(OOF, data.index[mask])\n\n        # retain only those columns needed\n        data.drop(['user_id'], axis=1, inplace=True)\n        gc.collect()\n        time.sleep(3)\n        \n        # save it and del from memory to save space\n        data.loc[mask].reset_index(drop=True).to_feather(\"train_intermediate.feather\") \n        data.loc[~mask].reset_index(drop=True).to_feather(\"val_intermediate.feather\")\n\n        del data, mask\n        gc.collect()\n        time.sleep(3)\n        \n        ####### LOGIC FOR CREATING NUMPY ARRAY WITHOUT MEMORY SPIKE #######\n        \n        print (\"Done saving! Beginning logic for LGB dataset creation..\")\n        \n        chunk_per = 0.025\n        \n        # load and reorder the columns\n        tr_data = pd.read_feather(\"train_intermediate.feather\")\n        tr_data = tr_data[FEATURES + ['answered_correctly']]\n\n        ## initialize an emtpy array to which we would be assigning\n        tr_result = np.empty(shape=(len(tr_data), len(FEATURES) + 1), dtype='float64')\n        step = int(len(tr_data) * chunk_per)\n        end = (len(tr_data) // step) + 1\n\n        for i in tqdm.tqdm(range(1, end + 1)):\n\n            # assing to numpy slice and remove dataframe\n            tr_result[(i - 1) * step: i * step] = tr_data.iloc[:step]\n            tr_data = tr_data.iloc[step:]\n\n            # clear memory and let it go for a while unused\n            gc.collect()\n            time.sleep(3)\n\n        del tr_data\n        gc.collect()\n        ! rm ./train_intermediate.feather\n        time.sleep(3)\n        \n        # create LGB dataset\n        tr_data = lgb.Dataset(\n            tr_result[:, :-1], label=tr_result[:, -1], feature_name=FEATURES, \n            categorical_feature=cat_cols, free_raw_data=not ITERATIVE_TRAIN)\n        \n        # construct the dataset (Memory Bottleneck)\n        tr_data.construct()\n        \n        del tr_result\n        gc.collect()\n        time.sleep(2)\n        \n        # load and reorder the columns\n        vl_data = pd.read_feather(\"./val_intermediate.feather\")\n        vl_data = vl_data[FEATURES + ['answered_correctly']]\n\n        #  Val Data Next\n        ## initialize an emtpy array to which we would be assigning\n        vl_result = np.empty(shape=(len(vl_data), len(FEATURES) + 1), dtype='float64')\n        step = int(len(vl_data) * chunk_per)\n        end = (len(vl_data) // step) + 1\n\n        for i in tqdm.tqdm(range(1, end + 1)):\n\n            # assign to numpy slice and remove dataframe\n            vl_result[(i - 1) * step: i * step] = vl_data.iloc[:step]\n            vl_data = vl_data.iloc[step:]\n\n            # clear memory and let it go for a while unused\n            gc.collect()\n            time.sleep(3)\n\n        del vl_data\n        gc.collect() \n        ! rm ./val_intermediate.feather\n        time.sleep(3)\n        \n        vl_data = lgb.Dataset(\n            vl_result[:, :-1], label=vl_result[:, -1], feature_name=FEATURES, \n            categorical_feature=cat_cols, reference=tr_data, \n            free_raw_data=not ITERATIVE_TRAIN)\n        \n        # construct the dataset (Memory Bottleneck)\n        vl_data.construct()        \n        \n        # no longer needed\n        del vl_result\n        gc.collect()\n        time.sleep(2)\n\n        ###################################################################\n\n        print (\"Done generating LGB datasets! Starting training..\")\n\n        model = lgb.train(\n            \n            params={'metric': 'auc', 'objective': 'binary', \n                    'num_leaves': 200, 'learning_rate': 0.075},\n            \n            train_set=tr_data,\n            feature_name=FEATURES,\n            categorical_feature=cat_cols,\n            num_boost_round=N_ROUNDS,\n            valid_sets=[tr_data, vl_data], \n            init_model=pre_model,\n            early_stopping_rounds=50,\n            verbose_eval=50\n        )\n        \n        del tr_data, vl_data\n        gc.collect()\n        \n        print (f\"Done training chunk {j + 1}!\")\n        \n        # save the model & append to list\n        model.save_model(f\"trained_model_c{j + 1}.txt\")\n        models.append(model)\n        \n        # reassign pre_model for iterative training\n        if ITERATIVE_TRAIN:\n            pre_model = model\n        \nif OOF_VALIDATE:\n        \n    ##################### MODEL(S) EVALUATION LOGIC #####################\n    \n    print (f\"\\nWe would be evaluating the model on {HOLD_OUT} rows out of {len(OOF)} OOF rows.\")\n    data = pd.read_feather(LOC3).iloc[-HOLD_OUT:]\n\n    actual = data['answered_correctly']\n    data.drop(['user_id', 'answered_correctly'], axis=1, inplace=True)\n\n    # empty numpy array for saving preds\n    preds = np.zeros(shape=(len(actual), len(models)))\n\n    print (\"\\nIndividual Model's OOF Scores:\")\n    for i in range(len(models)):\n        FEATURES = models[i].feature_name()\n        cat_cols = np.intersect1d(['tagF', 'tagS', 'tagT', 'tagL'], FEATURES)\n        preds[:, i] = models[i].predict(data[FEATURES], categorical_feature=cat_cols)\n        print(f\"\\tModel #{i+1}'s Score: {roc_auc_score(actual, preds[:, i]):.4f}\")\n\n    print (f\"\\nThe Ensemble's Score: {roc_auc_score(actual, preds.mean(1)):.4f}\")\n    \n    # clear memory after running, not needed\n    del data, actual, preds\n    gc.collect()\n    \n# only retain last model if iterative training            \nif ITERATIVE_TRAIN:\n    models = models[-1:]\n    \n# save OOF for future training & evaluation\nnp.save(\"OOF.npy\", OOF)\nprint (f\"\\nNumber of Models: {len(models)}\\n\")","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"A custom class to make ensemble predictions:"},{"metadata":{"trusted":true},"cell_type":"code","source":"class ensemble(object):\n    def __init__(self, models, weights=None, treelite=True):\n        self.n_models = len(models)\n        self.models = models\n        self.weights = [1 / self.n_models] * self.n_models if not weights else weights\n        self.cat_cols = ['tagF', 'tagS', 'tagL', 'tagT']\n        self.treelite = treelite\n        \n        # specify feature order\n        self.feats = ['repeat_c', 'tagF', 'tagS', 'tagL', 'tagT', 'response_time', 'prior_question_elapsed_time',\n                       'up_mean', 'up_count', 'uq_per_hr', 'uwrong_sum', 'lec_recent', 'pqet_mean', 'seen_ratio',\n                       'tmed', 'up_recency', 'ts_recency_10', 'ts_recency_5', 'timestamp', 'task_container_id',\n                       'content_c', 'che_sum', 'seen_exp_when_wrong', 'seen_exp_when_right', 'sessions', \n                      'sess_event_count', 'pqetmr_10', 'uf_bundle', 'u_mean', 'ummr_10', 'rt_per_task', 'c_mean',\n                       'c_std', 'h_mean', 'uph_mean', 'wrong_et_med', 'right_et_med']\n        \n        if not self.treelite:\n            assert len(np.setdiff1d(self.feats, self.models[0].feature_name())) == 0\n    \n    def __getattr__(self, attr):\n        '''Solely to prevent any errors by mistaken accessing in future.\n        Features such as feature_importances are not reliable!'''\n        \n        return getattr(self.models[0], attr)\n    \n    def feature_name(self):\n        return self.feats\n    \n    def predict(self, x, **kwargs):\n        \n        if not self.treelite:\n            return self.predict_lgb(x[self.feats])\n        \n        else:\n            return self.predict_treelite(x[self.feats].values)\n        \n    def predict_treelite(self, x):\n        \n        # cast to treelite compatible batch\n        X = treelite_runtime.Batch.from_npy2d(x)\n        \n        pred = np.zeros(len(x))\n        for m, w in zip(self.models, self.weights):\n            pred += m.predict(X) * w\n            \n        return pred\n        \n    def predict_lgb(self, x):\n        \n        pred = np.zeros(len(x))\n        for m, w in zip(self.models, self.weights):\n            pred += m.predict(x, categorical_feature=self.cat_cols) * w\n            \n        return pred","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"#### FTRL Model Training:\nThis part has been added from my other notebook. This part trains the FTRL model over the LGB ensemble. We feed in the LGB predictions. This is a bottleneck so we load the predictions we had precomputed:"},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\n\nimport glob\n\n# track time passing\nstart_time = time.time()\n\nif not os.path.exists(\"../input/col-sampled-train-dataset/lgb_pred_1.npy\"):\n\n    # we take only a subset of the 8 models for faster approximation\n    models = [models[3], models[6]]\n\n    # Rows taken at a time for model predictions\n    BATCH_SIZE = int(3.0e7)  \n\n    CHUNKS = list(range(0, N_ROWS, BATCH_SIZE))\n\n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | Predictions will be chunked for\", len(CHUNKS), \"Chunks!\")\n    print (CHUNKS)\n\n    for j, front in enumerate(CHUNKS):\n        rear = front + BATCH_SIZE\n\n        print (f\"\\nBeginning chunk {j+1}! Loading the Feather File to Memory => \")    \n        print (f\"Slicing from {front} to {rear}!\")\n\n        data = pd.DataFrame()\n        for file, start, end in return_chunk_indices(front, rear, nrows=N_ROWS):\n\n            file = f\"../input/riiid-final-model-inputs/model_train_c{file}.feather\"\n\n            if not os.path.exists(file):\n                continue\n\n            data = pd.concat([data, pd.read_feather(file).iloc[start: end]])\n\n        gc.collect()\n        time.sleep(2)\n\n        # rearange columns (only keep those needed)\n        data = data[LGB_COLS]\n\n        # compute len before we create treelite batch\n        data_shape = data.shape\n\n        print (f\"Time Elapsed: {time.time() - start_time:10.2f} s |\", end=' ')\n        print (f\"Done loading! Data Dimensions: {data_shape}\")\n        print (\"Beginning logic for treelite batch dataset creation..\")\n\n        gc.collect()\n        time.sleep(2)\n\n        chunk_per = 0.025\n\n        ## initialize an emtpy array to which we would be assigning\n        darray = np.empty(shape=(data_shape[0], data_shape[1]), dtype='float64')\n        step = int(data_shape[0] * chunk_per)\n        end = (data_shape[0] // step) + 1\n\n        for i in tqdm.tqdm(range(1, end + 1)):\n\n            # assing to numpy slice and remove dataframe\n            darray[(i - 1) * step: i * step] = data.iloc[:step]\n            data = data.iloc[step:]\n\n            # clear memory and let it go for a while unused\n            gc.collect()\n            time.sleep(3)\n\n        del data\n        gc.collect()\n        time.sleep(3)\n\n        # convert to tlite batch for prediction\n        darray = treelite_runtime.Batch.from_npy2d(darray)\n\n        # a simple buffer to prevent OOMs\n        gc.collect()\n        time.sleep(3)\n\n        lgb_pred = np.zeros(shape=data_shape[0], dtype='float32')\n\n        print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | Making Predictions using Treelite models =>\")\n\n        for k, model in enumerate(models):\n\n            lgb_pred = lgb_pred + model.predict(darray)\n\n            print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | Done with model# {k + 1}\")\n\n        # compute mean \n        lgb_pred = lgb_pred / len(models)\n\n        print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | Done with Chunk {j+1}! Saving predictions to Disk!\")\n        np.save(f\"lgb_pred_{j+1}.npy\", lgb_pred)\n\n        del darray, lgb_pred\n        gc.collect() \n        \nelse:\n    print(f\"Time Elapsed: {time.time() - start_time:10.2f} s | Copying LGB pred numpy arrays back to Disk!\")\n    for i in glob.glob(\"../input/col-sampled-train-dataset/lgb_pred_*\"):\n        ! cp {i} {i.split(\"/\")[-1]}\n    print(f\"Time Elapsed: {time.time() - start_time:10.2f} s | Done copying to Disk!\")","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\n\n# track time passing\nstart_time = time.time()\n\nif not os.path.exists(\"../input/col-sampled-train-dataset/lgb2_pred_1.npy\"):\n\n    # we take only a subset of the 8 models for faster approximation\n    models = [models[1], models[4]]\n\n    # Rows taken at a time for model predictions\n    BATCH_SIZE = int(3.0e7)  \n\n    CHUNKS = list(range(0, N_ROWS, BATCH_SIZE))\n\n    print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | Predictions will be chunked for\", len(CHUNKS), \"Chunks!\")\n    print (CHUNKS)\n\n    for j, front in enumerate(CHUNKS):\n        rear = front + BATCH_SIZE\n\n        print (f\"\\nBeginning chunk {j+1}! Loading the Feather File to Memory => \")    \n        print (f\"Slicing from {front} to {rear}!\")\n\n        data = pd.DataFrame()\n        for file, start, end in return_chunk_indices(front, rear, nrows=N_ROWS):\n\n            file = f\"../input/riiid-final-model-inputs/model_train_c{file}.feather\"\n\n            if not os.path.exists(file):\n                continue\n\n            data = pd.concat([data, pd.read_feather(file).iloc[start: end]])\n\n        gc.collect()\n        time.sleep(2)\n\n        # rearange columns (only keep those needed)\n        data = data[LGB_COLS]\n\n        # compute len before we create treelite batch\n        data_shape = data.shape\n\n        print (f\"Time Elapsed: {time.time() - start_time:10.2f} s |\", end=' ')\n        print (f\"Done loading! Data Dimensions: {data_shape}\")\n        print (\"Beginning logic for treelite batch dataset creation..\")\n\n        gc.collect()\n        time.sleep(2)\n\n        chunk_per = 0.025\n\n        ## initialize an emtpy array to which we would be assigning\n        darray = np.empty(shape=(data_shape[0], data_shape[1]), dtype='float64')\n        step = int(data_shape[0] * chunk_per)\n        end = (data_shape[0] // step) + 1\n\n        for i in tqdm.tqdm(range(1, end + 1)):\n\n            # assing to numpy slice and remove dataframe\n            darray[(i - 1) * step: i * step] = data.iloc[:step]\n            data = data.iloc[step:]\n\n            # clear memory and let it go for a while unused\n            gc.collect()\n            time.sleep(3)\n\n        del data\n        gc.collect()\n        time.sleep(3)\n\n        # convert to tlite batch for prediction\n        darray = treelite_runtime.Batch.from_npy2d(darray)\n\n        # a simple buffer to prevent OOMs\n        gc.collect()\n        time.sleep(3)\n\n        lgb_pred = np.zeros(shape=data_shape[0], dtype='float32')\n\n        print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | Making Predictions using Treelite models =>\")\n\n        for k, model in enumerate(models):\n\n            lgb_pred = lgb_pred + model.predict(darray)\n\n            print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | Done with model# {k + 1}\")\n\n        # compute mean \n        lgb_pred = lgb_pred / len(models)\n\n        print (f\"Time Elapsed: {time.time() - start_time:10.2f} s | Done with Chunk {j+1}! Saving predictions to Disk!\")\n        np.save(f\"lgb2_pred_{j+1}.npy\", lgb_pred)\n\n        del darray, lgb_pred\n        gc.collect() \n        \nelse:\n    print(f\"Time Elapsed: {time.time() - start_time:10.2f} s | Copying LGB pred numpy arrays back to Disk!\")\n    for i in glob.glob(\"../input/col-sampled-train-dataset/lgb_pred_*\"):\n        ! cp {i} {i.split(\"/\")[-1]}\n    print(f\"Time Elapsed: {time.time() - start_time:10.2f} s | Done copying to Disk!\")","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Creating dataset for FTRL training:"},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\n\ndata = pd.read_feather(\n    \"../input/riiid-train-data-multiple-formats/riiid_train.feather\", \n    columns=['content_id', 'content_type_id'])\n\ndata = data.loc[~data['content_type_id']].iloc[:N_ROWS]\n\ndata = data.merge(\n    ques[['question_id', 'part']].set_index(\"question_id\"), \n    left_on=['content_id'], right_index=True, how='left')\n\n# drop unncessary columns\ndata.drop(['content_type_id'], axis=1, inplace=True)\n\n# reset the indices\ndata.reset_index(drop=True, inplace=True)\n\n# temp save to save memory\ndata.to_feather(\"./intermediate_train.feather\")\ndel data\ngc.collect()\n\ndata = pd.DataFrame()\nfor LOC in [LOC1, LOC2, LOC3]:\n    \n    data = pd.concat([\n        data, \n        \n        (pd.read_feather(LOC, columns=['user_id', 'task_container_id', 'content_c', 'c_mean', 'h_mean', \n                                    'answered_correctly', 'ummr_10']))\n    ], axis=0)\n\n# reset the indices to match\ndata.reset_index(drop=True, inplace=True)\n\n# we create some meta features to help FTRL\ndata['ummr_10_50'] = data['ummr_10'] > 0\ndata['h_mean_50'] = data['h_mean'] > 0.5\ndata['c_mean_50'] = data['c_mean'] > 0.5\ndata['c_mean_25'] = data['c_mean'] > 0.25\ndata['c_mean_75'] = data['c_mean'] > 0.75\n\n# delete those not needed\ndata.drop(['ummr_10', 'c_mean', 'h_mean'], axis=1, inplace=True)\n\n# lets now concatenate both the frames\ndata = pd.concat([\n    data, pd.read_feather(\"./intermediate_train.feather\")\n], axis=1)\n\n# delete intermediate file\n! rm ./intermediate_train.feather\n\nlgb_pred = np.array([])\n\nfor file in sorted(glob.glob(\"../input/col-sampled-train-dataset/lgb_pred_*\")):\n    lgb_pred = np.concatenate([lgb_pred, np.load(file)])\n    \nlgb2_pred = np.array([])\n\nfor file in sorted(glob.glob(\"../input/col-sampled-train-dataset/lgb2_pred_*\")):\n    lgb2_pred = np.concatenate([lgb2_pred, np.load(file)])\n    \nassert len(lgb_pred) == len(lgb2_pred)\n\n# we divide them by two \nlgb_pred = (lgb_pred + lgb2_pred) / 2\n\ndel lgb2_pred\ngc.collect()\n\n# save the loaded values to new column\ndata['lgb_pred'] = lgb_pred\ndata['lgb_75'] = data['lgb_pred'] > 0.75\ndata['lgb_50'] = data['lgb_pred'] > 0.5\ndata['lgb_25'] = data['lgb_pred'] > 0.25\n\ndel lgb_pred\ngc.collect()\n\nprint (data.shape)\nprint (data.columns)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Uint8, uint16 can't work with datatable frames. We need to type cast them. let's write a utility function for that:"},{"metadata":{"trusted":true},"cell_type":"code","source":"def df_to_dt_format(df):\n    for i in df.columns:\n        org = str(df[i].dtype)\n        converted = org.lstrip(\"u\")\n        if org != converted:\n            converted = converted[:3] + str(int(converted.lstrip(\"int\")) * 2)\n            df[i] = df[i].astype(converted)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# type cast in place\ndf_to_dt_format(data)\n\n# convert to datatable frame for training\ndata = dt.Frame(data)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"FTRL_COLS = [\n    'user_id', 'task_container_id', 'content_c', 'ummr_10_50', 'h_mean_50', \n    'c_mean_50', 'c_mean_25', 'c_mean_75', 'content_id', 'part', 'lgb_pred', 'lgb_75',\n    'lgb_50', 'lgb_25'\n]\n\nINTERACTIONS = None","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Train FTRL Model. Notice the speed at which FTRL trains!"},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\n\nfrom datatable.models import Ftrl\n\nftrl = Ftrl(\n    nepochs=1, interactions=INTERACTIONS, \n    alpha=0.005, double_precision=True,\n)\n\nftrl.fit(data[:int(9.5e7), FTRL_COLS], data[:int(9.5e7), ['answered_correctly']])","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"#### FTRL Evaluation:\n\nWithout online Learning:\n1. With only the 4 model predictions:"},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time \n\n# make predictions for validation\npreds = ftrl.predict(data[int(9.5e7):, FTRL_COLS]).to_pandas()\nactual = data[int(9.5e7):, 'answered_correctly'].to_numpy()\n\n# Base line to beat\nprint (\"Baseline Score to beat: {:.4f}\".format(roc_auc_score(\n    actual, data[int(9.5e7):, 'lgb_pred'].to_pandas()\n)))\n\n# FTRL score\nroc_auc_score(actual, preds)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"2. With all 8 model predictions.\n\nAt model inference we need to run the model using all the LGB models we have trained. So lets do an evaluation that would mimic that scenario:"},{"metadata":{"trusted":true},"cell_type":"code","source":"model = ensemble(models, treelite=False)\nmodel, type(model.models[0])","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"if not os.path.exists(\"../input/treelite-converted-ensemble-8-models-25m/lgb_pred_final.npy\"):\n    # we make predictions using the ensemble model\n    temp = pd.read_feather(LOC3).iloc[-(N_ROWS - int(9.5e7)):][model.feature_name()]\n    lgb_pred = model.predict(temp)\n\n    del temp\n    gc.collect()\n\nelse:\n    lgb_pred = np.load(\"../input/treelite-converted-ensemble-8-models-25m/lgb_pred_final.npy\")","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Our goal here is to make the last peice of data alone is uses all the LGB model predictions rather than just 4 of them:"},{"metadata":{"trusted":true},"cell_type":"code","source":"data[int(9.5e7):, 'lgb_pred'] = lgb_pred\ndata[int(9.5e7):, 'lgb_75'] = data[int(9.5e7):, dt.f.lgb_pred > 0.75] \ndata[int(9.5e7):, 'lgb_50'] = data[int(9.5e7):, dt.f.lgb_pred > 0.50] \ndata[int(9.5e7):, 'lgb_25'] = data[int(9.5e7):, dt.f.lgb_pred > 0.25] \n\n# make our predictions\npreds = ftrl.predict(data[int(9.5e7):, FTRL_COLS])\n\n# Base line to beat\nprint (\"Baseline Score to beat: {:.4f}\".format(roc_auc_score(\n    actual, data[int(9.5e7):, 'lgb_pred'].to_pandas()\n)))\n\n# FTRL score\nroc_auc_score(data[int(9.5e7):, 'answered_correctly'].to_pandas(), preds)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"We use FTRL for its ability to learn from test instances as well. So evaluation would be better performed if the model is trained and predicted batchwise. Let's do that:"},{"metadata":{"trusted":true},"cell_type":"code","source":"from copy import deepcopy\n\n# a save point\nol_ftrl = deepcopy(ftrl)\n\n# reduce learning rate\nol_ftrl.alpha = 0.005","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time \n\nol_preds = []\npreds = []\n\n# smaller batch sizes would give more closer approximation\nBATCH_SIZE = 1000\n\nfor front in range(int(9.5e7), N_ROWS, BATCH_SIZE):\n    pred = ol_ftrl.predict(data[front:front+BATCH_SIZE, FTRL_COLS]).to_list()[0]\n    ol_preds.append(pred)\n    \n    pred = ftrl.predict(data[front:front+BATCH_SIZE, FTRL_COLS]).to_list()[0]\n    preds.append(pred)\n    \n    # learn from the batches\n    ol_ftrl.fit(data[front:front+BATCH_SIZE, FTRL_COLS], data[front:front+BATCH_SIZE, 'answered_correctly'])","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# lets compute how they have performed\nactual = data[int(9.5e7):, 'answered_correctly'].to_numpy()\n\n# Base line to beat\nprint (\"Baseline Score to beat: {:.4f}\".format(roc_auc_score(\n    actual, data[int(9.5e7):, 'lgb_pred'].to_pandas()\n)))\n\nprint (\"\\nModel Score Comparison: Online: {:.4f} | Offline: {:.4f}\".format(\n    roc_auc_score(actual, np.concatenate(ol_preds)),\n    roc_auc_score(actual, np.concatenate(preds))\n))","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"### Model State Generation:\nLoad the models (Choose between LGB or Treelite)"},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time \n\nTREELITE = False\n\nif TREELITE:\n    models = []\n    for model in sorted(glob.glob(\"../input/treelite-converted-ensemble-8-models-25m/tl_*.so\")):\n        models.append(treelite_runtime.Predictor(model, verbose=False, nthread=1))\n    \nelse:\n    models = []\n    for file in sorted(glob.glob(\"../input/riiid-final-model-inputs/trained_model_*.txt\")):\n        models.append(lgb.Booster(model_file=file))\n\nmodel = ensemble(models, treelite=TREELITE)\n\nmodel, model.n_models, type(model.models[0])","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Generate the states for all the users and contents:"},{"metadata":{"trusted":true},"cell_type":"code","source":"# set to false in case the dataset changes\nload_from_file = True\ncat_cols = ['tagF', 'tagS', 'tagT', 'tagL']","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"start_time = time.time()\n\nif not load_from_file:    \n    \n    pq_shifted = pd.read_feather(\n        \"../input/riiid-train-data-multiple-formats/riiid_train.feather\", \n        columns=['user_id', 'task_container_id', 'content_type_id', \n                 'prior_question_elapsed_time', 'prior_question_had_explanation'])\n    \n    # question mask for future use\n    q_mask = pq_shifted['content_type_id'] == 0\n    \n    # filter out the lectures\n    pq_shifted = pq_shifted.loc[q_mask] \n    \n    # type cast to save memory and space\n    pq_shifted['prior_question_had_explanation'] = (\n        pq_shifted['prior_question_had_explanation']\n        .fillna(False).astype('uint8'))\n    \n    # we take only one value per task_container_id, user_id pair\n    temp = (pq_shifted.groupby(['user_id', 'task_container_id'])\n            [['prior_question_elapsed_time', 'prior_question_had_explanation']]\n            .mean())\n    \n    temp = temp.groupby(\"user_id\").shift(-1)\n    temp.columns = ['pqet_shifted', 'pqhe_shifted']\n    \n    # drop, no longer useful\n    pq_shifted.drop(\n        ['content_type_id', 'prior_question_elapsed_time', 'prior_question_had_explanation'], \n        axis=1, inplace=True)\n    \n    gc.collect()\n    \n    pq_shifted = pq_shifted.merge(temp, left_on=['user_id', 'task_container_id'], right_index=True, how='left')\n    pq_shifted.drop(['user_id'], axis=1, inplace=True)\n    \n    # saving it for futher use down the line\n    pq_shifted['pqet_shifted'] = pq_shifted['pqet_shifted'].astype('float32')\n    pq_shifted['pqhe_shifted'] = pq_shifted['pqhe_shifted'].fillna(False).astype(bool)\n    pq_shifted.reset_index().to_feather(\"./pq_shifted.feather\")\n    \n    content_df = (pd.read_feather(\n        \"../input/riiid-train-data-multiple-formats/riiid_train.feather\",\n        columns=['content_id', 'answered_correctly']).loc[q_mask])\n    \n    content_df = pd.concat([content_df, pq_shifted], axis=1)\n    \n    del temp, pq_shifted\n    gc.collect()\n    \n    # create the content mean response_time for right and wrong answers    \n    et_med = content_df.groupby([\"content_id\", 'answered_correctly'])['pqet_shifted'].median()\n    et_med = et_med.unstack()\n    et_med.columns = ['wrong_et_med', 'right_et_med']\n    \n    # create the che_sum for all contents\n    che_sum = content_df.groupby(['content_id'])['pqhe_shifted'].sum().rename(\"che_sum\")\n    \n    # create the task_container_id median\n    tmed = content_df.groupby('content_id')['task_container_id'].median().rename(\"tmed\")\n\n    # save the content_mean\n    content_df = (content_df.groupby('content_id')['answered_correctly'].agg(\n        content_c='count', c_mean='mean', c_std='std'))\n    \n    # merge the response_times and c_mean\n    content_df = pd.concat([content_df, et_med, tmed, che_sum], axis=1)\n    \n    del et_med, tmed, che_sum\n    gc.collect()\n\n    # save the categorical columns required\n    content_df = content_df.merge(\n        ql.loc[~ql.content_type_id, ['content_id', *cat_cols, 'part', 'bundle_q_count', 'bundle_id']]\n        .set_index(\"content_id\"), on='content_id', how='left')\n    \nelse:\n    \n    content_df = pd.read_csv(\"../input/riiid-final-model-inputs/content-df.csv\", index_col=0)\n\n# set size to minimum possible for saving space\ncontent_df[cat_cols + ['bundle_q_count', 'part']] = content_df[cat_cols + ['bundle_q_count', 'part']].astype('uint8')\ncontent_df[['wrong_et_med', 'right_et_med']] = content_df[['wrong_et_med', 'right_et_med']].astype('float32')\ncontent_df[['c_mean', 'c_std']] = content_df[['c_mean', 'c_std']].astype('float32')\ncontent_df[['content_c', 'tmed', 'che_sum', 'bundle_id']] = (\n    content_df[['content_c', 'tmed', 'che_sum', 'bundle_id']]).astype('uint16')\n\nprint (f\"Time Elapsed: {(time.time() - start_time):6.2f} Sec\")\nprint (f\"Memory Usage: {content_df.memory_usage(deep=True).sum() / (2**20):6.2f}\", 'MB')\nprint (f\"Length      : {len(content_df):7}\")\n\ncontent_df.sample(5)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"start_time = time.time()\nLEC_RECENT_ROLL = 10\nSESSION_DURATION = 15 * 60 * 1000\n\nif not load_from_file:\n\n    user_df = (pd.read_feather(\n        \"../input/riiid-train-data-multiple-formats/riiid_train.feather\",\n        columns=['user_id', 'answered_correctly', 'timestamp', 'content_id', \n                 'prior_question_elapsed_time', 'prior_question_had_explanation']))\n    \n    # merge with part to be able to gather part wise statistics\n    user_df = user_df.merge(\n        content_df[['part', 'c_mean', 'bundle_id']], left_on='content_id', \n        right_index=True, how='left')\n    \n    # create the uf_bundle feature\n    uf_bundle = user_df.groupby(\"user_id\")['bundle_id'].first().rename(\"uf_bundle\")\n    user_df.drop(['bundle_id', 'content_id'], axis=1, inplace=True)\n    \n    # before fitering out the lectures, we need to calc the timestamp\n    # based features, once done we filter them out\n    \n    user_df['sessions'] = user_df.groupby(\"user_id\")['timestamp'].transform(lambda x: rt_func(x.values))\n    user_df['sessions'] = (user_df['sessions'] > SESSION_DURATION).astype('uint8')\n    \n    sess = pd.concat([\n        user_df.groupby('user_id')['sessions'].sum().rename(\"sessions\"),    \n        user_df.groupby('user_id')['sessions'].agg(\n            lambda x: np.argmax(x.values[::-1]) if x.any() else len(x) - 1).rename(\"sess_event_count\")\n    ], axis=1)\n    \n    user_ts = user_df.groupby(['user_id', 'part'])['timestamp'].max().unstack().fillna(0)\n    user_ts.columns = [f'ts_max_{i}' for i in range(1, 8)]\n    \n    user_ts = pd.concat([\n        user_df.groupby(\"user_id\")['timestamp'].agg(ts_max='max', uic='count'), \n        user_ts\n    ], axis=1)\n    \n    # drop sessions, no longer needed\n    user_df.drop(['sessions', 'timestamp'], axis=1, inplace=True)\n    \n    # create the lec_recent_feature\n    lec_recent = user_df[['user_id', 'answered_correctly']].groupby(\"user_id\").tail(LEC_RECENT_ROLL)\n    lec_recent['content_type_id'] = lec_recent['answered_correctly'].map({-1: 1, 0: 0, 1: 0})\n    lec_recent.drop(\"answered_correctly\", axis=1, inplace=True)\n    lec_recent = (\n        lec_recent.groupby('user_id')['content_type_id']\n        .apply(lambda x: np.where(x)[0][-1] + 1 if x.any() else 0)\n        .rename(\"lec_recent\")\n    )\n    \n    ############ TIME STAMP BASED FEATURE GENERATION ENDS HERE ############\n    ## filtering out the lectures\n    user_df = user_df[user_df.answered_correctly != -1]\n    \n    # casting pqhe for successful aggregation\n    user_df['prior_question_had_explanation'] = user_df['prior_question_had_explanation'].astype(bool).fillna(False)\n    \n    # generate the pq features\n    pq_sum = user_df.groupby(\"user_id\")[['prior_question_had_explanation', 'prior_question_elapsed_time']].sum()\n    \n    # drop these features, no longer needed\n    user_df.drop(['prior_question_had_explanation', 'prior_question_elapsed_time'], axis=1, inplace=True)\n    gc.collect()\n    \n    # loading pqhe_shifted from save file, can concat only with filtered df\n    seen_exp = pd.read_feather(\"./pq_shifted.feather\", columns=['index', 'pqhe_shifted']).set_index('index')\n    user_df = pd.concat([user_df, seen_exp], axis=1)\n    \n    # creating seen_exp_when_* features\n    seen_exp = user_df.groupby(['user_id', 'answered_correctly'])['pqhe_shifted'].sum().unstack()\n    seen_exp.columns = ['seen_exp_when_wrong', 'seen_exp_when_right']\n    seen_exp.fillna(0, inplace=True)\n    user_df.drop(['pqhe_shifted'], axis=1, inplace=True)\n    \n    # calculate the sum of answered_correctly for h_mean later on, User Correct Sum\n    u_crct_sum = user_df.groupby(\"user_id\")['answered_correctly'].sum().rename(\"ucs\")\n    \n    # calculate up_org_sum (unmodified part wise) for uph_mean later on\n    up_org_sum = user_df.groupby(['user_id', 'part'])['answered_correctly'].agg('sum').unstack()\n    up_org_sum = up_org_sum.astype(\"float32\").fillna(0)\n    \n    # modify the answered_correctly (we are going to weigh it based on c_mean)\n    user_df['answered_correctly'] = modify_ac(user_df['answered_correctly'].values, user_df['c_mean'].values)\n    \n    # drop the timestamp, and c_mean no longer needed\n    user_df.drop([\"c_mean\"], axis=1, inplace=True)\n    gc.collect()\n    \n    user_df = user_df.groupby(['user_id', 'part'])['answered_correctly'].agg(['sum', 'count']).unstack()\n    \n    # impute missing values before adding\n    user_df = user_df.astype('float32').fillna(0)\n    user_df['u_sum'] = user_df['sum'].sum(1)\n    user_df['u_count'] = user_df['count'].sum(1)\n\n    # concat ts and u related data\n    user_df = pd.concat([\n        user_ts, sess, lec_recent, up_org_sum, \n        user_df, uf_bundle, u_crct_sum, \n        pq_sum, seen_exp\n    ], axis=1)\n    \n    del user_ts, pq_sum, u_crct_sum, lec_recent\n    del uf_bundle, sess, seen_exp\n    gc.collect()\n    \n    # rename the aggregated columns\n    user_df.columns = (\n        ['ts_max', 'uic'] + [f'ts_max_{i}' for i in range(1, 8)] + \n        ['sess_max', 'sess_event_sum'] + ['lec_recent'] +\n        [f'ups{i}' for i in range(1, 8)] + \n        [f's{i}' for i in range(1, 8)] + \n        [f'c{i}' for i in range(1, 8)] + \n        ['us', 'uc'] + ['uf_bundle'] + \n        ['ucs'] + ['pqhe_sum', 'pqet_sum'] + \n        ['seen_exp_when_wrong', 'seen_exp_when_right'])\n    \nelse:\n    \n    user_df = pd.read_csv(\"../input/riiid-final-model-inputs/user-df.csv\", index_col=0)\n\n## type cast to save space\nuser_df['lec_recent'] = user_df['lec_recent'].astype('uint8')\nuser_df[[f'ups{i}' for i in range(1, 8)]] = user_df[[f'ups{i}' for i in range(1, 8)]].astype(\"uint16\")\nuser_df[[f'c{i}' for i in range(1, 8)]] = user_df[[f'c{i}' for i in range(1, 8)]].astype('uint16')\nuser_df[['uic', 'uc', 'ucs', 'pqhe_sum']] = user_df[['uic', 'uc', 'ucs', 'pqhe_sum']].astype(\"uint16\")\nuser_df[['sess_event_sum', 'sess_max']] = user_df[['sess_event_sum', 'sess_max']].astype(\"uint16\")\nuser_df[['seen_exp_when_wrong', 'seen_exp_when_right', 'uf_bundle']] = (\n    user_df[['seen_exp_when_wrong', 'seen_exp_when_right', 'uf_bundle']].astype(\"uint16\"))\n\nprint (f\"Time Elapsed: {(time.time() - start_time):7.2f} Sec\")\nprint (f\"Memory Usage: {user_df.memory_usage(deep=True).sum() / (2**20):7.2f}\", 'MB')\nprint (f\"Length      : {len(user_df):8}\")\n\nuser_df.sample(5).T","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"\nPandas apply would take a lot of time. So let's try to parallelize them using multiprocessing. Source: https://stackoverflow.com/questions/26784164/pandas-multiprocessing-apply"},{"metadata":{"trusted":true},"cell_type":"code","source":"from multiprocessing import  Pool\nfrom functools import partial\nimport sys\n\nmax_q = ques.question_id.max() + 1\n\ndef to_ba(indices, max_q=max_q):\n    'Function to convert indices to bitarray'\n    ba = np.zeros(max_q, dtype=bool)\n    ba[indices] = 1\n    return bitarray(list(ba))\n\ndef parallelize(data, func, num_of_processes=8):\n    data_split = np.array_split(data, num_of_processes)\n    pool = Pool(num_of_processes)\n    data = pd.concat(pool.map(func, data_split))\n    pool.close()\n    pool.join()\n    return data\n\ndef run_on_subset(func, data_subset):\n    return data_subset.apply(func)\n\ndef parallelize_on_rows(data, func, num_of_processes=8):\n    return parallelize(data, partial(run_on_subset, func), num_of_processes)\n\nstart_time = time.time()\n\nif not load_from_file:\n\n    repeat_c = (pd.read_feather(\n        \"../input/riiid-train-data-multiple-formats/riiid_train.feather\",\n        columns=['user_id', 'content_id', 'answered_correctly']))\n\n    repeat_c = repeat_c[repeat_c['answered_correctly'] != -1]\n    repeat_c = repeat_c.groupby(\"user_id\")['content_id'].unique()\n\n    repeat_c = parallelize_on_rows(repeat_c, to_ba)\n    \nelse:\n    \n    repeat_c = pd.read_pickle(\"../input/riiid-final-model-inputs/repeat-c.pkl\")\n\nprint (f\"Time Elapsed: {(time.time() - start_time):6.2f} Sec\")\nprint (f\"Memory Usage: {sys.getsizeof(repeat_c) / (2 ** 20):6.2f} MB\")\nprint (f\"{len(repeat_c)} users have been extracted!\")\n\nrepeat_c.sample(3)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"start_time = time.time()\n\n# alter this window to alter the data gen\nROLL_WINDOW = 10\nUROLL_NULL_FILL = -2\n\nif not load_from_file:\n\n    u_roll = (pd.read_feather(\n        \"../input/riiid-train-data-multiple-formats/riiid_train.feather\",\n        columns=['user_id', 'answered_correctly', 'content_id']))\n    \n    # filtering out the lectures\n    u_roll = u_roll[u_roll.answered_correctly != -1]\n    \n    # retain only last window required number of elements\n    u_roll = u_roll.groupby(\"user_id\").tail(ROLL_WINDOW)\n    \n    # modify answered_correctly\n    u_roll = u_roll.merge(content_df['c_mean'], left_on='content_id', right_index=True, how='left')\n    u_roll['answered_correctly'] = modify_ac(u_roll['answered_correctly'].values, u_roll['c_mean'].values)\n    \n    # drop columns, no longer needed\n    u_roll.drop(['c_mean', 'content_id'], axis=1, inplace=True)\n    gc.collect()\n    \n    # we would be left shifting new responses as we receive\n    # so padding missing users from left\n    u_roll = u_roll.groupby(\"user_id\")['answered_correctly'].apply(\n        lambda x: np.concatenate([np.repeat(UROLL_NULL_FILL, ROLL_WINDOW - len(x)), x.values])\n    )\n    \n    # convert to csv friendly format to save space\n    u_roll = parallelize_on_rows(u_roll, pd.Series)\n    gc.collect()\n    \nelse:\n    \n    u_roll = pd.read_csv(\"../input/riiid-final-model-inputs/u_roll.csv\", index_col=0)\n    \n# perform a type conversion\nu_roll = u_roll.astype(\"float32\")\n\nprint (f\"Time Elapsed: {(time.time() - start_time):7.2f} Sec\")\nprint (f\"Memory Usage: {u_roll.memory_usage(deep=True).sum() / (2**20):7.2f}\", 'MB')\nprint (f\"Length      : {len(u_roll):7}\")\n\nu_roll.sample(5)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"start_time = time.time()\n\n# alter this window to alter the data gen\nROLL_WINDOW_PQET = 10\nPQET_ROLL_NULL_FILL = -1\n\nif not load_from_file:\n\n    pqet_roll = (pd.read_feather(\n        \"../input/riiid-train-data-multiple-formats/riiid_train.feather\",\n        columns=['user_id', 'prior_question_elapsed_time', 'answered_correctly']))\n    \n    # filtering out the lectures\n    pqet_roll = pqet_roll[pqet_roll.answered_correctly != -1]\n    \n    # drop columns, no longer needed\n    pqet_roll.drop(['answered_correctly'], axis=1, inplace=True)\n    \n    # retain only last window required number of elements\n    pqet_roll = pqet_roll.groupby(\"user_id\").tail(ROLL_WINDOW_PQET)\n    \n    # we would be left shifting new responses as we receive\n    # so padding missing users from left\n    pqet_roll = pqet_roll.groupby(\"user_id\")['prior_question_elapsed_time'].apply(\n        lambda x: np.concatenate([np.repeat(PQET_ROLL_NULL_FILL, ROLL_WINDOW_PQET - len(x)), x.values])\n    )\n    \n    # convert to csv friendly format to save space\n    pqet_roll = parallelize_on_rows(pqet_roll, pd.Series)\n    gc.collect()\n    \n    # fill the missing values (pqet first value for new user is always missing)\n    pqet_roll = pqet_roll.fillna(PQET_ROLL_NULL_FILL)\n    \nelse:\n    \n    pqet_roll = pd.read_csv(\"../input/riiid-final-model-inputs/pqet_roll.csv\", index_col=0)\n    \n# perform a type conversion\npqet_roll = pqet_roll.astype(\"float32\")\n\nprint (f\"Time Elapsed: {(time.time() - start_time):7.2f} Sec\")\nprint (f\"Memory Usage: {pqet_roll.memory_usage(deep=True).sum() / (2**20):7.2f}\", 'MB')\nprint (f\"Length      : {len(pqet_roll):7}\")\n\npqet_roll.sample(5)","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"start_time = time.time()\n\n# alter this window to alter the data gen\nTS_RECENCY_PERIOD = 10\nTS_ROLL_NULL_FILL = np.nan\n\nif not load_from_file:\n\n    ts_roll = (pd.read_feather(\n        \"../input/riiid-train-data-multiple-formats/riiid_train.feather\",\n        columns=['user_id', 'timestamp']))\n    \n    # retain only last window required number of elements\n    ts_roll = ts_roll.groupby(\"user_id\").tail(TS_RECENCY_PERIOD)\n    \n    # we would be left shifting new responses as we receive\n    # so padding missing users from left\n    ts_roll = ts_roll.groupby(\"user_id\")['timestamp'].apply(\n        lambda x: np.concatenate([np.repeat(TS_ROLL_NULL_FILL, TS_RECENCY_PERIOD - len(x)), x.values])\n    )\n    \n    # convert to csv friendly format to save space\n    ts_roll = parallelize_on_rows(ts_roll, pd.Series)\n    gc.collect()\n    \n    # impute missing values with padding\n    ts_roll = ts_roll.fillna(TS_ROLL_NULL_FILL)\n    \nelse:\n    \n    ts_roll = pd.read_csv(\"../input/riiid-final-model-inputs/ts_roll.csv\", index_col=0)\n\nprint (f\"Time Elapsed: {(time.time() - start_time):7.2f} Sec\")\nprint (f\"Memory Usage: {ts_roll.memory_usage(deep=True).sum() / (2**20):7.2f}\", 'MB')\nprint (f\"Length      : {len(ts_roll):7}\")\n\nts_roll.sample(5)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"Write them down to seperate file for faster loading the next time around:"},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time \n\nuser_df.to_csv(\"user-df.csv\")\ncontent_df.to_csv(\"content-df.csv\")\nu_roll.to_csv(\"u_roll.csv\")\nrepeat_c.to_pickle(\"repeat-c.pkl\")\npqet_roll.to_csv(\"pqet_roll.csv\")\nts_roll.to_csv(\"ts_roll.csv\")\n\nif os.path.exists(\"../input/riiid-final-model-inputs/pq_shifted.feather\"):\n    ! cp ../input/riiid-final-model-inputs/pq_shifted.feather ./pq_shifted.feather\n    \nwith open(\"sample-batches.pkl\", 'wb') as f:\n    pickle.dump(batches, f)","execution_count":null,"outputs":[]},{"metadata":{},"cell_type":"markdown","source":"### The End Pipeline:\n\nThe sample test data given does not have lectures in them. This is the reason why *so many* end pipeline fail. We fail to account for the presence of lectures during inference. So we define a simple function that inserts a lecture randomly. We do this only when we are in the *validation Mode*.\n\nOur end pipeline has two modes:\n1. Validaton Mode: When we need to test if our end pipe works as intended. Since once we exhaust the iter_test, it can no longer be used.\n\n2. Prediction Mode: Whenever we need to submit the kernel for evaluation."},{"metadata":{"trusted":true},"cell_type":"code","source":"def insert_lecture(batches):\n    'A simple function to randomly insert a lecture in between, for debugging purposes!'\n    from copy import deepcopy\n    \n    temp = deepcopy(batches)\n    \n    i = np.random.choice(len(temp))\n    j = np.random.choice(len(temp[i][0]))\n    \n    print (f\"Lecture inserted at {i+1} batch at {j} index!\")\n    \n    # assign the content as a lecture\n    temp[i][0].iloc[j, 4] = 1 # content_type_id\n    temp[i][0].iloc[j, 3] = np.random.choice(lectures['lecture_id'].values)\n    temp[i][0].iloc[j, 6] = np.nan\n    temp[i][0].iloc[j, 7] = np.nan\n    \n    if len(temp) > i + 1:\n        pgac = eval(temp[i + 1][0].iloc[0]['prior_group_answers_correct'])\n        pgr = eval(temp[i + 1][0].iloc[0]['prior_group_responses'])\n        pgac[j] = -1\n        pgr[j] = -1\n        \n        temp[i + 1][0].iloc[0, -2] = str(pgac)\n        temp[i + 1][0].iloc[0, -1] = str(pgr)\n        \n    return temp","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"# set to False to bench mark pipeline speeds\nSUBMIT = False\n\nif not SUBMIT:\n    print (\"Validation Mode.\")\n    iter_test = iter(insert_lecture(batches))\n    \n    # list to hold the batch wise processed dataframes\n    op = []\n    \nelse:\n    print (\"Prediction Mode.\")\n    iter_test = env.iter_test()","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"%%time\n\ni, prev_test = 0, tuple()\n\nfor i, batch in enumerate(iter_test):\n\n    ## =========================================================== ##\n    ## ====================== Online Learning ==================== ##\n    ## =========================================================== ##\n\n    if len(prev_test):\n\n        # masks, u_missing, u_present, etc can be reused\n        processed_batch = post_process(prev_test[0], batch[0])\n        \n        ftrl.fit(dt.Frame(prev_test[1])[:, FTRL_COLS], \n                  dt.Frame(processed_batch.loc[q_mask, 'answered_correctly']))\n\n        for _, ts, user, content, ans, pqhe, pqet in (\n            processed_batch[[\n                'timestamp', 'user_id', 'content_id', 'answered_correctly', \n                'prior_question_had_explanation', 'prior_question_elapsed_time']].itertuples()):\n            \n            org_ans = ans\n            \n            if org_ans == -1:\n                part = lectures.loc[lectures.lecture_id == content, 'part'].values[0]\n                c_mean = None\n                \n            else:\n                part = content_df.at[content, 'part']\n                c_mean = content_df.at[content, 'c_mean']\n                \n                # adding weight to ans\n                ans = (1 - c_mean) if ans == 1 else (- c_mean)\n                \n                # alter pqhe, pqet\n                pqet = pqet if not pd.isna(pqet) else np.nan\n                pqhe = int(pqhe) if not pd.isna(pqhe) else 0\n                \n            if user not in user_df.index:\n                \n                # first user interaction is always a lecture, else code couldnt have run\n                # init with zeros or required padding\n                user_df.loc[user] = [0] * user_df.shape[1]\n                ts_roll.loc[user] = ([TS_ROLL_NULL_FILL] * (TS_RECENCY_PERIOD - 1)) + [ts]\n                              \n                repeat_c.at[user] = to_ba([content])\n                u_roll.loc[user] = ([UROLL_NULL_FILL] * (ROLL_WINDOW - 1)) + [ans]\n                \n                pqet_roll.loc[user] = [PQET_ROLL_NULL_FILL] * ROLL_WINDOW_PQET\n\n                # for first user save the bundle_id\n                user_df.at[user, 'uf_bundle'] = content_df.at[content, 'bundle_id'] \n            \n            else:\n                \n                ts_roll.loc[user] = ts_roll.loc[user].values.tolist()[1:] + [ts]\n                \n                if org_ans != -1:\n                    u_roll.loc[user] = u_roll.loc[user].values.tolist()[1:] + [ans]\n                    \n                    temp = repeat_c.at[user]\n                    temp[content] = True\n                    repeat_c.at[user] = temp\n                    \n                    pqet_roll.loc[user] = pqet_roll.loc[user].values.tolist()[1:] + [pqet]\n                    \n            ## update these features regardless of content being a lecture or question                \n            user_df.at[user, 'ts_max'] = ts\n            user_df.at[user, f'ts_max_{int(part)}'] = ts\n            user_df.at[user, 'uic'] = user_df.at[user, 'uic'] + 1\n            \n            temp = (ts - user_df.at[user, 'ts_max']) > SESSION_DURATION # True if it is a new session\n            user_df.at[user, 'sess_max'] = (user_df.at[user, 'sess_max'] + 1 if temp else user_df.at[user, 'sess_max'])\n            user_df.at[user, 'sess_event_sum'] = 0 if temp else user_df.at[user, 'sess_event_sum'] + 1\n                \n            if org_ans == -1: # a lecture\n                \n                user_df.at[user, 'lec_recent'] = LEC_RECENT_ROLL\n                \n            else: # not a lecture\n                \n                temp = user_df.at[user, 'lec_recent']\n                user_df.at[user, 'lec_recent'] = (temp -1) if temp > 0 else 0\n                \n                user_df.at[user, 'uc'] = user_df.at[user, 'uc'] + 1\n                user_df.at[user, 'us'] = user_df.at[user, 'us'] + ans\n\n                user_df.at[user, f'c{part}'] = user_df.at[user, f'c{part}'] + 1\n                user_df.at[user, f's{part}'] = user_df.at[user, f's{part}'] + ans\n\n                user_df.at[user, f'ups{part}'] = user_df.at[user, f'ups{part}'] + org_ans\n                user_df.at[user, 'ucs'] = user_df.at[user, 'ucs'] + org_ans\n                \n                user_df.at[user, 'pqhe_sum'] = user_df.at[user, 'pqhe_sum'] + pqhe\n                \n                # the complication is to replicate the expanding_mean of pandas\n                if np.isnan(pqet): # first time\n                    user_df.at[user, 'pqet_sum'] = np.nan\n                    \n                elif np.isnan(user_df.at[user, 'pqet_sum']): # when accessing the second time\n                    user_df.at[user, 'pqet_sum'] = pqet\n                    \n                else: # after second time\n                    user_df.at[user, 'pqet_sum'] = user_df.at[user, 'pqet_sum'] + pqet\n                    \n                if pqhe: \n                    if org_ans == 1: # right ans and seen exp\n                        user_df.at[user, 'seen_exp_when_right'] = user_df.at[user, 'seen_exp_when_right'] + 1\n                        \n                    else: # if wrong ans and seen exp\n                        user_df.at[user, 'seen_exp_when_wrong'] = user_df.at[user, 'seen_exp_when_wrong'] + 1\n\n    ## =========================================================== ##\n    ## =================== Make the predictions ================== ##\n    ## =========================================================== ##\n    \n    # rows mask (lectures filtered out)\n    q_mask = batch[0]['content_type_id'] == 0\n    \n    # we do a reset_index wthout drop, grp_num is saved\n    pred = batch[0].loc[q_mask,  [\n        'row_id', 'timestamp', 'task_container_id', 'content_id', \n        'user_id', 'prior_question_elapsed_time']].reset_index()\n    \n    # add all the required columns from content_df\n    pred = pd.concat([\n        pred, content_df.reindex(pred['content_id']).reset_index(drop=True)], \n        axis=1)\n    \n    # add u_mean_roll from u_roll\n    pred[f'ummr_{ROLL_WINDOW}'] = u_roll.reindex(pred['user_id']).apply(\n        lambda x: float(np.ma.masked_values(x, UROLL_NULL_FILL).mean()), axis=1).values\n    \n    # add u_mean_roll from u_roll\n    pred[f'pqetmr_{ROLL_WINDOW_PQET}'] = pqet_roll.reindex(pred['user_id']).apply(\n        lambda x: float(np.ma.masked_values(x, PQET_ROLL_NULL_FILL).mean()), axis=1).values\n    \n    pred['ts_recency_10'] = pred['timestamp'] - ts_roll.reindex(pred['user_id']).iloc[:, 0].values\n    pred['ts_recency_5'] = pred['timestamp'] - ts_roll.reindex(pred['user_id']).iloc[:, 4].values\n    \n    # we save the user_df seperately to perform ops on them\n    temp = user_df.reindex(pred['user_id']).reset_index(drop=True)\n\n    # concat part with it for future calc\n    temp = pd.concat([temp, pred['part']], axis=1)\n    \n    # up_mean, uq_count, uwrong_sum, upwrong_sum lec_recent\n    temp['up_count'] = temp.apply(lambda x: x[f\"c{int(x['part'])}\"] , axis=1)\n    temp['up_mean'] = temp.apply(lambda x: x[f\"s{int(x['part'])}\"] / x[\"up_count\"], axis=1)\n    temp['uph_mean'] = temp.apply(lambda x: x[f\"ups{int(x['part'])}\"] / x[\"up_count\"], axis=1)\n    temp['up_recency'] = temp.apply(lambda x: x[f\"ts_max_{int(x['part'])}\"], axis=1)\n    \n    temp['up_count'] = temp['up_count'] / temp['uc'].values\n    temp['uwrong_sum'] = temp['uc'] - temp['ucs'].values\n    \n    #  Impute the missing values appropriately    \n    ## could be na when den is 0 -> No questions attempted\n    ## WE donot impute to replicate the training scenario\n    temp['u_mean'] = (temp['us'] / temp['uc'].values)\n    temp['h_mean'] = (temp['ucs'] / temp['uc'].values)\n    \n    #  Compute mean using the sum\n    ## we do a -1 for pqet since we want to ignore nans, \n    ## similar to what expanding_mean in pandas does\n    temp['pqet_mean'] = temp['pqet_sum'] / (temp['uc'].values - 1)\n    temp['seen_ratio'] = temp['pqhe_sum'] / temp['uc'].values\n    \n    # type cast lec recent as False\n    temp['lec_recent'] = temp['lec_recent'].fillna(0).astype(bool) # when 0, it will be False\n    \n    # retain only those user features needed\n    temp = temp[['ts_max', 'up_count', 'up_mean', 'u_mean', 'uwrong_sum', 'lec_recent', \n                 'h_mean', 'pqet_mean', 'seen_ratio', 'uic', 'uph_mean', 'uf_bundle',\n                'seen_exp_when_right', 'seen_exp_when_wrong', 'up_recency', \n                 'sess_max', 'sess_event_sum']]\n    \n    pred = pd.concat([pred, temp], axis=1).set_index(\"group_num\")\n    \n    # adding the uq_per_hr feature\n    pred['uq_per_hr'] = pred['uic'] / (pred['timestamp'] / 60000)\n    \n    # create the harmonic mean feature\n    pred['h_mean'] = (2 * pred['h_mean'] * pred['c_mean']) / (pred['h_mean'] + pred['c_mean'])\n    pred['uph_mean'] = (2 * pred['uph_mean'] * pred['c_mean']) / (pred['uph_mean'] + pred['c_mean'])\n    \n    #  now we add the repeat_c feature, we donot use merge    \n    pred['repeat_c'] = repeat_c.reindex(pred['user_id']).values\n    pred['repeat_c'] = (pred[['repeat_c', 'content_id']].apply(\n        lambda x: x['repeat_c'][x['content_id']]  if type(x['repeat_c']) != float \n        else False, axis=1))\n    \n    pred['ts_max'] = pred['ts_max'].fillna(0)\n    pred['up_recency'] = pred['timestamp'] - pred['up_recency']\n    pred['response_time'] = pred['timestamp'] - pred['ts_max']\n    pred['rt_per_task'] = pred['response_time'] / pred['bundle_q_count']\n    \n    pred['sessions'] = (pred['response_time'] > SESSION_DURATION).astype(bool)\n    pred['sess_event_count'] = np.where(pred['sessions'], 0, pred['sess_event_sum'] + 1)\n    pred['sessions'] = pred['sess_max'] + pred['sessions']\n    \n    # Impute the features for missing users as required\n    pred['uf_bundle'] = pred['uf_bundle'].fillna(pred['bundle_id'])\n    \n    # TODO: Check if fillna works here as intended\n    \n    pred[['seen_exp_when_wrong', 'seen_exp_when_right', 'up_mean', 'up_count', \n          'up_recency', 'seen_ratio', 'rt_per_task', 'response_time', 'sessions', \n          'sess_event_count', 'uwrong_sum']] = (\n        pred[['seen_exp_when_wrong', 'seen_exp_when_right', 'up_mean', 'up_count', \n              'up_recency', 'seen_ratio', 'rt_per_task', 'response_time', 'sessions', \n              'sess_event_count', 'uwrong_sum']].fillna(0))\n    \n    pred['lgb_pred'] = model.predict(pred)\n    \n    # save the loaded values to new column\n    pred['lgb_75'] = pred['lgb_pred'] > 0.75\n    pred['lgb_50'] = pred['lgb_pred'] > 0.5\n    pred['lgb_25'] = pred['lgb_pred'] > 0.25\n\n    # we create some meta features to help FTRL\n    pred['ummr_10_50'] = pred['ummr_10'] > 0\n    pred['h_mean_50'] = pred['h_mean'] > 0.5\n    pred['c_mean_50'] = pred['c_mean'] > 0.5    \n    pred['c_mean_25'] = pred['c_mean'] > 0.25\n    pred['c_mean_75'] = pred['c_mean'] > 0.75\n    \n    pred['pqet_proximity'] = ((pred['prior_question_elapsed_time'] - pred['right_et_med']).abs() < \n                          (pred['prior_question_elapsed_time'] - pred['wrong_et_med']).abs())\n    \n    df_to_dt_format(pred) \n    pred['answered_correctly'] = ftrl.predict(dt.Frame(pred)[:, FTRL_COLS]).to_list()[0]\n\n    if SUBMIT:\n        env.predict(pred[['row_id', 'answered_correctly']])\n        \n    else:\n        op.append(pred)\n\n    # retaining current batch data for next batch\n    prev_test = (batch[0].copy(deep=True), pred)\n\nelse:\n    print (f\"Sucessfully completed after {i + 1} iterations!\")","execution_count":null,"outputs":[]},{"metadata":{"trusted":true},"cell_type":"code","source":"","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}