外观
链式中介
多中介变量的链式中介效应分析。
核心代码
py
def _fit_chain_mediation_analysis(
self,
df: pd.DataFrame,
entity_col: str,
time_col: str,
decimals: int,
title: str,
) -> Any:
x_var, controls = self._resolve_core_x_and_controls()
mediator_vars = self._unique_preserve_order(self.mediator_vars or self.chain_mediation_options.get('mediator_vars', []))
if len(mediator_vars) != 2:
raise ValueError(_('链式中介必须选择 2 个按顺序进入模型的中介变量。'))
mediator1, mediator2 = mediator_vars
model_kind, absorb_vars, panel_dims, base_rows = self._resolve_advanced_model_spec()
use_bootstrap = bool(self.chain_mediation_options.get('use_bootstrap'))
bootstrap_reps = int(self.chain_mediation_options.get('bootstrap_reps', 500) or 500)
bootstrap_progress_floor = max(1.0, bootstrap_reps * 0.20) if use_bootstrap else 0.0
bootstrap_prelude_steps = [
_('准备总效应模型'),
_('准备 %(mediator1)s 模型') % {'mediator1': mediator1},
_('准备 %(mediator2)s 模型') % {'mediator2': mediator2},
_('准备联合结果模型'),
]
def report_chain_bootstrap_prelude(step_index: int) -> None:
if not use_bootstrap or not callable(self.progress_callback):
return
step_label = bootstrap_prelude_steps[max(0, step_index - 1)]
self.progress_callback({
'completed': 0,
'total': bootstrap_reps,
'successful': 0,
'skipped': 0,
'current_iteration': 0,
'current_stage_index': step_index,
'current_stage_total': len(bootstrap_prelude_steps),
'current_stage_label': step_label,
'progress_value': max(1.0, bootstrap_reps * 0.05 * step_index),
'message': _('正在准备链式中介 Bootstrap：%(step_label)s') % {'step_label': step_label},
})
report_chain_bootstrap_prelude(1)
total_fit = self._fit_linear_model(
df,
self.y_var,
[x_var] + controls,
model_kind=model_kind,
absorb_vars=absorb_vars,
panel_dims=panel_dims,
header=_('链式中介 - 路径 c（总效应）'),
)
report_chain_bootstrap_prelude(2)
mediator1_fit = self._fit_linear_model(
df,
mediator1,
[x_var] + controls,
model_kind=model_kind,
absorb_vars=absorb_vars,
panel_dims=panel_dims,
header=_('链式中介 - 方程 1（X→M1）'),
)
report_chain_bootstrap_prelude(3)
mediator2_fit = self._fit_linear_model(
df,
mediator2,
[x_var, mediator1] + controls,
model_kind=model_kind,
absorb_vars=absorb_vars,
panel_dims=panel_dims,
header=_('链式中介 - 方程 2（X,M1→M2）'),
)
report_chain_bootstrap_prelude(4)
outcome_fit = self._fit_linear_model(
df,
self.y_var,
[x_var, mediator1, mediator2] + controls,
model_kind=model_kind,
absorb_vars=absorb_vars,
panel_dims=panel_dims,
header=_('链式中介 - 方程 3（X,M1,M2→Y）'),
)
self._result_model_snapshots = []
self._store_result_model_snapshot(
method='chain_total',
method_label=_('路径 c：总效应'),
y_name=self.y_var,
x_vars=[x_var] + controls,
params=total_fit['params'],
std_errors=total_fit['std_errors'],
test_stats=total_fit['test_stats'],
pvalues=total_fit['pvalues'],
stats=total_fit['stats'],
custom_rows=base_rows,
)
self._store_result_model_snapshot(
method='chain_m1',
method_label=_('方程 1：X→M1'),
y_name=mediator1,
x_vars=[x_var] + controls,
params=mediator1_fit['params'],
std_errors=mediator1_fit['std_errors'],
test_stats=mediator1_fit['test_stats'],
pvalues=mediator1_fit['pvalues'],
stats=mediator1_fit['stats'],
custom_rows=base_rows,
)
self._store_result_model_snapshot(
method='chain_m2',
method_label=_('方程 2：X,M1→M2'),
y_name=mediator2,
x_vars=[x_var, mediator1] + controls,
params=mediator2_fit['params'],
std_errors=mediator2_fit['std_errors'],
test_stats=mediator2_fit['test_stats'],
pvalues=mediator2_fit['pvalues'],
stats=mediator2_fit['stats'],
custom_rows=base_rows,
)
self._store_result_model_snapshot(
method='chain_outcome',
method_label=_('方程 3：X,M1,M2→Y'),
y_name=self.y_var,
x_vars=[x_var, mediator1, mediator2] + controls,
params=outcome_fit['params'],
std_errors=outcome_fit['std_errors'],
test_stats=outcome_fit['test_stats'],
pvalues=outcome_fit['pvalues'],
stats=outcome_fit['stats'],
custom_rows=base_rows,
)
a1 = float(mediator1_fit['params'].get(x_var, np.nan))
a2 = float(mediator2_fit['params'].get(x_var, np.nan))
d21 = float(mediator2_fit['params'].get(mediator1, np.nan))
b1 = float(outcome_fit['params'].get(mediator1, np.nan))
b2 = float(outcome_fit['params'].get(mediator2, np.nan))
total_effect = float(total_fit['params'].get(x_var, np.nan))
direct_effect = float(outcome_fit['params'].get(x_var, np.nan))
indirect_m1 = a1 * b1
indirect_m2 = a2 * b2
indirect_chain = a1 * d21 * b2
total_indirect = indirect_m1 + indirect_m2 + indirect_chain
summary_rows = [
{'项目': f'{x_var}→{mediator1}→{self.y_var}', '估计值': indirect_m1},
{'项目': f'{x_var}→{mediator2}→{self.y_var}', '估计值': indirect_m2},
{'项目': f'{x_var}→{mediator1}→{mediator2}→{self.y_var}', '估计值': indirect_chain},
{'项目': _('总间接效应'), '估计值': total_indirect},
{'项目': _('直接效应'), '估计值': direct_effect},
{'项目': _('总效应'), '估计值': total_effect},
{'项目': _('中介占总效应比例'), '估计值': total_indirect / total_effect if np.isfinite(total_effect) and not np.isclose(total_effect, 0.0) else np.nan},
]
html_parts = [
'<div style="margin-top: 30px;"></div>',
self._render_dataframe_table(pd.DataFrame(summary_rows), _('%(title)s - 效应分解') % {'title': title or _('链式中介')}, decimals=decimals),
]
use_bootstrap = bool(self.chain_mediation_options.get('use_bootstrap'))
bootstrap_reps = int(self.chain_mediation_options.get('bootstrap_reps', 500) or 500)
if use_bootstrap:
bootstrap_progress_floor = max(1.0, bootstrap_reps * 0.20)
if callable(self.progress_callback):
prelude_steps = [
_('准备总效应模型'),
_('准备 %(mediator1)s 模型') % {'mediator1': mediator1},
_('准备 %(mediator2)s 模型') % {'mediator2': mediator2},
_('准备联合结果模型'),
]
for step_index, step_label in enumerate(prelude_steps, start=1):
self.progress_callback({
'completed': 0,
'total': bootstrap_reps,
'successful': 0,
'skipped': 0,
'current_iteration': 0,
'current_stage_index': step_index,
'current_stage_total': len(prelude_steps),
'current_stage_label': step_label,
'progress_value': max(1.0, bootstrap_reps * 0.05 * step_index),
'message': _('正在准备链式中介 Bootstrap：%(step_label)s') % {'step_label': step_label},
})
def estimate_chain_effects(sample_df: pd.DataFrame) -> Dict[str, float]:
self._notify_bootstrap_stage(1, 4, _('总效应回归'))
sample_total = self._fit_linear_model(sample_df, self.y_var, [x_var] + controls, model_kind=model_kind, absorb_vars=absorb_vars, panel_dims=panel_dims)
self._notify_bootstrap_stage(2, 4, _('%(mediator1)s 回归') % {'mediator1': mediator1})
sample_m1 = self._fit_linear_model(sample_df, mediator1, [x_var] + controls, model_kind=model_kind, absorb_vars=absorb_vars, panel_dims=panel_dims)
self._notify_bootstrap_stage(3, 4, _('%(mediator2)s 回归') % {'mediator2': mediator2})
sample_m2 = self._fit_linear_model(sample_df, mediator2, [x_var, mediator1] + controls, model_kind=model_kind, absorb_vars=absorb_vars, panel_dims=panel_dims)
self._notify_bootstrap_stage(4, 4, _('联合结果回归'))
sample_y = self._fit_linear_model(sample_df, self.y_var, [x_var, mediator1, mediator2] + controls, model_kind=model_kind, absorb_vars=absorb_vars, panel_dims=panel_dims)
s_a1 = float(sample_m1['params'].get(x_var, np.nan))
s_a2 = float(sample_m2['params'].get(x_var, np.nan))
s_d21 = float(sample_m2['params'].get(mediator1, np.nan))
s_b1 = float(sample_y['params'].get(mediator1, np.nan))
s_b2 = float(sample_y['params'].get(mediator2, np.nan))
s_total = float(sample_total['params'].get(x_var, np.nan))
s_direct = float(sample_y['params'].get(x_var, np.nan))
s_indirect_m1 = s_a1 * s_b1
s_indirect_m2 = s_a2 * s_b2
s_indirect_chain = s_a1 * s_d21 * s_b2
return {
'经 M1 的间接效应': s_indirect_m1,
'经 M2 的间接效应': s_indirect_m2,
'链式间接效应': s_indirect_chain,
'总间接效应': s_indirect_m1 + s_indirect_m2 + s_indirect_chain,
'直接效应': s_direct,
'总效应': s_total,
}
bootstrap_df, failures = self._run_bootstrap(
df,
bootstrap_reps,
estimate_chain_effects,
progress_label=_('链式中介 Bootstrap 重复抽样'),
progress_floor=bootstrap_progress_floor,
)
if bootstrap_df.empty:
warnings.warn("链式中介的 Bootstrap 重复抽样全部失败，已仅保留点估计结果。", UserWarning)
else:
bootstrap_rows = []
effect_map = {
'经 M1 的间接效应': indirect_m1,
'经 M2 的间接效应': indirect_m2,
'链式间接效应': indirect_chain,
'总间接效应': total_indirect,
'直接效应': direct_effect,
'总效应': total_effect,
}
for effect_name, point_estimate in effect_map.items():
effect_values = bootstrap_df[effect_name].dropna()
if effect_values.empty:
continue
bootstrap_rows.append({
'效应': effect_name,
'点估计': point_estimate,
'Bootstrap SE': effect_values.std(ddof=1),
'95% CI 下限': effect_values.quantile(0.025),
'95% CI 上限': effect_values.quantile(0.975),
})
html_parts.append(
self._render_dataframe_table(
pd.DataFrame(bootstrap_rows),
_('%(title)s - Bootstrap 检验') % {'title': title or _('链式中介')},
decimals=decimals,
)
)
if failures:
warnings.warn(f"链式中介 Bootstrap 过程中有 {failures} 次重复抽样失败，结果基于成功抽样重复次数生成。", UserWarning)
self.result = outcome_fit['result']
self.model_stats = dict(outcome_fit['stats'])
self.display_x_vars = [x_var, mediator1, mediator2] + controls
self.table_custom_rows = base_rows
self.diagnostic_html = ''.join(html_parts)
self.raw_output = "\n\n".join([
total_fit['raw_output'],
mediator1_fit['raw_output'],
mediator2_fit['raw_output'],
outcome_fit['raw_output'],
])
return self.result