Spot workload imbalances across your team

Analyse team workload distribution

This identifies workload imbalances across a team. Your CSV should include columns: staff_member, date, active_cases, case_complexity_avg, hours_worked. The scipy library (used for trend analysis) is included in Colab by default.

import pandas as pd
import numpy as np
import matplotlib.pyplot as plt

# Load workload data
# Columns: staff_member, date, active_cases, case_complexity_avg, hours_worked, etc.
workload = pd.read_csv('team_workload.csv')
workload['date'] = pd.to_datetime(workload['date'])

print(f"Loaded {len(workload)} workload records")
print(f"Team members: {workload['staff_member'].nunique()}")
print(f"Date range: {workload['date'].min()} to {workload['date'].max()}")

# Get most recent month for current snapshot
latest_month = workload['date'].max()
current_workload = workload[workload['date'] == latest_month]

print(f"\nAnalysing workload as of {latest_month.strftime('%B %Y')}")
print("="*60)

# 1. Case distribution analysis
case_stats = current_workload.groupby('staff_member').agg({
    'active_cases': 'sum',
    'case_complexity_avg': 'mean',
    'hours_worked': 'sum'
}).round(1)

team_avg_cases = case_stats['active_cases'].mean()
team_avg_complexity = case_stats['case_complexity_avg'].mean()
team_avg_hours = case_stats['hours_worked'].mean()

print("\nCurrent Workload Distribution:")
print(f"{'Staff Member':<20} {'Active Cases':<15} {'Avg Complexity':<18} {'Hours':<10}")
print("-"*60)

for staff, row in case_stats.iterrows():
    cases_indicator = "⚠️ " if row['active_cases'] > team_avg_cases * 1.2 else "  "
    complexity_indicator = "⚠️ " if row['case_complexity_avg'] > team_avg_complexity * 1.15 else "  "
    hours_indicator = "⚠️ " if row['hours_worked'] > team_avg_hours * 1.1 else "  "

    print(f"{staff:<20} {cases_indicator}{row['active_cases']:<13.0f} "
          f"{complexity_indicator}{row['case_complexity_avg']:<16.1f} "
          f"{hours_indicator}{row['hours_worked']:<8.0f}")

print(f"\n{'Team Average:':<20} {team_avg_cases:<13.1f} {team_avg_complexity:<16.1f} {team_avg_hours:<8.1f}")

# 2. Identify imbalances
print("\n" + "="*60)
print("Workload Imbalances Identified:")
print("="*60)

# Cases imbalance
case_std = case_stats['active_cases'].std()
case_cv = (case_std / team_avg_cases) * 100  # Coefficient of variation

if case_cv > 25:  # More than 25% variation
    print(f"\n⚠️  High Case Distribution Inequality")
    print(f"   Variation: {case_cv:.1f}% (above 25% threshold)")
    highest_cases = case_stats['active_cases'].max()
    lowest_cases = case_stats['active_cases'].min()
    print(f"   Range: {lowest_cases:.0f} to {highest_cases:.0f} cases")
    print(f"   Ratio: Highest has {highest_cases/lowest_cases:.1f}x more cases than lowest")

# Complexity concentration
high_complexity_staff = case_stats[case_stats['case_complexity_avg'] > team_avg_complexity * 1.15]
if len(high_complexity_staff) > 0 and len(high_complexity_staff) < len(case_stats) * 0.5:
    print(f"\n⚠️  Complex Case Concentration")
    print(f"   {len(high_complexity_staff)} of {len(case_stats)} staff handling disproportionate complexity")
    print(f"   Staff with high complexity:")
    for staff in high_complexity_staff.index:
        complexity = high_complexity_staff.loc[staff, 'case_complexity_avg']
        print(f"     {staff}: {complexity:.1f} (team avg: {team_avg_complexity:.1f})")

# Hours worked imbalance
hours_over_40 = current_workload[current_workload['hours_worked'] > 40]
if len(hours_over_40) > 0:
    print(f"\n⚠️  Excess Hours Pattern")
    print(f"   {len(hours_over_40)} staff members working over 40 hours")
    for _, row in hours_over_40.iterrows():
        print(f"     {row['staff_member']}: {row['hours_worked']:.0f} hours")

# 3. Capacity analysis
total_cases = current_workload['active_cases'].sum()
total_capacity_hours = len(current_workload) * 37.5  # Assuming 37.5hr contracts
total_worked_hours = current_workload['hours_worked'].sum()

capacity_utilisation = (total_worked_hours / total_capacity_hours) * 100

print(f"\n" + "="*60)
print("Team Capacity Analysis:")
print("="*60)
print(f"Total active cases: {total_cases:.0f}")
print(f"Team size: {len(current_workload)} staff")
print(f"Contracted capacity: {total_capacity_hours:.0f} hours")
print(f"Actual hours worked: {total_worked_hours:.0f} hours")
print(f"Capacity utilisation: {capacity_utilisation:.1f}%")

if capacity_utilisation > 105:
    print(f"\n⚠️  Team Operating Over Capacity")
    print(f"   Working {capacity_utilisation - 100:.1f}% above contracted hours")
    print(f"   Indicates understaffing or unsustainable workload")
elif capacity_utilisation < 85:
    print(f"\n✅ Team Has Spare Capacity")
    print(f"   Could accommodate {100 - capacity_utilisation:.1f}% more work")
else:
    print(f"\n✅ Capacity Utilisation Healthy")

# 4. Trend analysis (last 6 months)
recent_data = workload[workload['date'] >= (latest_month - pd.DateOffset(months=6))]

trend_by_month = recent_data.groupby('date').agg({
    'active_cases': 'sum',
    'hours_worked': 'sum'
})

# Calculate trend
from scipy import stats
months_numeric = np.arange(len(trend_by_month))
cases_slope, _, _, _, _ = stats.linregress(months_numeric, trend_by_month['active_cases'])
hours_slope, _, _, _, _ = stats.linregress(months_numeric, trend_by_month['hours_worked'])

print(f"\n" + "="*60)
print("6-Month Trends:")
print("="*60)

if cases_slope > 2:
    print(f"⚠️  Cases Increasing: +{cases_slope:.1f} per month")
elif cases_slope < -2:
    print(f"✅ Cases Decreasing: {cases_slope:.1f} per month")
else:
    print(f"→ Cases Stable: {cases_slope:+.1f} per month")

if hours_slope > 5:
    print(f"⚠️  Hours Increasing: +{hours_slope:.1f} per month")
    print(f"   Suggests growing workload pressure")
elif hours_slope < -5:
    print(f"✅ Hours Decreasing: {hours_slope:.1f} per month")
else:
    print(f"→ Hours Stable: {hours_slope:+.1f} per month")

# 5. Recommendations
print(f"\n" + "="*60)
print("Recommendations:")
print("="*60)

if capacity_utilisation > 105:
    additional_staff_needed = ((total_worked_hours - total_capacity_hours) / 37.5)
    print(f"\n1. Consider hiring {additional_staff_needed:.1f} additional FTE")
    print(f"   Current team working unsustainable hours")

if case_cv > 25:
    print(f"\n2. Redistribute cases to balance workload")
    print(f"   Current distribution shows significant inequality")

if len(high_complexity_staff) > 0 and len(high_complexity_staff) < len(case_stats) * 0.5:
    print(f"\n3. Spread complex cases more evenly across team")
    print(f"   Current concentration creates burnout risk and knowledge silos")

if cases_slope > 2:
    print(f"\n4. Monitor growing demand")
    print(f"   Cases increasing {cases_slope:.1f} per month - plan capacity accordingly")

print(f"\n5. Review findings with team")
print(f"   Ask: Does this match your experience? What would help?")

# Save analysis
case_stats.to_csv('workload_analysis_current.csv')
trend_by_month.to_csv('workload_trends.csv')

# Create visualisation
fig, axes = plt.subplots(2, 1, figsize=(12, 10))

# Chart 1: Current case distribution
ax1 = axes[0]
case_stats['active_cases'].plot(kind='bar', ax=ax1, color='skyblue')
ax1.axhline(team_avg_cases, color='red', linestyle='--', label='Team Average')
ax1.set_title('Active Cases per Staff Member')
ax1.set_xlabel('Staff Member')
ax1.set_ylabel('Active Cases')
ax1.legend()
ax1.grid(True, alpha=0.3, axis='y')

# Chart 2: Trend over time
ax2 = axes[1]
ax2.plot(trend_by_month.index, trend_by_month['active_cases'], marker='o', label='Total Cases')
ax2.set_title('Team Workload Trend (Last 6 Months)')
ax2.set_xlabel('Month')
ax2.set_ylabel('Total Active Cases')
ax2.legend()
ax2.grid(True, alpha=0.3)

plt.tight_layout()
plt.savefig('workload_analysis.png', dpi=150)

print(f"\nAnalysis complete!")
print(f"Results saved to:")
print(f"  - workload_analysis_current.csv")
print(f"  - workload_trends.csv")
print(f"  - workload_analysis.png")