💰 Currency Transaction Report Template¶

from pyspark.sql.functions import *
from pyspark.sql.window import Window
from datetime import datetime, timedelta

def extract_ctr_candidates(process_date: str = None):
    """
    Extract CTR candidates for a given date.

    Args:
        process_date: Date to process (YYYY-MM-DD), defaults to yesterday
    """
    if process_date is None:
        process_date = (datetime.now() - timedelta(days=1)).strftime("%Y-%m-%d")

    # Read financial transactions
    df_transactions = spark.table("bronze_financial_transactions") \
        .filter(col("transaction_type").isin(["CASH_IN", "CASH_OUT"])) \
        .filter(to_date("transaction_timestamp") == process_date)

    # Aggregate by player and date
    df_daily = df_transactions.groupBy(
        "player_id",
        to_date("transaction_timestamp").alias("transaction_date")
    ).agg(
        sum(when(col("transaction_type") == "CASH_IN", col("amount")).otherwise(0)).alias("cash_in"),
        sum(when(col("transaction_type") == "CASH_OUT", col("amount")).otherwise(0)).alias("cash_out"),
        sum("amount").alias("total_cash"),
        count("*").alias("transaction_count")
    )

    # Filter CTR threshold
    df_ctr = df_daily.filter(col("total_cash") >= 10000)

    # Join with player dimension
    df_player = spark.table("dim_player") \
        .filter(col("is_current") == True) \
        .select(
            "player_id", "first_name", "last_name", "date_of_birth",
            "address_line1", "city", "state", "zip_code",
            "id_type", "id_state"
        )

    df_ctr_report = df_ctr.join(df_player, "player_id", "left")

    # Add metadata
    df_ctr_report = df_ctr_report \
        .withColumn("filing_status", lit("CTR_REQUIRED")) \
        .withColumn("report_generated_at", current_timestamp()) \
        .withColumn("filing_deadline", date_add(col("transaction_date"), 15))

    return df_ctr_report

# Execute
ctr_data = extract_ctr_candidates()
ctr_data.write.format("delta").mode("append").saveAsTable("compliance_ctr_filings")
print(f"CTR candidates extracted: {ctr_data.count()}")