| 1 | #!/usr/bin/env python3 |
| 2 | """Track press-correlated repo momentum over time. |
| 3 | |
| 4 | Reads correlation data to find press-correlated repos, checks their star |
| 5 | trajectory at week +2 and +4, and classifies growth as "sustained" or "faded". |
| 6 | |
| 7 | Usage: |
| 8 | python scripts/momentum_tracker.py [--topic ai-ml] [--week 2026-W21] [--lag 4] |
| 9 | """ |
| 10 | |
| 11 | from __future__ import annotations |
| 12 | |
| 13 | import argparse |
| 14 | import json |
| 15 | import sys |
| 16 | from datetime import datetime, timedelta |
| 17 | from pathlib import Path |
| 18 | from typing import Any |
| 19 | |
| 20 | from scripts import topic_paths |
| 21 | |
| 22 | |
| 23 | def parse_args(argv: list[str] | None = None) -> argparse.Namespace: |
| 24 | parser = argparse.ArgumentParser( |
| 25 | description="Track press-correlated repo momentum" |
| 26 | ) |
| 27 | parser.add_argument( |
| 28 | "--topic", |
| 29 | default=None, |
| 30 | help="Topic ID for path resolution (default: from config or general).", |
| 31 | ) |
| 32 | parser.add_argument( |
| 33 | "--week", |
| 34 | default=None, |
| 35 | help="Base week to track from (YYYY-WNN). Defaults to current ISO week.", |
| 36 | ) |
| 37 | parser.add_argument( |
| 38 | "--lag", |
| 39 | type=int, |
| 40 | default=4, |
| 41 | help="Maximum lag in weeks to check trajectory (default: 4).", |
| 42 | ) |
| 43 | return parser.parse_args(argv) |
| 44 | |
| 45 | |
| 46 | def current_iso_week() -> str: |
| 47 | """Return the current ISO week as YYYY-WNN.""" |
| 48 | now = datetime.now() |
| 49 | cal = now.isocalendar() |
| 50 | return f"{cal[0]}-W{cal[1]:02d}" |
| 51 | |
| 52 | |
| 53 | def iso_week_to_date(week_str: str) -> datetime: |
| 54 | """Convert YYYY-WNN to a datetime (Monday of that week).""" |
| 55 | year, week_num = week_str.split("-W") |
| 56 | return datetime.strptime(f"{year}-W{int(week_num):02d}-1", "%G-W%V-%u") |
| 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_json_safe(path: Path) -> dict[str, Any] | None: |
| 68 | """Load JSON file, returning None on missing or invalid.""" |
| 69 | if not path.exists(): |
| 70 | return None |
| 71 | try: |
| 72 | with open(path, encoding="utf-8") as f: |
| 73 | return json.load(f) |
| 74 | except (json.JSONDecodeError, OSError): |
| 75 | return None |
| 76 | |
| 77 | |
| 78 | def find_correlation_file(analyzed_dir: Path, week: str) -> Path | None: |
| 79 | """Find correlation file for a given week.""" |
| 80 | path = analyzed_dir / f"{week}-correlations.json" |
| 81 | if path.exists(): |
| 82 | return path |
| 83 | matches = sorted(analyzed_dir.glob(f"{week}*correlation*.json")) |
| 84 | return matches[0] if matches else None |
| 85 | |
| 86 | |
| 87 | def extract_correlated_repos(correlations: dict[str, Any]) -> list[dict[str, Any]]: |
| 88 | """Extract press-correlated repos from correlation data.""" |
| 89 | repos = [] |
| 90 | entries = correlations.get("correlations", []) |
| 91 | for entry in entries: |
| 92 | if entry.get("press_correlated", False): |
| 93 | repos.append(entry) |
| 94 | return repos |
| 95 | |
| 96 | |
| 97 | def get_repo_stars_gained(raw_data: dict[str, Any] | None, repo_name: str) -> int | None: |
| 98 | """Extract stars_gained for a repo from raw week data.""" |
| 99 | if raw_data is None: |
| 100 | return None |
| 101 | for key in ("repos", "repositories", "new_repos", "trending_repos"): |
| 102 | for repo in raw_data.get(key, []): |
| 103 | name = repo.get("full_name") or repo.get("repo") or repo.get("name", "") |
| 104 | if name == repo_name: |
| 105 | return repo.get("stars_gained") |
| 106 | if isinstance(raw_data, list): |
| 107 | for repo in raw_data: |
| 108 | name = repo.get("full_name") or repo.get("repo") or repo.get("name", "") |
| 109 | if name == repo_name: |
| 110 | return repo.get("stars_gained") |
| 111 | return None |
| 112 | |
| 113 | |
| 114 | def compute_decay_rate(initial: int, current: int) -> float: |
| 115 | """Compute decay rate: 1 - (current / initial). Clamped to [0, 1].""" |
| 116 | if initial <= 0: |
| 117 | return 0.0 |
| 118 | rate = 1.0 - (current / initial) |
| 119 | return round(max(0.0, min(1.0, rate)), 4) |
| 120 | |
| 121 | |
| 122 | def classify_momentum( |
| 123 | initial_gained: int, |
| 124 | week2_gained: int | None, |
| 125 | week4_gained: int | None, |
| 126 | lag: int, |
| 127 | ) -> str: |
| 128 | """Classify as 'sustained' or 'faded' based on trajectory.""" |
| 129 | if initial_gained <= 0: |
| 130 | return "faded" |
| 131 | |
| 132 | check_gained = None |
| 133 | if lag >= 4 and week4_gained is not None: |
| 134 | check_gained = week4_gained |
| 135 | elif week2_gained is not None: |
| 136 | check_gained = week2_gained |
| 137 | |
| 138 | if check_gained is None: |
| 139 | return "faded" |
| 140 | |
| 141 | if check_gained >= initial_gained * 0.2: |
| 142 | return "sustained" |
| 143 | return "faded" |
| 144 | |
| 145 | |
| 146 | def track_repo_momentum( |
| 147 | repo_name: str, |
| 148 | initial_gained: int, |
| 149 | raw_dir: Path, |
| 150 | base_week: str, |
| 151 | lag: int, |
| 152 | ) -> dict[str, Any]: |
| 153 | """Track a single repo's momentum over time.""" |
| 154 | w2 = week_offset(base_week, 2) |
| 155 | w2_data = load_json_safe(raw_dir / f"{w2}.json") |
| 156 | week2_gained = get_repo_stars_gained(w2_data, repo_name) |
| 157 | |
| 158 | w4 = week_offset(base_week, 4) |
| 159 | w4_data = load_json_safe(raw_dir / f"{w4}.json") |
| 160 | week4_gained = get_repo_stars_gained(w4_data, repo_name) |
| 161 | |
| 162 | classification = classify_momentum(initial_gained, week2_gained, week4_gained, lag) |
| 163 | |
| 164 | best_later = week4_gained if (lag >= 4 and week4_gained is not None) else week2_gained |
| 165 | decay_rate = compute_decay_rate(initial_gained, best_later or 0) if initial_gained > 0 else 0.0 |
| 166 | |
| 167 | return { |
| 168 | "repo": repo_name, |
| 169 | "initial_stars_gained": initial_gained, |
| 170 | "week2_stars_gained": week2_gained, |
| 171 | "week4_stars_gained": week4_gained, |
| 172 | "classification": classification, |
| 173 | "decay_rate": decay_rate, |
| 174 | } |
| 175 | |
| 176 | |
| 177 | def update_predictions_validated( |
| 178 | predictions_path: Path, |
| 179 | tracked_repos: list[dict[str, Any]], |
| 180 | ) -> int: |
| 181 | """Update predictions.jsonl with momentum validation results.""" |
| 182 | if not predictions_path.exists(): |
| 183 | return 0 |
| 184 | |
| 185 | predictions = [] |
| 186 | with open(predictions_path, encoding="utf-8") as f: |
| 187 | for line in f: |
| 188 | line = line.strip() |
| 189 | if line: |
| 190 | predictions.append(json.loads(line)) |
| 191 | |
| 192 | if not predictions: |
| 193 | return 0 |
| 194 | |
| 195 | repo_results = {r["repo"]: r["classification"] for r in tracked_repos} |
| 196 | |
| 197 | updated = 0 |
| 198 | for pred in predictions: |
| 199 | if pred.get("validated") is not None: |
| 200 | continue |
| 201 | repo = pred.get("repo", "") |
| 202 | if repo in repo_results: |
| 203 | pred["validated"] = repo_results[repo] == "sustained" |
| 204 | updated += 1 |
| 205 | |
| 206 | if updated > 0: |
| 207 | with open(predictions_path, "w", encoding="utf-8") as f: |
| 208 | for pred in predictions: |
| 209 | f.write(json.dumps(pred, ensure_ascii=False) + "\n") |
| 210 | |
| 211 | return updated |
| 212 | |
| 213 | |
| 214 | def run_momentum_tracking( |
| 215 | topic_id: str | None = None, |
| 216 | week: str | None = None, |
| 217 | lag: int = 4, |
| 218 | ) -> dict[str, Any]: |
| 219 | """Main tracking logic. Returns momentum report.""" |
| 220 | base_week = week or current_iso_week() |
| 221 | raw_directory = topic_paths.raw_dir(topic_id) |
| 222 | analyzed_directory = topic_paths.analyzed_dir(topic_id) |
| 223 | metrics_directory = topic_paths.metrics_dir(topic_id) |
| 224 | |
| 225 | corr_file = find_correlation_file(analyzed_directory, base_week) |
| 226 | if corr_file is None: |
| 227 | print(f"No correlation data for week {base_week} in {analyzed_directory}", file=sys.stderr) |
| 228 | return {"week": base_week, "tracked_repos": [], "summary": {"total": 0, "sustained": 0, "faded": 0}} |
| 229 | |
| 230 | correlations = load_json_safe(corr_file) |
| 231 | if correlations is None: |
| 232 | print(f"Failed to load {corr_file}", file=sys.stderr) |
| 233 | return {"week": base_week, "tracked_repos": [], "summary": {"total": 0, "sustained": 0, "faded": 0}} |
| 234 | |
| 235 | correlated = extract_correlated_repos(correlations) |
| 236 | if not correlated: |
| 237 | print(f"No press-correlated repos found for {base_week}", file=sys.stderr) |
| 238 | return {"week": base_week, "tracked_repos": [], "summary": {"total": 0, "sustained": 0, "faded": 0}} |
| 239 | |
| 240 | base_raw = load_json_safe(raw_directory / f"{base_week}.json") |
| 241 | |
| 242 | tracked_repos = [] |
| 243 | for entry in correlated: |
| 244 | repo_name = entry.get("repo", "") |
| 245 | if not repo_name: |
| 246 | continue |
| 247 | |
| 248 | initial_gained = get_repo_stars_gained(base_raw, repo_name) |
| 249 | if initial_gained is None: |
| 250 | initial_gained = entry.get("stars_gained", 0) or 0 |
| 251 | |
| 252 | result = track_repo_momentum(repo_name, initial_gained, raw_directory, base_week, lag) |
| 253 | tracked_repos.append(result) |
| 254 | |
| 255 | sustained = sum(1 for r in tracked_repos if r["classification"] == "sustained") |
| 256 | faded = sum(1 for r in tracked_repos if r["classification"] == "faded") |
| 257 | |
| 258 | report = { |
| 259 | "week": base_week, |
| 260 | "tracked_repos": tracked_repos, |
| 261 | "summary": { |
| 262 | "total": len(tracked_repos), |
| 263 | "sustained": sustained, |
| 264 | "faded": faded, |
| 265 | }, |
| 266 | } |
| 267 | |
| 268 | metrics_directory.mkdir(parents=True, exist_ok=True) |
| 269 | output_path = metrics_directory / f"momentum-{base_week}.json" |
| 270 | with open(output_path, "w", encoding="utf-8") as f: |
| 271 | json.dump(report, f, indent=2, ensure_ascii=False) |
| 272 | f.write("\n") |
| 273 | print(f"Wrote momentum report to {output_path}") |
| 274 | |
| 275 | predictions_path = metrics_directory / "predictions.jsonl" |
| 276 | updated = update_predictions_validated(predictions_path, tracked_repos) |
| 277 | if updated: |
| 278 | print(f"Updated {updated} predictions in {predictions_path}") |
| 279 | |
| 280 | return report |
| 281 | |
| 282 | |
| 283 | def main(argv: list[str] | None = None) -> dict[str, Any]: |
| 284 | """CLI entry point.""" |
| 285 | args = parse_args(argv) |
| 286 | return run_momentum_tracking( |
| 287 | topic_id=args.topic, |
| 288 | week=args.week, |
| 289 | lag=args.lag, |
| 290 | ) |
| 291 | |
| 292 | |
| 293 | if __name__ == "__main__": |
| 294 | main() |