| 1 | #!/usr/bin/env python3 |
| 2 | """Generate prediction ledger entries from analyzed summaries and raw data. |
| 3 | |
| 4 | Reads an analyzed summary markdown and its corresponding raw JSON to produce |
| 5 | 3-5 heuristic predictions about repos likely to gain momentum. Predictions |
| 6 | are appended to a per-topic JSONL file for later validation. |
| 7 | |
| 8 | Usage: |
| 9 | python scripts/prediction_ledger.py [--input FILE] [--topic TOPIC] [--raw FILE] |
| 10 | """ |
| 11 | |
| 12 | from __future__ import annotations |
| 13 | |
| 14 | import argparse |
| 15 | import json |
| 16 | import re |
| 17 | from pathlib import Path |
| 18 | from typing import Any |
| 19 | |
| 20 | from scripts.topic_paths import analyzed_dir, metrics_dir, raw_dir |
| 21 | |
| 22 | FRONTMATTER_PATTERN = re.compile(r"^---\n(.*?)\n---\n(.*)\Z", re.DOTALL) |
| 23 | WEEK_PATTERN = re.compile(r"\d{4}-W\d{2}") |
| 24 | REPO_LINK_PATTERN = re.compile(r"\[(?P<full_name>[^\]]+/[^\]]+)\]\(https://github\.com/[^\)]+\)") |
| 25 | |
| 26 | PREDICTION_TYPES = [ |
| 27 | "rising_star", |
| 28 | "emerging_topic", |
| 29 | "momentum_shift", |
| 30 | "breakout_candidate", |
| 31 | "declining_signal", |
| 32 | ] |
| 33 | |
| 34 | MAX_PREDICTIONS = 5 |
| 35 | MIN_PREDICTIONS = 3 |
| 36 | |
| 37 | |
| 38 | def parse_args(argv: list[str] | None = None) -> argparse.Namespace: |
| 39 | parser = argparse.ArgumentParser( |
| 40 | description="Generate prediction ledger entries from analyzed summaries." |
| 41 | ) |
| 42 | parser.add_argument( |
| 43 | "--input", |
| 44 | default=None, |
| 45 | help="Path to analyzed summary markdown (data/analyzed/{topic}/YYYY-WNN-summary.md).", |
| 46 | ) |
| 47 | parser.add_argument( |
| 48 | "--topic", |
| 49 | default=None, |
| 50 | help="Topic ID for path resolution. Defaults to general.", |
| 51 | ) |
| 52 | parser.add_argument( |
| 53 | "--raw", |
| 54 | default=None, |
| 55 | help="Path to raw JSON data. Inferred from summary week if not given.", |
| 56 | ) |
| 57 | return parser.parse_args(argv) |
| 58 | |
| 59 | |
| 60 | def find_latest_summary(topic_id: str | None) -> Path: |
| 61 | """Find the most recent analyzed summary for a topic.""" |
| 62 | search = analyzed_dir(topic_id) |
| 63 | candidates = sorted(search.glob("*-summary.md")) |
| 64 | if not candidates: |
| 65 | raise FileNotFoundError(f"No summaries found in {search}") |
| 66 | return candidates[-1] |
| 67 | |
| 68 | |
| 69 | def extract_week(text: str) -> str | None: |
| 70 | """Extract YYYY-WNN week identifier from text.""" |
| 71 | match = WEEK_PATTERN.search(text) |
| 72 | return match.group(0) if match else None |
| 73 | |
| 74 | |
| 75 | def infer_raw_path(summary_path: Path, topic_id: str | None) -> Path: |
| 76 | """Infer the raw JSON path from the summary filename.""" |
| 77 | week = extract_week(summary_path.name) |
| 78 | if not week: |
| 79 | raise ValueError(f"Cannot infer week from {summary_path.name}") |
| 80 | return raw_dir(topic_id) / f"{week}.json" |
| 81 | |
| 82 | |
| 83 | def parse_summary(content: str) -> dict[str, Any]: |
| 84 | """Parse an analyzed summary markdown into frontmatter and body.""" |
| 85 | match = FRONTMATTER_PATTERN.match(content) |
| 86 | if not match: |
| 87 | return {"frontmatter": {}, "body": content} |
| 88 | |
| 89 | fm_text, body = match.group(1), match.group(2) |
| 90 | frontmatter: dict[str, Any] = {} |
| 91 | for line in fm_text.splitlines(): |
| 92 | if ":" in line: |
| 93 | key, _, value = line.partition(":") |
| 94 | value = value.strip().strip('"').strip("'") |
| 95 | frontmatter[key.strip()] = value |
| 96 | return {"frontmatter": frontmatter, "body": body} |
| 97 | |
| 98 | |
| 99 | def extract_repos_from_summary(body: str) -> list[str]: |
| 100 | """Extract repo full_names mentioned in the summary body.""" |
| 101 | seen: set[str] = set() |
| 102 | repos: list[str] = [] |
| 103 | for match in REPO_LINK_PATTERN.finditer(body): |
| 104 | name = match.group("full_name") |
| 105 | if name not in seen: |
| 106 | seen.add(name) |
| 107 | repos.append(name) |
| 108 | return repos |
| 109 | |
| 110 | |
| 111 | def load_raw_data(raw_path: Path) -> dict[str, Any]: |
| 112 | """Load and return raw JSON data.""" |
| 113 | with open(raw_path, encoding="utf-8") as f: |
| 114 | return json.load(f) |
| 115 | |
| 116 | |
| 117 | def build_repo_index(raw_data: dict[str, Any]) -> dict[str, dict[str, Any]]: |
| 118 | """Index repos from raw data by full_name for quick lookup.""" |
| 119 | index: dict[str, dict[str, Any]] = {} |
| 120 | for section in ("new_repos", "trending_repos"): |
| 121 | for repo in raw_data.get(section, []): |
| 122 | full_name = repo.get("full_name", "") |
| 123 | if full_name: |
| 124 | entry = index.get(full_name, {}) |
| 125 | entry.update(repo) |
| 126 | entry["_source"] = section |
| 127 | index[full_name] = entry |
| 128 | return index |
| 129 | |
| 130 | |
| 131 | def score_rising_star(repo: dict[str, Any]) -> float: |
| 132 | """Score a repo for rising_star potential.""" |
| 133 | stars = repo.get("stars", 0) |
| 134 | is_new = repo.get("_source") == "new_repos" |
| 135 | # High stars on a new repo is a strong signal |
| 136 | if is_new and stars >= 1000: |
| 137 | return min(0.9, 0.5 + (stars / 10000)) |
| 138 | if is_new and stars >= 100: |
| 139 | return min(0.7, 0.3 + (stars / 5000)) |
| 140 | if stars >= 5000: |
| 141 | return 0.4 |
| 142 | return 0.2 |
| 143 | |
| 144 | |
| 145 | def score_breakout_candidate(repo: dict[str, Any]) -> float: |
| 146 | """Score a repo for breakout_candidate potential.""" |
| 147 | stars = repo.get("stars", 0) |
| 148 | forks = repo.get("forks", 0) |
| 149 | is_new = repo.get("_source") == "new_repos" |
| 150 | fork_ratio = forks / max(stars, 1) |
| 151 | if is_new and fork_ratio > 0.1 and stars >= 50: |
| 152 | return min(0.8, 0.4 + fork_ratio) |
| 153 | if stars >= 500 and fork_ratio > 0.15: |
| 154 | return 0.6 |
| 155 | return 0.2 |
| 156 | |
| 157 | |
| 158 | def score_momentum_shift(repo: dict[str, Any]) -> float: |
| 159 | """Score a repo for momentum_shift (trending but established).""" |
| 160 | stars = repo.get("stars", 0) |
| 161 | is_trending = repo.get("_source") == "trending_repos" |
| 162 | if is_trending and stars >= 10000: |
| 163 | return 0.6 |
| 164 | if is_trending and stars >= 1000: |
| 165 | return 0.5 |
| 166 | return 0.2 |
| 167 | |
| 168 | |
| 169 | def classify_prediction(repo: dict[str, Any]) -> tuple[str, float, str]: |
| 170 | """Classify a repo into a prediction type with confidence and reason.""" |
| 171 | scores = { |
| 172 | "rising_star": score_rising_star(repo), |
| 173 | "breakout_candidate": score_breakout_candidate(repo), |
| 174 | "momentum_shift": score_momentum_shift(repo), |
| 175 | } |
| 176 | |
| 177 | best_type = max(scores, key=scores.get) # type: ignore[arg-type] |
| 178 | confidence = scores[best_type] |
| 179 | |
| 180 | reasons = { |
| 181 | "rising_star": f"New repo with {repo.get('stars', 0)} stars and active development", |
| 182 | "breakout_candidate": ( |
| 183 | f"High fork ratio ({repo.get('forks', 0)} forks / " |
| 184 | f"{repo.get('stars', 0)} stars) suggests community adoption" |
| 185 | ), |
| 186 | "momentum_shift": (f"Established repo ({repo.get('stars', 0)} stars) trending this week"), |
| 187 | } |
| 188 | |
| 189 | return best_type, round(confidence, 2), reasons[best_type] |
| 190 | |
| 191 | |
| 192 | def generate_predictions( |
| 193 | summary_content: str, |
| 194 | raw_data: dict[str, Any], |
| 195 | week: str, |
| 196 | ) -> list[dict[str, Any]]: |
| 197 | """Generate 3-5 predictions from analyzed summary and raw data.""" |
| 198 | parsed = parse_summary(summary_content) |
| 199 | mentioned_repos = extract_repos_from_summary(parsed["body"]) |
| 200 | repo_index = build_repo_index(raw_data) |
| 201 | |
| 202 | predictions: list[dict[str, Any]] = [] |
| 203 | |
| 204 | # Score mentioned repos that exist in raw data |
| 205 | candidates: list[tuple[str, str, float, str]] = [] |
| 206 | for repo_name in mentioned_repos: |
| 207 | if repo_name in repo_index: |
| 208 | pred_type, confidence, reason = classify_prediction(repo_index[repo_name]) |
| 209 | candidates.append((repo_name, pred_type, confidence, reason)) |
| 210 | |
| 211 | # Sort by confidence descending, take top entries |
| 212 | candidates.sort(key=lambda x: x[2], reverse=True) |
| 213 | |
| 214 | for repo_name, pred_type, confidence, reason in candidates[:MAX_PREDICTIONS]: |
| 215 | predictions.append( |
| 216 | { |
| 217 | "week": week, |
| 218 | "repo": repo_name, |
| 219 | "prediction": pred_type, |
| 220 | "confidence": confidence, |
| 221 | "reason": reason, |
| 222 | "validated": None, |
| 223 | } |
| 224 | ) |
| 225 | |
| 226 | # If we have fewer than MIN_PREDICTIONS from mentioned repos, |
| 227 | # supplement from raw data's new_repos |
| 228 | if len(predictions) < MIN_PREDICTIONS: |
| 229 | existing = {p["repo"] for p in predictions} |
| 230 | for repo in raw_data.get("new_repos", []): |
| 231 | if len(predictions) >= MIN_PREDICTIONS: |
| 232 | break |
| 233 | full_name = repo.get("full_name", "") |
| 234 | if full_name and full_name not in existing: |
| 235 | pred_type, confidence, reason = classify_prediction(repo) |
| 236 | if confidence >= 0.3: |
| 237 | predictions.append( |
| 238 | { |
| 239 | "week": week, |
| 240 | "repo": full_name, |
| 241 | "prediction": pred_type, |
| 242 | "confidence": confidence, |
| 243 | "reason": reason, |
| 244 | "validated": None, |
| 245 | } |
| 246 | ) |
| 247 | existing.add(full_name) |
| 248 | |
| 249 | return predictions[:MAX_PREDICTIONS] |
| 250 | |
| 251 | |
| 252 | def append_predictions(predictions: list[dict[str, Any]], output_path: Path) -> None: |
| 253 | """Append predictions to a JSONL file.""" |
| 254 | output_path.parent.mkdir(parents=True, exist_ok=True) |
| 255 | with open(output_path, "a", encoding="utf-8") as f: |
| 256 | for pred in predictions: |
| 257 | f.write(json.dumps(pred, ensure_ascii=False) + "\n") |
| 258 | |
| 259 | |
| 260 | def main(argv: list[str] | None = None) -> list[dict[str, Any]]: |
| 261 | """Main entry point. Returns the generated predictions.""" |
| 262 | args = parse_args(argv) |
| 263 | topic_id = args.topic |
| 264 | |
| 265 | # Resolve input summary |
| 266 | if args.input: |
| 267 | summary_path = Path(args.input) |
| 268 | else: |
| 269 | summary_path = find_latest_summary(topic_id) |
| 270 | |
| 271 | # Read summary |
| 272 | summary_content = summary_path.read_text(encoding="utf-8") |
| 273 | |
| 274 | # Resolve raw data path |
| 275 | if args.raw: |
| 276 | raw_path = Path(args.raw) |
| 277 | else: |
| 278 | raw_path = infer_raw_path(summary_path, topic_id) |
| 279 | |
| 280 | raw_data = load_raw_data(raw_path) |
| 281 | |
| 282 | # Determine week |
| 283 | week = extract_week(summary_path.name) or raw_data.get("week", "unknown") |
| 284 | |
| 285 | # Generate predictions |
| 286 | predictions = generate_predictions(summary_content, raw_data, week) |
| 287 | |
| 288 | # Write output |
| 289 | output_path = metrics_dir(topic_id) / "predictions.jsonl" |
| 290 | append_predictions(predictions, output_path) |
| 291 | |
| 292 | # Print summary |
| 293 | print(f"Generated {len(predictions)} predictions for {week}") |
| 294 | for p in predictions: |
| 295 | print(f" [{p['prediction']}] {p['repo']} (confidence: {p['confidence']})") |
| 296 | |
| 297 | return predictions |
| 298 | |
| 299 | |
| 300 | if __name__ == "__main__": |
| 301 | main() |