平行趋势检验

基于事件研究的平行趋势假设检验。
核心代码

    def _fit_parallel_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') or None
        time_col = self._get_time_column()
        if not treat_var:
            raise ValueError(_('平行趋势检验需要选择处理状态变量或 DID 变量'))

        event_min = int(self.did_options.get('event_study_min', -4))
        event_max = int(self.did_options.get('event_study_max', 4))
        event_ref = int(self.did_options.get('event_study_ref', -1))
        if event_min >= event_max:
            raise ValueError(_('事件期最小值必须小于最大值'))
        if not event_min <= event_ref <= event_max:
            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
        post_var = self.did_options.get('post_var') or None
        if not policy_year_var and policy_year_value is None and not entity_col:
            raise ValueError(_('未选择 policy_year 时，多时点事件研究需要选择个体变量和时间变量，用 DID 信号变量自动识别政策时点'))

        absorb_vars = self._get_absorb_vars(entity_col, time_col)
        controls = self._get_control_vars(exclude=[treat_var, policy_year_var, did_signal_var, post_var, entity_col, time_col] + absorb_vars)
        cols = [self.y_var, treat_var, policy_year_var, did_signal_var, post_var, entity_col, time_col] + controls + self._cluster_cols() + absorb_vars
        numeric_cols = [self.y_var, treat_var] + controls
        numeric_cols.extend([var for var in [did_signal_var, post_var] if var])
        required_cols = [self.y_var, treat_var, time_col] + controls + self._cluster_cols() + absorb_vars
        if not policy_year_var and policy_year_value is None:
            required_cols.append(entity_col)
            required_cols.extend([var for var in [did_signal_var, post_var] if var])
        df = self._numeric_clean_frame(cols, numeric_cols=numeric_cols, required_cols=required_cols)
        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:
            ever_by_treat = self._ever_treated_series(df, treat_var, None)
            policy_series = pd.Series(np.nan, index=df.index, name='policy_year', dtype=object)
            policy_series.loc[ever_by_treat] = policy_year_value
            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)
            self._validate_event_treatment_identity(df, treat_var, None, ever_by_treat)
            df['__did_ever_treated__'] = ever_by_treat
            df['__did_event_time__'] = df['__did_event_time_base__'] - df['__did_policy_time__']
            policy_source = _('使用用户手动填写的固定政策时点：%(policy_year_value)s') % {'policy_year_value': policy_year_value}
        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)
            ever_by_treat = self._ever_treated_series(df, treat_var, None)
            if not ever_by_treat.any():
                if entity_col:
                    ever_by_treat = df.groupby('__did_entity__')['__did_policy_time__'].transform(lambda s: s.notna().any()).astype(bool)
                else:
                    ever_by_treat = df['__did_policy_time__'].notna()
                warnings.warn("处理组变量未识别到处理组，已临时使用 policy_year 非缺失个体作为处理组。", UserWarning)
            self._validate_event_treatment_identity(df, treat_var, None, ever_by_treat)
            df['__did_ever_treated__'] = ever_by_treat
            df['__did_event_time__'] = df['__did_event_time_base__'] - df['__did_policy_time__']
            policy_source = _('使用用户选择的 policy_year 变量：%(policy_year_var)s') % {'policy_year_var': policy_year_var}
        else:
            if did_signal_var:
                signal_col = did_signal_var
                signal_label = did_signal_var
            elif post_var:
                df['__did_signal_fallback__'] = (
                    pd.to_numeric(df[treat_var], errors='coerce').fillna(0.0)
                    * pd.to_numeric(df[post_var], errors='coerce').fillna(0.0)
                )
                signal_col = '__did_signal_fallback__'
                signal_label = f"{treat_var} × {post_var}"
            else:
                raise ValueError(_('未选择 policy_year 时，多时点事件研究需要选择 DID 信号变量，用其首次取 1 的时期推断政策时点。'))
            treat_series = pd.to_numeric(df[signal_col], errors='coerce').fillna(0.0) > 0
            first_treat_time = (
                df.loc[treat_series, ['__did_entity__', '__did_event_time_base__']]
                .groupby('__did_entity__')['__did_event_time_base__']
                .min()
            )
            if first_treat_time.empty:
                raise ValueError(_('没有 policy_year 时，需要 DID 信号变量中至少存在取值为 1 或正数的观测'))

            df['__did_first_treat_time__'] = df['__did_entity__'].map(first_treat_time)
            df['__did_ever_treated__'] = df['__did_first_treat_time__'].notna()
            df['__did_event_time__'] = df['__did_event_time_base__'] - df['__did_first_treat_time__']
            policy_source = _('未选择 policy_year，已按个体和时间变量使用 %(signal_label)s 首次取 1 的时间推断政策时点。') % {'signal_label': signal_label}
            warnings.warn(policy_source, UserWarning)

        event_cols: List[str] = []
        event_map: Dict[str, int] = {}
        for event_time in range(event_min, event_max + 1):
            if event_time == event_ref:
                continue
            col = self._event_dummy_name(event_time)
            df[col] = (
                df['__did_ever_treated__']
                & np.isclose(df['__did_event_time__'], float(event_time), atol=1e-8)
            ).astype(float)
            if df[col].sum() > 0:
                event_cols.append(col)
                event_map[col] = event_time
                self._param_label_map[col] = self._event_display_label(event_time)

        if not event_cols:
            raise ValueError(_('当前事件期范围内没有可估计的事件虚拟变量，请检查处理变量或扩大事件期范围'))

        if entity_col and time_col and set(absorb_vars) == {entity_col, time_col} and len(absorb_vars) == 2:
            result, used_df, exog_cols = self._fit_panel_regression(
                df=df,
                y_col=self.y_var,
                x_cols=event_cols + controls,
                entity_col=entity_col,
                time_col=time_col,
            )
        else:
            result, used_df, exog_cols = self._fit_absorbing_regression(
                df=df,
                y_col=self.y_var,
                x_cols=event_cols + controls,
                absorb_vars=absorb_vars,
            )
        self.result = result
        self._capture_panel_result(result)
        self._set_model_stats(result)
        self.display_x_vars = [self._param_label_map.get(col, col) for col in event_cols] + controls
        self.table_custom_rows = [
            {'label': 'Absorbed FE', 'value': ', '.join(absorb_vars) if absorb_vars else 'None'},
            {'label': 'Policy year source', 'value': policy_source},
            {'label': 'Reference period', 'value': self._event_display_label(event_ref)},
        ]

        ci_level = self._get_ci_level()
        ci_multiplier = self._ci_multiplier(ci_level)
        summary_df = pd.DataFrame([
            {'项目': _('政策时点来源'), '取值': policy_source},
            {'项目': _('吸收固定效应'), '取值': ', '.join(absorb_vars) if absorb_vars else _('未选择')},
            {'项目': _('参考期'), '取值': self._event_display_label(event_ref)},
            {'项目': _('置信区间'), '取值': f'{ci_level:g}%'},
            {'项目': _('事件前后分割线'), '取值': _('0期')},
        ])
        event_table = self._build_event_table(
            result=result,
            event_map=event_map,
            event_ref=event_ref,
            ci_multiplier=ci_multiplier,
            ci_level=ci_level,
            decimals=decimals,
            title=title or _('平行趋势检验')
        )
        self.diagnostic_html = (
            self._render_dataframe_table(summary_df, _('%(title)s - 设定说明') % {'title': title or _('平行趋势检验')}, decimals=decimals)
            + event_table
        )

        chart = self._build_event_study_chart(
            result=result,
            event_map=event_map,
            event_ref=event_ref,
            ci_multiplier=ci_multiplier,
            title=self.did_options.get('chart_title') or title or _('平行趋势检验')
        )
        if chart:
            self.chart_images.append(chart)

        self.raw_output = str(result.summary)
        return self.result
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
平行趋势检验 ​

核心代码 ​

平行趋势检验

核心代码
