{"metadata": {"language_info": {"pygments_lexer": "ipython3", "version": "3.5.4", "mimetype": "text/x-python", "name": "python", "codemirror_mode": {"version": 3, "name": "ipython"}, "nbconvert_exporter": "python", "file_extension": ".py"}, "kernelspec": {"language": "python", "name": "conda-root-py", "display_name": "Python [conda root]"}}, "nbformat": 4, "cells": [{"source": ["This script simply reads the log files in manageable chunks, assembles a running set of metrics for each MSNO (see 'cols' below) and then outputs a CSV with one row per user to be used in downstream analyses.\n", "\n", "This workbook is intended to be copied and run locally as it will exceed the Kaggle limits.  I ran it overnight on my MacBook Pro with 16GB RAM."], "metadata": {}, "cell_type": "markdown"}, {"source": ["%matplotlib inline\n", "import numpy as np\n", "import pandas as pd\n", "\n", "from sklearn.model_selection import train_test_split\n", "from sklearn.preprocessing import LabelEncoder\n", "import lightgbm as lgb\n", "from tqdm import tqdm\n", "import gc\n", "import seaborn as sbn\n", "import datetime as dt\n", "import pdb\n", "from collections import OrderedDict"], "metadata": {"collapsed": true}, "execution_count": null, "cell_type": "code", "outputs": []}, {"source": ["# Replace these paths with relevant paths on your local host\n", "data_path = '../../../input/kkboxchurnprediction/data/'\n", "output_path = '../../../output/kkboxchurnprediction/out/'\n"], "metadata": {"collapsed": true}, "execution_count": null, "cell_type": "code", "outputs": []}, {"source": ["# these are the columns that will be output to the CSV\n", "cols = ['num_25_std', 'num_25_sum', 'num_25_min', 'num_25_max', 'num_25_mean',\n", "       'num_50_std', 'num_50_sum', 'num_50_min', 'num_50_max', 'num_50_mean',\n", "       'num_75_std', 'num_75_sum', 'num_75_min', 'num_75_max', 'num_75_mean',\n", "       'num_985_std', 'num_985_sum', 'num_985_min', 'num_985_max',\n", "       'num_985_mean', 'num_100_std', 'num_100_sum', 'num_100_min',\n", "       'num_100_max', 'num_100_mean', 'num_unq_std', 'num_unq_sum',\n", "       'num_unq_min', 'num_unq_max', 'num_unq_mean', 'total_secs_std',\n", "       'total_secs_sum', 'total_secs_min', 'total_secs_max', 'total_secs_mean',\n", "       'earliest_date', 'latest_date', 'log_count', 'msno']\n"], "metadata": {"collapsed": true}, "execution_count": null, "cell_type": "code", "outputs": []}, {"source": ["num_stat_grps = 7\n", "stat_grp_size = 5\n", "\n", "def update_user_batch(old, new):\n", "#     pdb.set_trace()\n", "    \n", "    if old is None:\n", "        old = np.zeros((num_stat_grps*stat_grp_size) + 3)\n", "\n", "    # run the stats\n", "    xct = old[-1] + new[-1]\n", "    \n", "    for i in range(0,num_stat_grps):\n", "        # move the index to the next set of stats\n", "        ni = i*stat_grp_size\n", "        oi = i*stat_grp_size \n", "        xsum = old[oi+1] + new[ni+1]\n", "        xmean = float(xsum)/float(xct)\n", "        xstd = 0.0 # wasn't sure how to do this on an incremental basis... feel free to add this if your math skills are better than mine\n", "        old[oi+0] = xstd\n", "        old[oi+1] = xsum\n", "        old[oi+2] = new[ni+2] if xct == 1 else new[ni+2] if new[ni+2] < old[oi+2] else old[oi+2]\n", "        old[oi+3] = new[ni+3] if xct == 1 else new[ni+3] if new[ni+3] > old[oi+3] else old[oi+3]\n", "        old[oi+4] = xmean\n", "    \n", "    # increment row count\n", "    old[-1] = xct\n", "    \n", "    # set the earliest date if needed\n", "    if not old[-3]:\n", "        old[-3] = new[-3]\n", "    elif new[-3] < old[-3]:\n", "        old[-3] = new[-3]\n", "\n", "    # update latest date\n", "    if new[-2] > old[-2]:\n", "        old[-2] = new[-2] # assumes all filed entries are sequential\n", "    \n", "    return old\n", "\n", "            \n", "# simple unit test; uncomment to run\n", "# told = [0.0,6,2,2,2.0, 1.0,6,1,3,2.0, 3.0,20,5,10,6.66, 0,0,0,0,0, 0,0,0,0,0, 0,0,0,0,0, 0,0,0,0,0, 100000, 100000, 3]\n", "# tnew = [0.0,4,2,2,2.0, 1.0,6,3,3,3.0, 3.0,20,10,10,10.0, 0,0,0,0,0, 0,0,0,0,0, 0,0,0,0,0, 0,0,0,0,0, 100, 900000, 2]\n", "# told2 = update_user_batch(told, tnew)\n", "# print(told2)\n", "# assert told2[-1] == 5\n", "# assert told2[1] == 10\n", "# assert told2[2] == 2\n", "# assert told2[3] == 2\n", "# assert told2[4] == 2.0\n", "# assert told2[6] == 12\n", "# assert told2[7] == 1\n", "# assert told2[8] == 3\n", "# assert told2[9] == 2.40\n", "# assert told2[11] == 40\n", "# assert told2[12] == 5\n", "# assert told2[13] == 10\n", "# assert told2[14] == 8.0\n", "\n", "# assert told2[-3] == 100\n", "# assert told2[-2] == 900000\n"], "metadata": {"collapsed": true}, "execution_count": null, "cell_type": "code", "outputs": []}, {"source": ["# process the logs\n", "print('getting iterator for user_logs...')\n", "\n", "# dictionary for running MSNO stats\n", "userlogs = {}\n", "\n", "# drop the second file if you don't want the latest logs\n", "files = ['user_logs.csv', 'user_logs_v2.csv']\n", "\n", "start = dt.datetime.now()\n", "for f in files:\n", "    print('READING {0}'.format(f))\n", "    log_reader = pd.read_csv(data_path + f, chunksize=1000000)\n", "    for idx, df in enumerate(log_reader):\n", "        print('chunk {0}; num users: {1}; total duration (min): {2:0.1f}'.format(idx, len(userlogs), (dt.datetime.now() - start).total_seconds()/60.0))\n", "        dfg = df.groupby('msno').agg(OrderedDict([\n", "            ('num_25', {'sum', 'mean', 'std', 'min', 'max'}),\n", "            ('num_50', {'sum', 'mean', 'std', 'min', 'max'}),\n", "            ('num_75', {'sum', 'mean', 'std', 'min', 'max'}),\n", "            ('num_985', {'sum', 'mean', 'std', 'min', 'max'}),\n", "            ('num_100', {'sum', 'mean', 'std', 'min', 'max'}),\n", "            ('num_unq', {'sum', 'mean', 'std', 'min', 'max'}),\n", "            ('total_secs', {'sum', 'mean', 'std', 'min', 'max'}),\n", "            ('date', {'first', 'last', 'count'}),\n", "        ]))\n", "        dfg.columns = ['_'.join(col) for col in dfg.columns.ravel()]        \n", "        dfg.reset_index(inplace=True)\n", "        for row in dfg.iterrows():\n", "            newvals = row[1]\n", "            msno = newvals['msno']\n", "            if msno in userlogs:\n", "                userlogs[msno] = update_user_batch(userlogs[msno], newvals[1::]) # col 0 is msno\n", "            else:\n", "                userlogs[msno] = update_user_batch(None, newvals[1::]) # col 0 is msno"], "metadata": {}, "execution_count": null, "cell_type": "code", "outputs": []}, {"source": ["del df, dfg; gc.collect()"], "metadata": {"_kg_hide-output": false}, "execution_count": null, "cell_type": "code", "outputs": []}, {"source": ["dfo = pd.DataFrame(data=[np.append(userlogs[key], key) for key in userlogs.keys()], columns=cols)"], "metadata": {"collapsed": true}, "execution_count": null, "cell_type": "code", "outputs": []}, {"source": ["del userlogs; gc.collect();"], "metadata": {"collapsed": true}, "execution_count": null, "cell_type": "code", "outputs": []}, {"source": ["for i in range(0,(35-4)):\n", "    dfo[dfo.columns[i]] = dfo[dfo.columns[i]].astype('float')\n", "\n", "dfo.dtypes"], "metadata": {"collapsed": true}, "execution_count": null, "cell_type": "code", "outputs": []}, {"source": ["# output results\n", "dfo.to_csv(data_path + 'user_logs_summary.csv', index=False)"], "metadata": {"collapsed": true}, "execution_count": null, "cell_type": "code", "outputs": []}], "nbformat_minor": 1}