Every public symbol

AcceptanceRegion(n: int, p: float, alpha: float, lower: int, upper: int)

An exact binomial acceptance region for a rate criterion.

region: AcceptanceRegion = clopper_pearson(n=500, p=0.05, alpha=0.001)

Add(node: Literal['add'] = 'add', terms: tuple[Const | Data | Param | Add | Mul | Div | Pow | Apply | Convolve | Reduce | Gather | Link | Opaque, ...])

Sum of terms sharing one dimension.

response = Equation(lhs=y, rhs=Add(terms=(Param(name="a", dimension=D.outcome), hill)), name="response")

Apply(node: Literal['apply'] = 'apply', fn: Literal['exp', 'log', 'log1p', 'expm1', 'tanh', 'sigmoid', 'logit', 'softplus', 'neg', 'relu', 'step', 'sin', 'cos'], arg: Const | Data | Param | Add | Mul | Div | Pow | Apply | Convolve | Reduce | Gather | Link | Opaque)

A scalar function of one argument. Argument and result are dimensionless.

print(fn, value(Apply(fn=fn, arg=Const(value=0.0, dimension=dimensionless()))))

ApplyFn(*args, **kwargs)

fns: list[ApplyFn] = ["exp", "sigmoid", "softplus"]

Assumption(name: str, facet: str, statement: str, challenged_by: str = '', state: Literal['unverified', 'satisfied', 'violated', 'asserted'] = 'unverified', detail: dict[str, str] = {})

A named, falsifiable condition that licenses a step.

overlap = Assumption(
    name="overlap",
    facet="population",
    statement="every target stratum has positive support in the source",
    challenged_by="propensity / dose overlap diagnostic",
)

The declarable set of base dimensions.

print(list(BASES))

BaseRegistry(defaults: tuple[tuple[str, str], ...] = ...) -> None

The declarable set of base dimensions.

private_bases = BaseRegistry()

Blocked(reason: str, detail: dict[str, str] = {}, status: Literal['blocked'] = 'blocked')

The request is not licensed: no route, no transfer, no identification.

Blocked(reason="   ")

Capability(*values)

What a producer can do. A request needing an absent one degrades, typed.

return frozenset({Capability.COUNTERFACTUAL})

Const(node: Literal['const'] = 'const', value: float | tuple[float, ...], dimension: Dimension)

A literal with a declared dimension: a scalar, or a short vector (a lag index, knots).

one = Const(value=1.0, dimension=dimensionless())

Constraint(name: str, expr: Const | Data | Param | Add | Mul | Div | Pow | Apply | Convolve | Reduce | Gather | Link | Opaque, family: Literal['normal', 'lognormal'], observed: float, scale: float, detail: dict[str, str] = {})

A soft constraint on a scalar function of the parameters: ``observed ~ p(expr(θ), scale)``.

lift = Constraint(
    name="mean_lift@experiment_1",
    expr=Reduce(op="mean", arg=hill),
    family="normal",
    observed=6.0,
    scale=0.5,
    detail={"measurement": "experiment_1", "doses": "observed"},
)

Convolve(node: Literal['convolve'] = 'convolve', signal: Const | Data | Param | Add | Mul | Div | Pow | Apply | Convolve | Reduce | Gather | Link | Opaque, kernel: Const | Data | Param | Add | Mul | Div | Pow | Apply | Convolve | Reduce | Gather | Link | Opaque)

Causal convolution of ``signal`` with ``kernel`` along the time axis.

carry = Convolve(signal=dose, kernel=weights)

Covariate(name: str, dimension: Dimension | None = None, unit: str | None = None, description: str = '')

Measured, not intervened on.

rainfall = Covariate(name="rainfall", dimension=dimensionless(), description="standardized")

The declarable set of base dimensions.

print(D.currency, D.time, D.outcome, D.entity)

int([x]) -> integer int(x, base=10) -> integer

print("digits kept of the uncertainty:", DIGITS)

Data(node: Literal['data'] = 'data', name: str, dimension: Dimension)

A column of the panel. Its dimension is the one its role declares.

dose = Data(name="dose", dimension=D.currency)

DesignMatrix(X: NDArray[float64], columns: tuple[str, ...], offset: NDArray[float64], at: Mapping[str, NDArray[float64]]) -> None

``mu = offset + X @ theta[columns]`` at a fixed point of the nonlinear parameters.

dm = DesignMatrix(X=X, columns=("a", "b"), offset=np.zeros(4), at={"k": np.array(50.0)})

Dimension(exponents: dict[str, Fraction])

A product of declared base dimensions raised to rational exponents.

Dimension(exponents={"temperature": 1})

Two quantities of different dimension were combined.

try:
    D.currency.require_equal(D.time, context="Add(term_1, term_2)")
except DimensionError as e:
    print(e)

Div(node: Literal['div'] = 'div', numerator: Const | Data | Param | Add | Mul | Div | Pow | Apply | Convolve | Reduce | Gather | Link | Opaque, denominator: Const | Data | Param | Add | Mul | Div | Pow | Apply | Convolve | Reduce | Gather | Link | Opaque)

Quotient; exponents subtract.

x = Div(numerator=dose, denominator=k)

Dose(name: str, dimension: Dimension | None = None, unit: str | None = None, description: str = '', numeraire: str | None = None)

The magnitude of an intervention on a treatment, with units and a numeraire.

dose = Dose(name="fertilizer_dose", dimension=D.currency, unit="USD", numeraire="USD")

Entity(*args, **kwargs)

What every entity spec provides: a name, an optional dimension and unit.

print(isinstance(dose, Entity), Treatment.__mro__[1].__name__)

EntityName(*args, **kwargs)

Runtime representation of an annotated type.

name: EntityName = "tv-ads.v2"

Equation(node: Literal['equation'] = 'equation', lhs: Const | Data | Param | Add | Mul | Div | Pow | Apply | Convolve | Reduce | Gather | Link | Opaque, rhs: Const | Data | Param | Add | Mul | Div | Pow | Apply | Convolve | Reduce | Gather | Link | Opaque, name: str = '')

``lhs = rhs``; both sides must share a dimension.

response = Equation(lhs=y, rhs=Add(terms=(Param(name="a", dimension=D.outcome), hill)), name="response")

Expr(*args, **kwargs)

Runtime representation of an annotated type.

hill: Expr = Mul(factors=(beta, Div(numerator=Pow(base=x, exponent=s),
                                    denominator=Add(terms=(one, Pow(base=x, exponent=s))))))

Represent a PEP 604 union type

def realize(value_ok: bool) -> float | Failure:
    if not value_ok:
        return Blocked(reason="adjustment set contains an unmeasured variable", detail={"node": "U"})
    return 0.42

Gather(node: Literal['gather'] = 'gather', source: Const | Data | Param | Add | Mul | Div | Pow | Apply | Convolve | Reduce | Gather | Link | Opaque, index: Data)

``source[..., index]``: pick entries of a vector-valued expression by an integer column.

mean = Add(terms=(Gather(source=alpha, index=unit), hill))

Interval(lower: float, upper: float, definition: Literal['eti', 'hdi', 'wald'], mass: float)

``[lower, upper]`` with the definition and the mass it was computed at.

Interval(lower=0.0, upper=1.0)

Intervention(doses: dict[str, float], mode: Literal['set', 'scale', 'shift'] = 'set', version: str = 'unspecified', window: TimeWindow | None = None)

What is set, to what value, how, over what support.

iv = Intervention(doses={"fertilizer": 100.0, "irrigation": 20.0}, version="granular-v2")

LedgerLine(kind: str, statement: str, assumption: Assumption | None = None, detail: dict[str, str] = {}, source: str = '', target: str = '')

One entry in the assumption ledger.

line = LedgerLine(
    kind="facet:population",
    statement="transferred from region=north to region=all under overlap",
    assumption=overlap.asserted(),
    detail={"from": "north", "to": "all"},
)

Likelihood(family: Literal['normal', 'lognormal', 'student_t', 'poisson'], scale: str | None = None, scale_expr: Optional[Const | Data | Param | Add | Mul | Div | Pow | Apply | Convolve | Reduce | Gather | Link | Opaque] = None, df: float | None = None)

How the outcome column is distributed around the mean expression.

Likelihood(family="normal", scale="sigma", scale_expr=se_col)

LikelihoodFamily(*args, **kwargs)

family: LikelihoodFamily = "normal"

Link(node: Literal['link'] = 'link', fn: Literal['identity', 'log', 'logit'], arg: Const | Data | Param | Add | Mul | Div | Pow | Apply | Convolve | Reduce | Gather | Link | Opaque)

A GLM link applied to a dimensionless argument (after a reference divide).

print(dimension(Link(fn=link, arg=Div(numerator=y, denominator=y_ref))))

LinkFn(*args, **kwargs)

link: LinkFn = "log"

Model(*args, **kwargs)

bad_trees: list[Model] = [
    Apply(fn="log", arg=dose),
    Add(terms=(beta, Mul(factors=(beta, Apply(fn="log", arg=dose))))),
    Pow(base=dose, exponent=s),
    Convolve(signal=dose, kernel=k),
]

ModelSpec(name: str, mean: Const | Data | Param | Add | Mul | Div | Pow | Apply | Convolve | Reduce | Gather | Link | Opaque, outcome: Data, likelihood: Likelihood, parameters: tuple[Param, ...], constraints: tuple[Constraint, ...] = ())

Mean expression + outcome column + likelihood + every parameter's prior.

model = ModelSpec(
    name="panel_hill",
    mean=mean,
    outcome=y,
    likelihood=Likelihood(family=family, scale="sigma"),
    parameters=(k, s, beta, lam, a_mean, a_sd, alpha, sigma),
)

Mul(node: Literal['mul'] = 'mul', factors: tuple[Const | Data | Param | Add | Mul | Div | Pow | Apply | Convolve | Reduce | Gather | Link | Opaque, ...])

Product; exponents add.

print(dimension(Pow(base=Mul(factors=(dose, dose)), exponent=Fraction(1, 2))))

NonEmptyStr(*args, **kwargs)

Runtime representation of an annotated type.

reason: NonEmptyStr = "a reason that is not blank"

ODESystem(node: Literal['ode'] = 'ode', states: tuple[Data, ...], rhs: tuple[Const | Data | Param | Add | Mul | Div | Pow | Apply | Convolve | Reduce | Gather | Link | Opaque, ...], time: Data)

``d(state_i)/dt = rhs_i`` for each state. Dimension-check-only in 1.0 (C1).

ode = ODESystem(states=(S,), rhs=(Mul(factors=(r, S)),), time=t)

Opaque(node: Literal['opaque'] = 'opaque', name: str, inputs: tuple[Const | Data | Param | Add | Mul | Div | Pow | Apply | Convolve | Reduce | Gather | Link | Opaque, ...] = (), dimension: Dimension)

A user-supplied function the tree cannot express.

weights = Opaque(name="geometric", inputs=(lam,), dimension=dimensionless())

OpaqueFn(*args, **kwargs)

geometric: OpaqueFn = lambda lam, L=6: lam ** np.arange(L)

OpaqueRegistry(*args, **kwargs)

A Mapping is a generic container for associating key/value pairs.

registry: OpaqueRegistry = {"geometric": geometric}

Outcome(name: str, dimension: Dimension | None = None, unit: str | None = None, description: str = '', aggregation: Literal['sum', 'mean', 'rate'] = 'sum')

What you are trying to move.

yield_total = Outcome(name="yield_total", dimension=D.outcome, unit="kg", aggregation="sum")

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

print(PRIOR_HYPER)

Param(node: Literal['param'] = 'param', name: str, dimension: Dimension, prior: Prior | None = None, shape: tuple[int, ...] = ())

A parameter to be inferred, with a declared dimension and an optional prior.

k = Param(name="k", dimension=D.currency, prior=Prior(family="lognormal", hyper={"mu": 3.0, "sigma": 1.0}))

Population(name: str, description: str = '', strata: dict[str, dict[str, float]] = {})

A target population: a name and, where known, the strata weights that define it.

north = Population(name="north", strata={"soil": {"clay": 0.3, "loam": 0.7}})

Posterior(draws: Mapping[str, ArrayLike], *, coords: Mapping[str, Sequence[Any]] | None = None, provenance: Mapping[str, Any] | None = None) -> None

Draws keyed by name, each ``(chain, draw, *shape)``, plus coords and provenance.

post = Posterior(
    {
        "beta": rng.normal(0.3, 0.1, size=(4, 500, 2)),
        "sigma": rng.lognormal(-1, 0.2, size=(4, 500)),
    },
    coords={"treatment": ["fertilizer", "irrigation"]},
    provenance={"seed": 42, "backend": "hand-built"},
)

Pow(node: Literal['pow'] = 'pow', base: Const | Data | Param | Add | Mul | Div | Pow | Apply | Convolve | Reduce | Gather | Link | Opaque, exponent: Union[Const | Data | Param | Add | Mul | Div | Pow | Apply | Convolve | Reduce | Gather | Link | Opaque, Fraction])

``base ** exponent``.

print(dimension(Pow(base=Mul(factors=(dose, dose)), exponent=Fraction(1, 2))))

PredictiveDraws(values: NDArray[float64], intervention: Intervention, window: TimeWindow | None = None, coords: Mapping[str, Sequence[Any]] = <factory>, seed: int | None = None) -> None

Outcome draws under an intervention: ``values`` is ``(chain, draw, *shape)``.

return PredictiveDraws(values=beta @ dose, intervention=iv, window=window, seed=seed)

Prior(family: Literal['normal', 'halfnormal', 'lognormal', 'beta', 'gamma', 'uniform', 'fixed'], hyper: dict[str, float | str])

A named prior family with hyperparameters.

k = Param(name="k", dimension=D.currency, prior=Prior(family="lognormal", hyper={"mu": 3.0, "sigma": 1.0}))

PriorFamily(*args, **kwargs)

fam: PriorFamily = "halfnormal"

Reduce(node: Literal['reduce'] = 'reduce', op: Literal['sum', 'mean', 'max'], arg: Const | Data | Param | Add | Mul | Div | Pow | Apply | Convolve | Reduce | Gather | Link | Opaque, keepdims: bool = False)

Reduce ``arg`` along its last axis: ``sum``, ``mean``, ``max``. Dimension is preserved.

weights = Div(numerator=raw, denominator=Reduce(op="sum", arg=raw))

SchemaVersionError(spec: str, found: str, expected: str) -> None

A serialized spec's version has no migration to the current class version.

try:
    ProbeDesignV2.from_json(json.dumps(v1_payload))
except SchemaVersionError as e:
    print("no migration:", e)

Spec()

Frozen, versioned, hashable configuration object. Subclass freely.

Spec.from_json('{"not": "an envelope"}')

SpecDiff(spec: str, changed: dict[str, tuple[Any, Any]])

Field-level differences between two specs of the same type.

d: SpecDiff = design.diff(
    design.model_copy(update={"levels": (0.0, 50.0, 150.0), "window": TimeWindow(start=0, stop=12)})
)

Base class for spec construction and (de)serialization failures.

try:
    Spec.from_json('{"not": "an envelope"}')
except SpecError as e:
    print("bad envelope refused:", e)

Summary(mean: float, median: float, sd: float, interval: Interval, n: int)

Point summaries of a set of draws plus one provenance-carrying interval.

s: Summary = summarize(draws, definition="hdi", mass=0.95)

SupportsEstimands(*args, **kwargs)

A fitted producer: posterior draws + counterfactual predictions + declared estimands.

print([m for m in dir(SupportsEstimands) if not m.startswith("_")])

SupportsForward(*args, **kwargs)

A deterministic response surface: ``forward`` == ``interpret.value(expr)``.

print(isinstance(surface, SupportsForward), surface.forward({"dose": d}, {"k": 50.0, "s": 2.0, "beta": 10.0}))

SupportsIntervention(*args, **kwargs)

Anything you can ask a counterfactual of.

print(isinstance(surface, SupportsIntervention))

SupportsPosterior(*args, **kwargs)

Anything with draws you can ask questions of.

print(isinstance(post, SupportsPosterior), post.names(), post.n_chains, post.n_draws())

System(node: Literal['system'] = 'system', equations: tuple[Equation, ...])

A set of equations, each checked independently.

model = System(equations=(response, cost))

TimeWindow(start: int, stop: int, basis: Literal['per_period', 'cumulative'] = 'cumulative')

A half-open index window ``[start, stop)`` on the panel's time axis, with a basis.

d: SpecDiff = design.diff(
    design.model_copy(update={"levels": (0.0, 50.0, 150.0), "window": TimeWindow(start=0, stop=12)})
)

Treatment(name: str, dimension: Dimension | None = None, unit: str | None = None, description: str = '')

The thing you can intervene on. Its dimension is the dose's dimension.

design = ProbeDesign(
    treatment=Treatment(name="fertilizer", dimension=D.currency, unit="USD"),
    levels=(0.0, 50.0, 100.0),
    window=TimeWindow(start=0, stop=8),
)

Units of measure, each attached to a base dimension, with conversions.

UNITS.declare("USD", "currency")

A dimension named a base that the registry has not declared.

try:
    Dimension(exponents={"temperature": 1})
except UndeclaredBaseError as e:
    print(type(e).__name__, "->", e)

An entity was declared without a dimension and is treated as dimensionless (D6).

print(loose.dim, "|", [w.category.__name__ for w in caught], "|", issubclass(caught[0].category, UndimensionedWarning))

Unit(name: str, dimension: Dimension | None = None, unit: str | None = None, description: str = '', kind: Literal['individual', 'cluster', 'aggregate'] = 'individual')

The randomization / observation unit: a plot, a patient, a region.

plot = Unit(name="plot", dimension=D.entity, kind="cluster")

No conversion is registered between two units of the same dimension.

try:
    UNITS.convert(1.0, src, dst)
except (DimensionError, UnitConversionError) as e:
    print(f"{src} -> {dst}: {type(e).__name__}: {e}")

UnitSystem() -> None

Units of measure, each attached to a base dimension, with conversions.

private_units = UnitSystem()

The envelope names a class that cannot be imported or is not a Spec.

try:
    TimeWindow.from_json(design.to_json())
except UnknownSpecError as e:
    print("wrong class refused:", e)

Unsupported(reason: str, detail: dict[str, str] = {}, missing: tuple[str, ...] = (), status: Literal['unsupported'] = 'unsupported')

The producer lacks a capability the request needs.

result = Unsupported(reason=f"surface lacks {', '.join(missing)}", missing=missing)

Unverified(reason: str, detail: dict[str, str] = {}, status: Literal['unverified'] = 'unverified')

An assumption the machinery cannot check; the caller decides.

print(bool(Unsupported(reason="no marginal capability", missing=("marginal",))), bool(Unverified(reason="overlap not checked")))

Verdict(status: Literal['identified', 'downgraded', 'blocked', 'unsupported', 'unverified'], reason: str = '', assumptions: tuple[Assumption, ...] = (), route: str = '')

An honest answer about whether a quantity is licensed, and under what.

v = Verdict(status="downgraded", route="backdoor", assumptions=(overlap,))

causal_convolve(signal: Array, weights: Array) -> Array

``y[..., t] = Σ_l w[..., l] · x[..., t-l]`` with zero history before ``t=0``.

print(causal_convolve(np.ones(4), np.array([0.5, 0.25])))

check(model: Model, expected: Dimension | None = None) -> Dimension

``dimension`` plus an optional assertion against a declared dimension.

print(check(hill, D.outcome))

children(node: Model) -> tuple[Model, ...]

Direct sub-nodes, in a stable order.

print(children(hill)[0], "|", node_path(hill, s))

clopper_pearson(n: int, p: float, alpha: float = 0.001) -> AcceptanceRegion

Exact two-sided binomial acceptance region for ``k ~ Binomial(n, p)``.

region: AcceptanceRegion = clopper_pearson(n=500, p=0.05, alpha=0.001)

compile_jax(model: Model, *, opaque: JaxOpaqueRegistry | None = None) -> JaxFn

Build a traceable ``f(data, params)`` for ``model``.

f_mean = compile_jax(mean)

compile_log_density(model: ModelSpec, *, opaque: JaxOpaqueRegistry | None = None) -> Callable[[Mapping[str, Any], Mapping[str, Any]], Any]

``f(data, z) -> log posterior`` in unconstrained coordinates, traceable by jax.

f_ld = compile_log_density(model)

constrain(model: ModelSpec, z: Mapping[str, ArrayLike]) -> tuple[dict[str, Array], float]

Inverse of ``unconstrain``; also returns the summed log-Jacobian ``log|dθ/dz|``.

back, log_jac = constrain(model, z)

data_names(node: Model) -> tuple[str, ...]

Every distinct ``Data`` name in the tree, by first appearance.

print([p.name for p in params(model)], data_names(model))

decimals_for(uncertainty: float, *, digits: int = 2) -> int

Decimal places warranted by ``uncertainty``, for ``round`` and ``format``.

print(f"12.3456789 ± {u:<6} -> {format_measured(12.3456789, u):>8}   "
      f"decimals={decimals_for(u)}")

dimension(model: Model, path: str = '') -> Dimension

The dimension of ``model``; raises ``DimensionError`` naming the failing node.

print(dimension(hill))

dimension_of(entity: Entity) -> Dimension

The entity's dimension; dimensionless when none was declared (D6).

print(UNITS.factor("USD", "GBP"), "|", UNITS.dimension_of("week"))

dimensionless() -> Dimension

print("x / x is dimensionless:", (rate / rate) == dimensionless(), (rate / rate).is_dimensionless)

effective_sample_size(draws: ArrayLike) -> float

Bulk ESS via initial positive sequence of autocorrelations (Geyer 1992).

print("ESS iid   :", round(effective_sample_size(chains)))

eti(draws: ArrayLike, mass: float) -> Interval

Equal-tailed interval: the ``(1-mass)/2`` and ``1-(1-mass)/2`` quantiles.

e = eti(draws, 0.9)

format_interval(lower: float, upper: float, *, uncertainty: float | None = None, digits: int = 2, fallback: int = 6, group: bool = False) -> str

``[lower, upper]`` with both bounds at the precision the width supports.

print(format_interval(h.lower, h.upper), "from a half-width of", round(h.half_width, 4))

format_measured(value: float, uncertainty: float | None = None, *, digits: int = 2, fallback: int = 6, group: bool = False) -> str

``value`` as text, carrying only the digits ``uncertainty`` supports.

print("no stated uncertainty ->", format_measured(12.3456789))

free_parameters(model: ModelSpec) -> tuple[Param, ...]

Parameters that are inferred (everything not ``fixed``), in declaration order.

print([p.name for p in free_parameters(model)])

hdi(draws: ArrayLike, mass: float) -> Interval

Highest-density interval: the narrowest window containing ``mass`` of the draws.

h = hdi(draws, 0.9)

interval(draws: ArrayLike, *, definition: IntervalDefinition, mass: float) -> Interval

i = interval(draws, definition="eti", mass=0.5)

is_failure(x: object) -> TypeGuard[Failure]

if is_failure(r):
    print(r.status, "->", r.reason, r.detail)
else:
    print("value", r)

jax_available() -> bool

print("jax available:", jax_available(), "|", require_jax())

latex(model: Model) -> str

print(latex(hill))

latex_or_unsupported(model: Model) -> str | Unsupported

``latex``, unless the tree contains an ``Opaque`` node — then a typed refusal.

print(latex_or_unsupported(carry))

likelihood_scale(model: ModelSpec, data: Mapping[str, ArrayLike], theta: Mapping[str, ArrayLike]) -> Array

The likelihood's scale at constrained values: the named parameter, or ``scale_expr`` evaluated on ``data`` and ``theta``. ``ValueError`` for a family without one.

print("scale per row:", likelihood_scale(known_scale, {"se": np.array([0.1, 0.2, 0.4])}, {"mu": 1.0}))

load_spec(s: str) -> Spec

Deserialize any spec from its envelope, resolving the class by name.

generic = load_spec(design.to_json())

log_density(model: ModelSpec, data: Mapping[str, ArrayLike], z: Mapping[str, ArrayLike]) -> float

Unnormalized log posterior in unconstrained coordinates: prior + Jacobian + likelihood.

print("log prior:", round(log_prior(model, theta), 3), "| log density:", round(log_density(model, data, z), 3))

log_likelihood(model: ModelSpec, data: Mapping[str, ArrayLike], theta: Mapping[str, ArrayLike]) -> float

Outcome log-likelihood at constrained values, plus every soft constraint's term.

print('decomposition available:', callable(log_likelihood), callable(log_prior))

log_prior(model: ModelSpec, theta: Mapping[str, ArrayLike]) -> float

Sum of log prior densities of the free parameters at constrained values.

print("log prior:", round(log_prior(model, theta), 3), "| log density:", round(log_density(model, data, z), 3))

mc_standard_error(draws: ArrayLike) -> float

Monte-Carlo SE of the mean, ``sd / sqrt(ESS)``.

print("MCSE      :", mc_standard_error(ar))

missing_capabilities(producer: SupportsIntervention, required: frozenset[Capability] | set[Capability]) -> tuple[str, ...]

Names of required capabilities the producer lacks, sorted; empty means go.

missing = missing_capabilities(surface, needed)

node_path(root: Model, target: Model) -> str

print(children(hill)[0], "|", node_path(hill, s))

params(node: Model) -> tuple[Param, ...]

Every distinct ``Param`` in the tree, by first appearance.

print([p.name for p in params(model)], data_names(model))

require_jax() -> Unsupported | None

print("jax available:", jax_available(), "|", require_jax())

round_to(value: float, uncertainty: float, *, digits: int = 2) -> float

``value`` rounded to the place ``uncertainty`` reaches.

print(round_to(s.mean, s.sd), "vs the stored", s.mean)

spec_type_name(cls: type[Spec]) -> str

``"<module>:<qualname>"`` — the name written into the envelope.

print(spec_type_name(ProbeDesign))

summarize(draws: ArrayLike, *, definition: IntervalDefinition = 'hdi', mass: float = 0.9) -> Summary

s: Summary = summarize(draws, definition="hdi", mass=0.95)

unconstrain(model: ModelSpec, theta: Mapping[str, ArrayLike]) -> dict[str, Array]

Map constrained parameter values to R^k (log / logit per support).

z = unconstrain(model, theta)

value(model: Model, *, data: Mapping[str, Any] | None = None, params: Mapping[str, Any] | None = None, opaque: OpaqueRegistry | None = None) -> Array

Evaluate ``model`` numerically. Missing inputs raise ``KeyError`` naming the name.

value(hill, data={"dose": d}, params={"k": 50.0, "s": 2.0, "beta": 10.0})

wald(estimate: float, se: float, mass: float) -> Interval

Frequentist ``estimate ± z_{(1+mass)/2} · se``, labelled as such.

print(wald(estimate=1.2, se=0.3, mass=0.95))

walk(node: Model, path: str = '') -> Iterator[tuple[str, Model]]

Pre-order traversal yielding ``(path, node)``; paths look like ``add[1].mul[0]``.

print([path for path, _ in walk(hill)][:5])

z_score(value: float, reference: float, reference_se: float) -> float

``(value - reference) / se``: how the statistical golden fixtures compare.

print("z vs ref  :", z_score(value=ar.mean(), reference=0.0, reference_se=mc_standard_error(ar)))

ColumnScaling(method: Literal['none', 'max', 'mean', 'standardize'], loc: float = 0.0, scale: float = 1.0)

``scaled = (x - loc) / scale``. ``loc`` is 0 unless standardizing.

manual = ScalingParameters(columns={"x": ColumnScaling(method=ref, scale=1000.0)})

Completeness(n_units: int, n_periods: int, n_rows: int, balanced: bool, missing_cells: int, null_cells: int, gaps: dict[str, int] = {}, duplicate_rows: int = 0)

What the panel covers, so an estimator can decide whether it accepts it (review D5).

c: Completeness = panel.completeness()

Panel(frame: DataFrame, roles: RoleMap) -> None

A ``DataFrame`` plus a ``RoleMap``. Sorted by (unit, time) on construction.

panel = Panel(df, roles)

The frame does not satisfy the role map.

try:
    holey.require_balanced(context="a synthetic-control estimator")
except PanelError as e:
    print("refused:", e)

RoleKind(*args, **kwargs)

kind: RoleKind = roles.kind_of("fert_usd")

RoleMap(unit: str, time: str, outcome: tuple[str, Outcome], treatments: dict[str, Treatment] = {}, covariates: dict[str, Covariate] = {})

Column → role for a panel.

roles = RoleMap(
    unit="plot",
    time="week",
    outcome=("yield_kg", Outcome(name="yield_total", dimension=D.outcome, unit="kg")),
    treatments={
        "fert_usd": Treatment(name="fertilizer", dimension=D.currency, unit="USD"),
        "irrig_usd": Treatment(name="irrigation", dimension=D.currency, unit="USD"),
    },
    covariates={"rain_z": Covariate(name="rainfall", dimension=dimensionless())},
)

ScalingMethod(*args, **kwargs)

ref: ScalingMethod = "max"

ScalingParameters(columns: dict[str, ColumnScaling])

Per-column scaling for a panel's measured columns.

manual = ScalingParameters(columns={"x": ColumnScaling(method=ref, scale=1000.0)})

fit_scaling(panel: Panel, *, treatments: ScalingMethod = 'max', outcome: ScalingMethod = 'max', covariates: ScalingMethod = 'standardize') -> ScalingParameters

Fit scaling per role. Indices (unit, time) are never scaled.

params: ScalingParameters = fit_scaling(panel)

Analysis(specs: Mapping[str, Spec] = <factory>, panel: Panel | None = None, posterior: Posterior | None = None, evidence: tuple[Spec, ...] = (), ledger: tuple[LedgerLine, ...] = (), provenance: Provenance | None = None) -> None

Everything an analysis produced, keyed and hashed.

save_analysis(Analysis(specs={"m": Treatment(name="seed_mass", dimension=D.mass, unit="kg")}), root2)

ArtifactRegistry(root: str | Path) -> None

``root/<hash>.json`` per spec, plus ``root/index.tsv`` of ``hash\ttype``.

reg = ArtifactRegistry(root.parent / "registry")

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

print(FORMAT_VERSION, "|", saved.provenance.seed, saved.provenance.created)

The directory is not a readable ``analysis.axiom`` of a known format version.

try:
    load_analysis(root)
except FormatError as e:
    print("FormatError:", e)

Provenance(axiom_version: str, created: str, hashes: dict[str, str] = {}, seed: int | None = None, environment: dict[str, str] = {}, notes: tuple[str, ...] = ())

Who made this, from what, with which seed.

prov = Provenance(axiom_version="0.0.0", created="2026-08-21T00:00:00+00:00", hashes=analysis.hashes(), seed=3)

environment_fingerprint() -> dict[str, str]

print(environment_fingerprint())

load_analysis(path: str | Path) -> Analysis

loaded = load_analysis(root)

save_analysis(analysis: Analysis, path: str | Path, *, seed: int | None = None, overwrite: bool = False) -> Analysis

Write an analysis and return it with its provenance stamped.

saved = save_analysis(analysis, root, seed=3)

Built-in immutable sequence.

print(BACKEND_NAMES)

Backend(*args, **kwargs)

Something that turns a ``ModelSpec`` plus data into a ``Posterior``.

print(isinstance(lap, Backend), getattr(lap, "name", lap))

ConvergenceReport(rows: tuple[ParameterDiagnostics, ...], divergences: int, thresholds: ConvergenceThresholds, converged: bool, n_chains: int, n_draws: int, nonfinite_draw_frac: float = 0.0)

Per-parameter diagnostics, the divergence count, and a verdict with its thresholds.

report: ConvergenceReport = diagnose(post)

ConvergenceThresholds(rhat_max: float = 1.01, ess_min: float = 400.0, divergences_max: int = 0)

What ``converged`` means. Defaults follow Vehtari et al. (2021).

print(row.passes(ConvergenceThresholds(rhat_max=1.05, ess_min=100.0)))

LaplaceBackend()

The always-available backend: no sampler, a Gaussian at the mode.

backend = LaplaceBackend()

Built-in immutable sequence.

print("PyMC can dispatch NUTS to:", NUTS_SAMPLERS, "\n")

NumpyroBackend(*, target_accept: float = 0.9, max_tree_depth: int = 10) -> None

``Backend`` over NumPyro's NUTS. ``optimize`` and ``laplace`` reuse ``infer.laplace``.

nuts = NumpyroBackend()

NutsSampler(*args, **kwargs)

choice: NutsSampler = "pymc"

ParameterDiagnostics(name: str, rhat: float | None, ess_bulk: float | None, ess_tail: float | None, mcse_mean: float | None, mean: float | None, sd: float | None, n_nonfinite: int = 0)

One row of the report: a scalar variable or one element of a vector one.

row: ParameterDiagnostics = report.row("tau")

PointEstimate(theta: dict[str, float | tuple[float, ...]], log_density: float, converged: bool, method: str, n_iter: int, hessian_pd: bool | None = None, min_eigenvalue: float | None = None, newton_decrement: float | None = None)

A mode (MAP) in *constrained* coordinates with the optimizer's verdict.

pe: PointEstimate = backend.optimize(model, data, seed=0)

Posterior(draws: Mapping[str, ArrayLike], *, coords: Mapping[str, Sequence[Any]] | None = None, provenance: Mapping[str, Any] | None = None) -> None

Draws keyed by name, each ``(chain, draw, *shape)``, plus coords and provenance.

post = Posterior({"mu": good, "tau": np.exp(ar)}, provenance={"seed": 0, "divergences": 0})

PymcBackend(*, target_accept: float = 0.9, nuts_sampler: NutsSampler = 'pymc') -> None

``Backend`` over PyMC's NUTS. ``optimize`` and ``laplace`` reuse ``infer.laplace``.

pymc_backend = PymcBackend(nuts_sampler=choice, target_accept=0.9)

SampleSettings(draws: int = 1000, tune: int = 1000, chains: int = 4, target_accept: float = 0.9)

How much to sample. The defaults are the usual NUTS defaults.

settings = SampleSettings(draws=300, tune=300, chains=2)

SamplerStatus(name: str, missing: tuple[str, ...] = (), error: str = '') -> None

Whether one NUTS sampler can actually be used, and why not when it cannot.

probed: SamplerStatus = samplers(probe=True)[name] if status.usable else status

diagnose(posterior: SupportsPosterior, *, divergences: int | None = None, rhat_max: float = 1.01, ess_min: float = 400.0, divergences_max: int = 0) -> ConvergenceReport

Diagnose every variable of ``posterior`` and judge it against the thresholds.

report: ConvergenceReport = diagnose(post)

ess_bulk(draws: ArrayLike) -> float

Bulk ESS: ESS of the rank-normalized split chains.

print(f"{name:8s} rhat={split_rhat(d):.3f} ess_bulk={ess_bulk(d):7.1f} ess_tail={ess_tail(d):7.1f} mcse={mcse_mean(d):.4f}")

ess_tail(draws: ArrayLike) -> float

Tail ESS: the smaller of the ESS of the 5% and 95% quantile indicators.

print(f"{name:8s} rhat={split_rhat(d):.3f} ess_bulk={ess_bulk(d):7.1f} ess_tail={ess_tail(d):7.1f} mcse={mcse_mean(d):.4f}")

find_mode(model: ModelSpec, data: Mapping[str, ArrayLike], *, init: Mapping[str, ArrayLike] | None = None, derivatives: Derivatives = 'auto', seed: int | None = None, allow_unverified: bool = False) -> PointEstimate | Unverified

Posterior mode in unconstrained space, reported in constrained coordinates.

mode = find_mode(model, data)

from_inference_data(idata: Any) -> Posterior

Read the ``posterior`` group of an arviz object back into a ``Posterior``.

back = from_inference_data(idata)

get_backend(name: str) -> Backend | Unsupported

Resolve a backend by name; ``Unsupported`` names the missing extra.

lap = get_backend("laplace")

hessian_at(model: ModelSpec, data: Mapping[str, ArrayLike], z: Mapping[str, ArrayLike], *, derivatives: Derivatives = 'auto') -> Array | Unverified

Hessian of the negative log density at unconstrained ``z``, in ``flat_layout`` order.

H_fd = hessian_at(model, data, z, derivatives="finite_difference")

laplace(model: ModelSpec, data: Mapping[str, ArrayLike], *, draws: int, seed: int | None, init: Mapping[str, ArrayLike] | None = None, chains: int = 1, derivatives: Derivatives = 'auto', allow_unverified: bool = False) -> Posterior | Unverified

Gaussian approximation at the mode, drawn and mapped back to constrained space.

post = backend.laplace(model, data, draws=4000, seed=0)

mcse_mean(draws: ArrayLike) -> float

Monte-Carlo standard error of the posterior mean, ``sd / sqrt(ess)`` on the raw draws.

print(f"{name:8s} rhat={split_rhat(d):.3f} ess_bulk={ess_bulk(d):7.1f} ess_tail={ess_tail(d):7.1f} mcse={mcse_mean(d):.4f}")

samplers(*, probe: bool = False) -> dict[str, SamplerStatus]

Every NUTS sampler PyMC can dispatch to, and whether it is usable here.

probed: SamplerStatus = samplers(probe=True)[name] if status.usable else status

split_rhat(draws: ArrayLike) -> float

Rank-normalized split-R̂: the max of the bulk and the folded (tail) versions.

print(f"{name:8s} rhat={split_rhat(d):.3f} ess_bulk={ess_bulk(d):7.1f} ess_tail={ess_tail(d):7.1f} mcse={mcse_mean(d):.4f}")

to_inference_data(posterior: Posterior) -> Any | Unsupported

Build an arviz object with a ``posterior`` group; provenance goes into its attrs.

idata = to_inference_data(post)

AffineForm(coefficients: Mapping[str, Expr], intercept: Expr) -> None

``expr = sum_j coefficients[j] * unknown_j + intercept``, with expression coefficients.

form: AffineForm | None = affine_split(market.equation("quantity").rhs, ["quantity", "price"])

Block(variables: tuple[str, ...], simultaneous: bool)

One set of variables solved together, and whether it is a cycle.

first: Block = order.blocks[0]

BlockOrder(blocks: tuple[Block, ...] = (), system: str = '')

The system's blocks in solution order, sources first.

order: BlockOrder = block_order(system)

BlockSolution(variables: tuple[str, ...], method: Literal['substitution', 'linear', 'fixed_point'], expressions: dict[str, Const | Data | Param | Add | Mul | Div | Pow | Apply | Convolve | Reduce | Gather | Link | Opaque], residuals: dict[str, Const | Data | Param | Add | Mul | Div | Pow | Apply | Convolve | Reduce | Gather | Link | Opaque] = {}, sweeps: int = 0, node_count: int, detail: dict[str, str] = {})

How one block was compiled, and into what.

solution: BlockSolution = solve_block(
    {v: market.equation(v).rhs for v in ("quantity", "price")},
    dims,
    simultaneous=True,
)

DynamicEquation(target: str, rhs: Const | Data | Param | Add | Mul | Div | Pow | Apply | Convolve | Reduce | Gather | Link | Opaque, name: str = '')

``target[t] = rhs``, where ``rhs`` reads variables at ``t`` and earlier.

No executed statement uses this symbol — it appears in the notebooks only inside an import, which is all gate 12 requires. Worth a worked line.

DynamicSystem(name: str = '', variables: tuple[Variable, ...], equations: tuple[DynamicEquation, ...] = (), description: str = '')

A set of variables and one equation per endogenous variable.

DynamicSystem(variables=(Variable(name="y", dimension=NONE),), equations=())

The system is malformed: an unknown reference, a missing equation, a bad name.

No executed statement uses this symbol — it appears in the notebooks only inside an import, which is all gate 12 requires. Worth a worked line.

Form(*args, **kwargs)

form: Form = marginal.form

An equation refers to a future value; solving that needs rational expectations.

try:
    parse_ref("y.f1")
except LeadNotSupportedError as e:
    print("lead ->", e)

The equation text does not parse, or names something undeclared.

try:
    parse_equations("y = wobble", variables=(Variable(name="y", dimension=NONE),))
except ParseError as e:
    print("unknown name ->", e)

Role(*args, **kwargs)

role: Role = system.variable("inflow").role

SolveMethod(*args, **kwargs)

method: SolveMethod = solution.method

Unrolled(system: str = '', system_hash: str = '', form: Literal['conditional', 'marginal'], periods: int, expressions: dict[str, Const | Data | Param | Add | Mul | Div | Pow | Apply | Convolve | Reduce | Gather | Link | Opaque], solutions: tuple[BlockSolution, ...] = (), columns: tuple[str, ...] = (), edges: tuple[tuple[str, str], ...] = (), node_count: int, detail: dict[str, str] = {})

A compiled dynamic system: one expression per solved node, and how it got there.

No executed statement uses this symbol — it appears in the notebooks only inside an import, which is all gate 12 requires. Worth a worked line.

Variable(name: str, dimension: Dimension, role: Literal['endogenous', 'exogenous'] = 'endogenous', observed: bool = True, initial: float = 0.0, description: str = '')

One named quantity of the system, with its dimension and its role.

parse_equations("y = wobble", variables=(Variable(name="y", dimension=NONE),))

affine_split(expr: Expr, unknowns: Sequence[str]) -> AffineForm | None

Split ``expr`` into coefficients on ``unknowns`` and a remainder, or ``None``.

form: AffineForm | None = affine_split(market.equation("quantity").rhs, ["quantity", "price"])

block_order(system: DynamicSystem) -> BlockOrder

Block-triangular decomposition of the contemporaneous dependency graph.

order: BlockOrder = block_order(system)

conditional_form(system: DynamicSystem, *, sweeps: int = 0, max_nodes: int = 20000) -> Unrolled | Unsupported

Compile one generic period: ``y = f(exogenous at t, every lag as a column)``.

print("without sweeps:", conditional_form(saturating).reason[:120])

lag_ref(var: str, lag: int) -> str

``("y", 0) -> "y"``; ``("y", 2) -> "y.l2"``.

print(lag_ref("stock", 2), parse_ref("stock.l2"), time_ref("stock", 3))

lagged_columns(system: DynamicSystem) -> tuple[str, ...]

Every lagged reference the conditional form needs as a materialized column.

print("columns:", conditional.columns, "| lagged:", lagged_columns(system))

parse_equations(text: str, *, variables: Sequence[Variable], parameters: Sequence[Param] = ()) -> tuple[DynamicEquation, ...]

Parse ``target = expression`` lines against declared variables and parameters.

parse_equations("y = wobble", variables=(Variable(name="y", dimension=NONE),))

parse_ref(name: str) -> tuple[str, int]

``"y.l2" -> ("y", 2)``; ``"y" -> ("y", 0)``. A lead raises.

print(lag_ref("stock", 2), parse_ref("stock.l2"), time_ref("stock", 3))

parse_system(text: str, *, variables: Sequence[Variable], parameters: Sequence[Param] = (), name: str = '', description: str = '') -> DynamicSystem

Parse equations and assemble the ``DynamicSystem``, dimension checks included.

parse_system(
    "y = x",
    variables=(
        Variable(name="y", dimension=D.outcome),
        Variable(name="x", dimension=D.currency, role="exogenous"),
    ),
)

prepare_panel(system: DynamicSystem, frame: DataFrame, *, unit: str | None = None, time: str) -> DataFrame

Add the lagged columns the conditional form reads, per unit, in time order.

prepared = prepare_panel(system, frame, unit="unit", time="t")

reduced_form(equations: Mapping[str, Expr], dimensions: Mapping[str, Dimension], *, max_nodes: int = 20000) -> dict[str, Expr] | Unsupported

Solve ``v = rhs_v(v, ...)`` exactly when every right-hand side is affine in the unknowns.

solved = reduced_form(
    {v: market.equation(v).rhs for v in ("quantity", "price")},
    dims,
)

refs_in(expr: Model) -> tuple[tuple[str, int], ...]

Every ``(variable, lag)`` the expression reads, by first appearance.

print(refs_in(system.equation("stock").rhs))

solve_block(equations: Mapping[str, Expr], dimensions: Mapping[str, Dimension], *, simultaneous: bool, start: Mapping[str, Expr] | None = None, sweeps: int = 0, max_nodes: int = 20000) -> BlockSolution | Unsupported

Compile one block: substitution, exact reduced form, or declared fixed-point sweeps.

solution: BlockSolution = solve_block(
    {v: market.equation(v).rhs for v in ("quantity", "price")},
    dims,
    simultaneous=True,
)

strongly_connected_components(nodes: tuple[str, ...], edges: tuple[tuple[str, str], ...]) -> tuple[tuple[str, ...], ...]

Tarjan's SCCs, each sorted, the tuple in reverse topological order (sources last).

print(strongly_connected_components(("a", "b", "c"), (("a", "b"), ("b", "a"), ("b", "c"))))

time_ref(var: str, period: int) -> str

``("y", 3) -> "y.t3"`` — the name of ``y`` at absolute period 3.

print(lag_ref("stock", 2), parse_ref("stock.l2"), time_ref("stock", 3))

to_model_spec(unrolled: Unrolled, outcome: str, *, likelihood: Likelihood, priors: Mapping[str, Prior], extra_parameters: Sequence[Param] = (), period: int | None = None, name: str = '') -> ModelSpec

Wrap one compiled node as a ``ModelSpec`` that any backend fits.

model = to_model_spec(
    conditional,
    "stock",
    likelihood=Likelihood(family="normal", scale="sigma"),
    priors={
        "decay": Prior(family="beta", hyper={"alpha": 1.5, "beta": 1.5}),
        "beta": Prior(family="lognormal", hyper={"mu": 0.0, "sigma": 1.0}),
    },
    extra_parameters=(
        Param(name="sigma", dimension=D.outcome, prior=Prior(family="halfnormal", hyper={"sigma": 1.0})),
    ),
)

unroll(system: DynamicSystem, periods: int, *, sweeps: int = 0, max_nodes: int = 20000) -> Unrolled | Unsupported

Compile every period into expressions over exogenous inputs alone.

marginal = unroll(system, periods=4)

unrolled_edges(system: DynamicSystem, periods: int, *, reduced: bool = True) -> tuple[tuple[str, str], ...]

Edges of the time-indexed graph, ``("x.t1", "y.t2")``, sorted and deduplicated.

print("as written (may be cyclic):", unrolled_edges(system, periods=2, reduced=False))

CausalGraph(nodes: tuple[str, ...] = (), edges: tuple[tuple[str, str], ...] = (), bidirected: tuple[tuple[str, str], ...] = (), unmeasured: tuple[str, ...] = (), selection: tuple[str, ...] = (), feedback: bool = False, name: str = '')

A directed acyclic graph over named nodes, with optional bidirected edges.

g = CausalGraph.from_edges("Z -> X, Z -> Y, X -> M, M -> Y, X <-> W, W -> Y", unmeasured=["W"], name="toy")

ClusterDAG(clusters: dict[str, tuple[str, ...]], edges: tuple[tuple[str, str], ...] = (), bidirected: tuple[tuple[str, str], ...] = (), unmeasured: tuple[str, ...] = (), name: str = '')

A DAG over clusters of variables, with the micro-level structure left open.

market = ClusterDAG(
    clusters={"cost": ("wage", "fuel"), "price": ("price",), "demand": ("units",)},
    edges=(("cost", "price"), ("price", "demand")),
    name="a two-block market",
)

Density(node: Literal['density'] = 'density', outcomes: tuple[str, ...], given: tuple[str, ...] = ())

``P(outcomes | given)`` — a conditional of the observational distribution.

ratio_example: Ratio = Ratio(
    numerator=Density(outcomes=("X", "Y")), denominator=Density(outcomes=("X",))
)

EndogeneityTest(statistic: float, p_value: float, df: int, method: str, conclusion: Literal['endogenous', 'exogenous'], alpha: float = 0.05, treatment: str = '', outcome: str = '', detail: dict[str, float] = {})

Outcome of an endogeneity test, with the level the conclusion used.

print(dwh if not isinstance(dwh, EndogeneityTest) else (dwh.conclusion, round(dwh.statistic, 2), dwh.p_value))

Formula(*args, **kwargs)

Runtime representation of an annotated type.

formula: Formula = front_door.formula

FrontDoorRoute(mediators: tuple[str, ...], treatment: str, outcome: str)

A front-door identification route: ``P(y | do(x))`` through ``mediators``.

route = FrontDoorRoute(mediators=("M",), treatment="X", outcome="Y")

The graph is malformed (cycle, unknown node, self-edge).

try:
    CausalGraph.from_edges("A -> B, B -> C, C -> A")
except GraphError as e:
    print("refused:", e)

Hedge(root: tuple[str, ...], subset: tuple[str, ...], variables: tuple[str, ...])

The witness that an effect is *not* identifiable — a proof, not a failed search.

hedge: Hedge = blocked.hedge

IdentificationVerdict(treatment: str, outcome: str, graph_hash: str, verdict: Verdict, route: Literal['no_causal_path', 'backdoor', 'frontdoor', 'instrument', 'backdoor_unmeasured', 'none'], adjustment_set: tuple[str, ...] = (), mediators: tuple[str, ...] = (), instrument: str = '', conditioning: tuple[str, ...] = (), unmeasured_required: tuple[str, ...] = (), alternatives: tuple[Literal['no_causal_path', 'backdoor', 'frontdoor', 'instrument', 'backdoor_unmeasured', 'none'], ...] = (), search_limits_hit: tuple[str, ...] = (), ... and 1 more field)

What the graph says about ``treatment -> outcome``, and by which route.

v: IdentificationVerdict = identify(simple, "X", "Y")

IdentifiedEffect(treatment: tuple[str, ...], outcome: tuple[str, ...], conditioned: tuple[str, ...] = (), graph_hash: str = '', formula: Optional[Density | Product | Marginal | Ratio] = None, hedge: Hedge | None = None, verdict: Verdict, detail: dict[str, str] = {})

What ID returned: a formula, or a hedge, and a verdict either way.

result: IdentifiedEffect = identify_effect(graph, "X", "Y")

InstrumentRoute(instrument: str, conditioning: tuple[str, ...] = (), treatment: str, outcome: str)

An instrumental-variable route: ``instrument`` for ``treatment -> outcome``.

print(InstrumentRoute(instrument="Z", conditioning=("W",), treatment="X", outcome="Y"))

JointTable(names: tuple[str, ...], table: Array) -> None

A joint distribution over discrete variables, as an array with one axis per variable.

joint = JointTable(names=("X", "Y", "Z"), table=observational)

LinearEstimate(estimate: float, se: float, n: int, method: Literal['ols', '2sls', 'frontdoor'], treatment: str, outcome: str, covariates: tuple[str, ...] = (), detail: dict[str, float] = {})

A point estimate with its classical standard error and provenance.

naive: LinearEstimate = ols(frame, "Y", "X")

Marginal(node: Literal['marginal'] = 'marginal', over: tuple[str, ...], term: Density | Product | Marginal | Ratio)

``sum over `over` of `term``` — marginalization.

print("is it a Marginal of a Product of Density terms?", isinstance(formula, Marginal))

MixedGraph(nodes: tuple[str, ...] = (), edges: tuple[tuple[str, str], ...] = (), bidirected: tuple[tuple[str, str], ...] = (), unmeasured: tuple[str, ...] = (), name: str = '')

A directed mixed graph. Directed edges may form cycles; separation is sigma.

loop = MixedGraph.from_edges("x -> a, a -> b, b -> a, b -> y")

Product(node: Literal['product'] = 'product', factors: tuple[Density | Product | Marginal | Ratio, ...])

A product of formulas.

product_example: Product = Product(
    factors=(Density(outcomes=("Y",), given=("X",)), Density(outcomes=("X",)))
)

Ratio(node: Literal['ratio'] = 'ratio', numerator: Density | Product | Marginal | Ratio, denominator: Density | Product | Marginal | Ratio)

``numerator / denominator``; how a conditional of a derived expression is written.

ratio_example: Ratio = Ratio(
    numerator=Density(outcomes=("X", "Y")), denominator=Density(outcomes=("X",))
)

Role(*args, **kwargs)

r: dict[str, Role] = roles(m_bias, "X", "Y")

RoleAssignment(graph_hash: str, treatment: str, outcome: str, roles: dict[str, Literal['treatment', 'outcome', 'confounder', 'mediator', 'collider', 'instrument', 'proxy', 'descendant_of_outcome', 'neutral']])

The roles of every node of one graph relative to one treatment–outcome pair.

ra: RoleAssignment = assign_roles(g, "X", "Y")

Route(*args, **kwargs)

pref: tuple[Route, ...] = ("frontdoor", "backdoor")

SequentialPlan(stages: tuple[str, ...], outcome: str, graph_hash: str, adjustments: tuple[tuple[str, ...], ...], verdict: Verdict, static_adjustment_fails: bool = False, detail: dict[str, str] = {})

Per-stage adjustment sets for a sequence of treatments, and the verdict on them.

plan: SequentialPlan = sequential_plan(graph, ["dose.t0", "dose.t1", "dose.t2"], "outcome.t2")

TransportVerdict(treatment: str, outcome: str, verdict: Verdict, s_admissible_set: tuple[str, ...] = (), mediators: tuple[str, ...] = (), given: tuple[str, ...] | None = None, selection: tuple[str, ...] = (), graph_hash: str = '', formula: str = '', missing: tuple[str, ...] = ())

Whether, and by what formula, an effect transports from source to target.

unadjusted: TransportVerdict = transport_verdict(g, "X", "Y", given=())

acyclify(graph: MixedGraph | CausalGraph) -> CausalGraph

The acyclification of a directed mixed graph (Mooij & Claassen 2020, Def. 3).

print("acyclified:", acyclify(loop).to_text())

adjustment_sets(graph: CausalGraph, x: str, y: str, *, measured_only: bool = True, max_candidates: int = 16) -> AdjustmentSets | Unsupported

Every back-door admissible set among the candidates, deterministically ordered.

print(adjustment_sets(simple, "X", "Y"), minimal_adjustment_sets(simple, "X", "Y"))

admissible_set_exists(graph: CausalGraph, x: str, y: str, *, measured_only: bool = True) -> bool

Whether any back-door admissible set exists, in polynomial time.

print(canonical_adjustment_set(simple, "X", "Y"), admissible_set_exists(simple, "X", "Y"))

assign_roles(graph: CausalGraph, x: str, y: str) -> RoleAssignment

``roles`` packaged as a ``Spec`` with the graph's content hash.

ra: RoleAssignment = assign_roles(g, "X", "Y")

backdoor_admissible(graph: CausalGraph, x: str, y: str, z: Iterable[str]) -> bool

Pearl's back-door criterion for ``z`` relative to ``(x, y)``.

print(backdoor_admissible(simple, "X", "Y", ["Z"]), backdoor_admissible(simple, "X", "Y", []))

canonical_adjustment_set(graph: CausalGraph, x: str, y: str, *, measured_only: bool = True) -> frozenset[str] | None

The canonical back-door set, or ``None`` when no admissible set exists.

print(canonical_adjustment_set(simple, "X", "Y"), admissible_set_exists(simple, "X", "Y"))

compatible(cdag: ClusterDAG, graph: CausalGraph) -> bool

Whether a micro-level DAG is compatible with the cluster-level claim.

print("compatible:", compatible(market, micro))

conditional_instruments(graph: CausalGraph, x: str, y: str, *, measured_only: bool = True, max_candidates: int = 12, max_conditioning: int = 2) -> tuple[tuple[str, frozenset[str]], ...] | Unsupported

All ``(z, W)`` pairs with ``|W| <= max_conditioning`` that satisfy the criterion.

print(instruments(civ, "X", "Y"), conditional_instruments(civ, "X", "Y"))

directly_transportable(graph: CausalGraph, x: str, y: str) -> bool

Whether ``P*(y | do(x)) = P(y | do(x))`` (B&P 2014 Def. 7).

print("direct:", directly_transportable(g, "X", "Y"))

districts(graph: CausalGraph) -> tuple[tuple[str, ...], ...]

The C-components: connected components of the bidirected part (Tian & Pearl 2002).

print("districts:", districts(admg))

durbin_wu_hausman(frame: DataFrame, y: str, x: str, instruments: str | Sequence[str], covariates: str | Sequence[str] = (), alpha: float = 0.05) -> EndogeneityTest | Unverified

Control-function Durbin–Wu–Hausman test of the exogeneity of ``x``.

dwh = durbin_wu_hausman(frame, "Y", "X", instruments=["Z"])

frontdoor_admissible(graph: CausalGraph, x: str, y: str, m: Iterable[str]) -> bool

Whether ``m`` satisfies the front-door criterion relative to ``(x, y)``.

print(frontdoor_admissible(fd_graph, "X", "Y", ["M"]), frontdoor_sets(fd_graph, "X", "Y"))

frontdoor_linear(frame: DataFrame, y: str, x: str, mediators: str | Sequence[str], covariates: str | Sequence[str] = ()) -> LinearEstimate

Linear front-door estimate of the effect of ``x`` on ``y`` through ``mediators``.

fd = frontdoor_linear(frame, "Y", "X", mediators=v.mediators)

frontdoor_sets(graph: CausalGraph, x: str, y: str, *, measured_only: bool = True, max_candidates: int = 12) -> tuple[frozenset[str], ...] | Unsupported

Every front-door admissible mediator set, smallest first, then lexicographic.

print(frontdoor_admissible(fd_graph, "X", "Y", ["M"]), frontdoor_sets(fd_graph, "X", "Y"))

hausman_iv_vs_ols(est_ols: LinearEstimate, est_2sls: LinearEstimate, alpha: float = 0.05) -> EndogeneityTest | Unverified

Classic Hausman contrast of an OLS and a 2SLS estimate of the same effect.

h = hausman_iv_vs_ols(ols(frame, "Y", "X"), two_stage_least_squares(frame, "Y", "X", instruments=["Z"]))

identify(graph: CausalGraph, treatment: str, outcome: str, *, prefer: Sequence[Route] = ..., has_carryover: bool = False, max_candidates: int = 16) -> IdentificationVerdict

Decide how (and whether) ``treatment -> outcome`` is identified in ``graph``.

v: IdentificationVerdict = identify(simple, "X", "Y")

identify_cluster_effect(cdag: ClusterDAG, treatment: str | Iterable[str], outcome: str | Iterable[str]) -> IdentifiedEffect

``P(outcome cluster | do(treatment cluster))`` by running ID on the cluster graph.

print("effect of the cost block on demand:", identify_cluster_effect(market, "cost", "demand").to_text())

identify_conditional_effect(graph: CausalGraph, treatment: str | Iterable[str], outcome: str | Iterable[str], conditioned: Iterable[str]) -> IdentifiedEffect

``P(outcome | do(treatment), conditioned)`` by IDC (Shpitser & Pearl 2008).

conditional = identify_conditional_effect(
    CausalGraph.from_edges("Z -> X, Z -> Y, X -> Y, X -> W, W -> Y"), "X", "Y", ["W"]
)

identify_effect(graph: CausalGraph, treatment: str | Iterable[str], outcome: str | Iterable[str]) -> IdentifiedEffect

``P(outcome | do(treatment))`` by the ID algorithm: the estimand, or the hedge.

result: IdentifiedEffect = identify_effect(graph, "X", "Y")

instrument_admissible(graph: CausalGraph, x: str, y: str, z: str, conditioning: Iterable[str] = ()) -> bool

Whether ``z`` is an instrument for the total effect of ``x`` on ``y`` given ``W``.

print(instruments(iv_graph, "X", "Y"), instrument_admissible(iv_graph, "X", "Y", "Z"))

instruments(graph: CausalGraph, x: str, y: str, *, measured_only: bool = True) -> tuple[str, ...]

All unconditional instruments (``W = ∅``) for ``x -> y``, sorted by name.

print(instruments(iv_graph, "X", "Y"), instrument_admissible(iv_graph, "X", "Y", "Z"))

latent_projection(graph: CausalGraph, keep: Iterable[str] | None = None) -> CausalGraph

Project the unmeasured nodes out, leaving an ADMG over the ones kept.

admg = latent_projection(explicit)

minimal_adjustment_sets(graph: CausalGraph, x: str, y: str, *, measured_only: bool = True, max_candidates: int = 16) -> AdjustmentSets | Unsupported

Inclusion-minimal back-door admissible sets, ordered as ``adjustment_sets``.

print(adjustment_sets(simple, "X", "Y"), minimal_adjustment_sets(simple, "X", "Y"))

minimal_s_admissible_sets(graph: CausalGraph, x: str, y: str, *, measured_only: bool = True, max_candidates: int = 14) -> tuple[frozenset[str], ...] | Unsupported

The S-admissible sets with no S-admissible proper subset, smallest first.

print(s_admissible_sets(g, "X", "Y"), minimal_s_admissible_sets(g, "X", "Y"))

ols(frame: DataFrame, y: str, x: str, covariates: str | Sequence[str] = ()) -> LinearEstimate

Coefficient on ``x`` in OLS of ``y`` on ``[1, x, covariates]``.

naive: LinearEstimate = ols(frame, "Y", "X")

parse_edges(text: str) -> tuple[tuple[Edge, ...], tuple[Edge, ...]]

Parse ``"Z -> X, Z -> Y, X -> Y, A <-> B"`` into (directed, bidirected) edge tuples.

print(parse_edges("A -> B; C <- B"))

requires_unmeasured(graph: CausalGraph, x: str, y: str) -> bool

True iff adjustment is possible only through a node the panel does not observe.

print("needs an unmeasured node:", requires_unmeasured(g, "X", "Y"))

roles(graph: CausalGraph, x: str, y: str) -> dict[str, Role]

Assign each node a descriptive role relative to ``(x, y)``.

r: dict[str, Role] = roles(m_bias, "X", "Y")

s_admissible(graph: CausalGraph, x: str, y: str, z: Iterable[str]) -> bool

Whether ``Z`` is S-admissible for the effect of ``x`` on ``y`` (B&P 2014 Def. 8).

print("S-admissible {Z}:", s_admissible(g, "X", "Y", ["Z"]), "| {}:", s_admissible(g, "X", "Y", []))

s_admissible_sets(graph: CausalGraph, x: str, y: str, *, measured_only: bool = True, max_candidates: int = 14) -> tuple[frozenset[str], ...] | Unsupported

Every S-admissible set, smallest first, then lexicographic.

print(s_admissible_sets(g, "X", "Y"), minimal_s_admissible_sets(g, "X", "Y"))

selection_diagram(graph: CausalGraph) -> CausalGraph

The selection diagram of ``graph`` as a plain DAG (B&P 2014 Def. 4).

sd = selection_diagram(g)

sequential_backdoor_admissible(graph: CausalGraph, treatments: Sequence[str], outcome: str, adjustments: Sequence[Iterable[str]], *, measured_only: bool = True) -> tuple[bool, str]

The sequential back-door criterion; ``(True, "")`` or ``(False, why)``.

ok, why = sequential_backdoor_admissible(
    graph, ["dose.t0", "dose.t1"], "outcome.t2", [("outcome.t1",), ()]
)

sequential_ignorability(graph: CausalGraph, stages: Sequence[str], outcome: str, adjustments: Sequence[Iterable[str]]) -> tuple[bool, str]

Sequential ignorability, read off one SWIG with every stage split.

print("SWIG agrees   :", sequential_ignorability(graph, stages, "outcome.t2", plan.adjustments))

sequential_plan(graph: CausalGraph, treatments: Sequence[str], outcome: str, *, candidates: Iterable[str] | None = None, prune: bool = True) -> SequentialPlan

Search for per-stage adjustment sets satisfying the sequential back-door criterion.

plan: SequentialPlan = sequential_plan(graph, ["dose.t0", "dose.t1", "dose.t2"], "outcome.t2")

sigma_separated(graph: MixedGraph | CausalGraph, x: str | Iterable[str], y: str | Iterable[str], z: Iterable[str] = ()) -> bool

Sigma-separation of ``X`` and ``Y`` given ``Z`` (Forré & Mooij 2017).

print("   quantity indep price given the block's parents?",
      structural.sigma_separated("quantity.t0", "price.t0", ["cost.t0"]))

single_world_ignorability(graph: CausalGraph, treatment: str, outcome: str, adjustment: Iterable[str] = (), *, label: str = '') -> bool

Whether ``Y(x)`` is independent of ``X`` given ``Z`` on the SWIG.

print(f"Y(x) indep X | {adjustment or '{}'}:", single_world_ignorability(g, "X", "Y", adjustment))

swig(graph: CausalGraph, interventions: Mapping[str, str] | Iterable[str]) -> CausalGraph

Split every intervened node; relabel its descendants as potential outcomes.

print("swig    :", swig(g, ["X"]).to_text())

transport_verdict(graph: CausalGraph, x: str, y: str, *, given: Iterable[str] | None = None, max_candidates: int = 14) -> TransportVerdict

Read the transportability of ``P*(y | do(x))`` off ``graph``'s selection diagram.

unadjusted: TransportVerdict = transport_verdict(g, "X", "Y", given=())

trivially_transportable(graph: CausalGraph, x: str, y: str, *, max_candidates: int = 14) -> bool

Whether ``P*(y | do(x))`` is identifiable from the target alone (B&P 2014 Def. 6).

print("trivially (from target data alone):", trivially_transportable(g, "X", "Y"))

two_stage_least_squares(frame: DataFrame, y: str, x: str, instruments: str | Sequence[str], covariates: str | Sequence[str] = ()) -> LinearEstimate

2SLS of ``y`` on ``x`` with ``instruments``, adjusting for ``covariates``.

iv = two_stage_least_squares(frame, "Y", "X", instruments=[v.instrument])

unrolled_graph(system: DynamicSystem, periods: int, *, name: str = '') -> CausalGraph

The acyclification of the time-indexed graph: an acyclic graph, honestly obtained.

graph = unrolled_graph(system, periods=3)

unrolled_mixed_graph(system: DynamicSystem, periods: int, *, name: str = '') -> MixedGraph

The time-indexed graph exactly as the equations were written, cycles included.

structural = unrolled_mixed_graph(market, periods=2)

weak_instrument_check(est: LinearEstimate, threshold: float = 10.0) -> Assumption

The instrument-strength assumption, checked against the first-stage F.

strength = weak_instrument_check(iv)

Dataset(values: Array, names: tuple[str, ...], regimes: tuple[frozenset[str], ...]) -> None

Rows of measurements, each labelled with what was intervened on when it was taken.

return Dataset(np.column_stack([values[n] for n in order]), tuple(order), (frozenset(do),) * n)

DiscoveryResult(essential: EssentialGraph, score: float, n_observations: int, targets: tuple[tuple[str, ...], ...] = (), phases: tuple[str, ...] = (), steps: tuple[str, ...] = (), penalty: float, limits_hit: tuple[str, ...] = (), detail: dict[str, str] = {})

What a structure search found, and how far it is entitled to be believed.

found: DiscoveryResult = ges(GaussianBIC(observational))

EdgeSupport(a: str, b: str, adjacent: float, forward: float, backward: float, undirected: float)

How often one pair of variables was joined, and which way it pointed.

assert isinstance(support, EdgeSupport)

EssentialGraph(nodes: tuple[str, ...] = (), directed: tuple[tuple[str, str], ...] = (), undirected: tuple[tuple[str, str], ...] = (), targets: tuple[tuple[str, ...], ...] = (), name: str = '')

A partially directed graph: compelled edges directed, the rest undirected.

partial = EssentialGraph(nodes=truth.nodes, directed=(("a", "b"),), undirected=(("b", "c"),))

GaussianBIC(data: Dataset, penalty: float = 1.0) -> None

Decomposable BIC for linear-Gaussian structure, aware of which rows were randomized.

found: DiscoveryResult = ges(GaussianBIC(observational))

IndependenceResult(x: str, y: str, given: tuple[str, ...] = (), correlation: float, p_value: float, n: int, statistic: float = 0.0)

One conditional-independence test: what was asked, and both answers.

assert isinstance(check, IndependenceResult)

Mark(*args, **kwargs)

left: Mark = edge.mark_a

PAG(nodes: tuple[str, ...] = (), edges: tuple[PagEdge, ...] = (), limits_hit: tuple[str, ...] = (), detail: dict[str, str] = {})

A partial ancestral graph: what the data determines when latents are possible.

pag: PAG = fci(observed, oracle_independence(world))

PagEdge(a: str, b: str, mark_a: Literal['circle', 'arrow', 'tail'] = 'circle', mark_b: Literal['circle', 'arrow', 'tail'] = 'circle')

One edge of a PAG: two variables and the mark at each end.

assert isinstance(edge, PagEdge)

PartialCorrelation(data: Dataset) -> None

A Gaussian conditional-independence test over a dataset.

check = PartialCorrelation(measured).test("seed", "harvest")

StabilityReport(variables: tuple[str, ...], edges: tuple[EdgeSupport, ...] = (), n_bootstrap: int, n_rows: int, targets: tuple[tuple[str, ...], ...] = (), penalty: float, seed: int | None = None, detail: dict[str, str] = {})

Per-edge support across bootstrap resamples of the same data.

assert isinstance(stability, StabilityReport)

consistent_extension(graph: EssentialGraph) -> CausalGraph

One DAG from the class, by Dor & Tarsi's (1992) algorithm.

print("a representative:", consistent_extension(essential).to_text())

consistent_extensions(graph: EssentialGraph, *, limit: int = 4096) -> tuple[CausalGraph, ...]

Every DAG in the class, by orienting the undirected edges every way.

for member in consistent_extensions(essential):
    print(" ", member.to_text())

cpdag(graph: CausalGraph) -> EssentialGraph

The essential graph of a DAG: what observational data could ever orient.

print("chain    :", cpdag(CausalGraph.from_edges("a -> b, b -> c")).to_text())

edge_stability(data: Dataset, *, n_bootstrap: int = 100, penalty: float = 1.0, targets: Sequence[Iterable[str]] | None = None, seed: int | None = None) -> StabilityReport

Resample the rows, re-run the search, and count how often each edge returns.

stability = edge_stability(trial, n_bootstrap=60, seed=7)

fci(variables: Sequence[str], independent: Independent, *, max_conditioning: int = 3, name: str = '') -> PAG

Discover a PAG without assuming causal sufficiency.

pag: PAG = fci(observed, oracle_independence(world))

fci_from_data(data: Dataset, *, alpha: float = 0.05, max_conditioning: int = 3, variables: Iterable[str] | None = None, name: str = '') -> PAG

Run FCI with a partial-correlation test at level ``alpha``.

from_data = fci_from_data(measured, alpha=0.01)

ges(score: GaussianBIC, *, name: str = '') -> DiscoveryResult

Greedy equivalence search on observational data: forward, then backward.

found: DiscoveryResult = ges(GaussianBIC(observational))

gies(score: GaussianBIC, targets: Sequence[Iterable[str]] | None = None, *, name: str = '') -> DiscoveryResult

Greedy interventional equivalence search: the same, over I-essential graphs.

one = gies(GaussianBIC(pooled))

interventional_essential_graph(graph: CausalGraph, targets: Sequence[Iterable[str]]) -> EssentialGraph

What the data could orient after intervening on each target in the family.

after = interventional_essential_graph(truth, [["a"]])

markov_equivalent(first: CausalGraph, second: CausalGraph) -> bool

Same skeleton and same v-structures — the classical characterization.

print("all Markov equivalent to the truth:",
      all(markov_equivalent(m, truth) for m in consistent_extensions(essential)))

meek_closure(graph: EssentialGraph) -> EssentialGraph

Orient every edge forced by what is already oriented (Meek 1995, rules 1-4).

print("R1 propagation:", meek_closure(partial).to_text())

oracle_independence(graph: CausalGraph) -> Independent

A perfect conditional-independence test, answered by d-separation in a known graph.

pag: PAG = fci(observed, oracle_independence(world))

orientation_gain(graph: CausalGraph, targets: Sequence[Iterable[str]], *, baseline: Sequence[Iterable[str]] = ()) -> tuple[Edge, ...]

The edges an experiment would orient that are undirected without it.

gain = orientation_gain(truth, [target])

v_structures(graph: CausalGraph) -> frozenset[tuple[str, str, str]]

``(a, c, b)`` with ``a -> c <- b`` and ``a``, ``b`` non-adjacent; ``a < b``.

print("v-structures of the collider:", v_structures(CausalGraph.from_edges("a -> c, b -> c")))

Estimand(name: str, quantity: Quantity, treatment: Treatment, intervention: Intervention, reference: Intervention | None = None, outcome: Outcome, population: Population, window: TimeWindow, level: Level, conditioning: tuple[str, ...] = (), dimension: Dimension, description: str = '')

A named, versioned, content-hashed counterfactual quantity. All facets required.

lift = Estimand(
    name="lift_at_100",
    quantity=Quantity(kind="contrast"),
    treatment=fertilizer,
    intervention=Intervention(doses={"fertilizer": 100.0}, version="granular"),
    reference=Intervention(doses={"fertilizer": 0.0}, version="granular"),
    outcome=yield_total,
    population=north,
    window=season,
    level=Level(unit="cluster", interference="none"),
    conditioning=(),
    dimension=D.outcome,
    description="season-total yield lift from 100 USD of granular fertilizer vs none, north region",
)

EstimandRegistry(estimands: Iterable[Estimand] = ()) -> None

Estimands by name, with their content hashes.

registry: EstimandRegistry = standard_estimands(
    treatment=spec.treatments[0], outcome=spec.outcome, population=Population(name="all"),
    window=TimeWindow(start=0, stop=1), level=Level(unit="individual"), dose=60.0, reference_dose=0.0,
)

EstimandResult(estimand_hash: str, estimand_name: str, kind: Literal['contrast', 'marginal', 'ratio', 'elasticity', 'area'], summary: Summary, dimension: Dimension, unit: str | None = None, status: Literal['identified', 'downgraded', 'blocked', 'unsupported'], identification: Verdict | None = None, assumptions: tuple[Assumption, ...] = (), ledger: tuple[LedgerLine, ...] = (), n_draws: int, producer_hash: str = '', ... and 1 more field)

A realized estimand: the summary, its dimension and unit, and how it was licensed.

if isinstance(asserted, EstimandResult):
    status: ResultStatus = asserted.status
    print(status, [line.kind for line in asserted.ledger], [a.name for a in asserted.assumptions])

Built-in immutable sequence.

print(FACETS)

Facet(*args, **kwargs)

f: Facet = "intervention"

FacetDiff(facet: Literal['quantity', 'intervention', 'outcome', 'population', 'window', 'level', 'conditioning', 'dimension'], statement: str, assumptions: tuple[Assumption, ...] = (), blocked: Blocked | None = None, corrections: tuple[str, ...] = ())

What one differing facet requires: assumptions, or a block, plus any correction.

entry: FacetDiff = plan.entry("intervention")

Level(unit: Literal['individual', 'cluster', 'aggregate'], interference: Literal['none', 'within_cluster', 'declared'] = 'none', interference_model: str = '')

Unit of analysis and interference model (review B2: SUTVA-2).

lift = Estimand(
    name="lift_at_100",
    quantity=Quantity(kind="contrast"),
    treatment=fertilizer,
    intervention=Intervention(doses={"fertilizer": 100.0}, version="granular"),
    reference=Intervention(doses={"fertilizer": 0.0}, version="granular"),
    outcome=yield_total,
    population=north,
    window=season,
    level=Level(unit="cluster", interference="none"),
    conditioning=(),
    dimension=D.outcome,
    description="season-total yield lift from 100 USD of granular fertilizer vs none, north region",
)

Quantity(kind: Literal['contrast', 'marginal', 'ratio', 'elasticity', 'area'], scale: Literal['natural', 'log'] = 'natural')

The functional applied to the response.

mroi = lift.model_copy(update={
    "name": "marginal_at_100",
    "quantity": Quantity(kind="marginal"),
    "reference": None,
    "dimension": D.outcome / D.currency,
})

QuantityKind(*args, **kwargs)

kinds: list[QuantityKind] = ["contrast", "marginal", "ratio", "elasticity", "area"]

RealizedDraws(result: EstimandResult, draws: Array) -> None

``realize(..., keep_draws=True)``: the result plus the draws it summarizes, flat.

if isinstance(rd, RealizedDraws):
    print(rd.draws.shape, rd.result.summary.n)

ResultStatus(*args, **kwargs)

status: ResultStatus = asserted.status

TransferPlan(status: Literal['identified', 'downgraded', 'blocked', 'unsupported', 'unverified'], source: str, target: str, differing: tuple[Literal['quantity', 'intervention', 'outcome', 'population', 'window', 'level', 'conditioning', 'dimension'], ...], entries: tuple[FacetDiff, ...], assumptions: tuple[Assumption, ...], corrections: tuple[str, ...] = (), ledger_lines: tuple[LedgerLine, ...], reason: str = '')

The typed diff between two estimands and what licenses reading one as the other.

plan: TransferPlan = experiment.transfer_to(national)

check_estimand_dimension(estimand: Estimand, expr: Expr) -> Dimension

``dimension(expr)``, asserted equal to the estimand's declared dimension (gate 10).

print(dimension(tree), "==", check_estimand_dimension(lift, tree))

derived_dimension(kind: QuantityKind, outcome: Dimension, dose: Dimension) -> Dimension

print(f"{kind:11s} -> {derived_dimension(kind, D.outcome, D.currency)}")

estimand_expr(estimand: Estimand, *, mean: Expr, treatment_data: Data, reference: Mapping[str, float] | None = None, derivative: Expr | None = None) -> Expr | Unsupported

The estimand as a per-row expression over ``mean``, or a typed ``Unsupported``.

tree = estimand_expr(lift, mean=world.model.mean, treatment_data=Data(name="dose", dimension=D.currency))

evaluate(estimands: Sequence[Estimand], producer: SupportsEstimands, *, verdicts: Mapping[str, Verdict] | None = None, assume_identified: bool = False, definition: IntervalDefinition = 'hdi', mass: float = 0.9, seed: int | None = None, unit_weights: Mapping[str, float] | None = None, unit_strata: Mapping[str, str] | None = None) -> dict[str, EstimandResult | Unsupported | Blocked]

``realize`` each estimand, keyed by name; ``verdicts`` maps estimand name to its verdict.

results = evaluate(list(registry), res, verdicts={name: verdict for name in registry.names()}, definition="hdi", mass=0.9)

realize(estimand: Estimand, producer: SupportsEstimands, *, verdict: Verdict | None = None, assume_identified: bool = False, definition: IntervalDefinition = 'hdi', mass: float = 0.9, seed: int | None = None, unit_weights: Mapping[str, float] | None = None, unit_strata: Mapping[str, str] | None = None, keep_draws: bool = False) -> EstimandResult | RealizedDraws | Unsupported | Blocked

Realize one estimand against a producer; see the module docstring for the arithmetic.

print(realize(lift, crippled))

standard_estimands(treatment: Treatment, outcome: Outcome, population: Population, window: TimeWindow, level: Level, *, dose: float, reference_dose: float = 0.0, version: str = 'unspecified') -> EstimandRegistry

The five domain-general estimands for one treatment at one dose.

registry: EstimandRegistry = standard_estimands(
    treatment=spec.treatments[0], outcome=spec.outcome, population=Population(name="all"),
    window=TimeWindow(start=0, stop=1), level=Level(unit="individual"), dose=60.0, reference_dose=0.0,
)

substitute(expr: Expr, replacements: Mapping[str, Expr]) -> Expr

Rebuild ``expr`` with every ``Data`` node named in ``replacements`` swapped out.

pinned = substitute(world.model.mean, {"dose": __import__("axiom.core", fromlist=["Const"]).Const(value=60.0, dimension=D.currency)})

Allocation(doses: dict[str, float], expected_outcome: float, budget: float, objective: Literal['mean', 'quantile'], q: float = 0.5, method: str, status: Literal['converged'] = 'converged', seed: int | None = None, detail: dict[str, float] = {})

One allocation: a dose per treatment and the objective it achieves.

if isinstance(alloc, Allocation):
    print(alloc.doses, round(alloc.expected_outcome, 3), alloc.status, round(alloc.total_dose, 3))
else:
    show(alloc)

AllocationMethod(*args, **kwargs)

meth: AllocationMethod = "slsqp"

AnyCarryover(*args, **kwargs)

Runtime representation of an annotated type.

cc: AnyCarryover = c

AnyKernel(*args, **kwargs)

Runtime representation of an annotated type.

k: AnyKernel = fam

AnyNuisanceTerm(*args, **kwargs)

Runtime representation of an annotated type.

any_term: AnyNuisanceTerm = FourierSeasonality(period=12.0, order=1)

AscentPath(treatments: tuple[str, ...], points: tuple[tuple[float, ...], ...], values: tuple[float, ...], stop: Literal['decrease', 'converged', 'boundary', 'n_steps'])

The visited points of a steepest-ascent walk and the surface value at each.

if isinstance(path, AscentPath):
    print(path.n, path.stop, path.best(), round(path.values[-1], 3))
else:
    show(path)

Build an immutable unordered collection of unique elements.

print("\ncoefficient priors may be signed:", sorted(BASIS_PRIOR_FAMILIES))

BandKind(*args, **kwargs)

kind: BandKind = band.kind

Bounds(treatments: tuple[str, ...], low: tuple[float, ...], high: tuple[float, ...])

The experimental region: a box ``[low_i, high_i]`` per treatment, in dose units.

bounds = Bounds(treatments=("a", "b"), low=(0.0, 0.0), high=(100.0, 40.0))

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

print(sorted(KERNELS), "|", sorted(CARRYOVERS))

CarryoverKernel(*args, **kwargs)

A lag-weight family that builds its own expression tree.

geo: CarryoverKernel = GeometricCarryover(max_lag=6)

CarryoverRole(*args, **kwargs)

role: CarryoverRole = "shape"

CovarianceFamily(*args, **kwargs)

covariance: CovarianceFamily = gp.covariance

Criterion(*args, **kwargs)

crit: Criterion = "d"

DelayedCarryover(name: Literal['delayed'] = 'delayed', max_lag: int)

``w_l ∝ λ^((l − θ)^2)``: a peak at lag ``θ`` with symmetric geometric decay around it.

delayed = DelayedCarryover(max_lag=8)

Design(treatments: tuple[str, ...], points: tuple[tuple[float, ...], ...], kind: str, detail: dict[str, float] = {})

A set of dose points (``points[i][j]`` is treatment ``j`` at run ``i``) and its recipe.

ccd: Design = central_composite(bounds, alpha="rotatable", center_points=3)

EventIndicators(kind: Literal['events'] = 'events', events: tuple[str, ...], coefficient_scale: float = 1.0)

User-supplied 0/1 columns, one coefficient each.

No executed statement uses this symbol — it appears in the notebooks only inside an import, which is all gate 12 requires. Worth a worked line.

ExponentialKernel(name: Literal['exponential'] = 'exponential', reference_dose: float = 1.0, amplitude_scale: float = 1.0, amplitude_prior: Prior | None = None)

``f(u) = 1 − exp(−u)``, ``u = dose / k``. Saturating; ``f(0) = 0``; concave everywhere.

for fam in (LogisticKernel(reference_dose=50.0), ExponentialKernel(reference_dose=50.0), PowerKernel(reference_dose=50.0), LinearKernel(reference_dose=50.0)):
    k: AnyKernel = fam
    names = [p.name for p in k.parameters("a", D.currency, D.outcome)]
    print(f"{k.name:12s} saturating={k.saturating!s:5s} params={names}")

FitResult(surface: Surface, posterior: Posterior | Unverified | Unsupported, data: Mapping[str, NDArray[Any]], report: ConvergenceReport | None, provenance: dict[str, Any], dropped: frozenset[Capability] = frozenset(), declared: tuple[str, ...] = ()) -> None

What ``fit`` returns: the surface, its posterior (or a typed failure), and the record.

result: FitResult = fit(world.spec, world.panel, backend="laplace", draws=2000, seed=0)

FourierSeasonality(kind: Literal['seasonality'] = 'seasonality', period: float, order: int, coefficient_scale: float = 1.0)

``sin(2πk t / period)``, ``cos(2πk t / period)`` for ``k = 1 … order``.

any_term: AnyNuisanceTerm = FourierSeasonality(period=12.0, order=1)

Frontier(budgets: tuple[float, ...], outcomes: tuple[float, ...], allocations: tuple[Allocation, ...])

Optimal outcome per budget: ``budgets`` ascend; ``allocations[i]`` reaches ``outcomes[i]``.

if isinstance(fr, Frontier):
    print(fr.as_frame())
    print("shadow prices:", np.round(fr.shadow_prices(), 4))
else:
    show(fr)

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

print("predict marks a dose grid with version", repr(GRID_VERSION))

GaussianProcessKernel(name: Literal['gaussian_process'] = 'gaussian_process', reference_dose: float = 1.0, amplitude_scale: float = 1.0, amplitude_prior: Prior | None = None, n_basis: int = 24, boundary_factor: float = 3.0, covariance: Literal['squared_exponential', 'matern32', 'matern52'] = 'squared_exponential', lengthscale_median: float = 0.3, lengthscale_spread: float = 0.5)

A Gaussian process over the dose axis, as its Hilbert-space basis approximation.

gp = GaussianProcessKernel(reference_dose=40.0, amplitude_scale=8.0)

GeometricCarryover(name: Literal['geometric'] = 'geometric', max_lag: int)

``w_l ∝ λ^l``: exponential decay from lag zero. ``λ ~ beta(2, 2)`` on (0, 1).

geo: CarryoverKernel = GeometricCarryover(max_lag=6)

HillKernel(name: Literal['hill'] = 'hill', reference_dose: float = 1.0, amplitude_scale: float = 1.0, amplitude_prior: Prior | None = None)

``f(u) = u^s / (1 + u^s)``, ``u = dose / k``. Saturating; ``f(0) = 0``, ``f(1) = 1/2``.

hill: ResponseKernel = HillKernel(reference_dose=50.0, amplitude_scale=10.0)

InterceptKind(*args, **kwargs)

kind: InterceptKind = "hierarchical"

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

print(sorted(KERNELS), "|", sorted(CARRYOVERS))

KernelRole(*args, **kwargs)

role: KernelRole = hill.roles[p.name.rsplit("_", 1)[0]] if p.name.rsplit("_", 1)[0] in hill.roles else "shape"

LinearKernel(name: Literal['linear'] = 'linear', reference_dose: float = 1.0, amplitude_scale: float = 1.0, amplitude_prior: Prior | None = None)

``response = beta_rate · dose``. No scale, no shape; never saturates.

for fam in (LogisticKernel(reference_dose=50.0), ExponentialKernel(reference_dose=50.0), PowerKernel(reference_dose=50.0), LinearKernel(reference_dose=50.0)):
    k: AnyKernel = fam
    names = [p.name for p in k.parameters("a", D.currency, D.outcome)]
    print(f"{k.name:12s} saturating={k.saturating!s:5s} params={names}")

LinearTrend(kind: Literal['trend'] = 'trend', origin: float | None = None, scale: float | None = None, coefficient_scale: float = 1.0)

``trend = (t − origin) / scale``: one dimensionless basis column.

term: NuisanceTerm = LinearTrend()

LogisticKernel(name: Literal['logistic'] = 'logistic', reference_dose: float = 1.0, amplitude_scale: float = 1.0, amplitude_prior: Prior | None = None)

``f(u) = 2 / (1 + exp(-u)) − 1 = tanh(u / 2)``, ``u = dose / k``. Saturating; ``f(0) = 0``.

for fam in (LogisticKernel(reference_dose=50.0), ExponentialKernel(reference_dose=50.0), PowerKernel(reference_dose=50.0), LinearKernel(reference_dose=50.0)):
    k: AnyKernel = fam
    names = [p.name for p in k.parameters("a", D.currency, D.outcome)]
    print(f"{k.name:12s} saturating={k.saturating!s:5s} params={names}")

MarginalHorizon(*args, **kwargs)

horizon: MarginalHorizon = "shift"

Method(*args, **kwargs)

method: Method = "auto"

NoCarryover(name: Literal['none'] = 'none')

The identity: all weight at lag zero. ``apply`` returns the dose expression itself.

for c in (DelayedCarryover(max_lag=6), WeibullCarryover(max_lag=6), NoCarryover()):
    cc: AnyCarryover = c
    print(cc.name, [p.name for p in cc.parameters("a")])

NuisanceSet(kind: Literal['set'] = 'set', terms: tuple[FourierSeasonality | LinearTrend | EventIndicators, ...])

Several nuisance terms with one ``augment`` / ``expr`` / ``parameters``.

trend_spec = spec.model_copy(update={"nuisance": NuisanceSet(terms=(LinearTrend(),)), "carryover": {}})

NuisanceTerm(*args, **kwargs)

A baseline term: basis columns in the frame, coefficients in the tree.

term: NuisanceTerm = LinearTrend()

Objective(*args, **kwargs)

obj: Objective = "mean"

ParameterRole(*args, **kwargs)

roles: dict[str, ParameterRole] = parameter_roles(spec)

PiecewiseLinearKernel(name: Literal['piecewise_linear'] = 'piecewise_linear', reference_dose: float = 1.0, amplitude_scale: float = 1.0, amplitude_prior: Prior | None = None, knots: tuple[float, ...] = (0.5,))

A **linear spline** in ``u = dose / reference_dose``: a broken stick on fixed knots.

broken = PiecewiseLinearKernel(reference_dose=40.0, knots=(15.0, 25.0))

PolynomialKernel(name: Literal['polynomial'] = 'polynomial', reference_dose: float = 1.0, amplitude_scale: float = 1.0, amplitude_prior: Prior | None = None, degree: int = 3)

``response = Σ_{j=1..degree} beta_j · u^j``, ``u = dose / reference_dose``.

poly = PolynomialKernel(reference_dose=40.0, degree=3)

PowerKernel(name: Literal['power'] = 'power', reference_dose: float = 1.0, amplitude_scale: float = 1.0, amplitude_prior: Prior | None = None)

``f(u) = u^s``, ``u = dose / k``. **Not saturating**: unbounded as the dose grows.

for fam in (LogisticKernel(reference_dose=50.0), ExponentialKernel(reference_dose=50.0), PowerKernel(reference_dose=50.0), LinearKernel(reference_dose=50.0)):
    k: AnyKernel = fam
    names = [p.name for p in k.parameters("a", D.currency, D.outcome)]
    print(f"{k.name:12s} saturating={k.saturating!s:5s} params={names}")

ResponseBand(treatment: str, outcome: str, kind: Literal['response', 'marginal'], doses: tuple[float, ...], mean: tuple[float, ...], median: tuple[float, ...], lower: tuple[float, ...], upper: tuple[float, ...], definition: Literal['eti', 'hdi', 'wald'], mass: float, n_draws: int, dimension: Dimension, ... and 2 more fields)

A dose grid, the curve on it, and the interval around the curve at every point.

assert isinstance(band, ResponseBand)

ResponseKernel(*args, **kwargs)

A dose-response family that builds its own expression tree.

hill: ResponseKernel = HillKernel(reference_dose=50.0, amplitude_scale=10.0)

SplineKernel(name: Literal['spline'] = 'spline', reference_dose: float = 1.0, amplitude_scale: float = 1.0, amplitude_prior: Prior | None = None, knots: tuple[float, ...] = (0.25, 0.5, 0.75))

A **natural cubic spline** in ``u = dose / reference_dose``, on fixed knots.

spline = SplineKernel(reference_dose=40.0, knots=(8.0, 16.0, 24.0, 32.0))

StationaryPoint(treatments: tuple[str, ...], origin: tuple[float, ...], point: tuple[float, ...], value: float, kind: Literal['maximum', 'minimum', 'saddle'], gradient: tuple[float, ...], eigenvalues: tuple[float, ...], eigenvectors: tuple[tuple[float, ...], ...])

Canonical analysis at ``origin``: the stationary point of the local quadratic and its shape.

No executed statement uses this symbol — it appears in the notebooks only inside an import, which is all gate 12 requires. Worth a worked line.

SupportsBands(*args, **kwargs)

What a band needs from a fit: draws under an intervention, and the fitted grid.

print("a fit satisfies the band protocol:", isinstance(result, SupportsBands))

Surface(spec: SurfaceSpec, *, provenance: Mapping[str, Any] | None = None) -> None

A ``SurfaceSpec`` with its built model: the ``SupportsForward`` the upper layers use.

surface = Surface(spec)

SurfaceSpec(name: str, treatments: tuple[Treatment, ...], outcome: Outcome, kernels: dict[str, HillKernel | LogisticKernel | ExponentialKernel | PowerKernel | LinearKernel | PolynomialKernel | SplineKernel | PiecewiseLinearKernel | GaussianProcessKernel] = {}, carryover: dict[str, GeometricCarryover | DelayedCarryover | WeibullCarryover | NoCarryover] = {}, nuisance: NuisanceSet = NuisanceSet(kind='set', terms=()), intercept: Literal['none', 'shared', 'per_unit', 'hierarchical'] = 'shared', interactions: tuple[tuple[str, str], ...] = (), likelihood: Likelihood = ..., time_column: str = 't', unit_column: str = 'unit', unit_labels: tuple[str, ...] = (), ... and 3 more fields)

A response surface over named treatments: what is fit, declared once.

spec = SurfaceSpec(
    name="two_treatment_panel",
    treatments=(Treatment(name="a", dimension=D.currency, unit="USD"), Treatment(name="b", dimension=D.currency, unit="USD")),
    outcome=Outcome(name="y", dimension=D.outcome, unit="units"),
    kernels={"a": HillKernel(reference_dose=50.0), "b": HillKernel(reference_dose=20.0)},
    carryover={"a": GeometricCarryover(max_lag=4)},
    nuisance=NuisanceSet(terms=(FourierSeasonality(period=12.0, order=1),)),
    intercept=kind,
    interactions=(("a", "b"),),
    likelihood=Likelihood(family="normal", scale="sigma"),
    unit_labels=("u0", "u1", "u2"),
)

WeibullCarryover(name: Literal['weibull'] = 'weibull', max_lag: int)

``w_l ∝ exp(−(l / λ)^κ)``: the Weibull survival function over lags.

for c in (DelayedCarryover(max_lag=6), WeibullCarryover(max_lag=6), NoCarryover()):
    cc: AnyCarryover = c
    print(cc.name, [p.name for p in cc.parameters("a")])

a_criterion(X: DesignMatrix | ArrayLike) -> float

``-trace((X'X)^-1)`` (larger is better); ``-inf`` when singular.

print("D:", round(d_criterion(X), 3), "A:", round(a_criterion(X), 3), "E:", round(e_criterion(X), 4))

allocate(surface: SupportsForward, posterior_or_theta: SupportsPosterior | Theta | Failure, *, budget: float, bounds: Bounds, objective: Objective = 'mean', q: float = 0.5, seed: int | None = None, method: AllocationMethod = 'slsqp', start: Mapping[str, float] | None = None, n_starts: int = 1, n_draws: int | None = None, maxiter: int = 200, tol: float = 1e-12, n_grid: int = 101, context: Context | None = None) -> Allocation | Failure

Maximize the posterior-expected outcome subject to ``Σ dose ≤ budget`` and ``bounds``.

alloc = allocate(surface, theta, budget=60.0, bounds=bounds, objective=obj, method=meth, seed=0)

bayesian_criterion(surface: SupportsForward, design: Design, theta_draws: Sequence[Mapping[str, ArrayLike]], criterion: Criterion = 'd') -> float

Mean over ``theta_draws`` of the criterion on ``surface.linearize(design, theta).X``.

print("naive   :", round(bayesian_criterion(surface, naive, prior_draws), 3))

box_behnken(bounds: Bounds, *, center_points: int = 3) -> Design

The published Box-Behnken design for ``k`` in 3..7: blocked two-level factorials.

print("box-behnken k=3:", box_behnken(b3, center_points=2).n)

build(spec: SurfaceSpec) -> ModelSpec

The ``ModelSpec`` a backend fits: ``mean = intercept + Σ_i beta_i f_i(carry_i(dose_i)) + Σ gamma_ij f_i f_j + nuisance``, every parameter with its prior, dimension-checked.

model = build(spec)

canonical_analysis(surface: SupportsForward, theta: Theta, x0: Mapping[str, float], *, method: Method = 'auto', tol: float = 1e-08, bounds: Bounds | None = None) -> StationaryPoint | Unsupported

Second-order expansion at ``x0``: stationary point ``x0 - H^-1 g`` and its nature.

sp = canonical_analysis(surface, theta, {"a": 30.0, "b": 10.0}, method=method)

carryover_from_name(name: str, **fields: Any) -> CarryoverKernel

Build a carryover by family name, e.g. ``carryover_from_name("geometric", max_lag=8)``.

print(carryover_from_name("geometric", max_lag=4))

central_composite(bounds: Bounds, *, alpha: float | Literal['rotatable', 'face'] = 'rotatable', center_points: int = 4, fraction: int | None = None, inscribed: bool = False) -> Design

Cube (2^k, or 2^(k-p) with ``fraction=p``), 2k axial points at ``±alpha``, centre runs.

ccd: Design = central_composite(bounds, alpha="rotatable", center_points=3)

check_linearization(surface: SupportsForward, dose: Mapping[str, ArrayLike], theta: Mapping[str, ArrayLike]) -> float

``‖X @ theta_lin + offset − forward(dose, theta)‖∞`` — the gate's number, for one point.

print("invariant:", check_linearization(ws, world.data, world.theta))

column_names(model: ModelSpec, linear: Sequence[str]) -> tuple[str, ...]

One column per scalar entry of every linear parameter, in ``linear`` order.

print(term.column_names("n0_"), any_term.column_names("n1_"))

counterfactual_doses(surface: Surface, data: Mapping[str, ArrayLike], iv: Intervention) -> dict[str, NDArray[Any]]

The fitted data dict with ``iv`` applied to the treatments it names.

cf = counterfactual_doses(surface, world.data, Intervention(doses={"a": 2.0}, mode="scale", window=TimeWindow(start=5, stop=10)))

d_criterion(X: DesignMatrix | ArrayLike) -> float

``log det(X'X)``; ``-inf`` when the information matrix is singular.

print("D:", round(d_criterion(X), 3), "A:", round(a_criterion(X), 3), "E:", round(e_criterion(X), 4))

defining_relation(generators: Sequence[str]) -> tuple[str, ...]

The words of ``I = ...`` implied by two-level generators, e.g. ``("ABCD",)`` for ``D=ABC``.

print("2^(4-1):", ff.n, ff.detail, defining_relation(("D=ABC",)))

design_matrix(model: ModelSpec, linear: Sequence[str], dose: Mapping[str, ArrayLike], theta_at: Mapping[str, ArrayLike]) -> DesignMatrix

``mu = offset + X @ theta[columns]`` for ``model.mean`` at ``theta_at``.

print(design_matrix(ws.model, ws.linear, world.data, world.theta).at.keys())

e_criterion(X: DesignMatrix | ArrayLike) -> float

Smallest eigenvalue of ``X'X``; ``0`` when singular (a zero eigenvalue, not a guess).

print("D:", round(d_criterion(X), 3), "A:", round(a_criterion(X), 3), "E:", round(e_criterion(X), 4))

equal_spacing(bounds: Bounds, n_per_treatment: int) -> Design

The naive comparator: a regular grid with ``n_per_treatment`` levels on every axis.

print("LHS:", lhs.n, "| equal spacing:", equal_spacing(bounds, 4).n)

fit(spec: SurfaceSpec, panel: Panel, *, backend: Backend | str = 'laplace', draws: int = 1000, tune: int = 1000, chains: int = 4, seed: int | None = None) -> FitResult

Prepare the panel, sample the built model through ``backend``, and record everything.

result: FitResult = fit(world.spec, world.panel, backend="laplace", draws=2000, seed=0)

forward(surface: SupportsForward, dose: Mapping[str, ArrayLike], theta: Mapping[str, ArrayLike]) -> Array

``value(surface.expr, data=dose, params=theta)`` — the one forward().

mu = forward(ws, world.data, world.theta)

fractional_factorial(bounds: Bounds, generators: tuple[str, ...]) -> Design

A two-level 2^(k-p) fraction from generators like ``("D=ABC",)``; A is the first treatment.

ff = fractional_factorial(Bounds(treatments=("A", "B", "C", "D"), low=(0.0,) * 4, high=(1.0,) * 4), generators=("D=ABC",))

frontier(surface: SupportsForward, posterior_or_theta: SupportsPosterior | Theta | Failure, budgets: Sequence[float], bounds: Bounds, *, objective: Objective = 'mean', q: float = 0.5, seed: int | None = None, method: AllocationMethod = 'slsqp', n_starts: int = 1, n_draws: int | None = None, maxiter: int = 200, tol: float = 1e-12, n_grid: int = 101, context: Context | None = None) -> Frontier | Failure

``allocate`` at every budget (sorted ascending), each warm-started from the last.

fr = frontier(surface, theta, budgets=[10.0, 30.0, 60.0, 120.0], bounds=bounds, seed=0)

full_factorial(bounds: Bounds, levels: int | tuple[int, ...]) -> Design

Every combination of equally spaced levels (``levels`` per treatment, or one count).

print("3^2 factorial:", full_factorial(bounds, 3).n)

gradient(surface: SupportsForward, theta: Theta, x: Mapping[str, float], *, method: Method = 'auto', bounds: Bounds | None = None) -> dict[str, float] | Unsupported

``d forward / d dose`` at ``x``, keyed like ``x``; the key order fixes treatment order.

print(gradient(surface, theta, {"a": 30.0, "b": 10.0}, method=method))

hessian(surface: SupportsForward, theta: Theta, x: Mapping[str, float], *, method: Method = 'auto', bounds: Bounds | None = None) -> Array | Unsupported

Second derivatives of ``forward`` at ``x``; rows and columns follow ``x``'s key order.

print(np.round(hessian(surface, theta, {"a": 30.0, "b": 10.0}, method=method), 5))

interaction_name(first: str, second: str) -> str

The parameter name of the pairwise interaction ``gamma`` between two treatments.

print(interaction_name("a", "b"), "|", dimension(model.mean))

kernel_from_name(name: str, **fields: Any) -> ResponseKernel

Build a kernel by family name, e.g. ``kernel_from_name("hill", reference_dose=50.0)``.

print(kernel_from_name("hill", reference_dose=10.0))

latin_hypercube(bounds: Bounds, n: int, *, seed: int | None = None, centered: bool = False, n_restarts: int = 10) -> Design

``n`` points with one per stratum per treatment; maximin over ``n_restarts`` draws.

lhs = latin_hypercube(bounds, 12, seed=0)

linear_coefficients(columns: Sequence[str], theta: Mapping[str, ArrayLike]) -> Array

The vector ``theta[columns]``: scalars by name, vector entries by ``name[i]``.

print(column_names(ws.model, ws.linear)[:4], linear_coefficients(dm.columns, world.theta)[:4].round(3))

marginal(surface: Surface, theta: Mapping[str, ArrayLike], dose: Mapping[str, ArrayLike], treatment: str) -> Array

``w_0 · ∂mu_t/∂c_t``: the effect of one more unit of dose in period ``t`` on outcome ``t``.

print(marginal(surface, world.theta, world.data, "a")[0, :4].round(4), marginal_total(surface, world.theta, world.data, "a")[0, :4].round(4))

marginal_band(result: SupportsBands, treatment: str, *, doses: Sequence[float] | None = None, n_grid: int = 25, mass: float = 0.9, definition: IntervalDefinition = 'eti', window: TimeWindow | None = None, seed: int | None = None) -> ResponseBand | Unsupported

``d outcome / d dose`` against dose, with the posterior interval at every point.

slope = marginal_band(result, "a", n_grid=9, mass=0.9)

marginal_expr(surface: Surface, treatment: str) -> Expr

``∂mu / ∂c`` with respect to a treatment's *carried* dose, as a tree.

print(dimension(marginal_expr(surface, "a")))

marginal_total(surface: Surface, theta: Mapping[str, ArrayLike], dose: Mapping[str, ArrayLike], treatment: str) -> Array

``Σ_l w_l · ∂mu_{t+l}/∂c_{t+l}``: the total effect of one more unit in period ``t``.

print(marginal(surface, world.theta, world.data, "a")[0, :4].round(4), marginal_total(surface, world.theta, world.data, "a")[0, :4].round(4))

optimal_exchange(surface: SupportsForward, candidates: Design, n: int, theta_draws: Sequence[Mapping[str, ArrayLike]], *, criterion: Criterion = 'd', seed: int | None = None, max_passes: int = 20, n_restarts: int = 3) -> Design | Unsupported

Fedorov point exchange: the best ``n`` candidate rows (replicates allowed) by the criterion.

opt = optimal_exchange(surface, candidates, n=naive.n, theta_draws=prior_draws, criterion=crit, seed=0)

parameter_roles(spec: SurfaceSpec) -> dict[str, ParameterRole]

``{parameter name: "linear" | "nonlinear" | "auxiliary"}`` in declaration order.

roles: dict[str, ParameterRole] = parameter_roles(spec)

predict(surface: Surface, posterior: SupportsPosterior, dose: Mapping[str, ArrayLike], *, seed: int | None = None, noise: bool = False) -> PredictiveDraws

Every draw pushed through ``forward``: ``values`` is ``(chain, draw, *forward shape)``.

print(np.allclose(dm.predict(world.theta).reshape(mu.shape), mu))

predict_marginal(surface: Surface, posterior: SupportsPosterior, dose: Mapping[str, ArrayLike], treatment: str, *, horizon: MarginalHorizon = 'total', support: TimeWindow | None = None, seed: int | None = None) -> PredictiveDraws

Every draw pushed through ``marginal`` / ``marginal_total`` / ``marginal_shift``.

pm = predict_marginal(surface, post, world.data, "a", horizon=horizon, support=TimeWindow(start=5, stop=10))

prepare(spec: SurfaceSpec, panel: Panel, *, conventions: Conventions | None = None) -> DataDict

The data dict the built model reads, laid out ``(n_units, n_periods)``.

data = prepare(world.spec, world.panel)

resolve_conventions(spec: SurfaceSpec, panel: Panel, conventions: Conventions | None = None) -> dict[str, dict[str, float]]

The trend conventions ``prepare`` lays the panel out with, one entry per trend column.

conv = resolve_conventions(world_t.spec, world_t.panel)

response_band(result: SupportsBands, treatment: str, *, doses: Sequence[float] | None = None, n_grid: int = 25, mass: float = 0.9, definition: IntervalDefinition = 'eti', window: TimeWindow | None = None, seed: int | None = None) -> ResponseBand | Unsupported

Expected outcome against dose, with the posterior interval at every grid point.

band = response_band(result, "a", n_grid=9, mass=0.9, definition="eti")

steepest_ascent(surface: SupportsForward, theta: Theta, start: Mapping[str, float], *, step: float, n_steps: int, bounds: Bounds | None = None, method: Method = 'auto', recompute: bool = False) -> AscentPath | Unsupported

Walk ``step`` dose units per move up the gradient until the value stops rising.

path = steepest_ascent(surface, theta, {"a": 10.0, "b": 5.0}, step=5.0, n_steps=20, bounds=bounds)

DoseDistribution(*args, **kwargs)

dist: DoseDistribution = "lognormal"

DosePlan(distribution: Literal['lognormal', 'uniform'] = 'lognormal', scale: float = 1.0, spread: float = 0.5, zero_fraction: float = 0.0)

How doses are drawn for one treatment: a family, a scale, a spread, and a zero share.

plan = DosePlan(distribution=dist, scale=50.0, spread=0.6, zero_fraction=0.1)

Horizon(*args, **kwargs)

horizon: Horizon = "total"

LinearSCM(graph: CausalGraph, coefficients: dict[str, float] = {}, noise_sd: dict[str, float] = {}, latent_sd: dict[str, float] = {}, intercepts: dict[str, float] = {})

A linear-Gaussian SCM over a ``CausalGraph`` with optional latent confounding.

scm = LinearSCM.from_text("Z -> X: 0.8, Z -> Y: 1.5, X -> Y: 2.0, X <-> Y: 0.7", name="demo")

The structural model is malformed (missing coefficient, unknown node, bad key).

try:
    LinearSCM.from_text("X -> Y: 1.0, X -> Y: 2.0")
except SCMError as e:
    print("refused:", e)

SurfaceWorld(spec: SurfaceSpec, model: ModelSpec, theta: dict[str, Array], panel: Panel, data: dict[str, Array], mean: Array, seed: int | None, noise_sd: float | None, provenance: dict[str, Any] = <factory>) -> None

A ``SurfaceSpec``, its true parameters, and a balanced panel generated from them.

world: SurfaceWorld = surface_world(
    n_units=4, n_periods=26, treatments=("a", "b"),
    kernels={"a": HillKernel(reference_dose=50.0), "b": HillKernel(reference_dose=20.0)},
    carryover={"a": GeometricCarryover(max_lag=4)},
    doses={"a": plan, "b": DosePlan(scale=20.0, spread=0.4)},
    nuisance=NuisanceSet(terms=(FourierSeasonality(period=13.0, order=1),)),
    intercept="hierarchical", truth_mode=mode, noise_sd=0.5, seed=3,
)

TruthMode(*args, **kwargs)

mode: TruthMode = "centre"

arms_world(*, n_units: int, treatments: Sequence[Treatment | str] = ('dose',), kernels: Mapping[str, Kernel] | Kernel | None = None, doses: Mapping[str, DosePlan] | DosePlan | Mapping[str, ArrayLike] | None = None, outcome: Outcome | None = None, truth: Mapping[str, ArrayLike] | None = None, truth_mode: TruthMode = 'centre', noise_sd: float | None = None, intercept_scale: float = 1.0, seed: int | None = None, name: str = 'arms_world') -> SurfaceWorld

The single-period world: one row per unit, a shared intercept, no carryover.

arms = arms_world(n_units=12, treatments=("dose",), doses={"dose": np.linspace(0, 100, 12)}, seed=0)

coefficient_key(a: str, b: str) -> str

The ``coefficients`` key for the directed edge ``a -> b``: ``"a->b"``.

print(coefficient_key("Z", "X"), latent_key("Y", "X"))

confounded_world() -> LinearSCM

``Z -> X`` (0.8), ``Z -> Y`` (1.5), ``X -> Y`` (2.0); everything measured.

for w in (confounded_world, hidden_confounder_world, iv_world, frontdoor_world, mediator_world, feedback_world):
    s = w()
    print(f"{w.__name__:24s} {s.graph.to_text():40s} truth={s.total_effect('X', 'Y'):.2f} unmeasured={s.graph.unmeasured} feedback={s.graph.feedback}")

draw_doses(plan: DosePlan, n_units: int, n_periods: int, *, seed: int | None = None) -> Array

``(n_units, n_periods)`` doses from ``plan``; deterministic given ``seed``.

print(draw_doses(plan, 2, 6, seed=0).round(1))

feedback_world() -> LinearSCM

The confounded world's summary graph, flagged ``feedback=True``.

for w in (confounded_world, hidden_confounder_world, iv_world, frontdoor_world, mediator_world, feedback_world):
    s = w()
    print(f"{w.__name__:24s} {s.graph.to_text():40s} truth={s.total_effect('X', 'Y'):.2f} unmeasured={s.graph.unmeasured} feedback={s.graph.feedback}")

frontdoor_world() -> LinearSCM

``X -> M`` (1.2), ``M -> Y`` (1.5), ``X <-> Y`` (latent scale 1.0).

for w in (confounded_world, hidden_confounder_world, iv_world, frontdoor_world, mediator_world, feedback_world):
    s = w()
    print(f"{w.__name__:24s} {s.graph.to_text():40s} truth={s.total_effect('X', 'Y'):.2f} unmeasured={s.graph.unmeasured} feedback={s.graph.feedback}")

hidden_confounder_world() -> LinearSCM

The confounded world with ``Z`` unmeasured.

print(scm.observed(frame).columns.tolist(), "| unmeasured dropped:", hidden_confounder_world().observed(hidden_confounder_world().simulate(3, seed=0)).columns.tolist())

iv_world() -> LinearSCM

``Z -> X`` (1.0), ``X -> Y`` (2.0), ``X <-> Y`` (latent scale 1.0).

for w in (confounded_world, hidden_confounder_world, iv_world, frontdoor_world, mediator_world, feedback_world):
    s = w()
    print(f"{w.__name__:24s} {s.graph.to_text():40s} truth={s.total_effect('X', 'Y'):.2f} unmeasured={s.graph.unmeasured} feedback={s.graph.feedback}")

latent_key(a: str, b: str) -> str

The ``latent_sd`` key for the bidirected edge ``a <-> b``: ``"a<->b"`` with ``a < b``.

print(coefficient_key("Z", "X"), latent_key("Y", "X"))

long_frame(columns: Mapping[str, ArrayLike], *, units: Sequence[str], periods: Sequence[int] | None = None, unit_column: str = 'unit', time_column: str = 't') -> DataFrame

Stack ``(n_units, n_periods)`` arrays into a unit-major, period-minor long table.

frame = long_frame({"x": [[1.0, 2.0], [3.0, 4.0]], "y": [[0.5, 0.6], [0.7, 0.8]]}, units=("u0", "u1"))

mediator_world() -> LinearSCM

``X -> M`` (1.0), ``M -> Y`` (0.5), ``X -> Y`` (1.0); everything measured, no confounding.

for w in (confounded_world, hidden_confounder_world, iv_world, frontdoor_world, mediator_world, feedback_world):
    s = w()
    print(f"{w.__name__:24s} {s.graph.to_text():40s} truth={s.total_effect('X', 'Y'):.2f} unmeasured={s.graph.unmeasured} feedback={s.graph.feedback}")

panel_from_arrays(*, doses: Mapping[str, ArrayLike], outcome: ArrayLike, treatments: Sequence[Treatment], outcome_entity: Outcome, covariates: Mapping[str, tuple[Covariate, ArrayLike]] | None = None, units: Sequence[str], periods: Sequence[int] | None = None, unit_column: str = 'unit', time_column: str = 't') -> Panel

A balanced ``Panel`` from wide arrays plus the entities that give the columns roles.

panel = panel_from_arrays(
    doses={"x": [[1.0, 2.0], [3.0, 4.0]]}, outcome=[[0.5, 0.6], [0.7, 0.8]],
    treatments=[Treatment(name="x", dimension=D.currency, unit="USD")],
    outcome_entity=Outcome(name="y", dimension=D.outcome), units=("u0", "u1"),
)

surface_world(*, n_units: int, n_periods: int, treatments: Sequence[Treatment | str] = ('a', 'b'), kernels: Mapping[str, Kernel] | Kernel | None = None, carryover: Mapping[str, Carryover] | Carryover | None = None, doses: Mapping[str, DosePlan] | DosePlan | None = None, nuisance: NuisanceSet | None = None, intercept: InterceptKind = 'hierarchical', interactions: Sequence[tuple[str, str]] = (), outcome: Outcome | None = None, likelihood: Likelihood | None = None, truth: Mapping[str, ArrayLike] | None = None, truth_mode: TruthMode = 'centre', noise_sd: float | None = None, intercept_scale: float = 1.0, seed: int | None = None, name: str = 'surface_world') -> SurfaceWorld

A units × periods panel world with saturation, carryover, intercepts, and nuisance.

world: SurfaceWorld = surface_world(
    n_units=4, n_periods=26, treatments=("a", "b"),
    kernels={"a": HillKernel(reference_dose=50.0), "b": HillKernel(reference_dose=20.0)},
    carryover={"a": GeometricCarryover(max_lag=4)},
    doses={"a": plan, "b": DosePlan(scale=20.0, spread=0.4)},
    nuisance=NuisanceSet(terms=(FourierSeasonality(period=13.0, order=1),)),
    intercept="hierarchical", truth_mode=mode, noise_sd=0.5, seed=3,
)

transport_pair() -> tuple[LinearSCM, LinearSCM]

A source and a target population that differ only in ``Z``'s distribution.

src, tgt = transport_pair()

true_parameters(model: ModelSpec, *, structural: Sequence[str] = (), truth: Mapping[str, ArrayLike] | None = None, mode: TruthMode = 'centre', seed: int | None = None) -> dict[str, Array]

A full true-parameter dictionary for ``model`` (see the module docstring for the rules).

theta_prior = true_parameters(world.model, structural=world.structural, mode="prior", seed=11)

unit_labels(n_units: int, prefix: str = 'u') -> tuple[str, ...]

``("u00", "u01", …)`` — zero-padded so that string order is numeric order.

print(unit_labels(3))

Read-only proxy of a mapping.

print("assumption 'parallel_trends':", ASSUMPTIONS["parallel_trends"].statement)

AnchoredEffect(parameter: str, posterior_mean: float, posterior_sd: float, posterior_interval: Interval, n_draws: int, mde: float, probability_exceeds_mde: float, probability_exceeds_mde_gaussian: float, credence: float, anchored_effect: float, already_believed: bool)

A posterior's view of an MDE.

assert isinstance(ae, AnchoredEffect)

Assignment(unit: Unit, labels: tuple[str, ...], treated: tuple[int, ...], control: tuple[int, ...], pairs: tuple[tuple[int, int], ...], unpaired: tuple[int, ...], metric: Literal['level', 'trajectory'], seed: int | None, pre_mean_treated: float, pre_mean_control: float, pre_smd: float, mean_pair_distance: float)

A matched-pair treatment assignment of clusters.

asg: Assignment = match_clusters(pre, labels=[f"r{i}" for i in range(9)], unit=region, metric=metric, seed=1)

Boundary(kind: Literal['efficacy', 'harm', 'futility'], side: Literal['upper', 'lower', 'two_sided'], z: tuple[float, ...], binding: bool = True, spent: tuple[float, ...] = (), detail: dict[str, str] = {})

Signed thresholds on the monitoring statistic, one per look.

futility = Boundary(kind="futility", side="lower", z=(-1.5, -0.8, -0.2, 0.3, 0.8), binding=False)

BoundaryKind(*args, **kwargs)

kind: BoundaryKind = with_margin.kind

A named, falsifiable condition that licenses a step.

print("named assumptions:", CANONICAL.name, "|", STOPPED_ESTIMATE_BIAS.name)

CalibrationResult(method: str, design: Literal['holdout', 'switchback', 'ghost'], n_simulations: int, n_unsupported: int, false_positive_count: int, false_positive_rate: float, alpha: float, region: AcceptanceRegion, passed: bool, reason: str = '', seed: int)

A method's A/A false-positive count against its exact binomial acceptance region.

cal: CalibrationResult = calibrate_method("difference_in_differences", spec, alpha=0.05)

CandidateScore(candidate: DesignCandidate, eig: float, evpi: float, evsi: float, opportunity_cost: float, cost: float, net_value: float, power: float, effect: float, alpha: float, numeraire: str = '', detail: dict[str, str] = {})

A candidate's information, value, cost, and power on one decision.

scores: list[CandidateScore] = [evaluate_candidate(c, decision, prior_mean=mean, prior_sd=sd, economics=economics) for c in designs]

ClusterDesign(unit: Unit, n_clusters: int, cluster_size: int, icc: float, allocation: float = 0.5)

A cluster-randomized design and its variance inflation.

cd = ClusterDesign(unit=region, n_clusters=24, cluster_size=40, icc=0.05, allocation=0.5)

Combination(exponents: dict[str, int], kind: Literal['symmetry', 'estimable'], scaling: Literal['log', 'absolute'] = 'log', score: float, label: str = '')

A monomial (or linear combination) of parameters, and how flat it is.

ratio = Combination(exponents={"alpha": 1, "k": -1}, kind="estimable", score=0.0)

CostPerOutcomeInterval(cost: float, effect: float, effect_se: float, alpha: float, estimate: float, lower: float, upper: float | None, method: Literal['fieller'] = 'fieller', status: Literal['bounded', 'unbounded'], effect_interval: Interval, naive_se: float, naive_interval: Interval)

``cost / effect`` with a Fieller interval and the naive delta-method interval beside it.

ci: CostPerOutcomeInterval = cost_per_outcome_interval(cost=1000.0, effect=25.0, effect_se=6.0)

CostPerOutcomePower(cost: float, true_effect: float, effect_se: float, alpha: float, threshold: float | None, effect_required: float, power: float)

Power to bound the cost per outcome unit (below ``threshold`` if one is set).

cp: CostPerOutcomePower = cost_per_outcome_power(1000.0, true_effect=25.0, effect_se=6.0, threshold=60.0)

CrossingProbabilities(drift: float, information: tuple[float, ...], per_look: dict[str, tuple[float, ...]], continue_probability: float, binding_only: bool)

Per-look probability that each boundary is the one that stops the study.

assert isinstance(spent, CrossingProbabilities)

Read-only proxy of a mapping.

print("relative SE by design kind:", dict(DESIGN_RELATIVE_SE))

Decision(*args, **kwargs)

decision: Decision = path.decision

DecisionSpec(name: str = 'act', threshold: float = 0.0, value_per_outcome_unit: float = 1.0, numeraire: str = '')

A two-action decision with value linear in the parameter.

decision = DecisionSpec(name="scale_up", threshold=0.0, value_per_outcome_unit=1000.0, numeraire="USD")

DerivativeMethod(*args, **kwargs)

method: DerivativeMethod = "finite"

DesignCandidate(name: str, method: str, n_units: int, n_periods: int, holdout_fraction: float, experiment_se: float, cost: float, n_clusters: int | None = None, cooldown_periods: int = 0, detail: dict[str, str] = {})

One concrete experiment design: method, size, horizon, holdout, precision, cost.

designs = [
    DesignCandidate(name="small_holdout", method="difference_in_differences", n_units=20, n_periods=6, holdout_fraction=0.2, experiment_se=0.35, cost=200.0, cooldown_periods=2),
    DesignCandidate(name="large_holdout", method="cluster_based_regression", n_units=60, n_periods=8, holdout_fraction=0.4, experiment_se=0.12, cost=600.0, n_clusters=12, cooldown_periods=4),
    DesignCandidate(name="switchback", method="switchback", n_units=20, n_periods=10, holdout_fraction=0.5, experiment_se=0.2, cost=300.0, cooldown_periods=1),
    DesignCandidate(name="ghost", method="ghost", n_units=40, n_periods=6, holdout_fraction=0.3, experiment_se=0.3, cost=450.0),
]

EIGEstimate(eig: float, se: float, n_sims: int, seed: int | None = None, method: Literal['gaussian', 'nested_monte_carlo'], n_prior_draws: int = 0, experiment_se: float, unit: Literal['nats'] = 'nats')

Expected information gain in nats, with the error of its estimator.

est: EIGEstimate = eig_monte_carlo(prior_draws, 0.3, n_sims=n_sims, seed=1)

EVOIResult(evpi: float, evsi: float, preposterior_sd: float, decision_threshold: float, n_sims: int, seed: int | None = None, method: Literal['gaussian', 'monte_carlo'], evsi_se: float = 0.0, experiment_se: float, prior_mean: float, prior_sd: float, value_per_outcome_unit: float, ... and 1 more field)

EVPI and EVSI of one experiment against one decision, in the numeraire.

res: EVOIResult = evoi_gaussian(decision, mean, sd, se)

EconomicInputs(value_per_outcome: ValuePerOutcome, dose_per_period: float, discount_rate: float, dose_unit: str = 'dose', dose_cost_per_unit: float = 0.0, marginal_value_ratio: float | None = None)

The economic inputs a candidate's opportunity cost needs, stated once.

economics = EconomicInputs(value_per_outcome=vpo, dose_per_period=20.0, discount_rate=0.02, dose_unit="L", dose_cost_per_unit=1.0, marginal_value_ratio=1.2)

EstimabilityReport(parameters: tuple[str, ...], observations: tuple[str, ...] = (), scaling: Literal['log', 'absolute'] = 'log', n_points: int = 1, rank: int, deficiency: int, persistent_deficiency: int, singular_values: tuple[float, ...] = (), condition_number: float, symmetries: tuple[Combination, ...] = (), estimable: tuple[Combination, ...] = (), null_basis: tuple[tuple[float, ...], ...] = (), ... and 4 more fields)

What this design can and cannot separate, at one parameter point and across several.

report: EstimabilityReport = estimable_combinations([low], theta, tolerance=3e-2)

ExperimentValue(information_value: float, opportunity_cost: float, fixed_cost: float, net: float, numeraire: str, detail: dict[str, str] = {})

``net = information_value − opportunity_cost − fixed_cost``, all in ``numeraire``.

ev: ExperimentValue = experiment_value(info, oc, fixed_cost=250.0)

FisherInformation(parameters: tuple[str, ...], matrix: tuple[tuple[float, ...], ...], noise_sd: float, n_observations: int, det: float, min_eigenvalue: float, singular: bool, method: Literal['jax', 'finite'], detail: dict[str, str] = {})

``J'J / noise_sd²`` for the named parameters at one design and one ``theta``.

assert isinstance(fi, FisherInformation)

HoldoutTradeoff(design: ClusterDesign, fractions: tuple[float, ...], mdes: tuple[float, ...], relative_mde: tuple[float, ...], n_holdout: tuple[int, ...], best_fraction: float, alpha: float, power: float, two_sided: bool)

MDE as a function of the share of clusters held out of treatment.

ht: HoldoutTradeoff = holdout_tradeoff(cd, sd=2.0, fractions=(0.1, 0.2, 0.3, 0.4, 0.5))

IdentifiabilityRidge(parameters: tuple[str, ...], direction: dict[str, float], min_eigenvalue: float, max_eigenvalue: float, condition_number: float, pairs: tuple[tuple[str, str], ...] = (), correlations: tuple[float, ...] = (), detail: dict[str, str] = {})

The flattest direction of the information matrix and the couplings it implies.

assert isinstance(ridge, IdentifiabilityRidge)

IdentifyingDesign(target: str, design: Design, indices: tuple[int, ...], expected_sd: float, expected_sds: dict[str, float], prior_sds: dict[str, float] = {}, noise_sd: float, seed: int | None = None, n_restarts: int, passes: int, detail: dict[str, str] = {})

The ``n`` candidate rows that minimize a target parameter's expected posterior sd.

assert isinstance(out, IdentifyingDesign)

Leaderboard(spec: SimulationSpec, alpha: float, effects: tuple[float, ...], rows: tuple[LeaderboardRow, ...], calibrations: tuple[CalibrationResult, ...], powers: tuple[tuple[SimulatedPower, ...], ...])

Methods ranked by mean power at a calibrated size; uncalibrated methods rank last.

board: Leaderboard = leaderboard(small, effects=(0.8, 1.6), alpha=0.05)

LeaderboardRow(method: str, design: Literal['holdout', 'switchback', 'ghost'], status: Literal['stable', 'experimental'], calibrated: bool, false_positive_rate: float, powers: tuple[float, ...], mean_power: float, biases: tuple[float, ...], coverages: tuple[float, ...], rmses: tuple[float, ...], n_unsupported: int, rank: int)

One method's line: its calibrated size and its power, bias and coverage per effect.

assert isinstance(row, LeaderboardRow)

LearningPriority(treatment: str, eig: float, evoi: float, net_value: float, cost: float, rank: int, numeraire: str = '', detail: dict[str, str] = {})

A candidate's EIG, EVSI, and net value, with its rank (``1`` is first).

assert isinstance(p, LearningPriority)

LookOutcome(look: int, label: str, information: float, z: float, decision: Decision, crossed: str = '', effect: float | None = None, se: float | None = None) -> None

One interim look: what it saw and what the rule said about it.

assert isinstance(outcome, LookOutcome)

LookSchedule(labels: tuple[str, ...], information: tuple[float, ...])

When the study looks at itself, and how much information it has when it does.

schedule = LookSchedule(labels=tuple(f"look_{k + 1}" for k in range(LOOKS)), information=information)

MDE(design: Literal['difference_in_means', 'coefficient'], effect: float, se: float, alpha: float, power: float, two_sided: bool, n: int | None = None, sd: float | None = None, allocation: float | None = None)

The smallest effect detected with probability ``power`` at level ``alpha``.

m: MDE = mde(200, sd=2.0, power=0.8)

Read-only proxy of a mapping.

for name, spec in METHODS.items():
    print(f"{name:28s} status={spec.status:12s} pre={spec.requires_pre_period!s:5} controls={spec.requires_controls!s:5} {spec.assumption_names()}")

MatchMetric(*args, **kwargs)

metric: MatchMetric = "trajectory"

MethodEstimate(method: str, effect: float, se: float, interval: Interval, n_treated: int, n_control: int, n_pre: int, n_post: int, se_method: str, critical: Literal['normal', 'student_t'] = 'normal', detail: dict[str, float] = {})

One method's estimate: the average effect per treated unit per post period.

assert isinstance(out, MethodEstimate)

MethodName(*args, **kwargs)

name: MethodName = "switchback"

MethodSpec(name: str, assumptions: tuple[Assumption, ...], requires_pre_period: bool, requires_controls: bool, status: Literal['stable', 'experimental'] = 'stable', description: str = '')

A registered experiment method: its assumptions, requirements, and status.

ms: MethodSpec = method_spec("switchback")

MethodStatus(*args, **kwargs)

status: MethodStatus = ms.status

MonitoringPath(rule: StoppingRule, looks: tuple[LookOutcome, ...], decision: Decision, stopped_at: int | None = None) -> None

A realized sequence of looks and the decision it produced.

assert isinstance(path, MonitoringPath)

Observation(label: str, expr: Expr, data: Mapping[str, Any] = <factory>, noise_sd: float = 1.0) -> None

One measurable quantity, the data it is measured at, and its noise.

low = Observation("low dose", hill, {"dose": np.linspace(0.01, 0.3, 12)}, noise_sd=0.05)

OperatingCharacteristics(rule: str, drift: float, crossings: CrossingProbabilities, expected_information: float, expected_looks: float)

What a rule does at one drift: where it stops, and what it costs to get there.

assert isinstance(oc, OperatingCharacteristics)

OpportunityCost(value: float, numeraire: str, holdout_fraction: float, n_periods: int, discount_rate: float, dose_per_period: float, dose_unit: str, dose_withheld: float, dose_withheld_discounted: float, dose_cost_per_unit: float, ratio_mean: float, ratio_sd: float, ... and 5 more fields)

What a holdout forgoes, in the numeraire, with every factor that produced it.

oc: OpportunityCost = opportunity_cost(0.25, 8, dose_per_period=20.0, marginal_value_ratio=ratio_draws,
                                      value_per_outcome=vpo, discount_rate=0.02, dose_unit="L", dose_cost_per_unit=1.0)

PanelArrays(outcome: Array, treated: tuple[int, ...], pre: slice, post: slice, assignment: Array | None = None, exposed: Array | None = None) -> None

The arrays every method reads.

arrays: PanelArrays = simulate_panel(spec, rng, design="holdout")

PanelDesign(*args, **kwargs)

d: PanelDesign = design_for_method(m)

Parameter(*args, **kwargs)

parameter: Parameter = "value_per_outcome"

Pattern(*args, **kwargs)

p: Pattern = s.pattern

PowerCurve(design: Literal['difference_in_means', 'coefficient'], effects: tuple[float, ...], powers: tuple[float, ...], se: float, alpha: float, two_sided: bool, n: int | None = None, sd: float | None = None, allocation: float | None = None)

Power as a function of effect size at a fixed design (``n``, ``sd``, ``allocation``).

curve: PowerCurve = power_curve(200, 2.0, effects=np.linspace(0.0, 1.2, 7))

PowerResult(design: Literal['difference_in_means', 'coefficient'], power: float, effect: float, se: float, alpha: float, two_sided: bool, n: int | None = None, sd: float | None = None, allocation: float | None = None)

Power to detect ``effect`` at ``alpha`` with standard error ``se``.

pr: PowerResult = power(200, effect=0.6, sd=2.0)

Prescription(added: tuple[str, ...] = (), rank_before: int, rank_after: int, n_parameters: int, broken: tuple[tuple[str, ...], ...] = (), still_flat: tuple[str, ...] = (), considered: tuple[str, ...] = (), complete: bool = False, detail: dict[str, str] = {})

The smallest set of extra measurements that restores rank, and what each one buys.

plan: Prescription = prescribe_measurements([low], [wide], theta, tolerance=3e-2)

ProfileReport(targets: tuple[str, ...], truth: dict[str, float], recovered: dict[str, float], interval: dict[str, tuple[float, float]], level: float, threshold: float, flat: tuple[str, ...] = (), n_observations: int, seed: int | None = None, converged: bool = True, detail: dict[str, str] = {})

Profile-likelihood intervals for named targets, from one simulated experiment.

profile: ProfileReport = simulated_identifiability(
    model, {"dose": doses}, truth,
    targets=["alpha", "k"], combinations=[ratio],
    seed=3, range_factor=30.0, n_grid=11,
)

ProgramSchedule(horizon_periods: int, slots: tuple[ScheduledExperiment, ...], skipped: tuple[str, ...], total_net_value: float, numeraire: str = '', detail: dict[str, str] = {})

A non-overlapping sequence of experiments within ``horizon_periods``.

sched: ProgramSchedule = schedule_with_cooldown(scores, horizon_periods=20)

ReExperimentTiming(posterior_sd: float, half_life_periods: float, experiment_se: float, min_eig: float, sd_at_threshold: float, periods: float, eig_now: float, design_kind: str = 'unspecified')

When an experiment with ``experiment_se`` is worth ``min_eig`` nats again.

timing: ReExperimentTiming = time_to_re_experiment(post_sd, 6.0, experiment_se=0.2, min_eig=0.3, design_kind="cluster_holdout")

Recommendation(selected: tuple[str, ...], priorities: tuple[LearningPriority, ...], budget: float | None = None, total_cost: float, total_net_value: float, numeraire: str = '', detail: dict[str, str] = {})

The selected treatments and why: the ranked priorities and the budget arithmetic.

assert isinstance(rec, Recommendation)

A named, falsifiable condition that licenses a step.

print("named assumptions:", CANONICAL.name, "|", STOPPED_ESTIMATE_BIAS.name)

SampleSize(design: Literal['difference_in_means', 'coefficient'], n: int, effect: float, se: float, alpha: float, power: float, target_power: float, two_sided: bool, sd: float | None = None, allocation: float | None = None, n_treated: int | None = None, n_control: int | None = None)

The smallest ``n`` whose achieved power reaches the target for ``effect``.

assert isinstance(ss, SampleSize)

Scaling(*args, **kwargs)

scaling: Scaling = "log"

Schedule(pattern: Literal['constant', 'pulse', 'ramp', 'random_switchback', 'alternating'], n_periods: int, doses: tuple[float, ...], treatment: str = 'dose', detail: dict[str, float] = {})

One treatment's dose per period: ``doses[t]`` for ``t = 0 … n_periods − 1``.

schedules: list[Schedule] = [
    constant(T, 50.0),
    pulse(T, 80.0, 20.0, on=2, off=2),
    alternating(T, 80.0, 20.0),
    ramp(T, 0.0, 100.0),
    random_switchback(T, 80.0, 20.0, seed=0),
]

ScheduledExperiment(name: str, start: int, end: int, free_at: int, net_value: float)

One slot of a program: ``[start, end)`` is the run, ``[end, free_at)`` the cooldown.

assert isinstance(slot, ScheduledExperiment)

SensitivityMatrix(matrix: Array, parameters: tuple[str, ...], observations: tuple[str, ...], scaling: Scaling, theta: dict[str, float], zero_columns: tuple[str, ...] = ()) -> None

``d f / d log theta`` (or ``d f / d theta``) stacked over observations.

sensitivity: SensitivityMatrix = sensitivity_matrix([low], theta, scaling=scaling)

SensitivityTable(parameter: Literal['prior_sd', 'experiment_se', 'value_per_outcome', 'discount_rate', 'holdout_fraction'], mode: Literal['absolute', 'multiplier'], grid: tuple[float, ...], candidates: tuple[str, ...], net_values: tuple[tuple[float, ...], ...], winners: tuple[str, ...], tipping_points: tuple[tuple[float, float], ...], base_value: float, base_winner: str, base_net_values: tuple[float, ...], numeraire: str = '', detail: dict[str, str] = {})

Net value of every candidate at every grid value of one input, and where the winner flips.

table: SensitivityTable = perturb(designs, decision, mean, sd, economics, parameter, grid=(0.5, 1.0, 2.0, 3.0, 5.0, 8.0))

Side(*args, **kwargs)

side: Side = with_margin.side

SimulatedPower(method: str, design: Literal['holdout', 'switchback', 'ghost'], effect: float, truth: float, n_simulations: int, n_unsupported: int, rejections: int, power: float, alpha: float, mean_effect: float, bias: float, coverage: float, ... and 5 more fields)

Realized A/B power and accuracy of a method at one effect size.

sp: SimulatedPower = simulated_power("difference_in_differences", ab, predicted_power=predicted)

SimulationSpec(n_units: int = 20, n_periods: int = 16, n_pre: int = 8, n_treated: int = 10, unit_sd: float = 1.0, noise_sd: float = 1.0, period_sd: float = 0.5, rho: float = 0.5, effect: float = 0.0, n_simulations: int = 200, seed: int = 0, mass: float = 0.95, ... and 2 more fields)

One simulated panel: its shape, its variance components, its effect, and its budget.

spec = SimulationSpec(n_units=16, n_periods=20, n_pre=10, n_treated=8, effect=1.0, noise_sd=1.0, seed=7)

SpendingFunction(*args, **kwargs)

kind_: SpendingFunction = kind

StoppingRule(name: str, looks: LookSchedule, boundaries: tuple[Boundary, ...])

A look schedule and the boundaries evaluated at every look.

rule = StoppingRule(name=name, looks=schedule, boundaries=(boundary,))

StudySummary(treatment: str, estimate: float, se: float, periods_ago: float, definition: str, name: str = '')

One past study of a treatment: its estimate, standard error, and age.

history = [
    StudySummary(treatment="fertilizer", estimate=1.4, se=0.3, periods_ago=12.0, definition="wald"),
    StudySummary(treatment="fertilizer", estimate=1.0, se=0.4, periods_ago=2.0, definition="wald"),
]

TreatmentCandidate(name: str, prior_mean: float, prior_sd: float, experiment_se: float, decision: DecisionSpec, opportunity_cost: float = 0.0, fixed_cost: float = 0.0)

A treatment one could run an experiment on, with the inputs its value needs.

candidates = [
    TreatmentCandidate(name="fertilizer", prior_mean=mean, prior_sd=sd, experiment_se=0.2, decision=decision, opportunity_cost=oc.value, fixed_cost=250.0),
    TreatmentCandidate(name="irrigation", prior_mean=0.8, prior_sd=0.6, experiment_se=0.3, decision=decision, opportunity_cost=100.0, fixed_cost=400.0),
    TreatmentCandidate(name="pruning", prior_mean=0.1, prior_sd=0.1, experiment_se=0.3, decision=decision, opportunity_cost=0.0, fixed_cost=60.0),
]

ValueFn(*args, **kwargs)

capped: ValueFn = lambda theta: np.minimum(decision.payoff(theta), 200.0)

ValuePerOutcome(value: float, outcome_unit: str, numeraire: str, source: str, assumption: Assumption | None = None)

``value`` numeraire units per ``outcome_unit``, and where that number came from.

vpo = ValuePerOutcome(value=3.0, outcome_unit="kg", numeraire="USD", source="contract price, 2026 season")

alpha_spending(alpha: float, information: Sequence[float] | LookSchedule, *, family: SpendingFunction = 'obrien_fleming', side: Side = 'two_sided', kind: BoundaryKind = 'efficacy', rho: float = 1.0, n_grid: int = 601, width: float = 8.0) -> Boundary

A Lan–DeMets boundary: solve each threshold to spend exactly what is budgeted.

efficacy = alpha_spending(ALPHA / 2, information, family="obrien_fleming", side="upper")

alternating(n_periods: int, high: float, low: float, *, treatment: str = 'dose') -> Schedule

``high, low, high, low, …`` — the maximal-contrast pulse with ``on = off = 1``.

schedules: list[Schedule] = [
    constant(T, 50.0),
    pulse(T, 80.0, 20.0, on=2, off=2),
    alternating(T, 80.0, 20.0),
    ramp(T, 0.0, 100.0),
    random_switchback(T, 80.0, 20.0, seed=0),
]

anchor_draws(draws: ArrayLike, parameter: str, mde: float, *, credence: float = 0.9, definition: IntervalDefinition = 'hdi', mass: float = 0.9) -> AnchoredEffect | Unsupported

``anchor_effect`` on a flat array of draws of one scalar parameter.

print("from flat draws:", round(anchor_draws(draws.ravel(), "beta", 0.5).anchored_effect, 3))

anchor_effect(posterior: SupportsPosterior, parameter: str, mde: float, *, index: int | None = None, credence: float = 0.9, definition: IntervalDefinition = 'hdi', mass: float = 0.9) -> AnchoredEffect | Unsupported

Anchor an MDE against the posterior of ``parameter``.

ae = anchor_effect(posterior, "beta", mde=0.5, credence=0.9)

bartlett_bandwidth(n_periods: int) -> int

Newey–West rule of thumb ``max(1, floor(4 · (T/100)^(2/9)))``.

print(f"switchback: {sb.effect:.3f} ± {sb.se:.3f}  bandwidth={sb.detail.get('bandwidth')}  (rule: {bartlett_bandwidth(spec.n_periods)})")

calibrate_method(method: str | MethodSpec, spec: SimulationSpec, *, alpha: float = 0.05, registry: Mapping[str, MethodSpec] = ..., design: PanelDesign | None = None) -> CalibrationResult

A/A calibration: ``spec`` with ``effect=0`` and intervals at mass ``1 − alpha``.

cal: CalibrationResult = calibrate_method("difference_in_differences", spec, alpha=0.05)

calibrate_registry(spec: SimulationSpec, registry: Mapping[str, MethodSpec] = ..., *, alpha: float = 0.05) -> tuple[Mapping[str, MethodSpec], tuple[CalibrationResult, ...]]

Calibrate every method in ``registry`` and return a new registry with statuses set.

registry, results = calibrate_registry(small, alpha=0.05)

cluster_mde(design: ClusterDesign, sd: float, *, alpha: float = 0.05, power: float = 0.8, two_sided: bool = True) -> MDE

Minimum detectable effect under ``design``; see ``cluster_power`` for the scale.

print("cluster MDE:", round(cluster_mde(cd, sd=2.0).effect, 4))

cluster_power(design: ClusterDesign, effect: float, sd: float, *, alpha: float = 0.05, two_sided: bool = True) -> PowerResult

Power to detect ``effect`` on the individual-level scale under ``design``.

print("cluster power for effect 0.5:", round(cluster_power(cd, effect=0.5, sd=2.0).power, 3))

clusters_needed(effect: float, sd: float, cluster_size: int, icc: float, *, alpha: float = 0.05, power: float = 0.8, two_sided: bool = True, allocation: float = 0.5) -> SampleSize | Unsupported

Smallest number of clusters of size ``cluster_size`` reaching ``power`` for ``effect``.

cn = clusters_needed(effect=0.5, sd=2.0, cluster_size=40, icc=0.05)

coefficient_mde(se: float, *, alpha: float = 0.05, power: float = 0.8, two_sided: bool = True) -> MDE

Minimum detectable coefficient given its design standard error.

print("coefficient_mde:", round(coefficient_mde(se=0.1, power=0.9).effect, 4))

coefficient_power(effect: float, se: float, *, alpha: float = 0.05, two_sided: bool = True) -> PowerResult

Power to detect a coefficient of size ``effect`` whose design SE is ``se``.

print("coefficient_power:", round(coefficient_power(0.3, se=0.1).power, 3))

coefficient_sample_size(effect: float, se: float, n: int, *, alpha: float = 0.05, power: float = 0.8, two_sided: bool = True) -> SampleSize | Unsupported

Observations needed for a coefficient whose SE is ``se`` at ``n`` observations.

css = coefficient_sample_size(effect=0.2, se=0.1, n=500)

constant(n_periods: int, dose: float, *, treatment: str = 'dose') -> Schedule

The same dose in every period — no temporal contrast.

flat = fisher_information(world.surface, data_with(constant(T, 50.0)), world.theta, NOISE_SD, method=method)

contrast_score(schedule: Schedule) -> float

Mean squared first difference over mean squared dose (module docstring); ``0`` if flat.

print(f"{p:18s} mean={s.mean:5.1f} total={s.total:6.1f} contrast={contrast_score(s):.3f}  {s.doses[:6]}...")

cost_per_outcome_interval(cost: float, effect: float, effect_se: float, *, alpha: float = 0.05) -> CostPerOutcomeInterval

Fieller and naive intervals for ``cost / effect`` at confidence ``1 − alpha``.

ci: CostPerOutcomeInterval = cost_per_outcome_interval(cost=1000.0, effect=25.0, effect_se=6.0)

cost_per_outcome_power(cost: float, true_effect: float, effect_se: float, *, alpha: float = 0.05, threshold: float | None = None) -> CostPerOutcomePower

Probability the Fieller upper bound is finite (``threshold=None``) or below ``threshold``.

cp: CostPerOutcomePower = cost_per_outcome_power(1000.0, true_effect=25.0, effect_se=6.0, threshold=60.0)

crossing_probabilities(rule: StoppingRule, drift: float, *, binding_only: bool = True, n_grid: int = 601, width: float = 8.0) -> CrossingProbabilities

Exact first-crossing probabilities for ``rule`` at ``drift``.

spent = crossing_probabilities(harm_only, 0.0)

decayed_sd(posterior_sd: float, periods_elapsed: float, half_life_periods: float) -> float

``posterior_sd · 2^(t / (2 · half_life))``: the posterior variance doubles every half-life.

print("decayed sd after 0, 6, 12 periods (hl=6):", [round(decayed_sd(post_sd, t, 6.0), 4) for t in (0, 6, 12)])

design_effect(cluster_size: int, icc: float) -> float

``1 + (m − 1) · icc`` — variance inflation of a cluster-randomized mean.

print("design effect:", design_effect(40, 0.05), "| effective n:", round(effective_sample_size(24, 40, 0.05), 1))

design_for_method(method: str | MethodSpec, registry: Mapping[str, MethodSpec] = ...) -> PanelDesign

The panel design a method's data requirements imply.

d: PanelDesign = design_for_method(m)

design_to_identify(surface: SupportsForward, candidates: Design, theta: Theta, noise_sd: float, target: str, n: int, *, prior_sds: Mapping[str, float] | None = None, seed: int | None = None, n_restarts: int = 3, max_passes: int = 20, method: DerivativeMethod = 'auto', parameter_scales: Mapping[str, float] | None = None) -> IdentifyingDesign | Unsupported

The ``n`` candidate rows (replicates allowed) minimizing ``target``'s expected posterior sd.

print("one row under flat priors ->", type(design_to_identify(steady, candidates, theta, NOISE_SD, target="k_a", n=1, seed=0, method=method)).__name__)

difference_in_differences_se(spec: SimulationSpec) -> float

Exact SE of the DiD estimator under the spec's DGP (derivation in the module docstring).

se = difference_in_differences_se(spec)

difference_se(n: int, sd: float, allocation: float = 0.5) -> float

``sd / sqrt(n · p · (1 − p))`` — the SE of a two-arm difference in means.

se = difference_se(200, sd=2.0, allocation=0.5)

discount_weights(n_periods: int, rate: float) -> Array

``(1 + rate)^(-t)`` for ``t = 0 … n_periods − 1``: period 0 is undiscounted.

print("discount weights (8 periods, 2%):", np.round(discount_weights(8, 0.02), 4))

effective_sample_size(n_clusters: int, cluster_size: int, icc: float) -> float

``k · m / DE`` — the number of independent observations the design is worth.

print("design effect:", design_effect(40, 0.05), "| effective n:", round(effective_sample_size(24, 40, 0.05), 1))

eig_gaussian(prior_sd: float, experiment_se: float) -> float

``0.5 · ln(1 + prior_sd² / experiment_se²)`` nats.

print(f"experiment_se={se:.2f}  EIG = {eig_gaussian(prior_sd, se):.4f} nats")

eig_monte_carlo(prior_draws: ArrayLike, experiment_se: float, n_sims: int = 2000, seed: int | None = None, *, chunk: int = 256) -> EIGEstimate

Nested Monte-Carlo EIG of ``y = theta + N(0, experiment_se²)`` under an empirical prior.

est: EIGEstimate = eig_monte_carlo(prior_draws, 0.3, n_sims=n_sims, seed=1)

elasticity(table: SensitivityTable) -> dict[str, float]

``(Δnet / net_base) / (Δx / x_base)`` per candidate, from the grid values around the base.

print("elasticities:", {k: round(v, 3) for k, v in elasticity(table).items()})

estimable_combinations(observations: Sequence[Observation], theta: Theta, *, at: Sequence[Theta] = (), parameters: Sequence[str] | None = None, scaling: Scaling = 'log', tolerance: float = 1e-08, direction_tolerance: float | None = None, max_support: int = 3, max_coefficient: int = 3, max_candidates: int = 200000, step: float = 0.0001, parameter_scales: Mapping[str, float] | None = None) -> EstimabilityReport | Unsupported

Rank, flat directions, and the named combinations this design can estimate.

report = estimable_combinations([low], theta, tolerance=tolerance)

estimate(method: str | MethodSpec, arrays: PanelArrays, *, mass: float = 0.95, registry: Mapping[str, MethodSpec] = ...) -> MethodEstimate | Unsupported

Run a registered method on ``arrays`` with a ``wald`` interval at ``mass``.

out = estimate(method, arrays, mass=0.95)

estimate_cluster_based_regression(outcome: ArrayLike, treated: Sequence[int], pre: slice, post: slice, *, mass: float = 0.95) -> MethodEstimate | Unsupported

Keyword convenience: build the ``PanelArrays`` and run ``estimate_arrays``.

cbr = estimate_cluster_based_regression(y, tr, pre, post)

estimate_difference_in_differences(outcome: ArrayLike, treated: Sequence[int], pre: slice, post: slice, *, mass: float = 0.95) -> MethodEstimate | Unsupported

Keyword convenience: build the ``PanelArrays`` and run ``estimate_arrays``.

did = estimate_difference_in_differences(y, tr, pre, post)

estimate_ghost(outcome: ArrayLike, treated: Sequence[int], post: slice, exposed: ArrayLike, *, pre: slice = slice(0, 0, None), mass: float = 0.95) -> MethodEstimate | Unsupported

Keyword convenience: build ``PanelArrays`` with ``exposed`` and run ``estimate_arrays``.

gh = estimate_ghost(gh_arrays.outcome, gh_arrays.treated, gh_arrays.post, gh_arrays.exposed)

estimate_switchback(outcome: ArrayLike, assignment: ArrayLike, *, treated: Sequence[int] = (), pre: slice = slice(0, 0, None), post: slice = slice(None, None, None), mass: float = 0.95) -> MethodEstimate | Unsupported

Keyword convenience: build ``PanelArrays`` with a schedule and run ``estimate_arrays``.

sb = estimate_switchback(sb_arrays.outcome, sb_arrays.assignment)

estimate_synthetic_control(outcome: ArrayLike, treated: Sequence[int], pre: slice, post: slice, *, mass: float = 0.95, rmspe_ratio_max: float | None = None) -> MethodEstimate | Unsupported

Keyword convenience with the optional RMSPE-ratio placebo filter.

sc = estimate_synthetic_control(y, tr, pre, post, rmspe_ratio_max=5.0)

estimate_time_based_regression(outcome: ArrayLike, treated: Sequence[int], pre: slice, post: slice, *, mass: float = 0.95) -> MethodEstimate | Unsupported

Keyword convenience: build the ``PanelArrays`` and run ``estimate_arrays``.

tbr = estimate_time_based_regression(y, tr, pre, post)

evaluate_candidate(candidate: DesignCandidate, decision: DecisionSpec, prior_mean: float, prior_sd: float, economics: EconomicInputs, *, alpha: float = 0.05, effect: float | None = None) -> CandidateScore

Score ``candidate`` on ``decision`` under ``theta ~ N(prior_mean, prior_sd²)``.

scores: list[CandidateScore] = [evaluate_candidate(c, decision, prior_mean=mean, prior_sd=sd, economics=economics) for c in designs]

evoi_gaussian(decision: DecisionSpec, prior_mean: float, prior_sd: float, experiment_se: float) -> EVOIResult

``evpi_gaussian`` and ``evsi_gaussian`` together, as an ``EVOIResult``.

res: EVOIResult = evoi_gaussian(decision, mean, sd, se)

evpi(decision: DecisionSpec, prior_draws: ArrayLike, *, value_fn: ValueFn | None = None) -> float

Monte-Carlo EVPI: ``mean(max(payoff, 0)) − max(mean(payoff), 0)`` over the prior draws.

print("MC EVPI:", round(evpi(decision, draws), 2), "vs closed", round(evpi_gaussian(decision, mean, sd), 2))

evpi_gaussian(decision: DecisionSpec, prior_mean: float, prior_sd: float) -> float

Expected value of perfect information under ``theta ~ N(prior_mean, prior_sd²)``.

print("EVPI:", round(evpi_gaussian(decision, mean, sd), 2), decision.numeraire)

evsi(decision: DecisionSpec, prior_draws: ArrayLike, experiment_se: float, *, n_sims: int = 1000, seed: int | None = None, value_fn: ValueFn | None = None, chunk: int = 256) -> EVOIResult

Monte-Carlo EVSI of observing ``y = theta + N(0, experiment_se²)``.

mc = evsi(decision, draws, se, n_sims=n_sims, seed=2)

evsi_gaussian(decision: DecisionSpec, prior_mean: float, prior_sd: float, experiment_se: float) -> float

Expected value of the sample information an experiment with ``experiment_se`` yields.

print("EVSI:", round(evsi_gaussian(decision, mean, sd, se), 2), decision.numeraire)

expected_posterior_sd(prior_sds: Mapping[str, float] | None, fisher: FisherInformation) -> dict[str, float] | Unsupported

Laplace posterior sds ``sqrt(diag((diag(1/sd²) + FI)⁻¹))`` per parameter coordinate.

print("expected posterior sd (flat prior) ->", type(expected_posterior_sd(None, flat)).__name__)

experiment_se_for_design(design_kind: str, reference_value: float, *, relative_se: float | None = None) -> float | Unsupported

``relative_se · |reference_value|``, with ``relative_se`` looked up by design kind.

se_holdout = experiment_se_for_design("cluster_holdout", reference_value=2.0)

experiment_value(information_value: float, opportunity_cost: OpportunityCost | float, fixed_cost: float, *, numeraire: str = '') -> ExperimentValue

``information_value − opportunity_cost − fixed_cost`` in one numeraire.

ev: ExperimentValue = experiment_value(info, oc, fixed_cost=250.0)

fisher_information(surface: SupportsForward, design_doses: Theta, theta: Theta, noise_sd: float, *, parameters: Sequence[str] | None = None, method: DerivativeMethod = 'auto', parameter_scales: Mapping[str, float] | None = None) -> FisherInformation | Unsupported

``J'J / noise_sd²`` at ``design_doses`` and ``theta`` for the mean's parameters.

fi = fisher_information(world.surface, data_with(s), world.theta, NOISE_SD, method=method)

harm_boundary(probability: float, information: Sequence[float] | LookSchedule, *, margin: float = 0.0, se_at_full_information: float = 1.0, kind: BoundaryKind = 'harm') -> Boundary

The Z threshold of a posterior-probability harm rule, look by look.

flat = harm_boundary(0.95, information)

holdout_tradeoff(design: ClusterDesign, sd: float, fractions: Sequence[float] = (0.1, 0.2, 0.3, 0.4, 0.5), *, alpha: float = 0.05, power: float = 0.8, two_sided: bool = True) -> HoldoutTradeoff

MDE at each holdout share for ``design`` (its own ``allocation`` is ignored).

ht: HoldoutTradeoff = holdout_tradeoff(cd, sd=2.0, fractions=(0.1, 0.2, 0.3, 0.4, 0.5))

identifiability_ridge(surface: SupportsForward, design_doses: Theta, theta: Theta, noise_sd: float, *, pairs: Sequence[tuple[str, str]] = (), parameters: Sequence[str] | None = None, method: DerivativeMethod = 'auto', parameter_scales: Mapping[str, float] | None = None) -> IdentifiabilityRidge | Unsupported

The flattest direction of the information matrix at this design, and pairwise couplings.

ridge = identifiability_ridge(world.surface, data_with(pulse(T, 80.0, 20.0, on=2, off=2)), world.theta, NOISE_SD, pairs=(("beta_a", "k_a"),), method=method)

information_fractions(counts: Sequence[float], total: float | None = None) -> tuple[float, ...]

Information fractions from cumulative counts (enrolled, completed, events).

print("from cumulative enrollment:", information_fractions([48, 96, 144, 192, 240]))

information_half_life(prior_sd: float, posterior_sd: float, half_life_periods: float) -> float

Periods until half the information the experiment bought has decayed.

print("information half-life:", round(information_half_life(prior_sd, post_sd, 6.0), 2), "periods")

information_value_of(decision: DecisionSpec, prior_mean: float, prior_sd: float, experiment_se: float) -> float

EVSI of an experiment measuring the decision parameter to ``experiment_se``.

info = information_value_of(decision, prior_mean=1.2, prior_sd=0.5, experiment_se=0.2)

leaderboard(spec: SimulationSpec, effects: Sequence[float], registry: Mapping[str, MethodSpec] = ..., *, alpha: float = 0.05) -> Leaderboard

Rank ``registry``'s methods by mean power over ``effects`` at the calibrated size.

board: Leaderboard = leaderboard(small, effects=(0.8, 1.6), alpha=0.05)

match_clusters(pre_outcomes: ArrayLike, *, labels: Sequence[str] | None = None, unit: Unit | None = None, metric: MatchMetric = 'level', seed: int | None = None) -> Assignment

Pair clusters by pre-period similarity and randomize one member of each pair.

asg: Assignment = match_clusters(pre, labels=[f"r{i}" for i in range(9)], unit=region, metric=metric, seed=1)

max_detectable_cost_per_outcome(cost: float, effect_mde: float) -> float

``cost / effect_mde``: the largest cost per outcome unit the experiment can bound.

print("largest bounded cost per outcome at the MDE:", round(max_detectable_cost_per_outcome(1000.0, m.effect), 2))

mde(n: int, sd: float, *, alpha: float = 0.05, power: float = 0.8, two_sided: bool = True, allocation: float = 0.5) -> MDE

Minimum detectable effect of a two-arm difference in means with ``n`` units in total.

m: MDE = mde(200, sd=2.0, power=0.8)

method_assumption(name: str, statement: str, challenged_by: str) -> Assumption

A method-facet ``Assumption`` in its default ``unverified`` state.

extra = method_assumption("balanced_exposure", "exposure is balanced across arms", "an imbalance test")

method_spec(method: str | MethodSpec, registry: Mapping[str, MethodSpec] = ...) -> MethodSpec

Resolve a name through ``registry`` (a ``MethodSpec`` passes through).

ms: MethodSpec = method_spec("switchback")

mid_horizon_factor(n_periods: int, rate: float) -> float

Mean of ``discount_weights``: undiscounted total × this = discounted total.

print("mid-horizon factor:", round(mid_horizon_factor(8, 0.02), 4))

monitor(rule: StoppingRule, z: Sequence[float], *, effects: Sequence[float] | None = None, ses: Sequence[float] | None = None) -> MonitoringPath

Walk realized statistics against ``rule`` and stop at the first crossing.

path = monitor(rule, observed_z[:3], effects=effects[:3], ses=ses[:3])

obrien_fleming(alpha: float, information: Sequence[float] | LookSchedule, *, side: Side = 'two_sided', kind: BoundaryKind = 'efficacy', n_grid: int = 601, width: float = 8.0) -> Boundary

Thresholds ``c / sqrt(t)``, solved so the total error is ``alpha``.

shapes = {
    "pocock": pocock(ALPHA, information),
    "obrien_fleming": obrien_fleming(ALPHA, information),
    "lan_demets_obf": alpha_spending(ALPHA, information, family="obrien_fleming"),
    "lan_demets_pocock": alpha_spending(ALPHA, information, family="pocock"),
}

observation_from_model(model: ModelSpec, data: Mapping[str, Any], *, noise_sd: float | None = None, theta: Theta | None = None, label: str = '') -> Observation

The observation a fitted ``ModelSpec`` represents: its mean over the panel's rows.

print(observation_from_model(model, {"dose": np.array([1.0])}, theta=truth).noise_sd)

operating_characteristics(rule: StoppingRule, drift: float, *, binding_only: bool = True, n_grid: int = 601, width: float = 8.0) -> OperatingCharacteristics

:func:`crossing_probabilities` plus the expected information and look count.

null = operating_characteristics(rule, 0.0)

opportunity_cost(holdout_fraction: float, n_periods: int, dose_per_period: float, marginal_value_ratio: float | ArrayLike, value_per_outcome: ValuePerOutcome, discount_rate: float, *, dose_unit: str = 'dose', dose_cost_per_unit: float = 0.0) -> OpportunityCost

Signed value forgone by withholding ``holdout_fraction`` of the dose for ``n_periods``.

negative = opportunity_cost(0.25, 8, 20.0, 0.2, vpo, 0.02, dose_cost_per_unit=1.0)

panel_arrays(outcome: ArrayLike, treated: Sequence[int], pre: slice, post: slice, *, assignment: ArrayLike | None = None, exposed: ArrayLike | None = None) -> PanelArrays

Validate and freeze the arrays a method reads; see ``PanelArrays``.

all_treated = panel_arrays(y, treated=range(y.shape[0]), pre=pre, post=post)

pareto_front(scores: Sequence[CandidateScore], objectives: Sequence[str] = ('net_value', '-cost', 'eig')) -> tuple[CandidateScore, ...]

The non-dominated scores: nothing is at least as good on every objective and better on one.

front = pareto_front(scores, objectives=("net_value", "-cost", "eig"))

perturb(candidates: Sequence[DesignCandidate], decision: DecisionSpec, prior_mean: float, prior_sd: float, economics: EconomicInputs, parameter: Parameter, grid: Sequence[float], *, alpha: float = 0.05, effect: float | None = None) -> SensitivityTable

Re-score every candidate at each grid value of ``parameter``; report winners and flips.

table: SensitivityTable = perturb(designs, decision, mean, sd, economics, parameter, grid=(0.5, 1.0, 2.0, 3.0, 5.0, 8.0))

pocock(alpha: float, information: Sequence[float] | LookSchedule, *, side: Side = 'two_sided', kind: BoundaryKind = 'efficacy', n_grid: int = 601, width: float = 8.0) -> Boundary

A constant critical value on the Z scale, solved so the total error is ``alpha``.

shapes = {
    "pocock": pocock(ALPHA, information),
    "obrien_fleming": obrien_fleming(ALPHA, information),
    "lan_demets_obf": alpha_spending(ALPHA, information, family="obrien_fleming"),
    "lan_demets_pocock": alpha_spending(ALPHA, information, family="pocock"),
}

power(n: int, effect: float, sd: float, *, alpha: float = 0.05, two_sided: bool = True, allocation: float = 0.5) -> PowerResult

Power of a two-arm difference in means with ``n`` units in total.

pr: PowerResult = power(200, effect=0.6, sd=2.0)

power_curve(n: int, sd: float, effects: Sequence[float], *, alpha: float = 0.05, two_sided: bool = True, allocation: float = 0.5) -> PowerCurve

Power over a grid of effect sizes for a fixed two-arm design.

curve: PowerCurve = power_curve(200, 2.0, effects=np.linspace(0.0, 1.2, 7))

power_from_se(effect: float, se: float, *, alpha: float = 0.05, two_sided: bool = True) -> PowerResult

Power for an effect whose estimator has standard error ``se``.

print("power_from_se:", round(power_from_se(0.3, 0.1).power, 3))

preposterior_sd(prior_sd: float, experiment_se: float) -> float

``prior_sd · preposterior_sd_ratio``: the sd of the posterior mean to come.

print("preposterior ratio:", round(preposterior_sd_ratio(sd, se), 4), "| preposterior sd:", round(preposterior_sd(sd, se), 4))

preposterior_sd_ratio(prior_sd: float, experiment_se: float) -> float

``sqrt(prior_sd² / (prior_sd² + experiment_se²))``.

print("preposterior ratio:", round(preposterior_sd_ratio(sd, se), 4), "| preposterior sd:", round(preposterior_sd(sd, se), 4))

prescribe_measurements(observations: Sequence[Observation], candidates: Sequence[Observation], theta: Theta, *, parameters: Sequence[str] | None = None, scaling: Scaling = 'log', tolerance: float = 1e-08, max_additions: int = 4, step: float = 0.0001, parameter_scales: Mapping[str, float] | None = None) -> Prescription | Unsupported

Greedily choose candidates until the design has full rank, or say what is left.

plan: Prescription = prescribe_measurements([low], [wide], theta, tolerance=3e-2)

prior_from_history(summaries: Sequence[StudySummary], half_life_periods: float) -> tuple[float, float]

``(mean, sd)`` of the inverse-variance combination of the decayed studies.

mean, sd = prior_from_history(history, half_life_periods=8.0)

profile_combination(model: ModelSpec, data: Mapping[str, Any], start: Mapping[str, Any], combination: Combination, *, grid: Sequence[float]) -> tuple[tuple[float, ...], tuple[float, ...]]

The profile of a *combination* of parameters — ``alpha / k`` rather than ``alpha``.

values, drops = profile_combination(model, simulated, truth, ratio, grid=ratio_grid)

profile_likelihood(model: ModelSpec, data: Mapping[str, Any], start: Mapping[str, Any], target: str, *, grid: Sequence[float]) -> tuple[tuple[float, ...], tuple[float, ...]]

``(values, drops)``: the profile of one parameter, as the drop in log-likelihood.

values, drops = profile_likelihood(model, simulated, truth, "alpha", grid=grid)

pulse(n_periods: int, high: float, low: float, *, on: int, off: int, start_on: bool = True, treatment: str = 'dose') -> Schedule

``on`` periods at ``high`` then ``off`` at ``low``, repeating.

pulsed = fisher_information(world.surface, data_with(pulse(T, 80.0, 20.0, on=2, off=2)), world.theta, NOISE_SD, method=method)

ramp(n_periods: int, start: float, stop: float, *, treatment: str = 'dose') -> Schedule

Linear from ``start`` at period 0 to ``stop`` at the last period (``start`` alone if one).

schedules: list[Schedule] = [
    constant(T, 50.0),
    pulse(T, 80.0, 20.0, on=2, off=2),
    alternating(T, 80.0, 20.0),
    ramp(T, 0.0, 100.0),
    random_switchback(T, 80.0, 20.0, seed=0),
]

random_switchback(n_periods: int, high: float, low: float, *, seed: int | None, p_high: float = 0.5, treatment: str = 'dose') -> Schedule

Each period independently at ``high`` with probability ``p_high``, else ``low``; seeded.

schedules: list[Schedule] = [
    constant(T, 50.0),
    pulse(T, 80.0, 20.0, on=2, off=2),
    alternating(T, 80.0, 20.0),
    ramp(T, 0.0, 100.0),
    random_switchback(T, 80.0, 20.0, seed=0),
]

rank_treatments(candidates: Sequence[TreatmentCandidate]) -> tuple[LearningPriority, ...]

Every candidate scored and ranked by net value, then EIG, then name.

for p in rank_treatments(candidates):
    assert isinstance(p, LearningPriority)
    print(f"#{p.rank} {p.treatment:12s} eig={p.eig:.3f} evsi={p.evoi:8.2f} net={p.net_value:8.2f} cost={p.cost:7.2f}")

recommend(candidates: Sequence[TreatmentCandidate], budget: float | None = None) -> Recommendation | Unsupported

Greedy knapsack over ``rank_treatments``: net value per cost, within ``budget``.

rec = recommend(candidates, budget=600.0)

ridge_of(fisher: FisherInformation, *, pairs: Sequence[tuple[str, str]] = ()) -> IdentifiabilityRidge | Unsupported

``identifiability_ridge`` from an information matrix already computed.

print("from the flat design:", ridge_of(flat).ridge_parameters, "| with pairs ->", type(ridge_of(flat, pairs=(("beta_a", "k_a"),))).__name__)

sample_size(effect: float, sd: float, *, alpha: float = 0.05, power: float = 0.8, two_sided: bool = True, allocation: float = 0.5) -> SampleSize | Unsupported

Smallest total ``n`` (integer arms at ``allocation``) reaching ``power`` for ``effect``.

ss = sample_size(effect=0.6, sd=2.0, power=0.8)

schedule_with_cooldown(scores: Sequence[CandidateScore], horizon_periods: int) -> ProgramSchedule

Greedy end-to-end sequence: best net value first, each run then its cooldown.

sched: ProgramSchedule = schedule_with_cooldown(scores, horizon_periods=20)

sensitivity_matrix(observations: Sequence[Observation], theta: Theta, *, parameters: Sequence[str] | None = None, scaling: Scaling = 'log', step: float = 0.0001, parameter_scales: Mapping[str, float] | None = None) -> SensitivityMatrix | Unsupported

Central differences of every observation with respect to every parameter.

sensitivity: SensitivityMatrix = sensitivity_matrix([low], theta, scaling=scaling)

simplex_weights(target: Array, donors: Array) -> Array

Non-negative weights summing to one that best reproduce ``target`` from ``donors`` rows.

w = simplex_weights(target, donors)

simulate_panel(spec: SimulationSpec, rng: Generator, *, design: PanelDesign = 'holdout') -> PanelArrays

Draw one panel from ``spec`` and apply ``spec.effect`` under ``design``.

arrays: PanelArrays = simulate_panel(spec, rng, design="holdout")

simulated_identifiability(model: ModelSpec, data: Mapping[str, Any], truth: Mapping[str, Any], *, targets: Sequence[str] | None = None, combinations: Sequence[Combination] = (), seed: int | None = None, level: float = 0.95, range_factor: float = 20.0, n_grid: int = 21, noise_sd: float | None = None) -> ProfileReport | Unsupported

Run the proposed experiment on simulated data and profile each target.

print(simulated_identifiability(model, {"dose": doses}, {"alpha": 4.0, "k": 10.0}).missing)

simulated_power(method: str | MethodSpec, spec: SimulationSpec, *, registry: Mapping[str, MethodSpec] = ..., design: PanelDesign | None = None, predicted_power: float | None = None) -> SimulatedPower

A/B simulation at ``spec.effect``: power, bias, RMSE and coverage at ``spec.mass``.

sp: SimulatedPower = simulated_power("difference_in_differences", ab, predicted_power=predicted)

spending(kind: SpendingFunction, t: float, alpha: float, *, rho: float = 1.0) -> float

Cumulative error a spending function allows to have been spent by fraction ``t``.

print(f"{'power (rho=3)':16s} " + " ".join(f"{spending('power', t, ALPHA, rho=3.0):7.4f}" for t in grid))

t_critical(mass: float, df: float) -> float

Student-t two-sided critical value ``t_{df, (1 + mass) / 2}``.

print(df, round(t_critical(0.95, df), 4))

time_to_re_experiment(posterior_sd: float, half_life_periods: float, experiment_se: float, min_eig: float, *, design_kind: str = 'unspecified') -> ReExperimentTiming

First period at which re-running the experiment is worth ``min_eig`` nats.

timing: ReExperimentTiming = time_to_re_experiment(post_sd, 6.0, experiment_se=0.2, min_eig=0.3, design_kind="cluster_holdout")

wald_t(effect: float, se: float, mass: float, df: float) -> Interval

``effect ± t_critical(mass, df) · se`` as a ``wald`` interval at ``mass``.

print(wald_t(1.0, 0.5, 0.95, df=10))

Agreement(estimand_name: str, estimand_hash: str, measurement_hash: str, source: str, estimate: float, se: float, posterior_mean: float, posterior_sd: float, interval: Interval, z: float, p: float, inside: bool, ... and 5 more fields)

How a measurement sits against the posterior of its own estimand.

assert isinstance(ok, Agreement)

AgreementVerdict(*args, **kwargs)

verdict: AgreementVerdict = ok.verdict

CalibratedSpec(spec: SurfaceSpec, treatment: str, parameter: str, prior: Prior, family: Literal['lognormal', 'gamma'], target: str, sources: tuple[str, ...], plan_statuses: dict[str, str], target_mean: float, target_se: float, design_factor: float, amplitude_mean: float, ... and 2 more fields)

The prior route's output: a new surface spec and how it was reached.

assert isinstance(cal, CalibratedSpec)

Correction(name: str, kind: Literal['multiplicative', 'additive', 'se_scale'], value: float, counterfactual: float, corrected: float, detail: dict[str, str] = {}, ledger_line: LedgerLine)

One correction operator's output: the number applied and the line that records it.

assert isinstance(c, Correction)

CorrectionKind(*args, **kwargs)

kind: CorrectionKind = c.kind

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

print(FACET_PREFIX, "->", [facet_of(line) for line in baseline.lines])

Ledger(lines: tuple[LedgerLine, ...] = ())

An immutable sequence of ``LedgerLine``; see the module docstring for the rule.

pruned = Ledger(lines=tuple(ln for ln in ledger.lines if facet_of(ln) != "level"))

Measurement(estimand: Estimand, estimate: float, se: float, definition: Literal['wald', 'eti', 'hdi'] = 'wald', mass: float = 0.95, method: str = '', n_units: int | None = None, n_periods: int | None = None, design_factor: float | None = None, source: str, assumptions: tuple[Assumption, ...] = ())

One external estimate of an estimand, with its uncertainty and provenance (D6.1).

m2 = Measurement(estimand=lift, estimate=0.9, se=0.5, definition="eti", mass=0.9, source="study-2")

PriorFamily(*args, **kwargs)

family: PriorFamily = fam

ResolvedTransfer(status: Literal['identified', 'downgraded', 'blocked', 'unsupported', 'unverified'], plan: TransferPlan, factor: float = 1.0, offset: float = 0.0, se_scale: float = 1.0, corrections: tuple[Correction, ...] = (), transport: Verdict | None = None, ledger: Ledger, completeness: Verdict, reason: str = '')

A plan with its corrections applied: the composite operator and the ledger that licenses it.

resolved: ResolvedTransfer = resolve(plan, corrections=[window, level])

str(object='') -> str str(bytes_or_buffer[, encoding[, errors]]) -> str

print("UNCORRECTED =", repr(UNCORRECTED))

aggregation_level(level_source: Level, level_target: Level, *, cluster_size: int, icc: float) -> Correction | Unsupported

Individual ↔ cluster ↔ aggregate: point factor 1, SE through ``variance_reweight``.

level = aggregation_level(source.level, target.level, cluster_size=25, icc=0.05)

agreement(producer: SupportsEstimands, measurement: Measurement, *, definition: IntervalDefinition = 'hdi', mass: float = 0.9, seed: int | None = None, tension_at: float = 1.0, disagrees_at: float = 2.0) -> Agreement | Unsupported | Blocked

Realize ``measurement.estimand`` on ``producer`` and compare (module docstring).

ok = agreement(calibrated, experiment, definition="hdi", mass=0.9, seed=0)

amplitude_prior(mean: float, sd: float, family: PriorFamily = 'lognormal') -> Prior

The moment-matched ``Prior`` for a positive amplitude with the given mean and sd.

print(amplitude_prior(amp_mean, amp_sd, family))

attach(measurement: Measurement, surface: Surface, data: Mapping[str, ArrayLike], *, doses: DoseMap | None = None) -> ModelSpec | Unsupported

``surface.model`` with the measurement's constraint appended, or why it cannot be.

model = attach(experiment, surface, data)

carryover_window_factor(carryover: CarryoverKernel, theta: Theta, window_periods: int, *, treatment: str) -> Correction | Unsupported

``factor = 1 / Σ_{l < window_periods} w_l``: a short-window total scaled to steady state.

window = carryover_window_factor(geo, {"lam_a": lam}, 2, treatment="a")

chord_to_marginal(surface: Surface, theta: Theta, treatment: str, d0: float, d1: float, at: float, *, held: Mapping[str, ArrayLike] | None = None) -> Correction | Unsupported

``factor = f'(at) / ((f(d1) − f(d0)) / (d1 − d0))``: read a chord as a marginal.

c = chord_to_marginal(surface, theta, "a", d0, d1, at)

combine_inverse_variance(targets: Sequence[float], ses: Sequence[float]) -> tuple[float, float]

Fixed-effect pooling of independent estimates of one quantity: ``(mean, se)``.

mean, se = combine_inverse_variance([m1.estimate, m2.estimate], [m1.se, m2.se])

combine_measurements(measurements: Sequence[Measurement], *, target: Estimand | None = None, corrections: Mapping[str, float] | None = None) -> tuple[float, float, tuple[LedgerLine, ...], dict[str, str]] | Unsupported

Pool measurements on one target estimand: ``(mean, se, ledger_lines, plan_statuses)``.

combined = combine_measurements([experiment, second])

constraint_for(measurement: Measurement, surface: Surface, data: Mapping[str, ArrayLike], *, doses: DoseMap | None = None) -> Constraint | Unsupported

The soft constraint that says the surface predicts ``measurement`` at its experiment.

c = constraint_for(experiment, surface, data)

derive_prior(measurements: Sequence[Measurement], spec: SurfaceSpec, treatment: str, *, beta_draws: ArrayLike, contribution_draws: ArrayLike, family: PriorFamily = 'lognormal', target: Estimand | None = None, corrections: Mapping[str, float] | None = None) -> CalibratedSpec | Unsupported

Turn measurements of a target estimand into an amplitude prior on ``spec`` (D6.2).

cal = derive_prior([experiment, second], spec, "a", beta_draws=beta_draws,
                   contribution_draws=contribution_draws, family=family)

design_factor(beta_samples: ArrayLike, contribution_samples: ArrayLike) -> float

``mean(contribution / beta)`` over paired posterior draws.

f = design_factor(beta_draws, contribution_draws)

dose_path_accumulation(carryover: CarryoverKernel, theta: Theta, dose_path: ArrayLike, *, treatment: str) -> Correction | Unsupported

``factor = Σ_t (w ∗ x)_t / Σ_t x_t``: the effective dose delivered inside the path's window.

acc = dose_path_accumulation(geo, {"lam_a": lam}, path, treatment="a")

facet_of(line: LedgerLine) -> str | None

The facet a line stands in for, or ``None`` for a non-facet line.

print(FACET_PREFIX, "->", [facet_of(line) for line in baseline.lines])

fit_calibrated(spec: SurfaceSpec, panel: Panel, measurements: Sequence[Measurement], *, backend: Backend | str = 'laplace', draws: int = 1000, tune: int = 1000, chains: int = 4, seed: int | None = None, doses: DoseMap | None = None) -> FitResult | Unsupported

``surface.fit`` with every measurement attached as a soft constraint.

calibrated = fit_calibrated(spec, world.panel, [experiment], backend="laplace", draws=500, seed=0)

lognormal_from_moments(mean: float, sd: float) -> tuple[float, float]

``(mu, sigma)`` of the lognormal with the given mean and standard deviation.

mu, sigma = lognormal_from_moments(amp_mean, amp_sd)

lognormal_mu_from_moments(value: float, se: float) -> float

``log(value) - sigma^2 / 2``: the log-scale location whose lognormal has mean ``value``.

mu = lognormal_mu_from_moments(2.0, 0.3)

lognormal_sigma_from_moments(value: float, se: float) -> float

``sqrt(log(1 + (se / value)^2))``: the log-scale sigma of a lognormal with that mean and sd.

sigma = lognormal_sigma_from_moments(2.0, 0.3)

mean_sd_to_gamma(mean: float, sd: float) -> tuple[float, float]

``(shape, rate)`` of the gamma with the given mean and standard deviation.

shape, rate = mean_sd_to_gamma(amp_mean, amp_sd)

resolve(plan: TransferPlan, *, transport: Verdict | None = None, corrections: Sequence[Correction] = ()) -> ResolvedTransfer

Compose the corrections over the plan and write the ledger (module docstring).

resolved: ResolvedTransfer = resolve(plan, corrections=[window, level])

variance_reweight(se: float, n_source: int, n_target: int, *, icc: float | None = None, cluster_size: int | None = None) -> Correction

``se · sqrt(n_source / n_target) · sqrt(1 + (m − 1)·icc)``: the SE at another size.

vr = variance_reweight(0.3, 40, 80, icc=0.05, cluster_size=10)

BaujatData(k: int, q_contribution: tuple[float, ...], influence: tuple[float, ...], detail: dict[str, str] = {})

Per-study ``Q`` contribution (x) against influence on the pooled estimate (y).

bj: BaujatData = baujat(y_asym, se)

Cell(records: tuple[tuple[str, float], ...], name: str = '')

The ``(contributor, value)`` pairs behind one publishable statistic.

small = Cell(records=(("c0", 0.4), ("c1", 0.5)), name="k-1")

Corpus(records: tuple[StudyRecord, ...], name: str = 'corpus')

An ordered collection of ``StudyRecord`` with unique study ids.

corpus = Corpus(records=records, name="demo")

CorpusStore(root: str | Path) -> None

Records and corpora by content hash; corpora also findable by name.

store = CorpusStore(tmp)

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

print("default mapping:", DEFAULT_COLUMNS)

EffectShrinkage(effect: str, studies: tuple[str, ...], estimate: float, se: float, theta: float, interval: Interval, analytic: float, empirical: float | None)

One study effect's raw and shrunk estimate.

s0: EffectShrinkage = rows[0]

EggerTest(k: int, df: int, intercept: float, se: float, t: float, p: float, slope: float, slope_se: float, mass: float, interval: Interval, detail: dict[str, str] = {})

Egger's regression test for funnel asymmetry.

eg: EggerTest = egger(yy, se)

EpsilonCharge(release_id: str, epsilon: float, mechanism: Literal['laplace', 'gaussian'], delta: float = 0.0)

One ε charge against the ledger, named by the release that made it (timestamp-free).

line: EpsilonCharge = l2.charges[-1]

EpsilonLedger(budget: float, charges: tuple[EpsilonCharge, ...] = ())

An immutable ε budget with its sequence of charges (basic sequential composition).

ledger = EpsilonLedger(budget=policy.epsilon_total)

EpsilonSplit(epsilon: float, parts: int, epsilons: tuple[float, ...], composition: str = 'sequential: the parts sum to epsilon')