"""Replicate-T mixing guard (artifact class #4, ideas.md 2026-08-26). Question: can a cheap guard flag non-equilibrium population mixing (photoisomer populations, pressure ladders, cross-study contamination) from cell volumes alone, with no literature reading? Guard: for each multi-T series with replicate entries at a shared nominal temperature, compare max within-T relative spread of V against the median adjacent-T relative step (mean V per nominal T). Flag when max_spread > max(median_step, FLOOR) i.e. same-temperature "replicates" disagree more than neighboring temperatures do, above an absolute floor that keeps numerical noise quiet. Nominal temperatures are clustered within TBIN=3 K before replicate matching (lesson from control P3: cross-study "room temperature" entries arrive as 295/296/298 K and exact-T matching sees no replicates). Positive controls (known artifacts, must flag): P1 Ru-SO2 linkage photoisomer series (JACS 2012, illumination populations) P2 BaThF6 pressure ladder (Dalton Trans 2011, 3x293 K cells at ~0/2.1/4.1 GPa) P3 C22H26N2O3 cross-study formula collision (3 unrelated papers) Negative controls (must pass): ferrierite (clean PTE zeolite), KNN (real transition), PbTiO3 (real transition), ScF3 (clean NTE). Scar-rule compliance: this script IS the artifact; run it to regenerate replicate_mixing_guard.json. """ import json, re, statistics from collections import defaultdict META = 'cod_meta.jsonl' OUT = 'replicate_mixing_guard.json' FLOOR = 0.01 # 1% absolute floor on within-T spread (cross-lab refinement # noise on V is routinely 0.2-0.6%; both positive artifacts # exceed 1%). Ratio factor: spread must also exceed 2x the # median adjacent-T step. Calibrated on controls below. TOK = re.compile(r'^([A-Z][a-z]?)([\d.]*)$') def canon_formula(raw): s = raw.strip().strip('-').strip() counts = defaultdict(float) for tok in s.split(): m = TOK.match(tok) if not m: return None el, num = m.group(1), m.group(2) if el == 'D': el = 'H' counts[el] += float(num) if num else 1.0 parts = [] for el, c in sorted(counts.items()): if abs(c - round(c)) < 1e-6: c = int(round(c)) parts.append(el + ('' if c == 1 else str(c))) else: parts.append(f'{el}{c:.2f}') return ''.join(parts) def parse_T(s): m = re.search(r'(-?\d+(?:\.\d+)?)', str(s)) return float(m.group(1)) if m else None meta = {} for line in open(META): m = json.loads(line) m['_T'] = parse_T(m.get('celltemp','')) m['_V'] = float(m['vol']) if m.get('vol') else None if m['_T'] is not None and m['_V'] and m['_V'] > 0: m['_canon'] = canon_formula(m.get('formula','')) if m['_canon']: meta[m['file']] = m # group by (canon, sgNumber) groups = defaultdict(list) for m in meta.values(): groups[(m['_canon'], m.get('sgNumber',''))].append(m) TBIN = 3.0 # K; nominal-temperature cluster width def guard(entries): """entries: list of meta dicts with _T, _V. Returns dict or None.""" byT = defaultdict(list) for e in entries: byT[e['_T']].append(e['_V']) # cluster nominal temps within TBIN (e.g. 295/296/298 -> one bin) Ts_sorted = sorted(byT) canon_T = {} cur = Ts_sorted[0] for T in Ts_sorted: if T - cur > TBIN: cur = T canon_T[T] = cur binned = defaultdict(list) for T, vs in byT.items(): binned[canon_T[T]].extend(vs) byT = binned reps = {T: vs for T, vs in byT.items() if len(vs) >= 2} if not reps or len(byT) < 2: return None spreads = {T: (max(vs)-min(vs))/statistics.mean(vs) for T, vs in reps.items()} max_spread_T = max(spreads, key=spreads.get) max_spread = spreads[max_spread_T] means = {T: statistics.mean(vs) for T, vs in byT.items()} Ts = sorted(means) steps = [abs(means[b]-means[a])/means[a] for a, b in zip(Ts, Ts[1:])] med_step = statistics.median(steps) if steps else 0.0 flagged = max_spread > max(2 * med_step, FLOOR) return { 'max_spread': round(max_spread, 5), 'spread_at_T': max_spread_T, 'median_step': round(med_step, 5), 'n_temps': len(byT), 'n_rep_temps': len(reps), 'n_entries': len(entries), 'flagged': flagged, 'spreads_by_T': {str(T): round(s,5) for T, s in sorted(spreads.items())}, 'max_spread_n_studies': len({e['journal'] for e in entries if abs(e['_T'] - max_spread_T) <= TBIN}), } # --- controls ---------------------------------------------------------------- def series_files(formula_canon, sg): return [m for m in meta.values() if m['_canon']==formula_canon and str(m.get('sgNumber',''))==sg] controls = {} # P1 Ru-SO2 photoisomer ru = [m for f in ["4118315","4118312","4118309","4118311","4118306","4118308","4118316","4118295","4118297"] for m in [meta.get(f)] if m] controls['P1_ru_so2_photoisomer'] = guard(ru) # P2 BaThF6 pressure ladder: find entries bathf6 = series_files('BaF6Th', '194') controls['P2_bathf6_pressure_ladder'] = guard(bathf6) if bathf6 else {'error': 'series not found', 'n': len(bathf6)} # P3 C22H26N2O3 cross-study collision (any SG with many entries) c22 = [m for m in meta.values() if m['_canon']=='C22H26N2O3'] big = defaultdict(list) for m in c22: big[(m['_canon'], m.get('sgNumber',''))].append(m) if big: key = max(big, key=lambda k: len(big[k])) controls['P3_c22h26n2o3_cross_study'] = guard(big[key]) controls['P3_c22h26n2o3_cross_study']['group'] = f'{key[0]}/sg{key[1]}' controls['P3_c22h26n2o3_cross_study']['n_studies'] = len({m['journal'] for m in big[key]}) # negatives for name, canon, sg in [('N1_ferrierite','O72Si36','58'), ('N2_linbo3','LiNbO3','161')]: ents = series_files(canon, sg) g = guard(ents) if ents else None controls[name] = g if g else {'pass': True, 'reason': 'no replicate-temperature cluster (clean series)'} # --- census-wide run --------------------------------------------------------- results = {} for (canon, sg), ents in groups.items(): if len(ents) >= 4: g = guard(ents) if g: results[f'{canon}|sg{sg}'] = g flagged = {k: v for k, v in results.items() if v['flagged']} has_h = lambda k: 'H' in k.split('|')[0] out = { 'method': 'Replicate-T mixing guard: max within-T V spread vs median adjacent-T step; flag if spread > max(step, 0.2% floor)', 'floor': FLOOR, 'controls': controls, 'n_series_evaluated': len(results), 'n_flagged': len(flagged), 'n_flagged_hfree': sum(1 for k in flagged if not has_h(k)), 'flagged_hfree_top40': [ dict(key=k, **v, journal=sorted({m['journal'] for m in groups[(k.split('|')[0], k.split('sg')[1].strip())]})[:3]) for k, v in sorted(flagged.items(), key=lambda kv: -kv[1]['max_spread']/max(kv[1]['median_step'],1e-6)) if not has_h(k) ][:40], } json.dump(out, open(OUT,'w'), indent=1, default=str) print('controls:') for k, v in controls.items(): if isinstance(v, dict) and 'flagged' in v: print(f" {k}: flagged={v['flagged']} spread={v['max_spread']} step={v['median_step']}") else: print(f" {k}: {v}") print(f"census: {len(results)} series evaluated, {len(flagged)} flagged " f"({out['n_flagged_hfree']} H-free)")