🏦 Tutorial 33: SBA Small Business Analytics¶

# Generate synthetic SBA loan data for development
# Reference: data_generation/generators/federal/sba_generator.py

import sys
sys.path.append("../../data_generation")

from generators.federal.sba_generator import SBAGenerator

# Initialize generator
generator = SBAGenerator(
    output_path="/lakehouse/default/Files/raw/sba/",
    num_records=50000,         # 50K loans for development
    programs=["PPP", "7A", "504", "SBIR"],
    date_range=("2020-01-01", "2024-12-31"),
    seed=42
)

# Generate data
df_generated = generator.generate_batch()

print(f"Generated {df_generated.count():,} SBA loan records")
print(f"Program distribution:")
df_generated.groupBy("program_type").count().show()

# Download real SBA data from data.sba.gov
# PPP data: https://data.sba.gov/dataset/ppp-foia
# 7(a)/504: https://data.sba.gov/dataset/7-a-504-foia

import requests
import os

SBA_DATA_URLS = {
    "ppp_150k_plus": "https://data.sba.gov/dataset/ppp-foia/resource/ppp-150k-plus.csv",
    "ppp_under_150k": "https://data.sba.gov/dataset/ppp-foia/resource/ppp-under-150k.csv",
    "sba_7a_504": "https://data.sba.gov/dataset/7-a-504-foia/resource/7a-504-summary.csv",
}

download_path = "/lakehouse/default/Files/raw/sba/downloads/"
os.makedirs(download_path, exist_ok=True)

for name, url in SBA_DATA_URLS.items():
    print(f"Downloading {name}...")
    # Note: PPP files are large (2GB+). Use streaming download.
    response = requests.get(url, stream=True)
    file_path = os.path.join(download_path, f"{name}.csv")

    with open(file_path, "wb") as f:
        for chunk in response.iter_content(chunk_size=8192):
            f.write(chunk)

    file_size_mb = os.path.getsize(file_path) / (1024 * 1024)
    print(f"  Saved: {file_path} ({file_size_mb:.1f} MB)")

# Unified SBA loan schema for Bronze layer
from pyspark.sql.types import *

sba_loan_schema = StructType([
    # Core identifiers
    StructField("loan_id", StringType(), False),             # SBA loan number
    StructField("program_type", StringType(), False),         # PPP, 7A, 504, SBIR
    StructField("approval_date", DateType(), True),           # Loan approval date
    StructField("disbursement_date", DateType(), True),       # Funds disbursed date

    # Borrower information
    StructField("business_name", StringType(), True),         # Borrower business name
    StructField("business_type", StringType(), True),         # Corporation, LLC, Sole Prop, etc.
    StructField("naics_code", StringType(), True),            # 6-digit NAICS industry code
    StructField("naics_description", StringType(), True),     # Industry description
    StructField("business_age", StringType(), True),          # New, Existing, Startup
    StructField("jobs_reported", IntegerType(), True),        # Jobs supported/retained

    # Geography
    StructField("borrower_city", StringType(), True),
    StructField("borrower_state", StringType(), True),        # 2-letter state code
    StructField("borrower_zip", StringType(), True),          # 5-digit ZIP
    StructField("congressional_district", StringType(), True), # State-District (e.g., CA-12)
    StructField("rural_urban", StringType(), True),            # Rural, Urban, Suburban

    # Loan details
    StructField("loan_amount", DoubleType(), True),           # Total loan amount ($)
    StructField("sba_guarantee_amount", DoubleType(), True),  # SBA guarantee portion ($)
    StructField("sba_guarantee_pct", DoubleType(), True),     # Guarantee percentage
    StructField("term_months", IntegerType(), True),          # Loan term in months
    StructField("interest_rate", DoubleType(), True),         # Interest rate (%)
    StructField("loan_status", StringType(), True),           # Active, PIF, Default, Charged Off

    # Lender information
    StructField("lender_name", StringType(), True),           # Lending institution
    StructField("lender_id", StringType(), True),             # SBA lender ID
    StructField("lender_city", StringType(), True),
    StructField("lender_state", StringType(), True),

    # PPP-specific fields
    StructField("forgiveness_amount", DoubleType(), True),    # PPP forgiveness amount
    StructField("forgiveness_status", StringType(), True),    # Full, Partial, Denied, Pending
    StructField("forgiveness_date", DateType(), True),
    StructField("ppp_draw", StringType(), True),              # First Draw, Second Draw

    # SBIR-specific fields
    StructField("sbir_phase", StringType(), True),            # Phase I, Phase II
    StructField("awarding_agency", StringType(), True),       # Federal agency
    StructField("research_topic", StringType(), True),

    # 504-specific fields
    StructField("cdc_name", StringType(), True),              # Certified Development Company
    StructField("project_cost", DoubleType(), True),          # Total project cost
    StructField("debenture_amount", DoubleType(), True),      # SBA debenture
])

print(f"Schema fields: {len(sba_loan_schema.fields)}")
print(f"Required fields: {sum(1 for f in sba_loan_schema.fields if not f.nullable)}")

# Bronze ingestion: unified pipeline for all SBA program types
# Handles both synthetic generator output and open data CSV downloads

from pyspark.sql.functions import *
from datetime import datetime

BATCH_ID = f"sba-batch-{datetime.utcnow().strftime('%Y%m%d-%H%M%S')}"
RUN_ID = str(uuid4())

# Detect source format and read
source_path = "/lakehouse/default/Files/raw/sba/"

# Read Parquet (from generator)
df_parquet = spark.read \
    .schema(sba_loan_schema) \
    .parquet(f"{source_path}*.parquet") \
    if len(mssparkutils.fs.ls(f"{source_path}*.parquet")) > 0 else None

# Read CSV (from open data download)
df_csv = spark.read \
    .option("header", "true") \
    .option("inferSchema", "false") \
    .csv(f"{source_path}downloads/*.csv") \
    if os.path.exists("/lakehouse/default/Files/raw/sba/downloads/") else None

# Union available sources
sources = [df for df in [df_parquet, df_csv] if df is not None]
df_raw = sources[0]
for df in sources[1:]:
    df_raw = df_raw.unionByName(df, allowMissingColumns=True)

print(f"Raw records loaded: {df_raw.count():,}")
print(f"Program types: {[r.program_type for r in df_raw.select('program_type').distinct().collect()]}")

# Add Bronze metadata and write
df_bronze = df_raw \
    .withColumn("_ingested_at", current_timestamp()) \
    .withColumn("_source_file", input_file_name()) \
    .withColumn("_batch_id", lit(BATCH_ID)) \
    .withColumn("_run_id", lit(RUN_ID)) \
    .withColumn("_load_date", current_timestamp().cast("date"))

# Write to Delta table (append-only)
df_bronze.write \
    .format("delta") \
    .mode("append") \
    .option("mergeSchema", "true") \
    .partitionBy("program_type", "_load_date") \
    .saveAsTable("lh_bronze.bronze_sba_loans")

# Verification
count = spark.table("lh_bronze.bronze_sba_loans").count()
print(f"Bronze table: {count:,} total records")

# Program distribution
spark.sql("""
    SELECT program_type, COUNT(*) as record_count,
           ROUND(SUM(loan_amount), 0) as total_amount
    FROM lh_bronze.bronze_sba_loans
    GROUP BY program_type
    ORDER BY record_count DESC
""").show()

# NAICS code validation and enrichment
NAICS_PATTERN = r"^\d{6}$"

# NAICS sector mapping (first 2 digits)
NAICS_SECTORS = {
    "11": "Agriculture, Forestry, Fishing",
    "21": "Mining, Quarrying, Oil/Gas",
    "22": "Utilities",
    "23": "Construction",
    "31": "Manufacturing", "32": "Manufacturing", "33": "Manufacturing",
    "42": "Wholesale Trade",
    "44": "Retail Trade", "45": "Retail Trade",
    "48": "Transportation/Warehousing", "49": "Transportation/Warehousing",
    "51": "Information",
    "52": "Finance and Insurance",
    "53": "Real Estate",
    "54": "Professional/Scientific/Technical",
    "55": "Management of Companies",
    "56": "Admin/Support/Waste Management",
    "61": "Educational Services",
    "62": "Health Care and Social Assistance",
    "71": "Arts, Entertainment, Recreation",
    "72": "Accommodation and Food Services",
    "81": "Other Services",
    "92": "Public Administration",
}

sector_expr = create_map([lit(x) for pair in NAICS_SECTORS.items() for x in pair])

df_naics = df_bronze \
    .withColumn("naics_code_clean", regexp_replace(col("naics_code"), "[^0-9]", "")) \
    .withColumn("naics_valid", col("naics_code_clean").rlike(NAICS_PATTERN)) \
    .withColumn("naics_sector_code", substring(col("naics_code_clean"), 1, 2)) \
    .withColumn("naics_sector_name",
        coalesce(sector_expr[col("naics_sector_code")], lit("Unknown"))
    )

# Validation report
valid_naics = df_naics.filter(col("naics_valid") == True).count()
invalid_naics = df_naics.filter(col("naics_valid") == False).count()
print(f"NAICS Validation: {valid_naics:,} valid, {invalid_naics:,} invalid")

# Standardize state codes, ZIP codes, and congressional districts
df_geo = df_naics \
    .withColumn("borrower_state_std", upper(trim(col("borrower_state")))) \
    .withColumn("borrower_zip_std",
        lpad(regexp_replace(col("borrower_zip"), "[^0-9]", ""), 5, "0")
    ) \
    .withColumn("borrower_zip_valid",
        col("borrower_zip_std").rlike(r"^\d{5}$")
    ) \
    .withColumn("congressional_district_std",
        upper(trim(col("congressional_district")))
    ) \
    .withColumn("sba_region",
        when(col("borrower_state_std").isin("CT","ME","MA","NH","RI","VT"), lit("Region I - New England"))
        .when(col("borrower_state_std").isin("NJ","NY","PR","VI"), lit("Region II - Mid-Atlantic"))
        .when(col("borrower_state_std").isin("DC","DE","MD","PA","VA","WV"), lit("Region III - Atlantic"))
        .when(col("borrower_state_std").isin("AL","FL","GA","KY","MS","NC","SC","TN"), lit("Region IV - Southeast"))
        .when(col("borrower_state_std").isin("IL","IN","MI","MN","OH","WI"), lit("Region V - Great Lakes"))
        .when(col("borrower_state_std").isin("AR","LA","NM","OK","TX"), lit("Region VI - South Central"))
        .when(col("borrower_state_std").isin("IA","KS","MO","NE"), lit("Region VII - Plains"))
        .when(col("borrower_state_std").isin("CO","MT","ND","SD","UT","WY"), lit("Region VIII - Mountain"))
        .when(col("borrower_state_std").isin("AZ","CA","GU","HI","NV"), lit("Region IX - Pacific"))
        .when(col("borrower_state_std").isin("AK","ID","OR","WA"), lit("Region X - Northwest"))
        .otherwise(lit("Unknown"))
    )

# Validate loan amounts against program-specific thresholds
PROGRAM_LIMITS = {
    "PPP": (1000, 10000000),        # $1K - $10M
    "7A": (1000, 5000000),          # $1K - $5M
    "504": (125000, 5500000),       # $125K - $5.5M
    "SBIR": (50000, 2000000),       # $50K - $2M
}

df_validated = df_geo
for program, (min_amt, max_amt) in PROGRAM_LIMITS.items():
    df_validated = df_validated \
        .withColumn(f"amount_valid_{program.lower()}",
            when(
                (col("program_type") == program) &
                (col("loan_amount") >= min_amt) &
                (col("loan_amount") <= max_amt),
                lit(True)
            ).when(col("program_type") != program, lit(None))
            .otherwise(lit(False))
        )

# Unified amount validation flag
df_validated = df_validated \
    .withColumn("loan_amount_valid",
        coalesce(
            col("amount_valid_ppp"),
            col("amount_valid_7a"),
            col("amount_valid_504"),
            col("amount_valid_sbir"),
            lit(True)
        )
    ) \
    .drop("amount_valid_ppp", "amount_valid_7a", "amount_valid_504", "amount_valid_sbir")

# Deduplication by loan_id (primary key)
before_dedup = df_validated.count()
df_deduped = df_validated.dropDuplicates(["loan_id"])
after_dedup = df_deduped.count()
print(f"Deduplication: {before_dedup:,} -> {after_dedup:,} ({before_dedup - after_dedup:,} removed)")

# Data quality scoring (0-100)
df_silver = df_deduped \
    .withColumn("_dq_score",
        when(col("loan_id").isNotNull(), lit(15)).otherwise(lit(0)) +
        when(col("program_type").isNotNull(), lit(10)).otherwise(lit(0)) +
        when(col("loan_amount").isNotNull() & (col("loan_amount") > 0), lit(15)).otherwise(lit(0)) +
        when(col("naics_valid") == True, lit(10)).otherwise(lit(0)) +
        when(col("borrower_state_std").isNotNull(), lit(10)).otherwise(lit(0)) +
        when(col("borrower_zip_valid") == True, lit(10)).otherwise(lit(0)) +
        when(col("lender_name").isNotNull(), lit(10)).otherwise(lit(0)) +
        when(col("approval_date").isNotNull(), lit(10)).otherwise(lit(0)) +
        when(col("loan_amount_valid") == True, lit(5)).otherwise(lit(0)) +
        when(col("jobs_reported").isNotNull(), lit(5)).otherwise(lit(0))
    ) \
    .withColumn("_silver_timestamp", current_timestamp())

# Write Silver
df_silver.write \
    .format("delta") \
    .mode("overwrite") \
    .option("overwriteSchema", "true") \
    .partitionBy("program_type") \
    .saveAsTable("lh_silver.silver_sba_loans_unified")

spark.sql("OPTIMIZE lh_silver.silver_sba_loans_unified ZORDER BY (borrower_state_std, naics_sector_code)")

print(f"Silver: {df_silver.count():,} records written")
print(f"Average DQ score: {df_silver.agg(avg('_dq_score')).collect()[0][0]:.1f}/100")

# Build lender master from Silver loan data
df_lender_master = df_silver \
    .filter(col("lender_id").isNotNull()) \
    .groupBy("lender_id", "lender_name", "lender_city", "lender_state") \
    .agg(
        count("*").alias("total_loans"),
        sum("loan_amount").alias("total_amount"),
        countDistinct("program_type").alias("programs_served"),
        min("approval_date").alias("earliest_loan"),
        max("approval_date").alias("latest_loan"),
    )

df_lender_master.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("lh_silver.silver_sba_lenders")

print(f"Lender master: {df_lender_master.count():,} unique lenders")

# Portfolio summary: loan counts, amounts, and averages by program, state, and sector
df_portfolio = df_silver \
    .groupBy("program_type", "borrower_state_std", "naics_sector_name") \
    .agg(
        count("*").alias("loan_count"),
        sum("loan_amount").alias("total_loan_amount"),
        avg("loan_amount").alias("avg_loan_amount"),
        sum("sba_guarantee_amount").alias("total_guarantee_amount"),

        # Jobs impact
        sum("jobs_reported").alias("total_jobs_reported"),
        avg("jobs_reported").alias("avg_jobs_per_loan"),

        # Loan outcomes
        sum(when(col("loan_status") == "PIF", 1).otherwise(0)).alias("paid_in_full"),
        sum(when(col("loan_status") == "Default", 1).otherwise(0)).alias("defaults"),
        sum(when(col("loan_status") == "Charged Off", 1).otherwise(0)).alias("charged_off"),
        sum(when(col("loan_status") == "Active", 1).otherwise(0)).alias("active_loans"),

        # Quality metrics
        avg("_dq_score").alias("avg_dq_score"),
    ) \
    .withColumn("default_rate",
        when(col("loan_count") > 0,
            round((col("defaults") + col("charged_off")) * 100.0 / col("loan_count"), 2)
        ).otherwise(lit(0.0))
    ) \
    .withColumn("cost_per_job",
        when(col("total_jobs_reported") > 0,
            round(col("total_loan_amount") / col("total_jobs_reported"), 2)
        ).otherwise(lit(None))
    )

df_portfolio.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("lh_gold.gold_sba_portfolio_summary")

# Economic impact by state and congressional district
df_economic = df_silver \
    .groupBy("borrower_state_std", "congressional_district_std", "sba_region") \
    .agg(
        count("*").alias("total_loans"),
        sum("loan_amount").alias("total_capital_deployed"),
        sum("jobs_reported").alias("total_jobs_supported"),
        countDistinct("naics_sector_name").alias("industry_sectors_served"),
        countDistinct("lender_id").alias("active_lenders"),

        # Program breakdown
        sum(when(col("program_type") == "PPP", col("loan_amount")).otherwise(0)).alias("ppp_amount"),
        sum(when(col("program_type") == "7A", col("loan_amount")).otherwise(0)).alias("s7a_amount"),
        sum(when(col("program_type") == "504", col("loan_amount")).otherwise(0)).alias("s504_amount"),
        sum(when(col("program_type") == "SBIR", col("loan_amount")).otherwise(0)).alias("sbir_amount"),

        # Small business density
        countDistinct("business_name").alias("unique_businesses"),
    ) \
    .withColumn("avg_capital_per_job",
        when(col("total_jobs_supported") > 0,
            round(col("total_capital_deployed") / col("total_jobs_supported"), 0)
        ).otherwise(lit(None))
    )

df_economic.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("lh_gold.gold_sba_economic_impact")

# Lender performance scorecards
df_lender_score = df_silver \
    .groupBy("lender_id", "lender_name", "lender_state") \
    .agg(
        count("*").alias("total_loans_originated"),
        sum("loan_amount").alias("total_amount_originated"),
        avg("loan_amount").alias("avg_loan_size"),
        countDistinct("borrower_state_std").alias("states_served"),
        countDistinct("program_type").alias("programs_participated"),

        # Portfolio health
        sum(when(col("loan_status") == "PIF", 1).otherwise(0)).alias("loans_paid_in_full"),
        sum(when(col("loan_status").isin("Default", "Charged Off"), 1).otherwise(0)).alias("loans_defaulted"),
        sum(when(col("loan_status") == "Active", 1).otherwise(0)).alias("loans_active"),

        # PPP performance
        sum(when(col("program_type") == "PPP", 1).otherwise(0)).alias("ppp_loans"),
        sum(when((col("program_type") == "PPP") & (col("forgiveness_status") == "Full"), 1).otherwise(0))
            .alias("ppp_fully_forgiven"),

        # Jobs impact
        sum("jobs_reported").alias("total_jobs_supported"),
        avg("_dq_score").alias("avg_data_quality"),
    ) \
    .withColumn("default_rate_pct",
        when(col("total_loans_originated") > 0,
            round(col("loans_defaulted") * 100.0 / col("total_loans_originated"), 2)
        ).otherwise(lit(0.0))
    ) \
    .withColumn("ppp_forgiveness_rate_pct",
        when(col("ppp_loans") > 0,
            round(col("ppp_fully_forgiven") * 100.0 / col("ppp_loans"), 2)
        ).otherwise(lit(None))
    ) \
    .withColumn("lender_tier",
        when(col("total_loans_originated") >= 10000, lit("Tier 1 - National"))
        .when(col("total_loans_originated") >= 1000, lit("Tier 2 - Regional"))
        .when(col("total_loans_originated") >= 100, lit("Tier 3 - Community"))
        .otherwise(lit("Tier 4 - Micro"))
    )

df_lender_score.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("lh_gold.gold_sba_lender_scorecards")

spark.sql("OPTIMIZE lh_gold.gold_sba_lender_scorecards ZORDER BY (lender_state, lender_tier)")

# PPP-specific forgiveness analysis
df_ppp_forgiveness = df_silver \
    .filter(col("program_type") == "PPP") \
    .groupBy("forgiveness_status", "borrower_state_std", "naics_sector_name", "ppp_draw") \
    .agg(
        count("*").alias("loan_count"),
        sum("loan_amount").alias("total_loan_amount"),
        sum("forgiveness_amount").alias("total_forgiveness_amount"),
        avg("loan_amount").alias("avg_loan_amount"),
        sum("jobs_reported").alias("jobs_reported"),
    ) \
    .withColumn("forgiveness_rate_pct",
        when(col("total_loan_amount") > 0,
            round(col("total_forgiveness_amount") * 100.0 / col("total_loan_amount"), 2)
        ).otherwise(lit(0.0))
    )

df_ppp_forgiveness.write \
    .format("delta") \
    .mode("overwrite") \
    .saveAsTable("lh_gold.gold_sba_ppp_forgiveness")

# Summary statistics
print("Gold layer tables written:")
for table in ["gold_sba_portfolio_summary", "gold_sba_economic_impact",
              "gold_sba_lender_scorecards", "gold_sba_ppp_forgiveness"]:
    count = spark.table(f"lh_gold.{table}").count()
    print(f"  {table}: {count:,} rows")

# Data quality checks specific to SBA lending data
from great_expectations.core import ExpectationSuite

sba_expectations = ExpectationSuite(expectation_suite_name="sba_loans_suite")

# Loan ID uniqueness
sba_expectations.add_expectation({
    "expectation_type": "expect_column_values_to_be_unique",
    "kwargs": {"column": "loan_id"}
})

# Program type values
sba_expectations.add_expectation({
    "expectation_type": "expect_column_values_to_be_in_set",
    "kwargs": {"column": "program_type", "value_set": ["PPP", "7A", "504", "SBIR"]}
})

# Loan amount positivity
sba_expectations.add_expectation({
    "expectation_type": "expect_column_values_to_be_between",
    "kwargs": {"column": "loan_amount", "min_value": 0, "strict_min": True}
})

# State code validation (2-letter)
sba_expectations.add_expectation({
    "expectation_type": "expect_column_value_lengths_to_equal",
    "kwargs": {"column": "borrower_state_std", "value": 2}
})

# NAICS code format (6 digits)
sba_expectations.add_expectation({
    "expectation_type": "expect_column_values_to_match_regex",
    "kwargs": {"column": "naics_code_clean", "regex": r"^\d{6}$", "mostly": 0.95}
})

# PPP forgiveness cannot exceed loan amount
sba_expectations.add_expectation({
    "expectation_type": "expect_column_pair_values_a_to_be_less_than_or_equal_to_b",
    "kwargs": {"column_A": "forgiveness_amount", "column_B": "loan_amount", "mostly": 0.99}
})

print(f"SBA validation suite: {len(sba_expectations.expectations)} expectations defined")

# Automated DQ monitoring
dq_summary = spark.sql("""
    SELECT
        program_type,
        COUNT(*) as total_records,
        ROUND(AVG(_dq_score), 1) as avg_dq_score,
        SUM(CASE WHEN _dq_score >= 80 THEN 1 ELSE 0 END) as high_quality,
        SUM(CASE WHEN _dq_score < 50 THEN 1 ELSE 0 END) as low_quality,
        SUM(CASE WHEN naics_valid = false THEN 1 ELSE 0 END) as invalid_naics,
        SUM(CASE WHEN loan_amount_valid = false THEN 1 ELSE 0 END) as invalid_amounts
    FROM lh_silver.silver_sba_loans_unified
    GROUP BY program_type
    ORDER BY program_type
""")

dq_summary.show(truncate=False)

# Purview sensitivity labels for SBA data
purview_classifications = {
    "lh_bronze.bronze_sba_loans": {
        "sensitivity_label": "Public - Federal Open Data",
        "columns_with_pii": ["business_name"],
        "data_owner": "sba_analytics_team@contoso.com",
        "data_steward": "federal_data_governance@contoso.com",
        "retention_policy": "7 years (federal records schedule)",
    },
    "lh_silver.silver_sba_loans_unified": {
        "sensitivity_label": "Internal - Enriched Federal Data",
        "data_lineage": "bronze_sba_loans -> NAICS enrichment -> geo standardization",
    },
    "lh_gold.gold_sba_lender_scorecards": {
        "sensitivity_label": "Internal - Derived Analytics",
        "data_lineage": "silver_sba_loans_unified -> lender aggregation -> scoring",
    },
}

for table, config in purview_classifications.items():
    print(f"\n{table}:")
    for key, value in config.items():
        print(f"  {key}: {value}")