外观
安慰剂检验
时间安慰剂检验(亦支持随机安慰剂)。
核心代码
py
def _fit_time_placebo(self, decimals: int, title: str) -> Any:
if not self.y_var:
raise ValueError(_('时间安慰剂检验需要选择因变量'))
treat_var = self.did_options.get('treat_var')
post_var = self.did_options.get('post_var')
entity_col, time_col = self._get_panel_columns()
if not treat_var:
raise ValueError(_('时间安慰剂检验需要选择处理变量或 DID 变量'))
if not post_var:
raise ValueError(_('时间安慰剂检验需要选择政策后虚拟变量 post'))
shift = max(int(self.did_options.get('placebo_shift', 1)), 1)
controls = self._get_control_vars(exclude=[treat_var, post_var, entity_col, time_col])
absorb_vars = self._get_placebo_absorb_vars(entity_col, time_col)
cols = [self.y_var, treat_var, post_var, entity_col, time_col] + controls + self._cluster_cols()
df = self._numeric_clean_frame(cols, numeric_cols=[self.y_var, treat_var, post_var] + controls)
df = self._add_panel_aliases(df, entity_col, time_col)
df['__did_event_time_base__'], note = self._event_time_numeric(df[time_col])
if note:
warnings.warn(note, UserWarning)
post_series = pd.to_numeric(df[post_var], errors='coerce').fillna(0.0) > 0
if not post_series.any():
raise ValueError(_('post 变量中没有识别到政策后时期(取值应包含 1 或正数)'))
policy_time = float(df.loc[post_series, '__did_event_time_base__'].min())
placebo_policy_time = policy_time - shift
treat_numeric = pd.to_numeric(df[treat_var], errors='coerce').fillna(0.0)
ever_treated = df.groupby('__did_entity__')[treat_var].transform(lambda s: float(pd.to_numeric(s, errors='coerce').fillna(0.0).max() > 0))
treat_varies = df.groupby('__did_entity__')[treat_var].transform(lambda s: pd.to_numeric(s, errors='coerce').fillna(0.0).nunique()).max() > 1
if treat_varies:
df['__did_real__'] = treat_numeric
else:
df['__did_real__'] = ever_treated * post_series.astype(float)
df['__did_placebo_time__'] = (
ever_treated
* (df['__did_event_time_base__'] >= placebo_policy_time).astype(float)
* (df['__did_event_time_base__'] < policy_time).astype(float)
)
pre_df = df[df['__did_event_time_base__'] < policy_time].copy()
if pre_df['__did_placebo_time__'].sum() <= 0:
raise ValueError(_('虚构政策期没有生成有效的安慰剂处理变量,请减小前移期数或检查时间变量'))
real_result, _unused_a, _unused_b = self._fit_placebo_regression(
df=df,
y_col=self.y_var,
x_cols=['__did_real__'] + controls,
entity_col=entity_col,
time_col=time_col,
absorb_vars=absorb_vars,
)
placebo_result, _unused_a, _unused_b = self._fit_placebo_regression(
df=pre_df,
y_col=self.y_var,
x_cols=['__did_placebo_time__'] + controls,
entity_col=entity_col,
time_col=time_col,
absorb_vars=absorb_vars,
)
self.result = placebo_result
self._param_label_map['__did_placebo_time__'] = 'Placebo DID'
self._capture_panel_result(placebo_result)
self._set_model_stats(placebo_result)
self.display_x_vars = ['Placebo DID'] + controls
self.table_custom_rows = [
{'label': 'Fixed effects', 'value': self._format_fixed_effects_label(absorb_vars)},
{'label': 'Placebo type', 'value': _('时间前移 %(shift)s 期') % {'shift': shift}},
]
diagnostic_df = pd.DataFrame([
self._coef_summary_row(_('真实 DID'), real_result, '__did_real__', decimals),
self._coef_summary_row(_('时间安慰剂 DID'), placebo_result, '__did_placebo_time__', decimals),
])
placebo_p = self._safe_series_get(placebo_result.pvalues, '__did_placebo_time__')
verdict = _('通过:安慰剂 DID 不显著') if np.isfinite(placebo_p) and placebo_p >= 0.1 else _('需关注:安慰剂 DID 显著')
summary_df = pd.DataFrame([
{'项目': _('真实政策期'), '取值': policy_time},
{'项目': _('虚构政策期'), '取值': placebo_policy_time},
{'项目': _('真实政策前样本量'), '取值': int(len(pre_df))},
{'项目': _('判定'), '取值': verdict},
])
header = title or _('时间安慰剂检验')
self.diagnostic_html = (
self._render_dataframe_table(summary_df, _('%(header)s - 设定摘要') % {'header': header}, decimals=decimals)
+ self._render_dataframe_table(diagnostic_df, _('%(header)s - 系数对比') % {'header': header}, decimals=decimals)
)
self.raw_output = (
_('真实 DID 回归\n')
+ str(real_result.summary)
+ _('\n\n时间安慰剂回归\n')
+ str(placebo_result.summary)
)
return self.result