"""Validate a real date without rounding integer identity or floating labels."""

if isinstance(value, (bool, np.bool_)) or not isinstance(

value, (int, float, np.integer, np.floating)

):

raise ValueError(f"{name} must be a numeric real time value, got {value!r}")

if isinstance(value, (int, np.integer)):

# Selectors can be arbitrarily large absent integers. Never send them

# through float() or isfinite(), even just to validate them.

return int(value)

if not np.isfinite(value):

raise ValueError(f"{name} must be a finite numeric time value, got {value!r}")

with _errstate():

native = float(value)

if not math.isfinite(native) or type(value)(native) != value:

raise ValueError(f"{name} floating time values must be exactly representable as float64")

"group_empty",

"zero_survival_baseline",

"zero_survival_last_pre",

"zero_control_increment",

"control_survival_zero",

"nonfinite_counterfactual",

"nonfinite_effect",

"group_empty",

"zero_survival_baseline",

"zero_survival_last_pre",

"zero_control_increment",

"nonfinite_contrast",

if isinstance(alpha, bool) or not isinstance(alpha, (int, float, np.floating)):

raise ValueError(f"alpha must be a float strictly between 0 and 1, got {alpha!r}")

if not (0.0 < float(alpha) < 1.0):

if seed is None:

return

if isinstance(seed, bool) or not isinstance(seed, (int, np.integer)) or seed < 0:

validate_n_bootstrap(n_bootstrap)

if int(n_bootstrap) == 1:

raise ValueError(

"n_bootstrap must be 0 (point estimates only, no inference) or at "

"least 2 (a bootstrap SD needs two draws); got 1"

"""Validate the full hyperparameter dict (used by ``__init__`` and ``fit``).

_validate_method(params["method"])

_validate_draws(params["n_bootstrap"])

_validate_alpha(params["alpha"])

data: pd.DataFrame,

outcome: str,

unit: str,

time: str,

treatment: str,

last_pre_period: Any,

) -> Dict[str, Any]:

"""Validate the balanced absorbing panel and arrange it as arrays.

if not isinstance(data, pd.DataFrame):

raise ValueError(f"data must be a pandas DataFrame, got {type(data).__name__}")

for name, col in (

("outcome", outcome),

("unit", unit),

("time", time),

("treatment", treatment),

):

if col not in data.columns:

raise ValueError(f"{name} column {col!r} not found in data")

# Identifier checks BEFORE any grouping/pivot (no silent groupby drop, no

# phantom NaN unit).

if data[unit].isna().any():

raise ValueError(

f"unit column {unit!r} contains missing values; every row needs a "

"unit identifier (no silent groupby drop)"

)

t_col = data[time]

if pd.api.types.is_datetime64_any_dtype(t_col) or pd.api.types.is_timedelta64_dtype(t_col):

raise ValueError(

f"time column {time!r} must be numeric; convert datetime/timedelta "

"values to a numeric elapsed scale (e.g. days since the spell "

"start) before fitting"

)

if (

pd.api.types.is_bool_dtype(t_col)

or pd.api.types.is_complex_dtype(t_col)

or not pd.api.types.is_numeric_dtype(t_col)

):

raise ValueError(f"time column {time!r} must be numeric (got dtype {t_col.dtype})")

if t_col.isna().any():

raise ValueError(f"time column {time!r} contains missing or non-finite values")

grid = np.unique(t_col.to_numpy())

labels = [_time_scalar(p, f"time column {time!r}") for p in grid]

if len(data) == 0 or len(grid) < _MIN_PERIODS:

raise ValueError(

"DurationDiD requires at least three distinct time periods (baseline, "

f"last_pre_period, and one post-period); found {len(grid)}"

)

if data.duplicated(subset=[unit, time]).any():

raise ValueError(

f"DurationDiD requires exactly one row per (unit, period); found duplicate "

f"({unit!r}, {time!r}) combinations"

)

counts = data.groupby(unit, sort=True, observed=True)[time].size()

n_periods = len(grid)

incomplete = counts[counts != n_periods]

if len(incomplete) > 0:

bad = incomplete.index.tolist()[:5]

raise ValueError(

"Unbalanced panel: every individual must be observed at every date of the "

f"common time grid ({n_periods} periods); {len(incomplete)} unit(s) are not "

f"(e.g. {bad}). Late entry, dropout, and missing cells are not supported; "

"an administrative end of a complete window is fine."

)

# Subtract native scalars BEFORE conversion: machine integers can wrap,

# while casting absolute dates to float64 can erase entire time steps.

diffs = [right - left for left, right in zip(labels, labels[1:])]

step = diffs[0]

if not all(math.isfinite(d) and d > 0 for d in diffs):

raise ValueError("time grid must have a positive finite common spacing")

if any(abs(d - step) > _SPACING_RTOL * abs(step) for d in diffs):

raise ValueError(

"DurationDiD requires an equally spaced time grid (relative tolerance "

f"{_SPACING_RTOL:g}); found spacings {sorted(set(diffs))[:5]}"

)

elapsed = np.asarray([p - labels[0] for p in labels], dtype=float)

if not np.all(np.isfinite(elapsed)) or np.any(elapsed[1:] <= elapsed[:-1]):

raise ValueError(

"time grid must give finite, strictly increasing float64 elapsed durations"

)

# Binary columns: explicit float coercion, then missing/non-finite, then

# the 0/1 domain (validate_binary strips NaN before its membership test).

coerced: Dict[str, np.ndarray] = {}

for name, col in (("outcome", outcome), ("treatment", treatment)):

if pd.api.types.is_complex_dtype(data[col]) or not (

pd.api.types.is_numeric_dtype(data[col]) or pd.api.types.is_bool_dtype(data[col])

):

bad_vals = pd.unique(data[col].astype(object))[:5].tolist()

raise ValueError(

f"{name} column {col!r} must be a numeric 0/1 column (got dtype "

f"{data[col].dtype}; values such as {bad_vals}); numeric strings are not "

"coerced"

)

try:

arr = data[col].to_numpy(dtype=float)

except (ValueError, TypeError) as exc:

bad_vals = pd.unique(data[col].astype(object))[:5].tolist()

raise ValueError(

f"{name} column {col!r} must be numeric 0/1; could not convert values "

f"such as {bad_vals} ({exc})"

) from None

nonfinite = ~np.isfinite(arr)

if nonfinite.any():

rows = data.loc[nonfinite, [unit, time]].head(5).values.tolist()

raise ValueError(

f"{name} column {col!r} contains {int(nonfinite.sum())} missing or "

f"non-finite value(s); first offending (unit, period) pairs: {rows}"

)

validate_binary(arr, name)

coerced[name] = arr

# Internal frame with fixed names, so a user column named like a temporary

# (or a role column named "unit"/"time") can never collide.

# Pandas indexes need float64 for some floating dtypes (e.g. float16).

# This conversion is lossless after _time_scalar validation; retain the

# original grid dtype in results. Integer labels never enter this path.

floating_clock = pd.api.types.is_float_dtype(t_col)

index_grid = grid.astype(float) if floating_clock else grid

frame = pd.DataFrame(

{

"unit": data[unit].to_numpy(),

"time": t_col.to_numpy(dtype=float) if floating_clock else t_col.to_numpy(),

"y": coerced["outcome"],

"g": coerced["treatment"],

}

)

g_nunique = frame.groupby("unit")["g"].nunique()

if (g_nunique > 1).any():

bad = g_nunique[g_nunique > 1].index.tolist()[:5]

raise ValueError(

f"treatment column {treatment!r} must be a fixed 0/1 group indicator "

f"(constant within unit), not a time-varying received-treatment variable; "

f"units with varying values include {bad}"

)

y_wide = frame.pivot(index="unit", columns="time", values="y").reindex(columns=index_grid)

g_units = frame.groupby("unit")["g"].first().reindex(y_wide.index)

Y = y_wide.to_numpy(dtype=float)

G = g_units.to_numpy(dtype=float)

n_treated = int(np.sum(G == 1.0))

n_control = int(np.sum(G == 0.0))

if n_treated == 0 or n_control == 0:

raise ValueError(

"both groups are required: found "

f"{n_treated} treated and {n_control} control individual(s)"

)

reversal = np.diff(Y, axis=1) < 0

if reversal.any():

bad_units = y_wide.index[reversal.any(axis=1)].tolist()[:5]

raise ValueError(

"outcome must be absorbing (once 1, always 1 within each individual); "

f"found reversals (1 -> 0) for {int(reversal.any(axis=1).sum())} unit(s), "

f"e.g. {bad_units}"

)

tstar_val = _time_scalar(last_pre_period, "last_pre_period")

if tstar_val not in labels:

raise ValueError(

f"last_pre_period {last_pre_period!r} is not a value of the time column "

f"(grid: {grid.tolist()[:8]}{'...' if len(grid) > 8 else ''})"

)

tstar_idx = labels.index(tstar_val)

if tstar_idx == 0:

raise ValueError(

"last_pre_period equals the first date; at least two pre-treatment dates "

"(the baseline and last_pre_period) are required"

)

if tstar_idx == n_periods - 1:

raise ValueError(

"last_pre_period equals the last date; at least one post-treatment date is required"

)

order = np.argsort(-G, kind="stable") # treated first, stable within group

return {

"Y": np.ascontiguousarray(Y[order]),

"n_treated": n_treated,

"grid": grid,

"elapsed": elapsed,

"tstar_idx": tstar_idx,

Y: np.ndarray, n_treated: int, W: np.ndarray

) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:

"""Group survival curves for count-weight rows ``W`` (C x n).

W = np.asarray(W, dtype=float)

W1, W2 = W[:, :n_treated], W[:, n_treated:]

n1, n2 = W1.sum(axis=1), W2.sum(axis=1)

with _errstate():

S1 = 1.0 - (W1 @ Y[:n_treated]) / n1[:, None]

S2 = 1.0 - (W2 @ Y[n_treated:]) / n2[:, None]

S: np.ndarray, elapsed: np.ndarray

) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:

"""``R = -log S``, ``D = R - R[..., :1]``, ``H = D / elapsed`` (baseline NaN)."""

with _errstate():

R = -np.log(S)

D = R - R[..., :1]

H = D / elapsed

S: np.ndarray,

R: np.ndarray,

D: np.ndarray,

elapsed: np.ndarray,

fit_idx: np.ndarray,

fit_weights: np.ndarray,

method: str,

) -> Tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:

"""Fitted coefficient and imputed counterfactual, vectorized over draws.

S1 = S[:, 0, :]

R1, D1, D2 = R[:, 0, :], D[:, 0, :], D[:, 1, :]

w = np.asarray(fit_weights, dtype=float)

with _errstate():

if method == "cd":

e_fit = elapsed[fit_idx]

H1 = D1[:, fit_idx] / e_fit

H2 = D2[:, fit_idx] / e_fit

c = (H1 - H2) @ w

R0 = R1[:, :1] + D2 + elapsed[None, :] * c[:, None]

elif method == "ph":

ratio = D1[:, fit_idx] / D2[:, fit_idx]

c = ratio @ w

R0 = R1[:, :1] + c[:, None] * D2

else: # pragma: no cover - fit() validates method first

raise ValueError(f"method must be 'cd' or 'ph', got {method!r}")

S0 = np.exp(-R0)

tau = S0 - S1

R: np.ndarray, D: np.ndarray, elapsed: np.ndarray, c: np.ndarray, method: str

) -> np.ndarray:

"""Per-date absolute tolerance for the curve-validity gate (leading draw axis).

with _errstate():

term = elapsed[None, :] * c[:, None] if method == "cd" else c[:, None] * D[:, 1, :]

scale = np.maximum.reduce(

[

np.ones_like(term),

np.abs(np.broadcast_to(R[:, 0, :1], term.shape)),

np.abs(D[:, 1, :]),

np.abs(term),

]

)

R0: np.ndarray, S2: np.ndarray, tstar_idx: int, tol: Optional[np.ndarray] = None

) -> List[str]:

"""Validity of the imputed curve at every date (first match wins).

n_periods = len(R0)

tol_arr = np.zeros(n_periods) if tol is None else np.asarray(tol, dtype=float)

status = ["ok"] * n_periods

for t in range(1, n_periods):

if t > tstar_idx and S2[t] <= 0:

status[t] = "control_survival_zero"

elif not np.isfinite(R0[t]):

status[t] = "counterfactual_nonfinite"

elif R0[t] < -tol_arr[t]:

status[t] = "counterfactual_survival_above_one"

elif np.isfinite(R0[t - 1]) and R0[t] < R0[t - 1] - tol_arr[t]:

status[t] = "counterfactual_nonmonotone"

S: np.ndarray,

D: np.ndarray,

method: str,

grid: np.ndarray,

tstar_idx: int,

pre_periods: Optional[Sequence[Any]],

pre_period_weights: Optional[Sequence[float]],

) -> Tuple[np.ndarray, np.ndarray, Dict[Any, str], List[str]]:

"""Validate the fitting selectors and resolve the eligible fitting set.

labels = [_native_time_label(p) for p in grid]

if pre_period_weights is not None and pre_periods is None:

raise ValueError("pre_period_weights requires pre_periods (the dates the weights refer to)")

if pre_periods is None:

cand_idx = np.arange(1, tstar_idx + 1)

requested = False

else:

req = _selector_array(pre_periods, "pre_periods", "time values")

if req.size == 0:

raise ValueError("pre_periods must name at least one pre-treatment date")

idx_list: List[int] = []

for p in req:

if p not in labels:

raise ValueError(f"Pre-period '{_fmt(p)}' not found in time column")

k = labels.index(p)

if k < 1 or k > tstar_idx:

raise ValueError(

f"pre_periods value {_fmt(p)} must lie strictly after the baseline "

f"date {_label(grid, 0)!r} and at or before last_pre_period {_label(grid, tstar_idx)!r}"

)

idx_list.append(k)

if len(set(idx_list)) != len(idx_list):

raise ValueError("pre_periods contains duplicate dates")

cand_idx = np.asarray(idx_list, dtype=int)

requested = True

if pre_period_weights is None:

cand_w = np.ones(len(cand_idx), dtype=float)

else:

cand_w = _selector_array(pre_period_weights, "pre_period_weights", "nonnegative weights")

if cand_w.shape != (len(cand_idx),):

raise ValueError(

f"pre_period_weights must have one entry per pre_periods date "

f"({len(cand_idx)}), got {cand_w.shape[0]}"

)

if not np.all(np.isfinite(cand_w)):

raise ValueError("pre_period_weights must be finite")

if np.any(cand_w < 0):

raise ValueError("pre_period_weights must be nonnegative")

if np.all(cand_w == 0):

raise ValueError("pre_period_weights must not all be zero")

excluded: Dict[Any, str] = {}

keep_idx: List[int] = []

keep_w: List[float] = []

S1, S2, D2 = S[0], S[1], D[1]

for k, w in zip(cand_idx.tolist(), cand_w.tolist()):

label = _label(grid, k)

if w == 0:

excluded[label] = "zero_weight"

elif S1[k] <= 0:

excluded[label] = "zero_treated_survival"

elif S2[k] <= 0:

excluded[label] = "zero_control_survival"

elif method == "ph" and D2[k] <= 0:

excluded[label] = "zero_control_increment"

else:

keep_idx.append(k)

keep_w.append(w)

if not keep_idx:

raise ValueError(

"no eligible fitting period: every candidate pre-treatment date was "

f"excluded ({excluded}). PH needs a positive control cumulative-hazard "

"increment at some pre-date after the baseline; both groups need positive "

"survival at the fitting dates."

)

order = np.argsort(keep_idx)

fit_idx = np.asarray(keep_idx, dtype=int)[order]

weights = np.asarray(keep_w, dtype=float)[order]

# Scale-invariant normalization: the kept weights are finite and strictly

# positive, so dividing by the maximum first keeps the sum finite even for

# weights near the float64 limit (a bare sum could overflow to inf and

# silently normalize to zeros).

weights = weights / weights.max()

weights = weights / weights.sum()

messages: List[str] = []

elig_excluded = {k: v for k, v in excluded.items() if v != "zero_weight"}

if elig_excluded:

who = "requested" if requested else "default"

messages.append(

f"Excluded {len(elig_excluded)} {who} fitting period(s) as ineligible "

f"{elig_excluded}; the remaining fitting weights were renormalized to sum to one."

)

if tstar_idx not in set(fit_idx.tolist()):

messages.append(

f"The fitting periods omit last_pre_period {_label(grid, tstar_idx)!r}; the "

"pre-treatment diagnostic keeps that date as its fixed anchor regardless."

)

"""Flatten ordered inputs, preserving date identity separately from weights.

if isinstance(value, (str, bytes)) or np.isscalar(value) or not hasattr(value, "__iter__"):

raise ValueError(f"{name} must be a list of {kind} (e.g. [...]), got {value!r}")

if isinstance(value, (set, frozenset)):

# Unordered: pre_periods and pre_period_weights are paired positionally.

raise ValueError(

f"{name} must be an ordered list of {kind} (a set has no positional order "

f"to align with its companion selector), got {value!r}"

)

try:

raw = np.asarray(list(value), dtype=object).ravel()

if name == "pre_periods":

arr = np.asarray([_time_scalar(p, name) for p in raw], dtype=object)

else:

if any(isinstance(p, (complex, np.complexfloating)) for p in raw):

raise ValueError("weights must be real")

with _errstate():

arr = np.asarray(raw, dtype=float)

except (TypeError, ValueError, OverflowError) as exc:

raise ValueError(f"{name} must be a list of numeric {kind}, got {value!r}: {exc}") from None

D: np.ndarray, H: np.ndarray, J: np.ndarray, tstar_idx: int, method: str

) -> np.ndarray:

"""Algorithm 2 fixed-anchor contrasts over ``J`` (leading draw axis)."""

with _errstate():

if method == "cd":

gap = H[:, 0, J] - H[:, 1, J]

anchor = H[:, 0, tstar_idx] - H[:, 1, tstar_idx]

else:

gap = D[:, 0, J] / D[:, 1, J]

anchor = D[:, 0, tstar_idx] / D[:, 1, tstar_idx]

"""Inverse empirical CDF quantile: the ``ceil(p*B)``-th order statistic.

xs = np.sort(np.asarray(x, dtype=float))

n = xs.shape[0]

if n == 0:

return float("nan")

k = int(math.ceil(p * n - 1e-9))

k = min(max(k, 1), n)

point: np.ndarray, draws: np.ndarray, alpha: float

) -> Dict[str, Any]:

"""Centered absolute-deviation bootstrap summary on COMPLETE draws.

point = np.asarray(point, dtype=float).ravel()

draws = np.asarray(draws, dtype=float).reshape(draws.shape[0], -1)

with _errstate():

vcov = np.atleast_2d(np.cov(draws, rowvar=False, ddof=1))

# Summing identical floats can round their mean away from the stored

# value, creating spurious positive variance. Enforce exact degeneracy

# without treating legitimate small variation as zero.

constant = np.all(draws == draws[0], axis=0)

vcov[constant, :] = 0.0

vcov[:, constant] = 0.0

se = np.sqrt(np.diag(vcov))

crit = np.full(point.shape, np.nan)

p = np.full(point.shape, np.nan)

crit_sim = p_joint = statistic = float("nan")

if np.all(np.isfinite(se)) and np.all(se > 0):

z = np.abs(draws - point[None, :]) / se[None, :]

crit = np.array(

[_quantile_inverted_cdf(z[:, k], 1.0 - alpha) for k in range(z.shape[1])]

)

t_abs = np.abs(point / se)

p = np.mean(z >= t_abs[None, :], axis=0)

m = np.max(z, axis=1)

crit_sim = _quantile_inverted_cdf(m, 1.0 - alpha)

p_joint = float(np.mean(m >= np.max(t_abs)))

statistic = float(np.max(t_abs))

return {

"vcov": vcov,

"se": se,

"crit": crit,

"ci_lower": point - crit * se,

"ci_upper": point + crit * se,

"p": p,

"crit_sim": crit_sim,

"band_lower": point - crit_sim * se,

"band_upper": point + crit_sim * se,

"p_joint": p_joint,

"statistic": statistic,

"""Per-draw reason string: the first predicate in ``order`` that fires."""

reason = np.full(n, "", dtype=object)

for name in order:

reason = np.where((reason == "") & masks[name], name, reason)

Y: np.ndarray,

n_treated: int,

elapsed: np.ndarray,

tstar_idx: int,

fit_idx: np.ndarray,

fit_weights: np.ndarray,

method: str,

J: np.ndarray,

n_bootstrap: int,

rng: np.random.Generator,

) -> Dict[str, Any]:

"""Algorithm 1: ``n_bootstrap`` pooled whole-history resamples.

n, n_periods = Y.shape

post_idx = np.arange(tstar_idx + 1, n_periods)

n_post = len(post_idx)

tau_star = np.full((n_bootstrap, n_post), np.nan)

head_star = np.full(n_bootstrap, np.nan)

delta_star = np.full((n_bootstrap, len(J)), np.nan)

ok_post = np.zeros(n_bootstrap, dtype=bool)

ok_pretest = np.zeros(n_bootstrap, dtype=bool)

reason_post = np.full(n_bootstrap, "", dtype=object)

reason_pretest = np.full(n_bootstrap, "", dtype=object)

n_invalid_curve = 0

chunk = int(min(compute_block_size(n, n_bootstrap), _MAX_CHUNK_ROWS))

pretest_dates = np.concatenate([J, [tstar_idx]]).astype(int)

for start in range(0, n_bootstrap, chunk):

size = min(chunk, n_bootstrap - start)

idx = _draw_indices(rng, n, size)

W = np.zeros((size, n), dtype=float)

for r in range(size):

W[r] = np.bincount(idx[r], minlength=n)

S, n1, n2 = _group_survival(Y, n_treated, W)

R, D, H = _log_survival_moments(S, elapsed)

c, R0, S0, tau = _estimate_from_survival(S, R, D, elapsed, fit_idx, fit_weights, method)

tau_post = tau[:, post_idx]

with _errstate():

head = tau_post.mean(axis=1)

delta = (

_pretest_contrasts(D, H, J, tstar_idx, method) if len(J) else np.zeros((size, 0))

)

S1, S2 = S[:, 0, :], S[:, 1, :]

group_empty = (n1 == 0) | (n2 == 0)

base_zero = ~(np.nan_to_num(S1[:, 0], nan=0.0) > 0) | ~(

np.nan_to_num(S2[:, 0], nan=0.0) > 0

)

tstar_zero = ~(np.nan_to_num(S1[:, tstar_idx], nan=0.0) > 0) | ~(

np.nan_to_num(S2[:, tstar_idx], nan=0.0) > 0

)

with _errstate():

if method == "ph":

zero_inc_post = ~np.all(D[:, 1, fit_idx] > 0, axis=1)

zero_inc_pre = ~np.all(D[:, 1, pretest_dates] > 0, axis=1)

else:

zero_inc_post = np.zeros(size, dtype=bool)

zero_inc_pre = np.zeros(size, dtype=bool)

ctrl_zero = ~np.all(np.nan_to_num(S2[:, post_idx], nan=0.0) > 0, axis=1)

nonfinite_cf = (

~np.isfinite(c)

| ~np.all(np.isfinite(R0[:, post_idx]), axis=1)

| ~np.all(np.isfinite(S0[:, post_idx]), axis=1)

)

nonfinite_eff = ~np.all(np.isfinite(tau_post), axis=1)

if method == "cd":

inputs_ok = np.all(np.isfinite(H[:, :, pretest_dates]), axis=(1, 2))

else:

inputs_ok = np.all(np.isfinite(D[:, :, pretest_dates]), axis=(1, 2))

nonfinite_delta = ~(inputs_ok & np.all(np.isfinite(delta), axis=1))

r_post = _first_match(

{

"group_empty": group_empty,

"zero_survival_baseline": base_zero,

"zero_survival_last_pre": tstar_zero,

"zero_control_increment": zero_inc_post,

"control_survival_zero": ctrl_zero,

"nonfinite_counterfactual": nonfinite_cf,

"nonfinite_effect": nonfinite_eff,

},

_POST_FAILURE_ORDER,

size,

)

r_pre = _first_match(

{

"group_empty": group_empty,

"zero_survival_baseline": base_zero,

"zero_survival_last_pre": tstar_zero,

"zero_control_increment": zero_inc_pre,

"nonfinite_contrast": nonfinite_delta,

},

_PRETEST_FAILURE_ORDER,

size,

)

okp = r_post == ""

okq = r_pre == ""

sl = slice(start, start + size)

ok_post[sl] = okp

ok_pretest[sl] = okq

reason_post[sl] = r_post

reason_pretest[sl] = r_pre

tau_star[sl][okp] = tau_post[okp]

head_star[sl][okp] = head[okp]

if len(J):

delta_star[sl][okq] = delta[okq]

# Diagnostic count: complete draws whose imputed curve leaves the

# domain (finite S0 > 1 or a decreasing step) — not failures.

if okp.any():

R0_ok = R0[okp]

tol_ok = _curve_tolerance(R, D, elapsed, c, method)[okp]

with _errstate():

bad = (R0_ok < -tol_ok).any(axis=1) | (np.diff(R0_ok, axis=1) < -tol_ok[:, 1:]).any(

axis=1

)

n_invalid_curve += int(bad.sum())

return {

"tau_star": tau_star,

"head_star": head_star,

"delta_star": delta_star,

"ok_post": ok_post,

"ok_pretest": ok_pretest,

"reason_post": reason_post,

"reason_pretest": reason_pretest,

"n_draws_invalid_counterfactual": n_invalid_curve,

out: Dict[str, int] = {}

for r in reasons.tolist():

if r:

out[r] = out.get(r, 0) + 1

"""Duration difference-in-differences (Deaner & Ku 2026) for absorbing outcomes.

def __init__(

self,

method: str = "cd",

n_bootstrap: int = 1000,

alpha: float = 0.05,

seed: Optional[int] = None,

):

_validate_all_params(

{"method": method, "n_bootstrap": n_bootstrap, "alpha": alpha, "seed": seed}

)

self.method = method

self.n_bootstrap = n_bootstrap

self.alpha = alpha

self.seed = seed

self.is_fitted_ = False

self.results_: Optional[DurationDiDResults] = None

# get_params/set_params come from BaseEstimator.

def fit(

self,

data: pd.DataFrame,

outcome: str,

unit: str,

time: str,

treatment: str,

*,

last_pre_period: Any,

pre_periods: Optional[Sequence[Any]] = None,

pre_period_weights: Optional[Sequence[float]] = None,

) -> DurationDiDResults:

"""Fit the estimator on a balanced long individual panel.

# Re-validate the configuration BEFORE any data work: attributes can

# be mutated directly after construction (bypassing set_params).

_validate_all_params(self.get_params())