API reference

Allocation

Portfolio allocation rules.

Provides decision-rule implementations for portfolio selection, a shared preprocessing pipeline, and the AllocationRule protocol that all rules satisfy.

The allocate() facade loads config and data, then dispatches to the appropriate rule.

class impact_engine_allocate.allocation.AllocationRule

Abstract base class for decision-rule solvers.

Subclass and implement __call__() to provide a custom allocation decision rule. Register the subclass with RULE_REGISTRY to make it available via string name in pipeline configs.

class impact_engine_allocate.allocation.BayesianAllocation(weights: dict[str, float])

Bayesian expected-return decision rule.

Maximizes the weighted sum of scenario returns, where weights represent the decision-maker’s prior beliefs about scenario likelihoods.

Parameters:: weights (dict[str, float]) – Mapping from scenario name to probability weight. Must be non-negative and sum to 1. Keys must match the scenario names in the preprocessed initiatives’ effective_returns.
Raises:: ValueError – If weights are negative or do not sum to 1.

class impact_engine_allocate.allocation.MinimaxRegretAllocation

Minimax regret decision rule.

Minimizes the maximum regret across best, median, and worst-case scenarios, subject to budget and minimum worst-case return constraints.

This rule receives preprocessed initiatives (with effective_returns already computed by the shared preprocessing step).

class impact_engine_allocate.allocation.RuleRegistry

Registry mapping names to AllocationRule subclasses.

Example

>>> class MyRule(AllocationRule):
...     def __call__(self, initiatives, total_budget, min_portfolio_worst_return):
...         ...
>>> RULE_REGISTRY.register("my_rule", MyRule)

get_class(name: str) → type[AllocationRule]

Return the rule class registered under name.

Parameters:: name (str) – Registered rule name.
Returns:: The rule class (not an instance — instantiate with desired kwargs).
Return type:: type[AllocationRule]
Raises:: ValueError – If name is not registered.

list() → list[str]

Return sorted list of registered rule names.

Return type:: list[str]

register(name: str, cls: type[AllocationRule]) → None

Register an allocation rule class under name.

Parameters:

name (str) – Registry key (used in pipeline configs as the rule field).
cls (type[AllocationRule]) – Class to register. Must be a subclass of AllocationRule.

Raises:

ValueError – If cls is not a subclass of AllocationRule.

class impact_engine_allocate.allocation.RuleResult

Common output contract all decision rules must satisfy.

Parameters:

status (str) – Solver termination status (e.g. "Optimal").
selected_initiatives (list[str]) – IDs of selected initiatives.
total_cost (float) – Aggregate cost of the selected portfolio.
objective_value (float | None) – Value of the rule’s objective function, or None if non-optimal.
total_actual_returns (dict[str, float]) – Per-scenario effective returns for the selected portfolio.
rule (str) – Identifier for the decision rule (e.g. "minimax_regret").
detail (dict[str, Any]) – Rule-specific diagnostics, opaque to the adapter.

detail: dict[str, Any]

objective_value: float | None

rule: str

selected_initiatives: list[str]

status: str

total_actual_returns: dict[str, float]

total_cost: float

impact_engine_allocate.allocation.allocate_portfolio(config: str | Path | dict[str, Any], data_dir: str | Path) → RuleResult

Run portfolio allocation end-to-end.

Loads configuration, reads initiative data from job directories, preprocesses, and dispatches to the configured decision rule.

Parameters:

config (str | Path | dict) – Path to a YAML config file or a raw config dict. Must contain an allocation: section with budget and costs.
data_dir (str | Path) – Root directory containing per-initiative subdirectories with impact_results.json and evaluate_result.json.

Returns:

Portfolio selection result from the decision rule.

Return type:

RuleResult

impact_engine_allocate.allocation.calculate_effective_returns(initiatives: list[dict[str, ~typing.Any]], confidence_penalty_func: ~collections.abc.Callable[[float], float] = <function calculate_gamma>) → list[dict[str, Any]]

Calculate confidence-penalized effective returns for each initiative.

Blends each scenario’s base return toward the worst-case return, weighted by the penalty factor gamma. Does not mutate input.

Parameters:

initiatives (list[dict[str, Any]]) – Each dict must have keys: confidence, R_best, R_med, R_worst.
confidence_penalty_func (Callable[[float], float], optional) – Maps confidence to penalty factor gamma. Default: 1 - confidence.

Returns:

New list of dicts, each augmented with gamma and effective_returns keys.

Return type:

list[dict[str, Any]]

impact_engine_allocate.allocation.calculate_gamma(confidence_score: float) → float

Convert a confidence score to a penalty factor.

Parameters:: confidence_score (float) – Confidence in the initiative’s estimates, between 0 and 1.
Returns:: Penalty factor gamma = 1 - confidence.
Return type:: float
Raises:: ValueError – If confidence_score is outside [0, 1].

impact_engine_allocate.allocation.empty_rule_result(status: str, rule: str, scenarios: list[str] | None = None) → RuleResult

Build a RuleResult with no selection.

Parameters:

status (str) – Descriptive status string.
rule (str) – Decision rule identifier.
scenarios (list[str], optional) – Scenario names for the empty returns dict. Defaults to SCENARIOS.

Return type:

RuleResult

impact_engine_allocate.allocation.extract_selection(x_vars: dict[str, LpVariable], initiatives: list[dict[str, Any]], scenarios: list[str]) → tuple[list[str], float, dict[str, float]]

Extract selected initiatives and aggregate returns from a solved BIP.

Parameters:

x_vars (dict[str, LpVariable]) – Binary selection variables keyed by initiative ID.
initiatives (list[dict[str, Any]]) – Preprocessed initiatives with effective_returns.
scenarios (list[str]) – Scenario names to aggregate over.

Returns:

(selected_ids, total_cost, total_actual_returns).

Return type:

tuple[list[str], float, dict[str, float]]

impact_engine_allocate.allocation.preprocess(initiatives: list[dict[str, ~typing.Any]], min_confidence_threshold: float = 0.0, confidence_penalty_func: ~collections.abc.Callable[[float], float] = <function calculate_gamma>) → list[dict[str, Any]]

Filter initiatives by confidence and compute effective returns.

Parameters:

initiatives (list[dict[str, Any]]) – Raw initiatives with allocation field names.
min_confidence_threshold (float) – Initiatives below this confidence are excluded.
confidence_penalty_func (Callable[[float], float], optional) – Maps confidence to penalty factor gamma.

Returns:

Preprocessed initiatives with effective_returns, or empty list if no initiatives pass the confidence threshold.

Return type:

list[dict[str, Any]]

Models

Data models for the ALLOCATE pipeline stage.

class impact_engine_allocate.models.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]

Config

Parse-once configuration for the allocation subsystem.

class impact_engine_allocate.config.AllocationConfig(budget: float, costs: dict[str, float]=<factory>, rule: str = 'minimax_regret', min_confidence_threshold: float = 0.0, min_portfolio_worst_return: float = 0.0, solver_kwargs: dict = <factory>)

Validated allocation configuration.

Parameters:

budget (float) – Total budget for the portfolio (must be > 0).
costs (dict[str, float]) – Per-initiative cost_to_scale mapping.
rule (str) – Decision rule identifier ("minimax_regret" or "bayesian").
min_confidence_threshold (float) – Initiatives below this confidence are excluded.
min_portfolio_worst_return (float) – Minimum aggregate worst-case return for the portfolio.
solver_kwargs (dict) – Extra keyword arguments forwarded to the decision rule constructor (e.g. {"weights": {"best": 0.33, "med": 0.33, "worst": 0.34}}).

budget: float

costs: dict[str, float]

min_confidence_threshold: float = 0.0

min_portfolio_worst_return: float = 0.0

rule: str = 'minimax_regret'

solver_kwargs: dict

impact_engine_allocate.config.load_config(source: str | Path | dict[str, Any]) → dict[str, Any]

Load allocation configuration from a YAML file or dict.

Parameters:

source (str | Path | dict) – A path to a YAML file or a raw dict. YAML files must contain an allocation: section.

Returns:

Fully validated configuration dictionary.

Return type:

dict

Raises:

ValueError – If required fields are missing or invalid.
FileNotFoundError – If the YAML file does not exist.

Job reader

Job directory reader for the allocation subsystem.

Reads initiative data from pipeline output directories (measure_result.json and evaluate_result.json) and converts them into the flat dict format expected by the allocation rules.

impact_engine_allocate.job_reader.load_initiatives(data_dir: str | Path, costs: dict[str, float]) → list[dict[str, Any]]

Load initiative data from pipeline output directories.

Scans data_dir for subdirectories, each representing one initiative. Reads measure_result.json (measure output) and evaluate_result.json (evaluate output) from each subdirectory.

Parameters:

data_dir (str | Path) – Root directory containing per-initiative subdirectories.
costs (dict[str, float]) – Mapping from initiative ID (directory name) to cost_to_scale.

Returns:

List of initiative dicts with keys id, cost, R_best, R_med, R_worst, confidence.

Return type:

list[dict[str, Any]]

Raises:

FileNotFoundError – If required files are missing from an initiative directory.
ValueError – If an initiative directory has no matching cost entry.