Analysing Traffic Accident Patterns

import pandas as pd
import numpy as np

df = pd.DataFrame({
    "accident_id": ["A001","A002","A003","A004","A005","A006","A007","A008",
                    "A009","A010","A011","A012","A013","A014","A015","A016",
                    "A017","A018","A019","A020"],
    "road_type":   ["Motorway","A-Road","Residential","Junction","Motorway",
                    "A-Road","Residential","Junction","Motorway","A-Road",
                    "Residential","Junction","Motorway","A-Road","Residential",
                    "Junction","Motorway","A-Road","Residential","Junction"],
    "weather":     ["Wet","Dry","Dry","Fog","Ice","Wet","Dry","Ice",
                    "Dry","Fog","Wet","Dry","Ice","Wet","Fog","Dry",
                    "Wet","Ice","Dry","Fog"],
    "hour":        [8,17,14,7,22,8,13,19,6,16,11,9,23,17,7,14,8,21,15,7],
    "day_of_week": ["Monday","Friday","Wednesday","Tuesday","Saturday",
                    "Monday","Thursday","Friday","Sunday","Wednesday",
                    "Tuesday","Monday","Saturday","Thursday","Tuesday",
                    "Wednesday","Friday","Saturday","Monday","Sunday"],
    "severity":    ["Serious","Slight","Slight","Fatal","Fatal",
                    "Serious","Slight","Serious","Fatal","Serious",
                    "Slight","Slight","Fatal","Serious","Fatal",
                    "Slight","Serious","Fatal","Slight","Serious"],
    "vehicles":    [2,2,1,3,4,2,1,2,3,2,1,2,4,2,3,1,2,4,1,3],
    "casualties":  [1,1,0,2,3,1,0,1,2,1,0,0,3,1,2,0,1,3,0,2],
    "speed_limit": [70,60,30,50,70,60,30,50,70,60,30,50,70,60,30,50,60,70,30,50]
})

# Map severity to numeric weights: Slight=1, Serious=3, Fatal=5
severity_weights = {"Slight": 1, "Serious": 3, "Fatal": 5}
df["severity_weight"] = df["severity"].map(severity_weights)

# Time-of-day banding using pd.cut()
df["time_period"] = pd.cut(
    df["hour"],
    bins   = [0, 6, 9, 15, 19, 24],
    labels = ["Night (0-6)","Morning Rush (7-9)",
              "Midday (10-15)","Eve Rush (16-19)","Late Eve (20-23)"],
    right  = False
)

print(f"Accidents: {len(df)} | Fatal: {(df['severity']=='Fatal').sum()} "
      f"| Serious: {(df['severity']=='Serious').sum()} "
      f"| Slight: {(df['severity']=='Slight').sum()}")
print(f"Total casualties: {df['casualties'].sum()} | Total severity score: {df['severity_weight'].sum()}")

# Road type: count vs weighted severity score
road_stats = df.groupby("road_type").agg(
    accidents        = ("accident_id",     "count"),
    severity_score   = ("severity_weight", "sum"),
    mean_severity    = ("severity_weight", "mean"),
    total_casualties = ("casualties",      "sum"),
    fatal_count      = ("severity",        lambda x: (x == "Fatal").sum())
).round(2).reset_index()
road_stats = road_stats.sort_values("severity_score", ascending=False)
print("\nRoad type — count vs weighted severity score:")
print(road_stats.to_string(index=False))

# Weather severity analysis
weather_stats = df.groupby("weather").agg(
    accidents        = ("accident_id",     "count"),
    severity_score   = ("severity_weight", "sum"),
    mean_severity    = ("severity_weight", "mean"),
    total_casualties = ("casualties",      "sum"),
    fatal_count      = ("severity",        lambda x: (x == "Fatal").sum())
).round(2).reset_index()

# Severity uplift vs dry baseline
dry_mean = weather_stats.loc[weather_stats["weather"] == "Dry", "mean_severity"].values[0]
weather_stats["uplift_vs_dry"] = (weather_stats["mean_severity"] / dry_mean).round(2)
weather_stats = weather_stats.sort_values("mean_severity", ascending=False)

print("Weather condition severity — ranked by mean harm score:")
print(weather_stats[["weather","accidents","mean_severity",
                      "fatal_count","uplift_vs_dry"]].to_string(index=False))

# Share of total harm in adverse vs dry weather
adverse = df[df["weather"] != "Dry"]
total_score = df["severity_weight"].sum()
adverse_score = adverse["severity_weight"].sum()
print(f"\nAdverse weather: {len(adverse)} accidents ({len(adverse)/len(df)*100:.0f}% of total)")
print(f"Adverse weather share of total severity score: {adverse_score}/{total_score} = {adverse_score/total_score*100:.0f}%")
print(f"Adverse weather casualties: {adverse['casualties'].sum()}/{df['casualties'].sum()} = {adverse['casualties'].sum()/df['casualties'].sum()*100:.0f}%")

# Severity by time period
time_stats = df.groupby("time_period", observed=True).agg(
    accidents      = ("accident_id",     "count"),
    severity_score = ("severity_weight", "sum"),
    mean_severity  = ("severity_weight", "mean"),
    casualties     = ("casualties",      "sum")
).round(2).reset_index()
time_stats = time_stats.sort_values("severity_score", ascending=False)

print("Severity by time period:")
print(time_stats.to_string(index=False))

# Day of week severity
day_order = ["Monday","Tuesday","Wednesday","Thursday","Friday","Saturday","Sunday"]
day_stats = df.groupby("day_of_week").agg(
    accidents      = ("accident_id",     "count"),
    severity_score = ("severity_weight", "sum"),
    mean_severity  = ("severity_weight", "mean"),
    casualties     = ("casualties",      "sum")
).round(2).reindex(day_order).reset_index()

print("\nSeverity by day of week:")
print(day_stats.to_string(index=False))

# Weekend vs weekday comparison
df["is_weekend"] = df["day_of_week"].isin(["Saturday","Sunday"])
weekend_comp = df.groupby("is_weekend").agg(
    accidents      = ("accident_id",     "count"),
    severity_score = ("severity_weight", "sum"),
    mean_severity  = ("severity_weight", "mean"),
    casualties     = ("casualties",      "sum")
).round(2)
weekend_comp.index = weekend_comp.index.map({False: "Weekday", True: "Weekend"})
print("\nWeekend vs Weekday:")
print(weekend_comp.to_string())

# Pivot: total severity score by road_type x weather
pivot_sev = pd.pivot_table(
    df,
    index      = "road_type",
    columns    = "weather",
    values     = "severity_weight",
    aggfunc    = "sum",
    fill_value = np.nan
).round(1)
print("Total severity score by road type x weather:")
print(pivot_sev.to_string())

# Pivot: accident count
pivot_count = pd.pivot_table(
    df,
    index      = "road_type",
    columns    = "weather",
    values     = "accident_id",
    aggfunc    = "count",
    fill_value = 0
)
print("\nAccident count by road type x weather:")
print(pivot_count.to_string())

# Find the single worst combination
worst = pivot_sev.stack().idxmax()
worst_score = pivot_sev.stack().max()
print(f"\nHighest-risk combination: {worst[0]} in {worst[1]} weather — score {worst_score}")

# Row and column totals for ranking
print("\nRow totals (road type severity ranking):")
print(pivot_sev.sum(axis=1).sort_values(ascending=False).to_string())
print("\nColumn totals (weather severity ranking):")
print(pivot_sev.sum(axis=0).sort_values(ascending=False).to_string())

TOTAL_BUDGET = 2_400_000  # £2.4m road safety budget

# Current harm scores by priority segment
priority_segments = df.groupby(["road_type","weather"]).agg(
    accidents      = ("accident_id",     "count"),
    severity_score = ("severity_weight", "sum"),
    casualties     = ("casualties",      "sum")
).reset_index()
priority_segments = priority_segments.sort_values("severity_score", ascending=False).head(5)
priority_segments["budget_share_pct"] = (
    priority_segments["severity_score"] / priority_segments["severity_score"].sum() * 100
).round(1)
priority_segments["allocated_budget"] = (
    priority_segments["severity_score"] / priority_segments["severity_score"].sum() * TOTAL_BUDGET
).round(0).astype(int)

print("Top 5 priority segments — severity-proportional budget allocation:")
print(priority_segments.to_string(index=False))

# Simulate 30% severity reduction at Motorway Ice (top segment)
top_score    = 15.0
reduction_30 = top_score * 0.30
points_saved = round(reduction_30, 1)
cost_per_point = TOTAL_BUDGET / priority_segments["severity_score"].sum()

print(f"\nSimulation: 30% severity reduction at Motorway Ice")
print(f"  Current severity score:      {top_score}")
print(f"  Harm points prevented:       {points_saved}")
print(f"  Proportional budget share:   £{int(priority_segments[priority_segments['road_type']=='Motorway']['allocated_budget'].values[0]):,}")
print(f"  Cost per harm point removed: £{cost_per_point:,.0f}")
print(f"\nTotal severity score currently: {df['severity_weight'].sum()}")
print(f"After intervention:             {df['severity_weight'].sum() - points_saved:.0f} "
      f"({points_saved/df['severity_weight'].sum()*100:.1f}% reduction)")