feat(experiments): added actuals evaluator to properly compare actuals costs against forecasted ones

Author: chripiermarini

experiments/actuals_evaluator.py

@@ -0,0 +1,183 @@
+"""Evaluate a completed experiment's plan against actual demand.
+
+The optimizer only sees forecasted demand, so its plan may over- or under-ship
+relative to what actually happens.  This module simulates the planned flows
+against ground-truth demand to produce realized (as-executed) metrics,
+both in aggregate and broken down per destination.
+
+Usage:
+    from actuals_evaluator import evaluate, save_realized_metrics
+    metrics = evaluate(output_path)
+    save_realized_metrics(metrics, output_path)
+"""
+
+import json
+from dataclasses import asdict, dataclass
+from pathlib import Path
+
+import polars as pl
+
+from data.ingestion import Reader
+from experiment_config import load_experiment_config
+from utils.system_paths import get_project_root
+
+
+@dataclass
+class DestinationActuals:
+    transport_cost: float
+    realized_holding_cost: float
+    realized_total_cost: float
+    total_actual_demand: float
+    total_fulfilled: float
+    total_shortage: float
+    fill_rate: float
+    cost_per_unit_demanded: float
+    cost_per_unit_fulfilled: float
+
+
+@dataclass
+class RealizedMetrics:
+    transport_cost: float
+    realized_holding_cost: float
+    realized_total_cost: float
+    total_actual_demand: float
+    total_fulfilled: float
+    total_shortage: float
+    fill_rate: float
+    cost_per_unit_demanded: float
+    cost_per_unit_fulfilled: float
+    per_destination: dict[str, DestinationActuals]
+
+
+def evaluate(experiment_output_path: Path) -> RealizedMetrics:
+    """Simulate the experiment's planned flows against actual demand.
+
+    Reads flows.parquet and metrics.json from experiment_output_path, plus the
+    original dataset referenced in config.yaml, and returns realized metrics.
+    """
+    project_root = get_project_root()
+    config = load_experiment_config(
+        project_root, experiment_output_path / "config.yaml"
+    )
+
+    flows = pl.read_parquet(experiment_output_path / "flows.parquet")
+
+    raw = Reader(config.dataset_path).read()
+
+    holding_cost_map: dict[str, float] = {}
+    if "holding_cost" in raw.destinations.columns:
+        holding_cost_map = dict(
+            zip(
+                raw.destinations["destination_id"].to_list(),
+                raw.destinations["holding_cost"].to_list(),
+            )
+        )
+
+    # Per-destination transport cost from planned flows × lane unit costs
+    lane_cost_map: dict[tuple, float] = {
+        (row["origin_id"], row["destination_id"]): row["unit_cost"]
+        for row in raw.lanes.to_dicts()
+    }
+    dest_transport_cost: dict[str, float] = {}
+    for row in flows.to_dicts():
+        d_id = row["destination_id"]
+        cost = lane_cost_map.get((row["origin_id"], d_id), 0.0) * row["flow"]
+        dest_transport_cost[d_id] = dest_transport_cost.get(d_id, 0.0) + cost
+
+    forecast_dates = sorted(flows["period"].unique().to_list())
+
+    actual_demand = raw.demand_history.filter(pl.col("date").is_in(forecast_dates))
+
+    # Aggregate planned inflows per (destination, period)
+    inflows = (
+        flows.group_by(["destination_id", "period"])
+        .agg(pl.col("flow").sum().alias("inflow"))
+        .rename({"period": "date"})
+    )
+
+    destinations = sorted(raw.destinations["destination_id"].to_list())
+
+    total_actual_demand = 0.0
+    total_fulfilled = 0.0
+    total_shortage = 0.0
+    realized_holding_cost = 0.0
+    per_destination: dict[str, DestinationActuals] = {}
+
+    for d_id in destinations:
+        h_cost = holding_cost_map.get(d_id, 0.0)
+
+        inflow_map: dict = {
+            row["date"]: row["inflow"]
+            for row in inflows.filter(pl.col("destination_id") == d_id).to_dicts()
+        }
+        demand_map: dict = {
+            row["date"]: row["demand"]
+            for row in actual_demand.filter(pl.col("destination_id") == d_id).to_dicts()
+        }
+
+        d_fulfilled = 0.0
+        d_shortage = 0.0
+        d_holding_cost = 0.0
+        d_demand = 0.0
+        inventory = 0.0
+
+        for t in forecast_dates:
+            inflow = inflow_map.get(t, 0.0)
+            actual_d = demand_map.get(t, 0.0)
+
+            available = inventory + inflow
+            fulfilled = min(actual_d, available)
+            shortage = max(0.0, actual_d - available)
+            inventory = available - fulfilled
+
+            d_demand += actual_d
+            d_fulfilled += fulfilled
+            d_shortage += shortage
+            d_holding_cost += h_cost * inventory
+
+        d_transport = dest_transport_cost.get(d_id, 0.0)
+        d_realized_total = d_transport + d_holding_cost
+        d_fill_rate = d_fulfilled / d_demand if d_demand > 0 else 1.0
+        d_cpu_demanded = d_realized_total / d_demand if d_demand > 0 else 0.0
+        d_cpu_fulfilled = d_realized_total / d_fulfilled if d_fulfilled > 0 else 0.0
+
+        per_destination[d_id] = DestinationActuals(
+            transport_cost=d_transport,
+            realized_holding_cost=d_holding_cost,
+            realized_total_cost=d_realized_total,
+            total_actual_demand=d_demand,
+            total_fulfilled=d_fulfilled,
+            total_shortage=d_shortage,
+            fill_rate=d_fill_rate,
+            cost_per_unit_demanded=d_cpu_demanded,
+            cost_per_unit_fulfilled=d_cpu_fulfilled,
+        )
+
+        total_actual_demand += d_demand
+        total_fulfilled += d_fulfilled
+        total_shortage += d_shortage
+        realized_holding_cost += d_holding_cost
+
+    transport_cost = sum(dest_transport_cost.values())
+    realized_total_cost = transport_cost + realized_holding_cost
+    fill_rate = total_fulfilled / total_actual_demand if total_actual_demand > 0 else 1.0
+    cpu_demanded = realized_total_cost / total_actual_demand if total_actual_demand > 0 else 0.0
+    cpu_fulfilled = realized_total_cost / total_fulfilled if total_fulfilled > 0 else 0.0
+
+    return RealizedMetrics(
+        transport_cost=transport_cost,
+        realized_holding_cost=realized_holding_cost,
+        realized_total_cost=realized_total_cost,
+        total_actual_demand=total_actual_demand,
+        total_fulfilled=total_fulfilled,
+        total_shortage=total_shortage,
+        fill_rate=fill_rate,
+        cost_per_unit_demanded=cpu_demanded,
+        cost_per_unit_fulfilled=cpu_fulfilled,
+        per_destination=per_destination,
+    )
+
+
+def save_realized_metrics(metrics: RealizedMetrics, output_path: Path) -> None:
+    with open(output_path / "realized_metrics.json", "w") as f:
+        json.dump(asdict(metrics), f, indent=2)

experiments/run_all.py

@@ -22,7 +22,6 @@
     _HERE / "configs" / "baseline_naive.yaml",
     _HERE / "configs" / "baseline_global_ets.yaml",
     _HERE / "configs" / "model_selection.yaml",
-    _HERE / "configs" / "scale_test.yaml",
 ]
 
 
@@ -53,37 +52,97 @@ def _print_summary() -> None:
     for config_path in CONFIGS:
         experiment_name = config_path.stem
         metrics_path = results_root / experiment_name / "metrics.json"
+        realized_path = results_root / experiment_name / "realized_metrics.json"
+
         if not metrics_path.exists():
-            rows.append((experiment_name, "MISSING", "-", "-", "-", "-"))
+            rows.append((experiment_name, "MISSING", "-", "-", "-", "-", "-"))
             continue
 
         with open(metrics_path) as f:
             m = json.load(f)
 
         agg = m.get("aggregated_forecast", {})
         costs = m.get("costs", {})
+        planned_cost = costs.get("total_cost", float("nan"))
+
+        if realized_path.exists():
+            with open(realized_path) as f:
+                r = json.load(f)
+            realized_cost = r.get("realized_total_cost", float("nan"))
+            fill_rate = r.get("fill_rate", float("nan"))
+        else:
+            realized_cost = float("nan")
+            fill_rate = float("nan")
+
         rows.append(
             (
                 experiment_name,
                 "OK",
                 f"{agg.get('mean_wape', float('nan')):.4f}",
-                f"{costs.get('total_cost', float('nan')):.2f}",
-                f"{costs.get('transportation_cost', float('nan')):.2f}",
-                f"{costs.get('holding_cost', float('nan')):.2f}",
+                f"{planned_cost:.2f}",
+                f"{realized_cost:.2f}",
+                f"{fill_rate:.4f}",
             )
         )
 
-    header = f"{'Experiment':<30} {'Status':<8} {'WAPE':>8} {'Total':>12} {'Transport':>12} {'Holding':>10}"
+    header = (
+        f"{'Experiment':<30} {'Status':<8} {'WAPE':>8}"
+        f" {'Planned Cost':>14} {'Realized Cost':>14} {'Fill Rate':>10}"
+    )
     separator = "-" * len(header)
     print("\n" + separator)
     print(header)
     print(separator)
     for row in rows:
         print(
-            f"{row[0]:<30} {row[1]:<8} {row[2]:>8} {row[3]:>12} {row[4]:>12} {row[5]:>10}"
+            f"{row[0]:<30} {row[1]:<8} {row[2]:>8}"
+            f" {row[3]:>14} {row[4]:>14} {row[5]:>10}"
         )
     print(separator + "\n")
 
+    _print_destination_breakdown(results_root)
+
+
+def _print_destination_breakdown(results_root: Path) -> None:
+    dest_header = (
+        f"  {'Destination':<14} {'Transport':>12} {'Holding':>10}"
+        f" {'Realized':>12} {'Demand':>10} {'Shortage':>10}"
+        f" {'Fill Rate':>10} {'$/unit dem':>11} {'$/unit ful':>11}"
+    )
+    dest_separator = "  " + "-" * (len(dest_header) - 2)
+
+    for config_path in CONFIGS:
+        experiment_name = config_path.stem
+        realized_path = results_root / experiment_name / "realized_metrics.json"
+        if not realized_path.exists():
+            continue
+
+        with open(realized_path) as f:
+            r = json.load(f)
+
+        per_dest = r.get("per_destination", {})
+        if not per_dest:
+            continue
+
+        print(f"  {experiment_name}")
+        print(dest_separator)
+        print(dest_header)
+        print(dest_separator)
+        for d_id in sorted(per_dest):
+            d = per_dest[d_id]
+            print(
+                f"  {d_id:<14}"
+                f" {d['transport_cost']:>12.2f}"
+                f" {d['realized_holding_cost']:>10.2f}"
+                f" {d['realized_total_cost']:>12.2f}"
+                f" {d['total_actual_demand']:>10.2f}"
+                f" {d['total_shortage']:>10.2f}"
+                f" {d['fill_rate']:>10.4f}"
+                f" {d['cost_per_unit_demanded']:>11.2f}"
+                f" {d['cost_per_unit_fulfilled']:>11.2f}"
+            )
+        print(dest_separator + "\n")
+
 
 if __name__ == "__main__":
     main()

experiments/run_experiment.py

@@ -24,6 +24,7 @@
 from forecasting import create_forecasting_pipeline, AggregatedForecastingResult
 from optimization import MultiPeriodOptimizer, MultiPeriodResult
 from experiment_config import ExperimentConfig, load_experiment_config
+from actuals_evaluator import evaluate as evaluate_realized, save_realized_metrics
 from utils.system_paths import get_project_root
 
 logging.basicConfig(
@@ -93,6 +94,15 @@ def run_experiment(config_path: Path) -> None:
 
     logger.info("Artifacts saved to: %s", config.output_path)
 
+    # --- Realized evaluation ---
+    realized = evaluate_realized(config.output_path)
+    save_realized_metrics(realized, config.output_path)
+    logger.info(
+        "Realized evaluation: fill_rate=%.4f, realized_cost=%.2f",
+        realized.fill_rate,
+        realized.realized_total_cost,
+    )
+
 
 def _save_metrics(
     config: ExperimentConfig,