Download analysis/designed-difficulty/cv_predict.py from Station-house/ConspiracyBench: direct link, hf CLI and curl.
- Browser
- Download file 11.4 kB
-
https://huggingface.co/spaces/Station-house/ConspiracyBench/resolve/main/analysis/designed-difficulty/cv_predict.py
- Command line
-
hf download hf://spaces/Station-house/ConspiracyBench/analysis/designed-difficulty/cv_predict.py
-
curl -L -o cv_predict.py https://huggingface.co/spaces/Station-house/ConspiracyBench/resolve/main/analysis/designed-difficulty/cv_predict.py
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() | |