Lab 08: E-Supply Chain Mapping & Disruption Analysis¶

# disruption_analysis.py
import networkx as nx
import matplotlib.pyplot as plt
import matplotlib.patches as mpatches

# Build supply chain graph
# Nodes = supply chain entities, Edges = material/product flow
G = nx.DiGraph()

# Add nodes with attributes
nodes = {
    # Raw materials
    "Copper Mine (Chile)":       {"tier": 3, "country": "CL", "risk": "medium"},
    "Lithium Mine (Australia)":  {"tier": 3, "country": "AU", "risk": "low"},
    "Rare Earth (China)":        {"tier": 3, "country": "CN", "risk": "high"},

    # Component manufacturers
    "TSMC (Taiwan)":             {"tier": 2, "country": "TW", "risk": "critical"},
    "Samsung Display (Korea)":   {"tier": 2, "country": "KR", "risk": "medium"},
    "Foxconn Assembly (China)":  {"tier": 1, "country": "CN", "risk": "high"},

    # Logistics
    "Port of Shanghai":          {"tier": 0, "country": "CN", "risk": "high"},
    "Port of Long Beach":        {"tier": 0, "country": "US", "risk": "medium"},

    # Fulfillment
    "Apple Fulfillment (CA)":    {"tier": -1, "country": "US", "risk": "low"},
    "FedEx Last Mile":           {"tier": -1, "country": "US", "risk": "low"},

    # Customer
    "End Customer":              {"tier": -2, "country": "US", "risk": "none"},
}

for node, attrs in nodes.items():
    G.add_node(node, **attrs)

# Add edges (flow relationships)
edges = [
    ("Copper Mine (Chile)",      "Foxconn Assembly (China)",  {"flow": "copper"}),
    ("Lithium Mine (Australia)", "Foxconn Assembly (China)",  {"flow": "lithium"}),
    ("Rare Earth (China)",       "TSMC (Taiwan)",             {"flow": "rare earths"}),
    ("TSMC (Taiwan)",            "Foxconn Assembly (China)",  {"flow": "chips", "critical": True}),
    ("Samsung Display (Korea)",  "Foxconn Assembly (China)",  {"flow": "displays"}),
    ("Foxconn Assembly (China)", "Port of Shanghai",          {"flow": "finished goods"}),
    ("Port of Shanghai",         "Port of Long Beach",        {"flow": "containers", "transit_days": 16}),
    ("Port of Long Beach",       "Apple Fulfillment (CA)",    {"flow": "inventory"}),
    ("Apple Fulfillment (CA)",   "FedEx Last Mile",           {"flow": "orders"}),
    ("FedEx Last Mile",          "End Customer",              {"flow": "delivery"}),
]

for src, dst, attrs in edges:
    G.add_edge(src, dst, **attrs)

# Color by risk level
risk_colors = {"critical": "#c62828", "high": "#f57c00",
               "medium": "#fbc02d", "low": "#388e3c", "none": "#1565c0"}
node_colors = [risk_colors[G.nodes[n]["risk"]] for n in G.nodes()]

plt.figure(figsize=(16, 10))
pos = nx.spring_layout(G, seed=42, k=2)
nx.draw_networkx(G, pos, node_color=node_colors, node_size=2000,
                 font_size=7, arrows=True, arrowsize=20,
                 edge_color="#555", width=1.5, with_labels=True)

legend = [mpatches.Patch(color=c, label=r) for r, c in risk_colors.items()]
plt.legend(handles=legend, loc="upper left", title="Risk Level")
plt.title("E-Commerce Supply Chain Network — Risk Heat Map", fontsize=14)
plt.axis("off")
plt.tight_layout()
plt.savefig("supply_chain_map.png", dpi=150)
print("Supply chain map saved: supply_chain_map.png")

# ── Single Points of Failure ─────────────────────────────────────
print("\n=== Single Points of Failure Analysis ===\n")

# Find nodes whose removal disconnects the graph
critical_nodes = []
for node in G.nodes():
    G_test = G.copy()
    G_test.remove_node(node)
    if not nx.is_weakly_connected(G_test):
        critical_nodes.append(node)

print("Critical nodes (removal disconnects supply chain):")
for node in critical_nodes:
    risk = G.nodes[node]["risk"]
    print(f"  ⚠ {node} [risk: {risk}]")

# Betweenness centrality — nodes most "in between" others
centrality = nx.betweenness_centrality(G)
print("\nTop 5 nodes by betweenness centrality (flow bottlenecks):")
for node, score in sorted(centrality.items(), key=lambda x: x[1], reverse=True)[:5]:
    print(f"  {score:.3f} — {node}")

# ── Disruption Cost Model ─────────────────────────────────────────
print("\n=== Disruption Cost Calculator ===\n")

scenarios = [
    {
        "name": "TSMC Production Stop (Taiwan conflict)",
        "duration_days": 90,
        "affected_revenue_pct": 0.85,
        "daily_revenue": 137_000_000,  # Apple ~$50B/quarter / 90 days
        "mitigation_cost": 2_000_000_000,
        "probability_annual": 0.05
    },
    {
        "name": "Port of Long Beach Strike (30 days)",
        "duration_days": 30,
        "affected_revenue_pct": 0.40,
        "daily_revenue": 137_000_000,
        "mitigation_cost": 50_000_000,
        "probability_annual": 0.15
    },
    {
        "name": "Foxconn Factory Shutdown (COVID-style)",
        "duration_days": 45,
        "affected_revenue_pct": 0.60,
        "daily_revenue": 137_000_000,
        "mitigation_cost": 500_000_000,
        "probability_annual": 0.10
    }
]

for s in scenarios:
    direct_loss = s["duration_days"] * s["daily_revenue"] * s["affected_revenue_pct"]
    expected_annual_loss = direct_loss * s["probability_annual"]
    roi = (expected_annual_loss - s["mitigation_cost"]) / s["mitigation_cost"] * 100

    print(f"Scenario: {s['name']}")
    print(f"  Direct loss:              ${direct_loss/1e9:.2f}B")
    print(f"  Probability (annual):     {s['probability_annual']*100:.0f}%")
    print(f"  Expected annual loss:     ${expected_annual_loss/1e6:.0f}M")
    print(f"  Mitigation cost:          ${s['mitigation_cost']/1e6:.0f}M")
    print(f"  Mitigation ROI:           {roi:.0f}%")
    print(f"  Decision: {'INVEST in mitigation ✓' if roi > 0 else 'Accept risk ✗'}\n")

import math

def eoq(demand_annual: float, order_cost: float, holding_cost_pct: float, unit_cost: float) -> dict:
    """
    EOQ = sqrt(2 * D * S / H)
    D = annual demand, S = order cost, H = holding cost per unit per year
    """
    H = holding_cost_pct * unit_cost
    eoq_qty = math.sqrt(2 * demand_annual * order_cost / H)
    orders_per_year = demand_annual / eoq_qty
    cycle_time_days = 365 / orders_per_year
    annual_order_cost = orders_per_year * order_cost
    annual_holding_cost = (eoq_qty / 2) * H
    total_cost = annual_order_cost + annual_holding_cost

    return {
        "EOQ": round(eoq_qty),
        "Orders per year": round(orders_per_year, 1),
        "Cycle time (days)": round(cycle_time_days, 1),
        "Annual order cost": round(annual_order_cost, 2),
        "Annual holding cost": round(annual_holding_cost, 2),
        "Total annual cost": round(total_cost, 2)
    }

print("=== EOQ Calculator ===\n")
products = [
    ("Safety Gloves", 5000, 85, 0.25, 9.00),
    ("Hard Hats",     2000, 95, 0.20, 18.00),
    ("Safety Vests",  3500, 75, 0.30, 12.00),
]
for name, D, S, h, c in products:
    result = eoq(D, S, h, c)
    print(f"{name}:")
    for k, v in result.items():
        print(f"  {k}: {v}")
    print()