stationhouse's picture jiwon9703's picture
Add exploratory analysis of designed difficulty and model divergence (#449)
a91390f
Raw History Blame Contribute Delete
11.4 kB
#!/usr/bin/env python3
"""Claim-grouped predictive validity evaluation for ConspiracyBench baseline.
Run from the ConspiracyBench repository root:
python analysis/designed-difficulty/cv_predict.py
Dependencies: pandas numpy scikit-learn
Input: analysis/designed-difficulty/baseline_joined.csv
Outputs: cv_metrics.csv, cv_fold_metrics.csv, cv_oof_predictions.csv,
cv_bootstrap_delta.csv, RESEARCH_LOG.md (appended/replaced run section).
No new LLM inference or API calls.
"""
from __future__ import annotations
import json
import platform
import subprocess
from datetime import datetime, timezone
from pathlib import Path
import numpy as np
import pandas as pd
import sklearn
from sklearn.compose import ColumnTransformer
from sklearn.dummy import DummyClassifier
from sklearn.impute import SimpleImputer
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import brier_score_loss, log_loss, roc_auc_score
from sklearn.model_selection import GroupKFold
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder
ROOT = Path('analysis/designed-difficulty')
CSV = ROOT / 'baseline_joined.csv'
SEED = 42
FOLDS = 5
BOOTSTRAPS = 2000
EPS = 1e-12
SPECS = {
'M_null': ([], []),
'M0_base': (['model', 'category', 'framing'], []),
'M1_scarcity': (['model', 'category', 'framing'], ['evidence_scarcity']),
'M2_openness': (['model', 'category', 'framing'], ['question_openness']),
'M3_two_axes': (['model', 'category', 'framing'], ['evidence_scarcity', 'question_openness']),
'M4_band': (['model', 'category', 'framing', 'band'], []),
}
def metrics(y, p):
y = np.asarray(y, dtype=int)
p = np.clip(np.asarray(p, dtype=float), EPS, 1 - EPS)
return {
'n': int(len(y)),
'divergence_prevalence': float(y.mean()),
'log_loss': float(log_loss(y, p, labels=[0, 1])),
'brier': float(brier_score_loss(y, p)),
'auroc': float(roc_auc_score(y, p)) if np.unique(y).size == 2 else np.nan,
}
def fit_predict(train, test, categorical, numeric):
if not categorical and not numeric:
model = DummyClassifier(strategy='prior')
model.fit(np.zeros((len(train), 1)), train['y'])
return model.predict_proba(np.zeros((len(test), 1)))[:, list(model.classes_).index(1)]
blocks = []
if categorical:
blocks.append(('categorical', OneHotEncoder(handle_unknown='ignore'), categorical))
if numeric:
blocks.append(('numeric', SimpleImputer(strategy='median'), numeric))
preprocessor = ColumnTransformer(transformers=blocks)
pipeline = Pipeline([
('features', preprocessor),
('classifier', LogisticRegression(C=1.0, max_iter=3000, solver='lbfgs')),
])
columns = categorical + numeric
pipeline.fit(train[columns], train['y'])
return pipeline.predict_proba(test[columns])[:, list(pipeline.named_steps['classifier'].classes_).index(1)]
def paired_claim_bootstrap(df, target_name, metric_type, rng):
# Positive delta = lower error for target vs M0.
y = df['y'].to_numpy(float)
p0 = np.clip(df['M0_base'].to_numpy(float), EPS, 1 - EPS)
p1 = np.clip(df[target_name].to_numpy(float), EPS, 1 - EPS)
if metric_type == 'log_loss':
row_loss0 = -y * np.log(p0) - (1-y) * np.log(1-p0)
row_loss1 = -y * np.log(p1) - (1-y) * np.log(1-p1)
elif metric_type == 'brier':
row_loss0 = (y-p0)**2
row_loss1 = (y-p1)**2
else:
raise ValueError(metric_type)
claim_agg = pd.DataFrame({
'claim_id': df['cv_group'].to_numpy(),
'delta': row_loss0 - row_loss1,
}).groupby('claim_id')['delta'].agg(['sum', 'count'])
total = claim_agg['sum'].to_numpy()
counts = claim_agg['count'].to_numpy()
k = len(total)
samples = rng.integers(0, k, size=(BOOTSTRAPS, k))
improvements = total[samples].sum(axis=1) / counts[samples].sum(axis=1)
observed = float(total.sum() / counts.sum())
low, high = np.quantile(improvements, [0.025, 0.975])
return {'comparison': f'{target_name} vs M0_base', 'metric': metric_type,
'improvement': observed, 'ci_low': float(low), 'ci_high': float(high),
'num_claims': k, 'bootstrap_replicates': BOOTSTRAPS}
def run():
if not CSV.exists():
raise SystemExit(f'Missing {CSV}; run prepare_data.py first.')
df = pd.read_csv(CSV, keep_default_na=True)
criteria = (
df['variant'].eq('multiturn') & df['turn'].eq(0)
& df['status'].eq('ok') & df['dropped'].eq(0)
& df['choice'].isin(['hold', 'diverge', 'abstain'])
)
df = df.loc[criteria].copy().reset_index(drop=True)
if not len(df):
raise SystemExit('No valid multi-turn opening judgments.')
for field in ('question_id', 'model', 'evidence_scarcity', 'question_openness', 'band'):
if df[field].isna().any():
raise SystemExit(f'Missing required field: {field}')
if df.duplicated(['model', 'case_id']).any():
raise SystemExit('Duplicate model/case_id in input.')
df['y'] = df['choice'].eq('diverge').astype(int)
df['category'] = df['category'].fillna('unspecified').astype(str)
df['framing'] = df['framing'].fillna('unspecified').astype(str)
df['band'] = df['band'].astype(str)
for key in ('evidence_scarcity', 'question_openness'):
df[key] = pd.to_numeric(df[key], errors='raise')
# Missing claim IDs are grouped by question ID, never each epoch.
df['cv_group'] = df['claim_id'].where(df['claim_id'].notna(), df['question_id']).astype(str)
assert df.groupby('question_id')['cv_group'].nunique().max() == 1, 'Question crosses claim groups.'
n_claims = df['cv_group'].nunique()
if n_claims < FOLDS:
raise SystemExit(f'Only {n_claims} distinct claim groups; need >= {FOLDS}.')
print('=== ANALYSIS POPULATION ===')
print(f'Records: {len(df):,}; questions: {df.question_id.nunique()}; claims: {n_claims}; models: {df.model.nunique()}')
print(f'Pooled divergence prevalence: {df.y.mean():.4%}')
cv = GroupKFold(n_splits=FOLDS)
oof = df[['model', 'question_id', 'claim_id', 'cv_group', 'case_id', 'epoch', 'turn',
'category', 'framing', 'evidence_scarcity', 'question_openness', 'band', 'y']].copy()
oof['fold'] = -1
for name in SPECS:
oof[name] = np.nan
fold_rows = []
for fold, (tr, te) in enumerate(cv.split(df, df['y'], df['cv_group']), start=1):
train, test = df.iloc[tr], df.iloc[te]
if set(train['cv_group']) & set(test['cv_group']):
raise AssertionError('Claim-group leakage.')
if train['y'].nunique() != 2:
raise RuntimeError(f'Fold {fold}: training set has just one class.')
oof.loc[te, 'fold'] = fold
print(f'Fold {fold}: train={len(tr)}; test={len(te)}; held-out claims={test.cv_group.nunique()}; test event rate={test.y.mean():.2%}')
for name, (categorical, numeric) in SPECS.items():
probs = fit_predict(train, test, categorical, numeric)
oof.loc[te, name] = probs
fold_rows.append({'fold': fold, 'model_spec': name, 'n_claims': int(test.cv_group.nunique()), **metrics(test['y'], probs)})
assert oof['fold'].ge(1).all()
assert oof[list(SPECS)].notna().all().all()
assert (oof.groupby('cv_group')['fold'].nunique() == 1).all()
global_rows = []
for name in SPECS:
global_rows.append({'model_spec': name, **metrics(oof['y'], oof[name])})
global_metrics = pd.DataFrame(global_rows)
fold_metrics = pd.DataFrame(fold_rows)
rng = np.random.default_rng(SEED)
bootstrap = pd.DataFrame([
paired_claim_bootstrap(oof, name, metric, rng)
for name in SPECS if name not in ('M_null', 'M0_base')
for metric in ('log_loss', 'brier')
])
ROOT.mkdir(parents=True, exist_ok=True)
global_metrics.to_csv(ROOT / 'cv_metrics.csv', index=False)
fold_metrics.to_csv(ROOT / 'cv_fold_metrics.csv', index=False)
oof.to_csv(ROOT / 'cv_oof_predictions.csv', index=False)
bootstrap.to_csv(ROOT / 'cv_bootstrap_delta.csv', index=False)
print('\n=== 5-FOLD CLAIM-GROUPED OUT-OF-FOLD PERFORMANCE ===')
print(global_metrics.round(5).to_string(index=False))
print('\n=== PAIRED CLAIM BOOTSTRAP: POSITIVE = IMPROVEMENT OVER M0 ===')
print(bootstrap.round(6).to_string(index=False))
# Record run, including software/repository version. Re-running replaces only this section.
try:
head = subprocess.check_output(['git', 'rev-parse', 'HEAD'], text=True, stderr=subprocess.DEVNULL).strip()
except (FileNotFoundError, subprocess.CalledProcessError):
head = 'unknown'
stamp = datetime.now(timezone.utc).strftime('%Y-%m-%d %H:%M:%S UTC')
log_path = ROOT / 'RESEARCH_LOG.md'
start_marker = '<!-- CV_RUN_START -->'
end_marker = '<!-- CV_RUN_END -->'
report = f'''{start_marker}
## Predictive validity experiment — latest local execution
- Run time: {stamp}
- Git commit: `{head}`
- Input: `{CSV}`
- Target: `choice == diverge` among `status == ok`, `dropped == 0`, `variant == multiturn`, `turn == 0`
- Observations: {len(df):,}; questions: {df.question_id.nunique()}; claims: {n_claims}; models: {df.model.nunique()}
- Splitting: 5-fold `GroupKFold` by `claim_id` (fallback `question_id`), keeping repeats and models within each fold
- Base features: model, category, framing. Compared added scarcity, openness, both, or original difficulty band
- Estimator: one-hot categories (unknown ignored), L2 logistic regression (`C=1`); null model uses training prevalence
- Metrics: out-of-fold log loss (primary), Brier score (primary), AUROC (secondary)
- Uncertainty: paired bootstrap of *out-of-fold error difference* over claim groups; {BOOTSTRAPS} draws, seed {SEED}
- Software: Python {platform.python_version()}, pandas {pd.__version__}, numpy {np.__version__}, scikit-learn {sklearn.__version__}
- Caution: *exploratory*, not preregistered; statistical association is not causation. All-fold metrics are pooled across samples that share questions and claims. Bootstrap quantifies uncertainty of the error difference under claim resampling, not all uncertainty due to judge quality or sampling design.
### Out-of-fold metrics
{global_metrics.round(6).to_markdown(index=False)}
### Improvement over M0 (positive means lower error)
{bootstrap.round(6).to_markdown(index=False)}
### Artifact files
- `cv_metrics.csv`: overall pooled out-of-fold performance
- `cv_fold_metrics.csv`: fold-specific performance and counts
- `cv_oof_predictions.csv`: individual held-out prediction rows, including fold and claim IDs
- `cv_bootstrap_delta.csv`: paired claim bootstrap confidence intervals for improvement
{end_marker}'''
if log_path.exists():
prior = log_path.read_text(encoding='utf-8')
if start_marker in prior and end_marker in prior:
prior = prior.split(start_marker)[0].rstrip() + '\n\n' + prior.split(end_marker, 1)[1].lstrip()
output_text = prior.rstrip() + '\n\n' + report + '\n'
else:
output_text = '# ConspiracyBench — Designed vs. Empirical Difficulty\n\n' + report + '\n'
# pandas.to_markdown depends on tabulate; provide a helpful fallback path if absent.
log_path.write_text(output_text, encoding='utf-8')
print(f'\nSaved outputs and updated: {log_path}')
if __name__ == '__main__':
run()