cleverly.longitudinal.DynamicRegimen

class cleverly.longitudinal.DynamicRegimen(label: str, plan: tuple[RuleNode, ...], rule_kind: FunctionKind | None = None)

A plan whose nodes may be rules rather than constants.

rule_kind="known" declares that every callable node is a fixed rowwise function of one unit’s available history, chosen independently of the analysis sample. A rule that estimates a threshold, or otherwise aggregates across the sample, defines a different, data-adaptive target, and inference for it needs conditions this path does not check. No code can inspect a closure, so the declaration is the check. refuse_regimen_rules() refuses None and "estimated" when the regimen is built, and LTMLE.fit refuses them again before any learner.

DynamicRegimen("treat once L2 rises", (0, lambda h: h["L2"] > 0), rule_kind="known")

A node can hold a known policy density instead, written as a Stochastic node. The unit then draws its arm at that node with probability \(q_t(a \mid H_t)\):

DynamicRegimen("mix", (Stochastic(q1, "q1", density_kind="known"), "low"))

A rule node is handed the history frame, which holds no earlier treatment. A rule that must read an arm drawn at an earlier policy node, such as “continue the arm drawn at the first node”, is written as a one-hot Stochastic node. That node reads the policy frame, and resolve_plans() resolves a one-hot density as the rule it equals.

Parameters:

labelstr

What the reported parameter is named by, exactly as for Regimen.

plantuple

One entry per time point. An entry is a categorical treatment label, meaning that label for everybody at that node, a callable \(d_t(H_t)\) handed that node’s history frame and returning one arm per row, or a Stochastic node declared density_kind="known", whose density function is handed that node’s policy frame and returns one probability per level. Mixing them is the ordinary case: “treat at the first node, then keep treating only while the biomarker stays high” is a constant followed by a rule. The regimen stores the plan as a tuple, so a list or an iterator is read once, when it is built. A single label, callable or policy is not a plan and raises DataError.

rule_kind{“known”, “estimated”} or None

The declaration that every callable node of plan is a known function. One declaration covers every node. "known" is the one value a fit accepts for a plan with a callable node. None, the default, and "estimated" raise CapabilityError for such a plan, and a plan of labels and policy nodes alone is exempt. Each policy node carries its own density_kind. Any other value raises DataError. It is the last field, so DynamicRegimen(label, plan) keeps its positional order.

Attributes:

n_timesint

The number of treatment nodes, one for each entry of plan.

has_policybool

Whether any node of plan is a stochastic policy or a modified treatment policy.

Methods

DynamicRegimen.assignment(data: LongitudinalData) → Any

Evaluate every node’s rule and return the (n, T) arm matrix.

Called once per fit, in cleverly.longitudinal.LTMLE.fit(), and the matrix is what every mask, mechanism design and clever covariate then reads. Evaluating it once rather than at each use is not only cheaper: a rule that is not a deterministic function of the frame would otherwise let the follower masks disagree with the designs the mechanism was evaluated at, and the fit would be answering for no single regimen at all. The callable receives a frame for ergonomic vectorization, but its supported contract is rowwise: it must not read sample summaries or learn a rule from those rows.

A rule is asked for an arm on every row that is still in the study before the node – uncensored through t - 1, and on a survival fit event-free through t - 1 as well, since a unit that has had the event has no treatment decision at t for a rule to make. Off that set the history is the zero fill covariate_history() puts there, so whatever the rule returns is meaningless; it is replaced by a sentinel rather than validated, because such a row is masked out of every regression and every influence curve, and the only way it could still matter is by putting a nan into a design matrix that a learner is called on.

A policy node assigns no arm. Its column holds the label each unit was observed to receive, because the mechanism of a policy plan is evaluated at the observed history. policy_density() evaluates the policy itself.

It runs refuse_regimen_rules() before it calls any rule, because a modified regimen can reach it directly with a declaration this version refuses.

Parameters:

dataLongitudinalData

The validated panel, which supplies each node’s history frame and the rows still in the study.

Returns:

ndarray

(n, T) object array of treatment labels, one column per node. A row that has left the study before a node holds None at that node.

Raises:

CapabilityError

If a node is callable and rule_kind is None or "estimated", or a policy node does not declare density_kind="known".

DataError

If rule_kind is not one of the three states, or a rule raises or returns other than one arm per row.

DynamicRegimen.is_mtp(time: int) → bool

Whether time’s arm is a modified treatment policy of the unit’s own treatment.

Parameters:

timeint

The treatment node, counted from one.

Returns:

bool

True when the entry of plan at time is a policy object, such as Shift or ModifiedPolicy.

DynamicRegimen.is_policy(time: int) → bool

Whether time’s arm is drawn from a known policy density.

Parameters:

timeint

The treatment node, counted from one.

Returns:

bool

True when the entry of plan at time is a Stochastic node.

DynamicRegimen.is_rule(time: int) → bool

Whether time’s arm is decided by a rule rather than declared.

Parameters:

timeint

The treatment node, counted from one.

Returns:

bool

True when the entry of plan at time is callable.

DynamicRegimen.is_tilt(time: int) → bool

Whether time’s arm is drawn from an odds tilt of the treatment mechanism.

Parameters:

timeint

The treatment node, counted from one.

Returns:

bool

True when the entry of plan at time is an Incremental node.

DynamicRegimen.policy_density(data: LongitudinalData, time: int) → FloatArray

Evaluate the policy node at time and check it before it is used.

The density function is handed policy_frame(). It returns an (n, K) array whose columns follow data.treatment_levels[time - 1], or a dataframe whose columns are exactly those levels, which this reorders by level. The label route exists because a permutation of the columns of an array is silent.

The check reads the rows still in the study before the node: uncensored and event-free through time - 1. Every other row is set to a zero row, which no regression and no influence curve reads.

Parameters:

dataLongitudinalData

The validated panel.

timeint

The policy node, counted from one.

Returns:

ndarray

(n, K) policy probabilities, zero on the rows that left the study.

Raises:

CapabilityError

If the node does not declare density_kind="known".

DataError

If time is not a policy node, if the density function raises, or if it returns other than a probability simplex per reachable row over the levels.

Attributes

DynamicRegimen.has_mtp

Whether any node of plan is a modified treatment policy.

DynamicRegimen.has_policy

Whether any node of plan is a stochastic policy or a modified treatment policy.

DynamicRegimen.has_rule

Whether any node of plan is a rule \(d_t(H_t)\).

DynamicRegimen.has_tilt

Whether any node of plan is an odds tilt of the treatment mechanism.

DynamicRegimen.n_times

The number of treatment nodes, one for each entry of plan.

DynamicRegimen.rule_kind: FunctionKind | None = None

DynamicRegimen.label: str

DynamicRegimen.plan: tuple[RuleNode, ...]