08_predict.ipynb

{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#default_exp live.predict"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Live Predict\n",
    "\n",
    "> This module implements the POST endpoint for the Fraud Prediction Service. It can be invoked with an Order payload,\n",
    "then this steps will retrieve concurrently features calculated on customer_id and email, update the features using received order information on the fly, and predict is_fraud label using trained XGBoost model."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#export\n",
    "from typing import Dict, Optional\n",
    "from datetime import datetime, timezone, timedelta\n",
    "import os\n",
    "import json\n",
    "import pickle\n",
    "import aioredis\n",
    "import asyncio\n",
    "\n",
    "import pandas as pd\n",
    "import numpy as np\n",
    "import xgboost as xgb\n",
    "from dataclasses import dataclass\n",
    "from hopeit.dataobjects import dataobject\n",
    "from hopeit.server.serialization import serialize, Serialization, deserialize\n",
    "from hopeit.server.compression import Compression\n",
    "from hopeit.app.context import EventContext, PostprocessHook\n",
    "from hopeit.app.events import Spawn, SHUFFLE\n",
    "from hopeit.app.api import event_api\n",
    "from hopeit.app.logger import app_logger"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#export\n",
    "@dataobject\n",
    "@dataclass\n",
    "class OrderInfo:\n",
    "    order_id: str\n",
    "    customer_id: str\n",
    "    order_date: datetime\n",
    "    email: str\n",
    "    ip_addr: str\n",
    "    order_amount: float\n",
    "    location_lat: float\n",
    "    location_long: float\n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#export\n",
    "__steps__ = ['lookup_features', 'predict']\n",
    "\n",
    "__api__ = event_api(\n",
    "    summary=\"Live: Predict Fraud\",\n",
    "    payload=(OrderInfo, \"Order Information\"),\n",
    "    responses={\n",
    "        200: (dict, \"features used for prediction contatining `is_fraud` field as result of prediction\"),\n",
    "        404: (str, \"customer or email not found (this example only works for known customer_id and email)\")\n",
    "    }\n",
    ")\n",
    "\n",
    "logger = app_logger()\n",
    "\n",
    "model = None\n",
    "db = None\n",
    "\n",
    "features = ['order_amount', \n",
    "            'num_email_by_customer_id', 'same_email_by_customer_id', 'known_email_by_customer_id', \n",
    "            'num_ip_addr_by_customer_id', 'same_ip_addr_by_customer_id', 'known_ip_addr_by_customer_id',\n",
    "            'num_customer_id_by_email', 'same_customer_id_by_email', 'known_customer_id_by_email',\n",
    "            'order_amount_mean_by_customer_id',\n",
    "            'order_amount_std_by_customer_id', 'order_amount_min_by_customer_id', 'order_amount_max_by_customer_id',\n",
    "            'order_amount_sum_by_customer_id', \n",
    "            'order_amount_mean_by_email',\n",
    "            'order_amount_std_by_email', 'order_amount_min_by_email', 'order_amount_max_by_email',\n",
    "            'order_amount_sum_by_email']"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#export\n",
    "async def __init_event__(context: EventContext):\n",
    "    global model, db\n",
    "    if model is None:\n",
    "        file_name = os.path.join(context.env['model']['path'], context.env['model']['name'])\n",
    "        logger.info(context, f\"Loading model for prediction from {file_name}...\")\n",
    "        with open(file_name, 'rb') as fb:\n",
    "            model = pickle.load(fb)\n",
    "    if db is None:\n",
    "        address = context.env['db']['url']\n",
    "        logger.info(context, f\"Connecting to database {address}...\")\n",
    "        db = await aioredis.create_redis_pool(address)\n",
    "        "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#export\n",
    "async def _lookup_db(key: str):\n",
    "    item = await db.get(key)\n",
    "    if item is None:\n",
    "        return None\n",
    "    return deserialize(item, Serialization.PICKLE4, Compression.LZ4, dict)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#export\n",
    "async def lookup_features(order: OrderInfo, context: EventContext) -> Optional[dict]:\n",
    "    logger.info(context, \"Looking up features in database...\")\n",
    "    assert db, \"Connection to database missing.\"\n",
    "    customer_id_features, email_features = await asyncio.gather(\n",
    "        _lookup_db(order.customer_id),\n",
    "        _lookup_db(order.email)\n",
    "    )\n",
    "    if customer_id_features is None or email_features is None:\n",
    "        return None\n",
    "    return {\n",
    "        **_update_features(order, email_features, 'email'), \n",
    "        **_update_features(order, customer_id_features, 'customer_id'),\n",
    "        **order.to_dict()\n",
    "    }\n",
    "\n",
    "def _append(data: dict, k: str, new_item: str):\n",
    "    x = data.get(k)\n",
    "    if isinstance(x, str):\n",
    "        x = json.loads(x)\n",
    "        x.append(new_item)\n",
    "        data[k] = list(set(x[-10:]))\n",
    "\n",
    "def _update_features(order: OrderInfo, data: dict, by: str):\n",
    "    _append(data, f'order_amount_by_{by}', order.order_amount)\n",
    "    _append(data, f'ip_addr_by_{by}', order.ip_addr)\n",
    "    _append(data, f'email_by_{by}', order.email)\n",
    "    _append(data, f'customer_id_by_{by}', order.customer_id)\n",
    "    _calc_counts(data, 'customer_id')\n",
    "    _calc_counts(data, 'email')\n",
    "    _calc_amount_stats(data, 'customer_id')\n",
    "    _calc_amount_stats(data, 'email')\n",
    "    return data\n",
    "\n",
    "def _calc_counts(data: dict, by: str):\n",
    "    for col in ['ip_addr', 'customer_id', 'email']:\n",
    "        x = data.get(f'{col}_by_{by}')\n",
    "        if x is not None:\n",
    "            data[f'num_{col}_by_{by}'] = len(x)\n",
    "\n",
    "def _calc_amount_stats(data: dict, by: str):\n",
    "    col = 'order_amount'\n",
    "    x = data.get(f'{col}_by_{by}')\n",
    "    if x is not None:\n",
    "        x = np.array(x)\n",
    "        data[f'{col}_max_by_{by}'] = np.max(x) \n",
    "        data[f'{col}_min_by_{by}'] = np.min(x) \n",
    "        data[f'{col}_mean_by_{by}'] = np.mean(x) \n",
    "        data[f'{col}_std_by_{by}'] = np.std(x) \n",
    "        data[f'{col}_sum_by_{by}'] = np.sum(x) \n"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#export\n",
    "async def predict(data: dict, context: EventContext) -> dict:\n",
    "    df = pd.DataFrame([data], columns=features)\n",
    "    x = xgb.DMatrix(df)\n",
    "    y = model.predict(x)\n",
    "    data['is_fraud'] = y[0].item()\n",
    "    return data\n",
    "    "
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#export\n",
    "async def __postprocess__(payload: Optional[dict], context: EventContext, response: PostprocessHook) -> dict:\n",
    "    if payload is None:\n",
    "        response.status = 404\n",
    "        return \"customer or email not found (this example only works for known customer_id and email)\"\n",
    "    return payload\n",
    "        "
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Test from notebook"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2020-07-08 11:37:25,377 | INFO | fraud-poc 0.0.1-service live.predict leo-legion 38272 | __init_event__ module=fraud_poc.live.predict... | track.operation_id=test_operation_id | track.request_id=test_request_id | track.request_ts=2020-07-08T11:37:25.377008+00:00\n",
      "2020-07-08 11:37:25,377 | INFO | fraud-poc 0.0.1-service live.predict leo-legion 38272 | Loading model for prediction from ./data/model/xgb/latest-ok.pkl... | track.operation_id=test_operation_id | track.request_id=test_request_id | track.request_ts=2020-07-08T11:37:25.377008+00:00\n",
      "2020-07-08 11:37:25,382 | INFO | fraud-poc 0.0.1-service live.predict leo-legion 38272 | Connecting to database redis://localhost:6379... | track.operation_id=test_operation_id | track.request_id=test_request_id | track.request_ts=2020-07-08T11:37:25.377008+00:00\n",
      "2020-07-08 11:37:25,384 | INFO | fraud-poc 0.0.1-service live.predict leo-legion 38272 | Looking up features in database... | track.operation_id=test_operation_id | track.request_id=test_request_id | track.request_ts=2020-07-08T11:37:25.377008+00:00\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "{'order_id': 'bb2de6c6-d655-4e75-8949-4ed792c024e5',\n",
       " 'order_date': '2020-07-08T11:37:25.376310+00:00',\n",
       " 'customer_id': 'd555b585-5511-4a16-9f22-819834110239',\n",
       " 'ip_addr': 'test',\n",
       " 'order_amount': 100.0,\n",
       " 'email': '1f5d34b02ef1975d5a82dcfe2e53fad6182e118c',\n",
       " 'customer_id_by_email': ['d555b585-5511-4a16-9f22-819834110239'],\n",
       " 'num_customer_id_by_email': 1,\n",
       " 'last_customer_id_by_email': 'd555b585-5511-4a16-9f22-819834110239',\n",
       " 'same_customer_id_by_email': 1,\n",
       " 'known_customer_id_by_email': 1,\n",
       " 'order_amount_mean_by_email': 438.30120590151927,\n",
       " 'order_amount_std_by_email': 274.2623429763596,\n",
       " 'order_amount_min_by_email': 13.426760485169664,\n",
       " 'order_amount_max_by_email': 879.7319538388477,\n",
       " 'order_amount_sum_by_email': 4383.012059015193,\n",
       " 'order_amount_by_email': [387.7647884248259,\n",
       "  100.0,\n",
       "  202.94798232836487,\n",
       "  13.426760485169664,\n",
       "  591.5459546297528,\n",
       "  879.7319538388477,\n",
       "  722.9025680336106,\n",
       "  627.8375917324881,\n",
       "  249.78620396645113,\n",
       "  607.0682555756823],\n",
       " 'key': 'd555b585-5511-4a16-9f22-819834110239',\n",
       " 'email_by_customer_id': ['63552fd2e3d7e9e71e761c50f8b7ff1c07f39abb',\n",
       "  'd33cfd28bb631654b8e59ac97a958555ad39ea78',\n",
       "  '1f5d34b02ef1975d5a82dcfe2e53fad6182e118c',\n",
       "  '239b8d32e104be4b7c0cc65e2cbc1ead42522578'],\n",
       " 'ip_addr_by_customer_id': ['4efa50345ed594d6f817f909e1b60473cb9d6c39',\n",
       "  '5c30206672cf95de1e41294ef3391a0f369e5fc8',\n",
       "  '6f63bf7b774f8594744b5d1fc17a0e1c43cfd809',\n",
       "  '785cd9911322a86079b9446e895b535490171a75',\n",
       "  '1404f14e3e04cdbbfadcd3ea70f014226bb9ac26',\n",
       "  '270e3d5cf5ddfde308b19442e225f8c3b9251ecc',\n",
       "  'test'],\n",
       " 'num_email_by_customer_id': 4,\n",
       " 'num_ip_addr_by_customer_id': 7,\n",
       " 'last_email_by_customer_id': '63552fd2e3d7e9e71e761c50f8b7ff1c07f39abb',\n",
       " 'last_ip_addr_by_customer_id': '5c30206672cf95de1e41294ef3391a0f369e5fc8',\n",
       " 'same_email_by_customer_id': 0,\n",
       " 'same_ip_addr_by_customer_id': 0,\n",
       " 'known_email_by_customer_id': 1,\n",
       " 'known_ip_addr_by_customer_id': 1,\n",
       " 'order_amount_mean_by_customer_id': 418.2469834416056,\n",
       " 'order_amount_std_by_customer_id': 259.19292283459004,\n",
       " 'order_amount_min_by_customer_id': 100.0,\n",
       " 'order_amount_max_by_customer_id': 879.7319538388477,\n",
       " 'order_amount_sum_by_customer_id': 4182.469834416056,\n",
       " 'order_amount_by_customer_id': [100.0,\n",
       "  453.97520881198227,\n",
       "  648.5861853978612,\n",
       "  591.5459546297528,\n",
       "  879.7319538388477,\n",
       "  305.8898989967592,\n",
       "  690.3395151993851,\n",
       "  147.61131120991155,\n",
       "  214.33601894031463,\n",
       "  150.4537873912415],\n",
       " 'location_lat': 0.0,\n",
       " 'location_long': 0.0,\n",
       " 'is_fraud': 0.542777955532074}"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "from hopeit.testing.apps import config, execute_event\n",
    "from fraud_poc.live.predict import OrderInfo\n",
    "from datetime import datetime, timezone\n",
    "import uuid\n",
    " \n",
    "def new_key(): return str(uuid.uuid4())\n",
    "\n",
    "app_config = config('config/fraud-service.json')\n",
    "payload = OrderInfo(\n",
    "    order_id=new_key(),\n",
    "    customer_id='d555b585-5511-4a16-9f22-819834110239',\n",
    "    order_date=datetime.now(tz=timezone.utc),\n",
    "    email='1f5d34b02ef1975d5a82dcfe2e53fad6182e118c',\n",
    "    ip_addr='test',\n",
    "    order_amount=100.0,\n",
    "    location_lat=0.0,\n",
    "    location_long=0.0\n",
    ") \n",
    "\n",
    "result = await execute_event(app_config, 'live.predict', payload)\n",
    "result"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "name": "stderr",
     "output_type": "stream",
     "text": [
      "2020-07-08 11:38:45,990 | INFO | fraud-poc 0.0.1-service live.predict leo-legion 38272 | __init_event__ module=fraud_poc.live.predict... | track.operation_id=test_operation_id | track.request_id=test_request_id | track.request_ts=2020-07-08T11:38:45.990505+00:00\n",
      "2020-07-08 11:38:45,991 | INFO | fraud-poc 0.0.1-service live.predict leo-legion 38272 | Looking up features in database... | track.operation_id=test_operation_id | track.request_id=test_request_id | track.request_ts=2020-07-08T11:38:45.990505+00:00\n"
     ]
    },
    {
     "data": {
      "text/plain": [
       "{'order_id': '29b1eb9a-17ed-4ea5-96b7-70b5c274dae5',\n",
       " 'order_date': '2020-07-08T11:38:45.990197+00:00',\n",
       " 'customer_id': '002d6c90-2680-494a-b65e-c969c48277c8',\n",
       " 'ip_addr': 'c7ee6b54b39b1b5edefa3aa85d24665ed8feadd5',\n",
       " 'order_amount': 10.0,\n",
       " 'email': '1f5d34b02ef1975d5a82dcfe2e53fad6182e118c',\n",
       " 'customer_id_by_email': ['002d6c90-2680-494a-b65e-c969c48277c8',\n",
       "  'd555b585-5511-4a16-9f22-819834110239'],\n",
       " 'num_customer_id_by_email': 2,\n",
       " 'last_customer_id_by_email': 'd555b585-5511-4a16-9f22-819834110239',\n",
       " 'same_customer_id_by_email': 1,\n",
       " 'known_customer_id_by_email': 1,\n",
       " 'order_amount_mean_by_email': 429.30120590151927,\n",
       " 'order_amount_std_by_email': 286.4232086984392,\n",
       " 'order_amount_min_by_email': 10.0,\n",
       " 'order_amount_max_by_email': 879.7319538388477,\n",
       " 'order_amount_sum_by_email': 4293.012059015193,\n",
       " 'order_amount_by_email': [387.7647884248259,\n",
       "  202.94798232836487,\n",
       "  10.0,\n",
       "  13.426760485169664,\n",
       "  591.5459546297528,\n",
       "  879.7319538388477,\n",
       "  722.9025680336106,\n",
       "  627.8375917324881,\n",
       "  249.78620396645113,\n",
       "  607.0682555756823],\n",
       " 'key': '002d6c90-2680-494a-b65e-c969c48277c8',\n",
       " 'email_by_customer_id': ['afe4b272a5def9d9b3cd9d8ac9b6136bbe90ffe9',\n",
       "  '1f5d34b02ef1975d5a82dcfe2e53fad6182e118c'],\n",
       " 'ip_addr_by_customer_id': ['c7ee6b54b39b1b5edefa3aa85d24665ed8feadd5'],\n",
       " 'num_email_by_customer_id': 2,\n",
       " 'num_ip_addr_by_customer_id': 1,\n",
       " 'last_email_by_customer_id': 'afe4b272a5def9d9b3cd9d8ac9b6136bbe90ffe9',\n",
       " 'last_ip_addr_by_customer_id': 'c7ee6b54b39b1b5edefa3aa85d24665ed8feadd5',\n",
       " 'same_email_by_customer_id': 1,\n",
       " 'same_ip_addr_by_customer_id': 1,\n",
       " 'known_email_by_customer_id': 1,\n",
       " 'known_ip_addr_by_customer_id': 1,\n",
       " 'order_amount_mean_by_customer_id': 512.1876094673013,\n",
       " 'order_amount_std_by_customer_id': 302.04578020233225,\n",
       " 'order_amount_min_by_customer_id': 10.0,\n",
       " 'order_amount_max_by_customer_id': 895.8665915672377,\n",
       " 'order_amount_sum_by_customer_id': 5121.876094673013,\n",
       " 'order_amount_by_customer_id': [551.3028970809393,\n",
       "  840.1112472933949,\n",
       "  10.0,\n",
       "  459.04884566221347,\n",
       "  620.4744338380846,\n",
       "  780.6965636312258,\n",
       "  145.4531898362139,\n",
       "  693.7183138797553,\n",
       "  125.20401188394759,\n",
       "  895.8665915672377],\n",
       " 'location_lat': 0.0,\n",
       " 'location_long': 0.0,\n",
       " 'is_fraud': 0.4534066617488861}"
      ]
     },
     "execution_count": null,
     "metadata": {},
     "output_type": "execute_result"
    }
   ],
   "source": [
    "payload = OrderInfo(\n",
    "    order_id=new_key(),\n",
    "    customer_id='002d6c90-2680-494a-b65e-c969c48277c8',\n",
    "    order_date=datetime.now(tz=timezone.utc),\n",
    "    email='1f5d34b02ef1975d5a82dcfe2e53fad6182e118c',\n",
    "    ip_addr='c7ee6b54b39b1b5edefa3aa85d24665ed8feadd5',\n",
    "    order_amount=10.0,\n",
    "    location_lat=0.0,\n",
    "    location_long=0.0\n",
    ") \n",
    "\n",
    "result = await execute_event(app_config, 'live.predict', payload)\n",
    "result"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}