@cryptotaxi247 / SquadScope-1 / commits / 1b24a1a

feat: momentum tracking and hype risk calibration (#81, #82) (#114)

* feat: momentum tracking and hype risk calibration (#81, #82) - Add scripts/momentum_tracker.py: tracks press-correlated repo star trajectory at week +2 and +4, classifies as sustained/faded, updates predictions.jsonl validated field - Add scripts/calibrate_hype_risk.py: reads momentum data, compares hype risk predictions vs actuals, outputs calibration report with recommended threshold adjustments - Add tests/test_momentum_tracker.py: 42 tests covering both scripts Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> * feat: budget alerts and tiered degradation (#83, #84) - budget_alerts.py: evaluates run cost and monthly spend against thresholds, emits GitHub Actions annotations, exits 1 only on hard $1.00 cap - tier_selector.py: selects service tier (normal/budget/minimal/emergency) based on cost estimates and monthly spend, outputs JSON config - Tests for both scripts (28 tests) Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --------- Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Juan Manuel Servera committed May 19, 2026 at 16:34 UTC 1b24a1a83b3bf5230b13597968b809f2536460f0
7 files changed +1277
scripts/budget_alerts.py new
+106
@@ -0,0 +1,106 @@
1 +#!/usr/bin/env python3
2 +"""Budget alerts for GitHub Actions CI.
3 +
4 +Evaluates current run cost and monthly cumulative spend against thresholds,
5 +emitting GitHub Actions annotations (::warning:: / ::error::) as appropriate.
6 +"""
7 +from __future__ import annotations
8 +
9 +import argparse
10 +import json
11 +import sys
12 +from datetime import UTC, datetime
13 +from pathlib import Path
14 +
15 +ROOT = Path(__file__).resolve().parent.parent
16 +DEFAULT_METRICS = ROOT / "data" / "metrics" / "token-usage.jsonl"
17 +
18 +# Thresholds
19 +SINGLE_RUN_WARNING = 0.50
20 +SINGLE_RUN_FAIL = 1.00
21 +MONTHLY_WARNING = 5.00
22 +MONTHLY_RECOMMEND_SWITCH = 10.00
23 +
24 +
25 +def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
26 + parser = argparse.ArgumentParser(description="Budget alerts for CI cost control.")
27 + parser.add_argument(
28 + "--run-cost",
29 + type=float,
30 + default=None,
31 + help="Cost of the current run in USD.",
32 + )
33 + parser.add_argument(
34 + "--metrics",
35 + type=Path,
36 + default=DEFAULT_METRICS,
37 + help="Path to token-usage.jsonl ledger.",
38 + )
39 + return parser.parse_args(argv)
40 +
41 +
42 +def load_monthly_spend(metrics_path: Path, now: datetime | None = None) -> float:
43 + """Sum estimated_cost for entries in the current month."""
44 + if not metrics_path.exists():
45 + return 0.0
46 + now = now or datetime.now(UTC)
47 + total = 0.0
48 + for line in metrics_path.read_text().splitlines():
49 + line = line.strip()
50 + if not line:
51 + continue
52 + try:
53 + entry = json.loads(line)
54 + except json.JSONDecodeError:
55 + continue
56 + ts = entry.get("timestamp", "")
57 + try:
58 + entry_dt = datetime.fromisoformat(ts.replace("Z", "+00:00"))
59 + except (ValueError, AttributeError):
60 + continue
61 + if entry_dt.year == now.year and entry_dt.month == now.month:
62 + total += entry.get("estimated_cost", 0.0)
63 + return total
64 +
65 +
66 +def evaluate(run_cost: float | None, monthly_spent: float) -> tuple[list[str], int]:
67 + """Return (annotations, exit_code)."""
68 + annotations: list[str] = []
69 + exit_code = 0
70 +
71 + if run_cost is not None:
72 + if run_cost > SINGLE_RUN_FAIL:
73 + annotations.append(
74 + f"::error::Single run cost ${run_cost:.2f} exceeds hard cap ${SINGLE_RUN_FAIL:.2f}"
75 + )
76 + exit_code = 1
77 + elif run_cost > SINGLE_RUN_WARNING:
78 + annotations.append(
79 + f"::warning::Single run cost ${run_cost:.2f} exceeds warning threshold ${SINGLE_RUN_WARNING:.2f}"
80 + )
81 +
82 + if monthly_spent > MONTHLY_RECOMMEND_SWITCH:
83 + annotations.append(
84 + f"::warning::Monthly spend ${monthly_spent:.2f} exceeds ${MONTHLY_RECOMMEND_SWITCH:.2f} — consider switching to a cheaper model"
85 + )
86 + elif monthly_spent > MONTHLY_WARNING:
87 + annotations.append(
88 + f"::warning::Monthly cumulative spend ${monthly_spent:.2f} exceeds ${MONTHLY_WARNING:.2f}"
89 + )
90 +
91 + return annotations, exit_code
92 +
93 +
94 +def main(argv: list[str] | None = None) -> int:
95 + args = parse_args(argv)
96 + monthly_spent = load_monthly_spend(args.metrics)
97 + if args.run_cost is not None:
98 + monthly_spent += args.run_cost
99 + annotations, exit_code = evaluate(args.run_cost, monthly_spent)
100 + for ann in annotations:
101 + print(ann)
102 + return exit_code
103 +
104 +
105 +if __name__ == "__main__":
106 + sys.exit(main())
scripts/calibrate_hype_risk.py new
+274
@@ -0,0 +1,274 @@
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 +sys.path.insert(0, str(Path(__file__).resolve().parent))
22 +import topic_paths # noqa: E402
23 +
24 +
25 +def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
26 + parser = argparse.ArgumentParser(
27 + description="Calibrate hype risk scoring model"
28 + )
29 + parser.add_argument(
30 + "--topic",
31 + default=None,
32 + help="Topic ID for path resolution.",
33 + )
34 + parser.add_argument(
35 + "--output",
36 + default=None,
37 + help="Output path for calibration report JSON.",
38 + )
39 + return parser.parse_args(argv)
40 +
41 +
42 +def load_momentum_files(metrics_directory: Path) -> list[dict[str, Any]]:
43 + """Load all momentum-*.json files from a metrics directory."""
44 + files = sorted(metrics_directory.glob("momentum-*.json"))
45 + results = []
46 + for f in files:
47 + try:
48 + with open(f, encoding="utf-8") as fh:
49 + data = json.load(fh)
50 + results.append(data)
51 + except (json.JSONDecodeError, OSError):
52 + continue
53 + return results
54 +
55 +
56 +def load_hype_risk_files(analyzed_directory: Path) -> list[dict[str, Any]]:
57 + """Load all *-hype-risk.json or hype risk assessment files."""
58 + files = sorted(analyzed_directory.glob("*hype*risk*.json"))
59 + results = []
60 + for f in files:
61 + try:
62 + with open(f, encoding="utf-8") as fh:
63 + data = json.load(fh)
64 + results.append(data)
65 + except (json.JSONDecodeError, OSError):
66 + continue
67 + return results
68 +
69 +
70 +def build_actual_outcomes(momentum_data: list[dict[str, Any]]) -> dict[str, str]:
71 + """Build repo -> actual outcome mapping from momentum data."""
72 + outcomes: dict[str, str] = {}
73 + for report in momentum_data:
74 + for repo in report.get("tracked_repos", []):
75 + repo_name = repo.get("repo", "")
76 + classification = repo.get("classification", "")
77 + if repo_name and classification:
78 + outcomes[repo_name] = classification
79 + return outcomes
80 +
81 +
82 +def build_predictions(
83 + hype_risk_data: list[dict[str, Any]],
84 + correlation_data: list[dict[str, Any]] | None = None,
85 +) -> dict[str, str]:
86 + """Build repo -> predicted risk level mapping from hype risk assessments."""
87 + predictions: dict[str, str] = {}
88 + for data in hype_risk_data:
89 + assessments = data.get("assessments", [])
90 + for assessment in assessments:
91 + repo = assessment.get("repo", "")
92 + risk = assessment.get("hype_risk", "")
93 + if repo and risk:
94 + predictions[repo] = risk
95 + return predictions
96 +
97 +
98 +def risk_to_expected_outcome(risk: str) -> str | None:
99 + """Map risk level to expected momentum outcome."""
100 + if risk in ("high",):
101 + return "faded"
102 + if risk in ("low", "very_low"):
103 + return "sustained"
104 + return None
105 +
106 +
107 +def compute_calibration(
108 + predictions: dict[str, str],
109 + actuals: dict[str, str],
110 +) -> dict[str, Any]:
111 + """Compare predictions vs actuals and compute accuracy by category."""
112 + accuracy_by_category: dict[str, dict[str, int]] = {}
113 + total_samples = 0
114 +
115 + for repo, risk_level in predictions.items():
116 + if repo not in actuals:
117 + continue
118 + expected = risk_to_expected_outcome(risk_level)
119 + if expected is None:
120 + continue
121 +
122 + actual = actuals[repo]
123 + total_samples += 1
124 +
125 + if risk_level not in accuracy_by_category:
126 + accuracy_by_category[risk_level] = {"predicted": 0, "correct": 0}
127 +
128 + accuracy_by_category[risk_level]["predicted"] += 1
129 + if expected == actual:
130 + accuracy_by_category[risk_level]["correct"] += 1
131 +
132 + for cat in accuracy_by_category.values():
133 + predicted = cat["predicted"]
134 + cat["accuracy"] = round(cat["correct"] / predicted, 4) if predicted > 0 else 0.0
135 +
136 + return {
137 + "samples": total_samples,
138 + "accuracy_by_category": accuracy_by_category,
139 + }
140 +
141 +
142 +def generate_recommendations(
143 + calibration: dict[str, Any],
144 + actuals: dict[str, str],
145 +) -> list[dict[str, Any]]:
146 + """Generate threshold adjustment recommendations based on calibration."""
147 + recommendations = []
148 + accuracy_by_cat = calibration.get("accuracy_by_category", {})
149 +
150 + high_stats = accuracy_by_cat.get("high", {})
151 + if high_stats.get("predicted", 0) > 0:
152 + high_acc = high_stats.get("accuracy", 0)
153 + if high_acc < 0.7:
154 + recommendations.append({
155 + "parameter": "high_risk_decay_threshold",
156 + "current": 0.5,
157 + "recommended": 0.6,
158 + "reason": (
159 + f"High-risk accuracy is {high_acc:.0%}, below 70% target. "
160 + "Raise decay threshold to reduce false positives."
161 + ),
162 + })
163 +
164 + low_stats = accuracy_by_cat.get("low", {})
165 + if low_stats.get("predicted", 0) > 0:
166 + low_acc = low_stats.get("accuracy", 0)
167 + if low_acc < 0.7:
168 + recommendations.append({
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 + total_sustained = sum(1 for v in actuals.values() if v == "sustained")
179 + total_faded = sum(1 for v in actuals.values() if v == "faded")
180 + if total_sustained + total_faded > 0:
181 + sustained_ratio = total_sustained / (total_sustained + total_faded)
182 + if sustained_ratio > 0.7:
183 + recommendations.append({
184 + "parameter": "press_correlation_confidence_floor",
185 + "current": 0.4,
186 + "recommended": 0.5,
187 + "reason": (
188 + f"Sustained ratio is {sustained_ratio:.0%}, suggesting most "
189 + "press-correlated repos maintain growth. Raise confidence "
190 + "floor to only flag truly risky repos."
191 + ),
192 + })
193 +
194 + if not recommendations:
195 + recommendations.append({
196 + "parameter": "no_changes",
197 + "current": None,
198 + "recommended": None,
199 + "reason": "Calibration shows acceptable accuracy. No adjustments needed.",
200 + })
201 +
202 + return recommendations
203 +
204 +
205 +def run_calibration(
206 + topic_id: str | None = None,
207 + output_path: str | None = None,
208 +) -> dict[str, Any]:
209 + """Main calibration logic."""
210 + metrics_directory = topic_paths.metrics_dir(topic_id)
211 + analyzed_directory = topic_paths.analyzed_dir(topic_id)
212 +
213 + momentum_data = load_momentum_files(metrics_directory)
214 + if not momentum_data:
215 + print(f"No momentum data found in {metrics_directory}", file=sys.stderr)
216 + report = {
217 + "calibration_date": datetime.now().strftime("%Y-%m-%d"),
218 + "samples": 0,
219 + "accuracy_by_category": {},
220 + "recommended_adjustments": [],
221 + }
222 + if output_path:
223 + out = Path(output_path)
224 + out.parent.mkdir(parents=True, exist_ok=True)
225 + with open(out, "w", encoding="utf-8") as f:
226 + json.dump(report, f, indent=2, ensure_ascii=False)
227 + f.write("\n")
228 + return report
229 +
230 + hype_risk_data = load_hype_risk_files(analyzed_directory)
231 +
232 + actuals = build_actual_outcomes(momentum_data)
233 + predictions = build_predictions(hype_risk_data)
234 +
235 + calibration = compute_calibration(predictions, actuals)
236 + recommendations = generate_recommendations(calibration, actuals)
237 +
238 + report = {
239 + "calibration_date": datetime.now().strftime("%Y-%m-%d"),
240 + "samples": calibration["samples"],
241 + "accuracy_by_category": calibration["accuracy_by_category"],
242 + "recommended_adjustments": recommendations,
243 + }
244 +
245 + if output_path:
246 + out = Path(output_path)
247 + else:
248 + metrics_directory.mkdir(parents=True, exist_ok=True)
249 + out = metrics_directory / "calibration-report.json"
250 +
251 + out.parent.mkdir(parents=True, exist_ok=True)
252 + with open(out, "w", encoding="utf-8") as f:
253 + json.dump(report, f, indent=2, ensure_ascii=False)
254 + f.write("\n")
255 +
256 + print(f"Calibration report: {calibration['samples']} samples")
257 + for cat, stats in calibration["accuracy_by_category"].items():
258 + print(f" [{cat}] {stats['correct']}/{stats['predicted']} ({stats['accuracy']:.0%})")
259 + print(f"Wrote report to {out}")
260 +
261 + return report
262 +
263 +
264 +def main(argv: list[str] | None = None) -> dict[str, Any]:
265 + """CLI entry point."""
266 + args = parse_args(argv)
267 + return run_calibration(
268 + topic_id=args.topic,
269 + output_path=args.output,
270 + )
271 +
272 +
273 +if __name__ == "__main__":
274 + main()
scripts/momentum_tracker.py new
+295
@@ -0,0 +1,295 @@
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 +sys.path.insert(0, str(Path(__file__).resolve().parent))
21 +import topic_paths # noqa: E402
22 +
23 +
24 +def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
25 + parser = argparse.ArgumentParser(
26 + description="Track press-correlated repo momentum"
27 + )
28 + parser.add_argument(
29 + "--topic",
30 + default=None,
31 + help="Topic ID for path resolution (default: from config or general).",
32 + )
33 + parser.add_argument(
34 + "--week",
35 + default=None,
36 + help="Base week to track from (YYYY-WNN). Defaults to current ISO week.",
37 + )
38 + parser.add_argument(
39 + "--lag",
40 + type=int,
41 + default=4,
42 + help="Maximum lag in weeks to check trajectory (default: 4).",
43 + )
44 + return parser.parse_args(argv)
45 +
46 +
47 +def current_iso_week() -> str:
48 + """Return the current ISO week as YYYY-WNN."""
49 + now = datetime.now()
50 + cal = now.isocalendar()
51 + return f"{cal[0]}-W{cal[1]:02d}"
52 +
53 +
54 +def iso_week_to_date(week_str: str) -> datetime:
55 + """Convert YYYY-WNN to a datetime (Monday of that week)."""
56 + year, week_num = week_str.split("-W")
57 + return datetime.strptime(f"{year}-W{int(week_num):02d}-1", "%G-W%V-%u")
58 +
59 +
60 +def week_offset(week_str: str, offset: int) -> str:
61 + """Return a week string offset by N weeks."""
62 + dt = iso_week_to_date(week_str)
63 + new_dt = dt + timedelta(weeks=offset)
64 + cal = new_dt.isocalendar()
65 + return f"{cal[0]}-W{cal[1]:02d}"
66 +
67 +
68 +def load_json_safe(path: Path) -> dict[str, Any] | None:
69 + """Load JSON file, returning None on missing or invalid."""
70 + if not path.exists():
71 + return None
72 + try:
73 + with open(path, encoding="utf-8") as f:
74 + return json.load(f)
75 + except (json.JSONDecodeError, OSError):
76 + return None
77 +
78 +
79 +def find_correlation_file(analyzed_dir: Path, week: str) -> Path | None:
80 + """Find correlation file for a given week."""
81 + path = analyzed_dir / f"{week}-correlations.json"
82 + if path.exists():
83 + return path
84 + matches = sorted(analyzed_dir.glob(f"{week}*correlation*.json"))
85 + return matches[0] if matches else None
86 +
87 +
88 +def extract_correlated_repos(correlations: dict[str, Any]) -> list[dict[str, Any]]:
89 + """Extract press-correlated repos from correlation data."""
90 + repos = []
91 + entries = correlations.get("correlations", [])
92 + for entry in entries:
93 + if entry.get("press_correlated", False):
94 + repos.append(entry)
95 + return repos
96 +
97 +
98 +def get_repo_stars_gained(raw_data: dict[str, Any] | None, repo_name: str) -> int | None:
99 + """Extract stars_gained for a repo from raw week data."""
100 + if raw_data is None:
101 + return None
102 + for key in ("repos", "repositories", "new_repos", "trending_repos"):
103 + for repo in raw_data.get(key, []):
104 + name = repo.get("full_name") or repo.get("repo") or repo.get("name", "")
105 + if name == repo_name:
106 + return repo.get("stars_gained")
107 + if isinstance(raw_data, list):
108 + for repo in raw_data:
109 + name = repo.get("full_name") or repo.get("repo") or repo.get("name", "")
110 + if name == repo_name:
111 + return repo.get("stars_gained")
112 + return None
113 +
114 +
115 +def compute_decay_rate(initial: int, current: int) -> float:
116 + """Compute decay rate: 1 - (current / initial). Clamped to [0, 1]."""
117 + if initial <= 0:
118 + return 0.0
119 + rate = 1.0 - (current / initial)
120 + return round(max(0.0, min(1.0, rate)), 4)
121 +
122 +
123 +def classify_momentum(
124 + initial_gained: int,
125 + week2_gained: int | None,
126 + week4_gained: int | None,
127 + lag: int,
128 +) -> str:
129 + """Classify as 'sustained' or 'faded' based on trajectory."""
130 + if initial_gained <= 0:
131 + return "faded"
132 +
133 + check_gained = None
134 + if lag >= 4 and week4_gained is not None:
135 + check_gained = week4_gained
136 + elif week2_gained is not None:
137 + check_gained = week2_gained
138 +
139 + if check_gained is None:
140 + return "faded"
141 +
142 + if check_gained >= initial_gained * 0.2:
143 + return "sustained"
144 + return "faded"
145 +
146 +
147 +def track_repo_momentum(
148 + repo_name: str,
149 + initial_gained: int,
150 + raw_dir: Path,
151 + base_week: str,
152 + lag: int,
153 +) -> dict[str, Any]:
154 + """Track a single repo's momentum over time."""
155 + w2 = week_offset(base_week, 2)
156 + w2_data = load_json_safe(raw_dir / f"{w2}.json")
157 + week2_gained = get_repo_stars_gained(w2_data, repo_name)
158 +
159 + w4 = week_offset(base_week, 4)
160 + w4_data = load_json_safe(raw_dir / f"{w4}.json")
161 + week4_gained = get_repo_stars_gained(w4_data, repo_name)
162 +
163 + classification = classify_momentum(initial_gained, week2_gained, week4_gained, lag)
164 +
165 + best_later = week4_gained if (lag >= 4 and week4_gained is not None) else week2_gained
166 + decay_rate = compute_decay_rate(initial_gained, best_later or 0) if initial_gained > 0 else 0.0
167 +
168 + return {
169 + "repo": repo_name,
170 + "initial_stars_gained": initial_gained,
171 + "week2_stars_gained": week2_gained,
172 + "week4_stars_gained": week4_gained,
173 + "classification": classification,
174 + "decay_rate": decay_rate,
175 + }
176 +
177 +
178 +def update_predictions_validated(
179 + predictions_path: Path,
180 + tracked_repos: list[dict[str, Any]],
181 +) -> int:
182 + """Update predictions.jsonl with momentum validation results."""
183 + if not predictions_path.exists():
184 + return 0
185 +
186 + predictions = []
187 + with open(predictions_path, encoding="utf-8") as f:
188 + for line in f:
189 + line = line.strip()
190 + if line:
191 + predictions.append(json.loads(line))
192 +
193 + if not predictions:
194 + return 0
195 +
196 + repo_results = {r["repo"]: r["classification"] for r in tracked_repos}
197 +
198 + updated = 0
199 + for pred in predictions:
200 + if pred.get("validated") is not None:
201 + continue
202 + repo = pred.get("repo", "")
203 + if repo in repo_results:
204 + pred["validated"] = repo_results[repo] == "sustained"
205 + updated += 1
206 +
207 + if updated > 0:
208 + with open(predictions_path, "w", encoding="utf-8") as f:
209 + for pred in predictions:
210 + f.write(json.dumps(pred, ensure_ascii=False) + "\n")
211 +
212 + return updated
213 +
214 +
215 +def run_momentum_tracking(
216 + topic_id: str | None = None,
217 + week: str | None = None,
218 + lag: int = 4,
219 +) -> dict[str, Any]:
220 + """Main tracking logic. Returns momentum report."""
221 + base_week = week or current_iso_week()
222 + raw_directory = topic_paths.raw_dir(topic_id)
223 + analyzed_directory = topic_paths.analyzed_dir(topic_id)
224 + metrics_directory = topic_paths.metrics_dir(topic_id)
225 +
226 + corr_file = find_correlation_file(analyzed_directory, base_week)
227 + if corr_file is None:
228 + print(f"No correlation data for week {base_week} in {analyzed_directory}", file=sys.stderr)
229 + return {"week": base_week, "tracked_repos": [], "summary": {"total": 0, "sustained": 0, "faded": 0}}
230 +
231 + correlations = load_json_safe(corr_file)
232 + if correlations is None:
233 + print(f"Failed to load {corr_file}", file=sys.stderr)
234 + return {"week": base_week, "tracked_repos": [], "summary": {"total": 0, "sustained": 0, "faded": 0}}
235 +
236 + correlated = extract_correlated_repos(correlations)
237 + if not correlated:
238 + print(f"No press-correlated repos found for {base_week}", file=sys.stderr)
239 + return {"week": base_week, "tracked_repos": [], "summary": {"total": 0, "sustained": 0, "faded": 0}}
240 +
241 + base_raw = load_json_safe(raw_directory / f"{base_week}.json")
242 +
243 + tracked_repos = []
244 + for entry in correlated:
245 + repo_name = entry.get("repo", "")
246 + if not repo_name:
247 + continue
248 +
249 + initial_gained = get_repo_stars_gained(base_raw, repo_name)
250 + if initial_gained is None:
251 + initial_gained = entry.get("stars_gained", 0) or 0
252 +
253 + result = track_repo_momentum(repo_name, initial_gained, raw_directory, base_week, lag)
254 + tracked_repos.append(result)
255 +
256 + sustained = sum(1 for r in tracked_repos if r["classification"] == "sustained")
257 + faded = sum(1 for r in tracked_repos if r["classification"] == "faded")
258 +
259 + report = {
260 + "week": base_week,
261 + "tracked_repos": tracked_repos,
262 + "summary": {
263 + "total": len(tracked_repos),
264 + "sustained": sustained,
265 + "faded": faded,
266 + },
267 + }
268 +
269 + metrics_directory.mkdir(parents=True, exist_ok=True)
270 + output_path = metrics_directory / f"momentum-{base_week}.json"
271 + with open(output_path, "w", encoding="utf-8") as f:
272 + json.dump(report, f, indent=2, ensure_ascii=False)
273 + f.write("\n")
274 + print(f"Wrote momentum report to {output_path}")
275 +
276 + predictions_path = metrics_directory / "predictions.jsonl"
277 + updated = update_predictions_validated(predictions_path, tracked_repos)
278 + if updated:
279 + print(f"Updated {updated} predictions in {predictions_path}")
280 +
281 + return report
282 +
283 +
284 +def main(argv: list[str] | None = None) -> dict[str, Any]:
285 + """CLI entry point."""
286 + args = parse_args(argv)
287 + return run_momentum_tracking(
288 + topic_id=args.topic,
289 + week=args.week,
290 + lag=args.lag,
291 + )
292 +
293 +
294 +if __name__ == "__main__":
295 + main()
scripts/tier_selector.py new
+74
@@ -0,0 +1,74 @@
1 +#!/usr/bin/env python3
2 +"""Tiered degradation selector.
3 +
4 +Determines the appropriate service tier based on estimated cost and
5 +monthly budget consumption, outputting a JSON configuration.
6 +"""
7 +from __future__ import annotations
8 +
9 +import argparse
10 +import json
11 +import sys
12 +
13 +# Tier definitions
14 +TIERS = {
15 + "normal": {"model": "claude-sonnet-4", "max_repos": None, "skip_ai": False},
16 + "budget": {"model": "gpt-4.1", "max_repos": 100, "skip_ai": False},
17 + "minimal": {"model": "gpt-5-mini", "max_repos": 30, "skip_ai": False},
18 + "emergency": {"model": None, "max_repos": None, "skip_ai": True},
19 +}
20 +
21 +DEFAULT_MONTHLY_BUDGET = 10.00
22 +
23 +
24 +def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
25 + parser = argparse.ArgumentParser(description="Select service tier based on budget status.")
26 + parser.add_argument(
27 + "--estimated-cost",
28 + type=float,
29 + default=0.0,
30 + help="Estimated cost of the upcoming run in USD.",
31 + )
32 + parser.add_argument(
33 + "--monthly-spent",
34 + type=float,
35 + default=0.0,
36 + help="Total USD spent this month so far.",
37 + )
38 + parser.add_argument(
39 + "--monthly-budget",
40 + type=float,
41 + default=DEFAULT_MONTHLY_BUDGET,
42 + help="Monthly budget cap in USD.",
43 + )
44 + return parser.parse_args(argv)
45 +
46 +
47 +def select_tier(estimated_cost: float, monthly_spent: float, monthly_budget: float = DEFAULT_MONTHLY_BUDGET) -> str:
48 + """Determine tier based on thresholds."""
49 + if monthly_spent >= monthly_budget:
50 + return "emergency"
51 + if monthly_spent >= 8.00:
52 + return "minimal"
53 + if estimated_cost >= 0.50 or monthly_spent >= 5.00:
54 + return "budget"
55 + return "normal"
56 +
57 +
58 +def build_config(tier: str) -> dict:
59 + """Build output config dict for the given tier."""
60 + cfg = TIERS[tier].copy()
61 + cfg["tier"] = tier
62 + return {"tier": cfg["tier"], "model": cfg["model"], "max_repos": cfg["max_repos"], "skip_ai": cfg["skip_ai"]}
63 +
64 +
65 +def main(argv: list[str] | None = None) -> int:
66 + args = parse_args(argv)
67 + tier = select_tier(args.estimated_cost, args.monthly_spent, args.monthly_budget)
68 + config = build_config(tier)
69 + print(json.dumps(config))
70 + return 0
71 +
72 +
73 +if __name__ == "__main__":
74 + sys.exit(main())
tests/test_budget_alerts.py new
+95
@@ -0,0 +1,95 @@
1 +"""Tests for scripts/budget_alerts.py."""
2 +from __future__ import annotations
3 +
4 +import json
5 +from datetime import UTC, datetime
6 +from pathlib import Path
7 +
8 +import pytest
9 +
10 +from scripts.budget_alerts import (
11 + MONTHLY_RECOMMEND_SWITCH,
12 + MONTHLY_WARNING,
13 + SINGLE_RUN_FAIL,
14 + SINGLE_RUN_WARNING,
15 + evaluate,
16 + load_monthly_spend,
17 + main,
18 +)
19 +
20 +
21 +@pytest.fixture
22 +def metrics_file(tmp_path: Path) -> Path:
23 + return tmp_path / "token-usage.jsonl"
24 +
25 +
26 +class TestLoadMonthlySpend:
27 + def test_missing_file(self, tmp_path: Path):
28 + assert load_monthly_spend(tmp_path / "nonexistent.jsonl") == 0.0
29 +
30 + def test_empty_file(self, metrics_file: Path):
31 + metrics_file.write_text("")
32 + assert load_monthly_spend(metrics_file) == 0.0
33 +
34 + def test_sums_current_month(self, metrics_file: Path):
35 + now = datetime(2026, 5, 19, tzinfo=UTC)
36 + entries = [
37 + {"timestamp": "2026-05-01T10:00:00Z", "estimated_cost": 0.30},
38 + {"timestamp": "2026-05-15T10:00:00Z", "estimated_cost": 0.50},
39 + {"timestamp": "2026-04-28T10:00:00Z", "estimated_cost": 1.00}, # prev month
40 + ]
41 + metrics_file.write_text("\n".join(json.dumps(e) for e in entries))
42 + assert load_monthly_spend(metrics_file, now=now) == pytest.approx(0.80)
43 +
44 + def test_handles_malformed_lines(self, metrics_file: Path):
45 + now = datetime(2026, 5, 19, tzinfo=UTC)
46 + metrics_file.write_text("not json\n" + json.dumps({"timestamp": "2026-05-01T10:00:00Z", "estimated_cost": 0.25}))
47 + assert load_monthly_spend(metrics_file, now=now) == pytest.approx(0.25)
48 +
49 +
50 +class TestEvaluate:
51 + def test_no_alerts_under_thresholds(self):
52 + annotations, code = evaluate(0.10, 2.00)
53 + assert annotations == []
54 + assert code == 0
55 +
56 + def test_single_run_warning(self):
57 + annotations, code = evaluate(0.60, 2.00)
58 + assert any("::warning::" in a and "Single run" in a for a in annotations)
59 + assert code == 0
60 +
61 + def test_single_run_fail(self):
62 + annotations, code = evaluate(1.50, 2.00)
63 + assert any("::error::" in a for a in annotations)
64 + assert code == 1
65 +
66 + def test_monthly_warning(self):
67 + annotations, code = evaluate(None, 6.00)
68 + assert any("::warning::" in a and "cumulative" in a for a in annotations)
69 + assert code == 0
70 +
71 + def test_monthly_recommend_switch(self):
72 + annotations, code = evaluate(None, 11.00)
73 + assert any("cheaper model" in a for a in annotations)
74 + assert code == 0
75 +
76 + def test_both_single_and_monthly(self):
77 + annotations, code = evaluate(0.60, 6.00)
78 + assert len(annotations) == 2
79 + assert code == 0
80 +
81 +
82 +class TestMain:
83 + def test_exit_0_no_issues(self, metrics_file: Path):
84 + metrics_file.write_text("")
85 + code = main(["--run-cost", "0.10", "--metrics", str(metrics_file)])
86 + assert code == 0
87 +
88 + def test_exit_1_over_cap(self, metrics_file: Path):
89 + metrics_file.write_text("")
90 + code = main(["--run-cost", "1.50", "--metrics", str(metrics_file)])
91 + assert code == 1
92 +
93 + def test_missing_metrics_file(self, tmp_path: Path):
94 + code = main(["--run-cost", "0.10", "--metrics", str(tmp_path / "missing.jsonl")])
95 + assert code == 0
tests/test_momentum_tracker.py new
+362
@@ -0,0 +1,362 @@
1 +"""Tests for momentum_tracker and calibrate_hype_risk."""
2 +
3 +import json
4 +import sys
5 +from pathlib import Path
6 +
7 +import pytest
8 +
9 +sys.path.insert(0, str(Path(__file__).resolve().parent.parent / "scripts"))
10 +from momentum_tracker import ( # noqa: E402
11 + classify_momentum,
12 + compute_decay_rate,
13 + current_iso_week,
14 + extract_correlated_repos,
15 + find_correlation_file,
16 + get_repo_stars_gained,
17 + iso_week_to_date,
18 + load_json_safe,
19 + run_momentum_tracking,
20 + track_repo_momentum,
21 + update_predictions_validated,
22 + week_offset,
23 +)
24 +from calibrate_hype_risk import ( # noqa: E402
25 + build_actual_outcomes,
26 + build_predictions,
27 + compute_calibration,
28 + generate_recommendations,
29 + risk_to_expected_outcome,
30 + run_calibration,
31 +)
32 +
33 +
34 +# ---------------------------------------------------------------------------
35 +# momentum_tracker tests
36 +# ---------------------------------------------------------------------------
37 +
38 +
39 +class TestWeekUtils:
40 + """Test ISO week utility functions."""
41 +
42 + def test_current_iso_week_format(self):
43 + """current_iso_week returns YYYY-WNN format."""
44 + week = current_iso_week()
45 + assert len(week) >= 7
46 + assert "-W" in week
47 +
48 + def test_week_offset_forward(self):
49 + """week_offset moves forward correctly."""
50 + assert week_offset("2026-W21", 2) == "2026-W23"
51 +
52 + def test_week_offset_backward(self):
53 + """week_offset moves backward correctly."""
54 + assert week_offset("2026-W03", -2) == "2026-W01"
55 +
56 + def test_week_offset_year_boundary(self):
57 + """week_offset crosses year boundary."""
58 + result = week_offset("2025-W52", 2)
59 + assert result.startswith("2026-W")
60 +
61 + def test_iso_week_to_date(self):
62 + """iso_week_to_date returns correct Monday."""
63 + dt = iso_week_to_date("2026-W01")
64 + assert dt.weekday() == 0 # Monday
65 +
66 +
67 +class TestClassifyMomentum:
68 + """Test momentum classification logic."""
69 +
70 + def test_sustained_strong_growth(self):
71 + """Still gaining 20%+ of initial → sustained."""
72 + assert classify_momentum(100, 50, 30, lag=4) == "sustained"
73 +
74 + def test_faded_no_growth(self):
75 + """Zero growth at checkpoints → faded."""
76 + assert classify_momentum(100, 5, 3, lag=4) == "faded"
77 +
78 + def test_faded_zero_initial(self):
79 + """Zero initial gained → faded."""
80 + assert classify_momentum(0, 10, 5, lag=4) == "faded"
81 +
82 + def test_sustained_week2_only(self):
83 + """With lag=2, only week2 data used."""
84 + assert classify_momentum(100, 30, None, lag=2) == "sustained"
85 +
86 + def test_faded_no_data(self):
87 + """No follow-up data → faded (conservative)."""
88 + assert classify_momentum(100, None, None, lag=4) == "faded"
89 +
90 + def test_sustained_at_threshold(self):
91 + """Exactly 20% of initial → sustained."""
92 + assert classify_momentum(100, 20, 20, lag=4) == "sustained"
93 +
94 + def test_faded_below_threshold(self):
95 + """Just below 20% → faded."""
96 + assert classify_momentum(100, 19, 19, lag=4) == "faded"
97 +
98 +
99 +class TestDecayRate:
100 + """Test decay rate computation."""
101 +
102 + def test_no_decay(self):
103 + assert compute_decay_rate(100, 100) == 0.0
104 +
105 + def test_full_decay(self):
106 + assert compute_decay_rate(100, 0) == 1.0
107 +
108 + def test_partial_decay(self):
109 + assert compute_decay_rate(100, 50) == 0.5
110 +
111 + def test_zero_initial(self):
112 + assert compute_decay_rate(0, 50) == 0.0
113 +
114 + def test_negative_clamped(self):
115 + assert compute_decay_rate(100, 150) == 0.0
116 +
117 +
118 +class TestExtractCorrelatedRepos:
119 + """Test extraction from correlation data."""
120 +
121 + def test_extracts_correlated(self):
122 + data = {
123 + "correlations": [
124 + {"repo": "org/a", "press_correlated": True},
125 + {"repo": "org/b", "press_correlated": False},
126 + {"repo": "org/c", "press_correlated": True},
127 + ]
128 + }
129 + result = extract_correlated_repos(data)
130 + assert len(result) == 2
131 + assert result[0]["repo"] == "org/a"
132 + assert result[1]["repo"] == "org/c"
133 +
134 + def test_empty_correlations(self):
135 + assert extract_correlated_repos({"correlations": []}) == []
136 +
137 +
138 +class TestGetRepoStarsGained:
139 + """Test star extraction from raw data."""
140 +
141 + def test_from_repos_key(self):
142 + data = {"repos": [{"full_name": "org/x", "stars_gained": 42}]}
143 + assert get_repo_stars_gained(data, "org/x") == 42
144 +
145 + def test_missing_repo(self):
146 + data = {"repos": [{"full_name": "org/x", "stars_gained": 42}]}
147 + assert get_repo_stars_gained(data, "org/y") is None
148 +
149 + def test_none_data(self):
150 + assert get_repo_stars_gained(None, "org/x") is None
151 +
152 +
153 +class TestTrackRepoMomentum:
154 + """Test single repo momentum tracking."""
155 +
156 + def test_with_data(self, tmp_path):
157 + w2_data = {"repos": [{"full_name": "org/x", "stars_gained": 50}]}
158 + (tmp_path / "2026-W23.json").write_text(json.dumps(w2_data))
159 + w4_data = {"repos": [{"full_name": "org/x", "stars_gained": 30}]}
160 + (tmp_path / "2026-W25.json").write_text(json.dumps(w4_data))
161 +
162 + result = track_repo_momentum("org/x", 200, tmp_path, "2026-W21", lag=4)
163 + assert result["repo"] == "org/x"
164 + assert result["initial_stars_gained"] == 200
165 + assert result["week2_stars_gained"] == 50
166 + assert result["week4_stars_gained"] == 30
167 + assert result["classification"] == "faded"
168 + assert result["decay_rate"] > 0
169 +
170 + def test_missing_weeks(self, tmp_path):
171 + result = track_repo_momentum("org/x", 100, tmp_path, "2026-W21", lag=4)
172 + assert result["week2_stars_gained"] is None
173 + assert result["week4_stars_gained"] is None
174 + assert result["classification"] == "faded"
175 +
176 +
177 +class TestUpdatePredictions:
178 + """Test predictions.jsonl update."""
179 +
180 + def test_updates_matching(self, tmp_path):
181 + preds = [
182 + {"repo": "org/a", "prediction": "rising_star", "week": "2026-W20"},
183 + {"repo": "org/b", "prediction": "rising_star", "week": "2026-W20"},
184 + ]
185 + pred_path = tmp_path / "predictions.jsonl"
186 + pred_path.write_text("\n".join(json.dumps(p) for p in preds) + "\n")
187 +
188 + tracked = [
189 + {"repo": "org/a", "classification": "sustained"},
190 + {"repo": "org/b", "classification": "faded"},
191 + ]
192 + updated = update_predictions_validated(pred_path, tracked)
193 + assert updated == 2
194 +
195 + lines = pred_path.read_text().strip().split("\n")
196 + result = [json.loads(line) for line in lines]
197 + assert result[0]["validated"] is True
198 + assert result[1]["validated"] is False
199 +
200 + def test_skips_already_validated(self, tmp_path):
201 + preds = [{"repo": "org/a", "validated": True, "week": "2026-W20"}]
202 + pred_path = tmp_path / "predictions.jsonl"
203 + pred_path.write_text(json.dumps(preds[0]) + "\n")
204 +
205 + tracked = [{"repo": "org/a", "classification": "faded"}]
206 + updated = update_predictions_validated(pred_path, tracked)
207 + assert updated == 0
208 +
209 + def test_missing_file(self, tmp_path):
210 + updated = update_predictions_validated(tmp_path / "nope.jsonl", [])
211 + assert updated == 0
212 +
213 +
214 +class TestRunMomentumTracking:
215 + """Integration test for full tracking run."""
216 +
217 + def test_no_correlations(self, tmp_path, monkeypatch):
218 + import scripts.topic_paths as tp
219 +
220 + monkeypatch.setattr(tp, "DATA_ROOT", tmp_path / "data")
221 + (tmp_path / "data" / "raw").mkdir(parents=True)
222 + (tmp_path / "data" / "analyzed").mkdir(parents=True)
223 + (tmp_path / "data" / "metrics").mkdir(parents=True)
224 +
225 + result = run_momentum_tracking(topic_id=None, week="2026-W21", lag=4)
226 + assert result["week"] == "2026-W21"
227 + assert result["tracked_repos"] == []
228 + assert result["summary"]["total"] == 0
229 +
230 +
231 +# ---------------------------------------------------------------------------
232 +# calibrate_hype_risk tests
233 +# ---------------------------------------------------------------------------
234 +
235 +
236 +class TestBuildActualOutcomes:
237 + """Test outcome extraction from momentum data."""
238 +
239 + def test_extracts_outcomes(self):
240 + data = [
241 + {
242 + "tracked_repos": [
243 + {"repo": "org/a", "classification": "sustained"},
244 + {"repo": "org/b", "classification": "faded"},
245 + ]
246 + }
247 + ]
248 + outcomes = build_actual_outcomes(data)
249 + assert outcomes["org/a"] == "sustained"
250 + assert outcomes["org/b"] == "faded"
251 +
252 + def test_latest_wins(self):
253 + data = [
254 + {"tracked_repos": [{"repo": "org/a", "classification": "faded"}]},
255 + {"tracked_repos": [{"repo": "org/a", "classification": "sustained"}]},
256 + ]
257 + outcomes = build_actual_outcomes(data)
258 + assert outcomes["org/a"] == "sustained"
259 +
260 +
261 +class TestBuildPredictions:
262 + """Test prediction extraction from hype risk data."""
263 +
264 + def test_extracts_risks(self):
265 + data = [
266 + {
267 + "assessments": [
268 + {"repo": "org/a", "hype_risk": "high"},
269 + {"repo": "org/b", "hype_risk": "low"},
270 + ]
271 + }
272 + ]
273 + preds = build_predictions(data)
274 + assert preds["org/a"] == "high"
275 + assert preds["org/b"] == "low"
276 +
277 +
278 +class TestRiskToExpectedOutcome:
279 + """Test risk level to outcome mapping."""
280 +
281 + def test_high_expects_faded(self):
282 + assert risk_to_expected_outcome("high") == "faded"
283 +
284 + def test_low_expects_sustained(self):
285 + assert risk_to_expected_outcome("low") == "sustained"
286 +
287 + def test_very_low_expects_sustained(self):
288 + assert risk_to_expected_outcome("very_low") == "sustained"
289 +
290 + def test_medium_uncertain(self):
291 + assert risk_to_expected_outcome("medium") is None
292 +
293 + def test_none_uncertain(self):
294 + assert risk_to_expected_outcome("none") is None
295 +
296 +
297 +class TestComputeCalibration:
298 + """Test calibration computation."""
299 +
300 + def test_perfect_accuracy(self):
301 + predictions = {"org/a": "high", "org/b": "low"}
302 + actuals = {"org/a": "faded", "org/b": "sustained"}
303 + result = compute_calibration(predictions, actuals)
304 + assert result["samples"] == 2
305 + assert result["accuracy_by_category"]["high"]["accuracy"] == 1.0
306 + assert result["accuracy_by_category"]["low"]["accuracy"] == 1.0
307 +
308 + def test_partial_accuracy(self):
309 + predictions = {"org/a": "high", "org/b": "high"}
310 + actuals = {"org/a": "faded", "org/b": "sustained"}
311 + result = compute_calibration(predictions, actuals)
312 + assert result["samples"] == 2
313 + assert result["accuracy_by_category"]["high"]["correct"] == 1
314 + assert result["accuracy_by_category"]["high"]["predicted"] == 2
315 +
316 + def test_no_overlap(self):
317 + predictions = {"org/a": "high"}
318 + actuals = {"org/z": "faded"}
319 + result = compute_calibration(predictions, actuals)
320 + assert result["samples"] == 0
321 +
322 +
323 +class TestGenerateRecommendations:
324 + """Test recommendation generation."""
325 +
326 + def test_low_high_accuracy_triggers_adjustment(self):
327 + calibration = {
328 + "accuracy_by_category": {
329 + "high": {"predicted": 10, "correct": 5, "accuracy": 0.5}
330 + }
331 + }
332 + actuals = {"org/a": "sustained", "org/b": "faded"}
333 + recs = generate_recommendations(calibration, actuals)
334 + params = [r["parameter"] for r in recs]
335 + assert "high_risk_decay_threshold" in params
336 +
337 + def test_good_accuracy_no_changes(self):
338 + calibration = {
339 + "accuracy_by_category": {
340 + "high": {"predicted": 10, "correct": 9, "accuracy": 0.9},
341 + "low": {"predicted": 10, "correct": 9, "accuracy": 0.9},
342 + }
343 + }
344 + actuals = {"org/a": "sustained", "org/b": "faded"}
345 + recs = generate_recommendations(calibration, actuals)
346 + params = [r["parameter"] for r in recs]
347 + assert "no_changes" in params
348 +
349 +
350 +class TestRunCalibration:
351 + """Integration test for calibration."""
352 +
353 + def test_no_data(self, tmp_path, monkeypatch):
354 + import scripts.topic_paths as tp
355 +
356 + monkeypatch.setattr(tp, "DATA_ROOT", tmp_path / "data")
357 + (tmp_path / "data" / "metrics").mkdir(parents=True)
358 + (tmp_path / "data" / "analyzed").mkdir(parents=True)
359 +
360 + result = run_calibration(topic_id=None, output_path=str(tmp_path / "out.json"))
361 + assert result["samples"] == 0
362 + assert result["recommended_adjustments"] == []
tests/test_tier_selector.py new
+71
@@ -0,0 +1,71 @@
1 +"""Tests for scripts/tier_selector.py."""
2 +from __future__ import annotations
3 +
4 +import json
5 +
6 +import pytest
7 +
8 +from scripts.tier_selector import build_config, main, select_tier
9 +
10 +
11 +class TestSelectTier:
12 + def test_normal(self):
13 + assert select_tier(0.10, 2.00) == "normal"
14 +
15 + def test_budget_by_estimated_cost(self):
16 + assert select_tier(0.50, 2.00) == "budget"
17 +
18 + def test_budget_by_monthly_spent(self):
19 + assert select_tier(0.10, 5.00) == "budget"
20 +
21 + def test_minimal(self):
22 + assert select_tier(0.10, 8.00) == "minimal"
23 +
24 + def test_emergency(self):
25 + assert select_tier(0.10, 10.00) == "emergency"
26 +
27 + def test_emergency_custom_budget(self):
28 + assert select_tier(0.10, 20.00, monthly_budget=20.00) == "emergency"
29 +
30 + def test_boundary_normal(self):
31 + assert select_tier(0.49, 4.99) == "normal"
32 +
33 + def test_boundary_budget(self):
34 + assert select_tier(0.50, 4.99) == "budget"
35 +
36 +
37 +class TestBuildConfig:
38 + def test_normal_config(self):
39 + cfg = build_config("normal")
40 + assert cfg == {"tier": "normal", "model": "claude-sonnet-4", "max_repos": None, "skip_ai": False}
41 +
42 + def test_budget_config(self):
43 + cfg = build_config("budget")
44 + assert cfg == {"tier": "budget", "model": "gpt-4.1", "max_repos": 100, "skip_ai": False}
45 +
46 + def test_minimal_config(self):
47 + cfg = build_config("minimal")
48 + assert cfg == {"tier": "minimal", "model": "gpt-5-mini", "max_repos": 30, "skip_ai": False}
49 +
50 + def test_emergency_config(self):
51 + cfg = build_config("emergency")
52 + assert cfg == {"tier": "emergency", "model": None, "max_repos": None, "skip_ai": True}
53 +
54 +
55 +class TestMain:
56 + def test_normal_output(self, capsys):
57 + code = main(["--estimated-cost", "0.10", "--monthly-spent", "2.00"])
58 + assert code == 0
59 + output = json.loads(capsys.readouterr().out)
60 + assert output["tier"] == "normal"
61 +
62 + def test_emergency_output(self, capsys):
63 + code = main(["--estimated-cost", "0.10", "--monthly-spent", "10.00"])
64 + assert code == 0
65 + output = json.loads(capsys.readouterr().out)
66 + assert output["tier"] == "emergency"
67 + assert output["skip_ai"] is True
68 +
69 + def test_always_exits_0(self, capsys):
70 + code = main(["--estimated-cost", "5.00", "--monthly-spent", "99.00"])
71 + assert code == 0