thenuke02/cs2-analyzer
0
1"""2Cross-Match Player Development Tracking.3 4Tracks player metrics over time across multiple demos, identifies5improvement trends and regression, provides role-specific benchmarks6by competitive level, and generates practice recommendations.7 8Builds on top of the existing MatchHistory table and DatabaseManager9infrastructure rather than creating separate storage.10"""11 12from __future__ import annotations13 14import logging15import statistics16from dataclasses import dataclass, field17from typing import Any18 19logger = logging.getLogger(__name__)20 21 22# =============================================================================23# Data Models24# =============================================================================25 26 27@dataclass28class MatchSnapshot:29 """Snapshot of a single match's key metrics for tracking.30 31 Can be constructed from orchestrator output or from MatchHistory DB rows.32 """33 34 # Identity35 steam_id: str36 demo_hash: str37 map_name: str | None = None38 result: str | None = None # "win", "loss", "draw"39 analyzed_at: str | None = None40 41 # Core stats42 kills: int = 043 deaths: int = 044 assists: int = 045 adr: float = 0.046 kast: float = 0.047 hs_pct: float = 0.048 rounds_played: int = 049 50 # Ratings51 hltv_rating: float = 1.052 aim_rating: float = 0.053 utility_rating: float = 0.054 55 # Advanced56 ttd_median_ms: float | None = None57 cp_median_deg: float | None = None58 59 # Duel stats60 entry_attempts: int = 061 entry_success: int = 062 clutch_situations: int = 063 clutch_wins: int = 064 trade_kill_success: int = 065 trade_kill_attempts: int = 066 67 # Utility68 enemies_flashed: int = 069 flash_assists: int = 070 he_damage: int = 071 72 def to_dict(self) -> dict[str, Any]:73 """Serialize to dict."""74 return {75 "steam_id": self.steam_id,76 "demo_hash": self.demo_hash,77 "map_name": self.map_name,78 "result": self.result,79 "analyzed_at": self.analyzed_at,80 "kills": self.kills,81 "deaths": self.deaths,82 "assists": self.assists,83 "adr": round(self.adr, 1),84 "kast": round(self.kast, 1),85 "hs_pct": round(self.hs_pct, 1),86 "rounds_played": self.rounds_played,87 "hltv_rating": round(self.hltv_rating, 2),88 "aim_rating": round(self.aim_rating, 1),89 "utility_rating": round(self.utility_rating, 1),90 "ttd_median_ms": (91 round(self.ttd_median_ms, 1) if self.ttd_median_ms is not None else None92 ),93 "cp_median_deg": (94 round(self.cp_median_deg, 1) if self.cp_median_deg is not None else None95 ),96 "entry_attempts": self.entry_attempts,97 "entry_success": self.entry_success,98 "clutch_situations": self.clutch_situations,99 "clutch_wins": self.clutch_wins,100 "trade_kill_success": self.trade_kill_success,101 "trade_kill_attempts": self.trade_kill_attempts,102 "enemies_flashed": self.enemies_flashed,103 "flash_assists": self.flash_assists,104 "he_damage": self.he_damage,105 }106 107 108@dataclass109class TrendAnalysis:110 """Trend analysis for a single metric across match windows.111 112 Compares: current match vs recent (last 5) vs historical (all matches).113 """114 115 metric_name: str116 current_value: float117 recent_avg: float # Last RECENT_WINDOW matches118 historical_avg: float # All matches119 recent_std: float = 0.0120 historical_std: float = 0.0121 direction: str = "stable" # "improving", "declining", "stable"122 change_pct: float = 0.0 # % change from historical to current123 sample_count: int = 0124 125 def to_dict(self) -> dict[str, Any]:126 """Serialize to dict."""127 return {128 "metric": self.metric_name,129 "current": round(self.current_value, 2),130 "recent_avg": round(self.recent_avg, 2),131 "historical_avg": round(self.historical_avg, 2),132 "recent_std": round(self.recent_std, 2),133 "historical_std": round(self.historical_std, 2),134 "direction": self.direction,135 "change_pct": round(self.change_pct, 1),136 "sample_count": self.sample_count,137 }138 139 140@dataclass141class RoleBenchmark:142 """Benchmark comparison for a metric against competitive levels."""143 144 metric_name: str145 player_value: float146 level: str # "beginner", "intermediate", "advanced", "elite"147 level_avg: float148 level_low: float149 level_high: float150 percentile_in_level: float # 0-100, where in this level the player falls151 verdict: str # "below", "at", "above" relative to level avg152 153 def to_dict(self) -> dict[str, Any]:154 """Serialize to dict."""155 return {156 "metric": self.metric_name,157 "player_value": round(self.player_value, 2),158 "level": self.level,159 "level_avg": round(self.level_avg, 2),160 "level_range": [round(self.level_low, 2), round(self.level_high, 2)],161 "percentile_in_level": round(self.percentile_in_level, 1),162 "verdict": self.verdict,163 }164 165 166@dataclass167class PracticeRecommendation:168 """A specific practice recommendation based on trend analysis."""169 170 area: str # "aim", "utility", "positioning", "economy", "trading", "entry"171 priority: str # "high", "medium", "low"172 description: str173 current_value: float174 target_value: float175 drill: str176 177 def to_dict(self) -> dict[str, Any]:178 """Serialize to dict."""179 return {180 "area": self.area,181 "priority": self.priority,182 "description": self.description,183 "current_value": round(self.current_value, 2),184 "target_value": round(self.target_value, 2),185 "drill": self.drill,186 }187 188 189@dataclass190class DevelopmentReport:191 """Comprehensive player development report across matches."""192 193 steam_id: str194 match_count: int195 date_range: tuple[str, str] | None = None # (earliest, latest) ISO dates196 current_snapshot: MatchSnapshot | None = None197 trends: list[TrendAnalysis] = field(default_factory=list)198 benchmarks: list[RoleBenchmark] = field(default_factory=list)199 recommendations: list[PracticeRecommendation] = field(default_factory=list)200 estimated_level: str = "intermediate"201 strengths: list[str] = field(default_factory=list)202 weaknesses: list[str] = field(default_factory=list)203 improvement_velocity: float = 0.0 # -1.0 to +1.0204 summary: str = ""205 206 def to_dict(self) -> dict[str, Any]:207 """Serialize to dict."""208 return {209 "steam_id": self.steam_id,210 "match_count": self.match_count,211 "date_range": list(self.date_range) if self.date_range else None,212 "current_snapshot": self.current_snapshot.to_dict() if self.current_snapshot else None,213 "trends": [t.to_dict() for t in self.trends],214 "benchmarks": [b.to_dict() for b in self.benchmarks],215 "recommendations": [r.to_dict() for r in self.recommendations],216 "estimated_level": self.estimated_level,217 "strengths": self.strengths,218 "weaknesses": self.weaknesses,219 "improvement_velocity": round(self.improvement_velocity, 2),220 "summary": self.summary,221 }222 223 224# =============================================================================225# Constants226# =============================================================================227 228# Metrics to track for trend analysis229# Maps: metric name -> (MatchSnapshot attr, higher_is_better)230TRACKED_METRICS: dict[str, tuple[str, bool]] = {231 "hltv_rating": ("hltv_rating", True),232 "adr": ("adr", True),233 "kast": ("kast", True),234 "hs_pct": ("hs_pct", True),235 "aim_rating": ("aim_rating", True),236 "utility_rating": ("utility_rating", True),237 "kills": ("kills", True),238 "deaths": ("deaths", False), # Lower is better239}240 241# History column names to track (for DB-based trend analysis)242TRACKED_HISTORY_METRICS: list[str] = [243 "kills",244 "deaths",245 "adr",246 "kast",247 "hs_pct",248 "hltv_rating",249 "aim_rating",250 "utility_rating",251 "trade_kill_success",252 "trade_kill_attempts",253 "entry_success",254 "entry_attempts",255 "clutch_wins",256 "clutch_situations",257 "he_damage",258 "enemies_flashed",259 "flash_assists",260 "ttd_median_ms",261 "cp_median_deg",262]263 264# Metrics where lower is better265LOWER_IS_BETTER = {"deaths", "ttd_median_ms", "cp_median_deg"}266 267# Threshold for trend detection (5% relative change)268TREND_THRESHOLD = 0.05269 270# Recent window size for trend comparison271RECENT_WINDOW = 5272 273# Minimum matches required for meaningful analysis274MIN_MATCHES = 3275 276# Competitive level benchmarks for CS2277# Based on industry data: ESEA ranks, FACEIT levels, community averages278LEVEL_BENCHMARKS: dict[str, dict[str, dict[str, float]]] = {279 "beginner": {280 "hltv_rating": {"low": 0.0, "avg": 0.75, "high": 0.90},281 "adr": {"low": 0.0, "avg": 55.0, "high": 70.0},282 "kast": {"low": 0.0, "avg": 55.0, "high": 65.0},283 "hs_pct": {"low": 0.0, "avg": 30.0, "high": 40.0},284 "entry_win_rate": {"low": 0.0, "avg": 35.0, "high": 45.0},285 "clutch_win_rate": {"low": 0.0, "avg": 10.0, "high": 20.0},286 "trade_rate": {"low": 0.0, "avg": 30.0, "high": 40.0},287 "utility_rating": {"low": 0.0, "avg": 25.0, "high": 40.0},288 },289 "intermediate": {290 "hltv_rating": {"low": 0.80, "avg": 1.00, "high": 1.15},291 "adr": {"low": 60.0, "avg": 75.0, "high": 85.0},292 "kast": {"low": 60.0, "avg": 68.0, "high": 75.0},293 "hs_pct": {"low": 35.0, "avg": 42.0, "high": 50.0},294 "entry_win_rate": {"low": 40.0, "avg": 48.0, "high": 55.0},295 "clutch_win_rate": {"low": 15.0, "avg": 22.0, "high": 30.0},296 "trade_rate": {"low": 35.0, "avg": 45.0, "high": 55.0},297 "utility_rating": {"low": 30.0, "avg": 45.0, "high": 55.0},298 },299 "advanced": {300 "hltv_rating": {"low": 1.00, "avg": 1.15, "high": 1.30},301 "adr": {"low": 75.0, "avg": 85.0, "high": 95.0},302 "kast": {"low": 68.0, "avg": 73.0, "high": 80.0},303 "hs_pct": {"low": 42.0, "avg": 48.0, "high": 55.0},304 "entry_win_rate": {"low": 48.0, "avg": 52.0, "high": 58.0},305 "clutch_win_rate": {"low": 20.0, "avg": 28.0, "high": 35.0},306 "trade_rate": {"low": 45.0, "avg": 52.0, "high": 60.0},307 "utility_rating": {"low": 45.0, "avg": 55.0, "high": 65.0},308 },309 "elite": {310 "hltv_rating": {"low": 1.15, "avg": 1.30, "high": 1.50},311 "adr": {"low": 85.0, "avg": 92.0, "high": 105.0},312 "kast": {"low": 73.0, "avg": 78.0, "high": 85.0},313 "hs_pct": {"low": 48.0, "avg": 52.0, "high": 60.0},314 "entry_win_rate": {"low": 52.0, "avg": 55.0, "high": 62.0},315 "clutch_win_rate": {"low": 25.0, "avg": 32.0, "high": 40.0},316 "trade_rate": {"low": 50.0, "avg": 58.0, "high": 65.0},317 "utility_rating": {"low": 55.0, "avg": 65.0, "high": 75.0},318 },319}320 321# Persona ID -> role mapping (for practice recommendations)322PERSONA_ROLE_MAP: dict[str, str] = {323 "the_opener": "entry_fragger",324 "the_headhunter": "entry_fragger",325 "the_anchor": "anchor",326 "the_survivor": "anchor",327 "the_utility_master": "support",328 "the_flash_master": "support",329 "the_cleanup": "trader",330 "the_terminator": "trader",331 "the_lurker": "lurker",332 "the_damage_dealer": "fragger",333 "the_competitor": "fragger",334}335 336# Role-specific practice targets (what a good player in this role should hit)337ROLE_PRACTICE_TARGETS: dict[str, dict[str, float]] = {338 "entry_fragger": {339 "hs_pct": 55.0,340 "entry_success_rate": 55.0,341 "adr": 80.0,342 "hltv_rating": 1.10,343 "ttd_median_ms": 300.0,344 "cp_median_deg": 10.0,345 },346 "anchor": {347 "kast": 75.0,348 "clutch_success_rate": 30.0,349 "adr": 70.0,350 "hltv_rating": 1.05,351 "ttd_median_ms": 350.0,352 "cp_median_deg": 12.0,353 },354 "support": {355 "flash_assists": 3.0,356 "enemies_flashed": 8.0,357 "he_damage": 30.0,358 "kast": 72.0,359 "hltv_rating": 1.0,360 "adr": 65.0,361 },362 "trader": {363 "trade_success_rate": 55.0,364 "adr": 75.0,365 "kast": 72.0,366 "hltv_rating": 1.05,367 "ttd_median_ms": 320.0,368 "cp_median_deg": 11.0,369 },370 "lurker": {371 "adr": 80.0,372 "deaths": 15.0, # lower is better373 "hltv_rating": 1.10,374 "kast": 72.0,375 "ttd_median_ms": 300.0,376 "cp_median_deg": 10.0,377 },378 "fragger": {379 "kills": 22.0,380 "adr": 85.0,381 "hltv_rating": 1.15,382 "hs_pct": 50.0,383 "ttd_median_ms": 300.0,384 "cp_median_deg": 10.0,385 },386}387 388 389# =============================================================================390# Player Tracker Engine391# =============================================================================392 393 394class PlayerTracker:395 """396 Cross-match player development tracker.397 398 Uses the existing DatabaseManager and MatchHistory table for storage.399 Provides snapshot extraction, trend analysis, level benchmarks,400 practice recommendations, and full development reports.401 """402 403 def __init__(self, db: Any | None = None) -> None:404 """Initialize with optional DatabaseManager instance.405 406 Args:407 db: DatabaseManager instance. If None, uses get_db().408 """409 self._db = db410 411 @property412 def db(self) -> Any:413 """Lazy-load database manager."""414 if self._db is None:415 from opensight.infra.database import get_db416 417 self._db = get_db()418 return self._db419 420 # =========================================================================421 # Snapshot Extraction422 # =========================================================================423 424 def extract_snapshot(425 self,426 orchestrator_result: dict[str, Any],427 steam_id: str,428 ) -> MatchSnapshot | None:429 """Extract a MatchSnapshot from an orchestrator result for a specific player.430 431 Args:432 orchestrator_result: Full orchestrator output dict433 steam_id: Player's Steam ID434 435 Returns:436 MatchSnapshot if player found, None otherwise437 """438 players = orchestrator_result.get("players", {})439 player = players.get(steam_id)440 if player is None:441 return None442 443 stats = player.get("stats", {})444 rating = player.get("rating", {})445 advanced = player.get("advanced", {})446 duels = player.get("duels", {})447 utility = player.get("utility", {})448 demo_info = orchestrator_result.get("demo_info", {})449 450 return MatchSnapshot(451 steam_id=steam_id,452 demo_hash=demo_info.get("demo_hash", ""),453 map_name=demo_info.get("map"),454 result=None, # Determined after score comparison455 kills=stats.get("kills", 0),456 deaths=stats.get("deaths", 0),457 assists=stats.get("assists", 0),458 adr=stats.get("adr", 0.0),459 kast=rating.get("kast_percentage", 0.0),460 hs_pct=stats.get("headshot_pct", 0.0),461 rounds_played=stats.get("rounds_played", 0),462 hltv_rating=rating.get("hltv_rating", 1.0),463 aim_rating=rating.get("aim_rating", 0.0),464 utility_rating=rating.get("utility_rating", 0.0),465 ttd_median_ms=advanced.get("ttd_median_ms"),466 cp_median_deg=advanced.get("cp_median_error_deg"),467 entry_attempts=duels.get("opening_kills", 0) + duels.get("opening_deaths", 0),468 entry_success=duels.get("opening_kills", 0),469 clutch_situations=duels.get("clutch_attempts", 0),470 clutch_wins=duels.get("clutch_wins", 0),471 trade_kill_success=duels.get("trade_kills", 0),472 trade_kill_attempts=duels.get("trade_kill_opportunities", 0),473 enemies_flashed=utility.get("enemies_flashed", 0),474 flash_assists=utility.get("flash_assists", 0),475 he_damage=utility.get("he_damage", 0),476 )477 478 def snapshot_from_history(self, history_row: dict[str, Any]) -> MatchSnapshot:479 """Convert a get_player_history_full() row to a MatchSnapshot.480 481 Args:482 history_row: Dict from DatabaseManager.get_player_history_full()483 484 Returns:485 MatchSnapshot486 """487 return MatchSnapshot(488 steam_id=history_row.get("steam_id", ""),489 demo_hash=history_row.get("demo_hash", ""),490 map_name=history_row.get("map_name"),491 result=history_row.get("result"),492 analyzed_at=history_row.get("analyzed_at"),493 kills=history_row.get("kills", 0),494 deaths=history_row.get("deaths", 0),495 assists=history_row.get("assists", 0),496 adr=history_row.get("adr", 0.0),497 kast=history_row.get("kast", 0.0),498 hs_pct=history_row.get("hs_pct", 0.0),499 rounds_played=history_row.get("rounds_played", 0),500 hltv_rating=history_row.get("hltv_rating", 1.0),501 aim_rating=history_row.get("aim_rating", 0.0),502 utility_rating=history_row.get("utility_rating", 0.0),503 ttd_median_ms=history_row.get("ttd_median_ms"),504 cp_median_deg=history_row.get("cp_median_deg"),505 entry_attempts=history_row.get("entry_attempts", 0),506 entry_success=history_row.get("entry_success", 0),507 clutch_situations=history_row.get("clutch_situations", 0),508 clutch_wins=history_row.get("clutch_wins", 0),509 trade_kill_success=history_row.get("trade_kill_success", 0),510 trade_kill_attempts=history_row.get("trade_kill_attempts", 0),511 enemies_flashed=history_row.get("enemies_flashed", 0),512 flash_assists=history_row.get("flash_assists", 0),513 he_damage=history_row.get("he_damage", 0),514 )515 516 # =========================================================================517 # Trend Analysis518 # =========================================================================519 520 def analyze_trends(521 self,522 snapshots: list[MatchSnapshot],523 current: MatchSnapshot | None = None,524 ) -> list[TrendAnalysis]:525 """Analyze metric trends across match snapshots.526 527 Compares current match performance against:528 - Recent window (last RECENT_WINDOW matches)529 - Historical average (all matches)530 531 Args:532 snapshots: List of MatchSnapshots ordered oldest-first533 current: Optional current match snapshot (if not already in list)534 535 Returns:536 List of TrendAnalysis for each tracked metric537 """538 if not snapshots:539 return []540 541 all_snapshots = list(snapshots)542 if current is not None:543 all_snapshots.append(current)544 545 if len(all_snapshots) < MIN_MATCHES:546 return []547 548 trends = []549 for metric_name, (attr_name, higher_is_better) in TRACKED_METRICS.items():550 values = []551 for s in all_snapshots:552 v = getattr(s, attr_name, None)553 if v is not None:554 values.append(float(v))555 556 if len(values) < MIN_MATCHES:557 continue558 559 current_value = values[-1]560 recent_values = values[-RECENT_WINDOW:]561 historical_values = values562 563 recent_avg = sum(recent_values) / len(recent_values)564 historical_avg = sum(historical_values) / len(historical_values)565 recent_std = statistics.stdev(recent_values) if len(recent_values) > 1 else 0.0566 historical_std = (567 statistics.stdev(historical_values) if len(historical_values) > 1 else 0.0568 )569 570 direction = _compute_trend_direction(recent_avg, historical_avg, higher_is_better)571 572 change_pct = 0.0573 if historical_avg != 0:574 change_pct = ((current_value - historical_avg) / abs(historical_avg)) * 100575 576 trends.append(577 TrendAnalysis(578 metric_name=metric_name,579 current_value=current_value,580 recent_avg=recent_avg,581 historical_avg=historical_avg,582 recent_std=recent_std,583 historical_std=historical_std,584 direction=direction,585 change_pct=change_pct,586 sample_count=len(historical_values),587 )588 )589 590 return trends591 592 def analyze_trends_from_db(593 self, steam_id: str, min_matches: int = MIN_MATCHES594 ) -> list[TrendAnalysis]:595 """Analyze trends directly from DB history (convenience method).596 597 Args:598 steam_id: Player's Steam ID599 min_matches: Minimum matches required600 601 Returns:602 List of TrendAnalysis603 """604 history_rows = self.db.get_player_history_full(steam_id, limit=30)605 if len(history_rows) < min_matches:606 return []607 608 # History is newest-first; reverse for chronological order609 snapshots = [self.snapshot_from_history(row) for row in reversed(history_rows)]610 return self.analyze_trends(snapshots)611 612 # =========================================================================613 # Level Estimation & Benchmarking614 # =========================================================================615 616 def estimate_level(self, snapshots: list[MatchSnapshot]) -> str:617 """Estimate a player's competitive level from their match history.618 619 Uses HLTV rating as the primary indicator:620 - <0.85: beginner621 - 0.85-1.05: intermediate622 - 1.05-1.20: advanced623 - >1.20: elite624 625 Args:626 snapshots: List of MatchSnapshots627 628 Returns:629 Level string: "beginner", "intermediate", "advanced", "elite"630 """631 if not snapshots:632 return "intermediate"633 634 ratings = [s.hltv_rating for s in snapshots if s.hltv_rating is not None]635 if not ratings:636 return "intermediate"637 638 avg_rating = sum(ratings) / len(ratings)639 640 if avg_rating < 0.85:641 return "beginner"642 elif avg_rating < 1.05:643 return "intermediate"644 elif avg_rating < 1.20:645 return "advanced"646 return "elite"647 648 def compute_benchmarks(649 self,650 snapshots: list[MatchSnapshot],651 level: str | None = None,652 ) -> list[RoleBenchmark]:653 """Compare player metrics against competitive level benchmarks.654 655 Args:656 snapshots: List of MatchSnapshots657 level: Competitive level to compare against.658 If None, auto-estimated from snapshots.659 660 Returns:661 List of RoleBenchmark comparisons662 """663 if not snapshots:664 return []665 666 if level is None:667 level = self.estimate_level(snapshots)668 669 if level not in LEVEL_BENCHMARKS:670 level = "intermediate"671 672 level_benchmarks = LEVEL_BENCHMARKS[level]673 benchmarks = []674 675 player_avgs = _calculate_player_averages(snapshots)676 677 for metric_name, bench in level_benchmarks.items():678 player_value = player_avgs.get(metric_name)679 if player_value is None:680 continue681 682 level_low = bench["low"]683 level_avg = bench["avg"]684 level_high = bench["high"]685 686 # Calculate percentile within level range687 level_range = level_high - level_low688 if level_range > 0:689 percentile = ((player_value - level_low) / level_range) * 100690 percentile = max(0.0, min(100.0, percentile))691 else:692 percentile = 50.0693 694 # Determine verdict695 if player_value >= level_avg * 1.05:696 verdict = "above"697 elif player_value <= level_avg * 0.95:698 verdict = "below"699 else:700 verdict = "at"701 702 benchmarks.append(703 RoleBenchmark(704 metric_name=metric_name,705 player_value=player_value,706 level=level,707 level_avg=level_avg,708 level_low=level_low,709 level_high=level_high,710 percentile_in_level=percentile,711 verdict=verdict,712 )713 )714 715 return benchmarks716 717 # =========================================================================718 # Practice Recommendations719 # =========================================================================720 721 def generate_recommendations(722 self,723 steam_id: str,724 trends: list[TrendAnalysis] | None = None,725 ) -> list[PracticeRecommendation]:726 """Generate practice recommendations based on trends and role.727 728 Args:729 steam_id: Player's Steam ID730 trends: Pre-computed trends (if None, computed from DB)731 732 Returns:733 Sorted list of PracticeRecommendation734 """735 if trends is None:736 trends = self.analyze_trends_from_db(steam_id)737 738 if not trends:739 return []740 741 role = self._get_player_role(steam_id)742 targets = ROLE_PRACTICE_TARGETS.get(role, ROLE_PRACTICE_TARGETS["fragger"])743 744 # Also need DB averages for rate-based metrics745 history = self.db.get_player_history_full(steam_id, limit=30)746 averages = _compute_history_averages(history) if history else {}747 748 trend_map = {t.metric_name: t for t in trends}749 recs: list[PracticeRecommendation] = []750 751 # Rule 1: HS% below target752 hs = trend_map.get("hs_pct")753 target_hs = targets.get("hs_pct", 50.0)754 if hs and hs.recent_avg < target_hs * 0.85:755 recs.append(756 PracticeRecommendation(757 area="aim",758 priority="medium",759 description="Headshot percentage below role target",760 current_value=hs.recent_avg,761 target_value=target_hs,762 drill="Practice headshot-only deathmatch to build muscle memory",763 )764 )765 766 # Rule 2: KAST declining767 kast = trend_map.get("kast")768 if kast and kast.direction == "declining":769 recs.append(770 PracticeRecommendation(771 area="positioning",772 priority="high",773 description="KAST is declining — less round impact",774 current_value=kast.recent_avg,775 target_value=targets.get("kast", 72.0),776 drill="Focus on staying alive and getting at least one contribution per round",777 )778 )779 780 # Rule 3: ADR declining781 adr = trend_map.get("adr")782 if adr and adr.direction == "declining":783 recs.append(784 PracticeRecommendation(785 area="aim",786 priority="high",787 description="Damage output is declining",788 current_value=adr.recent_avg,789 target_value=targets.get("adr", 80.0),790 drill="Be more aggressive in engagements, use utility to deal damage",791 )792 )793 794 # Rule 4: Deaths increasing795 deaths = trend_map.get("deaths")796 if deaths and deaths.direction == "declining": # "declining" = getting worse for deaths797 recs.append(798 PracticeRecommendation(799 area="positioning",800 priority="medium",801 description="Dying more often — improve survival discipline",802 current_value=deaths.recent_avg,803 target_value=targets.get("deaths", 15.0),804 drill="Focus on information gathering before peeking, use utility before engaging",805 )806 )807 808 # Rule 5: Entry success below target809 entry_rate = averages.get("entry_success_rate", 0)810 entry_target = targets.get("entry_success_rate", 55.0)811 if entry_rate > 0 and entry_rate < entry_target * 0.80:812 recs.append(813 PracticeRecommendation(814 area="entry",815 priority="medium",816 description="Entry success rate below role target",817 current_value=entry_rate,818 target_value=entry_target,819 drill="Practice entry routes with utility on specific maps",820 )821 )822 823 # Rule 6: Utility rating declining824 util = trend_map.get("utility_rating")825 if util and util.direction == "declining":826 recs.append(827 PracticeRecommendation(828 area="utility",829 priority="medium",830 description="Utility effectiveness is declining",831 current_value=util.recent_avg,832 target_value=targets.get("utility_rating", 50.0)833 if "utility_rating" in targets834 else 50.0,835 drill="Learn pop-flash lineups and HE/molotov spots for your most-played maps",836 )837 )838 839 # Sort by priority840 priority_order = {"high": 0, "medium": 1, "low": 2}841 recs.sort(key=lambda r: priority_order.get(r.priority, 3))842 843 return recs844 845 # =========================================================================846 # Development Report847 # =========================================================================848 849 def generate_report(850 self,851 steam_id: str,852 limit: int = 30,853 ) -> DevelopmentReport:854 """Generate a comprehensive development report for a player.855 856 Pulls match history from DB, analyzes trends, benchmarks against857 competitive levels, generates recommendations, and identifies858 strengths/weaknesses.859 860 Args:861 steam_id: Player's Steam ID (17 digits)862 limit: Max matches to analyze (default 30)863 864 Returns:865 DevelopmentReport866 """867 history_rows = self.db.get_player_history_full(steam_id, limit=limit)868 869 if not history_rows:870 return DevelopmentReport(steam_id=steam_id, match_count=0)871 872 # Convert to snapshots (history_rows are newest-first, reverse for oldest-first)873 snapshots = [self.snapshot_from_history(row) for row in reversed(history_rows)]874 875 current = snapshots[-1] if snapshots else None876 match_count = len(snapshots)877 878 # Date range879 latest = history_rows[0].get("analyzed_at", "unknown")880 earliest = history_rows[-1].get("analyzed_at", "unknown")881 date_range = (str(earliest), str(latest))882 883 # Analyze trends884 trends = self.analyze_trends(snapshots)885 886 # Estimate level and compute benchmarks887 level = self.estimate_level(snapshots)888 benchmarks = self.compute_benchmarks(snapshots, level)889 890 # Generate recommendations891 recommendations = self.generate_recommendations(steam_id, trends=trends)892 893 # Identify strengths and weaknesses894 strengths, weaknesses = _identify_strengths_weaknesses(benchmarks, trends)895 896 # Calculate improvement velocity897 improvement_velocity = _calculate_improvement_velocity(trends)898 899 # Build summary900 summary = _build_summary(trends, benchmarks, recommendations, match_count)901 902 return DevelopmentReport(903 steam_id=steam_id,904 match_count=match_count,905 date_range=date_range,906 current_snapshot=current,907 trends=trends,908 benchmarks=benchmarks,909 recommendations=recommendations,910 estimated_level=level,911 strengths=strengths,912 weaknesses=weaknesses,913 improvement_velocity=improvement_velocity,914 summary=summary,915 )916 917 # =========================================================================918 # Helpers919 # =========================================================================920 921 def _get_player_role(self, steam_id: str) -> str:922 """Get the player's role based on their persona."""923 try:924 persona = self.db.get_player_persona(steam_id)925 if persona:926 persona_id = persona.get("persona", "the_competitor")927 return PERSONA_ROLE_MAP.get(persona_id, "fragger")928 except Exception:929 logger.debug("Could not fetch persona for %s, defaulting to fragger", steam_id)930 return "fragger"931 932 933# =============================================================================934# Module-level Helpers935# =============================================================================936 937 938def _compute_trend_direction(939 recent_avg: float,940 historical_avg: float,941 higher_is_better: bool,942) -> str:943 """Determine trend direction by comparing recent vs historical average.944 945 Uses a 5% threshold relative to historical average to filter noise.946 """947 if historical_avg == 0:948 return "stable"949 950 pct_change = (recent_avg - historical_avg) / abs(historical_avg)951 952 if higher_is_better:953 if pct_change > TREND_THRESHOLD:954 return "improving"955 elif pct_change < -TREND_THRESHOLD:956 return "declining"957 else:958 # Lower is better (e.g., deaths)959 if pct_change < -TREND_THRESHOLD:960 return "improving"961 elif pct_change > TREND_THRESHOLD:962 return "declining"963 964 return "stable"965 966 967def _calculate_player_averages(968 snapshots: list[MatchSnapshot],969) -> dict[str, float]:970 """Calculate average metrics from snapshots for benchmarking."""971 if not snapshots:972 return {}973 974 n = len(snapshots)975 avgs: dict[str, float] = {}976 977 # Core averages978 avgs["hltv_rating"] = sum(s.hltv_rating for s in snapshots) / n979 avgs["adr"] = sum(s.adr for s in snapshots) / n980 avgs["kast"] = sum(s.kast for s in snapshots) / n981 avgs["hs_pct"] = sum(s.hs_pct for s in snapshots) / n982 983 # Utility rating (skip 0 values = missing data)984 util_values = [s.utility_rating for s in snapshots if s.utility_rating > 0]985 if util_values:986 avgs["utility_rating"] = sum(util_values) / len(util_values)987 988 # Entry win rate989 total_attempts = sum(s.entry_attempts for s in snapshots)990 total_success = sum(s.entry_success for s in snapshots)991 if total_attempts > 0:992 avgs["entry_win_rate"] = (total_success / total_attempts) * 100993 994 # Clutch win rate995 total_clutch_sit = sum(s.clutch_situations for s in snapshots)996 total_clutch_wins = sum(s.clutch_wins for s in snapshots)997 if total_clutch_sit > 0:998 avgs["clutch_win_rate"] = (total_clutch_wins / total_clutch_sit) * 100999 1000 # Trade rate1001 total_trade_attempts = sum(s.trade_kill_attempts for s in snapshots)1002 total_trade_success = sum(s.trade_kill_success for s in snapshots)1003 if total_trade_attempts > 0:1004 avgs["trade_rate"] = (total_trade_success / total_trade_attempts) * 1001005 1006 return avgs1007 1008 1009def _compute_history_averages(history: list[dict]) -> dict[str, float]:1010 """Compute metric averages from match history dicts, including derived rates."""1011 n = len(history) if len(history) > 0 else 11012 totals: dict[str, float] = {}1013 1014 for h in history:1015 for metric in TRACKED_HISTORY_METRICS:1016 _v = h.get(metric)1017 val = float(_v if _v is not None else 0)1018 totals[metric] = totals.get(metric, 0) + val1019 1020 averages = {m: round(totals.get(m, 0) / n, 2) for m in TRACKED_HISTORY_METRICS}1021 1022 # Derived rates1023 total_entry_attempts = totals.get("entry_attempts", 0)1024 if total_entry_attempts > 0:1025 averages["entry_success_rate"] = round(1026 (totals.get("entry_success", 0) / total_entry_attempts) * 100, 11027 )1028 else:1029 averages["entry_success_rate"] = 0.01030 1031 total_clutch_situations = totals.get("clutch_situations", 0)1032 if total_clutch_situations > 0:1033 averages["clutch_success_rate"] = round(1034 (totals.get("clutch_wins", 0) / total_clutch_situations) * 100, 11035 )1036 else:1037 averages["clutch_success_rate"] = 0.01038 1039 total_trade_attempts = totals.get("trade_kill_attempts", 0)1040 if total_trade_attempts > 0:1041 averages["trade_success_rate"] = round(1042 (totals.get("trade_kill_success", 0) / total_trade_attempts) * 100, 11043 )1044 else:1045 averages["trade_success_rate"] = 0.01046 1047 return averages1048 1049 1050def _identify_strengths_weaknesses(1051 benchmarks: list[RoleBenchmark],1052 trends: list[TrendAnalysis],1053) -> tuple[list[str], list[str]]:1054 """Identify player strengths and weaknesses from benchmarks and trends."""1055 strengths: list[str] = []1056 weaknesses: list[str] = []1057 1058 metric_labels = {1059 "hltv_rating": "HLTV Rating",1060 "adr": "ADR",1061 "kast": "KAST",1062 "hs_pct": "Headshot %",1063 "entry_win_rate": "Entry Duels",1064 "clutch_win_rate": "Clutch Play",1065 "trade_rate": "Trading",1066 "utility_rating": "Utility Usage",1067 "aim_rating": "Aim",1068 "kills": "Kill Count",1069 "deaths": "Survivability",1070 }1071 1072 for b in benchmarks:1073 label = metric_labels.get(b.metric_name, b.metric_name)1074 if b.verdict == "above":1075 strengths.append(1076 f"{label} above {b.level} average ({b.player_value:.1f} vs {b.level_avg:.1f})"1077 )1078 elif b.verdict == "below":1079 weaknesses.append(1080 f"{label} below {b.level} average ({b.player_value:.1f} vs {b.level_avg:.1f})"1081 )1082 1083 for t in trends:1084 label = metric_labels.get(t.metric_name, t.metric_name)1085 if t.direction == "improving" and abs(t.change_pct) > 5:1086 strengths.append(f"{label} trending up ({t.change_pct:+.1f}%)")1087 elif t.direction == "declining" and abs(t.change_pct) > 5:1088 weaknesses.append(f"{label} trending down ({t.change_pct:+.1f}%)")1089 1090 return strengths, weaknesses1091 1092 1093def _calculate_improvement_velocity(trends: list[TrendAnalysis]) -> float:1094 """Calculate overall improvement velocity from trends.1095 1096 Returns:1097 Float from -1.0 (all declining) to +1.0 (all improving)1098 """1099 if not trends:1100 return 0.01101 1102 direction_scores = []1103 for t in trends:1104 if t.direction == "improving":1105 direction_scores.append(1.0)1106 elif t.direction == "declining":1107 direction_scores.append(-1.0)1108 else:1109 direction_scores.append(0.0)1110 1111 return sum(direction_scores) / len(direction_scores)1112 1113 1114def _build_summary(1115 trends: list[TrendAnalysis],1116 benchmarks: list[RoleBenchmark],1117 recommendations: list[PracticeRecommendation],1118 match_count: int,1119) -> str:1120 """Build a human-readable summary string."""1121 improving = [t for t in trends if t.direction == "improving"]1122 declining = [t for t in trends if t.direction == "declining"]1123 1124 parts = [f"Based on {match_count} matches analyzed:"]1125 1126 if improving:1127 names = ", ".join(t.metric_name.replace("_", " ") for t in improving[:3])1128 parts.append(f"Improving in: {names}.")1129 1130 if declining:1131 names = ", ".join(t.metric_name.replace("_", " ") for t in declining[:3])1132 parts.append(f"Declining in: {names}.")1133 1134 if not improving and not declining:1135 parts.append("Performance is stable across all metrics.")1136 1137 weak = [b for b in benchmarks if b.verdict == "below"]1138 if weak:1139 names = ", ".join(b.metric_name.replace("_", " ") for b in weak[:3])1140 parts.append(f"Below benchmark in: {names}.")1141 1142 high_recs = [r for r in recommendations if r.priority == "high"]1143 if high_recs:1144 parts.append(f"{len(high_recs)} high-priority area(s) to focus on.")1145 1146 return " ".join(parts)1147 1148 1149# =============================================================================1150# Singleton Access1151# =============================================================================1152 1153_tracker: PlayerTracker | None = None1154 1155 1156def get_player_tracker() -> PlayerTracker:1157 """Get the singleton PlayerTracker instance."""1158 global _tracker1159 if _tracker is None:1160 _tracker = PlayerTracker()1161 return _tracker1162