Diagnostics

Metrics

deepbullwhip.diagnostics.metrics.bullwhip_ratio(orders, demand)

Variance ratio: Var(orders) / Var(demand).

Source code in deepbullwhip/diagnostics/metrics.py

def bullwhip_ratio(orders: TimeSeries, demand: TimeSeries) -> float:
    """Variance ratio: Var(orders) / Var(demand)."""
    vd = np.var(demand)
    return float(np.var(orders) / vd) if vd > 0 else 1.0

deepbullwhip.diagnostics.metrics.fill_rate(inventory_levels)

Fraction of periods with non-negative inventory.

Source code in deepbullwhip/diagnostics/metrics.py

def fill_rate(inventory_levels: TimeSeries) -> float:
    """Fraction of periods with non-negative inventory."""
    return float(np.mean(np.asarray(inventory_levels) >= 0))

deepbullwhip.diagnostics.metrics.cumulative_bullwhip(echelon_bw_ratios)

Product of per-echelon bullwhip ratios.

Source code in deepbullwhip/diagnostics/metrics.py

def cumulative_bullwhip(echelon_bw_ratios: list[float]) -> float:
    """Product of per-echelon bullwhip ratios."""
    result = 1.0
    for bw in echelon_bw_ratios:
        result *= bw
    return result

deepbullwhip.diagnostics.metrics.bullwhip_lower_bound(lead_time, sensitivity, phi)

Theorem 1 lower bound: 1 + 2Llamphi/(1+phi^2) + L^2lam^2.

Source code in deepbullwhip/diagnostics/metrics.py

def bullwhip_lower_bound(
    lead_time: int, sensitivity: float, phi: float
) -> float:
    """Theorem 1 lower bound: 1 + 2*L*lam*phi/(1+phi^2) + L^2*lam^2."""
    return (
        1
        + 2 * lead_time * sensitivity * phi / (1 + phi**2)
        + lead_time**2 * sensitivity**2
    )

Plots

deepbullwhip.diagnostics.plots.plot_demand_trajectory(demand, shock_period=104, width='double')

Demand trajectory with marginal distribution.

Left panel: time series with mean, +/-1 std band, and optional shock marker. Right panel: rotated histogram.

Source code in deepbullwhip/diagnostics/plots.py

def plot_demand_trajectory(
    demand: TimeSeries,
    shock_period: int | None = 104,
    width: Literal["single", "double"] = "double",
) -> matplotlib.figure.Figure:
    """Demand trajectory with marginal distribution.

    Left panel: time series with mean, +/-1 std band, and optional shock
    marker. Right panel: rotated histogram.
    """
    _apply_style()
    w = _col_width(width)
    fig, (ax1, ax2) = plt.subplots(
        1, 2, figsize=(w, w / GOLDEN / 1.6),
        gridspec_kw={"width_ratios": [4, 1], "wspace": 0.05},
    )

    weeks = np.arange(len(demand))
    mu, sigma = demand.mean(), demand.std()

    ax1.fill_between(
        weeks, mu - sigma, mu + sigma,
        color=COLORS["E2"], alpha=0.12, label=r"$\mu \pm \sigma$",
    )
    ax1.plot(weeks, demand, color=COLORS["demand"], linewidth=0.6, label="Demand")
    ax1.axhline(mu, color=COLORS["grid"], linestyle="--", linewidth=0.5)

    if shock_period is not None and shock_period < len(demand):
        ax1.axvline(
            shock_period, color=COLORS["backorder"], linestyle=":",
            linewidth=0.7, label=f"Shock ($t={shock_period}$)",
        )

    ax1.set_xlabel("Week")
    ax1.set_ylabel("Demand (units)")
    ax1.legend(loc="upper left")
    ax1.set_xlim(0, len(demand) - 1)

    ax2.hist(
        demand, bins=25, orientation="horizontal",
        color=COLORS["E2"], alpha=0.6, edgecolor="white", linewidth=0.3,
    )
    ax2.axhline(mu, color=COLORS["grid"], linestyle="--", linewidth=0.5)
    ax2.set_xlabel("Freq.")
    ax2.set_ylim(ax1.get_ylim())
    ax2.set_yticklabels([])

    return fig

deepbullwhip.diagnostics.plots.plot_order_quantities(demand, sim_result, width='double')

Stacked subplots: customer demand + order quantity per echelon.

Source code in deepbullwhip/diagnostics/plots.py

def plot_order_quantities(
    demand: TimeSeries,
    sim_result: SimulationResult,
    width: Literal["single", "double"] = "double",
) -> matplotlib.figure.Figure:
    """Stacked subplots: customer demand + order quantity per echelon."""
    _apply_style()
    K = len(sim_result.echelon_results)
    w = _col_width(width)
    fig, axes = plt.subplots(
        K + 1, 1, figsize=(w, w * 0.85), sharex=True,
    )
    weeks = np.arange(len(demand))

    # Customer demand
    axes[0].plot(weeks, demand, color=COLORS["demand"], linewidth=0.6)
    axes[0].set_ylabel("Demand")
    axes[0].text(
        0.98, 0.92, "Customer Demand", transform=axes[0].transAxes,
        ha="right", va="top", fontsize=7, fontstyle="italic",
    )

    for k, er in enumerate(sim_result.echelon_results):
        ax = axes[k + 1]
        ax.plot(weeks, er.orders, color=_echelon_color(k), linewidth=0.6)
        ax.axhline(er.orders.mean(), color=COLORS["grid"], linestyle="--", linewidth=0.4)
        ax.set_ylabel("Orders")
        ax.text(
            0.98, 0.92, f"E{k + 1}: {er.name}",
            transform=ax.transAxes, ha="right", va="top",
            fontsize=7, fontstyle="italic",
        )

    axes[-1].set_xlabel("Week")
    axes[0].set_xlim(0, len(demand) - 1)

    return fig

deepbullwhip.diagnostics.plots.plot_inventory_levels(sim_result, width='double')

Inventory on-hand per echelon with zero-line and backorder shading.

Source code in deepbullwhip/diagnostics/plots.py

def plot_inventory_levels(
    sim_result: SimulationResult,
    width: Literal["single", "double"] = "double",
) -> matplotlib.figure.Figure:
    """Inventory on-hand per echelon with zero-line and backorder shading."""
    _apply_style()
    K = len(sim_result.echelon_results)
    w = _col_width(width)
    fig, axes = plt.subplots(K, 1, figsize=(w, w * 0.7), sharex=True)
    if K == 1:
        axes = [axes]

    for k, er in enumerate(sim_result.echelon_results):
        ax = axes[k]
        T = len(er.inventory_levels)
        weeks = np.arange(T)
        inv = er.inventory_levels

        # Shade backorder region
        ax.fill_between(
            weeks, inv, 0,
            where=inv < 0, color=COLORS["backorder"], alpha=0.15,
            label="Backorder",
        )
        ax.fill_between(
            weeks, inv, 0,
            where=inv >= 0, color=COLORS["holding"], alpha=0.10,
            label="On-hand",
        )
        ax.plot(weeks, inv, color=_echelon_color(k), linewidth=0.6)
        ax.axhline(0, color=COLORS["demand"], linewidth=0.4)

        ax.set_ylabel("Inventory")
        ax.text(
            0.98, 0.92,
            f"E{k + 1}: {er.name}  |  FR={er.fill_rate:.0%}",
            transform=ax.transAxes, ha="right", va="top", fontsize=7,
        )
        if k == 0:
            ax.legend(loc="upper left", ncol=2)

    axes[-1].set_xlabel("Week")
    axes[0].set_xlim(0, T - 1)

    return fig

deepbullwhip.diagnostics.plots.plot_inventory_position(demand, sim_result, chain, width='double')

Inventory position = on-hand + pipeline for each echelon.

Parameters:

Name	Type	Description	Default
chain	SerialSupplyChain	The chain object after simulate() has been called, so that echelon pipeline state is available.	required

Source code in deepbullwhip/diagnostics/plots.py

def plot_inventory_position(
    demand: TimeSeries,
    sim_result: SimulationResult,
    chain,
    width: Literal["single", "double"] = "double",
) -> matplotlib.figure.Figure:
    """Inventory position = on-hand + pipeline for each echelon.

    Parameters
    ----------
    chain : SerialSupplyChain
        The chain object *after* simulate() has been called, so that
        echelon pipeline state is available.
    """
    _apply_style()
    K = len(sim_result.echelon_results)
    w = _col_width(width)
    fig, axes = plt.subplots(K, 1, figsize=(w, w * 0.7), sharex=True)
    if K == 1:
        axes = [axes]

    for k, er in enumerate(sim_result.echelon_results):
        ax = axes[k]
        T = len(er.inventory_levels)
        weeks = np.arange(T)

        # Reconstruct inventory position from on-hand + pipeline snapshot
        # We can approximate IP from orders and inventory arrays:
        # IP(t) = inv(t) + sum of orders in pipeline at time t
        inv = er.inventory_levels
        orders = er.orders
        L = chain.echelons[k].lead_time

        # Build pipeline sum: at each t, pipeline = orders[t-L+1 .. t]
        pipeline_sum = np.zeros(T)
        for t in range(T):
            start = max(0, t - L + 1)
            pipeline_sum[t] = orders[start : t + 1].sum()

        ip = inv + pipeline_sum

        ax.plot(weeks, ip, color=_echelon_color(k), linewidth=0.6, label="Inv. Position")
        ax.plot(weeks, inv, color=_echelon_color(k), linewidth=0.4, linestyle="--",
                alpha=0.5, label="On-hand only")
        ax.axhline(0, color=COLORS["demand"], linewidth=0.4)

        ax.set_ylabel("Units")
        ax.text(
            0.98, 0.92, f"E{k + 1}: {er.name} (L={L})",
            transform=ax.transAxes, ha="right", va="top", fontsize=7,
        )
        if k == 0:
            ax.legend(loc="upper left", ncol=2)

    axes[-1].set_xlabel("Week")
    axes[0].set_xlim(0, T - 1)

    return fig

deepbullwhip.diagnostics.plots.plot_order_streams(demand, sim_result, echelon_indices=None, width='double')

All echelon order streams overlaid on customer demand.

Source code in deepbullwhip/diagnostics/plots.py

def plot_order_streams(
    demand: TimeSeries,
    sim_result: SimulationResult,
    echelon_indices: list[int] | None = None,
    width: Literal["single", "double"] = "double",
) -> matplotlib.figure.Figure:
    """All echelon order streams overlaid on customer demand."""
    _apply_style()
    if echelon_indices is None:
        echelon_indices = list(range(len(sim_result.echelon_results)))

    w = _col_width(width)
    fig, ax = plt.subplots(figsize=(w, w / GOLDEN / 1.4))
    weeks = np.arange(len(demand))

    ax.plot(
        weeks, demand, color=COLORS["demand"], linewidth=1.0,
        label="Customer Demand", zorder=10,
    )
    for idx in echelon_indices:
        er = sim_result.echelon_results[idx]
        ax.plot(
            weeks, er.orders, color=_echelon_color(idx), linewidth=0.5,
            alpha=0.8, label=f"E{idx + 1}: {er.name}",
        )

    ax.set_xlabel("Week")
    ax.set_ylabel("Quantity (units)")
    ax.legend(loc="upper left", ncol=2)
    ax.set_xlim(0, len(demand) - 1)

    return fig

deepbullwhip.diagnostics.plots.plot_cost_timeseries(sim_result, width='double')

Per-period cost for each echelon, stacked vertically.

Source code in deepbullwhip/diagnostics/plots.py

def plot_cost_timeseries(
    sim_result: SimulationResult,
    width: Literal["single", "double"] = "double",
) -> matplotlib.figure.Figure:
    """Per-period cost for each echelon, stacked vertically."""
    _apply_style()
    K = len(sim_result.echelon_results)
    w = _col_width(width)
    fig, axes = plt.subplots(K, 1, figsize=(w, w * 0.7), sharex=True)
    if K == 1:
        axes = [axes]

    for k, er in enumerate(sim_result.echelon_results):
        ax = axes[k]
        T = len(er.costs)
        weeks = np.arange(T)
        inv = er.inventory_levels

        holding_mask = inv >= 0
        costs_h = np.where(holding_mask, er.costs, 0)
        costs_b = np.where(~holding_mask, er.costs, 0)

        ax.bar(weeks, costs_h, width=1.0, color=COLORS["holding"], alpha=0.6, label="Holding")
        ax.bar(weeks, costs_b, width=1.0, color=COLORS["backorder"], alpha=0.6, label="Backorder")

        ax.set_ylabel("Cost")
        ax.text(
            0.98, 0.92,
            f"E{k + 1}: {er.name}  |  Total={er.total_cost:,.0f}",
            transform=ax.transAxes, ha="right", va="top", fontsize=7,
        )
        if k == 0:
            ax.legend(loc="upper left", ncol=2)

    axes[-1].set_xlabel("Week")
    axes[0].set_xlim(0, T - 1)

    return fig

deepbullwhip.diagnostics.plots.plot_cost_decomposition(results_by_model, width='double')

Stacked bar: holding vs backorder cost per model.

Source code in deepbullwhip/diagnostics/plots.py

def plot_cost_decomposition(
    results_by_model: dict[str, SimulationResult],
    width: Literal["single", "double"] = "double",
) -> matplotlib.figure.Figure:
    """Stacked bar: holding vs backorder cost per model."""
    _apply_style()
    w = _col_width(width)
    fig, ax = plt.subplots(figsize=(w, w / GOLDEN / 1.4))

    names = list(results_by_model.keys())
    x = np.arange(len(names))

    holding_costs = []
    backorder_costs = []
    for result in results_by_model.values():
        h_total = 0.0
        b_total = 0.0
        for er in result.echelon_results:
            inv = er.inventory_levels
            costs = er.costs
            h_mask = inv >= 0
            h_total += float(costs[h_mask].sum())
            b_total += float(costs[~h_mask].sum())
        holding_costs.append(h_total)
        backorder_costs.append(b_total)

    bar_w = 0.6
    ax.bar(x, holding_costs, bar_w, label="Holding", color=COLORS["holding"], edgecolor="white", linewidth=0.3)
    ax.bar(x, backorder_costs, bar_w, bottom=holding_costs, label="Backorder",
           color=COLORS["backorder"], edgecolor="white", linewidth=0.3)

    ax.set_xticks(x)
    ax.set_xticklabels(names, rotation=30, ha="right")
    ax.set_ylabel("Total Cost")
    ax.legend(loc="upper right")

    return fig

deepbullwhip.diagnostics.plots.plot_bullwhip_amplification(results_by_model, echelon_labels=None, width='single')

Log-scale cumulative bullwhip ratio across echelons.

Source code in deepbullwhip/diagnostics/plots.py

def plot_bullwhip_amplification(
    results_by_model: dict[str, SimulationResult],
    echelon_labels: list[str] | None = None,
    width: Literal["single", "double"] = "single",
) -> matplotlib.figure.Figure:
    """Log-scale cumulative bullwhip ratio across echelons."""
    _apply_style()
    w = _col_width(width)
    fig, ax = plt.subplots(figsize=(w, w / GOLDEN))

    for name, result in results_by_model.items():
        bw_ratios = [er.bullwhip_ratio for er in result.echelon_results]
        cum_bw = np.cumprod(bw_ratios)
        x = np.arange(1, len(cum_bw) + 1)
        ax.plot(x, cum_bw, marker="o", markersize=4, linewidth=1.0, label=name)

    ax.set_yscale("log")
    ax.set_xlabel("Echelon")
    ax.set_ylabel("Cumulative Bullwhip Ratio")
    if echelon_labels:
        ax.set_xticks(range(1, len(echelon_labels) + 1))
        ax.set_xticklabels(echelon_labels)
    ax.legend()
    ax.yaxis.set_major_formatter(ticker.ScalarFormatter())
    ax.grid(True, alpha=0.2, linewidth=0.3)

    return fig

deepbullwhip.diagnostics.plots.plot_summary_dashboard(demand, sim_result, width='double')

4-panel dashboard: demand, orders overlay, inventory, BW ratios.

Source code in deepbullwhip/diagnostics/plots.py

def plot_summary_dashboard(
    demand: TimeSeries,
    sim_result: SimulationResult,
    width: Literal["single", "double"] = "double",
) -> matplotlib.figure.Figure:
    """4-panel dashboard: demand, orders overlay, inventory, BW ratios."""
    _apply_style()
    w = _col_width(width)
    fig, axes = plt.subplots(2, 2, figsize=(w, w * 0.75))
    weeks = np.arange(len(demand))
    K = len(sim_result.echelon_results)

    # (a) Demand
    ax = axes[0, 0]
    ax.plot(weeks, demand, color=COLORS["demand"], linewidth=0.6)
    ax.axhline(demand.mean(), color=COLORS["grid"], linestyle="--", linewidth=0.4)
    ax.set_ylabel("Demand")
    ax.set_title("(a) Customer Demand", fontsize=8)

    # (b) Order streams
    ax = axes[0, 1]
    ax.plot(weeks, demand, color=COLORS["demand"], linewidth=0.8, label="Demand")
    for k, er in enumerate(sim_result.echelon_results):
        ax.plot(weeks, er.orders, color=_echelon_color(k), linewidth=0.4,
                alpha=0.7, label=f"E{k + 1}")
    ax.set_ylabel("Orders")
    ax.set_title("(b) Order Streams", fontsize=8)
    ax.legend(loc="upper left", ncol=3, fontsize=6)

    # (c) Inventory
    ax = axes[1, 0]
    for k, er in enumerate(sim_result.echelon_results):
        ax.plot(weeks, er.inventory_levels, color=_echelon_color(k),
                linewidth=0.5, label=f"E{k + 1}")
    ax.axhline(0, color=COLORS["demand"], linewidth=0.4)
    ax.set_xlabel("Week")
    ax.set_ylabel("Inventory")
    ax.set_title("(c) Inventory On-Hand", fontsize=8)
    ax.legend(loc="lower left", ncol=2, fontsize=6)

    # (d) Bullwhip & fill rate
    ax = axes[1, 1]
    bw = [er.bullwhip_ratio for er in sim_result.echelon_results]
    fr = [er.fill_rate for er in sim_result.echelon_results]
    x = np.arange(1, K + 1)
    labels = [f"E{k + 1}" for k in range(K)]

    bar_w = 0.35
    bars1 = ax.bar(x - bar_w / 2, bw, bar_w, color=COLORS["E1"], label="BW Ratio")
    ax2 = ax.twinx()
    bars2 = ax2.bar(x + bar_w / 2, fr, bar_w, color=COLORS["E3"], alpha=0.7, label="Fill Rate")
    ax.set_xticks(x)
    ax.set_xticklabels(labels)
    ax.set_ylabel("Bullwhip Ratio")
    ax2.set_ylabel("Fill Rate")
    ax2.set_ylim(0, 1.1)
    ax.set_title("(d) Bullwhip & Fill Rate", fontsize=8)

    # Combined legend
    lines = [bars1, bars2]
    labs = [item.get_label() for item in lines]
    ax.legend(lines, labs, loc="upper left", fontsize=6)

    for a in [axes[0, 0], axes[0, 1]]:
        a.set_xlim(0, len(demand) - 1)
    axes[1, 0].set_xlim(0, len(demand) - 1)

    return fig

deepbullwhip.diagnostics.plots.plot_echelon_detail(demand, sim_result, echelon_index=0, width='double')

3-row detail for one echelon: orders vs demand, inventory, costs.

Source code in deepbullwhip/diagnostics/plots.py

def plot_echelon_detail(
    demand: TimeSeries,
    sim_result: SimulationResult,
    echelon_index: int = 0,
    width: Literal["single", "double"] = "double",
) -> matplotlib.figure.Figure:
    """3-row detail for one echelon: orders vs demand, inventory, costs."""
    _apply_style()
    er = sim_result.echelon_results[echelon_index]
    w = _col_width(width)
    fig, axes = plt.subplots(3, 1, figsize=(w, w * 0.65), sharex=True)
    weeks = np.arange(len(demand))
    color = _echelon_color(echelon_index)

    # Orders vs demand
    ax = axes[0]
    ax.plot(weeks, demand, color=COLORS["demand"], linewidth=0.6, label="Demand")
    ax.plot(weeks, er.orders, color=color, linewidth=0.6, label=f"E{echelon_index + 1} Orders")
    ax.set_ylabel("Quantity")
    ax.legend(loc="upper left", ncol=2)
    ax.set_title(f"Echelon {echelon_index + 1}: {er.name}", fontsize=9)

    # Inventory
    ax = axes[1]
    ax.fill_between(weeks, er.inventory_levels, 0,
                    where=er.inventory_levels >= 0, color=COLORS["holding"], alpha=0.15)
    ax.fill_between(weeks, er.inventory_levels, 0,
                    where=er.inventory_levels < 0, color=COLORS["backorder"], alpha=0.15)
    ax.plot(weeks, er.inventory_levels, color=color, linewidth=0.6)
    ax.axhline(0, color=COLORS["demand"], linewidth=0.4)
    ax.set_ylabel("Inventory")

    # Costs
    ax = axes[2]
    inv = er.inventory_levels
    holding_mask = inv >= 0
    ax.bar(weeks[holding_mask], er.costs[holding_mask], width=1.0,
           color=COLORS["holding"], alpha=0.6, label="Holding")
    ax.bar(weeks[~holding_mask], er.costs[~holding_mask], width=1.0,
           color=COLORS["backorder"], alpha=0.6, label="Backorder")
    ax.set_ylabel("Cost")
    ax.set_xlabel("Week")
    ax.legend(loc="upper left", ncol=2)

    axes[0].set_xlim(0, len(demand) - 1)

    return fig

Network & Map

deepbullwhip.diagnostics.network.NodeLocation dataclass

Geographic or schematic location of a supply chain node.

Source code in deepbullwhip/diagnostics/network.py

@dataclass
class NodeLocation:
    """Geographic or schematic location of a supply chain node."""

    name: str
    lat: float
    lon: float
    role: str = ""
    details: str = ""

deepbullwhip.diagnostics.network.SupplyChainNetwork dataclass

Describes the topology and geography of a supply chain.

This is the matplotlib-based visualization data model. For Graphviz-based rendering, see :mod:deepbullwhip.diagnostics.graphviz_viz.

Source code in deepbullwhip/diagnostics/network.py

@dataclass
class SupplyChainNetwork:
    """Describes the topology and geography of a supply chain.

    This is the matplotlib-based visualization data model. For
    Graphviz-based rendering, see
    :mod:`deepbullwhip.diagnostics.graphviz_viz`.
    """

    nodes: list[NodeLocation]
    edges: list[tuple[int, int]] = field(default_factory=list)

    def __post_init__(self) -> None:
        if not self.edges and len(self.nodes) > 1:
            # Default: serial chain E_K -> ... -> E_1
            self.edges = [(i + 1, i) for i in range(len(self.nodes) - 1)]

    @classmethod
    def from_graph(
        cls,
        graph: "SupplyChainGraph",
        locations: dict[str, tuple[float, float]] | None = None,
    ) -> SupplyChainNetwork:
        """Create a visualization network from a :class:`SupplyChainGraph`.

        Parameters
        ----------
        graph : SupplyChainGraph
            The supply chain graph to convert.
        locations : dict[str, tuple[float, float]] or None
            Optional mapping from node name to ``(lat, lon)``
            coordinates. Nodes without coordinates are placed at
            ``(0, 0)``.

        Returns
        -------
        SupplyChainNetwork
        """

        node_names = list(graph.nodes.keys())
        name_to_idx = {name: i for i, name in enumerate(node_names)}

        nodes = []
        for name, cfg in graph.nodes.items():
            lat, lon = (0.0, 0.0)
            if locations and name in locations:
                lat, lon = locations[name]
            nodes.append(
                NodeLocation(
                    name=name,
                    lat=lat,
                    lon=lon,
                    role=cfg.name,
                )
            )

        edges = [
            (name_to_idx[u], name_to_idx[v])
            for u, v in graph.edges
        ]

        return cls(nodes=nodes, edges=edges)

from_graph(graph, locations=None) classmethod

Create a visualization network from a :class:SupplyChainGraph.

Parameters:

Name	Type	Description	Default
graph	SupplyChainGraph	The supply chain graph to convert.	required
locations	dict[str, tuple[float, float]] or None	Optional mapping from node name to (lat, lon) coordinates. Nodes without coordinates are placed at (0, 0).	None

Returns:

Type	Description
SupplyChainNetwork

Source code in deepbullwhip/diagnostics/network.py

@classmethod
def from_graph(
    cls,
    graph: "SupplyChainGraph",
    locations: dict[str, tuple[float, float]] | None = None,
) -> SupplyChainNetwork:
    """Create a visualization network from a :class:`SupplyChainGraph`.

    Parameters
    ----------
    graph : SupplyChainGraph
        The supply chain graph to convert.
    locations : dict[str, tuple[float, float]] or None
        Optional mapping from node name to ``(lat, lon)``
        coordinates. Nodes without coordinates are placed at
        ``(0, 0)``.

    Returns
    -------
    SupplyChainNetwork
    """

    node_names = list(graph.nodes.keys())
    name_to_idx = {name: i for i, name in enumerate(node_names)}

    nodes = []
    for name, cfg in graph.nodes.items():
        lat, lon = (0.0, 0.0)
        if locations and name in locations:
            lat, lon = locations[name]
        nodes.append(
            NodeLocation(
                name=name,
                lat=lat,
                lon=lon,
                role=cfg.name,
            )
        )

    edges = [
        (name_to_idx[u], name_to_idx[v])
        for u, v in graph.edges
    ]

    return cls(nodes=nodes, edges=edges)

deepbullwhip.diagnostics.network.plot_network_diagram(network, sim_result=None, width='double', orientation='horizontal')

Abstract network diagram of the supply chain.

Nodes are drawn as rounded rectangles with role labels. Edges show material flow direction. If sim_result is provided, node annotations include BW ratio and fill rate.

Source code in deepbullwhip/diagnostics/network.py

def plot_network_diagram(
    network: SupplyChainNetwork,
    sim_result: SimulationResult | None = None,
    width: Literal["single", "double"] = "double",
    orientation: Literal["horizontal", "vertical"] = "horizontal",
) -> matplotlib.figure.Figure:
    """Abstract network diagram of the supply chain.

    Nodes are drawn as rounded rectangles with role labels. Edges show
    material flow direction. If sim_result is provided, node annotations
    include BW ratio and fill rate.
    """
    _apply_style()
    w = SINGLE_COL if width == "single" else DOUBLE_COL
    K = len(network.nodes)

    if orientation == "horizontal":
        fig, ax = plt.subplots(figsize=(w, w / GOLDEN / 1.8))
        positions = [(i * 1.0 / (K - 1) if K > 1 else 0.5, 0.5) for i in range(K)]
    else:
        fig, ax = plt.subplots(figsize=(w / 1.5, w * 0.9))
        positions = [(0.5, 1.0 - i * 1.0 / (K - 1) if K > 1 else 0.5) for i in range(K)]

    ax.set_xlim(-0.15, 1.15)
    ax.set_ylim(-0.1, 1.1)
    ax.set_aspect("equal")
    ax.axis("off")

    node_w = 0.18
    node_h = 0.22 if orientation == "horizontal" else 0.12

    # Draw edges first (behind nodes)
    for src, dst in network.edges:
        x0, y0 = positions[src]
        x1, y1 = positions[dst]
        ax.annotate(
            "", xy=(x1, y1), xytext=(x0, y0),
            arrowprops=dict(
                arrowstyle="-|>", color="#888888", lw=1.2,
                connectionstyle="arc3,rad=0.0",
            ),
        )

    # Draw nodes
    for k, (node, (x, y)) in enumerate(zip(network.nodes, positions)):
        color = _echelon_color(k)
        rect = mpatches.FancyBboxPatch(
            (x - node_w / 2, y - node_h / 2), node_w, node_h,
            boxstyle="round,pad=0.02",
            facecolor=color, edgecolor="black", linewidth=0.6, alpha=0.25,
        )
        ax.add_patch(rect)

        # Node label
        label_lines = [f"E{k + 1}: {node.role}"]
        if node.name:
            label_lines.insert(0, node.name)

        if sim_result is not None and k < len(sim_result.echelon_results):
            er = sim_result.echelon_results[k]
            label_lines.append(f"BW={er.bullwhip_ratio:.2f}  FR={er.fill_rate:.0%}")

        ax.text(
            x, y, "\n".join(label_lines),
            ha="center", va="center", fontsize=6,
            fontweight="bold" if k == 0 else "normal",
            linespacing=1.4,
        )

    # Flow label
    if orientation == "horizontal":
        ax.text(0.5, -0.02, r"Material flow $\longrightarrow$",
                ha="center", va="top", fontsize=7, color="#666666")
    else:
        ax.text(0.85, 0.5, r"Material flow $\downarrow$",
                ha="left", va="center", fontsize=7, color="#666666", rotation=90)

    return fig

deepbullwhip.diagnostics.network.plot_supply_chain_map(network, sim_result=None, width='double', map_bounds=None, show_country_outline=True)

Geographic visualization of supply chain nodes on a lat/lon plot.

Plots nodes at their geographic coordinates with connecting arcs. If sim_result is provided, node size scales with total cost and color intensity with bullwhip ratio.

Parameters:

Name	Type	Description	Default
map_bounds	(lat_min, lat_max, lon_min, lon_max) or None	If None, computed from node positions with padding.	None
show_country_outline	bool	If True, draws a simplified Saudi Arabia outline.	True

Source code in deepbullwhip/diagnostics/network.py

def plot_supply_chain_map(
    network: SupplyChainNetwork,
    sim_result: SimulationResult | None = None,
    width: Literal["single", "double"] = "double",
    map_bounds: tuple[float, float, float, float] | None = None,
    show_country_outline: bool = True,
) -> matplotlib.figure.Figure:
    """Geographic visualization of supply chain nodes on a lat/lon plot.

    Plots nodes at their geographic coordinates with connecting arcs.
    If sim_result is provided, node size scales with total cost and
    color intensity with bullwhip ratio.

    Parameters
    ----------
    map_bounds : (lat_min, lat_max, lon_min, lon_max) or None
        If None, computed from node positions with padding.
    show_country_outline : bool
        If True, draws a simplified Saudi Arabia outline.
    """
    _apply_style()
    w = SINGLE_COL if width == "single" else DOUBLE_COL
    fig, ax = plt.subplots(figsize=(w, w * 0.85))

    lats = [n.lat for n in network.nodes]
    lons = [n.lon for n in network.nodes]

    if map_bounds is None:
        pad_lat = max(1.5, (max(lats) - min(lats)) * 0.25)
        pad_lon = max(1.5, (max(lons) - min(lons)) * 0.25)
        map_bounds = (
            min(lats) - pad_lat, max(lats) + pad_lat,
            min(lons) - pad_lon, max(lons) + pad_lon,
        )

    if show_country_outline:
        _draw_saudi_outline(ax)

    # Draw edges (arcs)
    for src, dst in network.edges:
        n_src, n_dst = network.nodes[src], network.nodes[dst]
        ax.annotate(
            "", xy=(n_dst.lon, n_dst.lat), xytext=(n_src.lon, n_src.lat),
            arrowprops=dict(
                arrowstyle="-|>", color="#888888", lw=0.8,
                connectionstyle="arc3,rad=0.15",
            ),
            zorder=2,
        )

    # Draw nodes
    for k, node in enumerate(network.nodes):
        color = _echelon_color(k)
        size = 80
        alpha = 0.85

        if sim_result is not None and k < len(sim_result.echelon_results):
            er = sim_result.echelon_results[k]
            size = 60 + 40 * min(er.bullwhip_ratio, 5)
            alpha = 0.7 + 0.06 * min(er.bullwhip_ratio, 5)

        ax.scatter(
            node.lon, node.lat, s=size, c=color, edgecolors="black",
            linewidths=0.5, alpha=alpha, zorder=5,
        )

        # Label
        label = f"E{k + 1}: {node.name}"
        if sim_result is not None and k < len(sim_result.echelon_results):
            er = sim_result.echelon_results[k]
            label += f"\nBW={er.bullwhip_ratio:.2f}"

        ax.annotate(
            label, (node.lon, node.lat),
            textcoords="offset points", xytext=(8, 8),
            fontsize=5.5, ha="left", va="bottom",
            bbox=dict(boxstyle="round,pad=0.2", fc="white", ec=color, alpha=0.85, lw=0.4),
            zorder=6,
        )

    ax.set_xlim(map_bounds[2], map_bounds[3])
    ax.set_ylim(map_bounds[0], map_bounds[1])
    ax.set_xlabel("Longitude")
    ax.set_ylabel("Latitude")
    ax.set_aspect(1.0 / np.cos(np.radians(np.mean(lats))))
    ax.grid(True, alpha=0.15, linewidth=0.3)

    return fig

deepbullwhip.diagnostics.network.kfupm_petrochemical_network()

Example 4-echelon petrochemical supply chain in Saudi Arabia.

A petrochemical product (polyethylene) supply chain involving KFUPM as R&D partner, sourced from Eastern Province refineries, processed through Jubail industrial complex, and distributed domestically.

Source code in deepbullwhip/diagnostics/network.py

def kfupm_petrochemical_network() -> SupplyChainNetwork:
    """Example 4-echelon petrochemical supply chain in Saudi Arabia.

    A petrochemical product (polyethylene) supply chain involving KFUPM
    as R&D partner, sourced from Eastern Province refineries, processed
    through Jubail industrial complex, and distributed domestically.
    """
    nodes = [
        NodeLocation(
            name="KFUPM / Distributor",
            lat=26.3073, lon=50.1433,
            role="Distributor / OEM",
            details="KFUPM Dhahran campus\nR&D + regional distribution hub",
        ),
        NodeLocation(
            name="Jubail OSAT Plant",
            lat=27.0046, lon=49.6588,
            role="Assembly & Processing",
            details="Jubail Industrial City\nPolymer compounding & packaging",
        ),
        NodeLocation(
            name="SABIC / Yanbu Refinery",
            lat=24.0895, lon=38.0618,
            role="Foundry / Manufacturing",
            details="Yanbu Industrial City\nPetrochemical production (SABIC)",
        ),
        NodeLocation(
            name="Aramco Raw Materials",
            lat=25.3838, lon=49.9164,
            role="Raw Material Supplier",
            details="Abqaiq / Ras Tanura\nCrude oil & naphtha feedstock",
        ),
    ]
    return SupplyChainNetwork(nodes=nodes)