| 1 | #!/usr/bin/env python3 |
| 2 | """Calibrate hype risk scoring using accumulated momentum data. |
| 3 | |
| 4 | Reads momentum data from data/metrics/{topic}/momentum-*.json, compares |
| 5 | hype risk predictions vs actual outcomes, and outputs a calibration report |
| 6 | with recommended threshold adjustments. |
| 7 | |
| 8 | Usage: |
| 9 | python scripts/calibrate_hype_risk.py [--topic ai-ml] [--output calibration.json] |
| 10 | """ |
| 11 | |
| 12 | from __future__ import annotations |
| 13 | |
| 14 | import argparse |
| 15 | import json |
| 16 | import sys |
| 17 | from datetime import datetime |
| 18 | from pathlib import Path |
| 19 | from typing import Any |
| 20 | |
| 21 | from scripts import topic_paths |
| 22 | |
| 23 | |
| 24 | def parse_args(argv: list[str] | None = None) -> argparse.Namespace: |
| 25 | parser = argparse.ArgumentParser(description="Calibrate hype risk scoring model") |
| 26 | parser.add_argument( |
| 27 | "--topic", |
| 28 | default=None, |
| 29 | help="Topic ID for path resolution.", |
| 30 | ) |
| 31 | parser.add_argument( |
| 32 | "--output", |
| 33 | default=None, |
| 34 | help="Output path for calibration report JSON.", |
| 35 | ) |
| 36 | return parser.parse_args(argv) |
| 37 | |
| 38 | |
| 39 | def load_momentum_files(metrics_directory: Path) -> list[dict[str, Any]]: |
| 40 | """Load all momentum-*.json files from a metrics directory.""" |
| 41 | files = sorted(metrics_directory.glob("momentum-*.json")) |
| 42 | results = [] |
| 43 | for f in files: |
| 44 | try: |
| 45 | with open(f, encoding="utf-8") as fh: |
| 46 | data = json.load(fh) |
| 47 | results.append(data) |
| 48 | except (json.JSONDecodeError, OSError): |
| 49 | continue |
| 50 | return results |
| 51 | |
| 52 | |
| 53 | def load_hype_risk_files(analyzed_directory: Path) -> list[dict[str, Any]]: |
| 54 | """Load all *-hype-risk.json or hype risk assessment files.""" |
| 55 | files = sorted(analyzed_directory.glob("*hype*risk*.json")) |
| 56 | results = [] |
| 57 | for f in files: |
| 58 | try: |
| 59 | with open(f, encoding="utf-8") as fh: |
| 60 | data = json.load(fh) |
| 61 | results.append(data) |
| 62 | except (json.JSONDecodeError, OSError): |
| 63 | continue |
| 64 | return results |
| 65 | |
| 66 | |
| 67 | def build_actual_outcomes(momentum_data: list[dict[str, Any]]) -> dict[str, str]: |
| 68 | """Build repo -> actual outcome mapping from momentum data.""" |
| 69 | outcomes: dict[str, str] = {} |
| 70 | for report in momentum_data: |
| 71 | for repo in report.get("tracked_repos", []): |
| 72 | repo_name = repo.get("repo", "") |
| 73 | classification = repo.get("classification", "") |
| 74 | if repo_name and classification: |
| 75 | outcomes[repo_name] = classification |
| 76 | return outcomes |
| 77 | |
| 78 | |
| 79 | def build_predictions( |
| 80 | hype_risk_data: list[dict[str, Any]], |
| 81 | correlation_data: list[dict[str, Any]] | None = None, |
| 82 | ) -> dict[str, str]: |
| 83 | """Build repo -> predicted risk level mapping from hype risk assessments.""" |
| 84 | predictions: dict[str, str] = {} |
| 85 | for data in hype_risk_data: |
| 86 | assessments = data.get("assessments", []) |
| 87 | for assessment in assessments: |
| 88 | repo = assessment.get("repo", "") |
| 89 | risk = assessment.get("hype_risk", "") |
| 90 | if repo and risk: |
| 91 | predictions[repo] = risk |
| 92 | return predictions |
| 93 | |
| 94 | |
| 95 | def risk_to_expected_outcome(risk: str) -> str | None: |
| 96 | """Map risk level to expected momentum outcome.""" |
| 97 | if risk in ("high",): |
| 98 | return "faded" |
| 99 | if risk in ("low", "very_low"): |
| 100 | return "sustained" |
| 101 | return None |
| 102 | |
| 103 | |
| 104 | def compute_calibration( |
| 105 | predictions: dict[str, str], |
| 106 | actuals: dict[str, str], |
| 107 | ) -> dict[str, Any]: |
| 108 | """Compare predictions vs actuals and compute accuracy by category.""" |
| 109 | accuracy_by_category: dict[str, dict[str, int]] = {} |
| 110 | total_samples = 0 |
| 111 | |
| 112 | for repo, risk_level in predictions.items(): |
| 113 | if repo not in actuals: |
| 114 | continue |
| 115 | expected = risk_to_expected_outcome(risk_level) |
| 116 | if expected is None: |
| 117 | continue |
| 118 | |
| 119 | actual = actuals[repo] |
| 120 | total_samples += 1 |
| 121 | |
| 122 | if risk_level not in accuracy_by_category: |
| 123 | accuracy_by_category[risk_level] = {"predicted": 0, "correct": 0} |
| 124 | |
| 125 | accuracy_by_category[risk_level]["predicted"] += 1 |
| 126 | if expected == actual: |
| 127 | accuracy_by_category[risk_level]["correct"] += 1 |
| 128 | |
| 129 | for cat in accuracy_by_category.values(): |
| 130 | predicted = cat["predicted"] |
| 131 | cat["accuracy"] = round(cat["correct"] / predicted, 4) if predicted > 0 else 0.0 |
| 132 | |
| 133 | return { |
| 134 | "samples": total_samples, |
| 135 | "accuracy_by_category": accuracy_by_category, |
| 136 | } |
| 137 | |
| 138 | |
| 139 | def generate_recommendations( |
| 140 | calibration: dict[str, Any], |
| 141 | actuals: dict[str, str], |
| 142 | ) -> list[dict[str, Any]]: |
| 143 | """Generate threshold adjustment recommendations based on calibration.""" |
| 144 | recommendations = [] |
| 145 | accuracy_by_cat = calibration.get("accuracy_by_category", {}) |
| 146 | |
| 147 | high_stats = accuracy_by_cat.get("high", {}) |
| 148 | if high_stats.get("predicted", 0) > 0: |
| 149 | high_acc = high_stats.get("accuracy", 0) |
| 150 | if high_acc < 0.7: |
| 151 | recommendations.append( |
| 152 | { |
| 153 | "parameter": "high_risk_decay_threshold", |
| 154 | "current": 0.5, |
| 155 | "recommended": 0.6, |
| 156 | "reason": ( |
| 157 | f"High-risk accuracy is {high_acc:.0%}, below 70% target. " |
| 158 | "Raise decay threshold to reduce false positives." |
| 159 | ), |
| 160 | } |
| 161 | ) |
| 162 | |
| 163 | low_stats = accuracy_by_cat.get("low", {}) |
| 164 | if low_stats.get("predicted", 0) > 0: |
| 165 | low_acc = low_stats.get("accuracy", 0) |
| 166 | if low_acc < 0.7: |
| 167 | recommendations.append( |
| 168 | { |
| 169 | "parameter": "sustained_threshold_weeks", |
| 170 | "current": 2, |
| 171 | "recommended": 3, |
| 172 | "reason": ( |
| 173 | f"Low-risk (sustained) accuracy is {low_acc:.0%}. " |
| 174 | "Extend observation window to improve confidence." |
| 175 | ), |
| 176 | } |
| 177 | ) |
| 178 | |
| 179 | total_sustained = sum(1 for v in actuals.values() if v == "sustained") |
| 180 | total_faded = sum(1 for v in actuals.values() if v == "faded") |
| 181 | if total_sustained + total_faded > 0: |
| 182 | sustained_ratio = total_sustained / (total_sustained + total_faded) |
| 183 | if sustained_ratio > 0.7: |
| 184 | recommendations.append( |
| 185 | { |
| 186 | "parameter": "press_correlation_confidence_floor", |
| 187 | "current": 0.4, |
| 188 | "recommended": 0.5, |
| 189 | "reason": ( |
| 190 | f"Sustained ratio is {sustained_ratio:.0%}, suggesting most " |
| 191 | "press-correlated repos maintain growth. Raise confidence " |
| 192 | "floor to only flag truly risky repos." |
| 193 | ), |
| 194 | } |
| 195 | ) |
| 196 | |
| 197 | if not recommendations: |
| 198 | recommendations.append( |
| 199 | { |
| 200 | "parameter": "no_changes", |
| 201 | "current": None, |
| 202 | "recommended": None, |
| 203 | "reason": "Calibration shows acceptable accuracy. No adjustments needed.", |
| 204 | } |
| 205 | ) |
| 206 | |
| 207 | return recommendations |
| 208 | |
| 209 | |
| 210 | def run_calibration( |
| 211 | topic_id: str | None = None, |
| 212 | output_path: str | None = None, |
| 213 | ) -> dict[str, Any]: |
| 214 | """Main calibration logic.""" |
| 215 | metrics_directory = topic_paths.metrics_dir(topic_id) |
| 216 | analyzed_directory = topic_paths.analyzed_dir(topic_id) |
| 217 | |
| 218 | momentum_data = load_momentum_files(metrics_directory) |
| 219 | if not momentum_data: |
| 220 | print(f"No momentum data found in {metrics_directory}", file=sys.stderr) |
| 221 | report = { |
| 222 | "calibration_date": datetime.now().strftime("%Y-%m-%d"), |
| 223 | "samples": 0, |
| 224 | "accuracy_by_category": {}, |
| 225 | "recommended_adjustments": [], |
| 226 | } |
| 227 | if output_path: |
| 228 | out = Path(output_path) |
| 229 | out.parent.mkdir(parents=True, exist_ok=True) |
| 230 | with open(out, "w", encoding="utf-8") as f: |
| 231 | json.dump(report, f, indent=2, ensure_ascii=False) |
| 232 | f.write("\n") |
| 233 | return report |
| 234 | |
| 235 | hype_risk_data = load_hype_risk_files(analyzed_directory) |
| 236 | |
| 237 | actuals = build_actual_outcomes(momentum_data) |
| 238 | predictions = build_predictions(hype_risk_data) |
| 239 | |
| 240 | calibration = compute_calibration(predictions, actuals) |
| 241 | recommendations = generate_recommendations(calibration, actuals) |
| 242 | |
| 243 | report = { |
| 244 | "calibration_date": datetime.now().strftime("%Y-%m-%d"), |
| 245 | "samples": calibration["samples"], |
| 246 | "accuracy_by_category": calibration["accuracy_by_category"], |
| 247 | "recommended_adjustments": recommendations, |
| 248 | } |
| 249 | |
| 250 | if output_path: |
| 251 | out = Path(output_path) |
| 252 | else: |
| 253 | metrics_directory.mkdir(parents=True, exist_ok=True) |
| 254 | out = metrics_directory / "calibration-report.json" |
| 255 | |
| 256 | out.parent.mkdir(parents=True, exist_ok=True) |
| 257 | with open(out, "w", encoding="utf-8") as f: |
| 258 | json.dump(report, f, indent=2, ensure_ascii=False) |
| 259 | f.write("\n") |
| 260 | |
| 261 | print(f"Calibration report: {calibration['samples']} samples") |
| 262 | for cat, stats in calibration["accuracy_by_category"].items(): |
| 263 | print(f" [{cat}] {stats['correct']}/{stats['predicted']} ({stats['accuracy']:.0%})") |
| 264 | print(f"Wrote report to {out}") |
| 265 | |
| 266 | return report |
| 267 | |
| 268 | |
| 269 | def main(argv: list[str] | None = None) -> dict[str, Any]: |
| 270 | """CLI entry point.""" |
| 271 | args = parse_args(argv) |
| 272 | return run_calibration( |
| 273 | topic_id=args.topic, |
| 274 | output_path=args.output, |
| 275 | ) |
| 276 | |
| 277 | |
| 278 | if __name__ == "__main__": |
| 279 | main() |