Interventions

The intervention and metric helpers.

set

set(value: float | int | str | bool) -> dict[str, Any]

Set the variable to an exact value.

A bare value anywhere do= is accepted means the same thing.

Parameter	Type	Default	Description
value	float | int | str | bool	required	The value to pin the variable to.

pct

pct(value: float) -> dict[str, Any]

Relative percentage change: rc.pct(+15) means +15%.

Parameter	Type	Default	Description
value	float	required	The change, in percent.

add

add(value: float) -> dict[str, Any]

Relative absolute change: rc.add(-5) means minus five units.

Parameter	Type	Default	Description
value	float	required	The change, in the variable's own units.

prob

prob(
    category: str | int | bool | dict[Any, float],
    probability: float | None = None,
) -> dict[str, Any]

Set a category's probability: rc.prob("yes", 0.8) or rc.prob({"yes": 0.8}).

Parameter	Type	Default	Description
category	str | int | bool | dict[Any, float]	required	The category, or a single-pair {category: probability} dict.
probability	float | None	None	The probability, when category is not a dict.

Raises

• InvalidArgumentError: The dict form carried more than one pair, no probability was given, or it is not a probability.

adjust_prob

adjust_prob(category: str | int | bool, delta: float) -> dict[str, Any]

Shift a category's probability by percentage points: rc.adjust_prob("yes", +10).

Parameter	Type	Default	Description
category	str | int | bool	required	The category to shift.
delta	float	required	The shift, in percentage points.

members

members(
    include: list[str] | None = None,
    exclude: list[str] | None = None,
    size: int | None = None,
    replace: bool = False,
) -> dict[str, Any]

Set-valued column intervention: who is in the set, who is out, how big it is.

Parameter	Type	Default	Description
include	list[str] | None	None	Members that must be in the set.
exclude	list[str] | None	None	Members that must not be.
size	int | None	None	How large the set should be.
replace	bool	False	Replace the observed membership instead of amending it.

at

at(
    spec: Any,
    timestamp: int | None = None,
    persistent: bool | None = None,
    duration_steps: int | None = None,
) -> dict[str, Any]

Schedule an intervention in time, for temporal and panel twins.

Wraps a value spec (or bare value) with when it applies and for how long: rc.at(rc.pct(-10), persistent=True) applies from the first forecast step onwards.

Parameter	Type	Default	Description
spec	Any	required	The intervention to schedule, or a bare value.
timestamp	int | None	None	When it starts (ms epoch). Defaults to the first forecast step.
persistent	bool | None	None	Keep applying it for every later step.
duration_steps	int | None	None	Apply it for this many steps only.

range

range(
    from_: float | None = None,
    to: float | None = None,
    *,
    steps: int | None = None,
) -> dict[str, Any]

Sweep a numeric variable across a grid instead of pinning it: rc.range(15, 30).

A scenario carries at most one range intervention; read the curves back with result.sweep().

Parameter	Type	Default	Description
from_	float | None	None	Low end of the sweep. Defaults to the variable's observed p05.
to	float | None	None	High end of the sweep. Defaults to the observed p95.
steps	int | None	None	How many points to evaluate across the range.

metric

metric(
    name: str,
    sql: str,
    unit: str = 'count',
    higher_is_better: bool = True,
) -> dict[str, Any]

A simulation metric: SQL over the sampled frame, registered as df/data/dataset.

Parameter	Type	Default	Description
name	str	required	Name for the metric, as it appears on the result.
sql	str	required	SQL over the sampled frame, which is registered as df, data, and dataset.
unit	str	'count'	Unit label for the metric's value.
higher_is_better	bool	True	Which direction counts as an improvement.

Examples

>>> rc.metric("avg_revenue", "SELECT AVG(revenue) AS value FROM df", unit="USD")

objective

objective(
    name: str,
    sql: str,
    direction: str = 'maximise',
    unit: str | None = None,
    weight: float | None = None,
) -> dict[str, Any]

An optimisation objective: what to move, which way, measured by SQL.

Parameter	Type	Default	Description
name	str	required	Label for the objective, as it appears on the result. It names the metric, not a variable in the causal graph.
sql	str	required	SQL over the sampled frame, which is registered as df, data, and dataset.
direction	str	'maximise'	maximise or minimise. Both spellings are accepted.
unit	str | None	None	Unit label for the objective's value.
weight	float | None	None	Relative weight against the other objectives. Defaults to 1.

Raises

• InvalidArgumentError: The name is blank, the SQL is not a SELECT, or the direction is neither maximise nor minimise.

Examples

>>> rc.objective("Total revenue", "SELECT SUM(revenue) AS value FROM df")

A categorical outcome has to be counted rather than averaged: AVG over a text column is not a number, and the run fails in the engine.

>>> rc.objective(
...     "Churn share",
...     "SELECT AVG(CASE WHEN churn = 'Yes' THEN 1.0 ELSE 0.0 END) AS value FROM df",
...     "minimise",
... )

mean_metrics

mean_metrics(outcomes: list[str]) -> list[dict[str, Any]]

Mean-of-column metrics for each outcome variable, the common case.

Parameter	Type	Default	Description
outcomes	list[str]	required	Column names to average.

Returns (list[dict[str, Any]]): One mean-of-column metric per outcome, ready for metrics=.