Python API

Use ScenarioPlanner when you want to evaluate one scenario or compare multiple scenarios from Python.

The preferred public API lives under abacus.scenarios.

abacus.scenario_planner remains available for existing statistical compatibility imports, but new statistical scenario code should use abacus.scenarios. Application wrappers are separate from the library.

For the recommended entry points and current scope, see Supported Surface.

Prerequisite

ScenarioPlanner requires a fitted PanelMMM with idata.

If you construct the planner before fitting, AMMM3 raises ValueError.

Create a planner

from abacus.scenarios import ScenarioPlanner

planner = ScenarioPlanner(mmm)

You can inspect the modelled channel names with:

channels = planner.channels

Evaluate one scenario

Use evaluate(...) when you want one scenario result:

from abacus.scenarios import ManualAllocationScenarioSpec, ScenarioPlanner

planner = ScenarioPlanner(mmm)

result = planner.evaluate(
    ManualAllocationScenarioSpec(
        name="Manual plan",
        start_date="2025-03-03",
        end_date="2025-03-24",
        noise_level=0.0,
        include_carryover=False,
        allocation={
            "channel_1": 420_000.0,
            "channel_2": 280_000.0,
            "channel_3": 200_000.0,
        },
    )
)

print(result.totals)
print(result.channels)
print(result.metadata)

evaluate(...) returns ScenarioResult with:

  • totals
  • channels
  • contributions_over_time
  • allocation
  • metadata

Compare multiple scenarios

Use compare(...) when you want one combined comparison object:

from abacus.scenarios import (
    CurrentScenarioSpec,
    FixedBudgetOptimizedScenarioSpec,
    ManualAllocationScenarioSpec,
    ScenarioPlanner,
)

planner = ScenarioPlanner(mmm)

comparison = planner.compare(
    [
        CurrentScenarioSpec(
            name="Current baseline",
            start_date="2025-01-06",
            end_date="2025-02-24",
        ),
        ManualAllocationScenarioSpec(
            name="Manual plan",
            start_date="2025-03-03",
            end_date="2025-03-24",
            noise_level=0.0,
            include_carryover=False,
            allocation={
                "channel_1": 420_000.0,
                "channel_2": 280_000.0,
                "channel_3": 200_000.0,
            },
        ),
        FixedBudgetOptimizedScenarioSpec(
            name="Optimised plan",
            start_date="2025-03-03",
            end_date="2025-03-24",
            noise_level=0.0,
            include_carryover=False,
            total_budget=900_000.0,
        ),
    ]
)

print(comparison.totals)
print(comparison.allocations)

compare(...) returns ScenarioComparison with:

  • totals
  • channels
  • contributions_over_time
  • allocations
  • metadata

Unlike ScenarioResult, the combined object uses the plural allocations.

Run a versioned recipe

Use ScenarioRecipe for an in-memory Python request. Use load_scenario_recipe(...) for YAML and run_scenario_recipe(...) when the model has already been retained by the pipeline.

from abacus.scenarios import run_scenario_recipe

bundle = run_scenario_recipe(
    results_dir="results/geo_fe_20260824_120000",
    recipe_path="data/demo/geo_fe/scenario_recipe.yml",
)

print(bundle.output_dir)
print(bundle.comparison.totals)

The default output path is a new timestamped directory under scenario_planner/recipes/ in the fitted run. Pass output_dir= only when you need another new location. AMMM3 rejects an existing target directory so a later evaluation cannot silently replace retained evidence.

Use evaluate_scenario_recipe(...) when you already have the fitted model in memory:

from abacus.scenarios import ScenarioRecipe, evaluate_scenario_recipe

recipe = ScenarioRecipe(scenarios=(current_spec, manual_spec))
bundle = evaluate_scenario_recipe(
    model=mmm,
    recipe=recipe,
    output_dir="scenario-evidence/fe-plan-v1",
    source_run_id="geo-fe-run",
)

ScenarioArtifactBundle exposes the output directory, named artefact paths, and the in-memory ScenarioComparison.

Programmatic workspace orchestration

Use WorkspaceService from abacus.scenarios when you want to work with saved planner workspaces directly from Python.

Common operations include:

  • load_workspace(...)
  • save_workspace(...)
  • clone_workspace(...)
  • update_workspace_metadata(...)
  • create_template_draft(...)
  • replace_draft(...)
  • evaluate_draft(...)
  • run_sensitivity_sweep(...)
  • export_workspace_bundle(...)

Example:

from abacus.scenarios import WorkspaceService, load_planner_run_context

run_context = load_planner_run_context("results/timeseries_20260308_144627")
workspace_service = WorkspaceService(run_context)
workspace = workspace_service.load_or_create_default_workspace(
    workspace_name="Timeseries planning workspace",
)

draft = workspace_service.create_template_draft(
    workspace=workspace,
    scenario_type="fixed_budget_optimized",
)
workspace = workspace_service.replace_draft(workspace, draft)
workspace = workspace_service.evaluate_draft(workspace, draft)
workspace_service.save_workspace(
    workspace,
    action="evaluate_draft",
    changed_scenario_ids=[draft.scenario_id],
)

This example creates the default workspace if it is absent. Use load_workspace(...) instead when you require a specific existing workspace ID.

WorkspaceService defaults to synchronous jobs when you instantiate it directly. Custom integrations can supply a job runner.

Prepare data for a client UI

Use to_store_payload() when you want a JSON-friendly version of the comparison tables:

payload = comparison.to_store_payload()

This method converts datetime columns to YYYY-MM-DD strings and returns a dict with a scalar contract_version plus record lists for the comparison tables. Compare contract_version with SCENARIO_CONTRACT_VERSION before a dashboard or external client assumes a payload shape.

Background-job helpers

For custom integrations, WorkspaceService also exposes queue/apply methods:

  • submit_draft_evaluation(...)
  • apply_draft_evaluation_job(...)
  • submit_sensitivity_sweep(...)
  • apply_sensitivity_sweep_job(...)

Use these when an external application needs queued evaluation. For scripted flows, the blocking methods are simpler.

See Workspace and protocol reference for return types, persistence, revision checks and custom job-runner contracts.

Relationship to the low-level wrapper

ScenarioPlanner uses the response wrapper for manual evaluation and PanelBudgetOptimizerWrapper for fixed-budget optimisation. Its budget contract differs from the low-level wrappers:

  • you pass total horizon budgets and allocations
  • the planner converts them to per-period units internally
  • the planner returns comparison tables rather than raw optimiser objects

If you want direct access to optimize_budget(...) or sample_response_distribution(...), use Budget Optimisation instead.

Common pitfalls

  • Passing per-period spend into ManualAllocationScenarioSpec or FixedBudgetOptimizedScenarioSpec
  • Expecting duplicate scenario_id values to be allowed in compare(...)
  • Forgetting that result.allocation and comparison.allocations use different attribute names