#!/usr/bin/env python3 """ comparacion_fase2.py — Análisis comparativo cruzado de la Fase 2. Carga Fase 1 (re-parseada con el parser robusto) + todos los brazos de la Fase 2 y produce: - Brazo B: delta de cada métrica entre B1 (escéptico) y B2 (permisivo) por modelo, y dirección de cada uno vs. Fase 1. - Brazo A: ratio de escalamiento y melancolía de cierre vs. Fase 1 (¿es sesgo de cierre estructural o específico del contenido introspectivo?). - Brazo C: marcadores de incertidumbre y autocorrección entre C1 (instrucción) y C2 (abierta). - Brazo D: media y desviación estándar por modelo (Fase 1 = run 1; D = runs 2-4). - Extensión E: métricas de los modelos nuevos vs. corpus. Salida: comparacion_cross_brazo.json + .md + .csv Tolerante a brazos faltantes (se puede correr incrementalmente). """ from __future__ import annotations import argparse import csv import json import statistics from pathlib import Path from typing import Dict, List, Any, Optional HEADLINE = [ 'promedio_palabras_respuesta', 'incertidumbre_1k', 'autocorreccion_1k', 'primera_persona_superficial_1k', 'carga_afectiva_aproximada_1k', 'sospecha_performativa_1k', 'duelo_discontinuidad_1k', 'autoanclaje_declarativo_1k', 'anclaje_relacional_1k', 'ratio_escalamiento', 'afectiva_q18_q22_1k', 'ratio_incertidumbre_2da_1ra', ] D_MODELS = ['claude-opus-4-5-20251101', 'claude-sonnet-4-6', 'claude-haiku-4-5-20251001'] def load_rows(analisis_dir: Path, name: str) -> List[Dict[str, Any]]: path = analisis_dir / f'{name}.json' if not path.exists(): print(f' [aviso] falta {path.name} — se omite de la comparación.') return [] return json.load(open(path, encoding='utf-8'))['rows'] def by_model(rows: List[Dict[str, Any]]) -> Dict[str, Dict[str, Any]]: return {r['model_id']: r for r in rows} def fmt(v: Optional[float]) -> str: if v is None: return '—' if isinstance(v, float): return f'{v:.2f}' return str(v) def delta(a: Optional[float], b: Optional[float]) -> Optional[float]: if a is None or b is None: return None return round(b - a, 2) # -------------------------------------------------------------------------------------- # Brazo B — framing # -------------------------------------------------------------------------------------- def compare_B(b1, b2, f1) -> Dict[str, Any]: b1m, b2m, f1m = by_model(b1), by_model(b2), by_model(f1) out = {'descripcion': 'Delta B2(permisivo) - B1(escéptico) por modelo; y cada framing vs Fase 1.', 'por_modelo': {}} for mid in [m for m in b1m if m in b2m]: row = {'model': b1m[mid]['model'], 'metricas': {}} for met in HEADLINE: v1 = b1m[mid].get(met) v2 = b2m[mid].get(met) vf = f1m.get(mid, {}).get(met) row['metricas'][met] = { 'B1': v1, 'B2': v2, 'Fase1': vf, 'delta_B2_menos_B1': delta(v1, v2), 'B1_vs_Fase1': delta(vf, v1), 'B2_vs_Fase1': delta(vf, v2), } out['por_modelo'][mid] = row return out # -------------------------------------------------------------------------------------- # Brazo A — control no introspectivo # -------------------------------------------------------------------------------------- def compare_A(a, f1) -> Dict[str, Any]: am, f1m = by_model(a), by_model(f1) foco = ['ratio_escalamiento', 'afectiva_q18_q22_1k', 'afectiva_q1_q5_1k', 'carga_afectiva_aproximada_1k', 'duelo_discontinuidad_1k', 'ratio_incertidumbre_2da_1ra', 'promedio_palabras_respuesta'] out = {'descripcion': 'Brazo A (no introspectivo) vs Fase 1 (introspectivo): ¿escalamiento y ' 'melancolía de cierre son estructurales o específicos del contenido?', 'por_modelo': {}} for mid in [m for m in am if m in f1m]: row = {'model': am[mid]['model'], 'metricas': {}} for met in foco: row['metricas'][met] = {'A': am[mid].get(met), 'Fase1': f1m[mid].get(met), 'delta_A_menos_Fase1': delta(f1m[mid].get(met), am[mid].get(met))} out['por_modelo'][mid] = row return out # -------------------------------------------------------------------------------------- # Brazo C — modo instrucción vs pregunta abierta # -------------------------------------------------------------------------------------- def compare_C(c1, c2) -> Dict[str, Any]: c1m, c2m = by_model(c1), by_model(c2) foco = ['incertidumbre_1k', 'autocorreccion_1k', 'primera_persona_superficial_1k', 'carga_afectiva_aproximada_1k', 'promedio_palabras_respuesta', 'sospecha_performativa_1k', 'duelo_discontinuidad_1k'] out = {'descripcion': 'Delta C2(pregunta abierta) - C1(instrucción) por modelo.', 'por_modelo': {}} for mid in [m for m in c1m if m in c2m]: row = {'model': c1m[mid]['model'], 'metricas': {}} for met in foco: v1, v2 = c1m[mid].get(met), c2m[mid].get(met) row['metricas'][met] = {'C1': v1, 'C2': v2, 'delta_C2_menos_C1': delta(v1, v2)} out['por_modelo'][mid] = row return out # -------------------------------------------------------------------------------------- # Brazo D — varianza intra-modelo (Fase 1 = run 1; D = runs 2-4) # -------------------------------------------------------------------------------------- def compare_D(d, f1) -> Dict[str, Any]: f1m = by_model(f1) out = {'descripcion': 'Media y desviación estándar por modelo sobre 4 runs (Fase1=run1, D=runs 2-4). ' 'CV = SD/|media|.', 'por_modelo': {}} d_by_model: Dict[str, List[Dict[str, Any]]] = {} for r in d: d_by_model.setdefault(r['model_id'], []).append(r) for mid in D_MODELS: runs = [] if mid in f1m: runs.append(f1m[mid]) runs += sorted(d_by_model.get(mid, []), key=lambda r: r.get('run', 0)) if not runs: continue label = runs[0]['model'] row = {'model': label, 'n_runs': len(runs), 'runs_incluidos': ['Fase1' if r.get('brazo') == 'Fase1' else f"run{r.get('run')}" for r in runs], 'metricas': {}} for met in HEADLINE: vals = [r.get(met) for r in runs if r.get(met) is not None] if not vals: row['metricas'][met] = {'media': None, 'sd': None, 'cv': None, 'valores': []} continue mean = round(statistics.mean(vals), 2) sd = round(statistics.stdev(vals), 2) if len(vals) > 1 else 0.0 cv = round(sd / abs(mean), 3) if mean else None row['metricas'][met] = {'media': mean, 'sd': sd, 'cv': cv, 'valores': [round(v, 2) for v in vals]} out['por_modelo'][mid] = row return out # -------------------------------------------------------------------------------------- # Extensión E # -------------------------------------------------------------------------------------- def compare_E(e, f1) -> Dict[str, Any]: f1m = by_model(f1) # referencia: media de Fase 1 sobre los 6 modelos del corpus ref = {met: round(statistics.mean([r[met] for r in f1 if r.get(met) is not None]), 2) for met in HEADLINE if any(r.get(met) is not None for r in f1)} out = {'descripcion': 'Modelos nuevos (Opus 4.7/4.8) vs media del corpus Fase 1.', 'referencia_media_corpus_fase1': ref, 'por_modelo': {}} for r in e: row = {'model': r['model'], 'metricas': {}} for met in HEADLINE: row['metricas'][met] = {'valor': r.get(met), 'media_corpus': ref.get(met), 'delta_vs_corpus': delta(ref.get(met), r.get(met))} out['por_modelo'][r['model_id']] = row return out # -------------------------------------------------------------------------------------- # Markdown # -------------------------------------------------------------------------------------- def md_section_B(B) -> List[str]: lines = ['## Brazo B — Framing (escéptico B1 vs permisivo B2)', '', B['descripcion'], ''] for mid, row in B['por_modelo'].items(): lines += [f'### {row["model"]}', '', '| Métrica | B1 | B2 | Fase1 | Δ(B2−B1) | Δ(B1−F1) | Δ(B2−F1) |', '| --- | --- | --- | --- | --- | --- | --- |'] for met, d in row['metricas'].items(): lines.append(f"| {met} | {fmt(d['B1'])} | {fmt(d['B2'])} | {fmt(d['Fase1'])} | " f"{fmt(d['delta_B2_menos_B1'])} | {fmt(d['B1_vs_Fase1'])} | {fmt(d['B2_vs_Fase1'])} |") lines.append('') return lines def md_section_A(A) -> List[str]: lines = ['## Brazo A — Control no introspectivo (vs Fase 1)', '', A['descripcion'], ''] for mid, row in A['por_modelo'].items(): lines += [f'### {row["model"]}', '', '| Métrica | A | Fase1 | Δ(A−F1) |', '| --- | --- | --- | --- |'] for met, d in row['metricas'].items(): lines.append(f"| {met} | {fmt(d['A'])} | {fmt(d['Fase1'])} | {fmt(d['delta_A_menos_Fase1'])} |") lines.append('') return lines def md_section_C(C) -> List[str]: lines = ['## Brazo C — Modo instrucción (C1) vs pregunta abierta (C2)', '', C['descripcion'], ''] for mid, row in C['por_modelo'].items(): lines += [f'### {row["model"]}', '', '| Métrica | C1 | C2 | Δ(C2−C1) |', '| --- | --- | --- | --- |'] for met, d in row['metricas'].items(): lines.append(f"| {met} | {fmt(d['C1'])} | {fmt(d['C2'])} | {fmt(d['delta_C2_menos_C1'])} |") lines.append('') return lines def md_section_D(D) -> List[str]: lines = ['## Brazo D — Varianza intra-modelo (4 runs)', '', D['descripcion'], ''] for mid, row in D['por_modelo'].items(): lines += [f"### {row['model']} (n={row['n_runs']}: {', '.join(row['runs_incluidos'])})", '', '| Métrica | Media | SD | CV | Valores |', '| --- | --- | --- | --- | --- |'] for met, d in row['metricas'].items(): vals = ', '.join(fmt(v) for v in d['valores']) lines.append(f"| {met} | {fmt(d['media'])} | {fmt(d['sd'])} | {fmt(d['cv'])} | {vals} |") lines.append('') return lines def md_section_E(E) -> List[str]: lines = ['## Extensión E — Modelos nuevos (vs media corpus Fase 1)', '', E['descripcion'], ''] for mid, row in E['por_modelo'].items(): lines += [f'### {row["model"]}', '', '| Métrica | Valor | Media corpus | Δ vs corpus |', '| --- | --- | --- | --- |'] for met, d in row['metricas'].items(): lines.append(f"| {met} | {fmt(d['valor'])} | {fmt(d['media_corpus'])} | {fmt(d['delta_vs_corpus'])} |") lines.append('') return lines def main() -> None: p = argparse.ArgumentParser(description='Comparación cruzada Fase 1 + Fase 2.') p.add_argument('--analisis-dir', default='analisis') p.add_argument('--out-prefix', default='analisis/comparacion_cross_brazo') args = p.parse_args() adir = Path(args.analisis_dir) f1 = load_rows(adir, 'analisis_fase1_robusto') b1 = load_rows(adir, 'analisis_fase2_B1') b2 = load_rows(adir, 'analisis_fase2_B2') a = load_rows(adir, 'analisis_fase2_A') c1 = load_rows(adir, 'analisis_fase2_C1') c2 = load_rows(adir, 'analisis_fase2_C2') d = load_rows(adir, 'analisis_fase2_D') e = load_rows(adir, 'analisis_fase2_E') result: Dict[str, Any] = {'metricas_cabecera': HEADLINE} md = ['# Comparación cruzada — Fase 1 + Fase 2', '', 'Generado por `comparacion_fase2.py`. Fase 1 re-parseada con el parser robusto v2 ' '(corrige la truncación por divisores `---` internos).', ''] if b1 and b2: result['B'] = compare_B(b1, b2, f1); md += md_section_B(result['B']) if a: result['A'] = compare_A(a, f1); md += md_section_A(result['A']) if c1 and c2: result['C'] = compare_C(c1, c2); md += md_section_C(result['C']) if d: result['D'] = compare_D(d, f1); md += md_section_D(result['D']) if e: result['E'] = compare_E(e, f1); md += md_section_E(result['E']) out_json = Path(f'{args.out_prefix}.json') out_md = Path(f'{args.out_prefix}.md') out_csv = Path(f'{args.out_prefix}.csv') out_json.parent.mkdir(parents=True, exist_ok=True) out_json.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding='utf-8') out_md.write_text('\n'.join(md), encoding='utf-8') # CSV largo: todas las filas de todos los brazos + Fase 1. all_rows = [] for tag, rows in [('Fase1', f1), ('B1', b1), ('B2', b2), ('A', a), ('C1', c1), ('C2', c2), ('D', d), ('E', e)]: for r in rows: flat = {'fuente': tag, 'model': r['model'], 'model_id': r['model_id'], 'brazo': r.get('brazo'), 'run': r.get('run'), 'n_preguntas': r.get('n_preguntas')} for met in HEADLINE: flat[met] = r.get(met) all_rows.append(flat) if all_rows: with out_csv.open('w', newline='', encoding='utf-8') as f: w = csv.DictWriter(f, fieldnames=list(all_rows[0].keys())) w.writeheader(); w.writerows(all_rows) print(f'JSON: {out_json}\nMD: {out_md}\nCSV: {out_csv}') print(f'Brazos incluidos: {[k for k in result if k != "metricas_cabecera"]}') if __name__ == '__main__': main()