📅 Day 2: Transformations & Real-Time Analytics¶

# Notebook: 02_silver_player_master

from delta.tables import DeltaTable
from pyspark.sql.functions import *

# Read new data from Bronze
df_new = spark.table("lh_bronze.bronze_player_profile") \
    .withColumn("effective_date", current_date()) \
    .withColumn("end_date", lit(None).cast("date")) \
    .withColumn("is_current", lit(True))

# Check if Silver table exists
if spark.catalog.tableExists("lh_silver.silver_player_master"):
    # SCD Type 2 merge logic
    silver_table = DeltaTable.forName(spark, "lh_silver.silver_player_master")

    # Get current records
    df_current = silver_table.toDF().filter(col("is_current") == True)

    # Find changes (compare on key attributes)
    df_changes = df_new.alias("new") \
        .join(df_current.alias("current"), "player_id", "left") \
        .filter(
            (col("current.player_id").isNull()) |  # New player
            (col("new.loyalty_tier") != col("current.loyalty_tier")) |  # Tier change
            (col("new.email") != col("current.email")) |  # Email change
            (col("new.address") != col("current.address"))  # Address change
        )

    if df_changes.count() > 0:
        # Close current records
        silver_table.update(
            condition = (col("player_id").isin(df_changes.select("player_id").collect())) &
                       (col("is_current") == True),
            set = {
                "end_date": current_date(),
                "is_current": lit(False)
            }
        )

        # Insert new versions
        df_changes.select(df_new.columns).write \
            .format("delta") \
            .mode("append") \
            .saveAsTable("lh_silver.silver_player_master")
else:
    # Initial load
    df_new.write \
        .format("delta") \
        .mode("overwrite") \
        .saveAsTable("lh_silver.silver_player_master")

# Verification
spark.sql("""
    SELECT player_id, COUNT(*) as versions
    FROM lh_silver.silver_player_master
    GROUP BY player_id
    HAVING COUNT(*) > 1
    ORDER BY versions DESC
    LIMIT 10
""").show()

# Notebook: 03_silver_financial_reconciled

# Read Bronze financial data
df_txn = spark.table("lh_bronze.bronze_financial_txn")

# Reconciliation rules
df_reconciled = df_txn \
    .withColumn("reconciliation_status",
        when(col("source_amount") == col("destination_amount"), "MATCHED")
        .when(abs(col("source_amount") - col("destination_amount")) < 0.01, "MATCHED_ROUNDING")
        .otherwise("VARIANCE")) \
    .withColumn("variance_amount",
        col("source_amount") - col("destination_amount")) \
    .withColumn("variance_pct",
        when(col("source_amount") > 0,
             (col("variance_amount") / col("source_amount")) * 100)
        .otherwise(0))

# Flag high-risk transactions
df_silver = df_reconciled \
    .withColumn("risk_flags",
        array_distinct(
            filter(
                array(
                    when(col("amount") >= 10000, lit("CTR_REQUIRED")),
                    when(col("amount").between(8000, 9999), lit("NEAR_CTR")),
                    when(col("variance_pct") > 1, lit("HIGH_VARIANCE")),
                    when(col("transaction_type") == "MARKER", lit("CREDIT_TRANSACTION"))
                ),
                lambda x: x.isNotNull()
            )
        ))

df_silver.write \
    .format("delta") \
    .mode("overwrite") \
    .partitionBy("transaction_date") \
    .saveAsTable("lh_silver.silver_financial_reconciled")

# Reconciliation summary
spark.sql("""
    SELECT
        reconciliation_status,
        COUNT(*) as transactions,
        SUM(amount) as total_amount,
        AVG(variance_pct) as avg_variance_pct
    FROM lh_silver.silver_financial_reconciled
    GROUP BY reconciliation_status
""").show()

# Notebook: 01_gold_slot_performance

# Read Silver slot data
df_silver = spark.table("lh_silver.silver_slot_cleansed")

# Daily aggregations by machine
df_daily = df_silver \
    .withColumn("business_date", to_date("event_timestamp")) \
    .groupBy("machine_id", "zone", "denomination", "business_date") \
    .agg(
        sum("coin_in").alias("total_coin_in"),
        sum("coin_out").alias("total_coin_out"),
        sum("games_played").alias("total_games"),
        sum(when(col("event_type") == "JACKPOT", col("jackpot_amount"))).alias("jackpot_payouts"),
        countDistinct("player_id").alias("unique_players"),
        count("*").alias("total_events"),
        avg("_dq_score").alias("avg_data_quality")
    )

# Calculate KPIs
df_gold = df_daily \
    .withColumn("net_win", col("total_coin_in") - col("total_coin_out")) \
    .withColumn("actual_hold_pct",
        when(col("total_coin_in") > 0,
             (col("net_win") / col("total_coin_in")) * 100)
        .otherwise(0)) \
    .withColumn("avg_bet",
        when(col("total_games") > 0,
             col("total_coin_in") / col("total_games"))
        .otherwise(0)) \
    .withColumn("win_per_unit",
        col("net_win") / 1)  # Per machine per day

# Add theoretical calculations (example: 8% theoretical hold)
df_gold = df_gold \
    .withColumn("theoretical_win", col("total_coin_in") * 0.08) \
    .withColumn("hold_variance", col("net_win") - col("theoretical_win")) \
    .withColumn("hold_variance_pct",
        when(col("theoretical_win") > 0,
             ((col("net_win") - col("theoretical_win")) / col("theoretical_win")) * 100)
        .otherwise(0))

# Write to Gold
df_gold.write \
    .format("delta") \
    .mode("overwrite") \
    .partitionBy("business_date") \
    .saveAsTable("lh_gold.gold_slot_performance")

# OPTIMIZE for Direct Lake
spark.sql("OPTIMIZE lh_gold.gold_slot_performance ZORDER BY (machine_id)")

# Notebook: 02_gold_player_360

# Get current player records
df_players = spark.table("lh_silver.silver_player_master") \
    .filter(col("is_current") == True)

# Aggregate gaming activity
df_slot_activity = spark.table("lh_silver.silver_slot_cleansed") \
    .groupBy("player_id") \
    .agg(
        sum("coin_in").alias("slot_coin_in"),
        sum("coin_out").alias("slot_coin_out"),
        count("*").alias("slot_games_played"),
        max("event_timestamp").alias("last_slot_play")
    )

df_table_activity = spark.table("lh_silver.silver_table_enriched") \
    .groupBy("player_id") \
    .agg(
        sum("buy_in").alias("table_buy_in"),
        sum("cash_out").alias("table_cash_out"),
        sum("hours_played").alias("table_hours_played"),
        max("session_end").alias("last_table_play")
    )

df_financial = spark.table("lh_silver.silver_financial_reconciled") \
    .groupBy("player_id") \
    .agg(
        count("*").alias("total_transactions"),
        sum(when(col("transaction_type") == "CASH_IN", col("amount"))).alias("total_cash_in"),
        sum(when(col("transaction_type") == "MARKER", col("amount"))).alias("total_markers"),
        max("transaction_timestamp").alias("last_transaction")
    )

# Join all activities
df_360 = df_players \
    .join(df_slot_activity, "player_id", "left") \
    .join(df_table_activity, "player_id", "left") \
    .join(df_financial, "player_id", "left")

# Calculate metrics
df_gold = df_360 \
    .withColumn("total_gaming_activity",
        coalesce(col("slot_coin_in"), lit(0)) + coalesce(col("table_buy_in"), lit(0))) \
    .withColumn("total_theo_win",
        (coalesce(col("slot_coin_in"), lit(0)) * 0.08) +
        (coalesce(col("table_hours_played"), lit(0)) * 50)) \
    .withColumn("last_visit",
        greatest(col("last_slot_play"), col("last_table_play"), col("last_transaction"))) \
    .withColumn("days_since_visit",
        datediff(current_date(), col("last_visit"))) \
    .withColumn("churn_risk",
        when(col("days_since_visit") > 90, "High")
        .when(col("days_since_visit") > 30, "Medium")
        .otherwise("Low")) \
    .withColumn("player_value_score",
        (coalesce(col("total_theo_win"), lit(0)) / 1000) +
        (coalesce(col("total_visits"), lit(0)) * 2))

df_gold.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("lh_gold.gold_player_360")

# Notebook: 03_gold_compliance_reporting

df_compliance = spark.table("lh_silver.silver_compliance_validated")
df_financial = spark.table("lh_silver.silver_financial_reconciled")

# Daily compliance summary
df_gold = df_compliance \
    .withColumn("report_date", to_date("filing_timestamp")) \
    .groupBy("report_date") \
    .agg(
        sum(when(col("filing_type") == "CTR", 1).otherwise(0)).alias("ctr_count"),
        sum(when(col("filing_type") == "SAR", 1).otherwise(0)).alias("sar_count"),
        sum(when(col("filing_type") == "W2G", 1).otherwise(0)).alias("w2g_count"),
        sum(when(col("filing_type") == "CTR", col("amount"))).alias("ctr_total_amount"),
        sum(when(col("filing_type") == "W2G", col("amount"))).alias("w2g_total_amount")
    )

# Add potential CTRs from financial (not yet filed)
df_potential_ctr = df_financial \
    .filter(col("amount") >= 10000) \
    .filter(col("ctr_filed") == False) \
    .groupBy(to_date("transaction_timestamp").alias("report_date")) \
    .agg(count("*").alias("pending_ctrs"))

df_gold = df_gold.join(df_potential_ctr, "report_date", "left")

df_gold.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("lh_gold.gold_compliance_reporting")

# streaming_producer.py
import json
import time
import random
from datetime import datetime
from azure.eventhub import EventHubProducerClient, EventData

# For demo: using local file stream
# Production: use Event Hub connection string

def generate_event():
    return {
        "machine_id": f"SLOT-{random.randint(1000, 9999)}",
        "event_type": random.choices(
            ["GAME_PLAY", "JACKPOT", "DOOR_OPEN", "METER_UPDATE"],
            weights=[85, 2, 3, 10]
        )[0],
        "event_timestamp": datetime.utcnow().isoformat(),
        "coin_in": round(random.uniform(1, 100), 2),
        "coin_out": round(random.uniform(0, 80), 2),
        "games_played": random.randint(1, 5),
        "jackpot_amount": round(random.uniform(100, 50000), 2) if random.random() < 0.02 else 0,
        "zone": random.choice(["North", "South", "East", "West", "VIP"]),
        "denomination": random.choice([0.01, 0.05, 0.25, 1.00, 5.00]),
        "player_id": f"P{random.randint(10000, 99999)}" if random.random() > 0.3 else None
    }

# Generate events
while True:
    event = generate_event()
    print(json.dumps(event))
    time.sleep(0.1)  # 10 events per second