| 1 | #!/usr/bin/env python3 |
| 2 | """Validate predictions from N weeks ago against actual outcomes. |
| 3 | |
| 4 | Reads predictions.jsonl, finds unvalidated predictions older than --weeks-ago, |
| 5 | checks whether predicted outcomes occurred by comparing raw data from the |
| 6 | prediction week against subsequent weeks, and writes a scorecard summary. |
| 7 | |
| 8 | Usage: |
| 9 | python scripts/hindsight_validation.py [--topic ai-ml] [--weeks-ago 4] [--data-dir data/] |
| 10 | """ |
| 11 | |
| 12 | from __future__ import annotations |
| 13 | |
| 14 | import argparse |
| 15 | import json |
| 16 | from datetime import datetime, timedelta |
| 17 | from pathlib import Path |
| 18 | from typing import Any |
| 19 | |
| 20 | from scripts.topic_paths import metrics_dir, raw_dir |
| 21 | |
| 22 | |
| 23 | def parse_args(argv: list[str] | None = None) -> argparse.Namespace: |
| 24 | parser = argparse.ArgumentParser( |
| 25 | description="Validate predictions against actual outcomes." |
| 26 | ) |
| 27 | parser.add_argument( |
| 28 | "--topic", |
| 29 | default=None, |
| 30 | help="Topic ID for path resolution. Defaults to general.", |
| 31 | ) |
| 32 | parser.add_argument( |
| 33 | "--weeks-ago", |
| 34 | type=int, |
| 35 | default=4, |
| 36 | help="Minimum age in weeks for predictions to validate (default: 4).", |
| 37 | ) |
| 38 | parser.add_argument( |
| 39 | "--data-dir", |
| 40 | default="data/", |
| 41 | help="Base data directory (default: data/).", |
| 42 | ) |
| 43 | return parser.parse_args(argv) |
| 44 | |
| 45 | |
| 46 | def iso_week_to_date(week_str: str) -> datetime: |
| 47 | """Convert YYYY-WNN to a datetime (Monday of that week).""" |
| 48 | year, week_num = week_str.split("-W") |
| 49 | return datetime.strptime(f"{year}-W{int(week_num):02d}-1", "%G-W%V-%u") |
| 50 | |
| 51 | |
| 52 | def current_iso_week() -> str: |
| 53 | """Return the current ISO week as YYYY-WNN.""" |
| 54 | now = datetime.now() |
| 55 | cal = now.isocalendar() |
| 56 | return f"{cal[0]}-W{cal[1]:02d}" |
| 57 | |
| 58 | |
| 59 | def week_offset(week_str: str, offset: int) -> str: |
| 60 | """Return a week string offset by N weeks.""" |
| 61 | dt = iso_week_to_date(week_str) |
| 62 | new_dt = dt + timedelta(weeks=offset) |
| 63 | cal = new_dt.isocalendar() |
| 64 | return f"{cal[0]}-W{cal[1]:02d}" |
| 65 | |
| 66 | |
| 67 | def load_predictions(path: Path) -> list[dict[str, Any]]: |
| 68 | """Load predictions from a JSONL file.""" |
| 69 | if not path.exists(): |
| 70 | return [] |
| 71 | predictions = [] |
| 72 | with open(path, encoding="utf-8") as f: |
| 73 | for line in f: |
| 74 | line = line.strip() |
| 75 | if line: |
| 76 | predictions.append(json.loads(line)) |
| 77 | return predictions |
| 78 | |
| 79 | |
| 80 | def save_predictions(predictions: list[dict[str, Any]], path: Path) -> None: |
| 81 | """Write predictions back to a JSONL file.""" |
| 82 | path.parent.mkdir(parents=True, exist_ok=True) |
| 83 | with open(path, "w", encoding="utf-8") as f: |
| 84 | for pred in predictions: |
| 85 | f.write(json.dumps(pred, ensure_ascii=False) + "\n") |
| 86 | |
| 87 | |
| 88 | def load_raw_week(raw_directory: Path, week_str: str) -> dict[str, Any] | None: |
| 89 | """Load raw JSON for a given week. Returns None if missing.""" |
| 90 | path = raw_directory / f"{week_str}.json" |
| 91 | if not path.exists(): |
| 92 | return None |
| 93 | try: |
| 94 | with open(path, encoding="utf-8") as f: |
| 95 | return json.load(f) |
| 96 | except (json.JSONDecodeError, OSError): |
| 97 | return None |
| 98 | |
| 99 | |
| 100 | def build_repo_stars(raw_data: dict[str, Any]) -> dict[str, int]: |
| 101 | """Extract repo -> stars mapping from raw data.""" |
| 102 | stars: dict[str, int] = {} |
| 103 | for section in ("new_repos", "trending_repos"): |
| 104 | for repo in raw_data.get(section, []): |
| 105 | name = repo.get("full_name", "") |
| 106 | if name: |
| 107 | stars[name] = repo.get("stars", 0) |
| 108 | return stars |
| 109 | |
| 110 | |
| 111 | def build_repo_set(raw_data: dict[str, Any]) -> set[str]: |
| 112 | """Extract the set of repo names present in raw data.""" |
| 113 | repos: set[str] = set() |
| 114 | for section in ("new_repos", "trending_repos"): |
| 115 | for repo in raw_data.get(section, []): |
| 116 | name = repo.get("full_name", "") |
| 117 | if name: |
| 118 | repos.add(name) |
| 119 | return repos |
| 120 | |
| 121 | |
| 122 | def validate_rising_star( |
| 123 | repo: str, |
| 124 | prediction_stars: int, |
| 125 | raw_directory: Path, |
| 126 | prediction_week: str, |
| 127 | weeks_ahead: int, |
| 128 | ) -> bool | None: |
| 129 | """Validate rising_star: stars grew 20%+ in subsequent weeks.""" |
| 130 | for i in range(weeks_ahead, 0, -1): |
| 131 | w = week_offset(prediction_week, i) |
| 132 | raw_data = load_raw_week(raw_directory, w) |
| 133 | if raw_data is None: |
| 134 | continue |
| 135 | current_stars = build_repo_stars(raw_data).get(repo) |
| 136 | if current_stars is not None: |
| 137 | if prediction_stars == 0: |
| 138 | return current_stars > 0 |
| 139 | growth = (current_stars - prediction_stars) / prediction_stars |
| 140 | return growth >= 0.20 |
| 141 | return None |
| 142 | |
| 143 | |
| 144 | def validate_breakout_candidate( |
| 145 | repo: str, |
| 146 | raw_directory: Path, |
| 147 | prediction_week: str, |
| 148 | weeks_ahead: int, |
| 149 | ) -> bool | None: |
| 150 | """Validate breakout_candidate: appeared in trending in subsequent weeks.""" |
| 151 | for i in range(1, weeks_ahead + 1): |
| 152 | w = week_offset(prediction_week, i) |
| 153 | raw_data = load_raw_week(raw_directory, w) |
| 154 | if raw_data is None: |
| 155 | continue |
| 156 | trending = { |
| 157 | r.get("full_name", "") |
| 158 | for r in raw_data.get("trending_repos", []) |
| 159 | } |
| 160 | if repo in trending: |
| 161 | return True |
| 162 | return False |
| 163 | |
| 164 | |
| 165 | def validate_momentum_shift( |
| 166 | repo: str, |
| 167 | prediction_stars: int, |
| 168 | raw_directory: Path, |
| 169 | prediction_week: str, |
| 170 | weeks_ahead: int, |
| 171 | ) -> bool | None: |
| 172 | """Validate momentum_shift: trend continued in same direction.""" |
| 173 | star_history = [prediction_stars] |
| 174 | for i in range(1, weeks_ahead + 1): |
| 175 | w = week_offset(prediction_week, i) |
| 176 | raw_data = load_raw_week(raw_directory, w) |
| 177 | if raw_data is None: |
| 178 | continue |
| 179 | s = build_repo_stars(raw_data).get(repo) |
| 180 | if s is not None: |
| 181 | star_history.append(s) |
| 182 | |
| 183 | if len(star_history) < 2: |
| 184 | return None |
| 185 | return star_history[-1] >= star_history[0] |
| 186 | |
| 187 | |
| 188 | def validate_declining_signal( |
| 189 | repo: str, |
| 190 | raw_directory: Path, |
| 191 | prediction_week: str, |
| 192 | weeks_ahead: int, |
| 193 | ) -> bool | None: |
| 194 | """Validate declining_signal: repo disappeared from subsequent crawls.""" |
| 195 | found_count = 0 |
| 196 | checked_count = 0 |
| 197 | for i in range(1, weeks_ahead + 1): |
| 198 | w = week_offset(prediction_week, i) |
| 199 | raw_data = load_raw_week(raw_directory, w) |
| 200 | if raw_data is None: |
| 201 | continue |
| 202 | checked_count += 1 |
| 203 | if repo in build_repo_set(raw_data): |
| 204 | found_count += 1 |
| 205 | |
| 206 | if checked_count == 0: |
| 207 | return None |
| 208 | return found_count <= checked_count // 2 |
| 209 | |
| 210 | |
| 211 | def validate_prediction( |
| 212 | prediction: dict[str, Any], |
| 213 | raw_directory: Path, |
| 214 | weeks_ahead: int, |
| 215 | ) -> bool | None: |
| 216 | """Validate a single prediction. Returns True/False or None if insufficient data.""" |
| 217 | repo = prediction.get("repo", "") |
| 218 | prediction_week = prediction.get("week", "") |
| 219 | pred_type = prediction.get("prediction", "") |
| 220 | |
| 221 | if not repo or not prediction_week: |
| 222 | return None |
| 223 | |
| 224 | pred_raw = load_raw_week(raw_directory, prediction_week) |
| 225 | prediction_stars = 0 |
| 226 | if pred_raw: |
| 227 | prediction_stars = build_repo_stars(pred_raw).get(repo, 0) |
| 228 | |
| 229 | if pred_type == "rising_star": |
| 230 | return validate_rising_star( |
| 231 | repo, prediction_stars, raw_directory, prediction_week, weeks_ahead |
| 232 | ) |
| 233 | elif pred_type == "breakout_candidate": |
| 234 | return validate_breakout_candidate( |
| 235 | repo, raw_directory, prediction_week, weeks_ahead |
| 236 | ) |
| 237 | elif pred_type == "momentum_shift": |
| 238 | return validate_momentum_shift( |
| 239 | repo, prediction_stars, raw_directory, prediction_week, weeks_ahead |
| 240 | ) |
| 241 | elif pred_type in ("declining_signal", "emerging_topic"): |
| 242 | return validate_declining_signal( |
| 243 | repo, raw_directory, prediction_week, weeks_ahead |
| 244 | ) |
| 245 | return None |
| 246 | |
| 247 | |
| 248 | def is_old_enough(prediction_week: str, weeks_ago: int) -> bool: |
| 249 | """Check if a prediction is at least weeks_ago weeks old.""" |
| 250 | try: |
| 251 | pred_date = iso_week_to_date(prediction_week) |
| 252 | cutoff = datetime.now() - timedelta(weeks=weeks_ago) |
| 253 | return pred_date <= cutoff |
| 254 | except (ValueError, AttributeError): |
| 255 | return False |
| 256 | |
| 257 | |
| 258 | def generate_scorecard(predictions: list[dict[str, Any]]) -> dict[str, Any]: |
| 259 | """Generate a scorecard summary from validated predictions.""" |
| 260 | validated = [p for p in predictions if p.get("validated") is not None] |
| 261 | correct = [p for p in validated if p.get("validated") is True] |
| 262 | |
| 263 | by_type: dict[str, dict[str, int]] = {} |
| 264 | for p in validated: |
| 265 | t = p.get("prediction", "unknown") |
| 266 | if t not in by_type: |
| 267 | by_type[t] = {"total": 0, "correct": 0} |
| 268 | by_type[t]["total"] += 1 |
| 269 | if p.get("validated") is True: |
| 270 | by_type[t]["correct"] += 1 |
| 271 | |
| 272 | total = len(validated) |
| 273 | accuracy = round(len(correct) / total, 4) if total > 0 else 0.0 |
| 274 | |
| 275 | return { |
| 276 | "total_validated": total, |
| 277 | "correct": len(correct), |
| 278 | "incorrect": total - len(correct), |
| 279 | "accuracy": accuracy, |
| 280 | "by_type": { |
| 281 | k: { |
| 282 | **v, |
| 283 | "accuracy": round(v["correct"] / v["total"], 4) if v["total"] > 0 else 0.0, |
| 284 | } |
| 285 | for k, v in by_type.items() |
| 286 | }, |
| 287 | } |
| 288 | |
| 289 | |
| 290 | def save_scorecard(scorecard: dict[str, Any], metrics_directory: Path) -> Path: |
| 291 | """Save scorecard to data/metrics/{topic}/scorecards/YYYY-WNN-scorecard.json.""" |
| 292 | week = current_iso_week() |
| 293 | scorecards_dir = metrics_directory / "scorecards" |
| 294 | scorecards_dir.mkdir(parents=True, exist_ok=True) |
| 295 | path = scorecards_dir / f"{week}-scorecard.json" |
| 296 | with open(path, "w", encoding="utf-8") as f: |
| 297 | json.dump(scorecard, f, indent=2, ensure_ascii=False) |
| 298 | return path |
| 299 | |
| 300 | |
| 301 | def run_validation( |
| 302 | topic_id: str | None = None, |
| 303 | weeks_ago: int = 4, |
| 304 | data_dir: str = "data/", |
| 305 | ) -> dict[str, Any]: |
| 306 | """Main validation logic. Returns the scorecard.""" |
| 307 | import scripts.topic_paths as tp |
| 308 | |
| 309 | original_root = tp.DATA_ROOT |
| 310 | tp.DATA_ROOT = Path(data_dir) |
| 311 | |
| 312 | try: |
| 313 | mdir = metrics_dir(topic_id) |
| 314 | rdir = raw_dir(topic_id) |
| 315 | |
| 316 | predictions_path = mdir / "predictions.jsonl" |
| 317 | predictions = load_predictions(predictions_path) |
| 318 | |
| 319 | if not predictions: |
| 320 | scorecard = generate_scorecard([]) |
| 321 | save_scorecard(scorecard, mdir) |
| 322 | return scorecard |
| 323 | |
| 324 | validated_count = 0 |
| 325 | for pred in predictions: |
| 326 | if pred.get("validated") is not None: |
| 327 | continue |
| 328 | pred_week = pred.get("week", "") |
| 329 | if not pred_week or not is_old_enough(pred_week, weeks_ago): |
| 330 | continue |
| 331 | |
| 332 | result = validate_prediction(pred, rdir, weeks_ago) |
| 333 | if result is not None: |
| 334 | pred["validated"] = result |
| 335 | validated_count += 1 |
| 336 | |
| 337 | save_predictions(predictions, predictions_path) |
| 338 | |
| 339 | scorecard = generate_scorecard(predictions) |
| 340 | scorecard_path = save_scorecard(scorecard, mdir) |
| 341 | |
| 342 | print(f"Validated {validated_count} predictions") |
| 343 | print(f"Overall accuracy: {scorecard['accuracy']:.1%}") |
| 344 | print(f" Correct: {scorecard['correct']}") |
| 345 | print(f" Incorrect: {scorecard['incorrect']}") |
| 346 | for ptype, stats in scorecard.get("by_type", {}).items(): |
| 347 | print(f" [{ptype}] {stats['correct']}/{stats['total']} ({stats['accuracy']:.1%})") |
| 348 | print(f"Scorecard saved to {scorecard_path}") |
| 349 | |
| 350 | return scorecard |
| 351 | finally: |
| 352 | tp.DATA_ROOT = original_root |
| 353 | |
| 354 | |
| 355 | def main(argv: list[str] | None = None) -> dict[str, Any]: |
| 356 | """CLI entry point.""" |
| 357 | args = parse_args(argv) |
| 358 | return run_validation( |
| 359 | topic_id=args.topic, |
| 360 | weeks_ago=args.weeks_ago, |
| 361 | data_dir=args.data_dir, |
| 362 | ) |
| 363 | |
| 364 | |
| 365 | if __name__ == "__main__": |
| 366 | main() |