bqml/04_model_training_and_prediction.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "ur8xi4C7S06n"
   },
   "outputs": [],
   "source": [
    "# Copyright 2022 Google LLC\n",
    "#\n",
    "# Licensed under the Apache License, Version 2.0 (the \"License\");\n",
    "# you may not use this file except in compliance with the License.\n",
    "# You may obtain a copy of the License at\n",
    "#\n",
    "#     https://www.apache.org/licenses/LICENSE-2.0\n",
    "#\n",
    "# Unless required by applicable law or agreed to in writing, software\n",
    "# distributed under the License is distributed on an \"AS IS\" BASIS,\n",
    "# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n",
    "# See the License for the specific language governing permissions and\n",
    "# limitations under the License."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "JAPoU8Sm5E6e"
   },
   "source": [
    "# Fraudfinder - BigQuery ML - Model training and prediction\n",
    "\n",
    "<table align=\"left\">\n",
    "  <td>\n",
    "    <a href=\"https://console.cloud.google.com/ai-platform/notebooks/deploy-notebook?download_url=https://github.com/GoogleCloudPlatform/fraudfinder/raw/main/bqml/04_model_training_and_prediction.ipynb\">\n",
    "       <img src=\"https://www.gstatic.com/cloud/images/navigation/vertex-ai.svg\" alt=\"Google Cloud Notebooks\">Open in Cloud Notebook\n",
    "    </a>\n",
    "  </td> \n",
    "  <td>\n",
    "    <a href=\"https://colab.research.google.com/github/GoogleCloudPlatform/fraudfinder/blob/main/bqml/04_model_training_and_prediction.ipynb\">\n",
    "      <img src=\"https://cloud.google.com/ml-engine/images/colab-logo-32px.png\" alt=\"Colab logo\"> Open in Colab\n",
    "    </a>\n",
    "  </td>\n",
    "  <td>\n",
    "    <a href=\"https://github.com/GoogleCloudPlatform/fraudfinder/blob/main/bqml/04_model_training_and_prediction.ipynb\">\n",
    "        <img src=\"https://cloud.google.com/ml-engine/images/github-logo-32px.png\" alt=\"GitHub logo\">\n",
    "      View on GitHub\n",
    "    </a>\n",
    "  </td>\n",
    "</table>"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "tvgnzT1CKxrO"
   },
   "source": [
    "## Overview\n",
    "\n",
    "[Fraudfinder](https://github.com/googlecloudplatform/fraudfinder) is a series of labs on how to build a real-time fraud detection system on Google Cloud. Throughout the Fraudfinder labs, you will learn how to read historical bank transaction data stored in data warehouse, read from a live stream of new transactions, perform exploratory data analysis (EDA), do feature engineering, ingest features into a feature store, train a model using feature store, register your model in a model registry, evaluate your model, deploy your model to an endpoint, do real-time inference on your model with feature store, and monitor your model."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "66bbec2a5e37"
   },
   "source": [
    "### Objective\n",
    "\n",
    "After feature engineering and ingesting data into a feature store, you are now ready to train and deploy your machine learning model to predict whether a transaction is fraudulent or not.\n",
    "\n",
    "In this notebook, using the data in Vertex AI Feature Store that you previously ingested data into, you will train a model using BigQuery ML, register the model to Vertex AI Model Registry, and deploy it to an endpoint for real-time prediction. \n",
    "\n",
    "In this tutorial, you will learn how to:\n",
    "\n",
    "- Train a logistic regression model in BigQuery using BigQuery ML\n",
    "- Evaluate the model\n",
    "- Test a prediction \n",
    "- Deploy to an endpoint on Vertex AI\n",
    "- Make an online prediction\n",
    "\n",
    "This tutorial uses the following Google Cloud data analytics and services:\n",
    "\n",
    "- [BigQuery](https://cloud.google.com/bigquery/)\n",
    "- [BigQuery ML](https://cloud.google.com/bigquery-ml/)\n",
    "- [Vertex AI](https://cloud.google.com/vertex-ai/)\n",
    "\n",
    "The steps performed include:\n",
    "\n",
    "- Using Python & SQL to query the data in BigQuery\n",
    "- Preparing the data for modeling\n",
    "- Training a classification model using BigQuery ML and registering it to Vertex AI Model Registry\n",
    "- Inspecting the model on Vertex AI Model Registry\n",
    "- Deploying the model to an endpoint on Vertex AI\n",
    "- Making sample online predictions to the model endpoint\n",
    "\n",
    "### Costs \n",
    "\n",
    "This tutorial uses billable components of Google Cloud:\n",
    "\n",
    "* BigQuery\n",
    "* BigQuery ML\n",
    "* Vertex AI\n",
    "\n",
    "Learn about [BigQuery Pricing](https://cloud.google.com/bigquery/pricing), [BigQuery ML pricing](https://cloud.google.com/bigquery-ml/pricing), [Vertex AI pricing](https://cloud.google.com/vertex-ai/pricing), and use the [Pricing Calculator](https://cloud.google.com/products/calculator/) to generate a cost estimate based on your projected usage."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "ze4-nDLfK4pw"
   },
   "source": [
    "### Load configuration settings from the setup notebook\n",
    "\n",
    "Set the constants used in this notebook and load the config settings from the `00_environment_setup.ipynb` notebook."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "gCuSR8GkAgzl"
   },
   "outputs": [],
   "source": [
    "GCP_PROJECTS = !gcloud config get-value project\n",
    "PROJECT_ID = GCP_PROJECTS[0]\n",
    "BUCKET_NAME = f\"{PROJECT_ID}-fraudfinder\"\n",
    "config = !gsutil cat gs://{BUCKET_NAME}/config/notebook_env.py\n",
    "print(config.n)\n",
    "exec(config.n)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "170fc7eb629a"
   },
   "source": [
    "### Import libraries"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "2b4ef9b72d43"
   },
   "outputs": [],
   "source": [
    "import pandas as pd\n",
    "\n",
    "from datetime import datetime, timedelta\n",
    "from typing import Union\n",
    "\n",
    "from google.cloud import aiplatform as vertex_ai\n",
    "from google.cloud import bigquery\n",
    "\n",
    "pd.set_option(\"display.max_columns\", None)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "39705682a1a0"
   },
   "source": [
    "### Define constants"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "3dda2792a08c"
   },
   "outputs": [],
   "source": [
    "START_DATE_TRAIN = (datetime.today() - timedelta(days=1)).strftime(\"%Y-%m-%d\")\n",
    "END_DATE_TRAIN = (datetime.today() - timedelta(days=1)).strftime(\"%Y-%m-%d\")\n",
    "CUSTOMERS_TABLE_NAME = f\"customers_{END_DATE_TRAIN.replace('-', '')}\"\n",
    "TERMINALS_TABLE_NAME = f\"terminals_{END_DATE_TRAIN.replace('-', '')}\"\n",
    "VERTEX_AI_MODEL_ID = 'bqml_fraud_classifier'\n",
    "SERVING_FEATURE_IDS = {\"customer\": [\"*\"], \"terminal\": [\"*\"]}\n",
    "READ_INSTANCES_TABLE = f\"ground_truth_{END_DATE_TRAIN.replace('-', '')}\"\n",
    "READ_INSTANCES_URI = f\"bq://{PROJECT_ID}.tx.{READ_INSTANCES_TABLE}\"\n",
    "BQ_TABLE_NAME = f\"train_table_{END_DATE_TRAIN.replace('-', '')}\"\n",
    "TRAIN_TABLE_URI = f\"bq://{PROJECT_ID}.tx.{BQ_TABLE_NAME}\"\n",
    "\n",
    "print(f\"\"\"\n",
    "START_DATE_TRAIN \\t= {START_DATE_TRAIN}\n",
    "END_DATE_TRAIN \\t\\t= {END_DATE_TRAIN}\n",
    "CUSTOMERS_TABLE_NAME \\t= {CUSTOMERS_TABLE_NAME}\n",
    "TERMINALS_TABLE_NAME \\t= {TERMINALS_TABLE_NAME}\n",
    "VERTEX_AI_MODEL_ID \\t= {VERTEX_AI_MODEL_ID}\n",
    "SERVING_FEATURE_IDS \\t= {SERVING_FEATURE_IDS}\n",
    "READ_INSTANCES_TABLE \\t= {READ_INSTANCES_TABLE}\n",
    "READ_INSTANCES_URI \\t= {READ_INSTANCES_URI}\n",
    "BQ_TABLE_NAME \\t\\t= {BQ_TABLE_NAME}\n",
    "TRAIN_TABLE_URI \\t= {TRAIN_TABLE_URI}\n",
    "\"\"\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "d5d6e49cc474"
   },
   "source": [
    "#### Payload schema"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "The following schema will be used later in this notebook for prediction."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "7fd429262fab"
   },
   "outputs": [],
   "source": [
    "PAYLOAD_SCHEMA = {\n",
    "    \"tx_amount\": \"float64\",\n",
    "    \"customer_id_nb_tx_1day_window\": \"int64\",\n",
    "    \"customer_id_nb_tx_7day_window\": \"int64\",\n",
    "    \"customer_id_nb_tx_14day_window\": \"int64\",\n",
    "    \"customer_id_avg_amount_1day_window\": \"float64\",\n",
    "    \"customer_id_avg_amount_7day_window\": \"float64\",\n",
    "    \"customer_id_avg_amount_14day_window\": \"float64\",\n",
    "    \"customer_id_nb_tx_15min_window\": \"int64\",\n",
    "    \"customer_id_avg_amount_15min_window\": \"float64\",\n",
    "    \"customer_id_nb_tx_30min_window\": \"int64\",\n",
    "    \"customer_id_avg_amount_30min_window\": \"float64\",\n",
    "    \"customer_id_nb_tx_60min_window\": \"int64\",\n",
    "    \"customer_id_avg_amount_60min_window\": \"float64\",\n",
    "    \"terminal_id_nb_tx_1day_window\": \"int64\",\n",
    "    \"terminal_id_nb_tx_7day_window\": \"int64\",\n",
    "    \"terminal_id_nb_tx_14day_window\": \"int64\",\n",
    "    \"terminal_id_risk_1day_window\": \"float64\",\n",
    "    \"terminal_id_risk_7day_window\": \"float64\",\n",
    "    \"terminal_id_risk_14day_window\": \"float64\",\n",
    "    \"terminal_id_nb_tx_15min_window\": \"int64\",\n",
    "    \"terminal_id_avg_amount_15min_window\": \"float64\",\n",
    "    \"terminal_id_nb_tx_30min_window\": \"int64\",\n",
    "    \"terminal_id_avg_amount_30min_window\": \"float64\",\n",
    "    \"terminal_id_nb_tx_60min_window\": \"int64\",\n",
    "    \"terminal_id_avg_amount_60min_window\": \"float64\",\n",
    "}"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "init_aip:mbsdk,all"
   },
   "source": [
    "### Initialize Vertex AI and BigQuery SDKs for Python\n",
    "\n",
    "Initialize the Vertex AI SDK for Python for your project and corresponding bucket."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "init_aip:mbsdk,all"
   },
   "outputs": [],
   "source": [
    "vertex_ai.init(project=PROJECT_ID, location=REGION)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "5faea0ea0937"
   },
   "source": [
    "### Helpers"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "f94734ac9312"
   },
   "source": [
    "Use a helper function for sending queries to BigQuery."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "e364dab1d353"
   },
   "outputs": [],
   "source": [
    "# Wrapper to use BigQuery client to run query/job, return job ID or result as DF\n",
    "def run_bq_query(sql: str) -> Union[str, pd.DataFrame]:\n",
    "    \"\"\"\n",
    "    Run a BigQuery query and return the job ID or result as a DataFrame\n",
    "    Args:\n",
    "        sql: SQL query, as a string, to execute in BigQuery\n",
    "    Returns:\n",
    "        df: DataFrame of results from query,  or error, if any\n",
    "    \"\"\"\n",
    "\n",
    "    bq_client = bigquery.Client()\n",
    "\n",
    "    # Try dry run before executing query to catch any errors\n",
    "    job_config = bigquery.QueryJobConfig(dry_run=True, use_query_cache=False)\n",
    "    bq_client.query(sql, job_config=job_config)\n",
    "\n",
    "    # If dry run succeeds without errors, proceed to run query\n",
    "    job_config = bigquery.QueryJobConfig()\n",
    "    client_result = bq_client.query(sql, job_config=job_config)\n",
    "\n",
    "    job_id = client_result.job_id\n",
    "\n",
    "    # Wait for query/job to finish running. then get & return data frame\n",
    "    df = client_result.result().to_arrow().to_pandas()\n",
    "    print(f\"Finished job_id: {job_id}\")\n",
    "    return df"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "59b4103c0263"
   },
   "source": [
    "## Fetching feature values for model training\n",
    "\n",
    "In the previous labs `02_feature_engineering_batch.ipynb` and `03_feature_engineering_streaming`, you computed features and ingested them into Vertex AI Feature Store. You can now look back into Feature Store to any point in time and output those values from Feature Store in batch to a BigQuery table in order to train your machine learning model.\n",
    "\n",
    "Note that in these Fraudfinder labs, the ground-truth fraud labels are already available for all transactions, even as soon as a new transaction occurs. However, in practice, there is usually a delay (in th order of _days_) between when a transaction occurs until it gets officially flagged as a fraudulent transaction by a fraud department. Because of this, if the data you use to train on doesn't yet have the up-to-date information about whether it's truly fraud or not, you may end up training an ML model on incorrect data (i.e. on transactions that are actually fraud, but were labeled as not-fraud at the time of training). If this is the case, then when you choose what data you use for training a model, be sure you use the point-in-time by \"time-traveling\" an appropriate number of days into the past to a date when you're confident that the labels should be accurate when fetching feature values from Vertex AI Feature Store. For this notebook, you will simply be \"time traveling\" to yesterday for simplicity.\n",
    "\n",
    "To fetch data from Vertex AI Feature Store (via batch serving), you will need to provide the following inputs:\n",
    "\n",
    "- a file containing a \"query\", with the entities and timestamps for each transaction\n",
    "- a list of features to fetch values for\n",
    "- the destination location and format where you want your feature values to be outputted to\n"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "232ca3c3be7e"
   },
   "source": [
    "### Preparing the data to query Vertex AI Feature Store (the \"Read-Instance\")\n",
    "\n",
    "You will first need to prepare data that you want to use to query Vertex AI Feature Store. It will need to be a CSV file formatted like the table below:\n",
    "\n",
    "|customer                     |terminal|timestamp                                    |\n",
    "|-----------------------------|--------|---------------------------------------------|\n",
    "|xxx3859                         |xxx8811    |2021-07-07 00:01:10 UTC                      |\n",
    "|xxx4165                         |xxx8810    |2021-07-07 00:01:55 UTC                      |\n",
    "|xxx2289                         |xxx2081    |2021-07-07 00:02:12 UTC                      |\n",
    "|xxx3227                         |xxx3011    |2021-07-07 00:03:23 UTC                      |\n",
    "|xxx2819                         |xxx6263    |2021-07-07 00:05:30 UTC                      |\n",
    "\n",
    "where the column names are the name of entities in Feature Store and the timestamps represents the time an event occurred. Vertex AI Feature Store will then retrieve the last feature value up to (but not after) those timestamps."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "056bbbc7c579"
   },
   "outputs": [],
   "source": [
    "read_instances_query = f\"\"\"\n",
    "CREATE OR REPLACE TABLE {PROJECT_ID}.tx.{READ_INSTANCES_TABLE} as (\n",
    "    SELECT\n",
    "        raw_tx.TX_TS AS timestamp,\n",
    "        raw_tx.CUSTOMER_ID AS customer,\n",
    "        raw_tx.TERMINAL_ID AS terminal,\n",
    "        raw_tx.TX_AMOUNT AS tx_amount,\n",
    "        raw_lb.TX_FRAUD AS tx_fraud,\n",
    "    FROM \n",
    "        tx.tx as raw_tx\n",
    "    LEFT JOIN \n",
    "        tx.txlabels as raw_lb\n",
    "    ON raw_tx.TX_ID = raw_lb.TX_ID\n",
    "    WHERE\n",
    "        DATE(raw_tx.TX_TS) = \"{START_DATE_TRAIN}\"\n",
    ");\n",
    "\"\"\"\n",
    "print(read_instances_query)\n",
    "\n",
    "run_bq_query(read_instances_query)\n",
    "run_bq_query(f\"SELECT * FROM {PROJECT_ID}.tx.{READ_INSTANCES_TABLE} LIMIT 10\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "9539737e33ec"
   },
   "source": [
    "### Initiate Feature Store\n",
    "\n",
    "Initiate the feature store you created in the `02_feature_engineering_batch.ipynb` notebook."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "a1a98b7681b9"
   },
   "outputs": [],
   "source": [
    "try:\n",
    "    ff_feature_store = vertex_ai.Featurestore(f\"{FEATURESTORE_ID}\")\n",
    "    print(f\"\"\"The feature store {FEATURESTORE_ID} found!\"\"\")\n",
    "except NameError:\n",
    "    print(f\"\"\"The feature store {FEATURESTORE_ID} does not exist!\"\"\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "efb7b335cafd"
   },
   "source": [
    "### Export a sample of data to a BigQuery using Vertex AI Feature Store's point-in-time capability\n",
    "\n",
    "In this section, we will use batch serving of feature store to prepare a dataset for training by calling the `BatchReadFeatureValues` API. Batch serving is used to fetch a large set of feature values with high throughput, typically for training a model or batch prediction. For outputting the results to BigQuery, you will use the [`batch_serve_to_bq` method](https://cloud.google.com/python/docs/reference/aiplatform/latest/google.cloud.aiplatform.Featurestore#google_cloud_aiplatform_Featurestore_batch_serve_to_bq).\n",
    "\n",
    "In the follow cell, based on the table you created above, you will query feature store for all of the relevant feature values, and export the feature values into a new table in BigQuery, which can the be used for model training later on."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "c2a940d39f47"
   },
   "outputs": [],
   "source": [
    "ff_feature_store.batch_serve_to_bq(\n",
    "    bq_destination_output_uri=TRAIN_TABLE_URI,\n",
    "    serving_feature_ids=SERVING_FEATURE_IDS,\n",
    "    read_instances_uri=READ_INSTANCES_URI,\n",
    "    pass_through_fields=[\"tx_amount\", \"tx_fraud\"],\n",
    ")\n",
    "\n",
    "print(f\"Feature values from feature store outputted to: {TRAIN_TABLE_URI}.\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "13b6ce9f8d8b"
   },
   "source": [
    "### Inspect the resulting output table from Vertex AI Feature Store"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "49dd00d5fbe5"
   },
   "source": [
    "Run the query below to inspect the training table. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "1789398c4571"
   },
   "outputs": [],
   "source": [
    "sql_inspect = f\"\"\"\n",
    "SELECT\n",
    "    *\n",
    "FROM\n",
    "    `tx.{BQ_TABLE_NAME}`\n",
    "LIMIT\n",
    "    5\n",
    "\"\"\"\n",
    "run_bq_query(sql_inspect)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "As you can see, each row represents a transaction ID, and the columns represent the attributes of the transaction (i.e `customer_id`, `terminal_id`, `tx_amount`),  aggregated features (for the customer and terminal) and the ground-truth label `tx_fraud` (1 if fraud, else 0)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "a4a686de97f5"
   },
   "source": [
    "## Using BigQuery ML to train a machine learning model directly in BigQuery\n",
    "\n",
    "BigQuery ML (BQML) provides the capability to train machine learning models, such as classification, regression, forecasting, and matrix factorization, in BigQuery using SQL syntax directly. BigQuery ML uses the scalable infrastructure of BigQuery so you don't need to set up additional infrastructure for training or batch serving.\n",
    "\n",
    "Because the data from Vertex AI Feature Store was outputted to a BigQuery table, this makes the data conveniently available for training directly using BigQuery ML.\n",
    "\n",
    "Learn more about [BigQuery ML documentation](https://cloud.google.com/bigquery-ml/docs)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "02f304053600"
   },
   "source": [
    "### Train a logistic regression model using BigQuery ML"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "566f3395f20b"
   },
   "source": [
    "The query below trains a logistic regression model using BigQuery ML. BigQuery resources are used to train the model.\n",
    "\n",
    "In the `OPTIONS` parameter:\n",
    "* with `model_registry=\"vertex_ai\"`, the BigQuery ML model will automatically be [registered to Vertex AI Model Registry](https://cloud.google.com/vertex-ai/docs/model-registry/model-registry-bqml), which enables you to view all of your registered models and its versions on Google Cloud in one place.\n",
    "\n",
    "* `vertex_ai_model_version_aliases` allows you to set aliases to help you keep track of your model version ([documentation](https://cloud.google.com/vertex-ai/docs/model-registry/model-alias))."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "414c45011c1f"
   },
   "outputs": [],
   "source": [
    "# this cell may take ~4 min to run\n",
    "sql_train_model_bqml = f\"\"\"\n",
    "CREATE OR REPLACE MODEL `tx.{MODEL_NAME}` \n",
    "OPTIONS(\n",
    "  MODEL_TYPE=\"LOGISTIC_REG\",\n",
    "  INPUT_LABEL_COLS=[\"tx_fraud\"],\n",
    "  EARLY_STOP=TRUE,\n",
    "  MIN_REL_PROGRESS=0.01,\n",
    "  model_registry=\"vertex_ai\", \n",
    "  vertex_ai_model_id='{VERTEX_AI_MODEL_ID}',\n",
    "  vertex_ai_model_version_aliases=['logit', 'experimental']\n",
    ") AS\n",
    "\n",
    "SELECT\n",
    "  * EXCEPT(timestamp, entity_type_customer, entity_type_terminal)\n",
    "FROM\n",
    "   `tx.{BQ_TABLE_NAME}`\n",
    "\"\"\"\n",
    "\n",
    "print(sql_train_model_bqml)\n",
    "\n",
    "run_bq_query(sql_train_model_bqml)\n",
    "\n",
    "print(f\"Training job finished for: `tx.{MODEL_NAME}`\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "92f26cdecf4d"
   },
   "source": [
    "#### Inspect the model on Vertex AI Model Registry\n",
    "The model should now be automatically registered to Vertex AI Model Registry upon completion.\n",
    "\n",
    "You can view the model on the <a href=\"https://console.cloud.google.com/vertex-ai/models\" target=\"_blank\">Vertex AI Model Registry page</a>, or use the code below to check that it was successfully registered:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Initiate Vertex AI Model Registry for `VERTEX_AI_MODEL_ID` model entry\n",
    "registry = vertex_ai.models.ModelRegistry(VERTEX_AI_MODEL_ID)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Get model versions\n",
    "versions = registry.list_versions()\n",
    "\n",
    "for version in versions:\n",
    "    version_id = version.version_id\n",
    "    version_created_time = datetime.fromtimestamp(\n",
    "        version.version_create_time.timestamp()\n",
    "    ).strftime(\"%m/%d/%Y %H:%M:%S\")\n",
    "    version_aliases = version.version_aliases\n",
    "    print(\n",
    "        f\"Model version {version_id} was created at {version_created_time} with aliases {version_aliases}\",\n",
    "    )"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "4e179df915a9"
   },
   "outputs": [],
   "source": [
    "# Get the model\n",
    "model = registry.get_model(version=\"logit\")\n",
    "print(model)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "a90e98c72a05"
   },
   "source": [
    "### Model evaluation"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "2aaaae772f67"
   },
   "source": [
    "With the model created, you can now evaluate the logistic regression model. Behind the scenes, BigQuery ML automatically [split the data](https://cloud.google.com/bigquery-ml/docs/reference/standard-sql/bigqueryml-syntax-create#data_split_method), which makes it easier to quickly train and evaluate models."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "a1f8ac93d570"
   },
   "outputs": [],
   "source": [
    "sql_evaluate_model = f\"\"\"\n",
    "SELECT\n",
    "  *\n",
    "FROM\n",
    "  ML.EVALUATE(MODEL tx.{MODEL_NAME})\n",
    "\"\"\"\n",
    "\n",
    "print(sql_evaluate_model)\n",
    "\n",
    "run_bq_query(sql_evaluate_model)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "d9f807a50f38"
   },
   "source": [
    "These metrics help you understand the performance of the model. \n",
    "\n",
    "There are various metrics for logistic regression and other model types (full list of metrics can be found in the [documentation](https://cloud.google.com/bigquery-ml/docs/reference/standard-sql/bigqueryml-syntax-evaluate#mlevaluate_output))."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "7e806ebc48a2"
   },
   "source": [
    "### Batch prediction (with Explainable AI)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "d31605829283"
   },
   "source": [
    "Make a batch prediction in BigQuery ML on the original training data to check the probability of a transaction to be fraudulent for transaction, as seen in the `probability` column, with the predicted label under the `predicted_tx_fraud` column.\n",
    "\n",
    "[ML.EXPLAIN_PREDICT](https://cloud.google.com/bigquery-ml/docs/reference/standard-sql/bigqueryml-syntax-explain-predict) has built-in [Explainable AI](https://cloud.google.com/bigquery-ml/docs/reference/standard-sql/bigqueryml-syntax-xai-overview). This allows you to see the top contributing features to each prediction and interpret how it was computed."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "2d500fdbfb44"
   },
   "outputs": [],
   "source": [
    "sql_explain_predict = f\"\"\"\n",
    "SELECT\n",
    "  *\n",
    "FROM\n",
    "  ML.EXPLAIN_PREDICT(MODEL tx.{MODEL_NAME},\n",
    "    (SELECT * FROM  `tx.{BQ_TABLE_NAME}` LIMIT 10)\n",
    "    )\n",
    "\"\"\"\n",
    "\n",
    "print(sql_explain_predict)\n",
    "\n",
    "run_bq_query(sql_explain_predict)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "fa1f96c0f452"
   },
   "source": [
    "Since the `top_feature_attributions` is a nested column, you can unnest the array ([documentation](https://cloud.google.com/bigquery/docs/reference/standard-sql/arrays)) into separate rows for each of the features. In other words, since ML.EXPLAIN_PREDICT provides the top 5 most important features, using `UNNEST` results in 5 rows per prediction:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "278b3441084b"
   },
   "outputs": [],
   "source": [
    "sql_explain_predict = f\"\"\"\n",
    "SELECT\n",
    "  tfa.*,\n",
    "  predicted_tx_fraud,\n",
    "  probability,\n",
    "  baseline_prediction_value,\n",
    "  prediction_value,\n",
    "  approximation_error,\n",
    "FROM\n",
    "  ML.EXPLAIN_PREDICT(MODEL tx.{MODEL_NAME},\n",
    "    (SELECT * FROM `tx.{BQ_TABLE_NAME}` )\n",
    "    ),\n",
    "  UNNEST(top_feature_attributions) as tfa\n",
    "LIMIT 100\n",
    "\"\"\"\n",
    "\n",
    "print(sql_explain_predict)\n",
    "\n",
    "run_bq_query(sql_explain_predict)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "89455f708f54"
   },
   "source": [
    "### Deploy the model to an endpoint"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "b6120dcc1ff6"
   },
   "source": [
    "While BigQuery ML supports batch prediction with [ML.PREDICT](https://cloud.google.com/bigquery-ml/docs/reference/standard-sql/bigqueryml-syntax-predict) and [ML.EXPLAIN_PREDICT](https://cloud.google.com/bigquery-ml/docs/reference/standard-sql/bigqueryml-syntax-explain-predict), BigQuery ML is not suitable for real-time predictions where you need low latency predictions with potentially high frequency of requests.\n",
    "\n",
    "In other words, deploying the BigQuery ML model to an endpoint enables you to do online predictions."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "e1ab7e1ac83c"
   },
   "source": [
    "#### Create a Vertex AI endpoint"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "0a61ea55f685"
   },
   "source": [
    "To deploy your model to an endpoint, you will first need to create an endpoint before you deploy the model to it."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "3ce73125dff6"
   },
   "outputs": [],
   "source": [
    "endpoint = vertex_ai.Endpoint.create(\n",
    "    display_name=ENDPOINT_NAME,\n",
    "    project=PROJECT_ID,\n",
    "    location=REGION,\n",
    ")\n",
    "\n",
    "print(endpoint.display_name)\n",
    "print(endpoint.resource_name)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "b58a104207d2"
   },
   "source": [
    "#### List endpoints"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "951ed1693f6b"
   },
   "source": [
    "List the endpoints to make sure it has successfully been created. You can also view your endpoints on the [Vertex AI Endpoints page](https://console.cloud.google.com/vertex-ai/endpoints)."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "887e007ee661"
   },
   "outputs": [],
   "source": [
    "endpoint.list(order_by=\"update_time\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "ba0d40b26cfb"
   },
   "source": [
    "#### Deploy model to Vertex endpoint"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "6a90be5b77a2"
   },
   "source": [
    "With the model, you can now deploy it to an endpoint. "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "a54ceab19a6e"
   },
   "outputs": [],
   "source": [
    "# deploying the model to the endpoint may take 10-15 minutes\n",
    "model.deploy(endpoint=endpoint)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "c303d779477b"
   },
   "source": [
    "You can also check on the status of your model by visiting the [Vertex AI Endpoints page](https://console.cloud.google.com/vertex-ai/endpoints)."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "4cb04d49c6f1"
   },
   "source": [
    "### Make online predictions to the endpoint with pub/sub -> pull subscription -> notebook -> Vertex AI endpoint\n",
    "Using a sample of the training data, you can test the endpoint to make online predictions."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "b2197d74f6e4"
   },
   "source": [
    "Below are some helper functions to help make it easier to read streaming data and do online predictions:"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "54135f8399af"
   },
   "outputs": [],
   "source": [
    "# A function to read some sample messages (transaction) from the public Pub/Sub\n",
    "def read_from_sub(project_id, subscription_path, messages=10):\n",
    "    \"\"\"\n",
    "    Read messages from a Pub/Sub subscription\n",
    "    Args:\n",
    "        project_id: project ID\n",
    "        subscription_name: the name of a Pub/Sub subscription in your project\n",
    "        messages: number of messages to read\n",
    "    Returns:\n",
    "        msg_data: list of messages in your Pub/Sub subscription as a Python dictionary\n",
    "    \"\"\"\n",
    "    import ast\n",
    "\n",
    "    from google.api_core import retry\n",
    "    from google.cloud import pubsub_v1\n",
    "\n",
    "    subscriber = pubsub_v1.SubscriberClient()\n",
    "    subscription_path = subscriber.subscription_path(project_id, subscription_path)\n",
    "\n",
    "    # Wrap the subscriber in a 'with' block to automatically call close() to\n",
    "    # close the underlying gRPC channel when done.\n",
    "    with subscriber:\n",
    "        # The subscriber pulls a specific number of messages. The actual\n",
    "        # number of messages pulled may be smaller than max_messages.\n",
    "        response = subscriber.pull(\n",
    "            subscription=subscription_path,\n",
    "            max_messages=messages,\n",
    "            retry=retry.Retry(deadline=300),\n",
    "        )\n",
    "\n",
    "        if len(response.received_messages) == 0:\n",
    "            print(\"no messages\")\n",
    "            return\n",
    "\n",
    "        ack_ids = []\n",
    "        msg_data = []\n",
    "        for received_message in response.received_messages:\n",
    "            msg = ast.literal_eval(received_message.message.data.decode(\"utf-8\"))\n",
    "            print(f\"Received: {msg}.\")\n",
    "            msg_data.append(msg)\n",
    "            ack_ids.append(received_message.ack_id)\n",
    "\n",
    "        # Acknowledges the received messages so they will not be sent again.\n",
    "        subscriber.acknowledge(subscription=subscription_path, ack_ids=ack_ids)\n",
    "\n",
    "        print(\n",
    "            f\"Received and acknowledged {len(response.received_messages)} messages from {subscription_path}.\"\n",
    "        )\n",
    "\n",
    "        return msg_data\n",
    "\n",
    "\n",
    "# A function for pre-processing of payload before sending it to a Vertex AI endpoint\n",
    "def preprocess(payload):\n",
    "    \"\"\"\n",
    "    Preprocesses the payload before sending it to a Vertex AI endpoint\n",
    "    Args:\n",
    "        payload: payload to be preprocessed\n",
    "    Returns:\n",
    "        payload: preprocessed payload\n",
    "    \"\"\"\n",
    "    # replace NaN's\n",
    "    for key, value in payload.items():\n",
    "        if value is None:\n",
    "            payload[key] = 0.0\n",
    "    return payload\n",
    "\n",
    "\n",
    "# A function to lookup features in Vertex AI Feature Store\n",
    "def features_lookup(ff_feature_store, entity, entity_ids):\n",
    "    \"\"\"\n",
    "    Looks up features in Vertex AI Feature Store\n",
    "    Args:\n",
    "        ff_feature_store: Feature Store object\n",
    "        entity: entity type\n",
    "        entity_ids: list of entity IDs\n",
    "    Returns:\n",
    "        features: list of features\n",
    "    \"\"\"\n",
    "    entity_type = ff_feature_store.get_entity_type(entity)\n",
    "    aggregated_features = entity_type.read(entity_ids=entity_ids, feature_ids=\"*\")\n",
    "    aggregated_features_preprocessed = preprocess(aggregated_features)\n",
    "    features = aggregated_features_preprocessed.iloc[0].to_dict()\n",
    "    return features"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "f84067f6fe1f"
   },
   "source": [
    "You can now read some messages from Pub/Sub, preprocess it, augment it with some features from Vertex AI Feature Store, and submit to Vertex AI Endpoint for online predictions."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "f41ab5d61080"
   },
   "outputs": [],
   "source": [
    "messages = read_from_sub(\n",
    "    project_id=PROJECT_ID, subscription_path=\"ff-tx-sub\", messages=10\n",
    ")\n",
    "\n",
    "for payload_input in messages:\n",
    "    print(f\"The recieved payload from Pub/Sub is: {payload_input}\")\n",
    "    print(f\"-----------------------------------------\")\n",
    "    payload = {}\n",
    "    payload[\"tx_amount\"] = payload_input[\"TX_AMOUNT\"]\n",
    "    # look up the customer featues from FS (wrttie by batch ingestion daily and by Dataflow in real-time)\n",
    "    customer_features = features_lookup(\n",
    "        ff_feature_store, \"customer\", [payload_input[\"CUSTOMER_ID\"]]\n",
    "    )\n",
    "    payload[\"customer_id_nb_tx_1day_window\"] = customer_features[\n",
    "        \"customer_id_nb_tx_1day_window\"\n",
    "    ]\n",
    "    payload[\"customer_id_nb_tx_7day_window\"] = customer_features[\n",
    "        \"customer_id_nb_tx_7day_window\"\n",
    "    ]\n",
    "    payload[\"customer_id_nb_tx_14day_window\"] = customer_features[\n",
    "        \"customer_id_nb_tx_14day_window\"\n",
    "    ]\n",
    "    payload[\"customer_id_avg_amount_1day_window\"] = customer_features[\n",
    "        \"customer_id_avg_amount_1day_window\"\n",
    "    ]\n",
    "    payload[\"customer_id_avg_amount_7day_window\"] = customer_features[\n",
    "        \"customer_id_avg_amount_7day_window\"\n",
    "    ]\n",
    "    payload[\"customer_id_avg_amount_14day_window\"] = customer_features[\n",
    "        \"customer_id_avg_amount_14day_window\"\n",
    "    ]\n",
    "    payload[\"customer_id_nb_tx_15min_window\"] = customer_features[\n",
    "        \"customer_id_nb_tx_15min_window\"\n",
    "    ]\n",
    "    payload[\"customer_id_avg_amount_15min_window\"] = customer_features[\n",
    "        \"customer_id_avg_amount_15min_window\"\n",
    "    ]\n",
    "    payload[\"customer_id_nb_tx_30min_window\"] = customer_features[\n",
    "        \"customer_id_nb_tx_30min_window\"\n",
    "    ]\n",
    "    payload[\"customer_id_avg_amount_30min_window\"] = customer_features[\n",
    "        \"customer_id_avg_amount_30min_window\"\n",
    "    ]\n",
    "    payload[\"customer_id_nb_tx_60min_window\"] = customer_features[\n",
    "        \"customer_id_nb_tx_60min_window\"\n",
    "    ]\n",
    "    payload[\"customer_id_avg_amount_60min_window\"] = customer_features[\n",
    "        \"customer_id_avg_amount_60min_window\"\n",
    "    ]\n",
    "    # look up the terminal featues from FS (wrttie by batch ingestion daily and by Dataflow in real-time)\n",
    "    terminal_features = features_lookup(\n",
    "        ff_feature_store, \"terminal\", [payload_input[\"TERMINAL_ID\"]]\n",
    "    )\n",
    "    payload[\"terminal_id_nb_tx_1day_window\"] = terminal_features[\n",
    "        \"terminal_id_nb_tx_1day_window\"\n",
    "    ]\n",
    "    payload[\"terminal_id_nb_tx_7day_window\"] = terminal_features[\n",
    "        \"terminal_id_nb_tx_7day_window\"\n",
    "    ]\n",
    "    payload[\"terminal_id_nb_tx_14day_window\"] = terminal_features[\n",
    "        \"terminal_id_nb_tx_14day_window\"\n",
    "    ]\n",
    "    payload[\"terminal_id_risk_1day_window\"] = terminal_features[\n",
    "        \"terminal_id_risk_1day_window\"\n",
    "    ]\n",
    "    payload[\"terminal_id_risk_7day_window\"] = terminal_features[\n",
    "        \"terminal_id_risk_7day_window\"\n",
    "    ]\n",
    "    payload[\"terminal_id_risk_14day_window\"] = terminal_features[\n",
    "        \"terminal_id_risk_14day_window\"\n",
    "    ]\n",
    "    payload[\"terminal_id_nb_tx_15min_window\"] = terminal_features[\n",
    "        \"terminal_id_nb_tx_15min_window\"\n",
    "    ]\n",
    "    payload[\"terminal_id_avg_amount_15min_window\"] = terminal_features[\n",
    "        \"terminal_id_avg_amount_15min_window\"\n",
    "    ]\n",
    "    payload[\"terminal_id_nb_tx_30min_window\"] = terminal_features[\n",
    "        \"terminal_id_nb_tx_30min_window\"\n",
    "    ]\n",
    "    payload[\"terminal_id_avg_amount_30min_window\"] = terminal_features[\n",
    "        \"terminal_id_avg_amount_30min_window\"\n",
    "    ]\n",
    "    payload[\"terminal_id_nb_tx_60min_window\"] = terminal_features[\n",
    "        \"terminal_id_nb_tx_60min_window\"\n",
    "    ]\n",
    "    payload[\"terminal_id_avg_amount_60min_window\"] = terminal_features[\n",
    "        \"terminal_id_avg_amount_60min_window\"\n",
    "    ]\n",
    "    payload = preprocess(payload)\n",
    "    print(f\"The input payload to the Vertex AI endpoint: {payload}\")\n",
    "    print(f\"-----------------------------------------\")\n",
    "\n",
    "    result = endpoint.predict(instances=[payload])\n",
    "    print(f\"The prediction result: {result}\")\n",
    "    print(f\"===============================================================\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### END\n",
    "\n",
    "Now you can go to the next notebook `05_model_training_pipeline_formalization.ipynb`"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {
    "id": "094c24798d33",
    "tags": []
   },
   "source": [
    "### Clean up"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "f5191374613a"
   },
   "outputs": [],
   "source": [
    "# endpoint[-1].undeploy_all()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {
    "id": "75cb7aba1566"
   },
   "outputs": [],
   "source": [
    "# delete_model_sql = f\"\"\"\n",
    "# DROP MODEL `{PROJECT_ID}.{BQ_DATASET}.{BQML_MODEL_NAME}`\n",
    "# \"\"\"\n",
    "\n",
    "# bq_query(delete_model_sql)"
   ]
  }
 ],
 "metadata": {
  "colab": {
   "collapsed_sections": [],
   "name": "04_model_training_and_prediction.ipynb",
   "toc_visible": true
  },
  "environment": {
   "kernel": "python3",
   "name": "common-cpu.m96",
   "type": "gcloud",
   "uri": "gcr.io/deeplearning-platform-release/base-cpu:m96"
  },
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.12"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}