🏛️ MICS Compliance Template¶

from pyspark.sql import SparkSession
from pyspark.sql.functions import *

spark = SparkSession.builder.getOrCreate()

def daily_machine_reconciliation(report_date: str):
    """
    Generate daily machine reconciliation report per MICS 542.19.

    Compares:
    - Opening meters vs closing meters
    - Calculated vs reported hold
    - Documented fills and jackpots
    """

    # Get meter readings
    df_meters = spark.table("bronze_machine_meters") \
        .filter(to_date("reading_timestamp") == report_date)

    # Get activity from telemetry
    df_activity = spark.table("silver_slot_telemetry") \
        .filter(to_date("event_timestamp") == report_date) \
        .groupBy("machine_id") \
        .agg(
            sum("bet_amount").alias("calculated_coin_in"),
            sum("win_amount").alias("calculated_coin_out"),
            count("*").alias("total_events"),
            sum(when(col("event_type") == "JACKPOT", col("win_amount")).otherwise(0)).alias("jackpot_total")
        )

    # Get physical counts
    df_drops = spark.table("bronze_drop_counts") \
        .filter(to_date("count_timestamp") == report_date) \
        .select("machine_id", "physical_drop_amount")

    # Reconciliation
    df_reconciliation = df_meters.alias("m") \
        .join(df_activity.alias("a"), "machine_id", "left") \
        .join(df_drops.alias("d"), "machine_id", "left") \
        .select(
            col("m.machine_id"),
            col("m.opening_coin_in").alias("meter_opening_coin_in"),
            col("m.closing_coin_in").alias("meter_closing_coin_in"),
            (col("m.closing_coin_in") - col("m.opening_coin_in")).alias("meter_coin_in_diff"),
            col("a.calculated_coin_in"),
            col("a.calculated_coin_out"),
            col("d.physical_drop_amount"),
            # Variance calculations
            (col("a.calculated_coin_in") - (col("m.closing_coin_in") - col("m.opening_coin_in"))).alias("coin_in_variance"),
            # Hold percentage
            ((col("a.calculated_coin_in") - col("a.calculated_coin_out")) / col("a.calculated_coin_in") * 100).alias("actual_hold_pct"),
            col("m.theoretical_hold_pct"),
            # Variance flag
            when(
                abs(col("a.calculated_coin_in") - (col("m.closing_coin_in") - col("m.opening_coin_in"))) > 100,
                "VARIANCE_ALERT"
            ).otherwise("OK").alias("status")
        )

    return df_reconciliation

# Generate report
reconciliation = daily_machine_reconciliation("2024-01-15")
reconciliation.show()

# Flag variances for review
variances = reconciliation.filter(col("status") == "VARIANCE_ALERT")
print(f"Machines with variances: {variances.count()}")

def verify_drop_compliance(report_date: str):
    """
    Verify drop and count procedures per MICS 542.21.

    Checks:
    - All scheduled drops completed
    - Dual signature on count sheets
    - Variance within tolerance
    """

    # Scheduled drops
    df_schedule = spark.table("bronze_drop_schedule") \
        .filter(col("scheduled_date") == report_date)

    # Actual drops
    df_actual = spark.table("bronze_drop_counts") \
        .filter(to_date("count_timestamp") == report_date)

    # Compare
    df_compliance = df_schedule.alias("s") \
        .join(df_actual.alias("a"),
              (col("s.machine_id") == col("a.machine_id")) &
              (col("s.drop_type") == col("a.drop_type")),
              "left") \
        .select(
            col("s.machine_id"),
            col("s.drop_type"),
            col("s.scheduled_time"),
            col("a.count_timestamp").alias("actual_time"),
            col("a.counter_1_id"),
            col("a.counter_2_id"),
            col("a.physical_count"),
            col("a.system_expected"),
            (col("a.physical_count") - col("a.system_expected")).alias("variance"),
            # Compliance checks
            when(col("a.count_timestamp").isNull(), "MISSED_DROP")
                .when(col("a.counter_2_id").isNull(), "MISSING_DUAL_CONTROL")
                .when(abs(col("a.physical_count") - col("a.system_expected")) > 50, "VARIANCE")
                .otherwise("COMPLIANT").alias("compliance_status")
        )

    return df_compliance

def generate_mics_compliance_report(report_month: str):
    """
    Generate monthly MICS compliance summary for audit.
    """

    report = {
        "report_period": report_month,
        "generated_at": datetime.now().isoformat(),
        "sections": {}
    }

    # Section 542.19 - Gaming Machines
    machine_compliance = spark.sql(f"""
        SELECT
            COUNT(DISTINCT machine_id) as total_machines,
            SUM(CASE WHEN daily_reconciled = TRUE THEN 1 ELSE 0 END) as reconciled_days,
            AVG(ABS(variance_pct)) as avg_variance_pct,
            MAX(ABS(variance_pct)) as max_variance_pct
        FROM gold.machine_daily_compliance
        WHERE report_month = '{report_month}'
    """).collect()[0]

    report["sections"]["542.19_gaming_machines"] = {
        "total_machines": machine_compliance.total_machines,
        "reconciliation_rate": machine_compliance.reconciled_days / 30 * 100,
        "avg_variance": machine_compliance.avg_variance_pct,
        "status": "COMPLIANT" if machine_compliance.avg_variance_pct < 0.5 else "REVIEW"
    }

    # Section 542.21 - Drop/Count
    drop_compliance = spark.sql(f"""
        SELECT
            COUNT(*) as total_drops,
            SUM(CASE WHEN dual_control = TRUE THEN 1 ELSE 0 END) as dual_control_count,
            SUM(CASE WHEN variance_within_tolerance = TRUE THEN 1 ELSE 0 END) as within_tolerance
        FROM gold.drop_count_compliance
        WHERE report_month = '{report_month}'
    """).collect()[0]

    report["sections"]["542.21_drop_count"] = {
        "total_drops": drop_compliance.total_drops,
        "dual_control_rate": drop_compliance.dual_control_count / drop_compliance.total_drops * 100,
        "tolerance_rate": drop_compliance.within_tolerance / drop_compliance.total_drops * 100,
        "status": "COMPLIANT" if drop_compliance.dual_control_count == drop_compliance.total_drops else "NON-COMPLIANT"
    }

    # Section 542.31 - Surveillance
    surveillance_compliance = spark.sql(f"""
        SELECT
            AVG(uptime_pct) as avg_uptime,
            MIN(retention_days) as min_retention,
            SUM(CASE WHEN coverage_complete = TRUE THEN 1 ELSE 0 END) as full_coverage_days
        FROM gold.surveillance_compliance
        WHERE report_month = '{report_month}'
    """).collect()[0]

    report["sections"]["542.31_surveillance"] = {
        "avg_uptime": surveillance_compliance.avg_uptime,
        "min_retention_days": surveillance_compliance.min_retention,
        "full_coverage_rate": surveillance_compliance.full_coverage_days / 30 * 100,
        "status": "COMPLIANT" if surveillance_compliance.min_retention >= 7 else "NON-COMPLIANT"
    }

    return report