thenuke02/cs2-analyzer
0
1"""2Team Self-Review Module for CS2 Demo Analysis.3 4Analyzes your own team's demos to identify:5- Failed trades (teammate died, not traded within 5 seconds)6- Wasted utility (flashes/smokes with no impact)7- Economy mistakes (bad force buys, wrong saves)8- Positioning errors (crossfires, bad peeks)9- Communication failures (duplicate holds, gaps)10 11Generates brutally honest player report cards and practice priorities.12"""13 14import logging15from collections import Counter, defaultdict16from dataclasses import dataclass, field17 18logger = logging.getLogger(__name__)19 20# 67 Esports roster (from PROMPT.md)21TEAM_67_ROSTER = {22 "Luke": "IGL",23 "foe": "Entry",24 "kix": "Support",25 "dergs": "AWP",26 "tr1d": "Support/Lurk",27 "miasma": "Anchor",28}29 30 31@dataclass32class Mistake:33 """A detected mistake in a round."""34 35 round_number: int36 mistake_type: str # "failed_trade", "wasted_utility", "economy", "positioning"37 description: str38 players_involved: list[str]39 fix_suggestion: str40 severity: str = "medium" # "low", "medium", "high", "critical"41 42 43@dataclass44class PlayerReportCard:45 """Individual player performance report."""46 47 player_name: str48 role: str49 grade: str # A, B, C, D, F50 kills: int51 deaths: int52 adr: float53 rating: float54 strengths: list[str]55 weaknesses: list[str]56 focus_area: str # Primary improvement area57 58 59@dataclass60class SelfReviewReport:61 """Complete self-review report."""62 63 team_name: str64 map_name: str65 result: str # "win" or "loss"66 score: str # e.g., "13-16"67 mistakes: list[Mistake] = field(default_factory=list)68 report_cards: list[PlayerReportCard] = field(default_factory=list)69 practice_priorities: list[str] = field(default_factory=list)70 71 72class SelfReviewEngine:73 """74 Engine for analyzing your own team's demos and identifying mistakes.75 """76 77 def __init__(self, team_roster: dict[str, str] | None = None):78 """79 Initialize the self-review engine.80 81 Args:82 team_roster: Dict of player_name -> role (defaults to 67 Esports roster)83 """84 self.team_roster = team_roster or TEAM_67_ROSTER85 86 def analyze(87 self,88 match_data: dict,89 our_team: str | None = None,90 ) -> SelfReviewReport:91 """92 Analyze match data for team mistakes.93 94 Args:95 match_data: Parsed match data from CachedAnalyzer96 our_team: Name of our team in the demo (optional, auto-detects)97 98 Returns:99 SelfReviewReport with mistakes and report cards100 """101 match_info = match_data.get("demo_info", {})102 map_name = match_info.get("map", "unknown")103 round_timeline = match_data.get("round_timeline", [])104 players = match_data.get("players", {})105 106 # Auto-detect our team based on roster names107 if not our_team:108 our_team = self._detect_our_team(players)109 110 # Determine result111 ct_score = match_info.get("score_ct", 0)112 t_score = match_info.get("score_t", 0)113 score = f"{ct_score}-{t_score}"114 115 # Filter to our team's players116 our_players = self._get_our_players(players, our_team)117 118 # Detect mistakes119 mistakes = []120 mistakes.extend(self._detect_failed_trades(round_timeline, our_players))121 mistakes.extend(self._detect_wasted_utility(round_timeline, our_players))122 mistakes.extend(self._detect_economy_mistakes(round_timeline, our_players))123 mistakes.extend(self._detect_positioning_errors(round_timeline, our_players))124 125 # Sort by severity126 severity_order = {"critical": 0, "high": 1, "medium": 2, "low": 3}127 mistakes.sort(key=lambda m: severity_order.get(m.severity, 99))128 129 # Generate report cards130 report_cards = self._generate_report_cards(our_players, mistakes)131 132 # Generate practice priorities133 practice_priorities = self._generate_practice_priorities(mistakes, report_cards)134 135 # Determine result136 our_wins = sum(1 for r in round_timeline if r.get("winner") == "CT") # Simplified137 result = "win" if our_wins > len(round_timeline) / 2 else "loss"138 139 return SelfReviewReport(140 team_name=our_team or "Unknown",141 map_name=map_name,142 result=result,143 score=score,144 mistakes=mistakes,145 report_cards=report_cards,146 practice_priorities=practice_priorities,147 )148 149 def _detect_our_team(self, players: dict) -> str:150 """Detect our team based on roster names."""151 for _steam_id, player in players.items():152 name = player.get("name", "")153 if name in self.team_roster:154 return player.get("team", "Unknown")155 return "Unknown"156 157 def _get_our_players(self, players: dict, our_team: str) -> dict:158 """Filter to only our team's players."""159 return {160 sid: p161 for sid, p in players.items()162 if p.get("team", "").lower() == our_team.lower()163 or p.get("name", "") in self.team_roster164 }165 166 def _detect_failed_trades(self, round_timeline: list[dict], our_players: dict) -> list[Mistake]:167 """Detect rounds where teammates weren't traded."""168 mistakes = []169 our_names = {p.get("name", "") for p in our_players.values()}170 171 for round_data in round_timeline:172 round_num = round_data.get("round_num", 0)173 if round_num == 0:174 continue175 kills = round_data.get("kills") or []176 177 # Track our team's deaths as we iterate through kills in order178 team_dead_so_far: set[str] = set()179 first_team_death_seen = False180 181 for i, kill in enumerate(kills):182 victim = kill.get("victim", "")183 attacker = kill.get("killer", "")184 kill_time = kill.get("tick", 0)185 186 # Track our team deaths before any checks (order matters)187 is_our_death = victim in our_names188 if is_our_death:189 team_dead_so_far.add(victim)190 191 # Only care about our team's deaths192 if not is_our_death:193 continue194 195 # Skip suicides and world kills (no attacker to trade)196 if not attacker or attacker == victim:197 continue198 199 # Skip the first death on our team each round (entry frag —200 # dying on the opening duel is part of the role, not a201 # failed trade)202 if not first_team_death_seen:203 first_team_death_seen = True204 continue205 206 # Check if attacker was killed within 5 seconds (320 ticks at 64 tick)207 trade_window = 320208 was_traded = False209 210 for subsequent_kill in kills[i + 1 :]:211 if subsequent_kill.get("victim") == attacker:212 if subsequent_kill.get("tick", 0) - kill_time <= trade_window:213 was_traded = True214 break215 216 if was_traded:217 continue218 219 # Find teammates who were alive when victim died220 alive_teammates = [p for p in our_names if p not in team_dead_so_far]221 222 # Skip if no teammates alive to trade (last-alive / clutch situation)223 if not alive_teammates:224 continue225 226 nearby = alive_teammates[:2]227 228 mistakes.append(229 Mistake(230 round_number=round_num,231 mistake_type="failed_trade",232 description=f"{victim} died to {attacker} and wasn't traded",233 players_involved=nearby,234 fix_suggestion=f"Players {', '.join(nearby)} should have traded within 5 seconds",235 severity="high",236 )237 )238 239 return mistakes240 241 def _find_nearby_teammates(self, round_data: dict, victim: str, our_names: set) -> list[str]:242 """Find teammates who were alive when victim died (computed from kills list)."""243 kills = round_data.get("kills") or []244 245 # Find all players dead before or at the same time as victim246 dead_before_victim = set()247 for k in kills:248 dead_before_victim.add(k.get("victim", ""))249 if k.get("victim", "") == victim:250 break251 252 # Return alive teammates (not dead and not the victim)253 return [p for p in our_names if p not in dead_before_victim and p != victim][:2]254 255 def _detect_wasted_utility(256 self, round_timeline: list[dict], our_players: dict257 ) -> list[Mistake]:258 """Detect utility that had no impact."""259 mistakes = []260 our_names = {p.get("name", "") for p in our_players.values()}261 262 for round_data in round_timeline:263 round_num = round_data.get("round_num", 0)264 utility = round_data.get("utility") or []265 blinds = round_data.get("blinds") or []266 267 for util in utility:268 player = util.get("player", "")269 if player not in our_names:270 continue271 272 util_type = util.get("type", "")273 274 # Check if flash had effect275 if util_type == "flashbang":276 enemies_flashed = [277 b for b in blinds if b.get("player") == player and b.get("enemy", False)278 ]279 teammates_flashed = [280 b for b in blinds if b.get("player") == player and not b.get("enemy", False)281 ]282 283 if not enemies_flashed and teammates_flashed:284 mistakes.append(285 Mistake(286 round_number=round_num,287 mistake_type="wasted_utility",288 description=f"{player} threw a flash that only hit teammates",289 players_involved=[player],290 fix_suggestion="Practice flash lineups to avoid team flashes",291 severity="medium",292 )293 )294 295 return mistakes296 297 def _detect_economy_mistakes(298 self, round_timeline: list[dict], our_players: dict299 ) -> list[Mistake]:300 """Detect economy management mistakes."""301 mistakes = []302 303 for i, round_data in enumerate(round_timeline):304 round_num = round_data.get("round_num", 0)305 if round_num == 0:306 continue307 308 # Get next round if available309 next_round = round_timeline[i + 1] if i + 1 < len(round_timeline) else None310 311 round_type = round_data.get("round_type", "")312 winner = round_data.get("winner", "")313 lost = winner != "T" # Simplified assumption314 315 if round_type == "force" and lost and next_round:316 next_type = next_round.get("round_type", "")317 if next_type in ["eco", "semi_eco"]:318 # Force buy that lost AND ruined next round319 mistakes.append(320 Mistake(321 round_number=round_num,322 mistake_type="economy",323 description="Force buy lost and ruined next round's economy",324 players_involved=["IGL"],325 fix_suggestion="Consider saving to guarantee full buy next round",326 severity="high",327 )328 )329 330 return mistakes331 332 def _detect_positioning_errors(333 self, round_timeline: list[dict], our_players: dict334 ) -> list[Mistake]:335 """Detect positioning mistakes.336 337 Note: ``was_dry_peek`` is computed by the orchestrator based on338 whether friendly utility was used within 192 ticks of the kill339 event. If the field is absent we silently skip dry-peek340 detection rather than crashing.341 """342 our_names = {p.get("name", "") for p in our_players.values()}343 344 # Pre-check: if was_dry_peek is True on >80% of kills, the field345 # is unreliable (upstream detection broken) — skip entirely.346 all_kills = [k for r in round_timeline for k in (r.get("kills") or [])]347 dry_peek_count = sum(1 for k in all_kills if k.get("was_dry_peek") is True)348 if len(all_kills) > 10 and dry_peek_count / len(all_kills) > 0.8:349 logger.warning(350 "was_dry_peek=True on %d/%d kills (>80%%) — field unreliable, "351 "skipping positioning errors",352 dry_peek_count,353 len(all_kills),354 )355 return []356 357 mistakes = []358 for round_data in round_timeline:359 round_num = round_data.get("round_num", 0)360 kills = round_data.get("kills") or []361 round_positioning = 0362 363 for kill in kills:364 victim = kill.get("victim", "")365 if victim not in our_names:366 continue367 368 dry_peek = kill.get("was_dry_peek")369 if dry_peek is None:370 continue371 if dry_peek:372 # Cap at 2 per round — more than that is repetitive373 round_positioning += 1374 if round_positioning > 2:375 continue376 mistakes.append(377 Mistake(378 round_number=round_num,379 mistake_type="positioning",380 description=f"{victim} dry peeked without utility support",381 players_involved=[victim],382 fix_suggestion="Always peek with flash or wait for teammate utility",383 severity="medium",384 )385 )386 387 return mistakes388 389 def _generate_report_cards(390 self, our_players: dict, mistakes: list[Mistake]391 ) -> list[PlayerReportCard]:392 """Generate individual player report cards."""393 report_cards = []394 395 # Count mistakes per player396 player_mistakes: dict[str, list[Mistake]] = defaultdict(list)397 for mistake in mistakes:398 for player in mistake.players_involved:399 player_mistakes[player].append(mistake)400 401 for _steam_id, player in our_players.items():402 name = player.get("name", "Unknown")403 stats = player.get("stats", {})404 rating_data = player.get("rating", {})405 406 kills = stats.get("kills", 0)407 deaths = stats.get("deaths", 0)408 adr = stats.get("adr", 0)409 rating = rating_data.get("hltv_rating", 1.0)410 rounds_played = stats.get("rounds_played", 1) or 1411 412 my_mistakes = player_mistakes[name]413 grade = self._calculate_grade(rating, adr, len(my_mistakes), rounds_played)414 415 # --- Strengths (stat-based) ---416 strengths: list[str] = []417 if rating > 1.3:418 strengths.append("Star player (elite rating)")419 elif rating > 1.1:420 strengths.append("High impact (good rating)")421 if adr > 100:422 strengths.append("Dominant damage output")423 elif adr > 85:424 strengths.append("Consistent damage output")425 if kills > deaths * 1.5:426 strengths.append("Strong K/D ratio")427 kast = rating_data.get("kast_percentage", 0)428 if kast > 75:429 strengths.append(f"High KAST ({kast:.0f}%)")430 431 # --- Weaknesses (stat + mistake-based) ---432 weaknesses: list[str] = []433 if rating < 0.8:434 weaknesses.append("Low impact (poor rating)")435 elif rating < 0.9:436 weaknesses.append("Below average impact")437 if adr < 50:438 weaknesses.append("Very low damage output")439 elif adr < 65:440 weaknesses.append("Low damage output")441 if deaths > kills * 1.5:442 weaknesses.append("Dying too often")443 444 # Add per-type weakness labels for frequent mistakes445 type_counts = Counter(m.mistake_type for m in my_mistakes)446 for mtype, count in type_counts.most_common(2):447 if count >= 2:448 type_labels = {449 "failed_trade": f"Failed to trade {count} times",450 "wasted_utility": f"Wasted utility {count} times",451 "economy": f"Economy errors ({count})",452 "positioning": f"Positioning mistakes ({count})",453 }454 weaknesses.append(type_labels.get(mtype, f"{mtype} ({count})"))455 456 # --- Focus area: player's MOST COMMON mistake type ---457 # Sub-differentiate within failed_trade using player stats so458 # not everyone gets the same generic focus.459 focus_map = {460 "wasted_utility": "Utility effectiveness",461 "economy": "Economy decision-making",462 "positioning": "Peek discipline and angles",463 }464 if type_counts:465 most_common_type = type_counts.most_common(1)[0][0]466 if most_common_type == "failed_trade":467 if rating > 1.2:468 # Star player survives but teammates die untraded469 focus = "Trade execution — position closer for refrags"470 elif deaths > kills * 1.2:471 focus = "Peek discipline — take fewer isolated fights"472 else:473 focus = "Trade timing and positioning"474 else:475 focus = focus_map.get(most_common_type, "General improvement")476 elif rating < 0.9:477 focus = "Impact and damage output"478 else:479 focus = "Maintain current form"480 481 role = self.team_roster.get(name, "Unknown")482 483 report_cards.append(484 PlayerReportCard(485 player_name=name,486 role=role,487 grade=grade,488 kills=kills,489 deaths=deaths,490 adr=adr,491 rating=rating,492 strengths=strengths or ["Solid performance"],493 weaknesses=weaknesses or ["No major issues"],494 focus_area=focus,495 )496 )497 498 # Sort by rating (best first)499 report_cards.sort(key=lambda x: x.rating, reverse=True)500 return report_cards501 502 def _calculate_grade(503 self, rating: float, adr: float, mistake_count: int, rounds_played: int504 ) -> str:505 """Calculate a letter grade for a player.506 507 Normalizes mistake penalty by rounds played so a handful of508 mistakes across a long match doesn't tank an otherwise good grade.509 """510 score = 0511 512 # Rating contribution (0-40 points)513 if rating >= 1.3:514 score += 40515 elif rating >= 1.1:516 score += 30517 elif rating >= 0.9:518 score += 20519 elif rating >= 0.7:520 score += 10521 522 # ADR contribution (0-30 points)523 if adr >= 90:524 score += 30525 elif adr >= 75:526 score += 20527 elif adr >= 60:528 score += 10529 530 # Mistake penalty: normalize by rounds, cap at -30531 mistakes_per_round = mistake_count / max(1, rounds_played)532 penalty = min(30, int(mistakes_per_round * 40))533 score -= penalty534 535 score = max(0, score)536 537 if score >= 60:538 return "A"539 elif score >= 45:540 return "B"541 elif score >= 30:542 return "C"543 elif score >= 15:544 return "D"545 else:546 return "F"547 548 def _generate_practice_priorities(549 self, mistakes: list[Mistake], report_cards: list[PlayerReportCard]550 ) -> list[str]:551 """Generate prioritized practice recommendations."""552 priorities = []553 554 # Count mistake types555 type_counts = defaultdict(int)556 for mistake in mistakes:557 type_counts[mistake.mistake_type] += 1558 559 # Generate priorities based on most common mistakes560 if type_counts["failed_trade"] > 2:561 priorities.append(562 "Trade timing drills - Practice 2-man peek scenarios and refrag timing"563 )564 if type_counts["wasted_utility"] > 2:565 priorities.append(566 "Utility practice - Review flash lineups and practice timing with team pushes"567 )568 if type_counts["economy"] > 1:569 priorities.append(570 "Economy decisions - IGL should review force buy thresholds with team"571 )572 if type_counts["positioning"] > 2:573 priorities.append("Peek discipline - Practice waiting for utility before engaging")574 575 # Add generic if no specific issues576 if not priorities:577 priorities.append("Continue current practice routine - no major issues detected")578 579 return priorities580 581 def generate_review_report(582 self,583 match_data: dict,584 our_team: str | None = None,585 ) -> str:586 """587 Generate a full self-review report using Claude.588 589 Args:590 match_data: Parsed match data from CachedAnalyzer591 our_team: Name of our team in the demo592 593 Returns:594 Markdown-formatted self-review report595 """596 # Analyze the match597 analysis = self.analyze(match_data, our_team)598 599 # Build summary for Claude600 from opensight.ai.llm_client import get_tactical_ai_client601 from opensight.ai.tactical import SYSTEM_PROMPT_SELF_REVIEW602 603 ai = get_tactical_ai_client()604 605 # Augment match data with our analysis606 augmented_data = dict(match_data)607 augmented_data["self_review"] = {608 "team_name": analysis.team_name,609 "result": analysis.result,610 "score": analysis.score,611 "mistakes": [612 {613 "round": m.round_number,614 "type": m.mistake_type,615 "description": m.description,616 "severity": m.severity,617 "fix": m.fix_suggestion,618 }619 for m in analysis.mistakes[:10] # Top 10 mistakes620 ],621 "report_cards": [622 {623 "player": rc.player_name,624 "role": rc.role,625 "grade": rc.grade,626 "rating": rc.rating,627 "focus": rc.focus_area,628 }629 for rc in analysis.report_cards630 ],631 "practice_priorities": analysis.practice_priorities,632 }633 634 report = ai.analyze(635 match_data=augmented_data,636 analysis_type="self-review",637 focus=f"Team: {analysis.team_name}, Result: {analysis.result} ({analysis.score})",638 system_prompt=SYSTEM_PROMPT_SELF_REVIEW,639 )640 641 return report642 643 644# Singleton instance645_review_engine_instance: SelfReviewEngine | None = None646 647 648def get_self_review_engine(649 team_roster: dict[str, str] | None = None,650) -> SelfReviewEngine:651 """Get or create singleton SelfReviewEngine instance."""652 global _review_engine_instance653 if _review_engine_instance is None:654 _review_engine_instance = SelfReviewEngine(team_roster)655 return _review_engine_instance656