diff --git a/batch-ai-systems/credit_scores/1_credit_scores_feature_backfill.ipynb b/batch-ai-systems/credit_scores/1_credit_scores_feature_backfill.ipynb index d6db1e10..1669b49d 100644 --- a/batch-ai-systems/credit_scores/1_credit_scores_feature_backfill.ipynb +++ b/batch-ai-systems/credit_scores/1_credit_scores_feature_backfill.ipynb @@ -503,11 +503,11 @@ "previous_applications_fg = fs.get_or_create_feature_group(\n", " name='previous_applications',\n", " version=1,\n", - " primary_key=['sk_id_prev','sk_id_curr'],\n", + " primary_key=['sk_id_curr'],\n", " online_enabled=False,\n", " event_time='datetime',\n", ")\n", - "previous_applications_fg.insert(previous_applications_df)" + "previous_applications_fg.insert(collapse_to_customer(previous_applications_df))" ] }, { @@ -528,11 +528,11 @@ "pos_cash_balances_fg = fs.get_or_create_feature_group(\n", " name='pos_cash_balances',\n", " version=1,\n", - " primary_key=['sk_id_prev','sk_id_curr'],\n", + " primary_key=['sk_id_curr'],\n", " online_enabled=False,\n", ")\n", "pos_cash_balances_fg.insert(\n", - " pos_cash_balances_df,\n", + " collapse_to_customer(pos_cash_balances_df),\n", " write_options={\"wait_for_job\": True},\n", ")" ] @@ -555,11 +555,11 @@ "installment_payments_fg = fs.get_or_create_feature_group(\n", " name='installment_payments',\n", " version=1,\n", - " primary_key=['sk_id_prev','sk_id_curr'],\n", + " primary_key=['sk_id_curr'],\n", " online_enabled=False,\n", " event_time='datetime',\n", ")\n", - "installment_payments_fg.insert(installment_payments_df)" + "installment_payments_fg.insert(collapse_to_customer(installment_payments_df))" ] }, { @@ -580,10 +580,10 @@ "credit_card_balances_fg = fs.get_or_create_feature_group(\n", " name='credit_card_balances',\n", " version=1,\n", - " primary_key=['sk_id_prev','sk_id_curr'],\n", + " primary_key=['sk_id_curr'],\n", " online_enabled=False,\n", ")\n", - "credit_card_balances_fg.insert(credit_card_balances_df)" + "credit_card_balances_fg.insert(collapse_to_customer(credit_card_balances_df))" ] }, { diff --git a/batch-ai-systems/credit_scores/3_credit_scores_training_pipeline.ipynb b/batch-ai-systems/credit_scores/3_credit_scores_training_pipeline.ipynb index b1ba4d5d..8bc5f878 100644 --- a/batch-ai-systems/credit_scores/3_credit_scores_training_pipeline.ipynb +++ b/batch-ai-systems/credit_scores/3_credit_scores_training_pipeline.ipynb @@ -225,12 +225,12 @@ " 'amt_credit', 'weekday_appr_process_start',\n", " 'hour_appr_process_start']))\\\n", " .join(bureau_balances_fg.select_except(['sk_id_bureau','months_balance']))\\\n", - " .join(previous_applications_fg.select_except(['sk_id_prev', 'sk_id_curr','datetime',\n", + " .join(previous_applications_fg.select_except(['sk_id_curr','datetime',\n", " 'name_contract_type', 'name_contract_status']))\\\n", - " .join(pos_cash_balances_fg.select_except(['sk_id_prev','sk_id_curr', 'months_balance',\n", + " .join(pos_cash_balances_fg.select_except(['sk_id_curr', 'months_balance',\n", " 'name_contract_status', 'sk_dpd', 'sk_dpd_def']))\\\n", - " .join(installment_payments_fg.select_except(['sk_id_prev', 'sk_id_curr', 'datetime']))\\\n", - " .join(credit_card_balances_fg.select_except(['sk_id_prev', 'sk_id_curr']))\\\n", + " .join(installment_payments_fg.select_except(['sk_id_curr', 'datetime']))\\\n", + " .join(credit_card_balances_fg.select_except(['sk_id_curr']))\\\n", " .join(previous_loan_counts_fg.select_except('sk_id_curr'))\n", "\n", "selected_features_show5 = selected_features.show(5)\n", @@ -387,7 +387,14 @@ "source": [ "X_train, X_test, y_train, y_test = feature_view.train_test_split(\n", " test_size=0.2,\n", - ")" + ")", + "\n", + "# Drop rows with a missing label: bureau records for applicants not present in\n", + "# the labeled `applications` feature group yield a null `target` after the join.\n", + "train_valid = y_train[\"target\"].notna().to_numpy()\n", + "X_train, y_train = X_train[train_valid], y_train[train_valid]\n", + "test_valid = y_test[\"target\"].notna().to_numpy()\n", + "X_test, y_test = X_test[test_valid], y_test[test_valid]" ] }, { @@ -593,7 +600,7 @@ " name=\"credit_scores_model\",\n", " metrics={\"f1_score\": score}, \n", " description=\"XGB for Credit Scores Project\",\n", - " input_example=X_train.sample(),\n", + " input_example=X_train.dropna().sample(),\n", " feature_view=feature_view,\n", ")\n", "\n", diff --git a/batch-ai-systems/credit_scores/functions.py b/batch-ai-systems/credit_scores/functions.py index 199679cc..f7d298bf 100644 --- a/batch-ai-systems/credit_scores/functions.py +++ b/batch-ai-systems/credit_scores/functions.py @@ -2,6 +2,35 @@ import numpy as np import matplotlib.axes._axes as axes +def collapse_to_customer(df: pd.DataFrame) -> pd.DataFrame: + ''' + Collapse a feature group that has one row per previous credit + (grain sk_id_prev, sk_id_curr) down to one row per applicant (sk_id_curr), + keeping the most recent record per applicant. + + Hopsworks 5.0 requires a feature view join to cover every column of the + joined feature group's primary key. The previous_* tables are at + (sk_id_prev, sk_id_curr) grain while the training set is at sk_id_curr + grain, so they are reduced to sk_id_curr grain here before joining. + + Args: + ----- + df: pd.DataFrame + DataFrame at (sk_id_prev, sk_id_curr) grain. + + Returns: + -------- + pd.DataFrame + One row per sk_id_curr (most recent by datetime), without sk_id_prev. + ''' + if 'datetime' in df.columns: + df = df.sort_values('datetime') + df = df.groupby('sk_id_curr', as_index=False).last() + if 'sk_id_prev' in df.columns: + df = df.drop(columns=['sk_id_prev']) + return df + + def remove_nans(df: pd.DataFrame) -> pd.DataFrame: ''' Function which removes missing values. diff --git a/batch-ai-systems/fraud_batch/3_fraud_batch_inference.ipynb b/batch-ai-systems/fraud_batch/3_fraud_batch_inference.ipynb index fc31d386..f3e6a8f9 100644 --- a/batch-ai-systems/fraud_batch/3_fraud_batch_inference.ipynb +++ b/batch-ai-systems/fraud_batch/3_fraud_batch_inference.ipynb @@ -168,7 +168,7 @@ "outputs": [], "source": [ "# log both transformed and untransformed features\n", - "feature_view.log(untransformed_batch_data.head(1000), predictions[:1000], training_dataset_version=1, model=retrieved_model)\n", + "feature_view.log(untransformed_features=untransformed_batch_data.head(1000), predictions=predictions[:1000], training_dataset_version=1, model=retrieved_model)\n", "feature_view.log(transformed_features=transformed_batch_data.head(1000), predictions=predictions[:1000], training_dataset_version=1, model=retrieved_model)" ] }, diff --git a/batch-ai-systems/hospital_wait_time/2_training_pipeline.ipynb b/batch-ai-systems/hospital_wait_time/2_training_pipeline.ipynb index 9c52242f..2058e0bf 100644 --- a/batch-ai-systems/hospital_wait_time/2_training_pipeline.ipynb +++ b/batch-ai-systems/hospital_wait_time/2_training_pipeline.ipynb @@ -20,7 +20,7 @@ "import datetime\n", "import pandas as pd\n", "import numpy as np\n", - "from matplotlib import pyplot\n", + "import matplotlib.pyplot as plt\n", "\n", "from sklearn.metrics import mean_absolute_error\n", "from prophet import Prophet\n", @@ -269,6 +269,36 @@ " test_end=split_dict['test_end'], \n", " event_time=True,\n", ")\n", + "\n", + "# Hopsworks 5.0 renames transformed columns to \"__\" and\n", + "# prefixes joined event-time columns with \"___\". Restore the\n", + "# plain names this notebook references (the `date` column and the per-feature\n", + "# Prophet regressors).\n", + "import re\n", + "def restore_column_names(df):\n", + " renamed = {}\n", + " for col in df.columns:\n", + " match = re.match(r\"(?:label_encoder|standard_scaler|min_max_scaler|robust_scaler)_(.+)_$\", col)\n", + " if match:\n", + " renamed[col] = match.group(1)\n", + " elif col.endswith(\"patient_info_1_date\"):\n", + " renamed[col] = \"date\"\n", + " df = df.rename(columns=renamed)\n", + " # drop the duplicate event-time columns coming from the joined feature groups\n", + " return df.drop(columns=[c for c in df.columns if c.endswith(\"_date\")], errors=\"ignore\")\n", + "\n", + "X_train = restore_column_names(X_train)\n", + "X_test = restore_column_names(X_test)\n", + "\n", + "\n", + "# Prophet rejects NaN in regressor columns; drop rows with a missing regressor or label.\n", + "_regressors = ['age_at_list_registration', 'cpra', 'hla_a1', 'hla_a2', 'hla_b1', 'hla_b2', 'hla_dr1', 'hla_dr2']\n", + "def _drop_nan_rows(X, y):\n", + " keep = X[_regressors].notna().all(axis=1) & y['duration'].notna()\n", + " return X[keep], y[keep]\n", + "X_train, y_train = _drop_nan_rows(X_train, y_train)\n", + "X_test, y_test = _drop_nan_rows(X_test, y_test)\n", + "\n", "X_train.head(3)" ] }, @@ -372,32 +402,32 @@ { "cell_type": "code", "execution_count": null, - "id": "b580f954-98be-4199-888a-e6bf3d194216", + "id": "ce527621", "metadata": {}, "outputs": [], "source": [ - "# Calculate MAE between expected and predicted values for december\n", - "y_pred = forecast['yhat']\n", - "mae = mean_absolute_error(y_test, y_pred)\n", - "print('⛳️ MAE: %.3f' % mae)\n", + "forecast = model.predict(X_test)\n", "\n", - "metrics = {\n", - " \"mae\": round(mae,2)\n", - "}\n", - "metrics" + "# Summarize the forecast\n", + "print(forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper']].head())" ] }, { "cell_type": "code", "execution_count": null, - "id": "ce527621", + "id": "b580f954-98be-4199-888a-e6bf3d194216", "metadata": {}, "outputs": [], "source": [ - "forecast = model.predict(X_test)\n", + "# Calculate MAE between expected and predicted values for december\n", + "y_pred = forecast['yhat']\n", + "mae = mean_absolute_error(y_test, y_pred)\n", + "print('⛳️ MAE: %.3f' % mae)\n", "\n", - "# Summarize the forecast\n", - "print(forecast[['ds', 'yhat', 'yhat_lower', 'yhat_upper']].head())" + "metrics = {\n", + " \"mae\": round(mae,2)\n", + "}\n", + "metrics" ] }, { @@ -524,7 +554,7 @@ " name=\"waiting_time_forecast_model\", # Name for the model\n", " description=\"Waiting time for a deceased donor kidney transplant forecasting model\", # Description of the model\n", " metrics=metrics, # Metrics used for evaluation\n", - " input_example=X_test.sample(), # Example input data for reference\n", + " input_example=X_test.dropna().sample(), # Example input data for reference\n", " feature_view=feature_view,\n", " \n", ")\n", diff --git a/batch-ai-systems/hospital_wait_time/3_inference_pipeline.ipynb b/batch-ai-systems/hospital_wait_time/3_inference_pipeline.ipynb index 846e34d0..f4b4f67c 100644 --- a/batch-ai-systems/hospital_wait_time/3_inference_pipeline.ipynb +++ b/batch-ai-systems/hospital_wait_time/3_inference_pipeline.ipynb @@ -127,6 +127,25 @@ " event_time=True,\n", ")\n", "\n", + "# Hopsworks 5.0 renames transformed columns to \"__\" and\n", + "# prefixes joined event-time columns with \"___\". Restore the\n", + "# plain names the model expects and drop rows with a missing regressor.\n", + "import re\n", + "def restore_column_names(df):\n", + " renamed = {}\n", + " for col in df.columns:\n", + " match = re.match(r\"(?:label_encoder|standard_scaler|min_max_scaler|robust_scaler)_(.+)_$\", col)\n", + " if match:\n", + " renamed[col] = match.group(1)\n", + " elif col.endswith(\"patient_info_1_date\"):\n", + " renamed[col] = \"date\"\n", + " df = df.rename(columns=renamed)\n", + " return df.drop(columns=[c for c in df.columns if c.endswith(\"_date\")], errors=\"ignore\")\n", + "\n", + "batch_data = restore_column_names(batch_data)\n", + "_regressors = ['age_at_list_registration', 'cpra', 'hla_a1', 'hla_a2', 'hla_b1', 'hla_b2', 'hla_dr1', 'hla_dr2']\n", + "batch_data = batch_data[batch_data[_regressors].notna().all(axis=1)]\n", + "\n", "batch_data['ds'] = batch_data.date\n", "batch_data['ds'] = pd.to_datetime(batch_data.ds)\n", "batch_data['ds'] = batch_data.ds.map(lambda x: x.replace(tzinfo=None))\n",