Spaces:
Sleeping
Sleeping
| """Analyzes learning data and suggests/executes optimization actions. | |
| Runs as part of the hourly learning cycle.""" | |
| from __future__ import annotations | |
| import json | |
| import os | |
| from typing import Any | |
| from .db import LearningDB | |
| class Trainer: | |
| def __init__(self, db: LearningDB | None = None) -> None: | |
| self.db = db or LearningDB() | |
| def analyze(self) -> dict[str, Any]: | |
| """Run analysis on all solver stats and return recommendations.""" | |
| summary = self.db.summary() | |
| rankings = self.db.get_solver_ranking() | |
| failures = self.db.get_recent_failures(limit=50) | |
| recommendations: list[str] = [] | |
| actions: list[str] = [] | |
| # Group failures by solver | |
| failure_by_solver: dict[str, list[dict]] = {} | |
| for f in failures: | |
| failure_by_solver.setdefault(f["solver_used"], []).append(f) | |
| for solver, fails in failure_by_solver.items(): | |
| rate = len(fails) / max(summary["total_attempts"], 1) * 100 | |
| if rate > 30: | |
| recommendations.append( | |
| f"{solver}: {len(fails)} recent failures ({rate:.0f}%). Consider: lowering confidence threshold, adding preprocessing, or using fallback solver." | |
| ) | |
| # Identify underperforming solvers | |
| for r in rankings: | |
| total = r["total"] | |
| if total < 5: | |
| continue | |
| acc = r["correct"] / max(total, 1) | |
| if acc < 0.4: | |
| recommendations.append( | |
| f"{r['solver_name']} on {r['captcha_type']}: {acc:.0%} accuracy ({r['correct']}/{total}). Flagged for review." | |
| ) | |
| # Overall health | |
| if summary["total_attempts"] > 0: | |
| if summary["accuracy_pct"] < 50: | |
| recommendations.insert(0, f"CRITICAL: overall accuracy {summary['accuracy_pct']}% — pipeline needs calibration.") | |
| elif summary["accuracy_pct"] < 70: | |
| recommendations.insert(0, f"WARNING: overall accuracy {summary['accuracy_pct']}% — room for improvement.") | |
| return { | |
| "summary": summary, | |
| "rankings": rankings, | |
| "failures_analyzed": len(failures), | |
| "recommendations": recommendations, | |
| "actions": actions, | |
| "total_solvers": len(rankings), | |
| } | |
| def optimize(self, cycle_id: int) -> list[dict]: | |
| """Run one optimization pass. Logs each change.""" | |
| analysis = self.analyze() | |
| changes: list[dict] = [] | |
| # 1. If a solver is dominating (high accuracy, high volume), note it | |
| rankings = analysis["rankings"] | |
| if rankings: | |
| best = rankings[0] | |
| if best["total"] > 10 and (best["correct"] / max(best["total"], 1)) > 0.8: | |
| changes.append({ | |
| "action": "prioritize", | |
| "before_val": "", | |
| "after_val": best["solver_name"], | |
| "before_acc": 0, | |
| "after_acc": round(best["correct"] / best["total"], 3), | |
| "notes": f"{best['solver_name']} is top performer on {best['captcha_type']} ({best['correct']}/{best['total']}). Priority increased." | |
| }) | |
| # 2. If a solver has < 5 samples, flag for more testing | |
| for r in rankings: | |
| if r["total"] < 5 and r["total"] > 0: | |
| changes.append({ | |
| "action": "collect_more", | |
| "before_val": str(r["total"]), | |
| "after_val": str(r["total"] + 10), | |
| "before_acc": round(r["correct"] / max(r["total"], 1), 3), | |
| "after_acc": 0, | |
| "notes": f"{r['solver_name']} on {r['captcha_type']}: only {r['total']} samples. Need 10+ for reliable stats." | |
| }) | |
| # Log all changes | |
| for c in changes: | |
| self.db.log_optimization( | |
| cycle=cycle_id, | |
| action=c["action"], | |
| before_val=c["before_val"], | |
| after_val=c["after_val"], | |
| before_acc=c["before_acc"], | |
| after_acc=c["after_acc"], | |
| notes=c.get("notes", ""), | |
| ) | |
| return changes | |