外观
简单中介
分解X与Y之间的总效应、直接效应及经M传递的间接效应。
计算口径
首版针对连续观测变量的OLS模型。所有路径使用相同完整案例与分类编码，Bootstrap按个案重抽样。控制变量也要有研究依据。
指定一个X、一个M和一个Y。间接效应为a×b；分析设置可显示两条路径回归，并设置200至5000次Bootstrap与随机种子。
重点报告各间接效应的Bootstrap百分位区间是否跨0。总效应显著不是检验间接效应的必要条件；横截面关联本身不能确立时间顺序和因果机制。
同源实现
以下片段来自 core/statistical/mechanisms.py 的 mediation，由镜像脚本按语法树提取。共享辅助函数和分发逻辑包含在完整下载包中。
py
def mediation(data, method, options):
predictor = columns(data, options.get('predictors'), 1, 1)[0]
outcome = options.get('outcome')
mediators = columns(data, options.get('mediators'), 1, 10)
if method == 'stat_mediation_simple' and len(mediators) != 1:
raise ValueError(_('简单中介需要一个中介变量'))
if method == 'stat_mediation_parallel' and len(mediators) < 2:
raise ValueError(_('平行中介至少需要两个中介变量'))
if method == 'stat_mediation_serial' and len(mediators) != 2:
raise ValueError(_('链式中介首版需要按先后顺序指定两个中介变量'))
controls = options.get('controls') or []
if controls:
columns(data, controls)
names = [predictor]+mediators+controls
if len(names) != len(set(names)):
raise ValueError(_('自变量、中介变量和控制变量不能重复'))
categorical = options.get('categorical') or []
if predictor in categorical or any(name in categorical for name in mediators):
raise ValueError(_('首版中介分析的自变量和中介变量须按连续变量处理'))
sample = prepare_sample(data, outcome, names, options)
alpha = number(options, 'alpha', .05, .0001, .25)
repetitions = number(options, 'bootstrap_samples', 1000, 200, 5000, integer=True)
seed = number(options, 'seed', 20260926, 0, 4294967295, integer=True)
# 同一个编码矩阵用于全部方程和重抽样，分类控制变量的参照组不会漂移。
matrix, terms, references = design_matrix(sample, names, options)
position = {term['variable']: i for i, term in enumerate(terms) if term['kind'] == 'continuous'}
control_positions = [i for i, term in enumerate(terms) if i == 0 or term['variable'] in controls]
x_position = position[predictor]
mediator_positions = [position[name] for name in mediators]
equations = [('total', outcome, control_positions+[x_position])]
for i, mediator in enumerate(mediators):
previous = mediator_positions[:i] if method == 'stat_mediation_serial' else []
equations.append(('mediator'+str(i), mediator, control_positions+[x_position]+previous))
equations.append(('direct', outcome, control_positions+[x_position]+mediator_positions))
fits = {}
diagnostics = {}
path_rows = []
for key, response, indexes in equations:
fit = fit_ols(sample[response].values, matrix[:, indexes], 'standard')
fits[key] = fit
coeffs = coefficient_rows(fit, matrix[:, indexes], [terms[i] for i in indexes], alpha)
diagnostics[key] = {'outcome': response, **model_details(fit), 'coefficients': coeffs}
path_rows.extend({'equation': response, **row} for row in coeffs)
effect_names = [_('总效应'), _('直接效应'), _('总间接效应')]
labels = [predictor+' → '+mediator+' → '+outcome for mediator in mediators]
if method == 'stat_mediation_serial':
labels.append(predictor+' → '+mediators[0]+' → '+mediators[1]+' → '+outcome)
effect_names += labels
def estimates(index=None):
coefficients = {}
for key, response, positions in equations:
if index is None:
beta = fits[key].params
else:
design = matrix[index][:, positions]
if np.linalg.matrix_rank(design) < len(positions):
raise np.linalg.LinAlgError('重抽样设计矩阵秩不足')
beta = np.linalg.lstsq(design, sample[response].values[index], rcond=None)[0]
coefficients[key] = dict(zip(positions, beta))
total = coefficients['total'][x_position]
direct = coefficients['direct'][x_position]
indirect = [coefficients['mediator'+str(i)][x_position]*coefficients['direct'][mediator_positions[i]] for i in range(len(mediators))]
if method == 'stat_mediation_serial':
indirect.append(coefficients['mediator0'][x_position]*coefficients['mediator1'][mediator_positions[0]]*coefficients['direct'][mediator_positions[1]])
return np.array([total, direct, sum(indirect)]+indirect)
original = estimates()
random = np.random.default_rng(seed)
draws, failures = [], 0
progress = options.get('_progress')
for iteration in range(repetitions):
index = random.integers(0, len(sample), len(sample))
try:
draw = estimates(index)
if not np.all(np.isfinite(draw)):
raise np.linalg.LinAlgError('非有限估计')
draws.append(draw)
except np.linalg.LinAlgError:
failures += 1
if callable(progress) and (iteration % 20 == 0 or iteration+1 == repetitions):
progress(iteration+1, repetitions)
if len(draws) < max(180, repetitions*.9):
raise ValueError(_('有效Bootstrap抽样不足90%，请检查样本量、稀疏类别或共线性'))
bootstrap = np.asarray(draws)
intervals = np.quantile(bootstrap, [alpha/2, 1-alpha/2], axis=0)
standard_errors = np.std(bootstrap, axis=0, ddof=1)
rows = [{'effect': name, 'estimate': original[i], 'se': standard_errors[i],
'ci': (intervals[0, i], intervals[1, i])} for i, name in enumerate(effect_names)]
table = PaperTable(_('中介效应检验'), [Column('effect', _('效应或路径'), 'text'), Column('estimate', _('效应值')),
Column('se', 'Boot SE'), Column('ci', _('Bootstrap %(level)s%%置信区间', level='%g' % (100*(1-alpha))), 'interval')],
rows, _('N=%(n)s；个案Bootstrap有效抽样%(b)s次，百分位置信区间。', n=len(sample), b=len(draws)))
tables = [table]
if boolean(options, 'show_path_regressions', False):
tables.append(PaperTable(_('路径回归'), [Column('equation', _('因变量'), 'text')]+regression_headers(alpha), path_rows))
return StatisticalResult(tables, {'sample_n': len(sample), 'outcome': outcome, 'predictor': predictor,
'mediators': mediators, 'controls': controls, 'references': references,
'equations': diagnostics, 'effects': rows, 'bootstrap_requested': repetitions,
'bootstrap_valid': len(draws), 'bootstrap_failed': failures,
'seed': seed, 'ci_method': 'percentile_case_bootstrap',
'decomposition_error': original[0]-original[1]-original[2]})复现本方法
在解压目录安装 requirements.txt 后执行。样例为固定种子的模拟数据，仅供验证；输出不得冒充真实研究结果。
python
import examples._bootstrap
from examples.statistics_cases import build_data, method_cases
from core.statistical.runner import analyze
method = "stat_mediation_simple"
options = dict(method_cases())[method]
result = analyze(build_data(), method, options)
print(result.tables[0].html())