custom_metric.ipynb

{
 "cells": [
  {
   "cell_type": "markdown",
   "source": [
    "# Custom Evaluation Metric\n",
    "TurboML allows you to define your own aggregate metrics in Python."
   ],
   "id": "0",
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## Set up the environment and install TurboML's SDK.\n",
    "We use `turboml-installer` to set up the environment for TurboML's SDK."
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "source": [
    "!pip install -q turboml-installer\n",
    "import turboml_installer ; turboml_installer.install_on_colab()"
   ],
   "execution_count": null,
   "metadata": {},
   "outputs": []
  },
  {
   "cell_type": "markdown",
   "source": [
    "The kernel should now be restarted with TurboML's SDK installed."
   ],
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "## Login to your TurboML instance\n",
    "\n",
    "Note that you can copy and replace this snippet with one from your TurboML homepage."
   ],
   "metadata": {}
  },
  {
   "cell_type": "code",
   "source": [
    "import turboml as tb",
    "\ntb.init(backend_url=BACKEND_URL, api_key=API_KEY)"
   ],
   "execution_count": null,
   "id": "1",
   "metadata": {},
   "outputs": []
  },
  {
   "cell_type": "code",
   "source": [
    "from turboml.common import ModelMetricAggregateFunction\n",
    "import math\n",
    "import pandas as pd"
   ],
   "execution_count": null,
   "id": "2",
   "metadata": {},
   "outputs": []
  },
  {
   "cell_type": "markdown",
   "source": [
    "### Model Metric Aggregation function\n",
    "\n",
    "Metric aggregate functions are used to add and compute any custom metric over model predictions and labels.\n",
    "\n",
    "#### Overview of Metric Aggregate Functions\n",
    "A metric aggregate function consists of the following lifecycle methods:\n",
    "\n",
    "1. ```create_state()```: Initializes the aggregation state.\n",
    "2. ```accumulate(state, prediction, label)```: Updates the state based on input values.\n",
    "3. ```retract(state, prediction, label) (optional)```: Reverses the effect of previously accumulated values (useful in sliding windows or similar contexts).\n",
    "4. ```merge_states(state1, state2)```: Merges two states (for distributed computation).\n",
    "5. ```finish(state)```: Computes and returns the final metric value."
   ],
   "id": "3",
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "### Steps to Define a Metric Aggregate Function\n",
    "\n",
    "**1. Define a Subclass**\n",
    "\n",
    "Create a subclass of ```ModelMetricAggregateFunction``` and override its methods.\n",
    "\n",
    "**2. Implement Required Methods**\n",
    "\n",
    "At a minimum, one needs to implement:\n",
    "\n",
    "- create_state\n",
    "- accumulate\n",
    "- finish\n",
    "- merge_states"
   ],
   "id": "4",
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "### Example: Focal Loss Metric\n",
    "Here’s an example of a custom focal loss metric function."
   ],
   "id": "5",
   "metadata": {}
  },
  {
   "cell_type": "code",
   "source": [
    "class FocalLoss(ModelMetricAggregateFunction):\n",
    "    def __init__(self):\n",
    "        super().__init__()\n",
    "\n",
    "    def create_state(self):\n",
    "        \"\"\"\n",
    "        Initialize the aggregation state.\n",
    "        Returns:\n",
    "            Any: A serializable object representing the initial state of the metric.\n",
    "            This can be a tuple, dictionary, or any other serializable data structure.\n",
    "            Note:\n",
    "                - The serialized size of the state should be less than 8MB to ensure\n",
    "                  compatibility with distributed systems and to avoid exceeding storage\n",
    "                  or transmission limits.\n",
    "                - Ensure the state is lightweight and efficiently encodable for optimal\n",
    "                  performance.\n",
    "        \"\"\"\n",
    "        return (0.0, 0)\n",
    "\n",
    "    def _compute_focal_loss(self, prediction, label, gamma=2.0, alpha=0.25):\n",
    "        if prediction is None or label is None:\n",
    "            return None\n",
    "        pt = prediction if label == 1 else 1 - prediction\n",
    "        pt = max(min(pt, 1 - 1e-6), 1e-6)\n",
    "        return -alpha * ((1 - pt) ** gamma) * math.log(pt)\n",
    "\n",
    "    def accumulate(self, state, prediction, label):\n",
    "        \"\"\"\n",
    "        Update the state with a new prediction-target pair.\n",
    "        Args:\n",
    "            state (Any): The current aggregation state.\n",
    "            prediction (float): Predicted value.\n",
    "            label (float): Ground truth.\n",
    "        Returns:\n",
    "            Any: The updated aggregation state, maintaining the same format and requirements as `create_state`.\n",
    "        \"\"\"\n",
    "        loss_sum, weight_sum = state\n",
    "        focal_loss = self._compute_focal_loss(prediction, label)\n",
    "        if focal_loss is None:\n",
    "            return state\n",
    "        return loss_sum + focal_loss, weight_sum + 1\n",
    "\n",
    "    def finish(self, state):\n",
    "        \"\"\"\n",
    "        Compute the final metric value.\n",
    "        Args:\n",
    "            state (Any): Final state.\n",
    "        Returns:\n",
    "            float: The result.\n",
    "        \"\"\"\n",
    "        loss_sum, weight_sum = state\n",
    "        return 0 if weight_sum == 0 else loss_sum / weight_sum\n",
    "\n",
    "    def merge_states(self, state1, state2):\n",
    "        \"\"\"\n",
    "        Merge two states (for distributed computations).\n",
    "        Args:\n",
    "            state1 (Any): The first aggregation state.\n",
    "            state2 (Any): The second aggregation state.\n",
    "\n",
    "        Returns:\n",
    "            tuple: Merged state, maintaining the same format and requirements as `create_state`.\n",
    "        \"\"\"\n",
    "        loss_sum1, weight_sum1 = state1\n",
    "        loss_sum2, weight_sum2 = state2\n",
    "        return loss_sum1 + loss_sum2, weight_sum1 + weight_sum2"
   ],
   "execution_count": null,
   "id": "6",
   "metadata": {},
   "outputs": []
  },
  {
   "cell_type": "markdown",
   "source": [
    "### Guidelines for Implementation\n",
    "1. State Management:\n",
    "    - Ensure the state is serializable and the serialized size of the state should be less than 8MB\n",
    "2. Edge Cases:\n",
    "    - Handle cases where inputs might be None.\n",
    "    - Ensure finish() handles empty states gracefully."
   ],
   "id": "7",
   "metadata": {}
  },
  {
   "cell_type": "markdown",
   "source": [
    "We will create one model to test the metric. Please follow the quickstart doc for details."
   ],
   "id": "8",
   "metadata": {}
  },
  {
   "cell_type": "code",
   "source": [
    "transactions = tb.datasets.FraudDetectionDatasetFeatures().to_online(\n",
    "    \"transactions\", load_if_exists=True\n",
    ")\n",
    "labels = tb.datasets.FraudDetectionDatasetLabels().to_online(\n",
    "    \"transaction_labels\", load_if_exists=True\n",
    ")"
   ],
   "execution_count": null,
   "id": "9",
   "metadata": {},
   "outputs": []
  },
  {
   "cell_type": "code",
   "source": [
    "model = tb.HoeffdingTreeClassifier(n_classes=2)"
   ],
   "execution_count": null,
   "id": "10",
   "metadata": {},
   "outputs": []
  },
  {
   "cell_type": "code",
   "source": [
    "numerical_fields = [\n",
    "    \"transactionAmount\",\n",
    "    \"localHour\",\n",
    "]\n",
    "features = transactions.get_model_inputs(numerical_fields=numerical_fields)\n",
    "label = labels.get_model_labels(label_field=\"is_fraud\")"
   ],
   "execution_count": null,
   "id": "11",
   "metadata": {},
   "outputs": []
  },
  {
   "cell_type": "code",
   "source": [
    "deployed_model_hft = model.deploy(name=\"demo_model_hft\", input=features, labels=label)"
   ],
   "execution_count": null,
   "id": "12",
   "metadata": {},
   "outputs": []
  },
  {
   "cell_type": "code",
   "source": [
    "outputs = deployed_model_hft.get_outputs()"
   ],
   "execution_count": null,
   "id": "13",
   "metadata": {},
   "outputs": []
  },
  {
   "cell_type": "markdown",
   "source": [
    "We can register a metric and get evaluations"
   ],
   "id": "14",
   "metadata": {}
  },
  {
   "cell_type": "code",
   "source": [
    "tb.register_custom_metric(\"FocalLoss\", FocalLoss)"
   ],
   "execution_count": null,
   "id": "15",
   "metadata": {},
   "outputs": []
  },
  {
   "cell_type": "code",
   "source": [
    "model_scores = deployed_model_hft.get_evaluation(\"FocalLoss\")\n",
    "model_scores[-1]"
   ],
   "execution_count": null,
   "id": "16",
   "metadata": {},
   "outputs": []
  },
  {
   "cell_type": "code",
   "source": [
    "import matplotlib.pyplot as plt\n",
    "\n",
    "plt.plot([model_score.metric for model_score in model_scores])"
   ],
   "execution_count": null,
   "id": "17",
   "metadata": {},
   "outputs": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3 (ipykernel)",
   "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.11.9"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 5
}