Create tools.py

src/tools.py

@@ -0,0 +1,316 @@
+import pandas as pd
+from rapidfuzz import process, fuzz
+
+# Data Loading ---------------------------------------------------------------
+try:
+    knowledge_base = pd.read_csv("data/mcp_knowledge_base.csv")
+    knowledge_base_latest = pd.read_csv("data/mcp_knowledge_base_latest.csv")
+    ALL_PROD_NAMES = knowledge_base.prod_name.values
+    ALL_ARTIST_NAMES = knowledge_base.artist.values
+    ALL_SET_NAMES = knowledge_base.set_name.values
+
+except Exception as e:
+    # Handle data loading error
+    print(f"ERROR loading data for tools: {e}")
+    knowledge_base = pd.DataFrame() 
+    knowledge_base_latest = pd.DataFrame()
+    ALL_PROD_NAMES = []
+    ALL_ARTIST_NAMES = []
+    ALL_SET_NAMES = []
+# ---------------------------------------------------------------------------
+
+
+class PokemonAdvisorTools():
+    """
+    A central class to house all data retrieval and analysis tools
+    for the cAsh MCP Robo-Advisor.
+    """
+
+    knowledge_base = knowledge_base
+    knowledge_base_latest = knowledge_base_latest
+    ALL_PROD_NAMES = ALL_PROD_NAMES
+    ALL_ARTIST_NAMES = ALL_ARTIST_NAMES
+    ALL_SET_NAMES = ALL_SET_NAMES
+
+    def list_card_names(self, name_query: str) -> list:
+        """
+        Retrieves a list of card names from the database.
+        Use this tool when the user says the card you provided is not what they are looking for.
+        Args:
+            name_query (str): The name of the card to search for (e.g., "Umbreon GX").
+                              The tool uses fuzzy matching, so exact spelling is not required.
+        Returns:
+            list: A list of 'prod_name's that matches the 'name_query'              
+        """
+        if not self.ALL_PROD_NAMES.any(): return {"error": "Data not loaded."}
+
+        prod_names_match = process.extract(name_query, self.ALL_PROD_NAMES, scorer=fuzz.WRatio, limit=5)
+        return [name[0].replace("_", " ") for name in prod_names_match]
+
+    def get_card_info(self, name_query: str) -> dict:
+        """
+        Retrieves comprehensive financial and metadata for a specific Pokemon card.
+        
+        Use this tool when you need to know the current price, 6-month trend, or
+        general details of a card.
+        
+        Args:
+            name_query (str): The name of the card to search for (e.g., "Charizard VMAX").
+                              The tool uses fuzzy matching, so exact spelling is not required.
+        
+        Returns:
+            dict: A dictionary containing 'used_price', 'graded_price', 'trend_6',
+                  and other key metrics. Returns an 'error' key if not found.
+        """
+        if not self.ALL_PROD_NAMES.any(): return {"error": "Data not loaded."}
+
+        match = process.extractOne(name_query, self.ALL_PROD_NAMES, scorer=fuzz.WRatio)
+        if not match or match[1] < 70:
+            return {"error": f"Card '{name_query}' not found. Please check spelling."}
+            
+        prod_name = match[0]
+        card_df = self.knowledge_base_latest[self.knowledge_base_latest["prod_name"] == prod_name]
+        
+        if card_df.empty:
+             return {"error": f"Data missing for '{prod_name}'."}
+             
+        return card_df.to_dict(orient="records")[0]
+
+
+    def find_grading_opportunities(self, max_price: float = 100, min_profit: float = 20) -> list:
+        """
+        Scans the market for 'Arbitrage' opportunities where the gap between the Raw 
+        and Graded price is largest.
+        
+        Use this tool when the user asks for "buying recommendations," "profitable cards,"
+        or "what should I grade?".
+        
+        Args:
+            max_price (float): The maximum price willing to pay for the raw card. Default is 100.
+            min_profit (float): The minimum profit (Graded Price - Raw Price - Fees) desired. Default is 20.
+            
+        Returns:
+            list: A list of dictionaries representing the top 10 most profitable opportunities,
+                  sorted by 'grade_profit' descending.
+        """        
+        profitable_grades = self.knowledge_base_latest[self.knowledge_base_latest["is_grade_profitable"] == True]
+        profitable_grades = profitable_grades[profitable_grades["used_price"] <= max_price]
+        min_profit_grades = profitable_grades[profitable_grades["grade_profit"] >= min_profit]
+        min_profit_grades = min_profit_grades.sort_values(
+            by="grade_profit", ascending=False
+            ).head(10)
+        output_columns = [
+            "prod_name",
+            "used_price",
+            "graded_price",
+            "grade_profit",
+            "grade_profit_ratio",
+            "is_popular_pokemon",
+            "artist"
+        ]
+        min_profit_grades = min_profit_grades[output_columns]
+        return min_profit_grades.to_dict(orient="records")
+
+    def get_market_movers(self, sort_by: str ="uptrend", interval: int = 6, market_type: str ="used") -> list:
+        """
+        Identifies cards with the strongest positive or negative price trends over a sustained period (3 or 6 months).
+        
+        Use this tool when users ask about "long-term growth," "steady winners," "market crashers," 
+        or "which cards are consistently losing value."
+        
+        NOTE: Use this for TRENDS. Use `get_recent_price_spikes` for sudden, short-term JUMPS.
+        
+        Args:
+            sort_by (str): "uptrend" to find biggest gainers, "downtrend" to find biggest losers. Default is "uptrend".
+            interval (int): The time period in months to analyze (3 or 6). Default is 6.
+            market_type (str): "used" (Raw) or "graded" (Slab). Default is "used".
+            
+        Returns:
+            list: A list of the top 10 cards matching the trend criteria, including their percentage change.
+        """        
+        market_move_data = self.knowledge_base_latest.sort(by=f"{market_type}_trend_{interval}", ascending=(not sort_by=="uptrend")).head(10)
+        output_columns = ["prod_name", "used_price", "graded_price"]
+        market_move_data = market_move_data[output_columns]
+        return market_move_data.to_dict(orient="records")
+
+    def _calculate_risk_label(self, vol, low_threshold, high_threshold):
+        """Helper function for volatility assessment tool."""
+        if vol < low_threshold:
+            return "🟢 Low Volatility (Stable/Blue Chip)"
+        elif vol > high_threshold:
+            return "🔴 High Volatility (Speculative)"
+        else:
+            return "🟡 Medium Volatility"
+
+    def assess_risk_volatility(self, card_name: str, interval: int = 6) -> dict:
+        """
+        Calculates the risk profile of a card based on its price volatility over time.
+        
+        ALWAYS use this tool before recommending an investment.
+        
+        Args:
+            card_name (str): The name of the card to analyze.
+            interval (int): The time period in months to analyze (must be 3 or 6). Default is 6.
+            
+        Returns:
+            dict: Contains 'volatility_assessment' (Low/Medium/High) and raw metrics.
+        """        
+        try:
+            interval = int(interval)
+        except ValueError:
+            return {"error": "Invalid 'interval' value. Must be 3 or 6."}
+
+        card_info = self.get_card_info(card_name)
+        if not card_info:
+            return {"error": f"Card not found for query: {card_name}"}
+
+        if interval not in [3, 6]:
+            return {"error": f"Invalid interval requested: {interval}. Only 3 or 6 months are supported."}
+
+        if interval == 3:
+            # 3-Month Thresholds
+            used_vol_low_threshold = 0.533
+            used_vol_high_threshold = 4.969
+            graded_vol_low_threshold = 0.982
+            graded_vol_high_threshold = 4.367
+            used_volatility = card_info.get("used_vol_3")
+            graded_volatility = card_info.get("graded_vol_3")
+
+        elif interval == 6:
+            # 6-Month Threshold
+            used_vol_low_threshold = 0.785
+            used_vol_high_threshold = 9.092
+            graded_vol_low_threshold = 2.250
+            graded_vol_high_threshold = 11.905
+            used_volatility = card_info.get("used_vol_6")
+            graded_volatility = card_info.get("graded_vol_6")
+
+        if used_volatility is None or graded_volatility is None:
+            return {"error": f"Volatility data missing for {card_name} at {interval} months. Check if card exists in the full knowledge base."}
+
+        return {
+            f"used_volatility": used_volatility,
+            f"graded_volatility": graded_volatility,
+            f"used_volatility_assesment_{interval}_months": self._calculate_risk_label(used_volatility, used_vol_low_threshold, used_vol_high_threshold),
+            f"graded_volatility_assesment_{interval}_months": self._calculate_risk_label(graded_volatility, graded_vol_low_threshold, graded_vol_high_threshold),
+        }
+
+    def get_roi_metrics(self, card_name: str) -> dict:
+        """
+        Retrieves the historical Return on Investment (ROI) percentages.
+        
+        Use this tool to show how a card has performed in the past (e.g., "Is it going up?").
+        
+        Args:
+            card_name (str): The name of the card.
+            
+        Returns:
+            dict: Returns 3-month and 6-month ROI percentages for both Used and Graded conditions.
+        """        
+        card_info = self.get_card_info(card_name)
+        if not card_info:
+            return {"error": f"Card not found for query: {card_name}. Cannot calculate ROI."}
+        return {
+            "used_price": card_info.get("used_price"),
+            "used_return_3_months": card_info.get("used_return_3"),
+            "used_return_6_months": card_info.get("used_return_6"),
+            "graded_return_3_months": card_info.get("graded_return_3"),
+            "graded_return_6_months": card_info.get("graded_return_6")
+        }
+
+    def get_recent_price_spikes(self, market_type: str = "used") -> list:
+        """
+        Identifies cards that have recently experienced a significant price jump ("Spike").
+        
+        Use this tool when users ask about "market movers," "hype," or "what is popping right now."
+        
+        Args:
+            market_type (str): Either "used" (Raw) or "graded" (Slab). Default is "used".
+            
+        Returns:
+            list: Top 20 cards with the highest recent positive price change.
+        """        
+        market_type = market_type.lower().strip()
+        if market_type == "used":
+            jump_data = self.knowledge_base_latest[self.knowledge_base_latest["used_jump_up"] == True]
+            jump_data = jump_data.sort_values("used_price", ascending=False).head(20)
+            output_columns = ["prod_name", "set_name", "used_price"]
+            return jump_data[output_columns].to_dict(orient="records")
+
+        elif market_type == "graded":
+            jump_data = self.knowledge_base_latest[self.knowledge_base_latest["graded_jump_up"] == True]
+            jump_data = jump_data.sort_values("graded_price", ascending=False).head(20)
+            output_columns = ["prod_name", "set_name", "graded_price"]
+            return jump_data[output_columns].to_dict(orient="records")
+
+        # --- Error Handling ---
+        else:
+            return {"error": f"Invalid market_type '{market_type}'. Please use 'used' or 'graded'."}
+
+    def find_cards_by_artist(self, artist_name: str) -> dict:
+        """
+        Finds profitable or popular cards illustrated by a specific artist.
+        
+        Use this for "Niche" requests or when users ask about art styles.
+        
+        Args:
+            artist_name (str): The artist's name limited to ['Akira Egawa', 'Shinji Kanda', 'HYOGONOSUKE', 'sowsow', 'Tomokazu Komiya'].
+            
+        Returns:
+            dict: A list of cards by that artist, sorted by profitability.
+        """
+        artist_match = process.extractOne(artist_name, self.ALL_ARTIST_NAMES, scorer=fuzz.WRatio)
+
+        if not artist_match or artist_match[1] < 75:
+            return {"error": f"Artist '{artist_name}' not found or matched with low confidence."}
+
+        artist_name_match = artist_match[0]
+        artist_card_data = self.knowledge_base_latest[self.knowledge_base_latest["artist"] == artist_name_match]
+        profitable_cards = artist_card_data[artist_card_data["is_grade_profitable"] == True]
+        profitable_cards = profitable_cards.sort_values(by="grade_profit", ascending=False).head(20)
+
+        output_columns = [
+            "prod_name",
+            "set_name",
+            "used_price",
+            "grade_profit",
+            "grade_profit_ratio"
+        ]
+
+        # --- Error Handling ---
+        if profitable_cards.empty:
+            return {"result": f"No currently profitable cards found by artist {artist_name_match} in the latest data."}
+
+        return {
+            "artist": artist_name_match,
+            "cards": profitable_cards[output_columns].to_dict(orient="records")
+        }
+
+    def analyze_set_performance(self, set_name: str) -> dict:
+        """
+        Aggregates data to analyze the overall health and sentiment of a specific Card Set.
+        
+        Use this when users ask about broad trends like "How is Evolving Skies doing?" 
+        rather than specific cards.
+        
+        Args:
+            set_query (str): The name of the set (e.g., "Sun & Moon"). Fuzzy matched.
+            
+        Returns:
+            dict: Average trends, average profitability, and the set's 'Chase Card'.
+        """        
+        set_name_match = process.extractOne(set_name.lower(), self.ALL_SET_NAMES, scorer=fuzz.WRatio)[0]
+        set_card_data = self.knowledge_base_latest[self.knowledge_base_latest["set_name"] == set_name_match]
+        total_cards = len(set_card_data)
+        avg_trend_6 = set_card_data["used_trend_6"].mean()
+        avg_grade_profit = set_card_data["grade_profit"].mean()
+        chase_card_row = set_card_data.sort_values('used_price', ascending=False).iloc[0]
+
+        return {
+            "set_name": set_name_match.replace("-", " "),
+            "total_cards_tracked": total_cards,
+            "market_sentiment_6mo": f"{avg_trend_6:.2f}%",
+            "avg_grading_profit": f"${avg_grade_profit:.2f}",
+            "chase_card": chase_card_row['prod_name'],
+            "chase_card_price": chase_card_row['used_price']
+        }