{"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_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"## import numpy as np\nimport pandas as pd\nfrom itertools import groupby\nfrom sklearn.model_selection import train_test_split\nfrom pandas.api.types import is_datetime64_ns_dtype\nfrom imblearn.under_sampling import RandomUnderSampler\nimport lightgbm as lgb\nimport matplotlib.pyplot as plt\nimport plotly.express as px\nimport gc\nfrom sklearn.preprocessing import MinMaxScaler\nfrom scipy.signal import butter, filtfilt\nimport logging\n#import warnings\n#warnings.filterwarnings(\"ignore\")\nfrom scipy import signal\nformatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s')\nlogging.basicConfig(level=logging.INFO, format=formatter)\nfrom IPython.display import display\n\nSEED = 1992\ntrain_path = \"/kaggle/input/zzzs-lightweight-training-dataset-target/Zzzs_train_multi.parquet\"\ntest_path = \"/kaggle/input/child-mind-institute-detect-sleep-states/test_series.parquet\"\n\ncolumns_to_scale = [\"anglez\", \"enmo\"]","metadata":{"_uuid":"8f2839f25d086af736a60e9eeb907d3b93b6e0e5","_cell_guid":"b1076dfc-b9ad-4769-8c92-a6c4dae69d19","execution":{"iopub.status.busy":"2023-09-22T12:06:13.639835Z","iopub.execute_input":"2023-09-22T12:06:13.640265Z","iopub.status.idle":"2023-09-22T12:06:13.652743Z","shell.execute_reply.started":"2023-09-22T12:06:13.640230Z","shell.execute_reply":"2023-09-22T12:06:13.651689Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# 📖 Introduction\n## 🔨 Profiling Tools\n\nIn this notebook I will present how to build data pipelies and investigate the code performance in the more in depth way :)\nWe will use own decorators and apply it to monitor code performace.","metadata":{}},{"cell_type":"code","source":"import time\nfrom functools import wraps\nimport multiprocessing\nimport platform\nimport time\nimport os\nimport psutil\nimport inspect\n\n# Set this to True if you want to measure performance during profiling\nPERFORMANCE_MEASURE = True\n\ndef elapsed_since(start):\n    \"\"\"\n    Calculate and return the elapsed time since a given start time.\n\n    Args:\n        start (float): The start time as a Unix timestamp.\n\n    Returns:\n        str: A formatted string representing the elapsed time.\n            Examples: \"42.53s\", \"256.12ms\", \"0.03min\", \"0.001hrs\"\n    \"\"\"\n\n    elapsed = time.time() - start\n    if elapsed < 1:\n        return str(round(elapsed*1000,2)) + \"ms\"\n    if elapsed < 60:\n        return str(round(elapsed, 2)) + \"s\"\n    if elapsed < 3600:\n        return str(round(elapsed/60, 2)) + \"min\"\n    else:\n        return str(round(elapsed / 3600, 2)) + \"hrs\"\n\n\ndef get_process_memory():\n    \"\"\"\n    Get memory information for the current process.\n\n    Returns:\n        tuple: A tuple containing three memory-related values:\n            (rss, vms, shared)\n            - rss (int): Resident Set Size in bytes\n            - vms (int): Virtual Memory Size in bytes\n            - shared (int): Shared memory size in bytes\n    \"\"\"\n\n    process = psutil.Process(os.getpid())\n    mi = process.memory_info()\n    return mi.rss, mi.vms, mi.shared\n\n\ndef format_bytes(bytes):\n    \"\"\"\n    Format a byte count into a human-readable string.\n\n    Args:\n        bytes (int): The number of bytes to format.\n\n    Returns:\n        str: A human-readable string representing the byte count.\n            Examples: \"42B\", \"256.12kB\", \"3.14MB\", \"0.002GB\"\n    \"\"\"\n        \n    if abs(bytes) < 1000:\n        return str(bytes)+\"B\"\n    elif abs(bytes) < 1e6:\n        return str(round(bytes/1e3,2)) + \"kB\"\n    elif abs(bytes) < 1e9:\n        return str(round(bytes / 1e6, 2)) + \"MB\"\n    else:\n        return str(round(bytes / 1e9, 2)) + \"GB\"\n\n\ndef profile(func, *args, **kwargs):\n    \"\"\"\n    Profile a function by measuring its execution time and memory usage.\n\n    Args:\n        func (callable): The function to be profiled.\n        *args: Positional arguments to pass to the function.\n        **kwargs: Keyword arguments to pass to the function.\n\n    Returns:\n        callable: A wrapper function that profiles the input function.\n    \"\"\"\n    def wrapper(*args, **kwargs):\n        rss_before, vms_before, shared_before = get_process_memory()\n        start = time.time()\n        result = func(*args, **kwargs)\n        elapsed_time = elapsed_since(start)\n        rss_after, vms_after, shared_after = get_process_memory()\n        if PERFORMANCE_MEASURE:\n            print(\"Profiling: {:>20}  RSS: {:>8} | VMS: {:>8} | SHR {\"\n                  \":>8} | time: {:>8}\"\n                  .format(\"<\" + func.__name__ + \">\",\n                          format_bytes(rss_after - rss_before),\n                          format_bytes(vms_after - vms_before),\n                          format_bytes(shared_after - shared_before),\n                          elapsed_time))\n        return result\n    if inspect.isfunction(func):\n        return wrapper\n    elif inspect.ismethod(func):\n        return wrapper(*args,**kwargs)\n\ndef system_check():\n    \"\"\"\n    Check and print information about the system.\n\n    Returns:\n        int: The number of CPU cores available in the system.\n    \"\"\"\n    num_cores = multiprocessing.cpu_count()\n\n    print(f\"Operating System:\\t{str(platform.node())}\")\n    print(f\"Number of CPU Cores:\\t{num_cores}\")\n\n    return num_cores\nsystem_check()","metadata":{"execution":{"iopub.status.busy":"2023-09-22T12:06:14.315458Z","iopub.execute_input":"2023-09-22T12:06:14.315814Z","iopub.status.idle":"2023-09-22T12:06:14.338282Z","shell.execute_reply.started":"2023-09-22T12:06:14.315784Z","shell.execute_reply":"2023-09-22T12:06:14.337286Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"## 🔨 Feature Enginnering","metadata":{}},{"cell_type":"code","source":"import pandas as pd\nimport numpy as np\nfrom tqdm import tqdm\n\nclass FeatureEngineer:\n    \n    @profile\n    def __init__(self, df_path, columns_to_scale,train_mode=True):\n        \"\"\"\n        Initialize the FeatureEngineer class.\n\n        Args:\n            df_path (str): Path to the input data in Parquet format.\n            columns_to_scale (list): List of column names to scale using MinMaxScaler.\n            train_mode (bool, optional): Set to True for training mode, False for inference mode. Defaults to True.\n        \"\"\"\n        self.df_path = df_path\n        self.columns_to_scale = columns_to_scale\n        self.periods = 10\n        self.train_mode = train_mode\n    \n    @staticmethod\n    def reduce_mem_usage(df):\n        \"\"\"\n        Iterate through all numeric columns of a DataFrame and modify the data type\n        to reduce memory usage.\n\n        Args:\n            df (DataFrame): The input DataFrame.\n\n        Returns:\n            DataFrame: The DataFrame with reduced memory usage.\n        \"\"\"\n        for col in df.columns:\n            col_type = df[col].dtype\n\n            if col_type != object and not is_datetime64_ns_dtype(df[col]) and not 'category':\n                c_min = df[col].min()\n                c_max = df[col].max()\n                if str(col_type)[:3] == 'int':\n                    if c_min > np.iinfo(np.int8).min and c_max < np.iinfo(np.int8).max:\n                        df[col] = df[col].astype(np.int8)\n                    elif c_min > np.iinfo(np.int16).min and c_max < np.iinfo(np.int16).max:\n                        df[col] = df[col].astype(np.int16)\n                    elif c_min > np.iinfo(np.int32).min and c_max < np.iinfo(np.int32).max:\n                        df[col] = df[col].astype(np.int32)\n                    elif c_min > np.iinfo(np.int64).min and c_max < np.iinfo(np.int64).max:\n                        df[col] = df[col].astype(np.int32)  \n                else:\n\n                    df[col] = df[col].astype(np.float16)\n        logging.info('reduce_mem_usage sucessfully')\n\n        return df\n    \n    @staticmethod\n    def kalman_filter(data, process_variance, measurement_variance):\n        \"\"\"\n        Apply the Kalman filter to a sequence of data.\n\n        Args:\n            data (iterable): The input data sequence to be filtered.\n            process_variance (float): The process variance (prediction error) parameter.\n            measurement_variance\" (float): The measurement variance (measurement error) parameter.\n\n        Returns:\n            numpy.ndarray: An array containing the filtered data.\n        \"\"\"\n        # Initial state estimate\n        x_est = data[0]\n\n        # Initial error estimate\n        P_est = 1.0\n\n        # Kalman gain\n        K = 0.0\n\n        filtered_data = []\n\n        for z in data:\n            # Prediction step\n            x_pred = x_est\n            P_pred = P_est + process_variance\n\n            # Update step\n            K = P_pred / (P_pred + measurement_variance)\n            x_est = x_pred + K * (z - x_pred)\n            P_est = (1 - K) * P_pred\n\n            filtered_data.append(x_est)\n\n        return np.array(filtered_data)\n    \n    @profile\n    def read_data(self):\n        \"\"\"\n        Read data from the specified path and apply memory reduction.\n\n        Returns:\n            DataFrame: The loaded and memory-reduced DataFrame.\n        \"\"\"\n        try:\n            self.data = self.reduce_mem_usage(pd.read_parquet(self.df_path).sample(frac=0.05, random_state=SEED))  \n            display(self.data.head())\n        except Exception as e:\n            print(f\"Error reading data: {e}\")\n            \n    @profile\n    def scale_data(self):\n        \"\"\"\n        Scale selected columns in the data using MinMaxScaler.\n        \"\"\"\n        try:\n            scaler = MinMaxScaler()\n            self.data[self.columns_to_scale] = scaler.fit_transform(self.data[self.columns_to_scale])\n\n        except Exception as e:\n            print(f\"Error scaling data: {e}\")\n\n    @profile\n    def preprocess_data(self):\n        \"\"\"\n        Preprocess data, create features, and optimize memory usage.\n\n        Returns:\n            DataFrame: The preprocessed DataFrame.\n        \"\"\"\n        if self.data is None:\n            print(\"Data is None. Please check the read_data method.\")\n            return\n\n        self.data['timestamp'] = pd.to_datetime(self.data['timestamp']).apply(lambda t: t.tz_localize(None))\n\n        # 🕘 Parse the 'timestamp' column and create an \"hour\" feature\n        self.data[\"hour\"] = self.data[\"timestamp\"].dt.hour\n\n        print(\"Data parsed correcty.\")\n        # Calculate the differences and rolling statistics for 'anglez' and 'enmo'\n        grouped = self.data.groupby('series_id')\n        self.data[\"anglez_diff\"] = grouped['anglez'].diff(periods=self.periods).fillna(method=\"bfill\").astype('float16')\n        self.data[\"enmo_diff\"] = grouped['enmo'].diff(periods=self.periods).fillna(method=\"bfill\").astype('float16')\n        print(\"Diff calculation done.\")\n        \n        rolling_cols = [\"anglez\", \"enmo\", \"anglez_diff\", \"enmo_diff\"]\n        for col in tqdm(rolling_cols):\n            self.data[f\"{col}_rolling_mean\"] = grouped[col].transform(lambda x: x.rolling(self.periods, center=True).mean().fillna(method=\"bfill\").fillna(method=\"ffill\")).astype('float16')\n            self.data[f\"{col}_rolling_max\"] = grouped[col].transform(lambda x: x.rolling(self.periods, center=True).max().fillna(method=\"bfill\").fillna(method=\"ffill\")).astype('float16')\n            self.data[f\"{col}_rolling_std\"] = grouped[col].transform(lambda x: x.rolling(self.periods, center=True).std().fillna(method=\"bfill\").fillna(method=\"ffill\")).astype('float16')\n        \n        # Lag features\n        for col in tqdm(['anglez', 'enmo']):\n            for lag in [1, 2, 3]:\n                self.data[f'{col}_lag_{lag}'] = self.data.groupby('series_id')[col].shift(lag).astype('float16')\n        \n        # Interaction features\n        self.data['anglez_enmo_product'] = self.data['anglez'] * self.data['enmo']\n        self.data['anglez_enmo_sum'] = self.data['anglez'] + self.data['enmo']\n        self.data['anglez_enmo_difference'] = self.data['anglez'] - self.data['enmo']\n\n        if self.train_mode:\n            self.data.drop(columns=['series_id', 'step', 'timestamp'], inplace=True)\n            self.data.dropna(how='any',inplace=True)\n            X_train = self.data.drop(columns=[\"awake\"])\n            y_train = self.data[\"awake\"]\n            display(X_train.head())\n            display(y_train.head())\n            return X_train,y_train\n        else:\n            return self.data\n\n","metadata":{"execution":{"iopub.status.busy":"2023-09-22T12:06:28.000510Z","iopub.execute_input":"2023-09-22T12:06:28.000868Z","iopub.status.idle":"2023-09-22T12:06:28.031091Z","shell.execute_reply.started":"2023-09-22T12:06:28.000838Z","shell.execute_reply":"2023-09-22T12:06:28.030063Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"train_fe = FeatureEngineer(train_path, columns_to_scale,train_mode=True)\ntrain_fe.read_data()\nX,y = train_fe.preprocess_data()\ngc.collect()  ","metadata":{"execution":{"iopub.status.busy":"2023-09-22T12:06:28.033000Z","iopub.execute_input":"2023-09-22T12:06:28.033393Z","iopub.status.idle":"2023-09-22T12:06:58.276508Z","shell.execute_reply.started":"2023-09-22T12:06:28.033361Z","shell.execute_reply":"2023-09-22T12:06:58.275578Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"### Check Missing Data\nmissing_counts = X.isna().sum()\nmissing_counts","metadata":{"execution":{"iopub.status.busy":"2023-09-22T12:06:58.278544Z","iopub.execute_input":"2023-09-22T12:06:58.279232Z","iopub.status.idle":"2023-09-22T12:06:58.338357Z","shell.execute_reply.started":"2023-09-22T12:06:58.279196Z","shell.execute_reply":"2023-09-22T12:06:58.337235Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"#### Create Random Column for the feature importance check \nX['Random_Column'] = np.random.rand(len(X))","metadata":{"execution":{"iopub.status.busy":"2023-09-22T12:07:01.509378Z","iopub.execute_input":"2023-09-22T12:07:01.509740Z","iopub.status.idle":"2023-09-22T12:07:01.524182Z","shell.execute_reply.started":"2023-09-22T12:07:01.509711Z","shell.execute_reply":"2023-09-22T12:07:01.523260Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"from xgboost import XGBClassifier\nmodel = XGBClassifier(tree_method='gpu_hist', gpu_id=0)\nmodel.fit(X, y)\nfeature_importances = model.feature_importances_","metadata":{"execution":{"iopub.status.busy":"2023-09-22T12:07:07.065704Z","iopub.execute_input":"2023-09-22T12:07:07.066087Z","iopub.status.idle":"2023-09-22T12:07:13.546386Z","shell.execute_reply.started":"2023-09-22T12:07:07.066049Z","shell.execute_reply":"2023-09-22T12:07:13.545561Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"If the feature importance is at the simmiar level to Random Column it means the feature only brings noise and it have to be removed/transformed.\n","metadata":{}},{"cell_type":"code","source":"import matplotlib.pyplot as plt\n\n# Create a DataFrame to store feature names and their importances\nfeature_importance_df = pd.DataFrame({'Feature': X.columns, 'Importance': model.feature_importances_})\n\n# Sort the DataFrame by importance in descending order\nfeature_importance_df = feature_importance_df.sort_values(by='Importance', ascending=False)\n\n# Create a bar plot to visualize feature importances\nplt.figure(figsize=(12, 6))\nplt.barh(feature_importance_df['Feature'], feature_importance_df['Importance'], color='skyblue')\nplt.xlabel('Feature Importance')\nplt.ylabel('Feature')\nplt.title('XGBoost Feature Importances')\nplt.gca().invert_yaxis()  # Invert the y-axis to display the most important features at the top\nplt.show()\ndisplay(feature_importance_df)","metadata":{"execution":{"iopub.status.busy":"2023-09-22T12:07:28.435499Z","iopub.execute_input":"2023-09-22T12:07:28.435924Z","iopub.status.idle":"2023-09-22T12:07:29.065072Z","shell.execute_reply.started":"2023-09-22T12:07:28.435888Z","shell.execute_reply":"2023-09-22T12:07:29.064019Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"markdown","source":"# Make the rest of the code ....","metadata":{}},{"cell_type":"markdown","source":"Of course additionall steps can be made eg. feature reduction and apply stationarity tests eg ADF or seasonal decomposition. However in this notebook I just want to show the basics concepts :) ","metadata":{}},{"cell_type":"code","source":"# from sklearn.ensemble import RandomForestClassifier\n\n# @profile\n# def RandomForestClassifierFit(X_train, y_train):\n\n#     classifier = RandomForestClassifier(n_estimators=100,\n#                                         max_depth=6,\n#                                         min_samples_leaf=150,\n#                                         random_state=SEED,\n#                                         n_jobs=-1)\n\n#     classifier.fit(X_train, y_train)\n#     return classifier\n\n# classifier = RandomForestClassifierFit(X, y)","metadata":{"execution":{"iopub.status.busy":"2023-09-22T12:09:52.819074Z","iopub.execute_input":"2023-09-22T12:09:52.819740Z","iopub.status.idle":"2023-09-22T12:09:52.824366Z","shell.execute_reply.started":"2023-09-22T12:09:52.819705Z","shell.execute_reply":"2023-09-22T12:09:52.823232Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"","metadata":{"execution":{"iopub.status.busy":"2023-09-22T12:08:22.683520Z","iopub.status.idle":"2023-09-22T12:08:22.684305Z","shell.execute_reply.started":"2023-09-22T12:08:22.684065Z","shell.execute_reply":"2023-09-22T12:08:22.684089Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"","metadata":{},"execution_count":null,"outputs":[]}]}