""" Synergy Builder - Analyzes multiple deck builds and creates optimized "best-of" deck. Takes multiple builds of the same configuration and identifies cards that appear frequently across builds, scoring them for synergy based on: - Frequency of appearance (higher = more consistent with strategy) - EDHREC rank (lower rank = more popular/powerful) - Theme tag matches (more matching tags = better fit) """ from __future__ import annotations from dataclasses import dataclass, field from typing import Any, Dict, List, Optional from collections import Counter from code.logging_util import get_logger from code.deck_builder import builder_utils as bu import pandas as pd import os logger = get_logger(__name__) @dataclass class ScoredCard: """A card with its synergy score and metadata.""" name: str frequency: float # 0.0-1.0, percentage of builds containing this card appearance_count: int # Number of builds this card appears in synergy_score: float # 0-100+ calculated score category: str # Card type category (Creature, Land, etc.) role: str = "" # Card role from tagging tags: List[str] = field(default_factory=list) # Theme tags edhrec_rank: Optional[int] = None # EDHREC rank if available count: int = 1 # Number of copies (usually 1 for Commander) type_line: str = "" # Full type line (e.g., "Creature — Rabbit Scout") @dataclass class CardPool: """Aggregated pool of cards from multiple builds.""" cards: Dict[str, ScoredCard] # card_name -> ScoredCard total_builds: int config: Dict[str, Any] # Original build configuration themes: List[str] # Theme tags from config def get_by_category(self, category: str) -> List[ScoredCard]: """Get all cards in a specific category.""" return [card for card in self.cards.values() if card.category == category] def get_top_cards(self, limit: int = 100) -> List[ScoredCard]: """Get top N cards by synergy score.""" return sorted(self.cards.values(), key=lambda c: c.synergy_score, reverse=True)[:limit] def get_high_frequency_cards(self, min_frequency: float = 0.8) -> List[ScoredCard]: """Get cards appearing in at least min_frequency of builds.""" return [card for card in self.cards.values() if card.frequency >= min_frequency] class SynergyAnalyzer: """Analyzes multiple builds and scores cards for synergy.""" # Scoring weights FREQUENCY_WEIGHT = 0.5 EDHREC_WEIGHT = 0.25 THEME_WEIGHT = 0.25 HIGH_FREQUENCY_BONUS = 1.1 # 10% bonus for cards in 80%+ builds def __init__(self): """Initialize synergy analyzer.""" self._type_line_cache: Dict[str, str] = {} def _load_type_lines(self) -> Dict[str, str]: """ Load card type lines from parquet for all cards. Returns: Dict mapping card name (lowercase) to type_line """ if self._type_line_cache: return self._type_line_cache try: parquet_path = os.path.join("card_files", "processed", "all_cards.parquet") if not os.path.exists(parquet_path): logger.warning(f"[Synergy] Card parquet not found at {parquet_path}") return {} df = pd.read_parquet(parquet_path) # Try 'type' first, then 'type_line' type_col = None if 'type' in df.columns: type_col = 'type' elif 'type_line' in df.columns: type_col = 'type_line' if not type_col or 'name' not in df.columns: logger.warning(f"[Synergy] Card parquet missing required columns. Available: {list(df.columns)}") return {} # Build mapping: lowercase name -> type_line for _, row in df.iterrows(): name = str(row.get('name', '')).strip() type_line = str(row.get(type_col, '')).strip() if name and type_line: self._type_line_cache[name.lower()] = type_line logger.info(f"[Synergy] Loaded type lines for {len(self._type_line_cache)} cards from parquet") return self._type_line_cache except Exception as e: logger.warning(f"[Synergy] Error loading type lines from parquet: {e}") return {} def analyze_builds(self, builds: List[Dict[str, Any]], config: Dict[str, Any]) -> CardPool: """ Aggregate all cards from builds and calculate appearance frequencies. Args: builds: List of build results from BuildCache config: Original deck configuration Returns: CardPool with all unique cards and their frequencies """ logger.info(f"[Synergy] Analyzing {len(builds)} builds for synergy") if not builds: raise ValueError("Cannot analyze synergy with no builds") total_builds = len(builds) themes = config.get("tags", []) # Load type lines from card CSV type_line_map = self._load_type_lines() # Count card appearances and cumulative counts across all builds card_appearances: Counter = Counter() # card_name -> number of builds containing it card_total_counts: Counter = Counter() # card_name -> sum of counts across all builds card_metadata: Dict[str, Dict[str, Any]] = {} for build in builds: result = build.get("result", {}) summary = result.get("summary", {}) if not isinstance(summary, dict): logger.warning("[Synergy] Build missing summary, skipping") continue type_breakdown = summary.get("type_breakdown", {}) if not isinstance(type_breakdown, dict): continue type_cards = type_breakdown.get("cards", {}) if not isinstance(type_cards, dict): continue # Collect unique cards from this build unique_cards_in_build = set() for category, card_list in type_cards.items(): if not isinstance(card_list, list): continue for card in card_list: if not isinstance(card, dict): continue card_name = card.get("name") if not card_name: continue card_count = card.get("count", 1) unique_cards_in_build.add(card_name) # Track cumulative count across all builds (for multi-copy cards like basics) card_total_counts[card_name] += card_count # Store metadata (first occurrence) if card_name not in card_metadata: # Get type_line from parquet, fallback to card data (which won't have it from summary) type_line = type_line_map.get(card_name.lower(), "") if not type_line: type_line = card.get("type", card.get("type_line", "")) # Debug: Log first few cards if len(card_metadata) < 3: logger.info(f"[Synergy Debug] Card: {card_name}, Type line: {type_line}, From map: {card_name.lower() in type_line_map}") card_metadata[card_name] = { "category": category, "role": card.get("role", ""), "tags": card.get("tags", []), "type_line": type_line } # Increment appearance count for each unique card in this build for card_name in unique_cards_in_build: card_appearances[card_name] += 1 # Create ScoredCard objects with frequencies and average counts scored_cards: Dict[str, ScoredCard] = {} for card_name, appearance_count in card_appearances.items(): frequency = appearance_count / total_builds metadata = card_metadata.get(card_name, {}) scored_card = ScoredCard( name=card_name, frequency=frequency, appearance_count=appearance_count, synergy_score=0.0, # Will be calculated next category=metadata.get("category", "Unknown"), role=metadata.get("role", ""), tags=metadata.get("tags", []), count=1, # Default to 1 copy per card in synergy deck (basics override this later) type_line=metadata.get("type_line", "") ) # Debug: Log first few scored cards if len(scored_cards) < 3: logger.info(f"[Synergy Debug] ScoredCard: {scored_card.name}, type_line='{scored_card.type_line}', count={scored_card.count}, in_map={card_name.lower() in type_line_map}") # Calculate synergy score scored_card.synergy_score = self.score_card(scored_card, themes) scored_cards[card_name] = scored_card logger.info(f"[Synergy] Analyzed {len(scored_cards)} unique cards from {total_builds} builds") return CardPool( cards=scored_cards, total_builds=total_builds, config=config, themes=themes ) def score_card(self, card: ScoredCard, themes: List[str]) -> float: """ Calculate synergy score for a card. Score = frequency_weight * frequency * 100 + edhrec_weight * (1 - rank/max_rank) * 100 + theme_weight * (matching_tags / total_tags) * 100 Args: card: ScoredCard to score themes: Theme tags from config Returns: Synergy score (0-100+) """ # Frequency component (0-100) frequency_score = card.frequency * 100 # EDHREC component (placeholder - would need EDHREC data) # For now, assume no EDHREC data available edhrec_score = 50.0 # Neutral score # Theme component (0-100) theme_score = 0.0 if themes and card.tags: theme_set = set(themes) card_tag_set = set(card.tags) matching_tags = len(theme_set & card_tag_set) theme_score = (matching_tags / len(themes)) * 100 if themes else 0.0 # Calculate weighted score score = ( self.FREQUENCY_WEIGHT * frequency_score + self.EDHREC_WEIGHT * edhrec_score + self.THEME_WEIGHT * theme_score ) # Bonus for high-frequency cards (appear in 80%+ builds) if card.frequency >= 0.8: score *= self.HIGH_FREQUENCY_BONUS return round(score, 2) class SynergyDeckBuilder: """Builds an optimized deck from a synergy-scored card pool.""" def __init__(self, analyzer: Optional[SynergyAnalyzer] = None): """ Initialize synergy deck builder. Args: analyzer: SynergyAnalyzer instance (creates new if None) """ self.analyzer = analyzer or SynergyAnalyzer() def _allocate_basic_lands( self, selected_cards: List[ScoredCard], by_category: Dict[str, List[ScoredCard]], pool: CardPool, ideals: Optional[Dict[str, int]] ) -> List[ScoredCard]: """ Allocate basic lands based on color identity and remaining land slots. Separates basic lands from nonbasics, then allocates basics based on: 1. Total lands target from ideals 2. Color identity from config 3. Current nonbasic land count Args: selected_cards: Currently selected cards (may include basics from pool) by_category: Cards grouped by category pool: Card pool with configuration ideals: Ideal card counts Returns: Updated list of selected cards with properly allocated basics """ if not ideals: return selected_cards # No ideals, keep as-is # Get basic land names basic_names = bu.basic_land_names() # Separate basics from nonbasics nonbasic_cards = [c for c in selected_cards if c.name not in basic_names] # Calculate how many basics we need # Note: For nonbasics, count=1 per card (singleton rule), so count == number of unique cards target_lands = ideals.get("lands", 35) nonbasic_lands = [c for c in nonbasic_cards if c.category == "Land"] current_nonbasic_count = len(nonbasic_lands) # If we have too many nonbasics, trim them if current_nonbasic_count > target_lands: logger.info(f"[Synergy] Too many nonbasics ({current_nonbasic_count}), trimming to {target_lands}") # Keep the highest scoring nonbasics sorted_nonbasic_lands = sorted(nonbasic_lands, key=lambda c: c.synergy_score, reverse=True) trimmed_nonbasic_lands = sorted_nonbasic_lands[:target_lands] # Update nonbasic_cards to exclude trimmed lands other_nonbasics = [c for c in nonbasic_cards if c.category != "Land"] nonbasic_cards = other_nonbasics + trimmed_nonbasic_lands return nonbasic_cards # No room for basics needed_basics = max(0, target_lands - current_nonbasic_count) if needed_basics == 0: logger.info("[Synergy] No basic lands needed (nonbasics exactly fill target)") return nonbasic_cards logger.info(f"[Synergy] Need {needed_basics} basics to fill {target_lands} land target (have {current_nonbasic_count} nonbasics)") # Get color identity from config color_identity = pool.config.get("colors", []) if not color_identity: logger.warning(f"[Synergy] No color identity in config (keys: {list(pool.config.keys())}), skipping basic land allocation") return nonbasic_cards # Map colors to basic land names from code.deck_builder import builder_constants as bc basic_map = getattr(bc, 'BASIC_LAND_MAPPING', { 'W': 'Plains', 'U': 'Island', 'B': 'Swamp', 'R': 'Mountain', 'G': 'Forest' }) # Allocate basics evenly across colors allocation: Dict[str, int] = {} colors = [c.upper() for c in color_identity if c.upper() in basic_map] if not colors: logger.warning(f"[Synergy] No valid colors found in identity: {color_identity}") return nonbasic_cards # Distribute basics evenly, with remainder going to first colors n = len(colors) base = needed_basics // n rem = needed_basics % n for idx, color in enumerate(sorted(colors)): # sorted for deterministic allocation count = base + (1 if idx < rem else 0) land_name = basic_map.get(color) if land_name: allocation[land_name] = count # Create ScoredCard objects for basics basic_cards = [] for land_name, count in allocation.items(): # Try to get type_line from cache first (most reliable) type_line = self.analyzer._type_line_cache.get(land_name.lower(), "") if not type_line: # Fallback: construct from land name type_line = f"Basic Land — {land_name[:-1] if land_name.endswith('s') else land_name}" # Try to get existing scored data from pool, else create minimal entry if land_name in pool.cards: existing = pool.cards[land_name] basic_card = ScoredCard( name=land_name, frequency=existing.frequency, appearance_count=existing.appearance_count, synergy_score=existing.synergy_score, category="Land", role="basic", tags=[], count=count, type_line=type_line # Use looked-up type_line ) else: # Not in pool (common for basics), create minimal entry basic_card = ScoredCard( name=land_name, frequency=1.0, # Assume high frequency for basics appearance_count=pool.total_builds, synergy_score=50.0, # Neutral score category="Land", role="basic", tags=[], count=count, type_line=type_line ) basic_cards.append(basic_card) # Update by_category to replace old basics with new allocation land_category = by_category.get("Land", []) land_category = [c for c in land_category if c.name not in basic_names] # Remove old basics land_category.extend(basic_cards) # Add new basics by_category["Land"] = land_category # Combine and return result = nonbasic_cards + basic_cards logger.info(f"[Synergy] Allocated {needed_basics} basic lands across {len(colors)} colors: {allocation}") return result def build_deck( self, pool: CardPool, ideals: Optional[Dict[str, int]] = None, target_size: int = 99 # Commander + 99 cards = 100 ) -> Dict[str, Any]: """ Build an optimized deck from the card pool, respecting ideal counts. Selects highest-scoring cards by category to meet ideal distributions. Args: pool: CardPool with scored cards ideals: Target card counts by category (e.g., {"Creature": 25, "Land": 35}) target_size: Total number of cards to include (default 99, excluding commander) Returns: Dict with deck list and metadata """ logger.info(f"[Synergy] Building deck from pool of {len(pool.cards)} cards") # Map category names to ideal keys (case-insensitive matching) category_mapping = { "Creature": "creatures", "Land": "lands", "Artifact": "artifacts", "Enchantment": "enchantments", "Instant": "instants", "Sorcery": "sorceries", "Planeswalker": "planeswalkers", "Battle": "battles" } selected_cards: List[ScoredCard] = [] by_category: Dict[str, List[ScoredCard]] = {} if ideals: # Build by category to meet ideals (±2 tolerance) logger.info(f"[Synergy] Using ideals: {ideals}") # Get basic land names for filtering basic_names = bu.basic_land_names() for category in ["Land", "Creature", "Artifact", "Enchantment", "Instant", "Sorcery", "Planeswalker", "Battle"]: ideal_key = category_mapping.get(category, category.lower()) target_count = ideals.get(ideal_key, 0) if target_count == 0: continue # Get all cards in this category sorted by score all_category_cards = pool.get_by_category(category) # For lands: only select nonbasics (basics allocated separately based on color identity) if category == "Land": # Filter out basics nonbasic_lands = [c for c in all_category_cards if c.name not in basic_names] category_cards = sorted( nonbasic_lands, key=lambda c: c.synergy_score, reverse=True ) # Reserve space for basics - typically want 15-20 basics minimum # So select fewer nonbasics to leave room min_basics_estimate = 15 # Reasonable minimum for most decks max_nonbasics = max(0, target_count - min_basics_estimate) selected = category_cards[:max_nonbasics] logger.info(f"[Synergy] Land: selected {len(selected)} nonbasics (max {max_nonbasics}, leaving room for basics)") else: category_cards = sorted( all_category_cards, key=lambda c: c.synergy_score, reverse=True ) # Select top cards up to target count selected = category_cards[:target_count] selected_cards.extend(selected) by_category[category] = selected logger.info( f"[Synergy] {category}: selected {len(selected)}/{target_count} " f"(pool had {len(category_cards)} available)" ) # Calculate how many basics we'll need before filling remaining slots target_lands = ideals.get("lands", 35) current_land_count = len(by_category.get("Land", [])) estimated_basics = max(0, target_lands - current_land_count) # Fill remaining slots with highest-scoring cards from any category (except Land) # But reserve space for basic lands that will be added later remaining_slots = target_size - len(selected_cards) - estimated_basics if remaining_slots > 0: selected_names = {c.name for c in selected_cards} # Exclude Land category from filler to avoid over-selecting lands remaining_pool = [ c for c in pool.get_top_cards(limit=len(pool.cards)) if c.name not in selected_names and c.category != "Land" ] filler_cards = remaining_pool[:remaining_slots] selected_cards.extend(filler_cards) # Add filler cards to by_category for card in filler_cards: by_category.setdefault(card.category, []).append(card) logger.info(f"[Synergy] Filled {len(filler_cards)} remaining slots (reserved {estimated_basics} for basics)") else: # No ideals provided - fall back to top-scoring cards logger.info("[Synergy] No ideals provided, selecting top-scoring cards") sorted_cards = pool.get_top_cards(limit=len(pool.cards)) selected_cards = sorted_cards[:target_size] # Group by category for summary for card in selected_cards: by_category.setdefault(card.category, []).append(card) # Add basic lands after nonbasics are selected selected_cards = self._allocate_basic_lands(selected_cards, by_category, pool, ideals) # Calculate stats (accounting for multi-copy cards) unique_cards = len(selected_cards) total_cards = sum(c.count for c in selected_cards) # Actual card count including duplicates # Debug: Check for cards with unexpected counts cards_with_count = [(c.name, c.count) for c in selected_cards if c.count != 1] if cards_with_count: logger.info(f"[Synergy Debug] Cards with count != 1: {cards_with_count[:10]}") avg_frequency = sum(c.frequency for c in selected_cards) / unique_cards if unique_cards else 0 avg_score = sum(c.synergy_score for c in selected_cards) / unique_cards if unique_cards else 0 high_freq_count = len([c for c in selected_cards if c.frequency >= 0.8]) logger.info( f"[Synergy] Built deck: {total_cards} cards ({unique_cards} unique), " f"avg frequency={avg_frequency:.2f}, avg score={avg_score:.2f}, " f"high-frequency cards={high_freq_count}" ) return { "cards": selected_cards, "by_category": by_category, "total_cards": total_cards, # Actual count including duplicates "unique_cards": unique_cards, # Unique card types "avg_frequency": round(avg_frequency, 3), "avg_score": round(avg_score, 2), "high_frequency_count": high_freq_count, "commander": pool.config.get("commander"), "themes": pool.themes } # Global analyzer instance _analyzer = SynergyAnalyzer() _builder = SynergyDeckBuilder(_analyzer) def analyze_and_build_synergy_deck( builds: List[Dict[str, Any]], config: Dict[str, Any] ) -> Dict[str, Any]: """ Convenience function to analyze builds and create synergy deck in one call. Args: builds: List of build results config: Original deck configuration (includes ideals) Returns: Synergy deck result dict """ pool = _analyzer.analyze_builds(builds, config) ideals = config.get("ideals", {}) deck = _builder.build_deck(pool, ideals=ideals) return deck