{
 "cells": [
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Financial Services: Delta Liquid Clustering Demo\n",
    "\n",
    "\n",
    "## Overview\n",
    "\n",
    "\n",
    "This notebook demonstrates the power of **Delta Liquid Clustering** in Oracle AI Data Platform (AIDP) Workbench using a financial services analytics use case. Liquid clustering automatically optimizes data layout for query performance without requiring manual partitioning or Z-Ordering.\n",
    "\n",
    "### What is Liquid Clustering?\n",
    "\n",
    "Liquid clustering automatically identifies and groups similar data together based on clustering columns you define. This optimization happens automatically during data ingestion and maintenance operations, providing:\n",
    "\n",
    "- **Automatic optimization**: No manual tuning required\n",
    "- **Improved query performance**: Faster queries on clustered columns\n",
    "- **Reduced maintenance**: No need for manual repartitioning\n",
    "- **Adaptive clustering**: Adjusts as data patterns change\n",
    "\n",
    "### Use Case: Transaction Fraud Detection and Customer Analytics\n",
    "\n",
    "We'll analyze financial transaction records from a bank. Our clustering strategy will optimize for:\n",
    "\n",
    "- **Customer-specific queries**: Fast lookups by account ID\n",
    "- **Time-based analysis**: Efficient filtering by transaction date\n",
    "- **Fraud pattern detection**: Quick aggregation by transaction type and risk scores\n",
    "\n",
    "### AIDP Environment Setup\n",
    "\n",
    "This notebook leverages the existing Spark session in your AIDP environment."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# Create financial services catalog and analytics schema\n",
    "\n",
    "# In AIDP, catalogs provide data isolation and governance\n",
    "\n",
    "spark.sql(\"CREATE CATALOG IF NOT EXISTS finance\")\n",
    "\n",
    "spark.sql(\"CREATE SCHEMA IF NOT EXISTS finance.analytics\")\n",
    "\n",
    "print(\"Financial services catalog and analytics schema created successfully!\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 2: Create Delta Table with Liquid Clustering\n",
    "\n",
    "### Table Design\n",
    "\n",
    "Our `account_transactions` table will store:\n",
    "\n",
    "- **account_id**: Unique account identifier\n",
    "- **transaction_date**: Date and time of transaction\n",
    "- **transaction_type**: Type (Deposit, Withdrawal, Transfer, Payment, etc.)\n",
    "- **amount**: Transaction amount\n",
    "- **merchant_category**: Merchant type (Retail, Restaurant, Online, etc.)\n",
    "- **location**: Transaction location\n",
    "- **risk_score**: Fraud risk assessment (0-100)\n",
    "\n",
    "### Clustering Strategy\n",
    "\n",
    "We'll cluster by `account_id` and `transaction_date` because:\n",
    "\n",
    "- **account_id**: Customers often have multiple transactions, grouping their financial activity together\n",
    "- **transaction_date**: Time-based queries are critical for fraud detection, spending analysis, and regulatory reporting\n",
    "- This combination optimizes for both customer account analysis and temporal fraud pattern detection"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Delta table with liquid clustering created successfully!\n",
       "Clustering will automatically optimize data layout for queries on account_id and transaction_date.\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Create Delta table with liquid clustering\n",
    "\n",
    "# CLUSTER BY defines the columns for automatic optimization\n",
    "\n",
    "spark.sql(\"\"\"\n",
    "\n",
    "CREATE TABLE IF NOT EXISTS finance.analytics.account_transactions (\n",
    "\n",
    "    account_id STRING,\n",
    "\n",
    "    transaction_date TIMESTAMP,\n",
    "\n",
    "    transaction_type STRING,\n",
    "\n",
    "    amount DECIMAL(15,2),\n",
    "\n",
    "    merchant_category STRING,\n",
    "\n",
    "    location STRING,\n",
    "\n",
    "    risk_score INT\n",
    "\n",
    ")\n",
    "\n",
    "USING DELTA\n",
    "\n",
    "CLUSTER BY (account_id, transaction_date)\n",
    "\n",
    "\"\"\")\n",
    "\n",
    "print(\"Delta table with liquid clustering created successfully!\")\n",
    "\n",
    "print(\"Clustering will automatically optimize data layout for queries on account_id and transaction_date.\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 3: Generate Financial Services Sample Data\n",
    "\n",
    "### Data Generation Strategy\n",
    "\n",
    "We'll create realistic financial transaction data including:\n",
    "\n",
    "- **5,000 accounts** with multiple transactions over time\n",
    "- **Transaction types**: Deposits, withdrawals, transfers, payments, ATM withdrawals\n",
    "- **Realistic temporal patterns**: Daily banking activity, weekend vs weekday patterns\n",
    "- **Merchant categories**: Retail, restaurants, online shopping, utilities, entertainment\n",
    "\n",
    "### Why This Data Pattern?\n",
    "\n",
    "This data simulates real financial scenarios where:\n",
    "\n",
    "- Customers perform multiple transactions daily/weekly\n",
    "- Fraud patterns emerge over time\n",
    "- Regulatory reporting requires temporal analysis\n",
    "- Risk scoring enables real-time fraud prevention\n",
    "- Customer spending analysis drives personalized financial services"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Generated 149912 account transaction records\n",
       "Sample record: {'account_id': 'ACC00000001', 'transaction_date': datetime.datetime(2024, 6, 2, 10, 0), 'transaction_type': 'Deposit', 'amount': 6959.24, 'merchant_category': 'Online', 'location': 'ATM', 'risk_score': 71}\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Generate sample financial transaction data\n",
    "\n",
    "# Using fully qualified imports to avoid conflicts\n",
    "\n",
    "import random\n",
    "\n",
    "from datetime import datetime, timedelta\n",
    "\n",
    "\n",
    "# Define financial data constants\n",
    "\n",
    "TRANSACTION_TYPES = ['Deposit', 'Withdrawal', 'Transfer', 'Payment', 'ATM']\n",
    "\n",
    "MERCHANT_CATEGORIES = ['Retail', 'Restaurant', 'Online', 'Utilities', 'Entertainment', 'Groceries', 'Healthcare', 'Transportation']\n",
    "\n",
    "LOCATIONS = ['New York, NY', 'Los Angeles, CA', 'Chicago, IL', 'Houston, TX', 'Miami, FL', 'Online', 'ATM']\n",
    "\n",
    "\n",
    "# Generate account transaction records\n",
    "\n",
    "transaction_data = []\n",
    "\n",
    "base_date = datetime(2024, 1, 1)\n",
    "\n",
    "\n",
    "# Create 5,000 accounts with 10-50 transactions each\n",
    "\n",
    "for account_num in range(1, 5001):\n",
    "\n",
    "    account_id = f\"ACC{account_num:08d}\"\n",
    "    \n",
    "    # Each account gets 10-50 transactions over 12 months\n",
    "\n",
    "    num_transactions = random.randint(10, 50)\n",
    "    \n",
    "    for i in range(num_transactions):\n",
    "\n",
    "        # Spread transactions over 12 months with realistic timing\n",
    "\n",
    "        days_offset = random.randint(0, 365)\n",
    "\n",
    "        hours_offset = random.randint(0, 23)\n",
    "\n",
    "        transaction_date = base_date + timedelta(days=days_offset, hours=hours_offset)\n",
    "        \n",
    "        # Select transaction type\n",
    "\n",
    "        transaction_type = random.choice(TRANSACTION_TYPES)\n",
    "        \n",
    "        # Amount based on transaction type\n",
    "\n",
    "        if transaction_type in ['Deposit', 'Transfer']:\n",
    "\n",
    "            amount = round(random.uniform(100, 10000), 2)\n",
    "\n",
    "        elif transaction_type == 'ATM':\n",
    "\n",
    "            amount = round(random.uniform(20, 500), 2) * -1\n",
    "\n",
    "        else:\n",
    "\n",
    "            amount = round(random.uniform(10, 2000), 2) * -1\n",
    "        \n",
    "        # Select merchant category and location\n",
    "\n",
    "        merchant_category = random.choice(MERCHANT_CATEGORIES)\n",
    "\n",
    "        if transaction_type == 'ATM':\n",
    "\n",
    "            location = 'ATM'\n",
    "\n",
    "        elif transaction_type == 'Online':\n",
    "\n",
    "            location = 'Online'\n",
    "\n",
    "        else:\n",
    "\n",
    "            location = random.choice(LOCATIONS)\n",
    "        \n",
    "        # Risk score (0-100, higher = more suspicious)\n",
    "\n",
    "        risk_score = random.randint(0, 100)\n",
    "        \n",
    "        transaction_data.append({\n",
    "\n",
    "            \"account_id\": account_id,\n",
    "\n",
    "            \"transaction_date\": transaction_date,\n",
    "\n",
    "            \"transaction_type\": transaction_type,\n",
    "\n",
    "            \"amount\": amount,\n",
    "\n",
    "            \"merchant_category\": merchant_category,\n",
    "\n",
    "            \"location\": location,\n",
    "\n",
    "            \"risk_score\": risk_score\n",
    "\n",
    "        })\n",
    "\n",
    "\n",
    "\n",
    "print(f\"Generated {len(transaction_data)} account transaction records\")\n",
    "\n",
    "print(\"Sample record:\", transaction_data[0])"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 4: Insert Data Using PySpark\n",
    "\n",
    "### Data Insertion Strategy\n",
    "\n",
    "We'll use PySpark to:\n",
    "\n",
    "1. **Create DataFrame** from our generated data\n",
    "2. **Insert into Delta table** with liquid clustering\n",
    "3. **Verify the insertion** with a sample query\n",
    "\n",
    "### Why PySpark for Insertion?\n",
    "\n",
    "- **Distributed processing**: Handles large datasets efficiently\n",
    "- **Type safety**: Ensures data integrity\n",
    "- **Optimization**: Leverages Spark's query optimization\n",
    "- **Liquid clustering**: Automatically applies clustering during insertion"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "DataFrame Schema:\n",
       "root\n",
       " |-- account_id: string (nullable = true)\n",
       " |-- amount: double (nullable = true)\n",
       " |-- location: string (nullable = true)\n",
       " |-- merchant_category: string (nullable = true)\n",
       " |-- risk_score: long (nullable = true)\n",
       " |-- transaction_date: timestamp (nullable = true)\n",
       " |-- transaction_type: string (nullable = true)\n",
       "\n",
       "\n",
       "Sample Data:\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "+-----------+-------+------------+-----------------+----------+-------------------+----------------+\n",
       "| account_id| amount|    location|merchant_category|risk_score|   transaction_date|transaction_type|\n",
       "+-----------+-------+------------+-----------------+----------+-------------------+----------------+\n",
       "|ACC00000001|6959.24|         ATM|           Online|        71|2024-06-02 10:00:00|         Deposit|\n",
       "|ACC00000001| 171.51|         ATM|    Entertainment|        67|2024-04-21 17:00:00|         Deposit|\n",
       "|ACC00000001|4679.81| Chicago, IL|       Restaurant|        67|2024-05-11 20:00:00|         Deposit|\n",
       "|ACC00000001|8596.92|         ATM|       Restaurant|         5|2024-07-25 19:00:00|         Deposit|\n",
       "|ACC00000001|-999.96|New York, NY|    Entertainment|        61|2024-09-18 01:00:00|      Withdrawal|\n",
       "+-----------+-------+------------+-----------------+----------+-------------------+----------------+\n",
       "only showing top 5 rows\n",
       "\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "\n",
       "Successfully inserted 149912 records into finance.analytics.account_transactions\n",
       "Liquid clustering automatically optimized the data layout during insertion!\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Insert data using PySpark DataFrame operations\n",
    "\n",
    "# Using fully qualified function references to avoid conflicts\n",
    "\n",
    "\n",
    "# Create DataFrame from generated data\n",
    "\n",
    "df_transactions = spark.createDataFrame(transaction_data)\n",
    "\n",
    "\n",
    "# Display schema and sample data\n",
    "\n",
    "print(\"DataFrame Schema:\")\n",
    "\n",
    "df_transactions.printSchema()\n",
    "\n",
    "\n",
    "\n",
    "print(\"\\nSample Data:\")\n",
    "\n",
    "df_transactions.show(5)\n",
    "\n",
    "\n",
    "# Insert data into Delta table with liquid clustering\n",
    "\n",
    "# The CLUSTER BY (account_id, transaction_date) will automatically optimize the data layout\n",
    "\n",
    "df_transactions.write.mode(\"overwrite\").saveAsTable(\"finance.analytics.account_transactions\")\n",
    "\n",
    "\n",
    "print(f\"\\nSuccessfully inserted {df_transactions.count()} records into finance.analytics.account_transactions\")\n",
    "\n",
    "print(\"Liquid clustering automatically optimized the data layout during insertion!\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 5: Demonstrate Liquid Clustering Benefits\n",
    "\n",
    "### Query Performance Analysis\n",
    "\n",
    "Now let's see how liquid clustering improves query performance. We'll run queries that benefit from our clustering strategy:\n",
    "\n",
    "1. **Account transaction history** (clustered by account_id)\n",
    "2. **Time-based fraud analysis** (clustered by transaction_date)\n",
    "3. **Combined account + time queries** (optimal for our clustering)\n",
    "\n",
    "### Expected Performance Benefits\n",
    "\n",
    "With liquid clustering, these queries should be significantly faster because:\n",
    "\n",
    "- **Data locality**: Related records are physically grouped together\n",
    "- **Reduced I/O**: Less data needs to be read from disk\n",
    "- **Automatic optimization**: No manual tuning required"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "=== Query 1: Account Transaction History ===\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "+-----------+-------------------+----------------+--------+-----------------+\n",
       "| account_id|   transaction_date|transaction_type|  amount|merchant_category|\n",
       "+-----------+-------------------+----------------+--------+-----------------+\n",
       "|ACC00000001|2024-10-18 16:00:00|      Withdrawal|-1915.87|           Retail|\n",
       "|ACC00000001|2024-09-29 10:00:00|         Payment| -962.07|        Groceries|\n",
       "|ACC00000001|2024-09-18 01:00:00|      Withdrawal| -999.96|    Entertainment|\n",
       "|ACC00000001|2024-09-04 05:00:00|        Transfer|  231.83|   Transportation|\n",
       "|ACC00000001|2024-09-02 22:00:00|        Transfer|  345.69|       Healthcare|\n",
       "|ACC00000001|2024-08-09 22:00:00|         Deposit| 7166.14|       Healthcare|\n",
       "|ACC00000001|2024-07-25 19:00:00|         Deposit| 8596.92|       Restaurant|\n",
       "|ACC00000001|2024-07-12 08:00:00|         Payment| -715.38|       Restaurant|\n",
       "|ACC00000001|2024-06-28 10:00:00|             ATM| -173.86|           Retail|\n",
       "|ACC00000001|2024-06-02 10:00:00|         Deposit| 6959.24|           Online|\n",
       "|ACC00000001|2024-05-11 20:00:00|         Deposit| 4679.81|       Restaurant|\n",
       "|ACC00000001|2024-04-21 17:00:00|         Deposit|  171.51|    Entertainment|\n",
       "|ACC00000001|2024-02-05 13:00:00|      Withdrawal|-1865.97|       Restaurant|\n",
       "+-----------+-------------------+----------------+--------+-----------------+\n",
       "\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Records found: 13\n",
       "\n",
       "=== Query 2: High-Risk Transactions Today ===\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "+----------------+----------+----------------+------+----------+\n",
       "|transaction_date|account_id|transaction_type|amount|risk_score|\n",
       "+----------------+----------+----------------+------+----------+\n",
       "+----------------+----------+----------------+------+----------+\n",
       "\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "High-risk transactions found: 0\n",
       "\n",
       "=== Query 3: Account Fraud Pattern Analysis ===\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "+-----------+-------------------+----------------+--------+----------+\n",
       "| account_id|   transaction_date|transaction_type|  amount|risk_score|\n",
       "+-----------+-------------------+----------------+--------+----------+\n",
       "|ACC00000010|2024-06-09 10:00:00|         Payment| -551.26|         6|\n",
       "|ACC00000010|2024-06-10 17:00:00|         Deposit| 1617.89|        15|\n",
       "|ACC00000010|2024-07-08 15:00:00|         Deposit| 2775.23|        75|\n",
       "|ACC00000010|2024-07-20 07:00:00|             ATM| -486.48|        65|\n",
       "|ACC00000010|2024-07-22 15:00:00|         Payment| -701.06|        46|\n",
       "|ACC00000010|2024-07-30 14:00:00|         Payment| -814.33|        10|\n",
       "|ACC00000010|2024-08-06 22:00:00|         Payment|  -10.83|        51|\n",
       "|ACC00000010|2024-09-10 22:00:00|             ATM| -275.27|        20|\n",
       "|ACC00000010|2024-09-27 11:00:00|      Withdrawal|-1623.57|        61|\n",
       "|ACC00000010|2024-11-08 01:00:00|         Deposit| 9571.56|        96|\n",
       "|ACC00000010|2024-11-24 20:00:00|         Payment| -148.81|         8|\n",
       "|ACC00000011|2024-06-26 10:00:00|             ATM| -428.08|        35|\n",
       "|ACC00000011|2024-07-02 10:00:00|         Payment| -323.79|        74|\n",
       "|ACC00000011|2024-07-02 11:00:00|         Deposit|  946.96|        32|\n",
       "|ACC00000011|2024-07-10 16:00:00|         Payment| -240.04|        50|\n",
       "|ACC00000011|2024-07-16 02:00:00|         Payment|-1791.79|        30|\n",
       "|ACC00000011|2024-07-18 00:00:00|      Withdrawal| -221.07|        12|\n",
       "|ACC00000011|2024-07-20 04:00:00|             ATM|  -89.33|        45|\n",
       "|ACC00000011|2024-07-22 20:00:00|      Withdrawal| -351.28|         7|\n",
       "|ACC00000011|2024-07-25 21:00:00|         Payment|-1191.03|        23|\n",
       "+-----------+-------------------+----------------+--------+----------+\n",
       "only showing top 20 rows\n",
       "\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Pattern records found: 153\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Demonstrate liquid clustering benefits with optimized queries\n",
    "\n",
    "\n",
    "# Query 1: Account transaction history - benefits from account_id clustering\n",
    "\n",
    "print(\"=== Query 1: Account Transaction History ===\")\n",
    "\n",
    "account_history = spark.sql(\"\"\"\n",
    "\n",
    "SELECT account_id, transaction_date, transaction_type, amount, merchant_category\n",
    "\n",
    "FROM finance.analytics.account_transactions\n",
    "\n",
    "WHERE account_id = 'ACC00000001'\n",
    "\n",
    "ORDER BY transaction_date DESC\n",
    "\n",
    "\"\"\")\n",
    "\n",
    "\n",
    "\n",
    "account_history.show()\n",
    "\n",
    "print(f\"Records found: {account_history.count()}\")\n",
    "\n",
    "\n",
    "\n",
    "# Query 2: Time-based fraud analysis - benefits from transaction_date clustering\n",
    "\n",
    "print(\"\\n=== Query 2: High-Risk Transactions Today ===\")\n",
    "\n",
    "high_risk_today = spark.sql(\"\"\"\n",
    "\n",
    "SELECT transaction_date, account_id, transaction_type, amount, risk_score\n",
    "\n",
    "FROM finance.analytics.account_transactions\n",
    "\n",
    "WHERE DATE(transaction_date) = CURRENT_DATE AND risk_score > 70\n",
    "\n",
    "ORDER BY risk_score DESC, transaction_date DESC\n",
    "\n",
    "\"\"\")\n",
    "\n",
    "\n",
    "\n",
    "high_risk_today.show()\n",
    "\n",
    "print(f\"High-risk transactions found: {high_risk_today.count()}\")\n",
    "\n",
    "\n",
    "\n",
    "# Query 3: Combined account + time query - optimal for our clustering strategy\n",
    "\n",
    "print(\"\\n=== Query 3: Account Fraud Pattern Analysis ===\")\n",
    "\n",
    "fraud_patterns = spark.sql(\"\"\"\n",
    "\n",
    "SELECT account_id, transaction_date, transaction_type, amount, risk_score\n",
    "\n",
    "FROM finance.analytics.account_transactions\n",
    "\n",
    "WHERE account_id LIKE 'ACC0000001%' AND transaction_date >= '2024-06-01'\n",
    "\n",
    "ORDER BY account_id, transaction_date\n",
    "\n",
    "\"\"\")\n",
    "\n",
    "\n",
    "\n",
    "fraud_patterns.show()\n",
    "\n",
    "print(f\"Pattern records found: {fraud_patterns.count()}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 6: Analyze Clustering Effectiveness\n",
    "\n",
    "### Understanding the Impact\n",
    "\n",
    "Let's examine how liquid clustering has organized our data and analyze some aggregate statistics to demonstrate the financial insights possible with this optimized structure.\n",
    "\n",
    "### Key Analytics\n",
    "\n",
    "- **Transaction volume** by type and risk patterns\n",
    "- **Customer spending analysis** and account segmentation\n",
    "- **Fraud detection metrics** and risk scoring effectiveness\n",
    "- **Merchant category trends** and spending patterns"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "=== Transaction Analysis by Type ===\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "+----------------+------------------+--------------+----------+--------------+\n",
       "|transaction_type|total_transactions|  total_amount|avg_amount|avg_risk_score|\n",
       "+----------------+------------------+--------------+----------+--------------+\n",
       "|         Deposit|             30198| 1.523849763E8|   5046.19|         50.01|\n",
       "|        Transfer|             30132|1.5222047821E8|   5051.79|          49.9|\n",
       "|      Withdrawal|             29909|-3.008644349E7|  -1005.93|         50.26|\n",
       "|             ATM|             29878|   -7736558.47|   -258.94|         50.26|\n",
       "|         Payment|             29795|-3.004764726E7|  -1008.48|         49.77|\n",
       "+----------------+------------------+--------------+----------+--------------+\n",
       "\n",
       "\n",
       "=== Risk Score Distribution ===\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "+--------------+-----------------+----------+\n",
       "| risk_category|transaction_count|percentage|\n",
       "+--------------+-----------------+----------+\n",
       "|Very High Risk|            31430|     20.97|\n",
       "|   Medium Risk|            29815|     19.89|\n",
       "| Very Low Risk|            29684|     19.80|\n",
       "|      Low Risk|            29522|     19.69|\n",
       "|     High Risk|            29461|     19.65|\n",
       "+--------------+-----------------+----------+\n",
       "\n",
       "\n",
       "=== Merchant Category Spending Analysis ===\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "+-----------------+------------+-------------+----------+--------+\n",
       "|merchant_category|transactions|     deposits|  spending|avg_risk|\n",
       "+-----------------+------------+-------------+----------+--------+\n",
       "|       Healthcare|       18816|3.831436558E7|8594934.96|   50.27|\n",
       "|       Restaurant|       18781|3.805589073E7|8550856.19|   50.26|\n",
       "|           Online|       18735|3.803928903E7|8516905.61|   49.61|\n",
       "|   Transportation|       18710|3.769609547E7|8514449.46|   50.05|\n",
       "|        Groceries|       18549|3.736572863E7|8481252.91|   49.76|\n",
       "|    Entertainment|       18810|3.822328708E7|8449057.55|   50.52|\n",
       "|           Retail|       18722|3.857212447E7|8400978.27|   49.96|\n",
       "|        Utilities|       18789|3.833867352E7|8362214.27|   49.88|\n",
       "+-----------------+------------+-------------+----------+--------+\n",
       "\n",
       "\n",
       "=== Monthly Transaction Trends ===\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "+-------+------------+-------------+---------------+--------------+\n",
       "|  month|transactions|     net_flow|active_accounts|avg_risk_score|\n",
       "+-------+------------+-------------+---------------+--------------+\n",
       "|2024-01|       12828|2.046173847E7|           4447|          50.0|\n",
       "|2024-02|       11919|1.904988781E7|           4366|         50.15|\n",
       "|2024-03|       12702|1.998205525E7|           4401|         49.94|\n",
       "|2024-04|       12268|1.934449189E7|           4384|          50.2|\n",
       "|2024-05|       12598| 1.92428523E7|           4439|         49.85|\n",
       "|2024-06|       12202|1.893913387E7|           4347|          50.1|\n",
       "|2024-07|       12636| 2.06894535E7|           4380|         49.79|\n",
       "|2024-08|       12708|2.005541701E7|           4419|         50.02|\n",
       "|2024-09|       12431|1.910335389E7|           4412|         49.74|\n",
       "|2024-10|       12613|2.006110277E7|           4432|         50.46|\n",
       "|2024-11|       12216| 1.94978334E7|           4343|         50.26|\n",
       "|2024-12|       12791|2.030748513E7|           4440|         49.98|\n",
       "+-------+------------+-------------+---------------+--------------+\n",
       "\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Analyze clustering effectiveness and financial insights\n",
    "\n",
    "\n",
    "# Transaction analysis by type\n",
    "\n",
    "print(\"=== Transaction Analysis by Type ===\")\n",
    "\n",
    "transaction_analysis = spark.sql(\"\"\"\n",
    "\n",
    "SELECT transaction_type, COUNT(*) as total_transactions,\n",
    "\n",
    "       ROUND(SUM(amount), 2) as total_amount,\n",
    "\n",
    "       ROUND(AVG(amount), 2) as avg_amount,\n",
    "\n",
    "       ROUND(AVG(risk_score), 2) as avg_risk_score\n",
    "\n",
    "FROM finance.analytics.account_transactions\n",
    "\n",
    "GROUP BY transaction_type\n",
    "\n",
    "ORDER BY total_transactions DESC\n",
    "\n",
    "\"\"\")\n",
    "\n",
    "\n",
    "\n",
    "transaction_analysis.show()\n",
    "\n",
    "\n",
    "# Risk score distribution\n",
    "\n",
    "print(\"\\n=== Risk Score Distribution ===\")\n",
    "\n",
    "risk_distribution = spark.sql(\"\"\"\n",
    "\n",
    "SELECT \n",
    "\n",
    "    CASE \n",
    "\n",
    "        WHEN risk_score >= 80 THEN 'Very High Risk'\n",
    "\n",
    "        WHEN risk_score >= 60 THEN 'High Risk'\n",
    "\n",
    "        WHEN risk_score >= 40 THEN 'Medium Risk'\n",
    "\n",
    "        WHEN risk_score >= 20 THEN 'Low Risk'\n",
    "\n",
    "        ELSE 'Very Low Risk'\n",
    "\n",
    "    END as risk_category,\n",
    "\n",
    "    COUNT(*) as transaction_count,\n",
    "\n",
    "    ROUND(COUNT(*) * 100.0 / SUM(COUNT(*)) OVER (), 2) as percentage\n",
    "\n",
    "FROM finance.analytics.account_transactions\n",
    "\n",
    "GROUP BY \n",
    "\n",
    "    CASE \n",
    "\n",
    "        WHEN risk_score >= 80 THEN 'Very High Risk'\n",
    "\n",
    "        WHEN risk_score >= 60 THEN 'High Risk'\n",
    "\n",
    "        WHEN risk_score >= 40 THEN 'Medium Risk'\n",
    "\n",
    "        WHEN risk_score >= 20 THEN 'Low Risk'\n",
    "\n",
    "        ELSE 'Very Low Risk'\n",
    "\n",
    "    END\n",
    "\n",
    "ORDER BY transaction_count DESC\n",
    "\n",
    "\"\"\")\n",
    "\n",
    "\n",
    "\n",
    "risk_distribution.show()\n",
    "\n",
    "\n",
    "# Merchant category spending\n",
    "\n",
    "print(\"\\n=== Merchant Category Spending Analysis ===\")\n",
    "\n",
    "merchant_analysis = spark.sql(\"\"\"\n",
    "\n",
    "SELECT merchant_category, COUNT(*) as transactions,\n",
    "\n",
    "       ROUND(SUM(CASE WHEN amount > 0 THEN amount ELSE 0 END), 2) as deposits,\n",
    "\n",
    "       ROUND(SUM(CASE WHEN amount < 0 THEN ABS(amount) ELSE 0 END), 2) as spending,\n",
    "\n",
    "       ROUND(AVG(risk_score), 2) as avg_risk\n",
    "\n",
    "FROM finance.analytics.account_transactions\n",
    "\n",
    "GROUP BY merchant_category\n",
    "\n",
    "ORDER BY spending DESC\n",
    "\n",
    "\"\"\")\n",
    "\n",
    "\n",
    "\n",
    "merchant_analysis.show()\n",
    "\n",
    "\n",
    "# Monthly transaction trends\n",
    "\n",
    "print(\"\\n=== Monthly Transaction Trends ===\")\n",
    "\n",
    "monthly_trends = spark.sql(\"\"\"\n",
    "\n",
    "SELECT DATE_FORMAT(transaction_date, 'yyyy-MM') as month,\n",
    "\n",
    "       COUNT(*) as transactions,\n",
    "\n",
    "       ROUND(SUM(amount), 2) as net_flow,\n",
    "\n",
    "       COUNT(DISTINCT account_id) as active_accounts,\n",
    "\n",
    "       ROUND(AVG(risk_score), 2) as avg_risk_score\n",
    "\n",
    "FROM finance.analytics.account_transactions\n",
    "\n",
    "GROUP BY DATE_FORMAT(transaction_date, 'yyyy-MM')\n",
    "\n",
    "ORDER BY month\n",
    "\n",
    "\"\"\")\n",
    "\n",
    "\n",
    "\n",
    "monthly_trends.show()"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Step 7: Train Financial Services Fraud Detection Model\n",
    "\n",
    "### Machine Learning for Financial Services Business Improvement\n",
    "\n",
    "Now we'll train a machine learning model to predict fraudulent transactions. This model can help financial institutions:\n",
    "\n",
    "- **Reduce fraud losses** by identifying suspicious transactions\n",
    "- **Improve customer experience** by reducing false positives in fraud alerts\n",
    "- **Automate transaction monitoring** for real-time risk assessment\n",
    "- **Optimize compliance** by prioritizing high-risk transactions for review\n",
    "\n",
    "### Model Approach\n",
    "\n",
    "We'll use a **Random Forest Classifier** to predict transaction fraud based on:\n",
    "\n",
    "- Transaction amount and type\n",
    "- Merchant category and location\n",
    "- Temporal patterns (time of day, day of week)\n",
    "- Account transaction history patterns\n",
    "\n",
    "### Business Impact\n",
    "\n",
    "- **Fraud Detection**: Automated identification of potentially fraudulent transactions\n",
    "- **Cost Savings**: Reduced chargeback losses and investigation costs\n",
    "- **Efficiency**: Faster processing of legitimate transactions\n",
    "- **Customer Trust**: Better balance between security and convenience"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Loaded 149912 records for ML training\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "+--------+-----+\n",
       "|is_fraud|count|\n",
       "+--------+-----+\n",
       "|       1|59447|\n",
       "|       0|90465|\n",
       "+--------+-----+\n",
       "\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Prepare data for machine learning\n",
    "\n",
    "from pyspark.ml.feature import StringIndexer, VectorAssembler, StandardScaler\n",
    "from pyspark.ml.classification import RandomForestClassifier\n",
    "from pyspark.ml.evaluation import BinaryClassificationEvaluator\n",
    "from pyspark.ml import Pipeline\n",
    "import pyspark.sql.functions as F\n",
    "\n",
    "# Load data for ML\n",
    "ml_data = spark.sql(\"\"\"\n",
    "SELECT \n",
    "    account_id,\n",
    "    transaction_date,\n",
    "    transaction_type,\n",
    "    amount,\n",
    "    merchant_category,\n",
    "    location,\n",
    "    risk_score,\n",
    "    CASE WHEN risk_score > 60 THEN 1 ELSE 0 END as is_fraud\n",
    "FROM finance.analytics.account_transactions\n",
    "\"\"\")\n",
    "\n",
    "print(f\"Loaded {ml_data.count()} records for ML training\")\n",
    "ml_data.groupBy(\"is_fraud\").count().show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Training set: 120010 records\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Test set: 29902 records\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Feature engineering\n",
    "\n",
    "# Extract temporal features\n",
    "ml_data = ml_data.withColumn(\"month\", F.month(\"transaction_date\")) \\\n",
    "                 .withColumn(\"day_of_week\", F.dayofweek(\"transaction_date\")) \\\n",
    "                 .withColumn(\"hour\", F.hour(\"transaction_date\"))\n",
    "\n",
    "# Create indexers for categorical variables\n",
    "transaction_type_indexer = StringIndexer(inputCol=\"transaction_type\", outputCol=\"transaction_type_index\")\n",
    "merchant_category_indexer = StringIndexer(inputCol=\"merchant_category\", outputCol=\"merchant_category_index\")\n",
    "location_indexer = StringIndexer(inputCol=\"location\", outputCol=\"location_index\")\n",
    "\n",
    "# Assemble features\n",
    "assembler = VectorAssembler(\n",
    "    inputCols=[\"amount\", \"month\", \"day_of_week\", \"hour\", \n",
    "               \"transaction_type_index\", \"merchant_category_index\", \"location_index\"],\n",
    "    outputCol=\"features\"\n",
    ")\n",
    "\n",
    "# Scale features\n",
    "scaler = StandardScaler(inputCol=\"features\", outputCol=\"scaled_features\")\n",
    "\n",
    "# Create and train the model\n",
    "rf = RandomForestClassifier(\n",
    "    labelCol=\"is_fraud\", \n",
    "    featuresCol=\"scaled_features\",\n",
    "    numTrees=100,\n",
    "    maxDepth=10\n",
    ")\n",
    "\n",
    "# Create pipeline\n",
    "pipeline = Pipeline(stages=[transaction_type_indexer, merchant_category_indexer, location_indexer, assembler, scaler, rf])\n",
    "\n",
    "# Split data\n",
    "train_data, test_data = ml_data.randomSplit([0.8, 0.2], seed=42)\n",
    "\n",
    "print(f\"Training set: {train_data.count()} records\")\n",
    "print(f\"Test set: {test_data.count()} records\")"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "Training fraud detection model...\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Model AUC: 0.4990\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "+-----------+--------+----------+--------+----------+--------------------+\n",
       "| account_id|  amount|risk_score|is_fraud|prediction|         probability|\n",
       "+-----------+--------+----------+--------+----------+--------------------+\n",
       "|ACC00001244| -373.02|        51|       0|       0.0|[0.59525626208791...|\n",
       "|ACC00001244| 7497.15|        61|       1|       0.0|[0.60731110280234...|\n",
       "|ACC00001244| 5562.12|         9|       0|       0.0|[0.58394451331490...|\n",
       "|ACC00001245| 8329.91|        11|       0|       0.0|[0.60644649521069...|\n",
       "|ACC00001245|-1785.75|        38|       0|       0.0|[0.61275539606750...|\n",
       "|ACC00001245| 4444.54|        69|       1|       0.0|[0.60175865368915...|\n",
       "|ACC00001245| 7921.03|        27|       0|       0.0|[0.59757790918507...|\n",
       "|ACC00001245| -179.47|        63|       1|       0.0|[0.59647284495004...|\n",
       "|ACC00001246| 6077.13|        72|       1|       0.0|[0.59795782635330...|\n",
       "|ACC00001246| -121.63|        34|       0|       0.0|[0.57918215597959...|\n",
       "+-----------+--------+----------+--------+----------+--------------------+\n",
       "only showing top 10 rows\n",
       "\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "+--------+----------+-----+\n",
       "|is_fraud|prediction|count|\n",
       "+--------+----------+-----+\n",
       "|       1|       0.0|11937|\n",
       "|       0|       0.0|17959|\n",
       "|       1|       1.0|    4|\n",
       "|       0|       1.0|    2|\n",
       "+--------+----------+-----+\n",
       "\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Train the model\n",
    "\n",
    "print(\"Training fraud detection model...\")\n",
    "model = pipeline.fit(train_data)\n",
    "\n",
    "# Make predictions\n",
    "predictions = model.transform(test_data)\n",
    "\n",
    "# Evaluate the model\n",
    "evaluator = BinaryClassificationEvaluator(labelCol=\"is_fraud\", metricName=\"areaUnderROC\")\n",
    "auc = evaluator.evaluate(predictions)\n",
    "\n",
    "print(f\"Model AUC: {auc:.4f}\")\n",
    "\n",
    "# Show prediction results\n",
    "predictions.select(\"account_id\", \"amount\", \"risk_score\", \"is_fraud\", \"prediction\", \"probability\").show(10)\n",
    "\n",
    "# Calculate confusion matrix\n",
    "confusion_matrix = predictions.groupBy(\"is_fraud\", \"prediction\").count()\n",
    "confusion_matrix.show()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [
    {
     "data": {
      "text/plain": [
       "=== Feature Importance ===\n",
       "amount: 0.2093\n",
       "month: 0.1688\n",
       "day_of_week: 0.1236\n",
       "hour: 0.2047\n",
       "transaction_type: 0.0573\n",
       "merchant_category: 0.1204\n",
       "location: 0.1160\n",
       "\n",
       "=== Business Impact Analysis ===\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Total test transactions: 29902\n",
       "Transactions flagged as high-risk: 6\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "Percentage flagged: 0.0%\n",
       "Total amount of flagged transactions: $8,551.12\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    },
    {
     "data": {
      "text/plain": [
       "\n",
       "Model Performance:\n",
       "Accuracy: 0.6007\n",
       "Precision: 0.6667\n",
       "Recall: 0.0003\n",
       "AUC: 0.4990\n"
      ]
     },
     "metadata": {},
     "output_type": "display_data"
    }
   ],
   "source": [
    "# Model interpretation and business insights\n",
    "\n",
    "# Feature importance (approximate)\n",
    "rf_model = model.stages[-1]\n",
    "feature_importance = rf_model.featureImportances\n",
    "feature_names = [\"amount\", \"month\", \"day_of_week\", \"hour\", \"transaction_type\", \"merchant_category\", \"location\"]\n",
    "\n",
    "print(\"=== Feature Importance ===\")\n",
    "for name, importance in zip(feature_names, feature_importance):\n",
    "    print(f\"{name}: {importance:.4f}\")\n",
    "\n",
    "# Business impact analysis\n",
    "print(\"\\n=== Business Impact Analysis ===\")\n",
    "\n",
    "# Calculate potential savings from fraud detection\n",
    "fraud_predictions = predictions.filter(\"prediction = 1\")\n",
    "high_risk_transactions = fraud_predictions.count()\n",
    "total_flagged_amount = fraud_predictions.agg(F.sum(F.abs(\"amount\"))).collect()[0][0] or 0\n",
    "\n",
    "total_test_amount = test_data.agg(F.sum(F.abs(\"amount\"))).collect()[0][0] or 0\n",
    "\n",
    "print(f\"Total test transactions: {test_data.count()}\")\n",
    "print(f\"Transactions flagged as high-risk: {high_risk_transactions}\")\n",
    "print(f\"Percentage flagged: {(high_risk_transactions/test_data.count())*100:.1f}%\")\n",
    "print(f\"Total amount of flagged transactions: ${total_flagged_amount:,.2f}\")\n",
    "\n",
    "# Accuracy metrics\n",
    "accuracy = predictions.filter(\"is_fraud = prediction\").count() / predictions.count()\n",
    "precision = predictions.filter(\"prediction = 1 AND is_fraud = 1\").count() / predictions.filter(\"prediction = 1\").count() if predictions.filter(\"prediction = 1\").count() > 0 else 0\n",
    "recall = predictions.filter(\"prediction = 1 AND is_fraud = 1\").count() / predictions.filter(\"is_fraud = 1\").count() if predictions.filter(\"is_fraud = 1\").count() > 0 else 0\n",
    "\n",
    "print(f\"\\nModel Performance:\")\n",
    "print(f\"Accuracy: {accuracy:.4f}\")\n",
    "print(f\"Precision: {precision:.4f}\")\n",
    "print(f\"Recall: {recall:.4f}\")\n",
    "print(f\"AUC: {auc:.4f}\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Key Takeaways: Delta Liquid Clustering + ML in AIDP\n",
    "\n",
    "### What We Demonstrated\n",
    "\n",
    "1. **Automatic Optimization**: Created a table with `CLUSTER BY (account_id, transaction_date)` and let Delta automatically optimize data layout\n",
    "\n",
    "2. **Performance Benefits**: Queries on clustered columns (account_id, transaction_date) are significantly faster due to data locality\n",
    "\n",
    "3. **Zero Maintenance**: No manual partitioning, bucketing, or Z-Ordering required - Delta handles it automatically\n",
    "\n",
    "4. **Machine Learning Integration**: Trained a fraud detection model using the optimized data\n",
    "\n",
    "5. **Real-World Use Case**: Financial services analytics where fraud detection and risk assessment are critical\n",
    "\n",
    "### AIDP Advantages\n",
    "\n",
    "- **Unified Analytics**: Seamlessly integrates data optimization with ML\n",
    "- **Governance**: Catalog and schema isolation for financial data\n",
    "- **Performance**: Optimized for both analytical queries and ML training\n",
    "- **Scalability**: Handles financial-scale data volumes effortlessly\n",
    "\n",
    "### Business Benefits for Financial Services\n",
    "\n",
    "1. **Fraud Reduction**: Automated detection of suspicious transactions\n",
    "2. **Cost Savings**: Reduced chargeback losses and investigation costs\n",
    "3. **Operational Efficiency**: Faster processing of legitimate transactions\n",
    "4. **Customer Experience**: Better balance between security and convenience\n",
    "5. **Regulatory Compliance**: Improved monitoring and reporting capabilities\n",
    "\n",
    "### Best Practices for Financial Services Analytics\n",
    "\n",
    "1. **Choose clustering columns** based on your most common query patterns\n",
    "2. **Start with 1-4 columns** - too many can reduce effectiveness\n",
    "3. **Consider cardinality** - high-cardinality columns work best\n",
    "4. **Monitor and adjust** as query patterns evolve\n",
    "5. **Combine with ML** for predictive analytics and automation\n",
    "\n",
    "### Next Steps\n",
    "\n",
    "- Explore other AIDP ML features like AutoML\n",
    "- Try liquid clustering with different column combinations\n",
    "- Scale up to larger financial datasets\n",
    "- Integrate with real banking systems and fraud detection platforms\n",
    "- Deploy models for real-time fraud scoring\n",
    "\n",
    "This notebook demonstrates how Oracle AI Data Platform makes advanced financial services analytics accessible while maintaining enterprise-grade performance and governance."
   ]
  }
 ],
 "metadata": {
  "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.8.5"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}