#!/usr/bin/env python3 """Deterministic first-pass taxonomy classifier for lab-field divergence. Input: a pandas DataFrame containing field p75 metrics, lab metrics, and optional context columns (the contract is `REQUIRED_COLUMNS` / `OPTIONAL_COLUMNS` below, and it is the same naming the step-01/03 SQL emits — verify with `--check-contract`). Output: the same DataFrame with `divergence_type`, `divergence_reason` and `evidence_status`. What this is and is not ----------------------- This is a **triage labeller**, not a causal model. Several taxonomy classes cannot be decided from a public origin-month panel at all: they need evidence the panel does not contain (interaction traces, RUM attribution, repeated lab runs, per-template or per-country breakdowns). Those classes are declared in `EVIDENCE_REQUIREMENTS`; when the deciding pattern is present but the required evidence columns are absent, the row is labelled `UNRESOLVED_` with `evidence_status='insufficient_evidence'` rather than being given a causal label it has not earned. `UNCLASSIFIED` therefore means "no pattern matched", and is distinct from "pattern matched but the minimum evidence standard in the taxonomy CSV was not met". """ from __future__ import annotations import pandas as pd LCP_GOOD_MS = 2500 INP_GOOD_MS = 200 CLS_GOOD = 0.1 TBT_LAB_WARNING_MS = 200 TTFB_POOR_MS = 1800 # Conservative diagnostic threshold; tune by preregistered sensitivity analysis. RTT_HIGH_FRACTION = 0.15 BFCACHE_FRACTION = 0.15 # Data contract. These names match the step-01 CrUX panel and the step-03 join exactly. REQUIRED_COLUMNS = ['p75_lcp', 'p75_inp', 'p75_cls', 'lab_lcp_ms', 'lab_tbt_ms', 'lab_cls'] OPTIONAL_COLUMNS = [ 'p75_ttfb', # step 01 'rtt_high_fraction', # step 01 (materialized `high_rtt`) 'navigation_bfcache_fraction', # step 01 (materialized `navigation_type_back_forward_cache`) ] # Columns that carry the taxonomy's minimum evidence standard for each class. A class whose # evidence columns are all missing cannot be positively assigned from this panel. EVIDENCE_REQUIREMENTS: dict[str, list[str]] = { # "Synthetic user-flow or RUM attribution finds long event processing/presentation delay". 'INP_BEHAVIOR': ['rum_inp_attribution_ms', 'userflow_inp_ms'], # "Repeated lab runs show load-time long tasks that do not overlap user interactions". 'TBT_FALSE_ALARM': ['lab_run_count', 'lab_tbt_iqr_ms'], # "Field CLS p75 remains good across CrUX windows; repeated lab variance high". 'CLS_LAB_ONLY': ['lab_run_count', 'lab_cls_iqr'], # "RUM layout-shift attribution identifies elements after load or user-specific variants". 'CLS_FIELD_ONLY': ['rum_cls_attribution'], # "CrUX navigation-type fractions predict residual; cold/warm lab profiles diverge". # Navigation type is a population share, never proof of service-worker use: CrUX cannot break # performance metrics down by navigation type, so a warm/cold lab pair is the deciding evidence. 'BFCACHE_SW': ['lab_warm_profile_lcp_ms'], # "Country-level CrUX, RTT, and multi-region WPT reduce residual". 'GEO_RTT': ['crux_country_breakdown', 'multiregion_lab_lcp_ms'], # "Device/form-factor stratified validation; lab sensitivity to CPU throttle". 'DEVICE_MIX': ['device_class_breakdown', 'lab_cpu_throttle_sensitivity'], # "URL-level CrUX or RUM top-template panel changes conclusion". 'PAGE_MIX': ['url_level_crux_panel', 'template_traffic_weights'], # "Event-study alignment explains residual; future CrUX converges after intervention". 'TIME_WINDOW': ['deploy_event_date', 'crux_event_study_residual'], # "Repeated lab runs and stratified RUM show multi-modal outcomes". 'CONTENT_VARIANCE': ['lab_run_count', 'lab_lcp_iqr_ms'], } def _has_evidence(row: pd.Series, klass: str) -> bool: """True when at least one of the taxonomy's minimum-evidence columns is present and non-null.""" needed = EVIDENCE_REQUIREMENTS.get(klass) if not needed: return True return any(col in row.index and pd.notna(row.get(col)) for col in needed) def _decide(row: pd.Series, klass: str, reason: str) -> tuple[str, str, str]: """Emit a positive label only when the class's minimum evidence standard is satisfiable.""" if _has_evidence(row, klass): return klass, reason, 'evidence_present' return ( f'UNRESOLVED_{klass}', f'{reason} Minimum evidence for {klass} is absent from this panel ' f'(needs one of: {", ".join(EVIDENCE_REQUIREMENTS[klass])}).', 'insufficient_evidence', ) def classify_row(row: pd.Series) -> tuple[str, str, str]: field_lcp_good = row.get('p75_lcp', float('inf')) <= LCP_GOOD_MS field_inp_good = row.get('p75_inp', float('inf')) <= INP_GOOD_MS field_cls_good = row.get('p75_cls', float('inf')) <= CLS_GOOD lab_lcp_good = row.get('lab_lcp_ms', float('inf')) <= LCP_GOOD_MS lab_tbt_good = row.get('lab_tbt_ms', float('inf')) <= TBT_LAB_WARNING_MS lab_cls_good = row.get('lab_cls', float('inf')) <= CLS_GOOD ttfb = row.get('p75_ttfb') ttfb_poor = pd.notna(ttfb) and ttfb > TTFB_POOR_MS rtt_high = row.get('rtt_high_fraction') rtt_high_share = pd.notna(rtt_high) and rtt_high > RTT_HIGH_FRACTION bfcache = row.get('navigation_bfcache_fraction') bfcache_high = pd.notna(bfcache) and bfcache > BFCACHE_FRACTION all_field_good = field_lcp_good and field_inp_good and field_cls_good all_lab_good = lab_lcp_good and lab_tbt_good and lab_cls_good all_field_fail = not (field_lcp_good or field_inp_good or field_cls_good) all_lab_fail = not (lab_lcp_good or lab_tbt_good or lab_cls_good) # REP is the AGREEING diagonal: the taxonomy defines it as "lab and field both pass OR BOTH # FAIL within tolerance". Only the both-pass half used to be implemented, so an origin that # lab and field agreed was broken fell through every branch to UNCLASSIFIED. if all_field_good and all_lab_good: return 'REP', 'Lab and field both pass primary metric gates.', 'evidence_present' if all_field_fail and all_lab_fail: return 'REP', 'Lab and field both fail primary metric gates — agreement, not divergence.', 'evidence_present' if (not field_lcp_good) and lab_lcp_good and ttfb_poor: return _decide(row, 'TTFB_FIELD', 'Field LCP fails despite lab LCP pass and CrUX TTFB is poor.') if (not field_inp_good) and lab_tbt_good: return _decide(row, 'INP_BEHAVIOR', 'Field INP fails despite acceptable load-time TBT.') if field_inp_good and (not lab_tbt_good): return _decide(row, 'TBT_FALSE_ALARM', 'Lab TBT warning but field INP is good.') if field_cls_good and (not lab_cls_good): return _decide(row, 'CLS_LAB_ONLY', 'Lab CLS fails but field CLS is good.') if (not field_cls_good) and lab_cls_good: return _decide(row, 'CLS_FIELD_ONLY', 'Field CLS fails but lab CLS is good.') if bfcache_high: return _decide(row, 'BFCACHE_SW', 'High bfcache/repeat navigation share may affect lab-field transportability.') if rtt_high_share: return _decide(row, 'GEO_RTT', 'High field RTT share suggests geographic or network-population divergence.') # The remaining four classes are never decidable from a public origin-month panel: each needs # a breakdown the panel does not carry. They are reachable only when the study supplies the # evidence columns, and are listed here so the classifier covers all 12 taxonomy classes # rather than silently omitting a third of them. for klass, pattern in ( ('DEVICE_MIX', 'device_class_breakdown'), ('PAGE_MIX', 'url_level_crux_panel'), ('TIME_WINDOW', 'deploy_event_date'), ('CONTENT_VARIANCE', 'lab_lcp_iqr_ms'), ): if pattern in row.index and pd.notna(row.get(pattern)): return _decide(row, klass, f'Panel supplies {pattern}; stratified evidence indicates {klass}.') return ( 'UNCLASSIFIED', 'No conservative rule fired; inspect page mix, timing, content variance, and missing features.', 'no_pattern_matched', ) def check_contract(df: pd.DataFrame) -> list[str]: """Return the required columns the panel is missing. Empty list means the contract holds.""" return [c for c in REQUIRED_COLUMNS if c not in df.columns] def classify(df: pd.DataFrame) -> pd.DataFrame: missing = check_contract(df) if missing: raise SystemExit( f'Panel is missing required columns: {missing}. ' f'Expected the step-01/03 contract: {REQUIRED_COLUMNS + OPTIONAL_COLUMNS}' ) labels = df.apply(classify_row, axis=1, result_type='expand') labels.columns = ['divergence_type', 'divergence_reason', 'evidence_status'] return pd.concat([df.copy(), labels], axis=1) if __name__ == '__main__': import argparse parser = argparse.ArgumentParser(description=__doc__) parser.add_argument('input_csv') parser.add_argument('output_csv') parser.add_argument('--check-contract', action='store_true', help='Validate the input contract and exit without writing output.') args = parser.parse_args() data = pd.read_csv(args.input_csv) if args.check_contract: gaps = check_contract(data) print('contract OK' if not gaps else f'missing required columns: {gaps}') raise SystemExit(0 if not gaps else 1) classify(data).to_csv(args.output_csv, index=False)