| 1 | #!/usr/bin/env python3 |
| 2 | """Load and summarize prediction scorecards for reskill integration. |
| 3 | |
| 4 | Reads scorecard JSON files from data/metrics/{topic}/scorecards/ and produces |
| 5 | a markdown summary suitable for injection into reskill prompts. |
| 6 | """ |
| 7 | |
| 8 | from __future__ import annotations |
| 9 | |
| 10 | import json |
| 11 | from pathlib import Path |
| 12 | from typing import Any |
| 13 | |
| 14 | from scripts.topic_paths import metrics_dir |
| 15 | |
| 16 | DEFAULT_SCORECARD_COUNT = 4 |
| 17 | |
| 18 | |
| 19 | def scorecard_dir(topic_id: str | None = None) -> Path: |
| 20 | """Return the scorecards directory for a given topic.""" |
| 21 | return metrics_dir(topic_id) / "scorecards" |
| 22 | |
| 23 | |
| 24 | def load_scorecards( |
| 25 | topic_id: str | None = None, count: int = DEFAULT_SCORECARD_COUNT |
| 26 | ) -> list[dict[str, Any]]: |
| 27 | """Load the most recent N scorecards for a topic, sorted oldest-first.""" |
| 28 | directory = scorecard_dir(topic_id) |
| 29 | if not directory.exists(): |
| 30 | return [] |
| 31 | files = sorted(directory.glob("*-scorecard.json")) |
| 32 | if count > 0: |
| 33 | files = files[-count:] |
| 34 | cards: list[dict[str, Any]] = [] |
| 35 | for path in files: |
| 36 | try: |
| 37 | with open(path, encoding="utf-8") as f: |
| 38 | cards.append(json.load(f)) |
| 39 | except (json.JSONDecodeError, OSError): |
| 40 | continue |
| 41 | return cards |
| 42 | |
| 43 | |
| 44 | def _aggregate_stats( |
| 45 | cards: list[dict[str, Any]], |
| 46 | ) -> tuple[int, int, int, dict[str, dict[str, int]]]: |
| 47 | """Aggregate totals across multiple scorecards. |
| 48 | |
| 49 | Returns (total_validated, total_correct, total_incorrect, by_type). |
| 50 | """ |
| 51 | total_validated = 0 |
| 52 | total_correct = 0 |
| 53 | total_incorrect = 0 |
| 54 | by_type: dict[str, dict[str, int]] = {} |
| 55 | |
| 56 | for card in cards: |
| 57 | total_validated += card.get("validated", 0) |
| 58 | total_correct += card.get("correct", 0) |
| 59 | total_incorrect += card.get("incorrect", 0) |
| 60 | for pred_type, stats in card.get("by_type", {}).items(): |
| 61 | if pred_type not in by_type: |
| 62 | by_type[pred_type] = {"total": 0, "correct": 0} |
| 63 | by_type[pred_type]["total"] += stats.get("total", 0) |
| 64 | by_type[pred_type]["correct"] += stats.get("correct", 0) |
| 65 | |
| 66 | return total_validated, total_correct, total_incorrect, by_type |
| 67 | |
| 68 | |
| 69 | def _format_by_type_analysis(by_type: dict[str, dict[str, int]]) -> list[str]: |
| 70 | """Produce per-type accuracy lines for the summary.""" |
| 71 | lines: list[str] = [] |
| 72 | for pred_type, stats in sorted(by_type.items()): |
| 73 | total = stats["total"] |
| 74 | correct = stats["correct"] |
| 75 | if total == 0: |
| 76 | continue |
| 77 | accuracy = correct / total |
| 78 | pct = int(round(accuracy * 100)) |
| 79 | lines.append(f'- "{pred_type}" predictions: {pct}% accurate ({correct}/{total})') |
| 80 | return lines |
| 81 | |
| 82 | |
| 83 | def _format_recommendations( |
| 84 | by_type: dict[str, dict[str, int]], overall_accuracy: float |
| 85 | ) -> list[str]: |
| 86 | """Generate adjustment recommendations based on type performance.""" |
| 87 | recs: list[str] = [] |
| 88 | for pred_type, stats in sorted(by_type.items()): |
| 89 | total = stats["total"] |
| 90 | correct = stats["correct"] |
| 91 | if total == 0: |
| 92 | continue |
| 93 | accuracy = correct / total |
| 94 | if accuracy < 0.5: |
| 95 | recs.append( |
| 96 | f'- "{pred_type}" predictions are underperforming ({int(round(accuracy * 100))}%) ' |
| 97 | f"— raise confidence threshold or require additional signals" |
| 98 | ) |
| 99 | elif accuracy >= 0.8: |
| 100 | recs.append( |
| 101 | f'- "{pred_type}" predictions are strong ({int(round(accuracy * 100))}%) ' |
| 102 | f"— current heuristics are reliable" |
| 103 | ) |
| 104 | if not recs: |
| 105 | if overall_accuracy < 0.6: |
| 106 | recs.append( |
| 107 | "- Overall accuracy is low — review signal weighting across all prediction types" |
| 108 | ) |
| 109 | else: |
| 110 | recs.append("- No specific type-level adjustments needed at this time") |
| 111 | return recs |
| 112 | |
| 113 | |
| 114 | def format_scorecard_summary(cards: list[dict[str, Any]]) -> str: |
| 115 | """Format loaded scorecards into a markdown summary section. |
| 116 | |
| 117 | Returns empty string if cards is empty. |
| 118 | """ |
| 119 | if not cards: |
| 120 | return "" |
| 121 | |
| 122 | total_validated, total_correct, total_incorrect, by_type = _aggregate_stats(cards) |
| 123 | |
| 124 | if total_validated == 0: |
| 125 | return "" |
| 126 | |
| 127 | overall_accuracy = total_correct / total_validated if total_validated else 0.0 |
| 128 | pct = int(round(overall_accuracy * 100)) |
| 129 | weeks = len(cards) |
| 130 | week_range = f"last {weeks} week{'s' if weeks != 1 else ''}" |
| 131 | |
| 132 | lines: list[str] = [] |
| 133 | lines.append(f"## Prediction Performance ({week_range})") |
| 134 | lines.append(f"Overall accuracy: {pct}% ({total_correct}/{total_validated} correct)") |
| 135 | lines.append("") |
| 136 | lines.append("### Per-Type Accuracy:") |
| 137 | lines.extend(_format_by_type_analysis(by_type)) |
| 138 | lines.append("") |
| 139 | lines.append("### Recommended Adjustments:") |
| 140 | lines.extend(_format_recommendations(by_type, overall_accuracy)) |
| 141 | |
| 142 | return "\n".join(lines) |
| 143 | |
| 144 | |
| 145 | def render_scorecard_section( |
| 146 | topic_id: str | None = None, count: int = DEFAULT_SCORECARD_COUNT |
| 147 | ) -> str: |
| 148 | """Load scorecards and return formatted summary, or empty string if none exist.""" |
| 149 | from scripts.sanitize_repo_content import _escape_untrusted_boundaries |
| 150 | |
| 151 | cards = load_scorecards(topic_id, count) |
| 152 | summary = format_scorecard_summary(cards) |
| 153 | return _escape_untrusted_boundaries(summary) if summary else summary |