Small Business Lending Analytics¶

# Databricks notebook source
# MAGIC %md
# MAGIC # PPP Loan Distribution and Fraud Detection
# MAGIC Bronze-to-Silver transformation with fraud indicator scoring

# COMMAND ----------

from pyspark.sql import functions as F
from pyspark.sql.window import Window

# Load Bronze PPP data
df_ppp = spark.read.format("delta").load(
    "abfss://lh_bronze@onelake.dfs.fabric.microsoft.com/Tables/bronze_sba_ppp_loans"
)

# --- Fraud Indicator 1: Loan Amount Structuring ---
# Loans just below reporting thresholds ($150K boundary for detailed review)
df_structuring = df_ppp.withColumn(
    "structuring_flag",
    F.when(
        (F.col("loan_amount") >= 145000) & (F.col("loan_amount") < 150000), 1
    ).otherwise(0)
)

# --- Fraud Indicator 2: Velocity Analysis ---
# Multiple loans from same address within 24 hours
address_window = Window.partitionBy("borrower_address", "borrower_zip") \
    .orderBy("date_approved") \
    .rangeBetween(-86400, 86400)  # 24-hour window in seconds

df_velocity = df_structuring.withColumn(
    "same_address_loan_count",
    F.count("loan_number").over(address_window)
).withColumn(
    "velocity_flag",
    F.when(F.col("same_address_loan_count") > 2, 1).otherwise(0)
)

# --- Fraud Indicator 3: Jobs-to-Amount Ratio ---
# Flag loans where amount is disproportionate to reported employees
df_scored = df_velocity.withColumn(
    "amount_per_job",
    F.when(F.col("jobs_reported") > 0,
           F.col("loan_amount") / F.col("jobs_reported"))
    .otherwise(F.lit(None))
).withColumn(
    "ratio_flag",
    F.when(F.col("amount_per_job") > 125000, 1).otherwise(0)  # avg salary * 2.5x payroll multiplier
)

# --- Composite Fraud Risk Score ---
df_final = df_scored.withColumn(
    "fraud_risk_score",
    F.col("structuring_flag") + F.col("velocity_flag") + F.col("ratio_flag")
).withColumn(
    "fraud_risk_tier",
    F.when(F.col("fraud_risk_score") >= 3, "HIGH")
    .when(F.col("fraud_risk_score") == 2, "MEDIUM")
    .when(F.col("fraud_risk_score") == 1, "LOW")
    .otherwise("NONE")
)

# Write to Silver
df_final.write.format("delta").mode("overwrite").option(
    "overwriteSchema", "true"
).save("abfss://lh_silver@onelake.dfs.fabric.microsoft.com/Tables/silver_sba_ppp_fraud_scored")

print(f"PPP loans scored: {df_final.count():,}")
print(f"HIGH risk: {df_final.filter(F.col('fraud_risk_tier') == 'HIGH').count():,}")

# Databricks notebook source
# MAGIC %md
# MAGIC # 7(a)/504 Lending Trend Analysis
# MAGIC Gold-layer aggregations for lender performance and geographic access

# COMMAND ----------

from pyspark.sql import functions as F
from pyspark.sql.window import Window

# Load Silver 7(a)/504 data
df_loans = spark.read.format("delta").load(
    "abfss://lh_silver@onelake.dfs.fabric.microsoft.com/Tables/silver_sba_7a504_loans"
)

# --- Lender Concentration Index ---
# Calculate HHI by state to detect lending monopolies
lender_state = df_loans.groupBy("borrower_state", "lender_name").agg(
    F.sum("gross_approval").alias("total_approved"),
    F.count("*").alias("loan_count")
)

state_totals = lender_state.groupBy("borrower_state").agg(
    F.sum("total_approved").alias("state_total")
)

df_hhi = lender_state.join(state_totals, "borrower_state").withColumn(
    "market_share_pct",
    (F.col("total_approved") / F.col("state_total")) * 100
).withColumn(
    "share_squared",
    F.pow(F.col("market_share_pct"), 2)
)

state_hhi = df_hhi.groupBy("borrower_state").agg(
    F.sum("share_squared").alias("lending_hhi"),
    F.countDistinct("lender_name").alias("active_lenders"),
    F.sum("total_approved").alias("total_volume")
).withColumn(
    "concentration_tier",
    F.when(F.col("lending_hhi") > 2500, "HIGHLY_CONCENTRATED")
    .when(F.col("lending_hhi") > 1500, "MODERATELY_CONCENTRATED")
    .otherwise("COMPETITIVE")
)

# --- Charge-Off Analysis by Vintage ---
df_chargeoff = df_loans.withColumn(
    "approval_year", F.year("approval_date")
).groupBy("approval_year", "program").agg(
    F.count("*").alias("total_loans"),
    F.sum(F.when(F.col("loan_status") == "CHGOFF", 1).otherwise(0)).alias("charged_off"),
    F.sum("gross_approval").alias("total_approved"),
    F.sum(F.when(F.col("loan_status") == "CHGOFF", F.col("gross_charge_off_amount"))
          .otherwise(0)).alias("total_chargeoff_amount")
).withColumn(
    "chargeoff_rate",
    F.round(F.col("charged_off") / F.col("total_loans") * 100, 2)
).withColumn(
    "loss_severity",
    F.round(F.col("total_chargeoff_amount") / F.col("total_approved") * 100, 2)
)

# Write Gold tables
state_hhi.write.format("delta").mode("overwrite").save(
    "abfss://lh_gold@onelake.dfs.fabric.microsoft.com/Tables/gold_sba_lending_concentration"
)
df_chargeoff.write.format("delta").mode("overwrite").save(
    "abfss://lh_gold@onelake.dfs.fabric.microsoft.com/Tables/gold_sba_chargeoff_vintage"
)

print(f"States analyzed: {state_hhi.count()}")
print(f"Vintage cohorts: {df_chargeoff.count()}")

# Databricks notebook source
# MAGIC %md
# MAGIC # SBIR/STTR Innovation Grant Analytics
# MAGIC Correlating R&D grants with lending outcomes and job creation

# COMMAND ----------

from pyspark.sql import functions as F

# Load SBIR awards and SBA lending data
df_sbir = spark.read.format("delta").load(
    "abfss://lh_silver@onelake.dfs.fabric.microsoft.com/Tables/silver_sba_sbir_awards"
)
df_loans = spark.read.format("delta").load(
    "abfss://lh_silver@onelake.dfs.fabric.microsoft.com/Tables/silver_sba_7a504_loans"
)

# --- SBIR Award Distribution by Agency and Phase ---
agency_phase = df_sbir.groupBy("agency", "phase").agg(
    F.count("*").alias("award_count"),
    F.sum("award_amount").alias("total_funding"),
    F.avg("award_amount").alias("avg_award"),
    F.countDistinct("company_name").alias("unique_awardees")
).withColumn(
    "commercialization_ratio",
    F.when(F.col("phase") == "Phase III",
           F.col("award_count")).otherwise(0)
)

# --- Innovation-to-Lending Pipeline ---
# Match SBIR awardees to SBA loan recipients by fuzzy name + zip
df_sbir_entities = df_sbir.select(
    F.upper(F.trim("company_name")).alias("entity_name"),
    F.col("zip").alias("entity_zip"),
    F.col("award_amount").alias("sbir_total"),
    F.col("phase").alias("sbir_phase"),
    F.col("award_year")
).distinct()

df_loan_entities = df_loans.select(
    F.upper(F.trim("borrower_name")).alias("entity_name"),
    F.col("borrower_zip").alias("entity_zip"),
    F.col("gross_approval").alias("loan_amount"),
    F.year("approval_date").alias("loan_year")
).distinct()

# Exact match on normalized name + zip
df_pipeline = df_sbir_entities.join(
    df_loan_entities,
    on=["entity_name", "entity_zip"],
    how="inner"
).filter(
    F.col("loan_year") >= F.col("award_year")  # Loan came after SBIR award
).withColumn(
    "years_to_lending",
    F.col("loan_year") - F.col("award_year")
)

pipeline_stats = df_pipeline.groupBy("sbir_phase").agg(
    F.count("*").alias("matched_entities"),
    F.avg("years_to_lending").alias("avg_years_to_loan"),
    F.avg("loan_amount").alias("avg_subsequent_loan"),
    F.avg("sbir_total").alias("avg_sbir_award")
)

# Write Gold
pipeline_stats.write.format("delta").mode("overwrite").save(
    "abfss://lh_gold@onelake.dfs.fabric.microsoft.com/Tables/gold_sba_sbir_lending_pipeline"
)

print(f"SBIR awardees matched to SBA loans: {df_pipeline.count():,}")

# Cross-domain: SBA lending deserts in high-value agricultural counties
df_usda = spark.read.format("delta").load(
    "abfss://lh_gold@onelake.dfs.fabric.microsoft.com/Tables/gold_usda_crop_production"
)
df_sba = spark.read.format("delta").load(
    "abfss://lh_gold@onelake.dfs.fabric.microsoft.com/Tables/gold_sba_lending_concentration"
)

df_rural_gap = df_usda.join(
    df_sba,
    df_usda.state_alpha == df_sba.borrower_state,
    "left"
).withColumn(
    "lending_gap_score",
    F.col("crop_value_millions") / F.coalesce(F.col("total_volume"), F.lit(1))
).filter(
    F.col("lending_gap_score") > 2.0  # High ag value, low lending
).orderBy(F.desc("lending_gap_score"))

df_rural_gap.select(
    "state_alpha", "crop_value_millions", "total_volume",
    "active_lenders", "lending_gap_score"
).show(20)