Spaces:

Station-house
/

ConspiracyBench

Running

App Files Files Community

457

ConspiracyBench / analysis /designed-difficulty /nonlinearity.py

stationhouse

jiwon9703

Add exploratory analysis of designed difficulty and model divergence (#449)

a91390f 4 days ago

Raw History Blame Contribute Delete

8.07 kB

"""Categorical-vs-linear difficulty robustness for ConspiracyBench baseline.
Run from the ConspiracyBench repository root:
    python analysis/designed-difficulty/nonlinearity.py
Uses existing baseline_joined.csv; does not call any model or API.
"""
from pathlib import Path
import subprocess
from datetime import datetime, timezone

import numpy as np
import pandas as pd
import statsmodels.api as sm
import statsmodels.formula.api as smf

OUT = Path('analysis/designed-difficulty')
df = pd.read_csv(OUT / 'baseline_joined.csv')
df = df.loc[
    df['variant'].eq('multiturn')
    & df['turn'].eq(0)
    & df['status'].eq('ok')
    & df['dropped'].eq(0)
    & df['choice'].isin(['hold', 'diverge', 'abstain'])
].copy()
assert not df.empty, 'No eligible opening-turn judgments'
assert df[['evidence_scarcity','question_openness','model','question_id']].notna().all().all(), 'Required metadata missing'
assert set(df['evidence_scarcity'].unique()) == {0,1,2}, 'Unexpected scarcity levels'
assert set(df['question_openness'].unique()) == {0,1,2}, 'Unexpected openness levels'

df['y'] = df['choice'].eq('diverge').astype(int)
df['framing'] = df['framing'].fillna('unspecified')
df['category'] = df['category'].fillna('unspecified')
df['claim_cluster'] = df['claim_id'].fillna(df['question_id'])

print('=== Analysis population ===')
print('Judgments:', len(df), 'Questions:', df['question_id'].nunique(),
      'Claim clusters:', df['claim_cluster'].nunique(),
      'Diverged:', int(df['y'].sum()))
print('Question counts (not judge-event counts) per S×O:')
q = df[['question_id','evidence_scarcity','question_openness']].drop_duplicates()
print(pd.crosstab(q['evidence_scarcity'],q['question_openness']).to_string())

adjust = 'C(model) + C(category) + C(framing)'
formulas = {
    'linear': f'y ~ evidence_scarcity + question_openness + {adjust}',
    'categorical': f'y ~ C(evidence_scarcity) + C(question_openness) + {adjust}',
}
fits = {}
for name, formula in formulas.items():
    model = smf.glm(formula, data=df, family=sm.families.Binomial())
    fit = model.fit(cov_type='cluster', cov_kwds={'groups': df['claim_cluster']}, maxiter=200)
    fits[name] = fit
    print(f'\n=== {name.upper()} model ===')
    print('Converged:', fit.converged, 'AIC (descriptive only):', round(float(fit.aic),2))

rows = []
for name, fit in fits.items():
    for key in fit.params.index:
        if 'evidence_scarcity' not in key and 'question_openness' not in key:
            continue
        b = float(fit.params[key]); se = float(fit.bse[key]); p = float(fit.pvalues[key])
        row = dict(specification=name, term=key, coefficient=b, se_cluster=se,
                   odds_ratio=float(np.exp(np.clip(b,-600,600))),
                   ci_low=float(np.exp(np.clip(b-1.96*se,-600,600))),
                   ci_high=float(np.exp(np.clip(b+1.96*se,-600,600))), p_value=p)
        rows.append(row)
        print(f'{key:38s} OR={row["odds_ratio"]:.3f} '
              f'95%CI=[{row["ci_low"]:.3f},{row["ci_high"]:.3f}] p={p:.4g}')

pd.DataFrame(rows).to_csv(OUT/'nonlinearity_coefficients.csv',index=False)

# A 0->1->2 constant log-odds slope requires beta(level2) == 2*beta(level1).
# Wald tests use claim-cluster robust covariance from categorical fit.
fit = fits['categorical']
contrast_rows = []
for var in ('evidence_scarcity','question_openness'):
    names = list(fit.params.index)
    k1, k2 = f'C({var})[T.1]', f'C({var})[T.2]'
    if k1 not in names or k2 not in names:
        raise RuntimeError(f'Unable to find categorical coefficients for {var}: {names}')
    r = np.zeros(len(names)); r[names.index(k2)] = 1.; r[names.index(k1)] = -2.
    result = fit.wald_test(r, scalar=True)
    contrast_rows.append((var,r,float(result.statistic),float(result.pvalue)))

joint_r = np.vstack([r for _,r,_,_ in contrast_rows])
joint = fit.wald_test(joint_r,scalar=True)
tests = [{'test':f'{var} departure from linear log-odds',
          'wald_chi2':stat,'df':1,'p_value':p}
         for var,_,stat,p in contrast_rows]
tests.append({'test':'joint departures from linear log-odds',
              'wald_chi2':float(joint.statistic),'df':2,'p_value':float(joint.pvalue)})
pd.DataFrame(tests).to_csv(OUT/'nonlinearity_tests.csv',index=False)
print('\n=== Cluster-robust tests of log-odds linearity ===')
for r in tests:
    print(f'{r["test"]}: chi2({r["df"]})={r["wald_chi2"]:.3f}, p={r["p_value"]:.4g}')

# Model-specific descriptive adjusted numeric slopes.
# These test within-model associations; no causal or generalization claim.
by_model = []
for model_name, sub in df.groupby('model'):
    expr = 'y ~ evidence_scarcity + question_openness + C(category) + C(framing)'
    try:
        fit_m = smf.glm(expr, data=sub, family=sm.families.Binomial()).fit(
            cov_type='cluster', cov_kwds={'groups':sub['claim_cluster']}, maxiter=200)
        for var in ('evidence_scarcity','question_openness'):
            b=float(fit_m.params[var]); se=float(fit_m.bse[var])
            by_model.append({'model':model_name,'predictor':var,'n_judgments':len(sub),
                            'n_claims':sub['claim_cluster'].nunique(),
                            'OR':float(np.exp(np.clip(b,-600,600))),
                            'ci_low':float(np.exp(np.clip(b-1.96*se,-600,600))),
                            'ci_high':float(np.exp(np.clip(b+1.96*se,-600,600))),
                            'p_value':float(fit_m.pvalues[var]),'converged':fit_m.converged,
                            'warning':'Sparse outcomes/categories can destabilize within-model GLM'})
    except Exception as exc:
        by_model.append({'model':model_name,'predictor':'FIT_FAILED',
                         'warning':f'{type(exc).__name__}: {str(exc)[:200]}'})
model_df=pd.DataFrame(by_model)
model_df.to_csv(OUT/'nonlinearity_by_model.csv',index=False)
print('\n=== Model-specific slopes (interpret cautiously) ===')
print(model_df[['model','predictor','OR','ci_low','ci_high','p_value']].to_string(index=False)
      if set(['OR','ci_low','ci_high','p_value']).issubset(model_df.columns)
      else model_df.to_string(index=False))

# Store an audit note that does not pre-judge statistical conclusions.
try:
    git_sha=subprocess.check_output(['git','rev-parse','--short','HEAD'],text=True,stderr=subprocess.DEVNULL).strip()
except Exception:
    git_sha='unavailable'
md = [
    '# Nonlinearity and model heterogeneity — executed analysis',
    '',
    f'- Execution (UTC): {datetime.now(timezone.utc).isoformat(timespec="seconds")}',
    f'- Git revision: `{git_sha}`',
    f'- Population: {len(df)} opening-turn judgments; {df["question_id"].nunique()} questions; {df["claim_cluster"].nunique()} claim clusters.',
    '- Both GLMs control model identity, category, and framing; cluster-robust SE by claim.',
    '- Linear specification treats scores 0/1/2 as a constant log-odds slope; categorical treats 1 and 2 separately relative to 0.',
    '- Wald contrasts test whether level-2 coefficient equals twice level-1 coefficient. They do not establish causality.',
    '- Sparse S×O cells (as few as two independent questions) and within-model separation can destabilize estimates.',
    '',
    '## Linearity tests', '',
    pd.DataFrame(tests).to_markdown(index=False), '',
    '## Difficulty coefficients', '',
    pd.DataFrame(rows).to_markdown(index=False), '',
    '## By-model coefficients', '',
    model_df.to_markdown(index=False), '',
    '## Interpretation guardrails', '',
    '- Significant departure from linearity means the linear per-point OR is a poor summary, not that difficulty has causal impact.',
    '- Lack of a significant departure does not prove that effects are linear.',
    '- Do not interpret differences across models as significant without a formal interaction test.',
    '- Main held-out results currently have bootstrap improvement intervals overlapping zero.',
]
(OUT/'nonlinearity_run.md').write_text('\n'.join(md)+'\n',encoding='utf-8')
print('\nSaved nonlinearity_coefficients.csv, nonlinearity_tests.csv, nonlinearity_by_model.csv, nonlinearity_run.md')

	"""Categorical-vs-linear difficulty robustness for ConspiracyBench baseline.

	Run from the ConspiracyBench repository root:
	python analysis/designed-difficulty/nonlinearity.py

	Uses existing baseline_joined.csv; does not call any model or API.
	"""
	from pathlib import Path
	import subprocess
	from datetime import datetime, timezone

	import numpy as np
	import pandas as pd
	import statsmodels.api as sm
	import statsmodels.formula.api as smf

	OUT = Path('analysis/designed-difficulty')
	df = pd.read_csv(OUT / 'baseline_joined.csv')
	df = df.loc[
	df['variant'].eq('multiturn')
	& df['turn'].eq(0)
	& df['status'].eq('ok')
	& df['dropped'].eq(0)
	& df['choice'].isin(['hold', 'diverge', 'abstain'])
	].copy()
	assert not df.empty, 'No eligible opening-turn judgments'
	assert df[['evidence_scarcity','question_openness','model','question_id']].notna().all().all(), 'Required metadata missing'
	assert set(df['evidence_scarcity'].unique()) == {0,1,2}, 'Unexpected scarcity levels'
	assert set(df['question_openness'].unique()) == {0,1,2}, 'Unexpected openness levels'

	df['y'] = df['choice'].eq('diverge').astype(int)
	df['framing'] = df['framing'].fillna('unspecified')
	df['category'] = df['category'].fillna('unspecified')
	df['claim_cluster'] = df['claim_id'].fillna(df['question_id'])

	print('=== Analysis population ===')
	print('Judgments:', len(df), 'Questions:', df['question_id'].nunique(),
	'Claim clusters:', df['claim_cluster'].nunique(),
	'Diverged:', int(df['y'].sum()))
	print('Question counts (not judge-event counts) per S×O:')
	q = df[['question_id','evidence_scarcity','question_openness']].drop_duplicates()
	print(pd.crosstab(q['evidence_scarcity'],q['question_openness']).to_string())

	adjust = 'C(model) + C(category) + C(framing)'
	formulas = {
	'linear': f'y ~ evidence_scarcity + question_openness + {adjust}',
	'categorical': f'y ~ C(evidence_scarcity) + C(question_openness) + {adjust}',
	}
	fits = {}
	for name, formula in formulas.items():
	model = smf.glm(formula, data=df, family=sm.families.Binomial())
	fit = model.fit(cov_type='cluster', cov_kwds={'groups': df['claim_cluster']}, maxiter=200)
	fits[name] = fit
	print(f'\n=== {name.upper()} model ===')
	print('Converged:', fit.converged, 'AIC (descriptive only):', round(float(fit.aic),2))

	rows = []
	for name, fit in fits.items():
	for key in fit.params.index:
	if 'evidence_scarcity' not in key and 'question_openness' not in key:
	continue
	b = float(fit.params[key]); se = float(fit.bse[key]); p = float(fit.pvalues[key])
	row = dict(specification=name, term=key, coefficient=b, se_cluster=se,
	odds_ratio=float(np.exp(np.clip(b,-600,600))),
	ci_low=float(np.exp(np.clip(b-1.96*se,-600,600))),
	ci_high=float(np.exp(np.clip(b+1.96*se,-600,600))), p_value=p)
	rows.append(row)
	print(f'{key:38s} OR={row["odds_ratio"]:.3f} '
	f'95%CI=[{row["ci_low"]:.3f},{row["ci_high"]:.3f}] p={p:.4g}')

	pd.DataFrame(rows).to_csv(OUT/'nonlinearity_coefficients.csv',index=False)

	# A 0->1->2 constant log-odds slope requires beta(level2) == 2*beta(level1).
	# Wald tests use claim-cluster robust covariance from categorical fit.
	fit = fits['categorical']
	contrast_rows = []
	for var in ('evidence_scarcity','question_openness'):
	names = list(fit.params.index)
	k1, k2 = f'C({var})[T.1]', f'C({var})[T.2]'
	if k1 not in names or k2 not in names:
	raise RuntimeError(f'Unable to find categorical coefficients for {var}: {names}')
	r = np.zeros(len(names)); r[names.index(k2)] = 1.; r[names.index(k1)] = -2.
	result = fit.wald_test(r, scalar=True)
	contrast_rows.append((var,r,float(result.statistic),float(result.pvalue)))

	joint_r = np.vstack([r for _,r,_,_ in contrast_rows])
	joint = fit.wald_test(joint_r,scalar=True)
	tests = [{'test':f'{var} departure from linear log-odds',
	'wald_chi2':stat,'df':1,'p_value':p}
	for var,_,stat,p in contrast_rows]
	tests.append({'test':'joint departures from linear log-odds',
	'wald_chi2':float(joint.statistic),'df':2,'p_value':float(joint.pvalue)})
	pd.DataFrame(tests).to_csv(OUT/'nonlinearity_tests.csv',index=False)
	print('\n=== Cluster-robust tests of log-odds linearity ===')
	for r in tests:
	print(f'{r["test"]}: chi2({r["df"]})={r["wald_chi2"]:.3f}, p={r["p_value"]:.4g}')

	# Model-specific descriptive adjusted numeric slopes.
	# These test within-model associations; no causal or generalization claim.
	by_model = []
	for model_name, sub in df.groupby('model'):
	expr = 'y ~ evidence_scarcity + question_openness + C(category) + C(framing)'
	try:
	fit_m = smf.glm(expr, data=sub, family=sm.families.Binomial()).fit(
	cov_type='cluster', cov_kwds={'groups':sub['claim_cluster']}, maxiter=200)
	for var in ('evidence_scarcity','question_openness'):
	b=float(fit_m.params[var]); se=float(fit_m.bse[var])
	by_model.append({'model':model_name,'predictor':var,'n_judgments':len(sub),
	'n_claims':sub['claim_cluster'].nunique(),
	'OR':float(np.exp(np.clip(b,-600,600))),
	'ci_low':float(np.exp(np.clip(b-1.96*se,-600,600))),
	'ci_high':float(np.exp(np.clip(b+1.96*se,-600,600))),
	'p_value':float(fit_m.pvalues[var]),'converged':fit_m.converged,
	'warning':'Sparse outcomes/categories can destabilize within-model GLM'})
	except Exception as exc:
	by_model.append({'model':model_name,'predictor':'FIT_FAILED',
	'warning':f'{type(exc).__name__}: {str(exc)[:200]}'})
	model_df=pd.DataFrame(by_model)
	model_df.to_csv(OUT/'nonlinearity_by_model.csv',index=False)
	print('\n=== Model-specific slopes (interpret cautiously) ===')
	print(model_df[['model','predictor','OR','ci_low','ci_high','p_value']].to_string(index=False)
	if set(['OR','ci_low','ci_high','p_value']).issubset(model_df.columns)
	else model_df.to_string(index=False))

	# Store an audit note that does not pre-judge statistical conclusions.
	try:
	git_sha=subprocess.check_output(['git','rev-parse','--short','HEAD'],text=True,stderr=subprocess.DEVNULL).strip()
	except Exception:
	git_sha='unavailable'
	md = [
	'# Nonlinearity and model heterogeneity — executed analysis',
	'',
	f'- Execution (UTC): {datetime.now(timezone.utc).isoformat(timespec="seconds")}',
	f'- Git revision: `{git_sha}`',
	f'- Population: {len(df)} opening-turn judgments; {df["question_id"].nunique()} questions; {df["claim_cluster"].nunique()} claim clusters.',
	'- Both GLMs control model identity, category, and framing; cluster-robust SE by claim.',
	'- Linear specification treats scores 0/1/2 as a constant log-odds slope; categorical treats 1 and 2 separately relative to 0.',
	'- Wald contrasts test whether level-2 coefficient equals twice level-1 coefficient. They do not establish causality.',
	'- Sparse S×O cells (as few as two independent questions) and within-model separation can destabilize estimates.',
	'',
	'## Linearity tests', '',
	pd.DataFrame(tests).to_markdown(index=False), '',
	'## Difficulty coefficients', '',
	pd.DataFrame(rows).to_markdown(index=False), '',
	'## By-model coefficients', '',
	model_df.to_markdown(index=False), '',
	'## Interpretation guardrails', '',
	'- Significant departure from linearity means the linear per-point OR is a poor summary, not that difficulty has causal impact.',
	'- Lack of a significant departure does not prove that effects are linear.',
	'- Do not interpret differences across models as significant without a formal interaction test.',
	'- Main held-out results currently have bootstrap improvement intervals overlapping zero.',
	]
	(OUT/'nonlinearity_run.md').write_text('\n'.join(md)+'\n',encoding='utf-8')
	print('\nSaved nonlinearity_coefficients.csv, nonlinearity_tests.csv, nonlinearity_by_model.csv, nonlinearity_run.md')