外观
中介效应
中介效应检验，Bootstrap 置信区间。
核心代码
py
def _fit_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_var = str(self.mediator_var or '').strip()
if not mediator_var:
raise ValueError(_('中介机制必须选择 1 个中介变量。'))
model_kind, absorb_vars, panel_dims, base_rows = self._resolve_advanced_model_spec()
use_bootstrap = bool(self.mediation_options.get('use_bootstrap'))
bootstrap_reps = int(self.mediation_options.get('bootstrap_reps', 500) or 500)
bootstrap_progress_floor = max(1.0, bootstrap_reps * 0.15) if use_bootstrap else 0.0
bootstrap_prelude_steps = [
_('准备总效应模型'),
_('准备中介变量模型'),
_('准备直接效应模型'),
]
def report_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_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_bootstrap_prelude(2)
mediator_fit = self._fit_linear_model(
df,
mediator_var,
[x_var] + controls,
model_kind=model_kind,
absorb_vars=absorb_vars,
panel_dims=panel_dims,
header=_('中介机制 - 路径 a（X→M）'),
)
report_bootstrap_prelude(3)
direct_fit = self._fit_linear_model(
df,
self.y_var,
[x_var, mediator_var] + controls,
model_kind=model_kind,
absorb_vars=absorb_vars,
panel_dims=panel_dims,
header=_('中介机制 - 路径 b / c′（Y→X,M）'),
)
self._result_model_snapshots = []
self._store_result_model_snapshot(
method='mediation_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='mediation_a',
method_label=_('路径 a：X→M'),
y_name=mediator_var,
x_vars=[x_var] + controls,
params=mediator_fit['params'],
std_errors=mediator_fit['std_errors'],
test_stats=mediator_fit['test_stats'],
pvalues=mediator_fit['pvalues'],
stats=mediator_fit['stats'],
custom_rows=base_rows,
)
self._store_result_model_snapshot(
method='mediation_bc',
method_label=_('路径 b / c′：Y→X,M'),
y_name=self.y_var,
x_vars=[x_var, mediator_var] + controls,
params=direct_fit['params'],
std_errors=direct_fit['std_errors'],
test_stats=direct_fit['test_stats'],
pvalues=direct_fit['pvalues'],
stats=direct_fit['stats'],
custom_rows=base_rows,
)
a_coef = float(mediator_fit['params'].get(x_var, np.nan))
b_coef = float(direct_fit['params'].get(mediator_var, np.nan))
total_effect = float(total_fit['params'].get(x_var, np.nan))
indirect_effect = a_coef * b_coef
direct_effect = total_effect - indirect_effect
a_se = float(mediator_fit['std_errors'].get(x_var, np.nan))
b_se = float(direct_fit['std_errors'].get(mediator_var, np.nan))
direct_se = float(direct_fit['std_errors'].get(x_var, np.nan))
total_se = float(total_fit['std_errors'].get(x_var, np.nan))
a_var = a_se ** 2 if np.isfinite(a_se) else np.nan
b_var = b_se ** 2 if np.isfinite(b_se) else np.nan
sobel_se = np.sqrt((b_coef ** 2) * a_var + (a_coef ** 2) * b_var) if np.isfinite(a_var) and np.isfinite(b_var) else np.nan
sobel_z = indirect_effect / sobel_se if np.isfinite(sobel_se) and sobel_se > 0 else np.nan
sobel_p = 2 * (1 - norm.cdf(abs(sobel_z))) if np.isfinite(sobel_z) else np.nan
goodman1_se = np.sqrt((b_coef ** 2) * a_var + (a_coef ** 2) * b_var + a_var * b_var) if np.isfinite(a_var) and np.isfinite(b_var) else np.nan
goodman2_se = np.sqrt((b_coef ** 2) * a_var + (a_coef ** 2) * b_var - a_var * b_var) if np.isfinite(a_var) and np.isfinite(b_var) and (b_coef ** 2) * a_var + (a_coef ** 2) * b_var - a_var * b_var >= 0 else np.nan
goodman1_z = indirect_effect / goodman1_se if np.isfinite(goodman1_se) and goodman1_se > 0 else np.nan
goodman2_z = indirect_effect / goodman2_se if np.isfinite(goodman2_se) and goodman2_se > 0 else np.nan
summary_rows = [
{'项目': _('a 系数（X→M）'), '估计值': a_coef, '标准误': a_se, '统计量': mediator_fit['test_stats'].get(x_var, np.nan), 'P-value': mediator_fit['pvalues'].get(x_var, np.nan)},
{'项目': _('b 系数（M→Y）'), '估计值': b_coef, '标准误': b_se, '统计量': direct_fit['test_stats'].get(mediator_var, np.nan), 'P-value': direct_fit['pvalues'].get(mediator_var, np.nan)},
{'项目': _('间接效应（a×b）'), '估计值': indirect_effect, '标准误': sobel_se, '统计量': sobel_z, 'P-value': sobel_p},
{'项目': _('直接效应'), '估计值': direct_effect, '标准误': direct_se, '统计量': direct_effect / direct_se if np.isfinite(direct_se) and direct_se != 0 else np.nan, 'P-value': 2 * (1 - norm.cdf(abs(direct_effect / direct_se))) if np.isfinite(direct_se) and direct_se != 0 else np.nan},
{'项目': _('总效应'), '估计值': total_effect, '标准误': total_se, '统计量': total_fit['test_stats'].get(x_var, np.nan), 'P-value': total_fit['pvalues'].get(x_var, np.nan)},
{'项目': 'Goodman-1', '估计值': indirect_effect, '标准误': goodman1_se, '统计量': goodman1_z, 'P-value': 2 * (1 - norm.cdf(abs(goodman1_z))) if np.isfinite(goodman1_z) else np.nan},
{'项目': 'Goodman-2', '估计值': indirect_effect, '标准误': goodman2_se, '统计量': goodman2_z, 'P-value': 2 * (1 - norm.cdf(abs(goodman2_z))) if np.isfinite(goodman2_z) else np.nan},
{'项目': _('中介占总效应比例'), '估计值': indirect_effect / total_effect if np.isfinite(total_effect) and not np.isclose(total_effect, 0.0) else np.nan, '标准误': np.nan, '统计量': np.nan, 'P-value': np.nan},
{'项目': _('间接 / 直接效应比'), '估计值': indirect_effect / direct_effect if np.isfinite(direct_effect) and not np.isclose(direct_effect, 0.0) else np.nan, '标准误': np.nan, '统计量': np.nan, 'P-value': 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.mediation_options.get('use_bootstrap'))
bootstrap_reps = int(self.mediation_options.get('bootstrap_reps', 500) or 500)
if use_bootstrap:
bootstrap_progress_floor = max(1.0, bootstrap_reps * 0.15)
if callable(self.progress_callback):
prelude_steps = [
_('准备总效应模型'),
_('准备中介变量模型'),
_('准备直接效应模型'),
]
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_effects(sample_df: pd.DataFrame) -> Dict[str, float]:
self._notify_bootstrap_stage(1, 3, _('路径 c 回归'))
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, 3, _('路径 a 回归'))
sample_mediator = self._fit_linear_model(sample_df, mediator_var, [x_var] + controls, model_kind=model_kind, absorb_vars=absorb_vars, panel_dims=panel_dims)
self._notify_bootstrap_stage(3, 3, _('路径 b / c′ 回归'))
sample_direct = self._fit_linear_model(sample_df, self.y_var, [x_var, mediator_var] + controls, model_kind=model_kind, absorb_vars=absorb_vars, panel_dims=panel_dims)
sample_a = float(sample_mediator['params'].get(x_var, np.nan))
sample_b = float(sample_direct['params'].get(mediator_var, np.nan))
sample_total_effect = float(sample_total['params'].get(x_var, np.nan))
sample_indirect = sample_a * sample_b
sample_direct_effect = sample_total_effect - sample_indirect
return {
'间接效应': sample_indirect,
'直接效应': sample_direct_effect,
'总效应': sample_total_effect,
}
bootstrap_df, failures = self._run_bootstrap(
df,
bootstrap_reps,
estimate_effects,
progress_label=_('中介机制 Bootstrap 重复抽样'),
progress_floor=bootstrap_progress_floor,
)
if bootstrap_df.empty:
warnings.warn("Bootstrap 重复抽样全部失败，已仅保留 Sobel/Goodman 检验结果。", UserWarning)
else:
bootstrap_rows = []
for effect_name, point_estimate in [(_('间接效应'), indirect_effect), (_('直接效应'), direct_effect), (_('总效应'), total_effect)]:
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 = direct_fit['result']
self.model_stats = dict(direct_fit['stats'])
self.display_x_vars = [x_var, mediator_var] + controls
self.table_custom_rows = base_rows
self.diagnostic_html = ''.join(html_parts)
self.raw_output = "\n\n".join([
total_fit['raw_output'],
mediator_fit['raw_output'],
direct_fit['raw_output'],
])
return self.result