外观
DID 分组趋势
双重差分的分组趋势与事件研究分析。
核心代码
py
def _fit_parallel_group_trends(self, decimals: int, title: str) -> Any:
if not self.y_var:
raise ValueError(_('单时点组间平行趋势检验需要选择因变量'))
treat_var = self.did_options.get('treat_var')
entity_col = self.panel_ids.get('entity')
time_col = self._get_time_column()
if not treat_var:
raise ValueError(_('单时点组间平行趋势检验需要选择处理组变量'))
policy_year_var = self.did_options.get('policy_year_var') or None
policy_year_value = self._get_policy_year_value()
did_signal_var = self.did_options.get('did_var') or None
if not policy_year_var and policy_year_value is None and did_signal_var and not entity_col:
raise ValueError(_('未选择 policy_year 时，若要用 DID 信号变量自动识别政策时点，需要选择个体变量和时间变量'))
cols = [self.y_var, treat_var, did_signal_var, policy_year_var, entity_col, time_col]
df = self._numeric_clean_frame(
cols,
numeric_cols=[self.y_var, treat_var] + ([did_signal_var] if did_signal_var else []),
required_cols=[self.y_var, treat_var, time_col]
+ ([entity_col] if did_signal_var and not policy_year_var and policy_year_value is None else [])
+ ([did_signal_var] if did_signal_var and not policy_year_var and policy_year_value is None else [])
)
df['__did_event_time_base__'], note = self._event_time_numeric(df[time_col])
if note:
warnings.warn(note, UserWarning)
if entity_col:
df['__did_entity__'] = normalize_grouping_series(df[entity_col])
if policy_year_value is not None:
policy_series = pd.Series(policy_year_value, index=df.index, name='policy_year')
time_values, policy_values, note = self._coerce_time_and_policy(df[time_col], policy_series)
df['__did_event_time_base__'] = time_values
df['__did_policy_time__'] = policy_values
if note:
warnings.warn(note, UserWarning)
policy_source = _('使用用户手动填写的固定政策时点：%(policy_year_value)s') % {'policy_year_value': policy_year_value}
policy_marker = float(df['__did_policy_time__'].dropna().iloc[0]) if df['__did_policy_time__'].notna().any() else np.nan
elif policy_year_var:
time_values, policy_values, note = self._coerce_time_and_policy(df[time_col], df[policy_year_var])
df['__did_event_time_base__'] = time_values
df['__did_policy_time__'] = policy_values
if note:
warnings.warn(note, UserWarning)
policy_source = _('使用用户选择的 policy_year 变量：%(policy_year_var)s') % {'policy_year_var': policy_year_var}
policy_candidates = df['__did_policy_time__'].dropna().unique()
if len(policy_candidates) == 0:
policy_marker = np.nan
else:
policy_marker = float(np.nanmin(policy_candidates))
if len(policy_candidates) > 1:
warnings.warn(
_('单时点组间趋势图检测到多个 policy_year，图中的政策竖线使用最早政策时间；若政策分批发生，建议使用多时点事件研究平行趋势检验。'),
UserWarning,
)
elif did_signal_var:
policy_marker = self._infer_policy_marker_from_did(df, did_signal_var)
policy_source = _('未选择 policy_year，已按个体和时间变量使用 %(did_signal_var)s 首次取 1 的时间推断政策时点。') % {'did_signal_var': did_signal_var}
warnings.warn(policy_source, UserWarning)
else:
policy_marker = np.nan
policy_source = _('未选择 policy_year，也未选择 DID 信号变量，组间趋势图不绘制政策竖线。')
warnings.warn(policy_source, UserWarning)
if entity_col:
df['__did_group__'] = df.groupby('__did_entity__')[treat_var].transform(
lambda s: int((pd.to_numeric(s, errors='coerce').fillna(0.0) > 0).any())
)
else:
df['__did_group__'] = (pd.to_numeric(df[treat_var], errors='coerce').fillna(0.0) > 0).astype(int)
if df['__did_group__'].nunique() != 2:
raise ValueError(_('处理组变量无法同时识别处理组和对照组'))
summary = (
df.groupby(['__did_event_time_base__', '__did_group__'])[self.y_var]
.agg(['mean', 'count'])
.reset_index()
)
mean_pivot = summary.pivot(index='__did_event_time_base__', columns='__did_group__', values='mean')
count_pivot = summary.pivot(index='__did_event_time_base__', columns='__did_group__', values='count')
trend_rows = []
for time_value in sorted(mean_pivot.index.tolist()):
control_mean = mean_pivot.get(0, pd.Series(dtype=float)).get(time_value, np.nan)
treated_mean = mean_pivot.get(1, pd.Series(dtype=float)).get(time_value, np.nan)
control_n = count_pivot.get(0, pd.Series(dtype=float)).get(time_value, 0)
treated_n = count_pivot.get(1, pd.Series(dtype=float)).get(time_value, 0)
trend_rows.append({
'时间': time_value,
'处理组均值': treated_mean,
'对照组均值': control_mean,
'组间差异': treated_mean - control_mean if np.isfinite(treated_mean) and np.isfinite(control_mean) else np.nan,
'处理组N': int(treated_n) if pd.notna(treated_n) else 0,
'对照组N': int(control_n) if pd.notna(control_n) else 0,
})
trend_df = pd.DataFrame(trend_rows)
header = title or _('单时点组间平行趋势')
source_df = pd.DataFrame([
{'项目': _('政策时点来源'), '取值': policy_source},
{'项目': _('图中政策竖线'), '取值': policy_marker if np.isfinite(policy_marker) else _('未识别')},
{'项目': _('说明'), '取值': _('该检验展示处理组与对照组被解释变量绝对水平随时间的变化，不估计事件期回归系数。')},
])
self.custom_html = (
self._render_dataframe_table(source_df, _('%(header)s - 设定说明') % {'header': header}, decimals=decimals)
+ self._render_dataframe_table(trend_df, _('%(header)s - 组间均值趋势表') % {'header': header}, decimals=decimals)
)
chart = self._build_group_trend_chart(trend_df, policy_marker, header)
if chart:
self.chart_images.append(chart)
self.model_stats = {'N': int(len(df))}
self.raw_output = trend_df.to_string(index=False)
self.result = self
return self.result