#!/usr/bin/env python3 """Build exposure-matched controls for the historical MRI-002 predictive sample. This script is fully offline. It reads the accepted historical master comments and predictive 120-discussion sample, then writes new derived CSVs only. It does not modify master/calibration files and does not run NLP, sentiment analysis, topic labeling, LLM classification, MRI scoring, or interpretation. """ from __future__ import annotations import hashlib import re import sys from pathlib import Path from typing import Any import pandas as pd ROOT = Path("historical") COMMENTS_MASTER = ROOT / "comments_2017_2018_master.parquet" PREDICTIVE_DISCUSSIONS = ROOT / "mri_predictive_120_discussions.csv" MATCHED_CANDIDATES_CSV = ROOT / "mri_predictive_120_matched_candidates.csv" MATCHED_BLIND_CSV = ROOT / "mri_predictive_120_matched_blind.csv" EARLY25_CSV = ROOT / "mri_predictive_120_early25.csv" CONTROL_TARGET_PER_WINNER = 3 FUTURE_WINNERS_PER_DISCUSSION = 10 MATCH_WINDOWS = [0.01, 0.025, 0.05] BLIND_SEED = "mri002-historical-matched-v1" class MatchedControlsError(RuntimeError): """Expected preparation or validation failure.""" def require_inputs() -> None: missing = [ str(path) for path in [COMMENTS_MASTER, PREDICTIVE_DISCUSSIONS] if not path.exists() ] if missing: raise MatchedControlsError(f"Missing required input file(s): {', '.join(missing)}") def parse_bool(series: pd.Series) -> pd.Series: if series.dtype == bool: return series return series.astype(str).str.strip().str.lower().isin(["true", "1", "yes"]) def stable_token(prefix: str, *parts: Any, length: int = 14) -> str: text = "|".join("" if part is None else str(part) for part in parts) digest = hashlib.sha256(text.encode("utf-8")).hexdigest()[:length].upper() return f"{prefix}{digest}" def valid_text(series: pd.Series) -> pd.Series: return series.notna() & series.astype(str).str.strip().ne("") def word_count(value: Any) -> int: text = "" if value is None or pd.isna(value) else str(value) return len(re.findall(r"\S+", text.strip())) def load_discussions() -> pd.DataFrame: discussions = pd.read_csv(PREDICTIVE_DISCUSSIONS, dtype={"articleID": str}) if len(discussions) != 120: raise MatchedControlsError( f"Expected 120 predictive discussions, found {len(discussions)}." ) if discussions["articleID"].duplicated().any(): duplicates = discussions.loc[discussions["articleID"].duplicated(), "articleID"].tolist() raise MatchedControlsError( "Predictive discussion file contains duplicate articleID values: " + ", ".join(duplicates[:10]) ) return discussions def load_top_level_comments(article_ids: set[str]) -> pd.DataFrame: comments = pd.read_parquet( COMMENTS_MASTER, columns=[ "commentID", "articleID", "commentBody", "createDate", "recommendations", "editorsSelection", "is_top_level", ], ) comments["articleID"] = comments["articleID"].astype(str) comments = comments[comments["articleID"].isin(article_ids) & parse_bool(comments["is_top_level"])].copy() comments["commentID"] = comments["commentID"].astype(str) comments["createDate"] = pd.to_numeric(comments["createDate"], errors="coerce") comments["recommendations"] = pd.to_numeric(comments["recommendations"], errors="coerce") comments["editorsSelection"] = parse_bool(comments["editorsSelection"]) comments["has_valid_text"] = valid_text(comments["commentBody"]) comments["has_valid_timestamp"] = comments["createDate"].notna() comments["has_valid_recommendations"] = comments["recommendations"].notna() comments["is_valid_candidate"] = ( comments["has_valid_text"] & comments["has_valid_timestamp"] & comments["has_valid_recommendations"] ) return comments def add_order_and_future_target(comments: pd.DataFrame) -> pd.DataFrame: ordered = comments.sort_values( ["articleID", "createDate", "commentID"], ascending=[True, True, True], kind="mergesort", ).copy() ordered["comment_order"] = ordered.groupby("articleID").cumcount() + 1 ordered["top_level_comment_total"] = ordered.groupby("articleID")["commentID"].transform("size") ordered["comment_order_pct"] = ordered["comment_order"] / ordered["top_level_comment_total"] ordered["is_early_25"] = ordered["comment_order"].le(25) ordered["is_late_25"] = ordered["comment_order"].gt(ordered["top_level_comment_total"] - 25) future_candidates = ordered[ ordered["comment_order"].gt(25) & ordered["is_valid_candidate"] ].sort_values( ["articleID", "recommendations", "createDate", "commentID"], ascending=[True, False, True, True], kind="mergesort", ).copy() future_candidates["future_recognition_rank"] = ( future_candidates.groupby("articleID").cumcount() + 1 ) ordered["future_recognition_rank"] = pd.Series(pd.NA, index=ordered.index, dtype="Int64") ordered.loc[future_candidates.index, "future_recognition_rank"] = future_candidates[ "future_recognition_rank" ].astype("Int64") ordered["is_future_top_10_recognition"] = ( ordered["future_recognition_rank"].le(FUTURE_WINNERS_PER_DISCUSSION).fillna(False) ) ordered["word_count"] = ordered["commentBody"].map(word_count).astype("Int64") return ordered def validate_future_targets(flagged: pd.DataFrame, article_ids: set[str]) -> None: summary = flagged.groupby("articleID").agg( top_level_comments=("commentID", "size"), early_count=("is_early_25", "sum"), future_count=("is_future_top_10_recognition", "sum"), late_count=("is_late_25", "sum"), missing_text=("has_valid_text", lambda values: int((~values).sum())), missing_timestamp=("has_valid_timestamp", lambda values: int((~values).sum())), missing_recommendations=("has_valid_recommendations", lambda values: int((~values).sum())), duplicate_comment_ids=("commentID", lambda values: int(values.duplicated().sum())), ) missing_articles = sorted(article_ids - set(summary.index.astype(str))) if missing_articles: raise MatchedControlsError( "No top-level comments found for articleID(s): " + ", ".join(missing_articles[:10]) ) failures = summary[ summary["early_count"].ne(25) | summary["future_count"].ne(FUTURE_WINNERS_PER_DISCUSSION) | summary["late_count"].ne(25) | summary["missing_timestamp"].gt(0) | summary["missing_recommendations"].gt(0) | summary["duplicate_comment_ids"].gt(0) ] if not failures.empty: raise MatchedControlsError( "Future-target validation failed for articleID(s): " + ", ".join(failures.index.astype(str).tolist()[:10]) ) def choose_controls_for_winner( article_pool: pd.DataFrame, future_winner: pd.Series, used_control_ids: set[str], ) -> pd.DataFrame: base_pool = article_pool[ article_pool["comment_order"].gt(25) & article_pool["is_valid_candidate"] & ~article_pool["is_future_top_10_recognition"] ].copy() base_pool["_order_pct_difference"] = ( base_pool["comment_order_pct"] - future_winner["comment_order_pct"] ).abs() unused_pool = base_pool[~base_pool["commentID"].isin(used_control_ids)].copy() selected = pd.DataFrame(columns=base_pool.columns) for window in MATCH_WINDOWS: candidates = unused_pool[unused_pool["_order_pct_difference"].le(window)].sort_values( ["_order_pct_difference", "createDate", "commentID"], ascending=[True, True, True], kind="mergesort", ) if len(candidates) >= CONTROL_TARGET_PER_WINNER: selected = candidates.head(CONTROL_TARGET_PER_WINNER).copy() break if selected.empty: selected = unused_pool.sort_values( ["_order_pct_difference", "createDate", "commentID"], ascending=[True, True, True], kind="mergesort", ).head(CONTROL_TARGET_PER_WINNER).copy() if len(selected) < CONTROL_TARGET_PER_WINNER: needed = CONTROL_TARGET_PER_WINNER - len(selected) fallback = base_pool[~base_pool["commentID"].isin(set(selected["commentID"]))].sort_values( ["_order_pct_difference", "createDate", "commentID"], ascending=[True, True, True], kind="mergesort", ).head(needed) selected = pd.concat([selected, fallback], ignore_index=False) return selected.head(CONTROL_TARGET_PER_WINNER).copy() def candidate_id(article_id: str, pair_id: str, comment_id: str) -> str: return stable_token("CID_", BLIND_SEED, article_id, pair_id, comment_id) def build_matched_candidates( flagged: pd.DataFrame, discussions: pd.DataFrame ) -> tuple[pd.DataFrame, pd.DataFrame]: meta = discussions[["articleID", "headline"]].copy() comments = flagged.merge(meta, on="articleID", how="left") rows: list[dict[str, Any]] = [] pair_number = 0 for article_id, article in comments.groupby("articleID", sort=False): article = article.sort_values(["comment_order", "commentID"], kind="mergesort") future_winners = article[article["is_future_top_10_recognition"]].sort_values( ["future_recognition_rank", "createDate", "commentID"], ascending=[True, True, True], kind="mergesort", ) used_control_ids: set[str] = set() for _, winner in future_winners.iterrows(): pair_number += 1 pair_id = f"P{pair_number:04d}" winner_row = { "articleID": winner["articleID"], "headline": winner["headline"], "pair_id": pair_id, "candidate_role": "FUTURE_WINNER", "commentID": winner["commentID"], "commentBody": winner["commentBody"], "createDate": winner["createDate"], "comment_order": winner["comment_order"], "comment_order_pct": winner["comment_order_pct"], "recommendations": winner["recommendations"], "word_count": winner["word_count"], "matched_to_future_commentID": winner["commentID"], "order_pct_difference": 0.0, "future_recognition_rank": winner["future_recognition_rank"], } winner_row["candidate_id"] = candidate_id( str(winner["articleID"]), pair_id, str(winner["commentID"]) ) rows.append(winner_row) controls = choose_controls_for_winner(article, winner, used_control_ids) for _, control in controls.iterrows(): used_control_ids.add(str(control["commentID"])) control_row = { "articleID": control["articleID"], "headline": control["headline"], "pair_id": pair_id, "candidate_role": "MATCHED_CONTROL", "commentID": control["commentID"], "commentBody": control["commentBody"], "createDate": control["createDate"], "comment_order": control["comment_order"], "comment_order_pct": control["comment_order_pct"], "recommendations": control["recommendations"], "word_count": control["word_count"], "matched_to_future_commentID": winner["commentID"], "order_pct_difference": abs( control["comment_order_pct"] - winner["comment_order_pct"] ), "future_recognition_rank": pd.NA, } control_row["candidate_id"] = candidate_id( str(control["articleID"]), pair_id, str(control["commentID"]) ) rows.append(control_row) candidates = pd.DataFrame(rows) output_columns = [ "articleID", "headline", "pair_id", "candidate_role", "commentID", "commentBody", "createDate", "comment_order", "comment_order_pct", "recommendations", "word_count", "matched_to_future_commentID", "order_pct_difference", "candidate_id", ] candidates[output_columns].to_csv(MATCHED_CANDIDATES_CSV, index=False) return candidates, comments def write_blind_copy(candidates: pd.DataFrame) -> pd.DataFrame: blind = candidates[ [ "candidate_id", "articleID", "headline", "pair_id", "commentBody", "createDate", "comment_order", "comment_order_pct", "word_count", ] ].copy() blind["_blind_sort_key"] = blind.apply( lambda row: stable_token("", BLIND_SEED, "blind-sort", row["candidate_id"], length=16), axis=1, ) blind = blind.sort_values(["articleID", "pair_id", "_blind_sort_key"], kind="mergesort").drop( columns=["_blind_sort_key"] ) blind.to_csv(MATCHED_BLIND_CSV, index=False) return blind def write_early25(flagged_with_meta: pd.DataFrame) -> pd.DataFrame: early = flagged_with_meta[flagged_with_meta["is_early_25"]].copy() output = early[ ["articleID", "headline", "commentID", "commentBody", "comment_order"] ].sort_values(["articleID", "comment_order"], kind="mergesort") output.to_csv(EARLY25_CSV, index=False) return output def validation_report( candidates: pd.DataFrame, blind: pd.DataFrame, early25: pd.DataFrame, discussions: pd.DataFrame, ) -> dict[str, Any]: future = candidates[candidates["candidate_role"].eq("FUTURE_WINNER")] controls = candidates[candidates["candidate_role"].eq("MATCHED_CONTROL")] controls_per_pair = controls.groupby("pair_id").size() duplicate_controls = int(controls.duplicated(subset=["articleID", "commentID"]).sum()) missing_text = int( (candidates["commentBody"].isna() | candidates["commentBody"].astype(str).str.strip().eq("")).sum() ) missing_timestamps = int(candidates["createDate"].isna().sum()) missing_recommendations = int(candidates["recommendations"].isna().sum()) control_uses_future = int( controls["commentID"].isin(set(future["commentID"].astype(str))).sum() ) controls_before_or_at_25 = int(controls["comment_order"].le(25).sum()) blind_revealing_columns = sorted( set(blind.columns).intersection( { "candidate_role", "recommendations", "future_recognition_rank", "matched_to_future_commentID", "order_pct_difference", "commentID", } ) ) return { "discussion_count": len(discussions), "future_winners": len(future), "target_controls": len(future) * CONTROL_TARGET_PER_WINNER, "actual_controls": len(controls), "future_winners_with_3_controls": int(controls_per_pair.eq(CONTROL_TARGET_PER_WINNER).sum()), "future_winners_with_fewer_than_3_controls": int( (controls_per_pair < CONTROL_TARGET_PER_WINNER).sum() ), "median_order_pct_difference": float(controls["order_pct_difference"].median()), "p90_order_pct_difference": float(controls["order_pct_difference"].quantile(0.90)), "duplicate_controls_within_article": duplicate_controls, "missing_text": missing_text, "missing_timestamps": missing_timestamps, "missing_recommendations": missing_recommendations, "control_uses_future_winner_comment": control_uses_future, "controls_before_or_at_25": controls_before_or_at_25, "blind_rows": len(blind), "blind_revealing_columns": blind_revealing_columns, "early25_rows": len(early25), "early25_articles_with_25": int(early25.groupby("articleID").size().eq(25).sum()), } def print_report(report: dict[str, Any]) -> None: print("Historical exposure-matched controls") print(f"120 discussions: {report['discussion_count']}") print(f"future winners: {report['future_winners']}") print(f"target matched controls: {report['target_controls']}") print(f"actual matched controls: {report['actual_controls']}") print(f"future winners with 3 controls: {report['future_winners_with_3_controls']}") print( "future winners with fewer than 3 controls: " f"{report['future_winners_with_fewer_than_3_controls']}" ) print( "median absolute comment_order_pct difference: " f"{report['median_order_pct_difference']:.6f}" ) print( "90th percentile order_pct difference: " f"{report['p90_order_pct_difference']:.6f}" ) print(f"duplicate controls within article: {report['duplicate_controls_within_article']}") print( "missing text/timestamps/recommendations in candidate rows: " f"{report['missing_text']}/{report['missing_timestamps']}/{report['missing_recommendations']}" ) print(f"controls using future-winner comments: {report['control_uses_future_winner_comment']}") print(f"controls with comment_order <= 25: {report['controls_before_or_at_25']}") print(f"blind rows: {report['blind_rows']}") print(f"blind revealing columns present: {report['blind_revealing_columns']}") print(f"EARLY_25 rows: {report['early25_rows']}") print(f"articles with exactly 25 EARLY_25 rows: {report['early25_articles_with_25']}") print("control selection variables used: comment_order_pct difference, createDate, commentID") print("control selection NEVER used recommendations, comment text, semantics, or categories") print(f"wrote {MATCHED_CANDIDATES_CSV}") print(f"wrote {MATCHED_BLIND_CSV}") print(f"wrote {EARLY25_CSV}") def main() -> int: try: require_inputs() discussions = load_discussions() article_ids = set(discussions["articleID"].astype(str)) comments = load_top_level_comments(article_ids) flagged = add_order_and_future_target(comments) validate_future_targets(flagged, article_ids) candidates, flagged_with_meta = build_matched_candidates(flagged, discussions) blind = write_blind_copy(candidates) early25 = write_early25(flagged_with_meta) report = validation_report(candidates, blind, early25, discussions) print_report(report) except MatchedControlsError as exc: print(f"Stopped: {exc}", file=sys.stderr) return 1 return 0 if __name__ == "__main__": raise SystemExit(main())