Download the notebook here! Interactive online version:

Impact Loop

This notebook walks through a complete pipeline run: Measure, Evaluate, Allocate, Scale.

We use five simulated initiatives with known treatment effects so every result is reproducible.

Workflow overview

1. Simulate initiative data

2. Configure the pipeline

3. Run the pipeline

4. Inspect results by stage

5. Examine outcome reports

Initial Setup

Each initiative gets a simulated product catalog. The simulator uses a fixed seed per initiative so results are deterministic.

import yaml
from pathlib import Path

from online_retail_simulator.simulate.products_rule_based import simulate_products_rule_based

NUM_PRODUCTS = 100

config_raw = yaml.safe_load(Path("config.yaml").read_text())

for init in config_raw["INITIATIVES"]:
    with open(init["measure_config"]) as f:
        measure_config = yaml.safe_load(f)

    source_config = measure_config["DATA"]["SOURCE"]["CONFIG"]
    seed = source_config["seed"]
    output_path = Path(source_config["path"])
    output_path.parent.mkdir(parents=True, exist_ok=True)

    products = simulate_products_rule_based(
        {"RULE": {"PRODUCTS": {"PARAMS": {"num_products": NUM_PRODUCTS, "seed": seed}}}}
    )
    products.to_csv(output_path, index=False)
    print(f"  {init['initiative_id']} — {len(products)} products → {output_path}")

Step 1 — Configure the pipeline

The pipeline is driven by a single YAML file that specifies the budget, initiative list, and stage adapters.

print(f"Budget: ${config_raw['ALLOCATE']['budget']:,.0f}")
print(f"Initiatives: {len(config_raw['INITIATIVES'])}")
for init in config_raw["INITIATIVES"]:
    print(f"  {init['initiative_id']}  (cost to scale: ${init['cost_to_scale']:,.0f})")

Step 2 — Run the pipeline

A single call executes all four stages: measure causal effects, score evidence quality, optimize the portfolio, and scale the selected initiatives.

from impact_engine_orchestrator import run_pipeline

result = run_pipeline("config.yaml")

print("Pipeline complete.")
print(f"  Initiatives measured: {len(result.pilot_results)}")
print(f"  Initiatives selected: {len(result.outcome_reports)}")

Step 3 — Inspect results by stage

MEASURE — Causal effect estimates

import pandas as pd

pilots = pd.DataFrame(result.pilot_results)
pilots[["initiative_id", "effect_estimate", "ci_lower", "ci_upper", "p_value", "model_type"]]

EVALUATE — Confidence scores

evals = pd.DataFrame(result.evaluate_results)
evals[["initiative_id", "confidence", "strategy"]]

ALLOCATE — Portfolio selection

alloc = result.allocate_result
print(f"Selected: {alloc['selected_initiatives']}")
print(f"\nBudget allocation:")
for iid, budget in alloc["budget_allocated"].items():
    predicted = alloc["predicted_returns"][iid]
    print(f"  {iid}: ${budget:,.0f} (predicted return: {predicted:.2%})")

Step 4 — Examine outcome reports

The outcome report compares pilot predictions against scale actuals. This is the learning signal that calibrates future cycles.

import dataclasses

reports = pd.DataFrame([dataclasses.asdict(r) for r in result.outcome_reports])
reports[
    [
        "initiative_id",
        "predicted_return",
        "actual_return",
        "prediction_error",
        "confidence_score",
        "budget_allocated",
        "sample_size_pilot",
        "sample_size_scale",
    ]
]