API Reference

Orchestrator

Fan-out/fan-in pipeline runner.

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

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

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

run() → dict: 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 = '')

Single initiative with its scaling cost.

cost_to_scale: float

initiative_id: str

measure_config: str = ''

class impact_engine_orchestrator.config.PipelineConfig(budget: float, scale_sample_size: int, 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

budget: float

evaluate_stage: StageConfig | None = None

initiatives: list[InitiativeConfig]

max_workers: int = 4

measure_stage: StageConfig | None = None

scale_sample_size: int

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) → PipelineConfig: Load a PipelineConfig from a YAML file.

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 methodology used for measurement.

EXPERIMENT = 'experiment'

INTERRUPTED_TIME_SERIES = 'interrupted_time_series'

METRICS_APPROXIMATION = 'metrics_approximation'

NEAREST_NEIGHBOUR_MATCHING = 'nearest_neighbour_matching'

OBSERVATIONAL = 'observational'

QUASI_EXPERIMENT = 'quasi-experiment'

SUBCLASSIFICATION = 'subclassification'

SYNTHETIC_CONTROL = 'synthetic_control'

TIME_SERIES = 'time-series'

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, cost: float, return_best: float, return_median: float, return_worst: float, model_type: ModelType, sample_size: int)

Confidence-scored initiative with scenario returns.

confidence: float

cost: float

initiative_id: str

model_type: ModelType

return_best: float

return_median: float

return_worst: float

sample_size: int

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.

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