API

Dynamics

class whynot.dynamics.Run(states: list, times: list)[source]

Encapsulate a trajectory from a dynamical system simulator.

states

Sequence of states \(x_{t_1}, x_{t_2}, \dots\) produced by the simulator.

times

Sequence of sampled times \({t_1},{t_2}, \dots\) at which states are recorded.

Examples

>>> state_at_year_closest_to_2015 = run[2015]
__getitem__(time)[source]

Return the state closest to the given time.

initial_state

Return initial state of the run.

class whynot.dynamics.BaseState[source]

State of the simulator.

class whynot.dynamics.BaseConfig[source]

Parameters for the simulation dynamics.

class whynot.dynamics.BaseIntervention(config_class, time, **kwargs)[source]

Parameterization of an intervention to a config.

class whynot.dynamics.DynamicsExperiment(name, description, simulator_config, intervention, state_sampler, simulator, propensity_scorer, outcome_extractor, covariate_builder)[source]

Encapsulate a causal experiment on a system dynamics simulator.

__init__(name, description, simulator_config, intervention, state_sampler, simulator, propensity_scorer, outcome_extractor, covariate_builder)[source]

Create an experiment based on system dynamics simulator.

Parameters:
  • name (str) – Name of the experiment
  • description (str) – Short description of the experiment
  • simulator – WhyNot systems dynamics simulator
  • simulator_config (Config) – Instantiated simulator Config for all runs
  • intervention (Intervention) – Instantiated simulator Intervention for treated runs
  • state_sampler – Function to sample initial simulator State (s)
  • propensity_score – Either float or function to generate propensity scores for a set of rollouts
  • outcome_extractor – Function to extract outcome measurement from rollouts
  • covariate_builder – Function to build covariates from rollouts.

All of simulator_config, intervention, state_sampler, propensity_scorer, outcome_extractor, and covariate_builder support a rich set of patterns. We describe the expected API below.

Note: If you wish to use causal graph construction and want to use numpy in any of the user-defined functions, please import the thinly wrapped version of numpy whynot.traceable_numpy and use this for all of the numpy calls, e.g.

import whynot.traceable_numpy as np

def propensity_scorer(run):
    return run[0].covariate_1 / np.max(run[0].values())

Examples

First, simulator_config can either be a constant Config object, or a parameterized function that returns a Config. For instance:

# simulator_config can be a constant Config object,
DynamicsExperiment(
    ...
    simulator_config=Config(123),
    ...)

# Or simulator_config can be a (parameterized) function that
# returns a config.
@parameter(name="p", default=0.2)
def config(p):
    return Config(1234, parameter=p)

DynamicsExperiment(
    ...
    simulator_config=config,
    ...)

Similarly, intervention can either be a constant Intervention object, or a (parameterized) function that returns an Intervention.

DynamicsExperiment(
    ...
    intervention=Intervention(year=92),
    ...)

@parameter(name="p", default=0.2)
def intervention(p):
    return Intervention(year=92, parameter=p)

DynamicsExperiment(
    ...
    intervention=intervention,
    ...)

The state_sampler generates a sequence of initial states. If the sampler uses randomness, it must include rng in the signature and use it as the sole source of randomness. This is to ensure that DynamicsExperiments can be deterministic if desired.

If num_samples is in the signature, then the state_sampler must returns a sequence of num_samples states. Otherwise, it must returns a single state.

def state_sample(rng):
    # Return a single random sample, use randomness in rng
    return State(rng.rand())

def state_sampler(rng, num_samples):
    # Return a list of initial states
    return [State(rng.rand()) for _ in range(num_samples)]

The propensity_scorer produces, for each run, the probability of being assigned to treatment in the observational dataset. The propensity_scorer can be a constant, in which case the probability of treatment is uniform for all units. Otherwise, if propensity_scorer is a function, it either produces a propensity score for each run separately or all together (for correlated treatment assignment).

For individual runs, the function may include one or both of treated_run and untreated_run in the signature. For a whole population, use untreated_runs or treated_runs in the function signature. Both config and intervention can be passed as argument to access the simulator_config and intervention for the given experiment.

# self.propensity_scorer can be a constant
DynamicsExperiment(
    ...
    propensity_scorer=0.8),
    ...)

##
# Compute treatment assignment probability for each run separately.
##
def propensity_scorer(untreated_run):
    # Return propensity score for a single run, based on the
    # untreated rollout.
    return 0.5 if untreated_run[20].values()[0] > 0.2 else 0.1

def propensity_scorer(untreated_run, config, intervention):
    # You can pass the config and intervention
    intervention_covariate = untreated_run[intervention.year].values()[0]
    threshold = config.threshold
    return 1.0 if intervention_covariate > threshold else 0.0

def propensity_scorer(treated_run):
    # Return propensity score for a single run, based on treated run
    return 0.5 if treated_run[20].values()[0] > 0.2 else 0.1

def propensity_scorer(treated_run, untreated_run, config):
    # Assign treatment based on runs with both settings.
    if treated_run[10].values()[0] > 5 and untreated_run[10].values() < 10:
        return config.propensity
    return 1.0 - config.propensity

##
# Compute treatment assignment probability for all runs
# simultaneously. This allows for correlated propensity scores.
##
def propensity_scorer(untreated_runs):
    # Return propensity score for all runs, based on untreated run.
    # Only treat the top 10% of runs.
    covariates = [run.final_state.values()[0] for run in untreated_runs]
    top10 = np.argsort(covariates)[-10:]
    return [0.9 if idx in top10 else 0.1 for idx in range(len(untreated_runs))]

etc..

The outcome_extractor returns, for a given run, the outcome \(Y\). Both config and intervention can be passed as arguments. For instance,

def outcome_extractor(run):
    # Return the first state coordinate at time step 100
    return run[100].values()[0]

def outcome_extractor(run, config, intervention):
    # You can also specify one (or both) of config/intervention
    # First state coordinate 20 steps after intervention.
    return run[intervention.time + 20].values()[0]

Finally, the covariate_builder returns, for a given run, the observed covariates \(X\). If the signature includes runs, the method returns covariates for the entire sequence of runs. Otherwise, it produces covariates for a single run. Both config and intervention can be passed.

def covariate_builder(run):
    # Functions with the argument `run` return covariates for a
    # single rollout.
    return run.initial_state.values()

def covariate_builder(runs):
    # Functions with the argument `runs` return covariates for the
    # entire sequence of runs.
    return np.array([run.initial_state.value() for run in runs])

def covariate_builder(config, intervention, run):
    # Can optionally specify `config` or `intervention` in both cases.
    return run[intervention.year].values()
get_parameters()[source]

Inspect provided methods and gather parameters.

Returns:params – Collection of all of the parameters specified in the experiment.
Return type:whynot.framework.ParameterCollection
run(num_samples, seed=None, parallelize=True, show_progress=False, causal_graph=False, **parameter_args)[source]

Run a basic parameterized experiment on a dynamical system simulator.

Parameters:
  • num_samples (int) – Number of units to sample and simulate
  • show_progress (bool) – Should progress bar be shown?
  • parallelize (bool) – If true, the experiment class will run all of the simulations in parallel. Parallelization is performed with multiprocessing using a ProcessPool from the concurrent.futures module.
  • seed (int) – Random number generator seed used to set all internal randomness.
  • causal_graph – Whether to attempt to build the causal graph. Currently, this is only supported on the hiv, lotka_volterra, and opioid simulators.
  • **parameter_args – If the experiment is parameterized, additional arguments to select the parameters of a particular run.
Returns:

dataset – Dataset object encapsulating the covariates, treatment assignments, and outcomes observed in the experiment, as well as unit-level ground truth. The covariates are an array of size [num_samples, num_features] where num_features is determined by the covariate builder.

Return type:

whynot.framework.Dataset

Reinforcement learning

class whynot.gym.envs.ODEEnvBuilder(simulate_fn, config, action_space, observation_space, initial_state, intervention_fn, reward_fn, observation_fn=None, timestep=1.0)[source]

Environment builder for simulators derived from dynamical systems.

__init__(simulate_fn, config, action_space, observation_space, initial_state, intervention_fn, reward_fn, observation_fn=None, timestep=1.0)[source]

Initialize an environment class.

Parameters:
  • simulate_fn (Callable) – A function with signature simulate(initial_state, config, intervention=None, seed=None) -> whynot.dynamics.Run
  • config (whynot.dynamics.BaseConfig) – The base simulator configuration
  • action_space (whynot.gym.spaces.Space) – The action space for the reinforcement learner
  • observation_space (whynot.gym.spaces.Space) – The space of observations for the agent
  • initial_state (whynot.dynamics.BaseState) – The initial state of the simulator
  • intervention_fn (Callable) – A function that maps actions to simulator interventions with signature get_intervention(action, time) -> whynot.dynamics.BaseState
  • reward_fn (Callable) – A function that computes the cost/reward of taking an intervention in a particular state state with signature get_reward(intervention, state) -> float
  • observation_fn (Callable) – (Optional) A function that computes the observed state for the state of the simulator with signature observation_fn(state) -> np.ndarray. If ommitted, the entire simulator state is returned.
  • timestep (float) – Time between successive observations in the dynamical system.
reset()[source]

Reset the state.

seed(seed=None)[source]

Set internal randomness of the environment.

step(action)[source]

Perform one forward step in the environment.

Parameters:action (A numpy array reprsenting an action of shape) – [1, action_dim].
Returns:
  • observation (A numpy array of shape [1, obs_dim].)
  • reward (A numpy array of shape [1, 1].)
  • done (A numpy array of shape [1, 1])
  • info_dict (An empty dict.)

Framework

class whynot.framework.Dataset(covariates: numpy.ndarray, treatments: numpy.ndarray, outcomes: numpy.ndarray, true_effects: numpy.ndarray, causal_graph: Any = None)[source]

Observational dataset and grouth truth unit-level effects.

covariates

Float array of shape [num_samples, num_features] of covariates for each unit.

Type:np.ndarray
treatments

Integer 0/1 array of shape [num_samples] indicating treatment status for each unit. 1 indicates treated, 0 indicates unit not treated.

Type:np.ndarray
outcomes

Float array of shape [num_samples] containing the observed outcome for each unit.

Type:np.ndarray
true_effects

Float array of shape [num_samples] containing the unit-level treatment effects, \(Y_i(1) - Y_i(0)\) for each \(i\).

Type:np.ndarray
sate

Sample average treatment effect based on ground truth unit-level effects.

Type:float
causal_graph

If supported by the simulator and experiment, the causal graph associated with the data.

Type:networkx.DiGraph
class whynot.framework.ExperimentParameter(name: str, default: Any, values: Any = None, description: str = '')[source]

Container for a parameter to vary in an experiment.

name

Name of the parameter

description

Parameter description

default

Default (uninitialized) value of the parameter

values

Iterator of parameter values that supports sampling (for random search).

class whynot.framework.ParameterCollection(params=None)[source]

Lightweight wrapper class around a set of parameters.

Provides utility functions to support sampling and assigning subsets of the parameters.

Enforces name uniqueness. Every parameter should have a unique name.

Estimators

whynot.causal_suite(covariates, treatment, outcome, verbose=False)[source]

Run a collection of causal inference algorithms on the observational dataset.

By default, the suite only runs estimators implemented in Python
  • Ordinary least squares (ols)
  • Propensity score matching
  • Propensity weighted least squares
Depending on the estimators installed in whynot_estimators, the suite additionally runs:
  • An IP weighting estimator (ip_weighting)
  • Matching estimators with mahalanobis distance metrics
  • Causal Forest (causal_forest)
  • TMLE (tmle)
Parameters:
  • covariates (np.ndarray) – Array of shape num_samples x num_features.
  • treatment (np.ndarray) – Boolean array of shape [num_samples] indicating treatment status for each sample.
  • outcome (np.ndarray) – Array of shape [num_sample] containing the observed outcome for each sample.
  • verbose (bool) – If True, print incremental messages as estimators are executed.
Returns:

results – Dictionary with keys denoting the name of each method and values the corresponding InferenceResult:

results[‘causal_forest’] -> inference_results

Return type:

dict

class whynot.framework.InferenceResult(ate: float = None, stderr: float = None, ci: tuple = (None, None), individual_effects: numpy.ndarray = None, elapsed_time: float = None)[source]

Object to store results of causal inference method.

ate

Estimated average treatment effect

stderr

Reported standard error of the ATE estimate. Only available if supported by the method, otherwise None.

ci

Reported 95% confidence interval for the ATE (lower_bound, upper_bound). Only available if supported by the method, otherwise None. TODO: Ideally, we’d support various significance levels.

individual_effects

Heterogeneous treatment effect for each unit. Only available if supported by the method, otherwise None.

elapsed_time

How long in second (wall-clock time) it took to produce the estimate.