API Reference

Entry Point

Package-level entry point: run_pipeline().

impact_engine_orchestrator.api.run_pipeline(fname: str | Path) → PipelineResult

Load a pipeline config file and execute the full pipeline.

Parameters:: fname (str | Path) – Path to the YAML pipeline configuration file.
Returns:: Typed pipeline result with outcome_reports as the primary user-facing field, plus stage-level detail in pilot_results, evaluate_results, allocate_result, and scale_results.
Return type:: PipelineResult

Examples

>>> result = run_pipeline("pipeline.yaml")
>>> print(result.outcome_reports)
[...]

Orchestrator

Fan-out/fan-in pipeline runner.

class impact_engine_orchestrator.orchestrator.Orchestrator(measure: PipelineComponentProtocol, evaluate: PipelineComponentProtocol, allocate: PipelineComponentProtocol, config: dict)

Run the full MEASURE-EVALUATE-ALLOCATE-SCALE pipeline.

classmethod from_config(config: dict) → Orchestrator: Build an Orchestrator from a pipeline config dict with stage configs.

run() → PipelineResult: Execute all pipeline stages and return combined results.

Configuration

Pipeline and initiative configuration.

class impact_engine_orchestrator.config.InitiativeConfig(initiative_id: str, cost_to_scale: float, measure_config: str = '', evaluate_strategy: str = 'score')

Single initiative with its scaling cost.

cost_to_scale: float

evaluate_strategy: str = 'score'

initiative_id: str

measure_config: str = ''

class impact_engine_orchestrator.config.PipelineConfig(initiatives: list[InitiativeConfig], max_workers: int = 4, measure_stage: StageConfig | None = None, evaluate_stage: StageConfig | None = None, allocate_stage: StageConfig | None = None)

Problem-level parameters for a single orchestrator run.

allocate_stage: StageConfig | None = None

evaluate_stage: StageConfig | None = None

initiatives: list[InitiativeConfig]

max_workers: int = 4

measure_stage: StageConfig | None = None

class impact_engine_orchestrator.config.StageConfig(component: str, kwargs: dict = <factory>)

Parsed stage configuration (component name + constructor kwargs).

component: str

kwargs: dict

impact_engine_orchestrator.config.load_config(path: str) → dict[str, Any]

Load pipeline configuration from a YAML file.

The YAML must specify storage_url (where measure writes results) and an ALLOCATE block with the decision rule. Measure and Evaluate components are always the canonical implementations; only Allocate parameters vary.

Components

Abstract base class and protocol for pipeline components.

class impact_engine_orchestrator.components.base.PipelineComponent

Single-initiative processor conforming to the handler interface.

abstractmethod execute(event: dict) → dict: Process single initiative, return result.

class impact_engine_orchestrator.components.base.PipelineComponentProtocol(*args, **kwargs)

Structural interface for pipeline stage components.

execute(event: dict) → dict: Process event and return result.

Contracts

Shared types used across contracts.

class impact_engine_orchestrator.contracts.types.ModelType(*values)

Causal inference model type identifiers.

EXPERIMENT = 'experiment'

INTERRUPTED_TIME_SERIES = 'interrupted_time_series'

METRICS_APPROXIMATION = 'metrics_approximation'

NEAREST_NEIGHBOUR_MATCHING = 'nearest_neighbour_matching'

SUBCLASSIFICATION = 'subclassification'

SYNTHETIC_CONTROL = 'synthetic_control'

Contract for MEASURE stage output.

class impact_engine_orchestrator.contracts.measure.MeasureResult(initiative_id: str, effect_estimate: float, ci_lower: float, ci_upper: float, p_value: float, sample_size: int, model_type: ModelType, diagnostics: dict)

Causal effect estimate with confidence interval and diagnostics.

ci_lower: float

ci_upper: float

diagnostics: dict

effect_estimate: float

initiative_id: str

model_type: ModelType

p_value: float

sample_size: int

Contract for EVALUATE stage output.

class impact_engine_orchestrator.contracts.evaluate.EvaluateResult(initiative_id: str, confidence: float, confidence_range: tuple[float, float], strategy: str, report: str | ReviewResult = '')

Output of the EVALUATE pipeline stage (both strategies).

Parameters:

initiative_id (str) – Initiative identifier.
confidence (float) – Confidence score between 0.0 and 1.0.
confidence_range (tuple[float, float]) – (lower, upper) bounds from the method reviewer.
strategy (str) – Strategy that produced this result ("score" or "review").
report (str | ReviewResult) – Descriptive string for the score strategy; full ReviewResult for the review strategy.

confidence: float

confidence_range: tuple[float, float]

initiative_id: str

report: str | ReviewResult = ''

strategy: str

Contract for ALLOCATE stage output — re-exported from the allocate package.

class impact_engine_orchestrator.contracts.allocate.AllocateResult(selected_initiatives: list[str], predicted_returns: dict[str, float], budget_allocated: dict[str, float])

Portfolio selection with budget allocation.

Parameters:

selected_initiatives (list[str]) – Initiative IDs chosen for investment.
predicted_returns (dict[str, float]) – Predicted return for each selected initiative.
budget_allocated (dict[str, float]) – Budget allocated to each selected initiative.

budget_allocated: dict[str, float]

predicted_returns: dict[str, float]

selected_initiatives: list[str]

Contract for the final outcome report.

class impact_engine_orchestrator.contracts.report.OutcomeReport(initiative_id: str, predicted_return: float, actual_return: float, prediction_error: float, sample_size_pilot: int, sample_size_scale: int, budget_allocated: float, confidence_score: float, model_type: ModelType)

Comparison of pilot prediction vs scale actual for one initiative.

actual_return: float

budget_allocated: float

confidence_score: float

initiative_id: str

model_type: ModelType

predicted_return: float

prediction_error: float

sample_size_pilot: int

sample_size_scale: int