{"metadata":{"kernelspec":{"language":"python","display_name":"Python 3","name":"python3"},"language_info":{"name":"python","version":"3.10.12","mimetype":"text/x-python","codemirror_mode":{"name":"ipython","version":3},"pygments_lexer":"ipython3","nbconvert_exporter":"python","file_extension":".py"},"kaggle":{"accelerator":"none","dataSources":[{"sourceId":50160,"databundleVersionId":7921029,"sourceType":"competition"},{"sourceId":8394709,"sourceType":"datasetVersion","datasetId":4993963},{"sourceId":8527486,"sourceType":"datasetVersion","datasetId":5092396}],"isInternetEnabled":false,"language":"python","sourceType":"notebook","isGpuEnabled":false}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"import joblib\nimport gc\nimport pandas as pd\nimport polars as pl\nimport lightgbm as lgb\nimport matplotlib.pyplot as plt\nimport json\nfrom pandas.api.types import CategoricalDtype\nfrom datetime import datetime\nfrom datetime import date\nfrom sklearn.model_selection import train_test_split\nfrom sklearn.metrics import auc, roc_curve, roc_auc_score\n\ndef convert_to_ordinal(date_column, date_format):\n epoch_start = datetime(1970, 1, 1)\n date_parsed = date_column.str.strptime(pl.Date, date_format, strict=False)\n return pl.when(date_parsed.is_not_null()).then(\n (date_parsed - epoch_start).dt.total_days()\n ).otherwise(None)\n\nclass DataPipeline_Depth_0:\n def __init__(self, base_path, static_0_0_path, static_0_1_path, static_cb_0_path, schema_path):\n self.base_path = base_path\n self.static_0_0_path = static_0_0_path\n self.static_0_1_path = static_0_1_path\n self.static_cb_0_path = static_cb_0_path\n self.schema_path = schema_path\n self.global_schema = {}\n\n def load_data(self, path):\n try:\n df = pl.read_parquet(path)\n self.update_schema(df)\n return df\n except Exception as e:\n print(f\"Error loading data from {path}: {e}\")\n raise\n\n def update_schema(self, dataframe):\n for col, dtype in zip(dataframe.columns, dataframe.dtypes):\n if col not in self.global_schema:\n self.global_schema[col] = str(dtype)\n\n def preprocess_base(self, data):\n data = data.with_columns(\n convert_to_ordinal(pl.col('date_decision'), '%Y-%m-%d').alias('date_decision_ordinal')\n )\n data = data.drop(['date_decision'])\n return data\n\n def preprocess_static(self, data):\n date_columns = [col for col in data.columns if col.endswith('D') and data[col].dtype == pl.Utf8]\n for col in date_columns:\n data = data.with_columns(\n convert_to_ordinal(pl.col(col), '%Y-%m-%d').alias(col)\n )\n for col in data.columns:\n if data[col].dtype == pl.Boolean:\n data = data.with_columns(data[col].cast(pl.Int32).alias(col))\n return data.select([col for col in data.columns if data[col].dtype != pl.Utf8 or col in date_columns])\n\n def preprocess_static_cb_0(self, data):\n date_columns = [col for col in data.columns if col.endswith('D') and data[col].dtype == pl.Utf8]\n for col in date_columns:\n data = data.with_columns(\n convert_to_ordinal(pl.col(col), '%Y-%m-%d').alias(col)\n )\n categorical_columns = ['education_1103M', 'maritalst_385M']\n for col in categorical_columns:\n if col in data.columns:\n data = data.with_columns(data[col].cast(pl.Categorical))\n columns_to_drop = [col for col in data.columns if data[col].dtype == pl.Utf8 and col not in date_columns and col not in categorical_columns]\n data = data.drop(columns_to_drop)\n return data\n\n def merge_data(self, data_base, data_static_0_0, data_static_0_1, data_static_cb_0):\n concatenated_data = pl.concat([data_static_0_0, data_static_0_1], how='vertical')\n merged_data = data_base.join(concatenated_data, on='case_id', how='left')\n del concatenated_data # Free up memory\n gc.collect()\n\n merged_data = merged_data.join(data_static_cb_0, on='case_id', how='left')\n for col in ['education_1103M', 'maritalst_385M']:\n if merged_data[col].dtype != pl.Categorical:\n merged_data = merged_data.with_columns(merged_data[col].cast(pl.Categorical))\n dummies = merged_data[['education_1103M', 'maritalst_385M']].to_dummies()\n merged_data = merged_data.drop(['education_1103M', 'maritalst_385M'])\n merged_data = pl.concat([merged_data, dummies], how='horizontal')\n return merged_data\n\n def execute_pipeline(self):\n data_base = self.load_data(self.base_path)\n data_static_0_0 = self.load_data(self.static_0_0_path)\n data_static_0_1 = self.load_data(self.static_0_1_path)\n data_static_cb_0 = self.load_data(self.static_cb_0_path)\n data_base = self.preprocess_base(data_base)\n data_static_0_0 = self.preprocess_static(data_static_0_0)\n data_static_0_1 = self.preprocess_static(data_static_0_1)\n data_static_cb_0 = self.preprocess_static_cb_0(data_static_cb_0)\n\n # Save preprocessed data to disk to free up memory\n data_base.write_parquet('train_base_processed.parquet')\n data_static_0_0.write_parquet('train_static_0_0_processed.parquet')\n data_static_0_1.write_parquet('train_static_0_1_processed.parquet')\n data_static_cb_0.write_parquet('train_static_cb_0_processed.parquet')\n\n # Clean up memory by deleting intermediate dataframes\n del data_base, data_static_0_0, data_static_0_1, data_static_cb_0\n gc.collect()\n\n # Reload preprocessed data from disk for merging\n data_base = pl.read_parquet('train_base_processed.parquet')\n data_static_0_0 = pl.read_parquet('train_static_0_0_processed.parquet')\n data_static_0_1 = pl.read_parquet('train_static_0_1_processed.parquet')\n data_static_cb_0 = pl.read_parquet('train_static_cb_0_processed.parquet')\n\n return self.merge_data(data_base, data_static_0_0, data_static_0_1, data_static_cb_0)\n\nif __name__ == \"__main__\":\n dataPath = \"/kaggle/input/home-credit-credit-risk-model-stability/\"\n schema_path = \"/kaggle/input/unified-schema/unified_schema.json\"\n pipeline = DataPipeline_Depth_0(\n dataPath + \"parquet_files/train/train_base.parquet\",\n dataPath + \"parquet_files/train/train_static_0_0.parquet\",\n dataPath + \"parquet_files/train/train_static_0_1.parquet\",\n dataPath + \"parquet_files/train/train_static_cb_0.parquet\",\n schema_path\n )\n Depth_0 = pipeline.execute_pipeline()\n \n # Clean up memory after executing the pipeline\n del pipeline, dataPath, schema_path\n gc.collect()","metadata":{"execution":{"iopub.status.busy":"2024-05-27T18:11:52.332563Z","iopub.execute_input":"2024-05-27T18:11:52.333326Z","iopub.status.idle":"2024-05-27T18:12:22.116494Z","shell.execute_reply.started":"2024-05-27T18:11:52.333284Z","shell.execute_reply":"2024-05-27T18:12:22.115245Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"def convert_to_ordinal(date_column, date_format):\n epoch_start = datetime(1970, 1, 1)\n date_parsed = date_column.str.strptime(pl.Date, date_format, strict=False)\n return pl.when(date_parsed.is_not_null()).then(\n (date_parsed - epoch_start).dt.total_days()\n ).otherwise(None)\n\ndef dtype_mapping(dtype_str):\n mapping = {\n 'Int32': pl.Int32,\n 'Int64': pl.Int64,\n 'Float32': pl.Float32,\n 'Float64': pl.Float64,\n 'Utf8': pl.Utf8,\n 'Boolean': pl.Boolean,\n 'Date': pl.Date,\n 'Categorical': pl.Categorical\n }\n return mapping.get(dtype_str, pl.Utf8)\n\nclass DataPipeline_Depth_0:\n def __init__(self, base_path, static_paths, static_cb_0_path, schema_path):\n self.base_path = base_path\n self.static_paths = static_paths\n self.static_cb_0_path = static_cb_0_path\n self.schema_path = schema_path\n self.global_schema = self.load_schema()\n\n def load_schema(self):\n with open(self.schema_path, 'r') as file:\n return json.load(file)\n\n def load_data(self, path):\n df = pl.read_parquet(path)\n return self.ensure_schema(df)\n\n def ensure_schema(self, dataframe):\n for col, expected_dtype in self.global_schema.items():\n expected_pl_dtype = dtype_mapping(expected_dtype)\n if col in dataframe.columns:\n if dataframe[col].dtype != expected_pl_dtype:\n dataframe = dataframe.with_columns(dataframe[col].cast(expected_pl_dtype))\n return dataframe\n\n def preprocess_base(self, data):\n data = data.with_columns(\n convert_to_ordinal(pl.col('date_decision'), '%Y-%m-%d').alias('date_decision_ordinal')\n )\n data = data.drop(['date_decision'])\n return data\n\n def preprocess_static(self, data):\n date_columns = [col for col in data.columns if col.endswith('D') and data[col].dtype == pl.Utf8]\n for col in date_columns:\n data = data.with_columns(\n convert_to_ordinal(pl.col(col), '%Y-%m-%d').alias(col)\n )\n for col in data.columns:\n if data[col].dtype == pl.Boolean:\n data = data.with_columns(data[col].cast(pl.Int32).alias(col))\n columns_to_keep = [col for col in data.columns if data[col].dtype != pl.Utf8 or col in date_columns]\n data = data.select(columns_to_keep)\n return data\n\n def preprocess_static_cb_0(self, data):\n date_columns = [col for col in data.columns if col.endswith('D') and data[col].dtype == pl.Utf8]\n for col in date_columns:\n data = data.with_columns(\n convert_to_ordinal(pl.col(col), '%Y-%m-%d').alias(col)\n )\n categorical_columns = ['education_1103M', 'maritalst_385M']\n for col in categorical_columns:\n if col in data.columns:\n data = data.with_columns(data[col].cast(pl.Categorical))\n columns_to_drop = [col for col in data.columns if data[col].dtype == pl.Utf8 and col not in date_columns and col not in categorical_columns]\n data = data.drop(columns_to_drop)\n return data\n\n def merge_data(self, data_base, static_datas, data_static_cb_0):\n concatenated_static_data = pl.concat(static_datas, how='vertical')\n merged_data = data_base.join(concatenated_static_data, on='case_id', how='left')\n del concatenated_static_data # Free up memory\n gc.collect()\n\n merged_data = merged_data.join(data_static_cb_0, on='case_id', how='left')\n for col in ['education_1103M', 'maritalst_385M']:\n if merged_data[col].dtype != pl.Categorical:\n merged_data = merged_data.with_columns(merged_data[col].cast(pl.Categorical))\n dummies = merged_data[['education_1103M', 'maritalst_385M']].to_dummies()\n merged_data = merged_data.drop(['education_1103M', 'maritalst_385M'])\n merged_data = pl.concat([merged_data, dummies], how='horizontal')\n return merged_data\n\n def execute_pipeline(self):\n data_base = self.load_data(self.base_path)\n static_datas = [self.load_data(path) for path in self.static_paths]\n data_static_cb_0 = self.load_data(self.static_cb_0_path)\n data_base = self.preprocess_base(data_base)\n static_datas = [self.preprocess_static(data) for data in static_datas]\n data_static_cb_0 = self.preprocess_static_cb_0(data_static_cb_0)\n \n # Save preprocessed data to disk to free up memory\n data_base.write_parquet('test_base_processed.parquet')\n for i, static_data in enumerate(static_datas):\n static_data.write_parquet(f'test_static_0_{i}_processed.parquet')\n data_static_cb_0.write_parquet('test_static_cb_0_processed.parquet')\n\n # Clean up memory by deleting intermediate dataframes\n del data_base, static_datas, data_static_cb_0\n gc.collect()\n\n # Reload preprocessed data from disk for merging\n data_base = pl.read_parquet('test_base_processed.parquet')\n static_datas = [pl.read_parquet(f'test_static_0_{i}_processed.parquet') for i in range(len(self.static_paths))]\n data_static_cb_0 = pl.read_parquet('test_static_cb_0_processed.parquet')\n\n return self.merge_data(data_base, static_datas, data_static_cb_0)\n\nif __name__ == \"__main__\":\n dataPath = \"/kaggle/input/home-credit-credit-risk-model-stability/\"\n schema_path = \"/kaggle/input/unified-schema/unified_schema.json\"\n base_path = dataPath + \"parquet_files/test/test_base.parquet\"\n static_paths = [\n dataPath + \"parquet_files/test/test_static_0_0.parquet\",\n dataPath + \"parquet_files/test/test_static_0_1.parquet\",\n dataPath + \"parquet_files/test/test_static_0_2.parquet\"\n ]\n static_cb_0_path = dataPath + \"parquet_files/test/test_static_cb_0.parquet\"\n \n pipeline = DataPipeline_Depth_0(base_path, static_paths, static_cb_0_path, schema_path)\n Depth_0_test = pipeline.execute_pipeline()\n \n # Clean up memory after executing the pipeline\n del pipeline, dataPath, schema_path\n gc.collect()","metadata":{"execution":{"iopub.status.busy":"2024-05-27T18:12:22.118803Z","iopub.execute_input":"2024-05-27T18:12:22.119630Z","iopub.status.idle":"2024-05-27T18:12:22.749889Z","shell.execute_reply.started":"2024-05-27T18:12:22.119578Z","shell.execute_reply":"2024-05-27T18:12:22.748629Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"class DataPipeline_Depth_1:\n def __init__(self, applprev_paths, other_path, deposit_path, person_path, debitcard_path, tax_registry_a_1_path, tax_registry_b_1_path, \n tax_registry_c_1_path, credit_bureau_a_1_paths, credit_bureau_b_1_path, schema_path):\n self.applprev_paths = applprev_paths\n self.other_path = other_path\n self.deposit_path = deposit_path\n self.person_path = person_path\n self.debitcard_path = debitcard_path\n self.tax_registry_a_1_path = tax_registry_a_1_path\n self.tax_registry_b_1_path = tax_registry_b_1_path\n self.tax_registry_c_1_path = tax_registry_c_1_path\n self.credit_bureau_a_1_paths = credit_bureau_a_1_paths\n self.credit_bureau_b_1_path = credit_bureau_b_1_path\n self.schema_path = schema_path\n self.global_schema = {}\n\n @staticmethod\n def try_parse_date(col, fmt1, fmt2):\n date1 = col.str.strptime(pl.Date, fmt1, strict=False)\n date2 = col.str.strptime(pl.Date, fmt2, strict=False)\n return pl.when(date1.is_not_null()).then(date1).otherwise(date2)\n\n @staticmethod\n def convert_to_ordinal(date):\n return pl.when(date.is_not_null()).then(\n (date.dt.year() * 365) + (date.dt.month() * 30) + date.dt.day()\n ).otherwise(None)\n\n @staticmethod\n def optimize_dataframe(df):\n for col in df.columns:\n if df[col].dtype == pl.Int64:\n df = df.with_columns(pl.col(col).cast(pl.Int32))\n elif df[col].dtype == pl.Float64:\n df = df.with_columns(pl.col(col).cast(pl.Float32))\n return df\n\n def load_data(self):\n self.train_applprev_1 = self.batch_load_and_update_schema(self.applprev_paths)\n self.train_other_1 = self.load_and_update_schema(self.other_path).lazy()\n self.train_deposit_1 = self.load_and_update_schema(self.deposit_path).lazy()\n self.train_person_1 = self.load_and_update_schema(self.person_path).lazy()\n self.train_debitcard_1 = self.load_and_update_schema(self.debitcard_path).lazy()\n self.train_tax_registry_a_1 = self.load_and_update_schema(self.tax_registry_a_1_path).lazy()\n self.train_tax_registry_b_1 = self.load_and_update_schema(self.tax_registry_b_1_path).lazy()\n self.train_tax_registry_c_1 = self.load_and_update_schema(self.tax_registry_c_1_path).lazy()\n self.train_credit_bureau_a_1 = self.batch_load_and_update_schema(self.credit_bureau_a_1_paths)\n self.train_credit_bureau_b_1 = self.load_and_update_schema(self.credit_bureau_b_1_path).lazy()\n\n def load_and_update_schema(self, path):\n df = pl.read_parquet(path)\n df = self.optimize_dataframe(df)\n self.update_schema(df)\n return df\n\n def batch_load_and_update_schema(self, paths):\n batch_size = 2\n lazy_frames = []\n for i in range(0, len(paths), batch_size):\n batch_paths = paths[i:i+batch_size]\n batch_frames = [self.load_and_update_schema(path) for path in batch_paths]\n lazy_frames.append(pl.concat(batch_frames).lazy())\n return pl.concat(lazy_frames)\n\n def update_schema(self, dataframe):\n for col, dtype in zip(dataframe.columns, dataframe.dtypes):\n if col not in self.global_schema:\n self.global_schema[col] = str(dtype)\n\n def save_schema(self):\n with open(self.schema_path, 'w') as file:\n json.dump(self.global_schema, file)\n\n def preprocess_data(self):\n date_formats = (\"%m/%d/%Y\", \"%Y-%m-%d\")\n date_columns = [\"approvaldate_319D\", \"dateactivated_425D\", \"creationdate_885D\", \"dtlastpmt_581D\", \"employedfrom_700D\", \"dtlastpmtallstes_3545839D\", \"firstnonzeroinstldate_307D\"]\n\n self.train_applprev_1 = self.train_applprev_1.with_columns([\n self.try_parse_date(pl.col(col), *date_formats).alias(col) for col in date_columns\n ]).group_by(\"case_id\").agg([\n pl.col(\"actualdpd_943P\").mean().alias(\"actualdpd_943P_mean\"),\n pl.col(\"annuity_853A\").sum().alias(\"annuity_853A_sum\"),\n pl.col(\"childnum_21L\").sum().alias(\"childnum_21L_sum\"),\n pl.col(\"credacc_actualbalance_314A\").mean().alias(\"credacc_actualbalance_314A_mean\"),\n pl.col(\"credacc_credlmt_575A\").mean().alias(\"credacc_credlmt_575A_mean\"),\n pl.col(\"credacc_maxhisbal_375A\").max().alias(\"credacc_maxhisbal_375A_max\"),\n pl.col(\"credacc_minhisbal_90A\").min().alias(\"credacc_minhisbal_90A_min\"),\n pl.col(\"credacc_transactions_402L\").sum().alias(\"credacc_transactions_402L_sum\"),\n pl.col(\"credamount_590A\").mean().alias(\"credamount_590A_mean\"),\n pl.col(\"currdebt_94A\").mean().alias(\"currdebt_94A_mean\"),\n pl.col(\"downpmt_134A\").sum().alias(\"downpmt_134A_sum\"),\n pl.col(\"mainoccupationinc_437A\").mean().alias(\"mainoccupationinc_437A_mean\"),\n pl.col(\"outstandingdebt_522A\").sum().alias(\"outstandingdebt_522A_sum\"),\n pl.col(\"pmtnum_8L\").max().alias(\"pmtnum_8L_max\"),\n pl.col(\"tenor_203L\").min().alias(\"tenor_203L_min\"),\n pl.col(\"isbidproduct_390L\").cast(pl.UInt32).sum().alias(\"isbidproduct_390L_sum\"),\n pl.col(\"isdebitcard_527L\").cast(pl.UInt32).sum().alias(\"isdebitcard_527L_sum\"),\n pl.col(\"credacc_status_367L\").n_unique().alias(\"credacc_status_367L_n_unique\"),\n pl.col(\"credtype_587L\").n_unique().alias(\"credtype_587L_n_unique\"),\n pl.col(\"education_1138M\").n_unique().alias(\"education_1138M_n_unique\"),\n pl.col(\"familystate_726L\").n_unique().alias(\"familystate_726L_n_unique\"),\n pl.col(\"postype_4733339M\").n_unique().alias(\"postype_4733339M_n_unique\"),\n pl.col(\"profession_152M\").n_unique().alias(\"profession_152M_n_unique\"),\n pl.col(\"rejectreason_755M\").n_unique().alias(\"rejectreason_755M_n_unique\"),\n pl.col(\"rejectreasonclient_4145042M\").n_unique().alias(\"rejectreasonclient_4145042M_n_unique\"),\n pl.col(\"status_219L\").n_unique().alias(\"status_219L_n_unique\"),\n (pl.col(\"approvaldate_319D\").diff().abs().min()).alias(\"approval_to_activation_min_diff\"),\n (pl.col(\"creationdate_885D\").diff().abs().min()).alias(\"creation_min_diff\"),\n (pl.col(\"dtlastpmt_581D\").diff().abs().max()).alias(\"payment_max_diff\"),\n pl.col(\"employedfrom_700D\").min().alias(\"earliest_employment_date\"),\n pl.col(\"byoccupationinc_3656910L\").n_unique().alias(\"byoccupationinc_3656910L_n_unique\"),\n pl.col(\"cancelreason_3545846M\").n_unique().alias(\"cancelreason_3545846M_n_unique\"),\n pl.col(\"district_544M\").n_unique().alias(\"district_544M_n_unique\"),\n pl.col(\"dtlastpmtallstes_3545839D\").min().alias(\"earliest_last_payment_date\"),\n pl.col(\"firstnonzeroinstldate_307D\").min().alias(\"earliest_first_nonzero_installment_date\"),\n pl.col(\"inittransactioncode_279L\").n_unique().alias(\"inittransactioncode_279L_n_unique\"),\n pl.col(\"maxdpdtolerance_577P\").max().alias(\"maximum_dpd_tolerance\"),\n pl.col(\"revolvingaccount_394A\").sum().alias(\"sum_revolving_accounts\")\n ])\n\n # Collect the lazy frame and write to parquet\n self.train_applprev_1 = self.train_applprev_1.collect()\n self.train_applprev_1.write_parquet('train_applprev_1_processed.parquet')\n\n self.train_other_1 = self.train_other_1.group_by(\"case_id\").agg([\n pl.col(\"amtdebitincoming_4809443A\").sum().alias(\"sum_amtdebitincoming\"),\n pl.col(\"amtdebitoutgoing_4809440A\").sum().alias(\"sum_amtdebitoutgoing\"),\n pl.col(\"amtdepositbalance_4809441A\").mean().alias(\"avg_amtdepositbalance\"),\n pl.col(\"amtdepositincoming_4809444A\").sum().alias(\"sum_amtdepositincoming\"),\n pl.col(\"amtdepositoutgoing_4809442A\").sum().alias(\"sum_amtdepositoutgoing\")\n ])\n\n # Collect the lazy frame and write to parquet\n self.train_other_1 = self.train_other_1.collect()\n self.train_other_1.write_parquet('train_other_1_processed.parquet')\n\n self.train_deposit_1 = self.train_deposit_1.group_by(\"case_id\").agg([\n pl.col(\"amount_416A\").mean().alias(\"average_amount\"),\n pl.count(\"openingdate_313D\").alias(\"open_contracts_count\"),\n pl.count(\"contractenddate_991D\").alias(\"closed_contracts_count\")\n ])\n\n # Collect the lazy frame and write to parquet\n self.train_deposit_1 = self.train_deposit_1.collect()\n self.train_deposit_1.write_parquet('train_deposit_1_processed.parquet')\n\n date_format = (\"%m/%d/%Y\", \"%Y-%m-%d\")\n self.train_person_1 = self.train_person_1.with_columns(\n self.try_parse_date(pl.col(\"empl_employedfrom_271D\"), *date_format).alias(\"empl_employedfrom_271D\")\n )\n self.train_person_1 = self.train_person_1.with_columns(\n self.convert_to_ordinal(pl.col(\"empl_employedfrom_271D\")).alias(\"ordinal_employedfrom_271D\")\n )\n self.train_person_1 = self.train_person_1.group_by(\"case_id\").agg([\n pl.col(\"birth_259D\").n_unique().alias(\"unique_birth_dates\"),\n pl.col(\"birthdate_87D\").n_unique().alias(\"unique_birth_dates_87D\"),\n pl.col(\"childnum_185L\").max().alias(\"max_children\"),\n pl.col(\"education_927M\").n_unique().alias(\"unique_educations\"),\n pl.col(\"empl_employedtotal_800L\").n_unique().alias(\"avg_employment_length\"),\n pl.col(\"mainoccupationinc_384A\").sum().alias(\"total_main_income\"),\n pl.col(\"gender_992L\").n_unique().alias(\"unique_genders\"),\n pl.col(\"housetype_905L\").n_unique().alias(\"unique_house_types\"),\n pl.col(\"housingtype_772L\").n_unique().alias(\"unique_housing_types\"),\n pl.col(\"incometype_1044T\").n_unique().alias(\"unique_income_types\"),\n pl.col(\"maritalst_703L\").n_unique().alias(\"unique_marital_statuses\"),\n pl.col(\"persontype_1072L\").n_unique().alias(\"unique_person_types_1072L\"),\n pl.col(\"persontype_792L\").n_unique().alias(\"unique_person_types_792L\"),\n pl.col(\"relationshiptoclient_415T\").n_unique().alias(\"unique_relationships_415T\"),\n pl.col(\"relationshiptoclient_642T\").n_unique().alias(\"unique_relationships_642T\"),\n pl.col(\"remitter_829L\").sum().alias(\"sum_remitters\"),\n pl.col(\"role_1084L\").n_unique().alias(\"unique_roles_1084L\"),\n pl.col(\"role_993L\").n_unique().alias(\"unique_roles_993L\"),\n pl.col(\"safeguarantyflag_411L\").sum().alias(\"sum_safeguaranty_flags\"),\n pl.col(\"sex_738L\").n_unique().alias(\"unique_sexes\"),\n pl.col(\"type_25L\").n_unique().alias(\"unique_contact_types\"),\n pl.col(\"contaddr_district_15M\").n_unique().alias(\"unique_contact_address_districts\"),\n pl.col(\"empladdr_district_926M\").n_unique().alias(\"unique_employer_address_districts\"),\n pl.col(\"registaddr_district_1083M\").n_unique().alias(\"unique_registered_address_districts\"),\n pl.col(\"isreference_387L\").sum().alias(\"sum_is_reference_flags\"),\n pl.col(\"empl_industry_691L\").n_unique().alias(\"unique_industries\"),\n pl.col(\"empladdr_zipcode_114M\").n_unique().alias(\"unique_employer_zipcodes\"),\n pl.col(\"contaddr_zipcode_807M\").n_unique().alias(\"unique_contact_zipcodes\"),\n pl.col(\"registaddr_zipcode_184M\").n_unique().alias(\"unique_registered_zipcodes\"),\n pl.col(\"language1_981M\").n_unique().alias(\"unique_languages\"),\n pl.col(\"familystate_447L\").n_unique().alias(\"unique_family_states\"),\n pl.col(\"contaddr_matchlist_1032L\").sum().alias(\"sum_contact_address_matchlist\"),\n pl.col(\"contaddr_smempladdr_334L\").sum().alias(\"sum_contact_same_employer_address\"),\n pl.col(\"personindex_1023L\").n_unique().alias(\"unique_person_indices\"),\n pl.col(\"ordinal_employedfrom_271D\").max().alias(\"latest_employment_date_ordinal\")\n ])\n\n # Collect the lazy frame and write to parquet\n self.train_person_1 = self.train_person_1.collect()\n self.train_person_1.write_parquet('train_person_1_processed.parquet')\n\n self.train_debitcard_1 = self.train_debitcard_1.with_columns([\n self.convert_to_ordinal(\n pl.col(\"openingdate_857D\").str.strptime(pl.Date, \"%Y-%m-%d\")\n ).alias(\"ordinal_openingdate\")\n ])\n\n self.train_debitcard_1 = self.train_debitcard_1.group_by(\"case_id\").agg([\n pl.col(\"last180dayaveragebalance_704A\").sum().alias(\"total_180dayaveragebalance\"),\n pl.col(\"last180dayturnover_1134A\").sum().alias(\"total_180dayturnover\"),\n pl.col(\"last30dayturnover_651A\").sum().alias(\"total_30dayturnover\"),\n pl.min(\"ordinal_openingdate\").alias(\"earliest_openingdate\")\n ])\n\n # Collect the lazy frame and write to parquet\n self.train_debitcard_1 = self.train_debitcard_1.collect()\n self.train_debitcard_1.write_parquet('train_debitcard_1_processed.parquet')\n\n self.train_tax_registry_a_1 = self.train_tax_registry_a_1.with_columns([\n self.convert_to_ordinal(\n pl.col(\"recorddate_4527225D\").str.strptime(pl.Date, \"%Y-%m-%d\")\n ).alias(\"ordinal_recorddate_4527225D\")\n ])\n\n self.train_tax_registry_a_1 = self.train_tax_registry_a_1.group_by(\"case_id\").agg([\n pl.col(\"amount_4527230A\").sum().alias(\"total_amount_4527230A\"),\n pl.col(\"ordinal_recorddate_4527225D\").max().alias(\"ordinal_recorddate_4527225D\"),\n pl.col(\"name_4527232M\").n_unique().alias(\"unique_name_4527232M\")\n ])\n\n # Collect the lazy frame and write to parquet\n self.train_tax_registry_a_1 = self.train_tax_registry_a_1.collect()\n self.train_tax_registry_a_1.write_parquet('train_tax_registry_a_1_processed.parquet')\n\n self.train_tax_registry_b_1 = self.train_tax_registry_b_1.group_by(\"case_id\").agg([\n pl.col(\"amount_4917619A\").sum().alias(\"total_amount_4917619A\"),\n pl.col(\"deductiondate_4917603D\").n_unique().alias(\"unique_deductiondate_4917603D\"),\n pl.col(\"name_4917606M\").n_unique().alias(\"unique_name_4917606M\")\n ])\n\n # Collect the lazy frame and write to parquet\n self.train_tax_registry_b_1 = self.train_tax_registry_b_1.collect()\n self.train_tax_registry_b_1.write_parquet('train_tax_registry_b_1_processed.parquet')\n\n self.train_tax_registry_c_1 = self.train_tax_registry_c_1.group_by(\"case_id\").agg([\n pl.col(\"pmtamount_36A\").sum().alias(\"total_pmtamount_36A\"),\n pl.col(\"processingdate_168D\").n_unique().alias(\"unique_processingdate_168D\"),\n pl.col(\"employername_160M\").n_unique().alias(\"unique_employername_160M\")\n ])\n\n # Collect the lazy frame and write to parquet\n self.train_tax_registry_c_1 = self.train_tax_registry_c_1.collect()\n self.train_tax_registry_c_1.write_parquet('train_tax_registry_c_1_processed.parquet')\n\n date_columns_2 = [column for column in self.train_credit_bureau_a_1.columns if column[-1] == 'D']\n amount_column = [column for column in self.train_credit_bureau_a_1.columns if column[-1] == 'A']\n dpd_column = [column for column in self.train_credit_bureau_a_1.columns if column[-1] == 'P']\n mask_column = [column for column in self.train_credit_bureau_a_1.columns if column[-1] == 'M']\n other_columns = [column for column in self.train_credit_bureau_a_1.columns if column[-1] == 'T' or column[-1] == 'L']\n\n for col in date_columns_2:\n self.train_credit_bureau_a_1 = self.train_credit_bureau_a_1.with_columns(\n self.convert_to_ordinal(pl.col(col)).alias(f\"ordinal_{col}\")\n )\n\n aggregations = []\n\n for col in self.train_credit_bureau_a_1.columns:\n if col in amount_column:\n aggregations.append(pl.col(col).sum().alias(f\"total_{col}\"))\n elif col in dpd_column:\n aggregations.append(pl.col(col).sum().alias(f\"total_{col}\"))\n elif col in mask_column:\n aggregations.append(pl.col(col).n_unique().alias(f\"unique_{col}\"))\n elif col in date_columns_2:\n aggregations.append(pl.col(col).max().alias(f\"max_{col}\"))\n elif col in other_columns:\n aggregations.append(pl.col(col).n_unique().alias(f\"unique_{col}\"))\n\n self.train_credit_bureau_a_1 = self.train_credit_bureau_a_1.group_by(\"case_id\").agg(aggregations)\n self.train_credit_bureau_a_1 = self.train_credit_bureau_a_1.collect()\n self.train_credit_bureau_a_1.write_parquet('train_credit_bureau_a_1_processed.parquet')\n\n date_columns_3 = [column for column in self.train_credit_bureau_b_1.columns if column[-1] == 'D']\n amount_column_2 = [column for column in self.train_credit_bureau_b_1.columns if column[-1] == 'A']\n dpd_column_2 = [column for column in self.train_credit_bureau_b_1.columns if column[-1] == 'P']\n mask_column_2 = [column for column in self.train_credit_bureau_b_1.columns if column[-1] == 'M']\n other_columns_2 = [column for column in self.train_credit_bureau_b_1.columns if column[-1] == 'T' or column[-1] == 'L']\n\n for col in date_columns_3:\n self.train_credit_bureau_b_1 = self.train_credit_bureau_b_1.with_columns(\n self.convert_to_ordinal(pl.col(col)).alias(f\"ordinal_{col}\")\n )\n\n aggregations_2 = []\n\n for col in self.train_credit_bureau_b_1.columns:\n if col in amount_column_2:\n aggregations_2.append(pl.col(col).sum().alias(f\"total_{col}\"))\n elif col in dpd_column_2:\n aggregations_2.append(pl.col(col).sum().alias(f\"total_{col}\"))\n elif col in mask_column_2:\n aggregations_2.append(pl.col(col).n_unique().alias(f\"unique_{col}\"))\n elif col in date_columns_3:\n aggregations_2.append(pl.col(col).max().alias(f\"max_{col}\"))\n elif col in other_columns_2:\n aggregations_2.append(pl.col(col).n_unique().alias(f\"unique_{col}\"))\n\n self.train_credit_bureau_b_1 = self.train_credit_bureau_b_1.group_by(\"case_id\").agg(aggregations_2)\n self.train_credit_bureau_b_1 = self.train_credit_bureau_b_1.collect()\n self.train_credit_bureau_b_1.write_parquet('train_credit_bureau_b_1_processed.parquet')\n\n\n def merge_data(self):\n # Load the preprocessed dataframes from disk if they are too large\n self.train_applprev_1 = pl.read_parquet('train_applprev_1_processed.parquet')\n self.train_other_1 = pl.read_parquet('train_other_1_processed.parquet')\n self.train_deposit_1 = pl.read_parquet('train_deposit_1_processed.parquet')\n self.train_person_1 = pl.read_parquet('train_person_1_processed.parquet')\n self.train_debitcard_1 = pl.read_parquet('train_debitcard_1_processed.parquet')\n self.train_tax_registry_a_1 = pl.read_parquet('train_tax_registry_a_1_processed.parquet')\n self.train_tax_registry_b_1 = pl.read_parquet('train_tax_registry_b_1_processed.parquet')\n self.train_tax_registry_c_1 = pl.read_parquet('train_tax_registry_c_1_processed.parquet')\n self.train_credit_bureau_a_1 = pl.read_parquet('train_credit_bureau_a_1_processed.parquet')\n self.train_credit_bureau_b_1 = pl.read_parquet('train_credit_bureau_b_1_processed.parquet')\n\n df_joined = self.train_applprev_1.join(self.train_other_1, on=\"case_id\", how=\"left\")\n df_joined = df_joined.join(self.train_deposit_1, on=\"case_id\", how=\"left\")\n df_joined = df_joined.join(self.train_person_1, on=\"case_id\", how=\"left\")\n df_joined = df_joined.join(self.train_debitcard_1, on=\"case_id\", how=\"left\")\n df_joined = df_joined.join(self.train_tax_registry_a_1, on=\"case_id\", how=\"left\")\n df_joined = df_joined.join(self.train_tax_registry_b_1, on=\"case_id\", how=\"left\")\n df_joined = df_joined.join(self.train_tax_registry_c_1, on=\"case_id\", how=\"left\")\n df_joined = df_joined.join(self.train_credit_bureau_a_1, on=\"case_id\", how=\"left\")\n df_joined = df_joined.join(self.train_credit_bureau_b_1, on=\"case_id\", how=\"left\")\n\n string_date_columns = [\n 'max_dateofcredend_289D', 'max_dateofcredend_353D', 'max_dateofcredstart_181D', \n 'max_dateofcredstart_739D', 'max_dateofrealrepmt_138D', 'max_lastupdate_1112D', \n 'max_lastupdate_388D', 'max_numberofoverdueinstlmaxdat_148D', \n 'max_numberofoverdueinstlmaxdat_641D', 'max_overdueamountmax2date_1002D', \n 'max_overdueamountmax2date_1142D', 'max_refreshdate_3813885D', \n 'max_contractdate_551D', 'max_contractmaturitydate_151D', 'max_lastupdate_260D'\n ]\n\n for column in string_date_columns:\n df_joined = df_joined.with_columns(\n self.try_parse_date(pl.col(column), '%Y-%m-%d', '%m/%d/%Y').alias(column)\n )\n\n date_columns = [col for col, dtype in zip(df_joined.columns, df_joined.dtypes) if dtype == pl.Date]\n for col in date_columns:\n df_joined = df_joined.with_columns(\n self.convert_to_ordinal(df_joined[col]).alias(col)\n )\n\n duration_columns = [\"approval_to_activation_min_diff\", \"creation_min_diff\", \"payment_max_diff\"]\n for column in duration_columns:\n df_joined = df_joined.with_columns(\n pl.col(column).str.replace(\"d\", \"\").cast(pl.Int64) * 1440\n )\n return df_joined\n\n\n def execute_pipeline(self):\n self.load_data()\n self.preprocess_data()\n\n # Clean up memory by deleting intermediate dataframes after preprocessing\n gc.collect()\n\n merged_data = self.merge_data()\n\n # Final cleanup after merging\n gc.collect()\n\n return merged_data\n\nif __name__ == \"__main__\":\n dataPath = \"/kaggle/input/home-credit-credit-risk-model-stability/\"\n applprev_paths = [\n dataPath + \"parquet_files/train/train_applprev_1_0.parquet\",\n dataPath + \"parquet_files/train/train_applprev_1_1.parquet\"\n ]\n other_path = dataPath + \"parquet_files/train/train_other_1.parquet\"\n deposit_path = dataPath + \"parquet_files/train/train_deposit_1.parquet\"\n person_path = dataPath + \"parquet_files/train/train_person_1.parquet\"\n debitcard_path = dataPath + \"parquet_files/train/train_debitcard_1.parquet\"\n tax_registry_a_1_path = dataPath + \"parquet_files/train/train_tax_registry_a_1.parquet\"\n tax_registry_b_1_path = dataPath + \"parquet_files/train/train_tax_registry_b_1.parquet\"\n tax_registry_c_1_path = dataPath + \"parquet_files/train/train_tax_registry_c_1.parquet\"\n credit_bureau_a_1_paths = [\n dataPath + \"parquet_files/train/train_credit_bureau_a_1_0.parquet\",\n dataPath + \"parquet_files/train/train_credit_bureau_a_1_1.parquet\",\n dataPath + \"parquet_files/train/train_credit_bureau_a_1_2.parquet\",\n dataPath + \"parquet_files/train/train_credit_bureau_a_1_3.parquet\"\n ]\n credit_bureau_b_1_path = dataPath + \"parquet_files/train/train_credit_bureau_b_1.parquet\"\n schema_path = \"/kaggle/input/unified-schema/unified_schema_3.json\"\n\n pipeline = DataPipeline_Depth_1(applprev_paths, other_path, deposit_path, person_path, debitcard_path, tax_registry_a_1_path, tax_registry_b_1_path, \n tax_registry_c_1_path, credit_bureau_a_1_paths, credit_bureau_b_1_path, schema_path)\n Depth_1 = pipeline.execute_pipeline()\n\n # Clean up memory after executing the pipeline\n del pipeline, dataPath, schema_path\n gc.collect()","metadata":{"execution":{"iopub.status.busy":"2024-05-27T18:12:22.753016Z","iopub.execute_input":"2024-05-27T18:12:22.753486Z","iopub.status.idle":"2024-05-27T18:15:17.020459Z","shell.execute_reply.started":"2024-05-27T18:12:22.753443Z","shell.execute_reply":"2024-05-27T18:15:17.018994Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"class DataPipeline_Depth_1:\n def __init__(self, applprev_paths, other_path, deposit_path, person_path, debitcard_path, tax_registry_a_1_path, tax_registry_b_1_path, \n tax_registry_c_1_path, credit_bureau_a_1_paths, credit_bureau_b_1_path, schema_path):\n self.applprev_paths = applprev_paths\n self.other_path = other_path\n self.deposit_path = deposit_path\n self.person_path = person_path\n self.debitcard_path = debitcard_path\n self.tax_registry_a_1_path = tax_registry_a_1_path\n self.tax_registry_b_1_path = tax_registry_b_1_path\n self.tax_registry_c_1_path = tax_registry_c_1_path\n self.credit_bureau_a_1_paths = credit_bureau_a_1_paths\n self.credit_bureau_b_1_path = credit_bureau_b_1_path\n self.schema_path = schema_path\n self.global_schema = self.load_schema()\n\n def load_schema(self):\n with open(self.schema_path, 'r') as file:\n return json.load(file)\n \n @staticmethod\n def dtype_mapping(dtype_str):\n mapping = {\n 'Int32': pl.Int32,\n 'Int64': pl.Int64,\n 'Float32': pl.Float32,\n 'Float64': pl.Float64,\n 'Utf8': pl.Utf8,\n 'Boolean': pl.Boolean,\n 'Date': pl.Date,\n 'Categorical': pl.Categorical\n }\n return mapping.get(dtype_str, pl.Utf8)\n\n @staticmethod\n def try_parse_date(col, fmt1, fmt2):\n date1 = col.str.strptime(pl.Date, fmt1, strict=False)\n date2 = col.str.strptime(pl.Date, fmt2, strict=False)\n return pl.when(date1.is_not_null()).then(date1).otherwise(date2)\n\n @staticmethod\n def convert_to_ordinal(date):\n return pl.when(date.is_not_null()).then(\n (date.dt.year() * 365) + (date.dt.month() * 30) + date.dt.day()\n ).otherwise(None)\n \n def load_and_ensure_schema(self, path):\n df = pl.read_parquet(path)\n for col, expected_dtype in self.global_schema.items():\n expected_pl_dtype = self.dtype_mapping(expected_dtype)\n if col in df.columns:\n if df[col].dtype != expected_pl_dtype:\n df = df.with_columns(df[col].cast(expected_pl_dtype))\n return df\n\n def load_data(self):\n self.train_applprev_1 = pl.concat([self.load_and_ensure_schema(path) for path in self.applprev_paths]).lazy()\n self.train_other_1 = self.load_and_ensure_schema(self.other_path).lazy()\n self.train_deposit_1 = self.load_and_ensure_schema(self.deposit_path).lazy()\n self.train_person_1 = self.load_and_ensure_schema(self.person_path).lazy()\n self.train_debitcard_1 = self.load_and_ensure_schema(self.debitcard_path).lazy()\n self.train_tax_registry_a_1 = self.load_and_ensure_schema(self.tax_registry_a_1_path).lazy()\n self.train_tax_registry_b_1 = self.load_and_ensure_schema(self.tax_registry_b_1_path).lazy()\n self.train_tax_registry_c_1 = self.load_and_ensure_schema(self.tax_registry_c_1_path).lazy()\n self.train_credit_bureau_a_1 = pl.concat([self.load_and_ensure_schema(path) for path in self.credit_bureau_a_1_paths]).lazy()\n self.train_credit_bureau_b_1 = self.load_and_ensure_schema(self.credit_bureau_b_1_path).lazy()\n\n def preprocess_data(self):\n date_formats = (\"%m/%d/%Y\", \"%Y-%m-%d\")\n date_columns = [\"approvaldate_319D\", \"dateactivated_425D\", \"creationdate_885D\", \"dtlastpmt_581D\", \"employedfrom_700D\", \"dtlastpmtallstes_3545839D\", \"firstnonzeroinstldate_307D\"]\n self.train_applprev_1 = self.train_applprev_1.with_columns([\n DataPipeline_Depth_1.try_parse_date(pl.col(col), *date_formats).alias(col) for col in date_columns\n ]).group_by(\"case_id\").agg([\n pl.col(\"actualdpd_943P\").mean().alias(\"actualdpd_943P_mean\"),\n pl.col(\"annuity_853A\").sum().alias(\"annuity_853A_sum\"),\n pl.col(\"childnum_21L\").sum().alias(\"childnum_21L_sum\"),\n pl.col(\"credacc_actualbalance_314A\").mean().alias(\"credacc_actualbalance_314A_mean\"),\n pl.col(\"credacc_credlmt_575A\").mean().alias(\"credacc_credlmt_575A_mean\"),\n pl.col(\"credacc_maxhisbal_375A\").max().alias(\"credacc_maxhisbal_375A_max\"),\n pl.col(\"credacc_minhisbal_90A\").min().alias(\"credacc_minhisbal_90A_min\"),\n pl.col(\"credacc_transactions_402L\").sum().alias(\"credacc_transactions_402L_sum\"),\n pl.col(\"credamount_590A\").mean().alias(\"credamount_590A_mean\"),\n pl.col(\"currdebt_94A\").mean().alias(\"currdebt_94A_mean\"),\n pl.col(\"downpmt_134A\").sum().alias(\"downpmt_134A_sum\"),\n pl.col(\"mainoccupationinc_437A\").mean().alias(\"mainoccupationinc_437A_mean\"),\n pl.col(\"outstandingdebt_522A\").sum().alias(\"outstandingdebt_522A_sum\"),\n pl.col(\"pmtnum_8L\").max().alias(\"pmtnum_8L_max\"),\n pl.col(\"tenor_203L\").min().alias(\"tenor_203L_min\"),\n pl.col(\"isbidproduct_390L\").cast(pl.UInt32).sum().alias(\"isbidproduct_390L_sum\"),\n pl.col(\"isdebitcard_527L\").cast(pl.UInt32).sum().alias(\"isdebitcard_527L_sum\"),\n pl.col(\"credacc_status_367L\").n_unique().alias(\"credacc_status_367L_n_unique\"),\n pl.col(\"credtype_587L\").n_unique().alias(\"credtype_587L_n_unique\"),\n pl.col(\"education_1138M\").n_unique().alias(\"education_1138M_n_unique\"),\n pl.col(\"familystate_726L\").n_unique().alias(\"familystate_726L_n_unique\"),\n pl.col(\"postype_4733339M\").n_unique().alias(\"postype_4733339M_n_unique\"),\n pl.col(\"profession_152M\").n_unique().alias(\"profession_152M_n_unique\"),\n pl.col(\"rejectreason_755M\").n_unique().alias(\"rejectreason_755M_n_unique\"),\n pl.col(\"rejectreasonclient_4145042M\").n_unique().alias(\"rejectreasonclient_4145042M_n_unique\"),\n pl.col(\"status_219L\").n_unique().alias(\"status_219L_n_unique\"),\n (pl.col(\"approvaldate_319D\").diff().abs().min()).alias(\"approval_to_activation_min_diff\"),\n (pl.col(\"creationdate_885D\").diff().abs().min()).alias(\"creation_min_diff\"),\n (pl.col(\"dtlastpmt_581D\").diff().abs().max()).alias(\"payment_max_diff\"),\n pl.col(\"employedfrom_700D\").min().alias(\"earliest_employment_date\"),\n pl.col(\"byoccupationinc_3656910L\").n_unique().alias(\"byoccupationinc_3656910L_n_unique\"),\n pl.col(\"cancelreason_3545846M\").n_unique().alias(\"cancelreason_3545846M_n_unique\"),\n pl.col(\"district_544M\").n_unique().alias(\"district_544M_n_unique\"),\n pl.col(\"dtlastpmtallstes_3545839D\").min().alias(\"earliest_last_payment_date\"),\n pl.col(\"firstnonzeroinstldate_307D\").min().alias(\"earliest_first_nonzero_installment_date\"),\n pl.col(\"inittransactioncode_279L\").n_unique().alias(\"inittransactioncode_279L_n_unique\"),\n pl.col(\"maxdpdtolerance_577P\").max().alias(\"maximum_dpd_tolerance\"),\n pl.col(\"revolvingaccount_394A\").sum().alias(\"sum_revolving_accounts\")\n ])\n\n self.train_applprev_1 = self.train_applprev_1.collect()\n self.train_applprev_1.write_parquet('train_applprev_1_processed.parquet')\n\n self.train_other_1 = self.train_other_1.group_by(\"case_id\").agg([\n pl.col(\"amtdebitincoming_4809443A\").sum().alias(\"sum_amtdebitincoming\"),\n pl.col(\"amtdebitoutgoing_4809440A\").sum().alias(\"sum_amtdebitoutgoing\"),\n pl.col(\"amtdepositbalance_4809441A\").mean().alias(\"avg_amtdepositbalance\"),\n pl.col(\"amtdepositincoming_4809444A\").sum().alias(\"sum_amtdepositincoming\"),\n pl.col(\"amtdepositoutgoing_4809442A\").sum().alias(\"sum_amtdepositoutgoing\")\n ])\n\n self.train_other_1 = self.train_other_1.collect()\n self.train_other_1.write_parquet('train_other_1_processed.parquet')\n\n self.train_deposit_1 = self.train_deposit_1.group_by(\"case_id\").agg([\n pl.col(\"amount_416A\").mean().alias(\"average_amount\"),\n pl.count(\"openingdate_313D\").alias(\"open_contracts_count\"),\n pl.count(\"contractenddate_991D\").alias(\"closed_contracts_count\")\n ])\n\n self.train_deposit_1 = self.train_deposit_1.collect()\n self.train_deposit_1.write_parquet('train_deposit_1_processed.parquet')\n\n date_format = (\"%m/%d/%Y\", \"%Y-%m-%d\")\n self.train_person_1 = self.train_person_1.with_columns(\n DataPipeline_Depth_1.try_parse_date(pl.col(\"empl_employedfrom_271D\"), *date_format).alias(\"empl_employedfrom_271D\")\n )\n self.train_person_1 = self.train_person_1.with_columns(\n DataPipeline_Depth_1.convert_to_ordinal(pl.col(\"empl_employedfrom_271D\")).alias(\"ordinal_employedfrom_271D\")\n )\n self.train_person_1 = self.train_person_1.group_by(\"case_id\").agg([\n pl.col(\"birth_259D\").n_unique().alias(\"unique_birth_dates\"),\n pl.col(\"birthdate_87D\").n_unique().alias(\"unique_birth_dates_87D\"),\n pl.col(\"childnum_185L\").max().alias(\"max_children\"),\n pl.col(\"education_927M\").n_unique().alias(\"unique_educations\"),\n pl.col(\"empl_employedtotal_800L\").n_unique().alias(\"avg_employment_length\"),\n pl.col(\"mainoccupationinc_384A\").sum().alias(\"total_main_income\"),\n pl.col(\"gender_992L\").n_unique().alias(\"unique_genders\"),\n pl.col(\"housetype_905L\").n_unique().alias(\"unique_house_types\"),\n pl.col(\"housingtype_772L\").n_unique().alias(\"unique_housing_types\"),\n pl.col(\"incometype_1044T\").n_unique().alias(\"unique_income_types\"),\n pl.col(\"maritalst_703L\").n_unique().alias(\"unique_marital_statuses\"),\n pl.col(\"persontype_1072L\").n_unique().alias(\"unique_person_types_1072L\"),\n pl.col(\"persontype_792L\").n_unique().alias(\"unique_person_types_792L\"),\n pl.col(\"relationshiptoclient_415T\").n_unique().alias(\"unique_relationships_415T\"),\n pl.col(\"relationshiptoclient_642T\").n_unique().alias(\"unique_relationships_642T\"),\n pl.col(\"remitter_829L\").sum().alias(\"sum_remitters\"),\n pl.col(\"role_1084L\").n_unique().alias(\"unique_roles_1084L\"),\n pl.col(\"role_993L\").n_unique().alias(\"unique_roles_993L\"),\n pl.col(\"safeguarantyflag_411L\").sum().alias(\"sum_safeguaranty_flags\"),\n pl.col(\"sex_738L\").n_unique().alias(\"unique_sexes\"),\n pl.col(\"type_25L\").n_unique().alias(\"unique_contact_types\"),\n pl.col(\"contaddr_district_15M\").n_unique().alias(\"unique_contact_address_districts\"),\n pl.col(\"empladdr_district_926M\").n_unique().alias(\"unique_employer_address_districts\"),\n pl.col(\"registaddr_district_1083M\").n_unique().alias(\"unique_registered_address_districts\"),\n pl.col(\"isreference_387L\").sum().alias(\"sum_is_reference_flags\"),\n pl.col(\"empl_industry_691L\").n_unique().alias(\"unique_industries\"),\n pl.col(\"empladdr_zipcode_114M\").n_unique().alias(\"unique_employer_zipcodes\"),\n pl.col(\"contaddr_zipcode_807M\").n_unique().alias(\"unique_contact_zipcodes\"),\n pl.col(\"registaddr_zipcode_184M\").n_unique().alias(\"unique_registered_zipcodes\"),\n pl.col(\"language1_981M\").n_unique().alias(\"unique_languages\"),\n pl.col(\"familystate_447L\").n_unique().alias(\"unique_family_states\"),\n pl.col(\"contaddr_matchlist_1032L\").sum().alias(\"sum_contact_address_matchlist\"),\n pl.col(\"contaddr_smempladdr_334L\").sum().alias(\"sum_contact_same_employer_address\"),\n pl.col(\"personindex_1023L\").n_unique().alias(\"unique_person_indices\"),\n pl.col(\"ordinal_employedfrom_271D\").max().alias(\"latest_employment_date_ordinal\")\n ])\n\n self.train_person_1 = self.train_person_1.collect()\n self.train_person_1.write_parquet('train_person_1_processed.parquet')\n\n self.train_debitcard_1 = self.train_debitcard_1.with_columns([\n DataPipeline_Depth_1.convert_to_ordinal(\n pl.col(\"openingdate_857D\").str.strptime(pl.Date, \"%Y-%m-%d\")\n ).alias(\"ordinal_openingdate\")\n ])\n\n self.train_debitcard_1 = self.train_debitcard_1.group_by(\"case_id\").agg([\n pl.col(\"last180dayaveragebalance_704A\").sum().alias(\"total_180dayaveragebalance\"),\n pl.col(\"last180dayturnover_1134A\").sum().alias(\"total_180dayturnover\"),\n pl.col(\"last30dayturnover_651A\").sum().alias(\"total_30dayturnover\"),\n pl.min(\"ordinal_openingdate\").alias(\"earliest_openingdate\")\n ])\n\n self.train_debitcard_1 = self.train_debitcard_1.collect()\n self.train_debitcard_1.write_parquet('train_debitcard_1_processed.parquet')\n\n self.train_tax_registry_a_1 = self.train_tax_registry_a_1.with_columns([\n DataPipeline_Depth_1.convert_to_ordinal(\n pl.col(\"recorddate_4527225D\").str.strptime(pl.Date, \"%Y-%m-%d\")\n ).alias(\"ordinal_recorddate_4527225D\")\n ])\n\n self.train_tax_registry_a_1 = self.train_tax_registry_a_1.group_by(\"case_id\").agg([\n pl.col(\"amount_4527230A\").sum().alias(\"total_amount_4527230A\"),\n pl.col(\"ordinal_recorddate_4527225D\").max().alias(\"ordinal_recorddate_4527225D\"),\n pl.col(\"name_4527232M\").n_unique().alias(\"unique_name_4527232M\")\n ])\n\n self.train_tax_registry_a_1 = self.train_tax_registry_a_1.collect()\n self.train_tax_registry_a_1.write_parquet('train_tax_registry_a_1_processed.parquet')\n\n self.train_tax_registry_b_1 = self.train_tax_registry_b_1.group_by(\"case_id\").agg([\n pl.col(\"amount_4917619A\").sum().alias(\"total_amount_4917619A\"),\n pl.col(\"deductiondate_4917603D\").n_unique().alias(\"unique_deductiondate_4917603D\"),\n pl.col(\"name_4917606M\").n_unique().alias(\"unique_name_4917606M\")\n ])\n\n self.train_tax_registry_b_1 = self.train_tax_registry_b_1.collect()\n self.train_tax_registry_b_1.write_parquet('train_tax_registry_b_1_processed.parquet')\n\n self.train_tax_registry_c_1 = self.train_tax_registry_c_1.group_by(\"case_id\").agg([\n pl.col(\"pmtamount_36A\").sum().alias(\"total_pmtamount_36A\"),\n pl.col(\"processingdate_168D\").n_unique().alias(\"unique_processingdate_168D\"),\n pl.col(\"employername_160M\").n_unique().alias(\"unique_employername_160M\")\n ])\n\n self.train_tax_registry_c_1 = self.train_tax_registry_c_1.collect()\n self.train_tax_registry_c_1.write_parquet('train_tax_registry_c_1_processed.parquet')\n\n date_columns_2 = [column for column in self.train_credit_bureau_a_1.columns if column[-1] == 'D']\n amount_column = [column for column in self.train_credit_bureau_a_1.columns if column[-1] == 'A']\n dpd_column = [column for column in self.train_credit_bureau_a_1.columns if column[-1] == 'P']\n mask_column = [column for column in self.train_credit_bureau_a_1.columns if column[-1] == 'M']\n other_columns = [column for column in self.train_credit_bureau_a_1.columns if column[-1] == 'T' or column[-1] == 'L']\n\n for col in date_columns_2:\n self.train_credit_bureau_a_1 = self.train_credit_bureau_a_1.with_columns(\n DataPipeline_Depth_1.convert_to_ordinal(pl.col(col)).alias(f\"ordinal_{col}\")\n )\n \n aggregations = []\n\n for col in self.train_credit_bureau_a_1.columns:\n if col in amount_column:\n aggregations.append(pl.col(col).sum().alias(f\"total_{col}\"))\n elif col in dpd_column:\n aggregations.append(pl.col(col).sum().alias(f\"total_{col}\"))\n elif col in mask_column:\n aggregations.append(pl.col(col).n_unique().alias(f\"unique_{col}\"))\n elif col in date_columns_2:\n aggregations.append(pl.col(col).max().alias(f\"max_{col}\"))\n elif col in other_columns:\n aggregations.append(pl.col(col).n_unique().alias(f\"unique_{col}\"))\n\n self.train_credit_bureau_a_1 = self.train_credit_bureau_a_1.group_by(\"case_id\").agg(aggregations)\n\n self.train_credit_bureau_a_1 = self.train_credit_bureau_a_1.collect()\n self.train_credit_bureau_a_1.write_parquet('train_credit_bureau_a_1_processed.parquet')\n\n date_columns_3 = [column for column in self.train_credit_bureau_b_1.columns if column[-1] == 'D']\n amount_column_2 = [column for column in self.train_credit_bureau_b_1.columns if column[-1] == 'A']\n dpd_column_2 = [column for column in self.train_credit_bureau_b_1.columns if column[-1] == 'P']\n mask_column_2 = [column for column in self.train_credit_bureau_b_1.columns if column[-1] == 'M']\n other_columns_2 = [column for column in self.train_credit_bureau_b_1.columns if column[-1] == 'T' or column[-1] == 'L']\n\n for col in date_columns_3:\n self.train_credit_bureau_b_1 = self.train_credit_bureau_b_1.with_columns(\n DataPipeline_Depth_1.convert_to_ordinal(pl.col(col)).alias(f\"ordinal_{col}\")\n )\n \n aggregations_2 = []\n\n for col in self.train_credit_bureau_b_1.columns:\n if col in amount_column_2:\n aggregations_2.append(pl.col(col).sum().alias(f\"total_{col}\"))\n elif col in dpd_column_2:\n aggregations_2.append(pl.col(col).sum().alias(f\"total_{col}\"))\n elif col in mask_column_2:\n aggregations_2.append(pl.col(col).n_unique().alias(f\"unique_{col}\"))\n elif col in date_columns_3:\n aggregations_2.append(pl.col(col).max().alias(f\"max_{col}\"))\n elif col in other_columns_2:\n aggregations_2.append(pl.col(col).n_unique().alias(f\"unique_{col}\"))\n\n self.train_credit_bureau_b_1 = self.train_credit_bureau_b_1.group_by(\"case_id\").agg(aggregations_2)\n\n self.train_credit_bureau_b_1 = self.train_credit_bureau_b_1.collect()\n self.train_credit_bureau_b_1.write_parquet('train_credit_bureau_b_1_processed.parquet')\n\n def merge_data(self):\n # Load the preprocessed dataframes from disk if they are too large\n self.train_applprev_1 = pl.read_parquet('train_applprev_1_processed.parquet')\n self.train_other_1 = pl.read_parquet('train_other_1_processed.parquet')\n self.train_deposit_1 = pl.read_parquet('train_deposit_1_processed.parquet')\n self.train_person_1 = pl.read_parquet('train_person_1_processed.parquet')\n self.train_debitcard_1 = pl.read_parquet('train_debitcard_1_processed.parquet')\n self.train_tax_registry_a_1 = pl.read_parquet('train_tax_registry_a_1_processed.parquet')\n self.train_tax_registry_b_1 = pl.read_parquet('train_tax_registry_b_1_processed.parquet')\n self.train_tax_registry_c_1 = pl.read_parquet('train_tax_registry_c_1_processed.parquet')\n self.train_credit_bureau_a_1 = pl.read_parquet('train_credit_bureau_a_1_processed.parquet')\n self.train_credit_bureau_b_1 = pl.read_parquet('train_credit_bureau_b_1_processed.parquet')\n\n df_joined = self.train_applprev_1.join(self.train_other_1, on=\"case_id\", how=\"left\")\n df_joined = df_joined.join(self.train_deposit_1, on=\"case_id\", how=\"left\")\n df_joined = df_joined.join(self.train_person_1, on=\"case_id\", how=\"left\")\n df_joined = df_joined.join(self.train_debitcard_1, on=\"case_id\", how=\"left\")\n df_joined = df_joined.join(self.train_tax_registry_a_1, on=\"case_id\", how=\"left\")\n df_joined = df_joined.join(self.train_tax_registry_b_1, on=\"case_id\", how=\"left\")\n df_joined = df_joined.join(self.train_tax_registry_c_1, on=\"case_id\", how=\"left\")\n df_joined = df_joined.join(self.train_credit_bureau_a_1, on=\"case_id\", how=\"left\")\n df_joined = df_joined.join(self.train_credit_bureau_b_1, on=\"case_id\", how=\"left\")\n\n string_date_columns = [\n 'max_dateofcredend_289D', 'max_dateofcredend_353D', 'max_dateofcredstart_181D', \n 'max_dateofcredstart_739D', 'max_dateofrealrepmt_138D', 'max_lastupdate_1112D', \n 'max_lastupdate_388D', 'max_numberofoverdueinstlmaxdat_148D', \n 'max_numberofoverdueinstlmaxdat_641D', 'max_overdueamountmax2date_1002D', \n 'max_overdueamountmax2date_1142D', 'max_refreshdate_3813885D', \n 'max_contractdate_551D', 'max_contractmaturitydate_151D', 'max_lastupdate_260D'\n ]\n\n for column in string_date_columns:\n df_joined = df_joined.with_columns(\n DataPipeline_Depth_1.try_parse_date(pl.col(column), '%Y-%m-%d', '%m/%d/%Y').alias(column)\n )\n\n date_columns = [col for col, dtype in zip(df_joined.columns, df_joined.dtypes) if dtype == pl.Date]\n for col in date_columns:\n df_joined = df_joined.with_columns(\n DataPipeline_Depth_1.convert_to_ordinal(df_joined[col]).alias(col)\n )\n\n duration_columns = [\"approval_to_activation_min_diff\", \"creation_min_diff\", \"payment_max_diff\"]\n for column in duration_columns:\n df_joined = df_joined.with_columns(\n pl.col(column).str.replace(\"d\", \"\").cast(pl.Int64) * 1440\n )\n return df_joined\n\n def execute_pipeline(self):\n self.load_data()\n self.preprocess_data()\n\n merged_data = self.merge_data()\n\n # Clean up memory by deleting intermediate dataframes\n del self.train_applprev_1, self.train_other_1, self.train_deposit_1, self.train_person_1, self.train_debitcard_1\n del self.train_tax_registry_a_1, self.train_tax_registry_b_1, self.train_tax_registry_c_1\n del self.train_credit_bureau_a_1, self.train_credit_bureau_b_1\n gc.collect()\n\n return merged_data\n\nif __name__ == \"__main__\":\n dataPath = \"/kaggle/input/home-credit-credit-risk-model-stability/\"\n applprev_paths = [\n dataPath + \"parquet_files/test/test_applprev_1_0.parquet\",\n dataPath + \"parquet_files/test/test_applprev_1_1.parquet\",\n dataPath + \"parquet_files/test/test_applprev_1_2.parquet\"\n ]\n other_path = dataPath + \"parquet_files/test/test_other_1.parquet\"\n deposit_path = dataPath + \"parquet_files/test/test_deposit_1.parquet\"\n person_path = dataPath + \"parquet_files/test/test_person_1.parquet\"\n debitcard_path = dataPath + \"parquet_files/test/test_debitcard_1.parquet\"\n tax_registry_a_1_path = dataPath + \"parquet_files/test/test_tax_registry_a_1.parquet\"\n tax_registry_b_1_path = dataPath + \"parquet_files/test/test_tax_registry_b_1.parquet\"\n tax_registry_c_1_path = dataPath + \"parquet_files/test/test_tax_registry_c_1.parquet\"\n credit_bureau_a_1_paths = [\n dataPath + \"parquet_files/test/test_credit_bureau_a_1_0.parquet\",\n dataPath + \"parquet_files/test/test_credit_bureau_a_1_1.parquet\",\n dataPath + \"parquet_files/test/test_credit_bureau_a_1_2.parquet\",\n dataPath + \"parquet_files/test/test_credit_bureau_a_1_3.parquet\",\n dataPath + \"parquet_files/test/test_credit_bureau_a_1_4.parquet\"\n ]\n credit_bureau_b_1_path = dataPath + \"parquet_files/test/test_credit_bureau_b_1.parquet\"\n schema_path = \"/kaggle/input/unified-schemav2/unified_schema_3.json\"\n\n pipeline = DataPipeline_Depth_1(applprev_paths, other_path, deposit_path, person_path, debitcard_path, tax_registry_a_1_path, tax_registry_b_1_path, \n tax_registry_c_1_path, credit_bureau_a_1_paths, credit_bureau_b_1_path, schema_path)\n Depth_1_test = pipeline.execute_pipeline()\n\n # Clean up memory after executing the pipeline\n del pipeline, dataPath, schema_path\n gc.collect()","metadata":{"execution":{"iopub.status.busy":"2024-05-27T18:15:17.023448Z","iopub.execute_input":"2024-05-27T18:15:17.023821Z","iopub.status.idle":"2024-05-27T18:15:17.629278Z","shell.execute_reply.started":"2024-05-27T18:15:17.023780Z","shell.execute_reply":"2024-05-27T18:15:17.628070Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"class DataPipeline_Depth_2:\n def __init__(self, applprev_path, person_path, credit_bureau_a_2_paths, credit_bureau_b_2_path):\n self.applprev_path = applprev_path\n self.person_path = person_path\n self.credit_bureau_a_2_paths = credit_bureau_a_2_paths\n self.credit_bureau_b_2_path = credit_bureau_b_2_path\n\n def load_data(self):\n try:\n self.train_applprev_2 = pl.read_parquet(self.applprev_path).lazy()\n self.train_person_2 = pl.read_parquet(self.person_path).lazy()\n self.credit_bureau_b_2 = pl.read_parquet(self.credit_bureau_b_2_path).lazy()\n except Exception as e:\n print(f\"Error loading data: {e}\")\n raise\n\n def preprocess_applprev(self, data):\n data = data.group_by([\"case_id\", \"num_group1\"]).agg([\n pl.col(\"conts_type_509L\").n_unique().alias(\"unique_contact_types\"),\n pl.col(\"cacccardblochreas_147M\").first().alias(\"first_cacccardblochreas_147M\"),\n pl.col(\"credacc_cards_status_52L\").first().alias(\"first_credacc_cards_status_52L\")\n ]).with_columns(\n pl.col('first_credacc_cards_status_52L')\n .fill_null('UNKNOWN')\n .alias('status')\n ).with_columns([\n (pl.col('status') == 'ACTIVE').cast(pl.Int32).alias('is_active'),\n (pl.col('status') == 'CANCELLED').cast(pl.Int32).alias('is_cancelled')\n ]).group_by('case_id').agg([\n pl.col('unique_contact_types').max().alias('max_unique_contact_type'),\n pl.col('first_cacccardblochreas_147M').n_unique().alias('n_unique_cacccardblochreas_147M'),\n pl.sum('is_cancelled').alias('total_cancelled'),\n pl.sum('is_active').alias('total_active')\n ])\n return data.collect().write_parquet('applprev_2_processed.parquet')\n\n def preprocess_person(self, data):\n data = data.group_by('case_id').agg([\n pl.col('addres_district_368M').n_unique().alias('n_unique_addres_district_368M'),\n pl.col('addres_role_871L').n_unique().alias('n_unique_addres_role_871L'),\n pl.col('addres_zip_823M').n_unique().alias('n_unique_addres_zip_823M'),\n pl.col('conts_role_79M').n_unique().alias('n_unique_conts_role_79M'),\n pl.col('empls_economicalst_849M').n_unique().alias('n_unique_empls_economicalst_849M'),\n pl.col('empls_employedfrom_796D').n_unique().alias('n_unique_empls_employedfrom_796D'),\n pl.col('empls_employer_name_740M').n_unique().alias('n_unique_empls_employer_name_740M'),\n pl.col('relatedpersons_role_762T').n_unique().alias('n_unique_relatedpersons_role_762T')\n ])\n return data.collect().write_parquet('person_2_processed.parquet')\n \n def preprocess_bureau_a_file(self, path):\n data = pl.read_parquet(path).lazy()\n detailed_data = data.group_by([\"case_id\", \"num_group1\"]).agg([\n pl.col(\"collater_valueofguarantee_1124L\").sum().alias(\"sum_collater_valueofguarantee_1124L\"),\n pl.col(\"collater_valueofguarantee_876L\").sum().alias(\"sum_collater_valueofguarantee_876L\"),\n pl.col(\"pmts_overdue_1140A\").sum().alias(\"sum_pmts_overdue_1140A\"),\n pl.col(\"pmts_overdue_1152A\").sum().alias(\"sum_pmts_overdue_1152A\"),\n pl.col(\"pmts_dpd_1073P\").mean().alias(\"avg_pmts_dpd_1073P\"),\n pl.col(\"pmts_dpd_303P\").mean().alias(\"avg_pmts_dpd_303P\"),\n pl.col(\"collater_typofvalofguarant_298M\").n_unique().alias(\"n_unique_collater_typofvalofguarant_298M\"),\n pl.col(\"collater_typofvalofguarant_407M\").n_unique().alias(\"n_unique_collater_typofvalofguarant_407M\"),\n pl.col(\"collaterals_typeofguarante_359M\").n_unique().alias(\"n_unique_collaterals_typeofguarante_359M\"),\n pl.col(\"collaterals_typeofguarante_669M\").n_unique().alias(\"n_unique_collaterals_typeofguarante_669M\"),\n pl.col(\"subjectroles_name_541M\").n_unique().alias(\"n_unique_subjectroles_name_541M\"),\n pl.col(\"subjectroles_name_838M\").n_unique().alias(\"n_unique_subjectroles_name_838M\"),\n pl.col(\"pmts_month_158T\").n_unique().alias(\"n_unique_pmts_month_158T\"),\n pl.col(\"pmts_month_706T\").n_unique().alias(\"n_unique_pmts_month_706T\"),\n pl.col(\"pmts_year_1139T\").n_unique().alias(\"n_unique_pmts_year_1139T\"),\n pl.col(\"pmts_year_507T\").n_unique().alias(\"n_unique_pmts_year_507T\")\n ])\n return detailed_data.collect()\n\n def preprocess_bureau_a(self, paths):\n detailed_data_list = [self.preprocess_bureau_a_file(path) for path in paths]\n combined_detailed_data = pl.concat(detailed_data_list).lazy()\n data = combined_detailed_data.group_by(\"case_id\").agg([\n pl.sum(\"sum_collater_valueofguarantee_1124L\").alias(\"total_sum_collater_valueofguarantee_1124L\"),\n pl.sum(\"sum_collater_valueofguarantee_876L\").alias(\"total_sum_collater_valueofguarantee_876L\"),\n pl.sum(\"sum_pmts_overdue_1140A\").alias(\"total_sum_pmts_overdue_1140A\"),\n pl.sum(\"sum_pmts_overdue_1152A\").alias(\"total_sum_pmts_overdue_1152A\"),\n pl.mean(\"avg_pmts_dpd_1073P\").alias(\"overall_avg_pmts_dpd_1073P\"),\n pl.mean(\"avg_pmts_dpd_303P\").alias(\"overall_avg_pmts_dpd_303P\"),\n pl.max(\"n_unique_collater_typofvalofguarant_298M\").alias(\"max_n_unique_collater_typofvalofguarant_298M\"),\n pl.max(\"n_unique_collater_typofvalofguarant_407M\").alias(\"max_n_unique_collater_typofvalofguarant_407M\"),\n pl.max(\"n_unique_collaterals_typeofguarante_359M\").alias(\"max_n_unique_collaterals_typeofguarante_359M\"),\n pl.max(\"n_unique_collaterals_typeofguarante_669M\").alias(\"max_n_unique_collaterals_typeofguarante_669M\"),\n pl.max(\"n_unique_subjectroles_name_541M\").alias(\"max_n_unique_subjectroles_name_541M\"),\n pl.max(\"n_unique_subjectroles_name_838M\").alias(\"max_n_unique_subjectroles_name_838M\"),\n pl.max(\"n_unique_pmts_month_158T\").alias(\"max_n_unique_pmts_month_158T\"),\n pl.max(\"n_unique_pmts_month_706T\").alias(\"max_n_unique_pmts_month_706T\"),\n pl.max(\"n_unique_pmts_year_1139T\").alias(\"max_n_unique_pmts_year_1139T\"),\n pl.max(\"n_unique_pmts_year_507T\").alias(\"max_n_unique_pmts_year_507T\")\n ])\n return data.collect().write_parquet('credit_bureau_a_2_processed.parquet')\n\n def preprocess_bureau_b(self, data):\n detailed_data_2 = data.group_by([\"case_id\", \"num_group1\"]).agg([\n pl.col(\"pmts_date_1107D\").n_unique().alias(\"unique_pmts_date_1107D\"),\n pl.col(\"pmts_dpdvalue_108P\").sum().alias(\"sum_pmts_dpdvalue_108P\"),\n pl.col(\"pmts_pmtsoverdue_635A\").sum().alias(\"sum_pmts_pmtsoverdue_635A\")\n ])\n data = detailed_data_2.group_by(\"case_id\").agg([\n pl.col(\"unique_pmts_date_1107D\").mean().alias(\"avg_unique_pmts_date\"),\n pl.col(\"sum_pmts_dpdvalue_108P\").max().alias(\"max_sum_pmts_dpdvalue\"),\n pl.col(\"sum_pmts_pmtsoverdue_635A\").sum().alias(\"total_sum_pmts_overdue\")\n ])\n return data.collect().write_parquet('credit_bureau_b_2_processed.parquet')\n \n def merge_data(self):\n applprev_data = pl.read_parquet('applprev_2_processed.parquet')\n person_data = pl.read_parquet('person_2_processed.parquet')\n bureau_a_data = pl.read_parquet('credit_bureau_a_2_processed.parquet')\n bureau_b_data = pl.read_parquet('credit_bureau_b_2_processed.parquet')\n\n merged_data = applprev_data.join(person_data, on='case_id', how='left')\n merged_data = merged_data.join(bureau_a_data, on='case_id', how='left')\n merged_data = merged_data.join(bureau_b_data, on='case_id', how='left')\n return merged_data\n\n def execute_pipeline(self):\n self.load_data()\n self.preprocess_applprev(self.train_applprev_2)\n self.preprocess_person(self.train_person_2)\n self.preprocess_bureau_a(self.credit_bureau_a_2_paths)\n self.preprocess_bureau_b(self.credit_bureau_b_2)\n depth_2 = self.merge_data()\n\n # Clean up memory by deleting intermediate dataframes\n del self.train_applprev_2, self.train_person_2, self.credit_bureau_b_2\n gc.collect()\n\n return depth_2\n\nif __name__ == \"__main__\":\n dataPath = \"/kaggle/input/home-credit-credit-risk-model-stability/\"\n applprev_path = dataPath + \"parquet_files/train/train_applprev_2.parquet\"\n person_path = dataPath + \"parquet_files/train/train_person_2.parquet\"\n credit_bureau_a_2_paths = [\n dataPath + \"parquet_files/train/train_credit_bureau_a_2_0.parquet\",\n dataPath + \"parquet_files/train/train_credit_bureau_a_2_1.parquet\",\n dataPath + \"parquet_files/train/train_credit_bureau_a_2_2.parquet\",\n dataPath + \"parquet_files/train/train_credit_bureau_a_2_3.parquet\",\n dataPath + \"parquet_files/train/train_credit_bureau_a_2_4.parquet\",\n dataPath + \"parquet_files/train/train_credit_bureau_a_2_5.parquet\",\n dataPath + \"parquet_files/train/train_credit_bureau_a_2_6.parquet\",\n dataPath + \"parquet_files/train/train_credit_bureau_a_2_7.parquet\",\n dataPath + \"parquet_files/train/train_credit_bureau_a_2_8.parquet\",\n dataPath + \"parquet_files/train/train_credit_bureau_a_2_9.parquet\",\n dataPath + \"parquet_files/train/train_credit_bureau_a_2_10.parquet\"\n ]\n credit_bureau_b_2_path = dataPath + \"parquet_files/train/train_credit_bureau_b_2.parquet\"\n pipeline = DataPipeline_Depth_2(applprev_path, person_path, credit_bureau_a_2_paths, credit_bureau_b_2_path)\n Depth_2 = pipeline.execute_pipeline()\n\n # Clean up memory after executing the pipeline\n del pipeline, dataPath, applprev_path, person_path, credit_bureau_a_2_paths, credit_bureau_b_2_path\n gc.collect()","metadata":{"execution":{"iopub.status.busy":"2024-05-27T18:15:17.632873Z","iopub.execute_input":"2024-05-27T18:15:17.633581Z","iopub.status.idle":"2024-05-27T18:17:32.206080Z","shell.execute_reply.started":"2024-05-27T18:15:17.633537Z","shell.execute_reply":"2024-05-27T18:17:32.205120Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"class DataPipeline_Depth_2:\n def __init__(self, applprev_path, person_path, credit_bureau_a_2_paths, credit_bureau_b_2_path):\n self.applprev_path = applprev_path\n self.person_path = person_path\n self.credit_bureau_a_2_paths = credit_bureau_a_2_paths\n self.credit_bureau_b_2_path = credit_bureau_b_2_path\n\n def load_data(self):\n try:\n self.test_applprev_2 = pl.scan_parquet(self.applprev_path)\n self.test_person_2 = pl.scan_parquet(self.person_path)\n self.credit_bureau_a_2 = pl.concat([pl.scan_parquet(path) for path in self.credit_bureau_a_2_paths])\n self.credit_bureau_b_2 = pl.scan_parquet(self.credit_bureau_b_2_path)\n except Exception as e:\n print(f\"Error loading data: {e}\")\n raise\n\n def preprocess_applprev(self, data):\n data = data.group_by([\"case_id\", \"num_group1\"]).agg([\n pl.col(\"conts_type_509L\").n_unique().alias(\"unique_contact_types\"),\n pl.col(\"cacccardblochreas_147M\").first().alias(\"first_cacccardblochreas_147M\"),\n pl.col(\"credacc_cards_status_52L\").first().alias(\"first_credacc_cards_status_52L\")\n ]).with_columns(\n pl.col('first_credacc_cards_status_52L')\n .fill_null('UNKNOWN')\n .alias('status')\n ).with_columns([\n (pl.col('status') == 'ACTIVE').cast(pl.Int32).alias('is_active'),\n (pl.col('status') == 'CANCELLED').cast(pl.Int32).alias('is_cancelled')\n ]).group_by('case_id').agg([\n pl.col('unique_contact_types').max().alias('max_unique_contact_type'),\n pl.col('first_cacccardblochreas_147M').n_unique().alias('n_unique_cacccardblochreas_147M'),\n pl.sum('is_cancelled').alias('total_cancelled'),\n pl.sum('is_active').alias('total_active')\n ])\n return data.collect().write_parquet('test_applprev_2_processed.parquet')\n\n def preprocess_person(self, data):\n data = data.group_by('case_id').agg([\n pl.col('addres_district_368M').n_unique().alias('n_unique_addres_district_368M'),\n pl.col('addres_role_871L').n_unique().alias('n_unique_addres_role_871L'),\n pl.col('addres_zip_823M').n_unique().alias('n_unique_addres_zip_823M'),\n pl.col('conts_role_79M').n_unique().alias('n_unique_conts_role_79M'),\n pl.col('empls_economicalst_849M').n_unique().alias('n_unique_empls_economicalst_849M'),\n pl.col('empls_employedfrom_796D').n_unique().alias('n_unique_empls_employedfrom_796D'),\n pl.col('empls_employer_name_740M').n_unique().alias('n_unique_empls_employer_name_740M'),\n pl.col('relatedpersons_role_762T').n_unique().alias('n_unique_relatedpersons_role_762T')\n ])\n return data.collect().write_parquet('test_person_2_processed.parquet')\n \n def preprocess_bureau_a(self, data):\n detailed_data = data.group_by([\"case_id\", \"num_group1\"]).agg([\n pl.col(\"collater_valueofguarantee_1124L\").sum().alias(\"sum_collater_valueofguarantee_1124L\"),\n pl.col(\"collater_valueofguarantee_876L\").sum().alias(\"sum_collater_valueofguarantee_876L\"),\n pl.col(\"pmts_overdue_1140A\").sum().alias(\"sum_pmts_overdue_1140A\"),\n pl.col(\"pmts_overdue_1152A\").sum().alias(\"sum_pmts_overdue_1152A\"),\n pl.col(\"pmts_dpd_1073P\").mean().alias(\"avg_pmts_dpd_1073P\"),\n pl.col(\"pmts_dpd_303P\").mean().alias(\"avg_pmts_dpd_303P\"),\n pl.col(\"collater_typofvalofguarant_298M\").n_unique().alias(\"n_unique_collater_typofvalofguarant_298M\"),\n pl.col(\"collater_typofvalofguarant_407M\").n_unique().alias(\"n_unique_collater_typofvalofguarant_407M\"),\n pl.col(\"collaterals_typeofguarante_359M\").n_unique().alias(\"n_unique_collaterals_typeofguarante_359M\"),\n pl.col(\"collaterals_typeofguarante_669M\").n_unique().alias(\"n_unique_collaterals_typeofguarante_669M\"),\n pl.col(\"subjectroles_name_541M\").n_unique().alias(\"n_unique_subjectroles_name_541M\"),\n pl.col(\"subjectroles_name_838M\").n_unique().alias(\"n_unique_subjectroles_name_838M\"),\n pl.col(\"pmts_month_158T\").n_unique().alias(\"n_unique_pmts_month_158T\"),\n pl.col(\"pmts_month_706T\").n_unique().alias(\"n_unique_pmts_month_706T\"),\n pl.col(\"pmts_year_1139T\").n_unique().alias(\"n_unique_pmts_year_1139T\"),\n pl.col(\"pmts_year_507T\").n_unique().alias(\"n_unique_pmts_year_507T\")\n ])\n data = detailed_data.group_by(\"case_id\").agg([\n pl.sum(\"sum_collater_valueofguarantee_1124L\").alias(\"total_sum_collater_valueofguarantee_1124L\"),\n pl.sum(\"sum_collater_valueofguarantee_876L\").alias(\"total_sum_collater_valueofguarantee_876L\"),\n pl.sum(\"sum_pmts_overdue_1140A\").alias(\"total_sum_pmts_overdue_1140A\"),\n pl.sum(\"sum_pmts_overdue_1152A\").alias(\"total_sum_pmts_overdue_1152A\"),\n pl.mean(\"avg_pmts_dpd_1073P\").alias(\"overall_avg_pmts_dpd_1073P\"),\n pl.mean(\"avg_pmts_dpd_303P\").alias(\"overall_avg_pmts_dpd_303P\"),\n pl.max(\"n_unique_collater_typofvalofguarant_298M\").alias(\"max_n_unique_collater_typofvalofguarant_298M\"),\n pl.max(\"n_unique_collater_typofvalofguarant_407M\").alias(\"max_n_unique_collater_typofvalofguarant_407M\"),\n pl.max(\"n_unique_collaterals_typeofguarante_359M\").alias(\"max_n_unique_collaterals_typeofguarante_359M\"),\n pl.max(\"n_unique_collaterals_typeofguarante_669M\").alias(\"max_n_unique_collaterals_typeofguarante_669M\"),\n pl.max(\"n_unique_subjectroles_name_541M\").alias(\"max_n_unique_subjectroles_name_541M\"),\n pl.max(\"n_unique_subjectroles_name_838M\").alias(\"max_n_unique_subjectroles_name_838M\"),\n pl.max(\"n_unique_pmts_month_158T\").alias(\"max_n_unique_pmts_month_158T\"),\n pl.max(\"n_unique_pmts_month_706T\").alias(\"max_n_unique_pmts_month_706T\"),\n pl.max(\"n_unique_pmts_year_1139T\").alias(\"max_n_unique_pmts_year_1139T\"),\n pl.max(\"n_unique_pmts_year_507T\").alias(\"max_n_unique_pmts_year_507T\")\n ])\n return data.collect().write_parquet('test_credit_bureau_a_2_processed.parquet')\n\n def preprocess_bureau_b(self, data):\n detailed_data_2 = data.group_by([\"case_id\", \"num_group1\"]).agg([\n pl.col(\"pmts_date_1107D\").n_unique().alias(\"unique_pmts_date_1107D\"),\n pl.col(\"pmts_dpdvalue_108P\").sum().alias(\"sum_pmts_dpdvalue_108P\"),\n pl.col(\"pmts_pmtsoverdue_635A\").sum().alias(\"sum_pmts_pmtsoverdue_635A\")\n ])\n data = detailed_data_2.group_by(\"case_id\").agg([\n pl.col(\"unique_pmts_date_1107D\").mean().alias(\"avg_unique_pmts_date\"),\n pl.col(\"sum_pmts_dpdvalue_108P\").max().alias(\"max_sum_pmts_dpdvalue\"),\n pl.col(\"sum_pmts_pmtsoverdue_635A\").sum().alias(\"total_sum_pmts_overdue\")\n ])\n return data.collect().write_parquet('test_credit_bureau_b_2_processed.parquet')\n \n def merge_data(self):\n applprev_data = pl.read_parquet('test_applprev_2_processed.parquet')\n person_data = pl.read_parquet('test_person_2_processed.parquet')\n bureau_a_data = pl.read_parquet('test_credit_bureau_a_2_processed.parquet')\n bureau_b_data = pl.read_parquet('test_credit_bureau_b_2_processed.parquet')\n\n merged_data = applprev_data.join(person_data, on='case_id', how='left')\n merged_data = merged_data.join(bureau_a_data, on='case_id', how='left')\n merged_data = merged_data.join(bureau_b_data, on='case_id', how='left')\n return merged_data\n\n def execute_pipeline(self):\n self.load_data()\n self.preprocess_applprev(self.test_applprev_2)\n self.preprocess_person(self.test_person_2)\n self.preprocess_bureau_a(self.credit_bureau_a_2)\n self.preprocess_bureau_b(self.credit_bureau_b_2)\n depth_2 = self.merge_data()\n\n # Clean up memory by deleting intermediate dataframes\n del self.test_applprev_2, self.test_person_2, self.credit_bureau_a_2, self.credit_bureau_b_2\n gc.collect()\n\n return depth_2\n\nif __name__ == \"__main__\":\n dataPath = \"/kaggle/input/home-credit-credit-risk-model-stability/\"\n applprev_path = dataPath + \"parquet_files/test/test_applprev_2.parquet\"\n person_path = dataPath + \"parquet_files/test/test_person_2.parquet\"\n credit_bureau_a_2_paths = [\n dataPath + \"parquet_files/test/test_credit_bureau_a_2_0.parquet\",\n dataPath + \"parquet_files/test/test_credit_bureau_a_2_1.parquet\",\n dataPath + \"parquet_files/test/test_credit_bureau_a_2_2.parquet\",\n dataPath + \"parquet_files/test/test_credit_bureau_a_2_3.parquet\",\n dataPath + \"parquet_files/test/test_credit_bureau_a_2_4.parquet\",\n dataPath + \"parquet_files/test/test_credit_bureau_a_2_5.parquet\",\n dataPath + \"parquet_files/test/test_credit_bureau_a_2_6.parquet\",\n dataPath + \"parquet_files/test/test_credit_bureau_a_2_7.parquet\",\n dataPath + \"parquet_files/test/test_credit_bureau_a_2_8.parquet\",\n dataPath + \"parquet_files/test/test_credit_bureau_a_2_9.parquet\",\n dataPath + \"parquet_files/test/test_credit_bureau_a_2_10.parquet\",\n dataPath + \"parquet_files/test/test_credit_bureau_a_2_11.parquet\"\n ]\n credit_bureau_b_2_path = dataPath + \"parquet_files/test/test_credit_bureau_b_2.parquet\"\n pipeline = DataPipeline_Depth_2(applprev_path, person_path, credit_bureau_a_2_paths, credit_bureau_b_2_path)\n Depth_2_test = pipeline.execute_pipeline()\n\n # Clean up memory after executing the pipeline\n del pipeline, dataPath, applprev_path, person_path, credit_bureau_a_2_paths, credit_bureau_b_2_path\n gc.collect()","metadata":{"execution":{"iopub.status.busy":"2024-05-27T18:17:32.207903Z","iopub.execute_input":"2024-05-27T18:17:32.208669Z","iopub.status.idle":"2024-05-27T18:17:32.554626Z","shell.execute_reply.started":"2024-05-27T18:17:32.208598Z","shell.execute_reply":"2024-05-27T18:17:32.553562Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Ensure the 'case_id' columns have the same data type in all DataFrames\nDepth_0 = Depth_0.with_columns(pl.col('case_id').cast(pl.Int32))\nDepth_1 = Depth_1.with_columns(pl.col('case_id').cast(pl.Int32))\nDepth_2 = Depth_2.with_columns(pl.col('case_id').cast(pl.Int32))\n\nDepth_0_test = Depth_0_test.with_columns(pl.col('case_id').cast(pl.Int32))\nDepth_1_test = Depth_1_test.with_columns(pl.col('case_id').cast(pl.Int32))\nDepth_2_test = Depth_2_test.with_columns(pl.col('case_id').cast(pl.Int32))\n\n# Join the dataframes for training\nInternal_Final = Depth_0.join(Depth_1, on='case_id', how='left')\nInternal_Final = Internal_Final.join(Depth_2, on='case_id', how='left')\n\n# Convert the final Polars DataFrame to Pandas DataFrame for compatibility with LightGBM\ndf = Internal_Final.to_pandas()\n\n# Prepare the data for training\nX = df.drop(['target', 'case_id'], axis=1)\ny = df['target']\n\nX_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)\ntrain_data = lgb.Dataset(X_train, label=y_train)\nvalid_data = lgb.Dataset(X_test, label=y_test)\n\n# LightGBM parameters\nparams = {\n 'boosting_type': 'gbdt',\n 'objective': 'binary',\n 'metric': 'auc',\n 'num_leaves': 25,\n 'min_data_in_leaf': 100,\n 'learning_rate': 0.1,\n 'feature_fraction': 0.8,\n 'bagging_fraction': 0.7,\n 'bagging_freq': 10,\n 'verbose': 1,\n 'n_jobs': -1,\n 'reg_alpha': 0.1,\n 'reg_lambda': 10,\n 'random_state': 42\n}\n\nnum_round = 100\ncallbacks = [lgb.early_stopping(stopping_rounds=10, verbose=True)]\n\n# Train the model\nbst = lgb.train(params,\n train_data,\n num_boost_round=num_round,\n valid_sets=[valid_data],\n callbacks=callbacks)\n\n# Save the model\nbst.save_model('lgbm_model.txt')","metadata":{"execution":{"iopub.status.busy":"2024-05-27T18:17:32.556272Z","iopub.execute_input":"2024-05-27T18:17:32.556954Z","iopub.status.idle":"2024-05-27T18:20:01.516143Z","shell.execute_reply.started":"2024-05-27T18:17:32.556917Z","shell.execute_reply":"2024-05-27T18:20:01.514779Z"},"trusted":true},"execution_count":null,"outputs":[]}]}