{"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},{"sourceId":8533726,"sourceType":"datasetVersion","datasetId":5097081}],"dockerImageVersionId":30615,"isInternetEnabled":false,"language":"python","sourceType":"notebook","isGpuEnabled":false}},"nbformat_minor":4,"nbformat":4,"cells":[{"cell_type":"code","source":"import gc\nimport json\nimport logging\nimport pandas as pd\nimport polars as pl\nimport lightgbm as lgb\nimport os\nfrom pandas.api.types import CategoricalDtype\nfrom datetime import datetime\n\ndef convert_to_ordinal(date_column, date_format):\n    date_parsed = date_column.str.strptime(pl.Date, date_format, strict=False)\n    epoch_start = pl.lit(datetime(1970, 1, 1))\n    days_since_epoch = (date_parsed.cast(pl.Date) - epoch_start.cast(pl.Date)).dt.total_days()\n    return days_since_epoch\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        merged_data = self.merge_data(data_base, static_datas, data_static_cb_0)\n        \n        # Clean up temporary files\n        os.remove('test_base_processed.parquet')\n        for i in range(len(self.static_paths)):\n            os.remove(f'test_static_0_{i}_processed.parquet')\n        os.remove('test_static_cb_0_processed.parquet')\n\n        return merged_data\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-30T22:17:09.526942Z","iopub.execute_input":"2024-05-30T22:17:09.527292Z","iopub.status.idle":"2024-05-30T22:17:13.804688Z","shell.execute_reply.started":"2024-05-30T22:17:09.527262Z","shell.execute_reply":"2024-05-30T22:17:13.803548Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Configure logging\nlogging.basicConfig(level=logging.INFO)\nlogger = logging.getLogger(__name__)\n\nclass 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_paths, credit_bureau_b_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_paths = credit_bureau_a_paths\n        self.credit_bureau_b_path = credit_bureau_b_path\n        self.schema_path = schema_path\n\n        # Load the schema\n        with open(schema_path, 'r') as file:\n            self.global_schema = json.load(file)\n\n        self.dtype_mapping = {\n            'Int8': pl.Int8,\n            'Int16': pl.Int16,\n            'Int32': pl.Int32,\n            'Int64': pl.Int64,\n            'UInt8': pl.UInt8,\n            'UInt16': pl.UInt16,\n            'UInt32': pl.UInt32,\n            'UInt64': pl.UInt64,\n            'Float32': pl.Float32,\n            'Float64': pl.Float64,\n            'Utf8': pl.Utf8,\n            'Boolean': pl.Boolean,\n            'Date': pl.Date,\n            'Datetime': pl.Datetime,\n            'String': pl.Utf8,  # Ensure 'String' is mapped correctly\n            'Null': pl.Utf8  # Handle Null type by converting to Utf8 (or another appropriate type)\n        }\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_and_update_schema(self, path):\n        df = pl.read_parquet(path)\n        df = self.optimize_dataframe(df)\n        df = self.ensure_schema(df)\n        return df\n\n    def ensure_schema(self, df):\n        for col, dtype in self.global_schema.items():\n            if col in df.columns:\n                try:\n                    if df[col].dtype != self.dtype_mapping[dtype]:\n                        if self.dtype_mapping[dtype] is not None:\n                            df = df.with_columns(pl.col(col).cast(self.dtype_mapping[dtype]))\n                except KeyError as e:\n                    logger.error(f\"Column: {col}, Expected dtype: {dtype}, Error: {e}\")\n                    raise\n        return df\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 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_paths)\n        self.train_credit_bureau_b_1 = self.load_and_update_schema(self.credit_bureau_b_path).lazy()\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 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    def execute_pipeline(self):\n        logger.info(\"Loading data...\")\n        self.load_data()\n        logger.info(\"Data loaded successfully.\")\n\n        logger.info(\"Preprocessing data...\")\n        self.preprocess_data()\n        logger.info(\"Data preprocessed successfully.\")\n\n        # Clean up memory by deleting intermediate dataframes after preprocessing\n        gc.collect()\n\n        logger.info(\"Merging data...\")\n        merged_data = self.merge_data()\n        logger.info(\"Data merged successfully.\")\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/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_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_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_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_path, \n                                    tax_registry_c_1_path, credit_bureau_a_paths, credit_bureau_b_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-30T22:17:13.806850Z","iopub.execute_input":"2024-05-30T22:17:13.807486Z","iopub.status.idle":"2024-05-30T22:17:14.484332Z","shell.execute_reply.started":"2024-05-30T22:17:13.807425Z","shell.execute_reply":"2024-05-30T22:17:14.483329Z"},"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/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-30T22:17:14.486028Z","iopub.execute_input":"2024-05-30T22:17:14.486441Z","iopub.status.idle":"2024-05-30T22:17:14.813346Z","shell.execute_reply.started":"2024-05-30T22:17:14.486401Z","shell.execute_reply":"2024-05-30T22:17:14.811999Z"},"trusted":true},"execution_count":null,"outputs":[]},{"cell_type":"code","source":"# Load the saved model\nbst = lgb.Booster(model_file='/kaggle/input/lgbm-model/lgbm_model.txt')\n\n# Ensure the 'case_id' columns have the same data type in all DataFrames\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# Set chunk size\nchunk_size = 25000\n\n# Initialize the final dataframe\nchunk_filenames = []\ncase_ids = []\n\n# Process in chunks and save intermediate results to disk\nfor i in range(0, Depth_0_test.height, chunk_size):\n    chunk_0 = Depth_0_test.slice(i, chunk_size)\n    chunk_1 = chunk_0.join(Depth_1_test, on='case_id', how='left')\n    chunk_2 = chunk_1.join(Depth_2_test, on='case_id', how='left')\n    \n    # Save the chunk to disk\n    chunk_filename = f'temp_chunk_{i}.parquet'\n    chunk_2.write_parquet(chunk_filename)\n    chunk_filenames.append(chunk_filename)\n    \n    # Collect case_ids\n    case_ids.extend(chunk_2.select('case_id').to_series().to_list())\n    \n    # Clean up memory after processing each chunk\n    del chunk_0, chunk_1, chunk_2\n    gc.collect()\n\n# Clean up memory by deleting intermediate dataframes\ndel Depth_0_test, Depth_1_test, Depth_2_test\ngc.collect()\n\n# Function to convert string columns to categorical\ndef convert_strings(df: pd.DataFrame) -> pd.DataFrame:\n    for col in df.columns:\n        if df[col].dtype.name in ['object', 'string']:\n            df[col] = df[col].astype('category')\n            if 'Unknown' not in df[col].cat.categories:\n                df[col].cat.add_categories(['Unknown'], inplace=True)\n    return df\n\n# Function to batch process predictions\ndef batch_predict(model, data, batch_size=25000):\n    preds = []\n    for i in range(0, len(data), batch_size):\n        batch_data = data[i:i + batch_size]\n        batch_preds = model.predict(batch_data, num_iteration=model.best_iteration, predict_disable_shape_check=True)\n        preds.extend(batch_preds)\n        gc.collect()\n    return preds\n\n# Initialize the list to store predictions\nall_preds = []\n\n# Process each chunk file for predictions\nfor chunk_filename in chunk_filenames:\n    chunk_data = pl.read_parquet(chunk_filename)\n    \n    # Convert to Pandas for LightGBM compatibility\n    df_test = chunk_data.to_pandas()\n    \n    # Clean up memory by deleting the chunk data\n    del chunk_data\n    gc.collect()\n    \n    # Convert string columns to categorical\n    X_submission = convert_strings(df_test)\n    \n    # Clean up memory\n    del df_test\n    gc.collect()\n\n    categorical_cols = X_submission.select_dtypes(include=['category']).columns\n\n    for col in categorical_cols:\n        train_categories = set(X_train[col].cat.categories)\n        submission_categories = set(X_submission[col].cat.categories)\n        new_categories = submission_categories - train_categories\n        X_submission.loc[X_submission[col].isin(new_categories), col] = \"Unknown\"\n        new_dtype = CategoricalDtype(categories=train_categories, ordered=True)\n        X_submission[col] = X_submission[col].astype(new_dtype)\n\n    # Predict on the submission data in batches\n    chunk_preds = batch_predict(bst, X_submission)\n    all_preds.extend(chunk_preds)\n    \n    # Clean up memory after processing each chunk\n    del X_submission\n    gc.collect()\n\n# Create the submission file\nsubmission = pd.DataFrame({\n    \"case_id\": case_ids,\n    \"score\": all_preds\n}).set_index('case_id')\nsubmission.to_csv(\"./submission.csv\")\n\n# Clean up memory after executing the pipeline\ndel all_preds, case_ids\ngc.collect()\n\n# Delete temporary chunk files\nfor chunk_filename in chunk_filenames:\n    os.remove(chunk_filename)","metadata":{"execution":{"iopub.status.busy":"2024-05-30T22:17:14.815702Z","iopub.execute_input":"2024-05-30T22:17:14.816078Z","iopub.status.idle":"2024-05-30T22:17:15.775927Z","shell.execute_reply.started":"2024-05-30T22:17:14.816043Z","shell.execute_reply":"2024-05-30T22:17:15.774821Z"},"trusted":true},"execution_count":null,"outputs":[]}]}