Antitrust Analytics¶

# Databricks notebook source
# MAGIC %md
# MAGIC # HHI Market Concentration Analysis
# MAGIC Bronze → Silver → Gold pipeline for HHI computation

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

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

# Read market share data from Silver layer
df_market = spark.read.table("lh_silver.antitrust_market_shares")

# Calculate HHI per market per year
df_hhi = (
    df_market
    .withColumn("share_squared", F.col("market_share_pct") ** 2)
    .groupBy("market_id", "naics_code", "market_name", "year")
    .agg(
        F.sum("share_squared").alias("hhi"),
        F.count("firm_id").alias("firm_count"),
        F.max("market_share_pct").alias("top_firm_share"),
        F.collect_list(
            F.struct("firm_name", "market_share_pct")
        ).alias("firms")
    )
)

# Classify concentration levels per 2023 Guidelines
df_hhi_classified = (
    df_hhi
    .withColumn(
        "concentration_level",
        F.when(F.col("hhi") < 1500, "Unconcentrated")
         .when(F.col("hhi") <= 2500, "Moderately Concentrated")
         .otherwise("Highly Concentrated")
    )
    .withColumn(
        "enforcement_flag",
        F.when(F.col("hhi") > 2500, True).otherwise(False)
    )
)

# Write to Gold layer
df_hhi_classified.write.format("delta").mode("overwrite").saveAsTable(
    "lh_gold.antitrust_hhi_concentration"
)

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

# MAGIC %md
# MAGIC ## HHI Year-over-Year Change

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

w = Window.partitionBy("market_id").orderBy("year")

df_hhi_delta = (
    df_hhi_classified
    .withColumn("hhi_prior_year", F.lag("hhi").over(w))
    .withColumn("hhi_delta", F.col("hhi") - F.col("hhi_prior_year"))
    .withColumn(
        "trend",
        F.when(F.col("hhi_delta") > 100, "Concentrating")
         .when(F.col("hhi_delta") < -100, "Deconcentrating")
         .otherwise("Stable")
    )
)

df_hhi_delta.write.format("delta").mode("overwrite").saveAsTable(
    "lh_gold.antitrust_hhi_trends"
)

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

from pyspark.sql import functions as F

df_mergers = spark.read.table("lh_gold.merger_review_pipeline")

# Stage duration analysis
df_stage_stats = (
    df_mergers
    .groupBy("stage", "outcome")
    .agg(
        F.count("filing_id").alias("filing_count"),
        F.avg("days_in_stage").alias("avg_days"),
        F.percentile_approx("days_in_stage", 0.5).alias("median_days"),
        F.max("days_in_stage").alias("max_days")
    )
    .orderBy("stage", "outcome")
)

# Challenge rate by NAICS sector
df_challenge_rate = (
    df_mergers
    .filter(F.col("stage") == "Decision")
    .groupBy("naics_sector")
    .agg(
        F.count("filing_id").alias("total_reviews"),
        F.sum(
            F.when(F.col("outcome").isin("Challenged", "Blocked"), 1).otherwise(0)
        ).alias("challenged_count"),
    )
    .withColumn(
        "challenge_rate",
        F.round(F.col("challenged_count") / F.col("total_reviews") * 100, 2)
    )
    .orderBy(F.desc("challenge_rate"))
)

df_challenge_rate.write.format("delta").mode("overwrite").saveAsTable(
    "lh_gold.antitrust_challenge_rates"
)

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

# MAGIC %md
# MAGIC ## Second Request Analysis
# MAGIC Second Requests are the "deep dive" investigations. Tracking their
# MAGIC frequency and outcomes reveals enforcement intensity trends.

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

df_second_requests = (
    df_mergers
    .filter(F.col("stage") == "Second Request")
    .withColumn("year", F.year("stage_entry_date"))
    .groupBy("year")
    .agg(
        F.count("filing_id").alias("second_requests_issued"),
        F.avg("days_in_stage").alias("avg_review_days"),
        F.sum(
            F.when(F.col("outcome") == "Abandoned", 1).otherwise(0)
        ).alias("abandoned_after_2r")
    )
    .orderBy("year")
)

df_second_requests.write.format("delta").mode("overwrite").saveAsTable(
    "lh_gold.antitrust_second_request_trends"
)

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

# MAGIC %md
# MAGIC # Bid-Rigging Pattern Detection
# MAGIC Identifies suspicious patterns in government contract bidding.

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

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

df_bids = spark.read.table("lh_silver.government_contract_bids")

# Pattern 1: Rotating winners
# If a small group of firms take turns winning in a market, flag it
w_market = Window.partitionBy("market_id", "contract_category").orderBy("award_date")

df_winners = (
    df_bids
    .filter(F.col("is_winner") == True)
    .withColumn("prev_winner", F.lag("firm_id").over(w_market))
    .withColumn("prev_prev_winner", F.lag("firm_id", 2).over(w_market))
    .withColumn(
        "rotation_flag",
        F.when(
            (F.col("firm_id") == F.col("prev_prev_winner")) &
            (F.col("firm_id") != F.col("prev_winner")),
            True
        ).otherwise(False)
    )
)

# Pattern 2: Cover bids with consistent spreads
# If losing bids are always X% above the winner, that is suspicious
df_bid_spreads = (
    df_bids
    .join(
        df_bids.filter(F.col("is_winner") == True)
        .select(
            F.col("contract_id"),
            F.col("bid_amount").alias("winning_bid")
        ),
        on="contract_id"
    )
    .filter(F.col("is_winner") == False)
    .withColumn(
        "spread_pct",
        F.round(
            (F.col("bid_amount") - F.col("winning_bid"))
            / F.col("winning_bid") * 100, 2
        )
    )
)

# Flag markets where spread variance is suspiciously low
df_suspicious_markets = (
    df_bid_spreads
    .groupBy("market_id", "firm_id")
    .agg(
        F.avg("spread_pct").alias("avg_spread"),
        F.stddev("spread_pct").alias("spread_stddev"),
        F.count("contract_id").alias("bid_count")
    )
    .filter(
        (F.col("spread_stddev") < 1.0) &  # Very consistent spreads
        (F.col("bid_count") >= 5)           # Enough data points
    )
)

df_suspicious_markets.write.format("delta").mode("overwrite").saveAsTable(
    "lh_gold.antitrust_bidrigging_flags"
)

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

# MAGIC %md
# MAGIC ## Price-Fixing Detection: Parallel Pricing

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

# Pattern 3: Parallel pricing — competitors change prices
# in lockstep within a narrow time window
df_prices = spark.read.table("lh_silver.market_prices")

w_firm = Window.partitionBy("firm_id", "product_category").orderBy("effective_date")

df_price_changes = (
    df_prices
    .withColumn("prev_price", F.lag("price").over(w_firm))
    .withColumn("price_change_pct", 
        F.round((F.col("price") - F.col("prev_price")) / F.col("prev_price") * 100, 2)
    )
    .filter(F.col("price_change_pct").isNotNull())
)

# Find instances where multiple firms changed prices by similar amounts
# within a short window
df_parallel = (
    df_price_changes.alias("a")
    .join(
        df_price_changes.alias("b"),
        (F.col("a.product_category") == F.col("b.product_category")) &
        (F.col("a.firm_id") != F.col("b.firm_id")) &
        (F.abs(F.datediff(F.col("a.effective_date"), F.col("b.effective_date"))) <= 7) &
        (F.abs(F.col("a.price_change_pct") - F.col("b.price_change_pct")) < 0.5)
    )
    .select(
        F.col("a.product_category"),
        F.col("a.firm_id").alias("firm_a"),
        F.col("b.firm_id").alias("firm_b"),
        F.col("a.effective_date").alias("date_a"),
        F.col("b.effective_date").alias("date_b"),
        F.col("a.price_change_pct").alias("change_a"),
        F.col("b.price_change_pct").alias("change_b")
    )
)

df_parallel.write.format("delta").mode("overwrite").saveAsTable(
    "lh_gold.antitrust_parallel_pricing_flags"
)

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

from pyspark.sql import functions as F

df_sentences = spark.read.table("lh_silver.ussc_antitrust_sentences")

# Penalty analysis by offense type
df_penalties = (
    df_sentences
    .groupBy("offense_type", "fiscal_year")
    .agg(
        F.count("case_id").alias("case_count"),
        F.sum("fine_amount").alias("total_fines"),
        F.avg("fine_amount").alias("avg_fine"),
        F.max("fine_amount").alias("max_fine"),
        F.avg("prison_months").alias("avg_prison_months"),
        F.sum(
            F.when(F.col("prison_months") > 0, 1).otherwise(0)
        ).alias("incarceration_count")
    )
    .withColumn(
        "incarceration_rate",
        F.round(F.col("incarceration_count") / F.col("case_count") * 100, 2)
    )
    .orderBy("fiscal_year", "offense_type")
)

df_penalties.write.format("delta").mode("overwrite").saveAsTable(
    "lh_gold.antitrust_penalty_trends"
)

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

# Join antitrust HHI data with SBA lending data by geography and year
df_hhi = spark.read.table("lh_gold.antitrust_hhi_concentration")
df_sba = spark.read.table("lh_gold.sba_loan_origination_by_region")

df_cross = (
    df_hhi
    .join(df_sba, on=["state_fips", "year"], how="inner")
    .select(
        "state_fips", "year", "market_name",
        "hhi", "concentration_level",
        "total_loans", "total_loan_amount", "avg_loan_size"
    )
)

# Correlation analysis
from pyspark.ml.stat import Correlation
from pyspark.ml.feature import VectorAssembler

assembler = VectorAssembler(
    inputCols=["hhi", "total_loans", "total_loan_amount"],
    outputCol="features"
)
df_vector = assembler.transform(df_cross)
correlation_matrix = Correlation.corr(df_vector, "features").head()[0]
print("HHI vs Loan Count correlation:", correlation_matrix.toArray()[0][1])

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

df_hhi = spark.read.table("lh_gold.antitrust_hhi_concentration")
df_epa = spark.read.table("lh_gold.epa_enforcement_by_naics")

df_cross_epa = (
    df_hhi
    .join(df_epa, on=["naics_code", "year"], how="inner")
    .groupBy("concentration_level")
    .agg(
        F.avg("violation_count").alias("avg_violations"),
        F.avg("penalty_amount").alias("avg_penalty"),
        F.avg("compliance_rate").alias("avg_compliance_rate")
    )
)

df_cross_epa.show()