"""

Synthetic Difference-in-Differences estimator.

"""

import warnings

from typing import Any, Dict, List, Optional, Tuple

import numpy as np

import pandas as pd

from numpy.linalg import LinAlgError

from diff_diff.bootstrap_utils import generate_rao_wu_weights

from diff_diff.estimators import DifferenceInDifferences

from diff_diff.linalg import solve_ols

from diff_diff.results import SyntheticDiDResults, _SyntheticDiDFitSnapshot

from diff_diff.utils import (

_compute_regularization,

_sum_normalize,

compute_sdid_estimator,

compute_sdid_unit_weights,

compute_sdid_unit_weights_survey,

compute_time_weights,

compute_time_weights_survey,

safe_inference,

validate_binary,

)

class SyntheticDiD(DifferenceInDifferences):

"""

Synthetic Difference-in-Differences (SDID) estimator.

Combines the strengths of Difference-in-Differences and Synthetic Control

methods by re-weighting control units to better match treated units'

pre-treatment trends.

This method is particularly useful when:

- You have few treated units (possibly just one)

- Parallel trends assumption may be questionable

- Control units are heterogeneous and need reweighting

- You want robustness to pre-treatment differences

Parameters

----------

zeta_omega : float, optional

Regularization for unit weights. If None (default), auto-computed

from data as ``(N1 * T1)^(1/4) * noise_level`` matching R's synthdid.

zeta_lambda : float, optional

Regularization for time weights. If None (default), auto-computed

from data as ``1e-6 * noise_level`` matching R's synthdid.

alpha : float, default=0.05

Significance level for confidence intervals.

variance_method : str, default="placebo"

Method for variance estimation:

- "placebo": Placebo-based variance matching R's synthdid::vcov(method="placebo").

Implements Algorithm 4 from Arkhangelsky et al. (2021). Library default

(R's default is ``"bootstrap"``; we default to placebo because it is

unconditionally available on pweight-only survey designs and avoids the

~5–30× slowdown of the refit bootstrap). See REGISTRY.md §SyntheticDiD

``Note (default variance_method deviation from R)`` for rationale.

- "bootstrap": Paper-faithful pairs bootstrap — Arkhangelsky et al. (2021)

Algorithm 2 step 2, also the behavior of R's default

synthdid::vcov(method="bootstrap") (which rebinds ``attr(estimate, "opts")``

with ``update.omega=TRUE``, so the renormalized ω is only Frank-Wolfe

initialization). Re-estimates ω̂_b and λ̂_b via two-pass sparsified

Frank-Wolfe on each bootstrap draw. **Survey support (PR #352):**

pweight-only fits use the constant per-control survey weight as ``rw``;

full-design fits (strata/PSU/FPC) use Rao-Wu rescaled weights per draw.

Both compose with the **weighted Frank-Wolfe** kernel

(``min ||A·diag(rw)·ω - b||² + ζ²·Σ rw_i ω_i²``); the FW returns ω on the

standard simplex, then ``ω_eff = rw·ω/Σ(rw·ω)`` is composed for the SDID

estimator. See REGISTRY.md §SyntheticDiD ``Note (survey + bootstrap

composition)`` for the argmin-set caveat.

- "jackknife": Jackknife variance matching R's synthdid::vcov(method="jackknife").

Implements Algorithm 3 from Arkhangelsky et al. (2021). Deterministic

(N_control + N_treated iterations), uses fixed weights (no re-estimation).

The ``n_bootstrap`` parameter is ignored for this method.

n_bootstrap : int, default=200

Number of replications for variance estimation. Used for:

- Bootstrap: Number of bootstrap samples

- Placebo: Number of random permutations (matches R's `replications` argument)

Ignored when ``variance_method="jackknife"``.

seed : int, optional

Random seed for reproducibility. If None (default), results

will vary between runs.

Attributes

----------

results_ : SyntheticDiDResults

Estimation results after calling fit().

is_fitted_ : bool

Whether the model has been fitted.

Examples

--------

Basic usage with panel data:

>>> import pandas as pd

>>> from diff_diff import SyntheticDiD

>>>

>>> # Panel data with units observed over multiple time periods

>>> # Treatment occurs at period 5 for treated units

>>> data = pd.DataFrame({

... 'unit': [...], # Unit identifier

... 'period': [...], # Time period

... 'outcome': [...], # Outcome variable

... 'treated': [...] # 1 if unit is ever treated, 0 otherwise

... })

>>>

>>> # Fit SDID model

>>> sdid = SyntheticDiD()

>>> results = sdid.fit(

... data,

... outcome='outcome',

... treatment='treated',

... unit='unit',

... time='period',

... post_periods=[5, 6, 7, 8]

... )

>>>

>>> # View results

>>> results.print_summary()

>>> print(f"ATT: {results.att:.3f} (SE: {results.se:.3f})")

>>>

>>> # Examine unit weights

>>> weights_df = results.get_unit_weights_df()

>>> print(weights_df.head(10))

Notes

-----

The SDID estimator (Arkhangelsky et al., 2021) computes:

τ̂ = (Ȳ_treated,post - Σ_t λ_t * Y_treated,t)

- Σ_j ω_j * (Ȳ_j,post - Σ_t λ_t * Y_j,t)

Where:

- ω_j are unit weights (sum to 1, non-negative)

- λ_t are time weights (sum to 1, non-negative)

Unit weights ω are chosen to match pre-treatment outcomes:

min ||Σ_j ω_j * Y_j,pre - Y_treated,pre||²

This interpolates between:

- Standard DiD (uniform weights): ω_j = 1/N_control

- Synthetic Control (exact matching): concentrated weights

**Conley spatial-HAC rejection.** SyntheticDiD does not support the

Conley (1999) spatial-HAC analytical sandwich. Passing

``vcov_type="conley"`` or any non-``None`` Conley keyword

(``conley_coords``, ``conley_cutoff_km``, ``conley_metric``,

``conley_kernel``) to ``__init__`` or ``set_params`` raises

``TypeError``. Rationale: SyntheticDiD's variance is derived from

bootstrap / jackknife / placebo resampling (Arkhangelsky et al. 2021

Algorithms 2–4), not the sandwich identity Conley plugs into. Adding

Conley support would require either an analytical SDID sandwich path

or a spatial-block bootstrap (Politis-Romano 1994 territory). Tracked

as a follow-up in ``TODO.md``.

References

----------

Arkhangelsky, D., Athey, S., Hirshberg, D. A., Imbens, G. W., & Wager, S.

(2021). Synthetic Difference-in-Differences. American Economic Review,

111(12), 4088-4118.

"""

def __init__(

self,

zeta_omega: Optional[float] = None,

zeta_lambda: Optional[float] = None,

alpha: float = 0.05,

variance_method: str = "placebo",

n_bootstrap: int = 200,

seed: Optional[int] = None,

# Deprecated — accepted for backward compat, ignored with warning

lambda_reg: Optional[float] = None,

zeta: Optional[float] = None,

# Defensive guard against silently-ignored Conley kwargs. SyntheticDiD

# inherits __init__ from DifferenceInDifferences but overrides with

# literal `super().__init__(robust=True, cluster=None, alpha=alpha)`,

# so any user-passed `vcov_type=` or `conley_*=` would be silently

# dropped. Per `feedback_no_silent_failures`, raise loudly. Tracked

# in TODO.md for a follow-up that wires Conley to a non-bootstrap

# variance path on SyntheticDiD.

vcov_type: Optional[str] = None,

conley_coords: Optional[Tuple[str, str]] = None,

conley_cutoff_km: Optional[float] = None,

conley_metric: Optional[str] = None,

conley_kernel: Optional[str] = None,

):

if vcov_type == "conley" or any(

v is not None for v in (conley_coords, conley_cutoff_km, conley_metric, conley_kernel)

):

raise TypeError(

"SyntheticDiD does not yet support vcov_type='conley' or any "

"conley_* kwargs. SyntheticDiD uses bootstrap/jackknife/placebo "

"variance (variance_method=...), not the analytical sandwich "

"routed through compute_robust_vcov. Tracked in TODO.md as "

"a follow-up."

)

if vcov_type is not None and vcov_type != "conley":

raise TypeError(

f"SyntheticDiD does not accept vcov_type={vcov_type!r}. "

f"SyntheticDiD's variance is bootstrap/jackknife/placebo "

f"based; configure via variance_method=..."

)

if lambda_reg is not None:

warnings.warn(

"lambda_reg is deprecated and ignored. Regularization is now "

"auto-computed from data. Use zeta_omega to override unit weight "

"regularization. Will be removed in v4.0.0.",

DeprecationWarning,

stacklevel=2,

)

if zeta is not None:

warnings.warn(

"zeta is deprecated and ignored. Use zeta_lambda to override "

"time weight regularization. Will be removed in v4.0.0.",

DeprecationWarning,

stacklevel=2,

)

super().__init__(robust=True, cluster=None, alpha=alpha)

self.zeta_omega = zeta_omega

self.zeta_lambda = zeta_lambda

self.variance_method = variance_method

self.n_bootstrap = n_bootstrap

self.seed = seed

self._validate_config()

self._unit_weights = None

self._time_weights = None

_VALID_VARIANCE_METHODS = ("bootstrap", "jackknife", "placebo")

def _validate_config(self) -> None:

"""Validate ``variance_method`` and ``n_bootstrap`` on the current state.

Called from both ``__init__`` and ``set_params`` so updates via the

sklearn-style setter path enforce the same contract as construction.

"""

if self.variance_method not in self._VALID_VARIANCE_METHODS:

raise ValueError(

f"variance_method must be one of {self._VALID_VARIANCE_METHODS}, "

f"got '{self.variance_method}'"

)

if self.n_bootstrap < 2 and self.variance_method != "jackknife":

raise ValueError(

f"n_bootstrap must be >= 2 (got {self.n_bootstrap}). At least 2 "

f"iterations are needed to estimate standard errors."

)

def fit( # type: ignore[override]

self,

data: pd.DataFrame,

outcome: str,

treatment: str,

unit: str,

time: str,

post_periods: Optional[List[Any]] = None,

covariates: Optional[List[str]] = None,

survey_design=None,

) -> SyntheticDiDResults:

"""

Fit the Synthetic Difference-in-Differences model.

Parameters

----------

data : pd.DataFrame

Panel data with observations for multiple units over multiple

time periods.

outcome : str

Name of the outcome variable column.

treatment : str

Name of the treatment group indicator column (0/1).

Should be 1 for all observations of treated units

(both pre and post treatment).

unit : str

Name of the unit identifier column.

time : str

Name of the time period column.

post_periods : list, optional

List of time period values that are post-treatment.

If None, uses the last half of periods.

covariates : list, optional

List of covariate column names. Covariates are residualized

out before computing the SDID estimator.

survey_design : SurveyDesign, optional

Survey design specification. Only pweight weight_type is

supported. Replicate-weight designs are rejected. All three

variance methods support both pweight-only and full

strata/PSU/FPC designs:

method pweight-only strata/PSU/FPC

bootstrap ✓ weighted FW ✓ weighted FW + Rao-Wu (PR #355)

placebo ✓ ✓ stratified permutation + weighted FW

jackknife ✓ ✓ PSU-level LOO + stratum aggregation

- **Bootstrap** composes Rao-Wu rescaled weights per draw with

the weighted-Frank-Wolfe kernel; see REGISTRY.md §SyntheticDiD

``Note (survey + bootstrap composition)``.

- **Placebo** under full design uses within-stratum permutation

(pseudo-treated sampled from controls in each treated-containing

stratum) with weighted-FW refit per draw; fit-time feasibility

guards raise ``ValueError`` when a treated stratum has fewer

controls than treated units (see ``Note (survey + placebo

composition)``).

- **Jackknife** under full design uses PSU-level LOO with

stratum aggregation (Rust & Rao 1996); anti-conservative with

few PSUs per stratum — prefer ``bootstrap`` when tight SE

calibration matters in that regime (see ``Note (survey +

jackknife composition)``).

Returns

-------

SyntheticDiDResults

Object containing the ATT estimate, standard error,

unit weights, and time weights.

Raises

------

ValueError

If required parameters are missing, data validation fails,

or a non-pweight survey design is provided. Under survey

designs, also raises when:

- The total survey mass on either arm is zero

(``w_control.sum() == 0`` or ``w_treated.sum() == 0``).

Every unit on that arm would have weight 0, encoding an

unidentified target population (PR #355 R7 P1).

- The composed effective-control mass

``(unit_weights * w_control).sum()`` is zero. Frank-Wolfe

sparsifies ``unit_weights`` to exact zeros by design, so

even when at least one control has positive survey weight,

the FW solution may concentrate all mass on controls whose

survey weights are 0. Raising up front avoids a silent

``0/0`` in the ``omega_eff`` normalization (PR #355 R12 P1).

- ``survey_design`` declares ``fpc`` with no explicit

``psu=``. SDID Rao-Wu then treats each unit as its own

PSU, so ``fpc`` must be ``>=`` the number of units

(unstratified) or ``>=`` the per-stratum unit count

(stratified). Front-door checked after

``collapse_survey_to_unit_level`` so the user sees a

targeted error instead of a bootstrap-exhaustion

failure (PR #355 R8 P1).

NotImplementedError

If ``survey_design`` carries replicate weights (BRR/Fay/JK1/

JKn/SDR) — SyntheticDiD has no replicate-weight variance

path. All three variance methods (placebo, bootstrap,

jackknife) accept pweight-only and full strata/PSU/FPC

analytical designs; only replicate-weight designs are

rejected.

"""

# Validate inputs

if outcome is None or treatment is None or unit is None or time is None:

raise ValueError("Must provide 'outcome', 'treatment', 'unit', and 'time'")

# Check columns exist

required_cols = [outcome, treatment, unit, time]

if covariates:

required_cols.extend(covariates)

missing = [c for c in required_cols if c not in data.columns]

if missing:

raise ValueError(f"Missing columns: {missing}")

# Resolve survey design

from diff_diff.survey import (

_resolve_survey_for_fit,

_validate_unit_constant_survey,

)

# R11 P1 fix: FPC is a documented no-op on placebo (Pesarin 2001

# §1.5 — permutation tests condition on the observed sample), but

# ``SurveyDesign.resolve()`` itself enforces ``FPC >= n_PSU``

# design-validity constraints (survey.py:349-368). On placebo,

# those constraints would block legitimate fits for a design

# element that doesn't enter the placebo math. Drop FPC from a

# copy of the survey design before resolution so placebo

# bypasses the validator entirely; emit the FPC no-op warning

# at the same time. The original survey_design object is

# preserved (caller's reference unchanged).

survey_design_for_resolve = survey_design

if (

self.variance_method == "placebo"

and survey_design is not None

and getattr(survey_design, "fpc", None) is not None

):

# R13 P3 fix: validate the FPC column name exists in `data`

# before dropping. Otherwise a typoed ``fpc="fpc_typo"`` is

# silently ignored on the placebo path (the missing-column

# check inside ``SurveyDesign.resolve()`` never runs because

# we strip FPC pre-resolve). Raise the same targeted error

# ``resolve()`` would have raised so input-spec mistakes

# surface even when the value is mathematically a no-op.

fpc_col = survey_design.fpc

if fpc_col not in data.columns:

raise ValueError(f"FPC column '{fpc_col}' not found in data")

import dataclasses as _dc

warnings.warn(

"SurveyDesign(fpc=...) is a no-op on "

"variance_method='placebo': permutation tests are "

"conditional on the observed sample (Pesarin 2001 §1.5), "

"so the sampling fraction does not enter Algorithm 4 or "

"its stratified-permutation survey extension. The FPC "

"column is dropped from the resolved survey design for "

"the placebo fit (this also bypasses the FPC >= n_PSU "

"design-validity check in SurveyDesign.resolve()). Use "

"variance_method='bootstrap' or 'jackknife' if you need "

"FPC to participate in the variance computation.",

UserWarning,

stacklevel=2,

)

survey_design_for_resolve = _dc.replace(survey_design, fpc=None)

resolved_survey, survey_weights, survey_weight_type, survey_metadata = (

_resolve_survey_for_fit(survey_design_for_resolve, data, "analytical")

)

# Reject replicate-weight designs — SyntheticDiD has no replicate-

# weight variance path. Analytical (pweight / strata / PSU / FPC)

# designs are supported across all three variance methods:

# bootstrap via weighted-FW + Rao-Wu (PR #355); placebo via

# stratified permutation + weighted FW; jackknife via PSU-level

# LOO with stratum aggregation (Rust & Rao 1996).

if resolved_survey is not None and resolved_survey.uses_replicate_variance:

raise NotImplementedError(

"SyntheticDiD does not support replicate-weight survey "

"designs. Analytical designs are supported across all "

"three variance methods (placebo, bootstrap, jackknife), "

"for both pweight-only and full strata/PSU/FPC. See "

"docs/methodology/REGISTRY.md §SyntheticDiD for the "

"full survey support matrix."

)

# Validate pweight only

if resolved_survey is not None and resolved_survey.weight_type != "pweight":

raise ValueError(

"SyntheticDiD survey support requires weight_type='pweight'. "

f"Got '{resolved_survey.weight_type}'."

)

# Strata/PSU/FPC support matrix:

# bootstrap → supported via weighted Frank-Wolfe + hybrid

# pairs-bootstrap + Rao-Wu rescaling (PR #355;

# see _bootstrap_se Rao-Wu branch).

# placebo → supported via stratified permutation + weighted

# Frank-Wolfe (this PR; _placebo_variance_se_survey).

# jackknife → supported via PSU-level LOO with stratum

# aggregation (this PR; _jackknife_se_survey).

# Validate treatment is binary

validate_binary(data[treatment].values, "treatment")

# Get all unique time periods

all_periods = sorted(data[time].unique())

if len(all_periods) < 2:

raise ValueError("Need at least 2 time periods")

# Determine pre and post periods

if post_periods is None:

mid = len(all_periods) // 2

post_periods = list(all_periods[mid:])

pre_periods = list(all_periods[:mid])

else:

post_periods = list(post_periods)

pre_periods = [p for p in all_periods if p not in post_periods]

if len(post_periods) == 0:

raise ValueError("Must have at least one post-treatment period")

if len(pre_periods) == 0:

raise ValueError("Must have at least one pre-treatment period")

# Validate post_periods are in data

for p in post_periods:

if p not in all_periods:

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

# Identify treated and control units

# Treatment indicator should be constant within unit

unit_treatment = data.groupby(unit)[treatment].first()

# Validate treatment is constant within unit (SDID requires block treatment)

treatment_nunique = data.groupby(unit)[treatment].nunique()

varying_units = treatment_nunique[treatment_nunique > 1]

if len(varying_units) > 0:

example_unit = varying_units.index[0]

example_vals = sorted(data.loc[data[unit] == example_unit, treatment].unique())

raise ValueError(

f"Treatment indicator varies within {len(varying_units)} unit(s) "

f"(e.g., unit '{example_unit}' has values {example_vals}). "

f"SyntheticDiD requires 'block' treatment where treatment is "

f"constant within each unit across all time periods. "

f"For staggered adoption designs, use CallawaySantAnna or "

f"ImputationDiD instead."

)

treated_units = unit_treatment[unit_treatment == 1].index.tolist()

control_units = unit_treatment[unit_treatment == 0].index.tolist()

if len(treated_units) == 0:

raise ValueError("No treated units found")

if len(control_units) == 0:

raise ValueError("No control units found")

# Validate balanced panel (SDID requires all units observed in all periods)

periods_per_unit = data.groupby(unit)[time].nunique()

expected_n_periods = len(all_periods)

unbalanced_units = periods_per_unit[periods_per_unit != expected_n_periods]

if len(unbalanced_units) > 0:

example_unit = unbalanced_units.index[0]

actual_count = unbalanced_units.iloc[0]

raise ValueError(

f"Panel is not balanced: {len(unbalanced_units)} unit(s) do not "

f"have observations in all {expected_n_periods} periods "

f"(e.g., unit '{example_unit}' has {actual_count} periods). "

f"SyntheticDiD requires a balanced panel. Use "

f"diff_diff.prep.balance_panel() to balance the panel first."

)

# Validate and extract survey weights. Pweight-only fits feed

# placebo / jackknife / bootstrap via ``w_control`` directly.

# Strata/PSU/FPC fits feed bootstrap via the unit-collapsed

# ``resolved_survey_unit`` (PR #352) which Rao-Wu rescaling

# consumes per draw.

if resolved_survey is not None:

_validate_unit_constant_survey(data, unit, survey_design)

# Collapse to unit level for the bootstrap survey path. The

# row order is [control_units..., treated_units...] so

# boot_rw[:n_control] / boot_rw[n_control:] line up with the

# bootstrap loop's column ordering. See

# `collapse_survey_to_unit_level` in diff_diff/survey.py.

# Use `data` (not `working_data`) for the groupby — survey

# design columns are unit-constant (validated above) and

# covariate residualization doesn't shuffle row order, so the

# collapse is invariant to which view we group on.

from diff_diff.survey import collapse_survey_to_unit_level

all_units_for_bootstrap = list(control_units) + list(treated_units)

resolved_survey_unit = collapse_survey_to_unit_level(

resolved_survey,

data,

unit,

all_units_for_bootstrap,

)

# Front-door FPC validation for implicit-PSU Rao-Wu (PR #355

# R8 P1). When psu is None but fpc is set,

# ``generate_rao_wu_weights`` (bootstrap_utils.py L654-L655)

# treats each unit as its own PSU and rejects

# ``FPC < n_units`` per stratum mid-draw. ``_bootstrap_se``

# catches that ``ValueError`` and keeps retrying, so the user

# sees a generic bootstrap-exhaustion message instead of a

# targeted FPC/design error. Validate upstream so the user

# gets a clean error before the bootstrap loop even starts.

#

# R10 P1 fix: gate this validator on variance methods that

# actually use FPC. Bootstrap (Rao-Wu) and jackknife (Rust

# & Rao stratum aggregation) both consume FPC; placebo is

# documented as FPC-no-op (Pesarin 2001 §1.5 — permutation

# tests condition on the observed sample). Running the

# validator on placebo would block legitimate placebo fits

# for a constraint that doesn't apply to permutation math.

if (

self.variance_method in ("bootstrap", "jackknife")

and resolved_survey_unit.psu is None

and resolved_survey_unit.fpc is not None

):

if resolved_survey_unit.strata is None:

n_units_total = len(resolved_survey_unit.weights)

fpc_val = float(resolved_survey_unit.fpc[0])

if fpc_val < n_units_total:

raise ValueError(

f"FPC ({fpc_val}) is less than the number of "

f"units ({n_units_total}). With no explicit "

"psu= column, SDID Rao-Wu treats each unit as "

"its own PSU; FPC must be >= the number of "

"units. Declare an explicit psu= column or "

"increase FPC."

)

else:

unique_strata = np.unique(resolved_survey_unit.strata)

for h in unique_strata:

mask_h = resolved_survey_unit.strata == h

n_h_units = int(mask_h.sum())

fpc_h = float(resolved_survey_unit.fpc[mask_h][0])

if fpc_h < n_h_units:

raise ValueError(

f"FPC ({fpc_h}) in stratum {h} is less than "

f"the number of units in that stratum "

f"({n_h_units}). With no explicit psu= "

"column, SDID Rao-Wu treats each unit as "

"its own PSU within strata; FPC must be "

">= the per-stratum unit count. Declare an "

"explicit psu= column or increase FPC."

)

# Source w_control / w_treated from resolved_survey_unit.weights

# rather than re-extracting raw panel columns. resolved_survey.weights

# is normalized to mean=1 by SurveyDesign.resolve() (survey.py L189-

# L203), so the weighted-FW bootstrap objective — which is NOT

# invariant to a global rescaling of rw — produces identical SE /

# p-value / CI under SurveyDesign(weights="w") vs "c*w" (PR #355

# R4 P0). Placebo / jackknife paths also consume w_control /

# w_treated but are scale-invariant (np.average divides by sum;

# ω_eff normalization likewise), so switching to resolved weights

# doesn't change their numerics.

n_control_for_split = len(control_units)

w_control = resolved_survey_unit.weights[:n_control_for_split].astype(np.float64)

w_treated = resolved_survey_unit.weights[n_control_for_split:].astype(np.float64)

# Front-door positive-mass guard (PR #355 R7 P1). Survey weights

# are non-negative post-resolve() (survey.py L171-L176 rejects

# negatives), but all-zero mass on either arm is reachable — the

# user can assign unit survey weights of 0 to every treated or

# every control unit, which encodes an unidentified target

# population. The fit-time ATT formulas downstream

# (``np.average(..., weights=w_treated)`` around L551-L582 and

# ``omega_eff = unit_weights * w_control`` in the bootstrap /

# placebo / jackknife dispatchers) would otherwise hit 0/0

# normalization or propagate NaNs silently. The bootstrap loop

# already has per-draw zero-mass retries for degenerate resamples

# (PR #355 R2 P0); this guard is the fit-time analogue.

if w_control.sum() <= 0:

raise ValueError(

"Survey-weighted control arm has zero total mass "

f"(sum of w_control = {w_control.sum():.3g}). "

"Every control unit has survey weight 0, so the target "

"population is unidentified. Drop units with zero weight, "

"or omit survey_design if unweighted estimation is intended."

)

if w_treated.sum() <= 0:

raise ValueError(

"Survey-weighted treated arm has zero total mass "

f"(sum of w_treated = {w_treated.sum():.3g}). "

"Every treated unit has survey weight 0, so the target "

"population is unidentified. Drop units with zero weight, "

"or omit survey_design if unweighted estimation is intended."

)

else:

w_treated = None

w_control = None

resolved_survey_unit = None

# Residualize covariates if provided

working_data = data.copy()

if covariates:

working_data = self._residualize_covariates(

working_data,

outcome,

covariates,

unit,

time,

survey_weights=survey_weights,

survey_weight_type=survey_weight_type,

)

# Create outcome matrices

# Shape: (n_periods, n_units)

Y_pre_control, Y_post_control, Y_pre_treated, Y_post_treated = (

self._create_outcome_matrices(

working_data,

outcome,

unit,

time,

pre_periods,

post_periods,

treated_units,

control_units,

)

)

# --- Y normalization ---------------------------------------------

# τ is location-invariant and scale-equivariant in Y. Normalizing Y

# once before weight optimization, the estimator, and the variance

# procedures (and rescaling τ/SE/CI/effects by Y_scale at the end)

# is mathematically a no-op but prevents ~6-digit precision loss in

# the SDID double-difference when outcomes span millions-to-billions.

# Normalization constants come from controls' pre-period only so the

# reference is unaffected by treatment. See REGISTRY.md §SyntheticDiD

# edge cases and synth-inference/synthdid#71 for R's version.

Y_shift = float(np.mean(Y_pre_control))

Y_scale_raw = float(np.std(Y_pre_control))

# Relative floor: avoid amplifying roundoff when std is tiny but

# nonzero (near-constant Y_pre_control). Fall back to 1.0 in that

# case and on non-finite std.

_scale_floor = 1e-12 * max(abs(Y_shift), 1.0)

Y_scale = Y_scale_raw if np.isfinite(Y_scale_raw) and Y_scale_raw > _scale_floor else 1.0

Y_pre_control_n = (Y_pre_control - Y_shift) / Y_scale

Y_post_control_n = (Y_post_control - Y_shift) / Y_scale

Y_pre_treated_n = (Y_pre_treated - Y_shift) / Y_scale

Y_post_treated_n = (Y_post_treated - Y_shift) / Y_scale

# Auto-regularization on normalized Y. FW's argmin is invariant under

# (Y, ζ) -> (Y/s, ζ/s); auto-zetas computed on Y_n are already on the

# normalized scale. User-supplied zetas are in original-Y units and

# are divided by Y_scale for internal FW use. Original-scale values

# are stored on results_ / self so diagnostic methods (in_time_placebo,

# sensitivity_to_zeta_omega) — which operate on the stored original-

# scale fit snapshot — see the same zeta the user specified.

auto_zeta_omega_n, auto_zeta_lambda_n = _compute_regularization(

Y_pre_control_n, len(treated_units), len(post_periods)

)

zeta_omega_n = (

self.zeta_omega / Y_scale if self.zeta_omega is not None else auto_zeta_omega_n

)

zeta_lambda_n = (

self.zeta_lambda / Y_scale if self.zeta_lambda is not None else auto_zeta_lambda_n

)

# Report the user-supplied value exactly (no roundoff from /*Y_scale

# roundtrip); report auto-zeta rescaled to original Y units.

zeta_omega = self.zeta_omega if self.zeta_omega is not None else auto_zeta_omega_n * Y_scale

zeta_lambda = (

self.zeta_lambda if self.zeta_lambda is not None else auto_zeta_lambda_n * Y_scale

)

# Store noise level for diagnostics (reported on original Y scale).

from diff_diff.utils import _compute_noise_level

noise_level_n = _compute_noise_level(Y_pre_control_n)

noise_level = noise_level_n * Y_scale

# Data-dependent convergence threshold (matches R's 1e-5 * noise.level),

# evaluated on normalized Y since FW operates on normalized Y. Floor of

# 1e-5 when noise is zero: R would use 0.0, causing FW to run all

# max_iter iterations; the floor enables early stop on zero-variation

# inputs without changing the optimum.

min_decrease = 1e-5 * noise_level_n if noise_level_n > 0 else 1e-5

# Compute unit weights (Frank-Wolfe with sparsification) on normalized Y.

# Survey weights enter via the treated mean target.

if w_treated is not None:

Y_pre_treated_mean_n = np.average(Y_pre_treated_n, axis=1, weights=w_treated)

else:

Y_pre_treated_mean_n = np.mean(Y_pre_treated_n, axis=1)

unit_weights = compute_sdid_unit_weights(

Y_pre_control_n,

Y_pre_treated_mean_n,

zeta_omega=zeta_omega_n,

min_decrease=min_decrease,

)

# Compute time weights (Frank-Wolfe on collapsed form) on normalized Y.

time_weights = compute_time_weights(

Y_pre_control_n,

Y_post_control_n,

zeta_lambda=zeta_lambda_n,

min_decrease=min_decrease,

)

# Compose ω with control survey weights (WLS regression interpretation).

# Frank-Wolfe finds best trajectory match; survey weights reweight by

# population importance post-optimization.

if w_control is not None:

omega_eff = unit_weights * w_control

# Front-door effective-control guard (PR #355 R12 P1). The R7 P1

# check (``w_control.sum() > 0`` at the raw level, synthetic_did.py

# L470-L499) is insufficient here: Frank-Wolfe sparsifies

# ``unit_weights`` to exact zeros by design, so even when at least

# one control has positive survey weight, FW may concentrate all

# mass on a subset whose survey weights are all 0. The composed

# vector ``unit_weights * w_control`` would then sum to 0, the

# downstream ``omega_eff / omega_eff.sum()`` would emit NaN, and

# the fit would return NaN ATT / SE silently. The analogous

# guards already exist for the bootstrap loop

# (``omega_scaled.sum() <= 0`` retry) and jackknife

# (``effective_control > 0`` support gate); this restores the

# contract at fit time.

omega_eff_sum = float(omega_eff.sum())

if omega_eff_sum <= 0:

raise ValueError(

"SDID point estimate is unidentified: the Frank-Wolfe "

"solution concentrates all synthetic-control mass on "

"units with zero survey weight, so the composed "

"omega_eff = unit_weights * w_control sums to "

f"{omega_eff_sum:.3g}. Every control unit with positive "

"fit-time weight has survey weight 0. Drop zero-weight "

"controls, or omit survey_design if unweighted "

"estimation is intended."

)

omega_eff = omega_eff / omega_eff_sum

else:

omega_eff = unit_weights

# Compute SDID estimate on normalized Y, then rescale to original units.

if w_treated is not None:

Y_post_treated_mean_n = np.average(Y_post_treated_n, axis=1, weights=w_treated)

else:

Y_post_treated_mean_n = np.mean(Y_post_treated_n, axis=1)

att_n = compute_sdid_estimator(

Y_pre_control_n,

Y_post_control_n,

Y_pre_treated_mean_n,

Y_post_treated_mean_n,

omega_eff,

time_weights,

)

att = att_n * Y_scale

# Recover original-scale treated means for diagnostics / trajectories.

Y_pre_treated_mean = Y_pre_treated_mean_n * Y_scale + Y_shift

Y_post_treated_mean = Y_post_treated_mean_n * Y_scale + Y_shift

# Compute pre-treatment fit (RMSE) using composed weights on the

# original Y (user-visible scale). omega_eff is a simplex — applies

# cleanly to any linear rescale of Y — so trajectories live on the

# original outcome scale for plotting and the poor-fit warning.

synthetic_pre_trajectory = Y_pre_control @ omega_eff

synthetic_post_trajectory = Y_post_control @ omega_eff

pre_fit_rmse = np.sqrt(np.mean((Y_pre_treated_mean - synthetic_pre_trajectory) ** 2))

# Warn if pre-treatment fit is poor (Registry requirement).

# Threshold: 1× SD of treated pre-treatment outcomes — a natural baseline

# since RMSE exceeding natural variation indicates the synthetic control

# fails to reproduce the treated series' level or trend.

pre_treatment_sd = (

np.std(Y_pre_treated_mean, ddof=1) if len(Y_pre_treated_mean) > 1 else 0.0

)

if pre_treatment_sd > 0 and pre_fit_rmse > pre_treatment_sd:

warnings.warn(

f"Pre-treatment fit is poor: RMSE ({pre_fit_rmse:.4f}) exceeds "

f"the standard deviation of treated pre-treatment outcomes "

f"({pre_treatment_sd:.4f}). The synthetic control may not "

f"adequately reproduce treated unit trends. Consider adding "

f"more control units or adjusting regularization.",

UserWarning,

stacklevel=2,

)

# Treated-unit trajectories (the pre/post means already computed above).

treated_pre_trajectory = Y_pre_treated_mean

treated_post_trajectory = Y_post_treated_mean

# Detect full-design survey (strata/PSU/FPC). The unit-collapsed

# ``resolved_survey_unit`` carries the per-unit strata/psu/fpc

# arrays ordered as [control..., treated...] to match the

# downstream variance-method column layout.

_full_design_survey = resolved_survey_unit is not None and (

resolved_survey_unit.strata is not None

or resolved_survey_unit.psu is not None

or resolved_survey_unit.fpc is not None

)

if _full_design_survey:

_n_c = len(control_units)

_strata_control = (

resolved_survey_unit.strata[:_n_c]

if resolved_survey_unit.strata is not None

else None

)

_strata_treated = (

resolved_survey_unit.strata[_n_c:]

if resolved_survey_unit.strata is not None

else None

)

_psu_control = (

resolved_survey_unit.psu[:_n_c] if resolved_survey_unit.psu is not None else None

)

_psu_treated = (

resolved_survey_unit.psu[_n_c:] if resolved_survey_unit.psu is not None else None

)

_fpc_control = (

resolved_survey_unit.fpc[:_n_c] if resolved_survey_unit.fpc is not None else None

)

_fpc_treated = (

resolved_survey_unit.fpc[_n_c:] if resolved_survey_unit.fpc is not None else None

)

else:

_strata_control = None

_strata_treated = None

_psu_control = None

_psu_treated = None

_fpc_control = None

_fpc_treated = None

# Placebo routes to the survey allocator whenever **strata or

# PSU** is declared (FPC alone does NOT flip dispatch). For

# PSU-without-strata designs, the whole panel is synthesized

# as a single stratum (stratified permutation degenerates to

# global within-stratum permutation, still dispatched through

# the weighted-FW path).

#

# FPC handling on placebo (R8 P1 fix): permutation tests are

# conditional on the observed sample (Pesarin 2001 §1.5), so

# the sampling fraction does not enter Algorithm 4 or its

# stratified-permutation extension. Including FPC in the

# dispatch trigger would silently switch numerics (weighted-FW

# vs unweighted-FW + post-hoc composition) on a survey design

# element that has no place in the placebo math. Drop FPC from

# the dispatch condition; emit a ``UserWarning`` below if FPC

# is set with placebo to surface the no-op contract.

_placebo_use_survey_path = (

self.variance_method == "placebo"

and resolved_survey_unit is not None

and (resolved_survey_unit.strata is not None or resolved_survey_unit.psu is not None)

)

# NOTE: the FPC no-op warning for placebo is emitted earlier

# (before ``_resolve_survey_for_fit``); ``resolved_survey_unit.fpc``

# is already None on the placebo path because the FPC column is

# dropped from a copy of the survey design pre-resolve. No

# duplicate warning here.

# Jackknife routes to the survey allocator whenever PSU or FPC or

# strata is declared. PSU-without-strata is treated as a single

# stratum (Rust & Rao 1996 JK1 form) inside

# ``_jackknife_se_survey``.

_jackknife_use_survey_path = _full_design_survey and self.variance_method == "jackknife"

# Synthesize a single stratum for PSU/FPC-without-strata designs

# so the placebo / jackknife survey paths can treat them as the

# JK1 / global-permutation degenerate case of the stratified

# allocator. The `_strata_*_eff` arrays are passed to the survey

# methods; the original `_strata_*` arrays stay None so other

# code paths (REGISTRY, metadata) see the true design.

if _full_design_survey and _strata_control is None:

_strata_control_eff: np.ndarray = np.zeros(len(control_units), dtype=np.int64)

_strata_treated_eff: np.ndarray = np.zeros(len(treated_units), dtype=np.int64)

else:

_strata_control_eff = _strata_control # type: ignore[assignment]

_strata_treated_eff = _strata_treated # type: ignore[assignment]

# Fit-time feasibility guard for stratified-permutation placebo

# (per `feedback_front_door_over_retry_swallow.md`). Case B / Case C

# are hard failures — partial-permutation fallback would silently

# change the null-distribution semantics and produce an incoherent

# test. Must run *before* the retry loop below swallows ValueErrors

# via `except (ValueError, LinAlgError, ZeroDivisionError): continue`.

if _placebo_use_survey_path:

unique_treated_strata, treated_counts = np.unique(

_strata_treated_eff, return_counts=True

)

has_nondegenerate_stratum = False

assert w_control is not None # always set on full-design survey

for h, n_t_h in zip(unique_treated_strata, treated_counts):

n_c_h = int(np.sum(_strata_control_eff == h))

if n_c_h == 0:

raise ValueError(

"Stratified-permutation placebo requires at least "

f"one control per stratum containing treated units; "

f"stratum {h} has 0 controls and {int(n_t_h)} "

"treated units. Either rebalance the panel, drop "

f"stratum {h} from the design, or use "

"variance_method='bootstrap' (which supports the "

"same full survey design via weighted-FW + Rao-Wu "

"without a permutation-feasibility constraint)."

)

if n_c_h < int(n_t_h):

raise ValueError(

"Stratified-permutation placebo requires at least "

"n_treated controls per stratum containing treated "

"units (for exact-count within-stratum "

f"permutation); stratum {h} has {n_c_h} controls "

f"but {int(n_t_h)} treated units. Either rebalance "

"the panel, drop the undersupplied stratum, or use "

"variance_method='bootstrap' (which supports the "

"same full survey design via weighted-FW + Rao-Wu "

"without a permutation-feasibility constraint)."

)

# Case E (R9 P1) — row-count guards passed (n_c_h ≥ n_t_h)

# but the stratum has fewer positive-weight controls

# than treated. The placebo allocator computes pseudo-

# treated means as ``np.average(Y, weights=w_control[idx])``;

# if too few controls have positive weight, draws can

# pick all-zero-weight subsets (ZeroDivisionError on

# np.average) and the retry loop swallows them as a

# generic ``n_successful=0`` warning + ``SE=0.0``.

# Front-door the targeted error.

w_in_h = w_control[_strata_control_eff == h]

n_c_h_positive = int(np.sum(w_in_h > 0))

if n_c_h_positive < int(n_t_h):

raise ValueError(

"Stratified-permutation placebo requires at least "

"n_treated controls with positive survey weight "

"per stratum containing treated units (the "

"pseudo-treated mean uses survey-weighted "

f"averaging); stratum {h} has {n_c_h_positive} "

f"positive-weight controls (out of {n_c_h} total) "

f"but {int(n_t_h)} treated units. Either rebalance "

"the panel, drop the undersupplied stratum, or use "

"variance_method='bootstrap' (which supports the "

"same full survey design via weighted-FW + Rao-Wu "

"without a per-draw positive-mass constraint)."

)

# Non-degenerate iff this stratum yields ≥2 distinct

# positive-mass pseudo-treated draws. Two necessary

# conditions, both required:

# * ``n_c_h > n_t_h`` — raw without-replacement count

# allows multiple subsets (otherwise only the

# "all-controls-as-pseudo-treated" subset exists,

# regardless of weights — Case D classical shape).

# * ``n_c_h_positive >= 2`` — at least 2 distinct

# positive-mass means are reachable. With only 1

# positive-weight control, every successful pick

# reduces to that single control's mean (zero-

# weight cohabitants contribute 0 to numerator and

# denominator), regardless of how many subsets the

# raw allocator can construct (Case D effective

# single-support shape, R11 P1).

if n_c_h > int(n_t_h) and n_c_h_positive >= 2:

has_nondegenerate_stratum = True