"""
Practitioner guidance for Difference-in-Differences analysis.
Implements Baker et al. (2025) "Difference-in-Differences Designs:
A Practitioner's Guide" as context-aware runtime guidance. Call
``practitioner_next_steps(results)`` after estimation to get a
structured set of recommended next steps.
"""
import math
from typing import Any, Dict, List, Optional, Set
# ---------------------------------------------------------------------------
# Valid step names (Baker et al. 8-step framework)
# ---------------------------------------------------------------------------
STEPS: Set[str] = {
"target_parameter",
"assumptions",
"parallel_trends",
"estimator_selection",
"estimation",
"sensitivity",
"heterogeneity",
"robustness",
}
# ---------------------------------------------------------------------------
# Estimator name mapping
# ---------------------------------------------------------------------------
_ESTIMATOR_NAMES: Dict[str, str] = {
"DiDResults": "DifferenceInDifferences",
"MultiPeriodDiDResults": "MultiPeriodDiD (Event Study)",
"CallawaySantAnnaResults": "CallawaySantAnna",
"SunAbrahamResults": "SunAbraham",
"ImputationDiDResults": "ImputationDiD (Borusyak-Jaravel-Spiess)",
"TwoStageDiDResults": "TwoStageDiD (Gardner)",
"StackedDiDResults": "StackedDiD",
"SyntheticDiDResults": "SyntheticDiD",
"TROPResults": "TROP",
"EfficientDiDResults": "EfficientDiD",
"ContinuousDiDResults": "ContinuousDiD",
"TripleDifferenceResults": "TripleDifference (DDD)",
"BaconDecompositionResults": "BaconDecomposition",
"HeterogeneousAdoptionDiDResults": "HeterogeneousAdoptionDiD (HAD)",
"HeterogeneousAdoptionDiDEventStudyResults": "HeterogeneousAdoptionDiD (Event Study)",
}
# ---------------------------------------------------------------------------
# Public API
# ---------------------------------------------------------------------------
def practitioner_next_steps(
results: Any,
*,
completed_steps: Optional[List[str]] = None,
verbose: bool = True,
) -> Dict[str, Any]:
"""
Context-aware practitioner guidance based on Baker et al. (2025).
Inspects the type and attributes of *results* to recommend which
Baker et al. steps remain. Returns a structured dict and optionally
prints a human-readable summary.
Parameters
----------
results : Any
A diff-diff results object (e.g. ``DiDResults``,
``CallawaySantAnnaResults``, etc.).
completed_steps : list of str, optional
Steps the caller has already completed. Valid names:
``"target_parameter"``, ``"assumptions"``, ``"parallel_trends"``,
``"estimator_selection"``, ``"estimation"``, ``"sensitivity"``,
``"heterogeneity"``, ``"robustness"``.
verbose : bool, default True
If True, print a human-readable summary to stdout.
Returns
-------
dict
Keys: ``"estimator"`` (str), ``"completed"`` (list of str),
``"next_steps"`` (list of dict), ``"warnings"`` (list of str).
Each next_step dict has: ``"baker_step"`` (int), ``"label"`` (str),
``"why"`` (str), ``"code"`` (str), ``"priority"`` (str).
"""
completed = set(completed_steps or [])
unknown = completed - STEPS
if unknown:
raise ValueError(f"Unknown step names: {unknown}. Valid names: {sorted(STEPS)}")
# Estimation is always complete if we have a results object
completed.add("estimation")
type_name = type(results).__name__
handler = _HANDLERS.get(type_name, _handle_generic)
steps, warnings = handler(results)
# Prepend Steps 1-2 (pre-estimation reasoning) to every handler's output.
# These are always relevant and filterable via completed_steps.
pre_estimation = [
_step(
baker_step=1,
label="Define target parameter",
why=(
"State explicitly what causal effect you are estimating "
"(ATT, ATT(g,t), weighted/unweighted) and what policy "
"question it answers."
),
code="# What is the target parameter? ATT? Weighted or unweighted?",
priority="high",
step_name="target_parameter",
),
_step(
baker_step=2,
label="State identification assumptions",
why=(
"Name the parallel trends variant you are invoking "
"(unconditional, conditional, PT-GT-NYT, etc.), the "
"no-anticipation assumption, and any overlap conditions."
),
code="# Which PT variant? No-anticipation? Overlap?",
priority="high",
step_name="assumptions",
),
]
steps = pre_estimation + steps
# Filter out completed steps
steps = _filter_steps(steps, completed)
output = {
"estimator": _ESTIMATOR_NAMES.get(type_name, type_name),
"completed": sorted(completed),
"next_steps": steps,
"warnings": warnings,
}
if verbose:
_print_output(output)
return output
# ---------------------------------------------------------------------------
# Step builder helper
# ---------------------------------------------------------------------------
def _step(
baker_step: int,
label: str,
why: str,
code: str,
priority: str = "high",
step_name: str = "",
) -> Dict[str, Any]:
return {
"baker_step": baker_step,
"label": label,
"why": why,
"code": code,
"priority": priority,
"_step_name": step_name,
}
# ---------------------------------------------------------------------------
# Common steps reused across handlers
# ---------------------------------------------------------------------------
def _parallel_trends_step(staggered: bool = False) -> Dict[str, Any]:
if staggered:
return _step(
baker_step=3,
label="Test parallel trends (event-study pre-periods)",
why=(
"For staggered designs, inspect event-study pre-period "
"coefficients rather than the generic check_parallel_trends() "
"which assumes a single binary treatment with universal "
"pre-periods. Pre-treatment ATTs should be near zero. "
"Use CS with aggregate='event_study' or check the estimator's "
"event-study output directly."
),
code=(
"# Inspect pre-treatment event-study coefficients:\n"
"# (available after fitting with event-study aggregation)\n"
"# Pre-period effects should be near zero and insignificant."
),
step_name="parallel_trends",
)
return _step(
baker_step=3,
label="Test parallel trends assumption",
why=(
"Parallel trends is the core identifying assumption. "
"Insignificant pre-trends do NOT prove it holds. For "
"MultiPeriodDiD or CS results, use HonestDiD to bound "
"the impact of violations."
),
code=(
"from diff_diff import check_parallel_trends\n"
"pt = check_parallel_trends(data, outcome='y', time='period',\n"
" treatment_group='treated')"
),
step_name="parallel_trends",
)
def _honest_did_step() -> Dict[str, Any]:
return _step(
baker_step=6,
label="Run HonestDiD sensitivity analysis",
why=(
"Bounds the treatment effect under plausible violations of "
"parallel trends. Essential for assessing result robustness."
),
code=(
"from diff_diff import compute_honest_did\n"
"honest = compute_honest_did(results, method='relative_magnitude', M=1.0)\n"
"print(honest.summary())"
),
step_name="sensitivity",
)
def _placebo_step() -> Dict[str, Any]:
"""Placebo tests for simple 2x2 DiD designs only."""
return _step(
baker_step=6,
label="Run placebo tests",
why=(
"Falsification tests using fake timing, permutation, and "
"leave-one-out diagnostics to probe assumption validity."
),
code=(
"from diff_diff import run_all_placebo_tests\n"
"# Requires binary time indicator (post=0/1), not multi-period:\n"
"placebo = run_all_placebo_tests(\n"
" data, outcome='y', treatment='treated', time='post',\n"
" unit='unit_id', pre_periods=[0], post_periods=[1],\n"
" n_permutations=500, seed=42)"
),
priority="medium",
step_name="sensitivity",
)
def _robustness_compare_step(alternatives: str) -> Dict[str, Any]:
return _step(
baker_step=8,
label=f"Compare with alternative estimators ({alternatives})",
why=(
"Agreement across estimators with different assumptions "
"strengthens conclusions. Disagreement reveals sensitivity."
),
code=(
f"# Re-estimate with {alternatives} and compare ATT, SE, CI\n"
f"# If results agree, confidence increases.\n"
f"# If they disagree, investigate which assumptions differ."
),
step_name="robustness",
)
def _covariates_step() -> Dict[str, Any]:
return _step(
baker_step=8,
label="Report with and without covariates",
why=(
"Shows whether results are sensitive to covariate conditioning. "
"Large shifts suggest covariates are driving identification."
),
code=(
"# Re-estimate without covariates and compare:\n"
"result_no_cov = estimator.fit(data, ..., covariates=None)\n"
"# Compare ATT with and without covariates.\n"
"# Use .att (basic DiD) or .overall_att (staggered estimators)."
),
priority="medium",
step_name="robustness",
)
# ---------------------------------------------------------------------------
# Per-type handlers each returns (steps, warnings)
# ---------------------------------------------------------------------------
def _handle_did(results: Any):
steps = [
_step(
baker_step=3,
label="Test parallel trends assumption",
why=(
"Parallel trends is the core identifying assumption. "
"Insignificant pre-trends do NOT prove it holds."
),
code=(
"from diff_diff import check_parallel_trends\n"
"pt = check_parallel_trends(data, outcome='y', time='period',\n"
" treatment_group='treated')"
),
step_name="parallel_trends",
),
_placebo_step(), # valid: basic 2x2 DiD with binary time
_step(
baker_step=4,
label="Check if data is actually staggered",
why=(
"If treatment timing varies across units, basic DiD produces "
"biased estimates. Use CallawaySantAnna or another "
"heterogeneity-robust estimator instead."
),
code=(
"# Check if there are multiple treatment cohorts:\n"
"print(data.groupby('unit')['treatment_date'].first().nunique())\n"
"# If > 1 cohort, switch to CallawaySantAnna"
),
step_name="estimator_selection",
),
]
warnings = _check_nan_att(results)
return steps, warnings
def _handle_multi_period(results: Any):
steps = [
_parallel_trends_step(),
_honest_did_step(),
# Note: run_all_placebo_tests() requires binary time indicator,
# which MultiPeriodDiD does not use. Omit placebo for this type.
_robustness_compare_step("CS, SA, or BJS"),
]
warnings = _check_nan_att(results)
return steps, warnings
def _handle_cs(results: Any):
steps = [
_parallel_trends_step(staggered=True),
_step(
baker_step=6,
label="Run HonestDiD sensitivity analysis",
why=(
"Bounds the treatment effect under plausible violations of "
"parallel trends. Requires event study effects refit with "
"aggregate='event_study' or 'all' if not already done."
),
code=(
"from diff_diff import compute_honest_did\n"
"# CS results must have event_study_effects:\n"
"results = cs.fit(data, ..., aggregate='event_study')\n"
"honest = compute_honest_did(results, method='relative_magnitude', M=1.0)\n"
"print(honest.summary())"
),
step_name="sensitivity",
),
_step(
baker_step=7,
label="Examine group and event study effects",
why=(
"Aggregate ATT may mask heterogeneity across cohorts or "
"dynamic effects over time. Inspect group and event study "
"aggregations."
),
code=(
"# Re-fit with aggregate='all' to get all aggregations:\n"
"results = cs.fit(data, ..., aggregate='all')\n"
"print(results.group_effects) # Per-cohort ATTs\n"
"print(results.event_study_effects) # Dynamic effects"
),
step_name="heterogeneity",
),
_robustness_compare_step("SA, BJS, or Gardner"),
_covariates_step(),
]
warnings = _check_nan_att(results)
return steps, warnings
def _handle_sa(results: Any):
steps = [
_parallel_trends_step(staggered=True),
_step(
baker_step=6,
label="Specification-based falsification",
why=(
"Compare results across control group definitions "
"(never_treated vs not_yet_treated) and anticipation "
"settings to assess robustness."
),
code=(
"# Re-estimate with different control group / anticipation:\n"
"# sa_alt = SunAbraham(control_group='not_yet_treated')"
),
priority="medium",
# DR's sensitivity section runs HonestDiD, not specification
# variation; tagging this as ``sensitivity`` caused
# ``_collect_next_steps`` to suppress it after HonestDiD ran.
# Use ``specification_comparison`` so the recommendation
# persists alongside a completed HonestDiD sensitivity check.
step_name="specification_comparison",
),
_step(
baker_step=7,
label="Examine event-study and cohort effects",
why=(
"SunAbraham results include event_study_effects (dynamic "
"effects by relative period) and cohort_effects (per-cohort "
"effects). Note: SA does not have an aggregate parameter "
"these are computed automatically during fit()."
),
code=(
"# SA event-study effects:\n"
"sa_es_df = results.to_dataframe(level='event_study')\n"
"# SA cohort effects:\n"
"sa_cohort_df = results.to_dataframe(level='cohort')"
),
step_name="heterogeneity",
),
_robustness_compare_step("CS, BJS, or Gardner"),
_covariates_step(),
]
warnings = _check_nan_att(results)
return steps, warnings
def _handle_imputation(results: Any):
steps = [
_parallel_trends_step(staggered=True),
_step(
baker_step=6,
label="Specification-based falsification",
why=(
"ImputationDiD does not have a control_group parameter. "
"Compare results with and without covariates, vary the "
"sample (drop cohorts), and compare with CS/SA as "
"falsification checks."
),
code=(
"# Compare with alternative estimators as robustness:\n"
"# Leave-one-cohort-out sensitivity analysis"
),
priority="medium",
# See note on SA handler: DR completes ``sensitivity`` when
# HonestDiD runs, which is unrelated to this specification-
# variation recommendation. Tag separately.
step_name="specification_comparison",
),
_robustness_compare_step("CS, SA, or Gardner"),
_covariates_step(),
]
warnings = _check_nan_att(results)
return steps, warnings
def _handle_two_stage(results: Any):
steps = [
_parallel_trends_step(staggered=True),
_step(
baker_step=6,
label="Specification-based falsification",
why=(
"TwoStageDiD does not have a control_group parameter. "
"Compare results with and without covariates, vary the "
"sample (drop cohorts), and compare with CS/SA as "
"falsification checks."
),
code=(
"# Compare with alternative estimators as robustness:\n"
"# Leave-one-cohort-out sensitivity analysis"
),
priority="medium",
# See note on SA handler: DR completes ``sensitivity`` when
# HonestDiD runs, which is unrelated to this specification-
# variation recommendation. Tag separately.
step_name="specification_comparison",
),
_robustness_compare_step("CS, BJS, or SA"),
_covariates_step(),
]
warnings = _check_nan_att(results)
return steps, warnings
def _handle_stacked(results: Any):
steps = [
_parallel_trends_step(staggered=True),
_step(
baker_step=6,
label="Vary clean control definition",
why=(
"StackedDiD uses clean_control parameter (not control_group). "
"Compare results with different clean control definitions "
"and event window widths as falsification."
),
code=(
"# Re-estimate with different clean_control settings:\n"
"# stacked_alt = StackedDiD(clean_control='not_yet_treated')"
),
priority="medium",
# See note on SA handler: DR completes ``sensitivity`` when
# HonestDiD runs, which does not replay ``clean_control``
# variation. Tag separately.
step_name="specification_comparison",
),
_step(
baker_step=7,
label="Check sub-experiment balance",
why=(
"Stacked DiD constructs sub-experiments for each cohort. "
"Verify that each sub-experiment has sufficient controls."
),
code="# Check results.n_sub_experiments and inspect results.stacked_data",
priority="medium",
step_name="heterogeneity",
),
_robustness_compare_step("CS, SA, or BJS"),
]
warnings = _check_nan_att(results)
return steps, warnings
def _handle_synthetic(results: Any):
steps = [
_step(
baker_step=6,
label="Check pre-treatment fit and weight concentration",
why=(
"Synthetic DiD relies on pre-treatment fit to construct "
"weights. Poor fit or highly concentrated unit weights "
"suggest the synthetic control may not approximate the "
"counterfactual well."
),
code=(
"print(f'Pre-treatment fit (RMSE): {results.pre_treatment_fit:.4f}')\n"
"concentration = results.get_weight_concentration()\n"
"print(f\"Effective N: {concentration['effective_n']:.1f}\")\n"
"print(f\"Top-5 weight share: {concentration['top_k_share']:.2%}\")"
),
step_name="sensitivity",
),
_step(
baker_step=6,
label="In-time placebo",
why=(
"Re-estimate on shifted fake treatment dates in the "
"pre-period. A credible design yields near-zero placebo "
"ATTs departures signal that something is being picked "
"up pre-treatment, weakening the causal interpretation."
),
code=("placebo_df = results.in_time_placebo()\n" "print(placebo_df)"),
priority="medium",
step_name="sensitivity",
),
_step(
baker_step=6,
label="Leave-one-out influence (jackknife)",
why=(
"If the estimate is driven by a single unit, robustness "
"is weak. Fit with variance_method='jackknife' and inspect "
"which units move the ATT the most."
),
code=(
"# Requires variance_method='jackknife' AND enough support for LOO\n"
"# (n_treated >= 2 and >= 2 effective-weight controls).\n"
"if getattr(results, '_loo_unit_ids', None) is not None:\n"
" loo_df = results.get_loo_effects_df()\n"
" print(loo_df.head(10))\n"
"else:\n"
" print('LOO not available - re-fit with '\n"
" 'variance_method=\"jackknife\" and ensure >=2 treated units '\n"
" 'with positive effective support.')"
),
priority="medium",
# DR's SyntheticDiD native battery covers pre-treatment fit,
# weight concentration, in-time placebo, and zeta-omega
# sensitivity, but NOT the jackknife LOO workflow (which
# requires a separate ``variance_method='jackknife'`` fit
# via ``get_loo_effects_df``). Tagging this recommendation
# as ``sensitivity`` caused ``_collect_next_steps`` to
# suppress it as soon as the native block ran, even though
# the jackknife was never executed. Round-24 P2 CI review
# on PR #318; same class as round-20 Hausman mistag.
step_name="loo_jackknife",
),
_step(
baker_step=6,
label="Regularization sensitivity (zeta_omega)",
why=(
"The unit-weight regularization is auto-selected from "
"data. Show whether the ATT moves materially across a "
"grid of values to gauge robustness to this choice."
),
code=("sens_df = results.sensitivity_to_zeta_omega()\n" "print(sens_df)"),
priority="low",
step_name="sensitivity",
),
_step(
baker_step=8,
label="Compare with staggered estimators (CS, SA)",
why=(
"SyntheticDiD is for few treated units; compare with "
"staggered estimators if applicable. Use TROP only if "
"factor confounding is suspected (different use case)."
),
code=(
"from diff_diff import CallawaySantAnna\n"
"cs = CallawaySantAnna()\n"
"cs_result = cs.fit(data, ...)\n"
"print(f'SDiD ATT: {results.att:.4f}, CS ATT: {cs_result.overall_att:.4f}')"
),
step_name="robustness",
),
]
warnings = _check_nan_att(results)
return steps, warnings
def _handle_trop(results: Any):
steps = [
_step(
baker_step=6,
label="Verify factor structure assumptions",
why=(
"TROP assumes an approximate factor model for untreated "
"potential outcomes. If the factor structure is misspecified, "
"estimates may be biased."
),
code=(
"# Check LOOCV-selected number of factors:\n"
"# Compare with SyntheticDiD as a robustness check"
),
step_name="sensitivity",
),
_step(
baker_step=6,
label="In-time or in-space placebo",
why=(
"Test robustness by re-estimating on a placebo treatment "
"period or dropping treated units one at a time. These "
"are the natural falsification checks for factor-model "
"panel estimators."
),
code=(
"# In-time placebo: re-estimate with a fake treatment date\n"
"# Leave-one-out: drop each treated unit and re-estimate"
),
priority="medium",
# TROP's estimator-native diagnostics surface factor-model fit
# metrics, not in-time or in-space placebos; DR does not run
# placebos on TROP. Tag separately from ``sensitivity`` so the
# recommendation persists after DR marks the TROP native
# battery complete.
step_name="placebo",
),
_robustness_compare_step("SyntheticDiD or CS"),
]
warnings = _check_nan_att(results)
return steps, warnings
def _handle_efficient(results: Any):
steps = [
_parallel_trends_step(staggered=True),
_step(
baker_step=6,
label="Compare control group definitions",
why=(
"EfficientDiD supports never_treated and last_cohort "
"control groups (not not_yet_treated). Compare results "
"across both to assess robustness."
),
code=(
"# Re-estimate with alternative control group:\n"
"# edid_alt = EfficientDiD(control_group='last_cohort')"
),
priority="medium",
# See note on SA handler: DR completes ``sensitivity`` when
# HonestDiD runs, which does not re-estimate with an
# alternative control_group. Tag separately so this
# recommendation persists alongside a completed HonestDiD
# block.
step_name="specification_comparison",
),
_step(
baker_step=7,
label="Run Hausman pretest (PT-All vs PT-Post)",
why=(
"EfficientDiD supports both PT-All and PT-Post assumptions. "
"The Hausman pretest compares them report which was selected."
),
code=(
"# Hausman pretest is a classmethod on the estimator:\n"
"from diff_diff import EfficientDiD\n"
"pretest = EfficientDiD.hausman_pretest(\n"
" data, outcome='y', unit='id', time='t', first_treat='g')"
),
# The Hausman pretest is a parallel-trends diagnostic per
# REGISTRY.md EfficientDiD: it tests whether the stronger
# PT-All regime is tenable relative to PT-Post. ``DiagnosticReport``
# treats a ran Hausman block as ``parallel_trends`` completion
# (``_check_pt_hausman``), so tagging this practitioner step as
# ``parallel_trends`` keeps ``_collect_next_steps()`` from
# recommending a check the report already executed. Round-20 P2
# CI review on PR #318 flagged the earlier ``heterogeneity`` tag
# as a mismatched-step-name bug.
step_name="parallel_trends",
),
_robustness_compare_step("CS, SA, or BJS"),
_covariates_step(),
]
warnings = _check_nan_att(results)
return steps, warnings
def _handle_continuous(results: Any):
steps = [
_step(
baker_step=3,
label="Assess parallel trends for continuous treatment",
why=(
"ContinuousDiD has dose-specific parallel trends assumptions "
"(PT/SPT) that differ from the binary treatment case. No "
"built-in formal test exists; inspect dose-specific "
"pre-treatment outcome trends across dose groups manually."
),
code=(
"# No built-in formal PT test for continuous treatment.\n"
"# Inspect pre-treatment outcome trends by dose group."
),
step_name="parallel_trends",
),
_step(
baker_step=4,
label="Switch to HeterogeneousAdoptionDiD if no untreated units",
why=(
"ContinuousDiD's identification assumes a never-treated "
"comparison group exists (units with dose = 0). When every "
"unit is treated at some positive dose level a universal "
"rollout where treatment varies in intensity, not status "
"use HeterogeneousAdoptionDiD instead. HAD identifies a "
"Weighted Average Slope (WAS) at the dose support boundary "
"by leveraging dose variation across units."
),
code=(
"# If your panel has no units with first_treat == 0, switch:\n"
"from diff_diff import HeterogeneousAdoptionDiD\n"
"had = HeterogeneousAdoptionDiD()\n"
"had_results = had.fit(\n"
" data, outcome_col='y', unit_col='unit',\n"
" time_col='t', dose_col='d', first_treat_col='first_treat')"
),
step_name="estimator_selection",
),
_step(
baker_step=7,
label="Plot dose-response curve",
why=(
"Continuous DiD estimates treatment effects at each dose "
"level. The dose-response curve reveals the functional form "
"of the treatment-dose relationship."
),
code=("from diff_diff import plot_dose_response\n" "plot_dose_response(results)"),
step_name="heterogeneity",
),
_step(
baker_step=6,
label="Check dose distribution",
why=(
"Sparse regions of the dose distribution produce imprecise "
"estimates. Verify sufficient support across dose values."
),
code="# Inspect the distribution of treatment doses in your data",
priority="medium",
step_name="sensitivity",
),
]
warnings = _check_nan_att(results)
return steps, warnings
def _handle_triple(results: Any):
steps = [
_step(
baker_step=3,
label="Assess DDD identifying assumption",
why=(
"DDD identification is weaker than requiring separate "
"parallel trends for two DiDs it allows group-specific "
"and partition-specific PT violations as long as they "
"cancel in the triple difference. No built-in formal "
"test exists; inspect pre-treatment outcome patterns "
"across the treatment/eligibility/time cells."
),
code=(
"# No built-in formal DDD assumption test.\n"
"# Inspect pre-treatment means across treatment x eligibility\n"
"# cells to assess whether the DDD structure is plausible."
),
step_name="parallel_trends",
),
_step(
baker_step=7,
label="Test placebo group",
why=(
"Re-estimate using a placebo eligibility group to check "
"whether the DDD result could be an artifact of the "
"group structure rather than the treatment."
),
code="# Re-estimate with a placebo eligibility group",
step_name="heterogeneity",
),
_covariates_step(),
]
warnings = _check_nan_att(results)
return steps, warnings
def _handle_bacon(results: Any):
steps = [
_step(
baker_step=4,
label="Switch to heterogeneity-robust estimator",
why=(
"Bacon decomposition is diagnostic, not an estimator. "
"If substantial weight falls on 'later vs earlier' "
"comparisons, TWFE is biased. Use CS, SA, BJS, or another "
"heterogeneity-robust estimator for causal estimates."
),
code=(
"from diff_diff import CallawaySantAnna\n"
"cs = CallawaySantAnna(control_group='never_treated',\n"
" estimation_method='dr')\n"
"results = cs.fit(data, ...)"
),
step_name="estimator_selection",
),
]
warnings = []
# Check for forbidden comparisons (later vs earlier treated)
weight = getattr(results, "total_weight_later_vs_earlier", 0)
if isinstance(weight, (int, float)) and weight > 0.01:
warnings.append(
f"Forbidden comparisons (later vs earlier treated) carry "
f"{weight:.0%} of TWFE weight TWFE estimate is contaminated. "
f"Switch to a heterogeneity-robust estimator."
)
return steps, warnings
def _handle_had(results: Any):
"""HeterogeneousAdoptionDiD single-period guidance.
Five Baker et al. steps (3, 4, 6, 7, 8). HAD's design absence is
"no untreated unit" - comparison comes from dose variation across
units, not from an untreated holdout. Treatment varies in intensity,
not in status.
"""
steps = [
_step(
baker_step=3,
label="Run the HAD pretest battery",
why=(
"On a two-period unweighted panel did_had_pretest_workflow "
"runs paper Section 4.2 step 1 (QUG support-infimum test - "
"decides Design 1' vs Design 1) and step 3 (Stute / "
"Yatchew-HR Assumption 8 linearity tests). Step 2 "
"(Assumption 7 pre-trends) is NOT covered on the overall "
"path - a single pre-period cannot support the joint "
"Stute variant - and the returned verdict explicitly "
"flags that gap. To close step 2, refit on a multi-period "
"panel with aggregate='event_study' AND verify the panel "
"has at least one earlier placebo pre-period beyond F-1; "
"if only the base pre-period F-1 is available, the "
"workflow still sets pretrends_joint=None, all_pass=False, "
"and a 'joint pre-trends skipped (no earlier pre-period)' "
"verdict suffix - in that case step 2 stays uncovered "
"even on the event-study path. On supported survey-weighted "
"fits (pweight + PSU/FPC under survey_design= / survey= / "
"weights=) the workflow skips QUG with a UserWarning "
"(permanent Phase 4.5 C0 deferral - extreme order statistics "
"are not smooth functionals of the empirical CDF) and returns "
"a linearity-conditional verdict only - so step 1 coverage "
"is unweighted-only and the reported verdict on supported "
"weighted fits is conditional on QUG holding by assumption. "
"Stratified (SurveyDesign(strata=...)) and replicate-weight "
"(BRR/Fay/JK1/JKn/SDR) designs raise NotImplementedError on "
"the linearity kernels and have no pretest workflow path "
"yet - deferred to a follow-up. "
"Assumptions 3 / 5 / 6 (uniform continuity at the "
"boundary, Design 1 sign / WAS_d_lower identification) "
"are NOT testable via pre-trends - the workflow vets only "
"what can be vetted."
),
code=(
"from diff_diff import did_had_pretest_workflow\n"
"report = did_had_pretest_workflow(\n"
" data, outcome_col='y', unit_col='unit',\n"
" time_col='t', dose_col='d',\n"
" first_treat_col='first_treat')\n"
"print(report.summary())\n"
"# verdict explicitly flags the Assumption 7 gap on the\n"
"# overall path; aggregate='event_study' on a multi-period\n"
"# panel adds joint Stute pre-trends + joint homogeneity-linearity.\n"
"# Passing survey_design= / weights= skips QUG (Phase 4.5 C0)\n"
"# and returns a linearity-conditional verdict only."
),
step_name="parallel_trends",
),
_step(
baker_step=4,
label="Confirm WAS is the target estimand (vs ATT(d) for ContinuousDiD)",
why=(
"HAD targets WAS (Weighted Average Slope) at the dose "
"support boundary. If you specifically want per-dose "
"ATT(d) / ACRT(d) dose-response curves AND your panel "
"has never-treated controls (units with first_treat == 0), "
"ContinuousDiD is the alternative different estimand, "
"and ContinuousDiD's identification requires never-treated "
"controls. HAD itself remains valid even with a small "
"share of never-treated units (paper compatibility; see "
"REGISTRY HeterogeneousAdoptionDiD edge cases "
"Garrett et al. 2020 retained 12 untreated counties out "
"of 2,954). The choice is about estimand, not about "
"whether untreated units exist."
),
code=(
"# HAD reports WAS at the dose support boundary.\n"
"# If you instead want per-dose ATT(d)/ACRT(d) dose-response\n"
"# curves AND the panel has never-treated controls:\n"
"from diff_diff import ContinuousDiD\n"
"cdid = ContinuousDiD()\n"
"cdid_results = cdid.fit(\n"
" data, outcome='y', unit='unit', time='t',\n"
" first_treat='first_treat', dose='d',\n"
" aggregate='dose')"
),
step_name="estimator_selection",
),
_step(
baker_step=6,
label="Inspect bandwidth diagnostics (continuous designs)",
why=(
"Continuous-dose designs (continuous_at_zero / "
"continuous_near_d_lower) use an MSE-DPI bandwidth selector "
"for the bias-corrected local-linear estimator. Bandwidth "
"choice affects WAS - verify the selector landed on a "
"viable bandwidth (not boundary-clipped or near-degenerate). "
"results.bandwidth_diagnostics is None on the mass_point "
"design (parametric, no bandwidth)."
),
code=(
"# Inspect the auto-selected bandwidths:\n"
"results.bandwidth_diagnostics # None on mass_point"
),
priority="medium",
step_name="sensitivity",
),
_step(
baker_step=7,
label="Re-fit with aggregate='event_study' for per-horizon WAS",
why=(
"On multi-period panels, the event-study aggregate returns "
"per-event-time WAS estimates instead of a single scalar. "
"Reveals whether dose response grows, decays, or stabilizes "
"across post-treatment horizons. Pre-period placebos serve "
"as a parallel-trends sanity check."
),
code=(
"from diff_diff import HeterogeneousAdoptionDiD\n"
"est = HeterogeneousAdoptionDiD()\n"
"es = est.fit(\n"
" data, outcome_col='y', unit_col='unit',\n"
" time_col='t', dose_col='d',\n"
" first_treat_col='first_treat',\n"
" aggregate='event_study')"
),
priority="medium",
step_name="heterogeneity",
),
_step(
baker_step=8,
label="Verify design auto-detection with explicit design=",
why=(
"design='auto' picks one of {continuous_at_zero, "
"continuous_near_d_lower, mass_point} from the dose "
"support. Re-fit with an explicit design= to verify the "
"auto-detection matched your panel structure - WAS vs "
"WAS_d_lower target parameters, and the bias-corrected "
"local-linear vs 2SLS estimation paths, differ in "
"interpretation."
),
code=(
"# Refit with each candidate design and compare:\n"
"from diff_diff import HeterogeneousAdoptionDiD\n"
"for d in ['continuous_at_zero', 'continuous_near_d_lower',\n"
" 'mass_point']:\n"
" try:\n"
" alt = HeterogeneousAdoptionDiD(design=d).fit(...)\n"
" print(d, alt.att, alt.target_parameter)\n"
" except Exception as e:\n"
" print(d, 'not applicable:', e)"
),
priority="medium",
step_name="robustness",
),
]
warnings = _check_nan_att(results)
return steps, warnings
def _handle_had_event_study(results: Any):
"""HeterogeneousAdoptionDiD event-study guidance.
Five Baker et al. steps (3, 4, 6, 7, 8). Same framing convention as
_handle_had: "no untreated unit", dose variation, treatment varies
in intensity not status.
"""
steps = [
_step(
baker_step=3,
label="Run the HAD pretest battery (event-study mode)",
why=(
"On multi-period unweighted panels, did_had_pretest_workflow "
"with aggregate='event_study' runs QUG plus joint Stute "
"pre-trends plus joint homogeneity-linearity Stute. The "
"joint Stute pre-trends variant closes the paper Section "
"4.2 step-2 gap ONLY IF the panel carries at least one "
"earlier placebo pre-period beyond the base F-1. With "
"only the base F-1 pre-period present (e.g. a minimal "
"valid 3-period event-study fit, or a 4-period fit under "
"trends_lin=True where the consumed F-2 placebo gets "
"dropped), pretrends_joint=None, all_pass=False, and the "
"verdict carries 'joint pre-trends skipped (no earlier "
"pre-period)' - step 2 stays uncovered. On supported "
"survey-weighted fits (pweight + PSU/FPC under "
"survey_design= / survey= / weights=) the workflow skips "
"QUG with a UserWarning (permanent Phase 4.5 C0 deferral) "
"and returns a linearity-conditional verdict only - so "
"step 1 coverage is unweighted-only on the event-study "
"path too, and the weighted verdict is conditional on QUG "
"holding by assumption. Stratified (SurveyDesign("
"strata=...)) and replicate-weight (BRR/Fay/JK1/JKn/SDR) "
"designs raise NotImplementedError on the linearity "
"kernels and have no pretest workflow path on the "
"event-study path yet - deferred to a follow-up. "
"The joint Stute pre-trends and joint "
"homogeneity-linearity tests themselves remain available "
"under supported survey weighting via PSU-level Mammen "
"multiplier bootstrap."
),
code=(
"from diff_diff import did_had_pretest_workflow\n"
"report = did_had_pretest_workflow(\n"
" data, outcome_col='y', unit_col='unit',\n"
" time_col='t', dose_col='d',\n"
" first_treat_col='first_treat',\n"
" aggregate='event_study')\n"
"print(report.summary())"
),
step_name="parallel_trends",
),
_step(
baker_step=4,
label="Confirm WAS is the target estimand (vs ATT(d) for ContinuousDiD)",
why=(
"HAD targets per-event-time WAS at the dose support "
"boundary. If you instead want per-dose ATT(d) / ACRT(d) "
"dose-response curves AND your panel has never-treated "
"controls, ContinuousDiD(aggregate='eventstudy') is the "
"alternative different estimand, requires never-treated. "
"HAD itself remains valid even with a small share of "
"never-treated units (paper compatibility); on staggered "
"panels HAD's last-cohort filter explicitly RETAINS "
"never-treated units as the untreated-group comparison "
"(paper Appendix B.2). The choice is about estimand."
),
code=(
"# HAD reports per-event-time WAS at the dose boundary.\n"
"# If you instead want per-dose ATT(d)/ACRT(d) event-study\n"
"# curves AND the panel has never-treated controls:\n"
"from diff_diff import ContinuousDiD\n"
"cdid = ContinuousDiD()\n"
"cdid_es = cdid.fit(\n"
" data, outcome='y', unit='unit', time='t',\n"
" first_treat='first_treat', dose='d',\n"
" aggregate='eventstudy')"
),
step_name="estimator_selection",
),
_step(
baker_step=6,
label="Use simultaneous (sup-t) confidence bands when reading multiple horizons",
why=(
"Pointwise CIs over-reject when you read multiple horizons "
"as a joint pattern. On weighted fits (survey_design= or "
"weights=), fit(cband=True) constructs simultaneous (sup-t) "
"bands across horizons via multiplier bootstrap. "
"results.cband_low / results.cband_high give the band "
"endpoints; results.cband_crit_value reports the sup-t "
"critical value used."
),
code=(
"from diff_diff import HeterogeneousAdoptionDiD, SurveyDesign\n"
"# Construct your survey design (adapt to your data):\n"
"sd = SurveyDesign(weights='weight_col')\n"
"# vcov_type='hc1' is REQUIRED on the mass-point design under\n"
"# survey_design= (the default classical sandwich raises\n"
"# NotImplementedError on the survey path because the\n"
"# Binder-TSL composition consumes the HC1-scale IF -\n"
"# see had.py:3495-3507). On the continuous designs the\n"
"# vcov_type kwarg is unused (CCT-2014 robust SE is the\n"
"# only formula), so passing vcov_type='hc1' is a no-op\n"
"# there and a safe default for the survey-aware example.\n"
"est = HeterogeneousAdoptionDiD(\n"
" n_bootstrap=999, seed=42, vcov_type='hc1')\n"
"es = est.fit(\n"
" data, outcome_col='y', unit_col='unit',\n"
" time_col='t', dose_col='d',\n"
" first_treat_col='first_treat',\n"
" aggregate='event_study',\n"
" survey_design=sd, cband=True)\n"
"es.cband_low, es.cband_high # simultaneous band endpoints"
),
priority="medium",
step_name="sensitivity",
),
_step(
baker_step=7,
label="Inspect per-horizon WAS arrays + pre-period placebos",
why=(
"Per-horizon WAS reveals adoption-effect dynamics. "
"Pre-period placebo horizons (event_times