Update src/tools.py

src/tools.py

@@ -1,24 +1,7 @@
 import pandas as pd
 from rapidfuzz import process, fuzz
-
-# Data Loading ---------------------------------------------------------------
-try:
-    knowledge_base = pd.read_csv("mcp_knowledge_base.csv")
-    knowledge_base_latest = pd.read_csv("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 = []
-# ---------------------------------------------------------------------------
-
+import os
+from pathlib import Path
 
 class PokemonAdvisorTools():
     """
@@ -26,21 +9,66 @@ class PokemonAdvisorTools():
     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
-
-    print(len(knowledge_base))
-    print(len(knowledge_base_latest))
+    def __init__(self, data_dir: str = None):
+        """
+        Initialize the tools with data loading.
+        
+        Args:
+            data_dir: Directory containing the CSV files. If None, uses current directory.
+        """
+        # Determine data directory
+        if data_dir is None:
+            data_dir = Path(__file__).parent  # Same directory as tools.py
+        else:
+            data_dir = Path(data_dir)
+        
+        # Construct file paths
+        kb_path = data_dir / "mcp_knowledge_base.csv"
+        kb_latest_path = data_dir / "mcp_knowledge_base_latest.csv"
+        
+        # Load data with error handling
+        try:
+            print(f"Loading data from: {data_dir}")
+            print(f"Looking for: {kb_path}")
+            print(f"Looking for: {kb_latest_path}")
+            
+            if not kb_path.exists():
+                raise FileNotFoundError(f"File not found: {kb_path}")
+            if not kb_latest_path.exists():
+                raise FileNotFoundError(f"File not found: {kb_latest_path}")
+            
+            self.knowledge_base = pd.read_csv(kb_path)
+            self.knowledge_base_latest = pd.read_csv(kb_latest_path)
+            
+            # Extract unique values
+            self.ALL_PROD_NAMES = self.knowledge_base['prod_name'].values
+            self.ALL_ARTIST_NAMES = self.knowledge_base['artist'].values
+            self.ALL_SET_NAMES = self.knowledge_base['set_name'].values
+            
+            print(f"✓ Successfully loaded {len(self.knowledge_base)} records from knowledge_base")
+            print(f"✓ Successfully loaded {len(self.knowledge_base_latest)} records from knowledge_base_latest")
+            print(f"✓ Found {len(self.ALL_SET_NAMES)} unique sets")
+            
+        except Exception as e:
+            print(f"❌ ERROR loading data: {e}")
+            print(f"Current working directory: {os.getcwd()}")
+            print(f"Files in data directory: {list(data_dir.glob('*.csv')) if data_dir.exists() else 'Directory not found'}")
+            
+            # Initialize empty DataFrames as fallback
+            self.knowledge_base = pd.DataFrame()
+            self.knowledge_base_latest = pd.DataFrame()
+            self.ALL_PROD_NAMES = []
+            self.ALL_ARTIST_NAMES = []
+            self.ALL_SET_NAMES = []
+            
+            raise RuntimeError(f"Failed to load Pokemon card data: {e}")
 
     def get_data_shape(self) -> int:
         """
         Retrieves the number of set names.
         Use this when user asks about the number of unique sets.
         Returns:
-            int: lenght of knowledge base
+            int: length of knowledge base
         """
         return len(self.ALL_SET_NAMES)
 
@@ -54,7 +82,8 @@ class PokemonAdvisorTools():
         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."}
+        if len(self.ALL_PROD_NAMES) == 0:
+            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]
@@ -74,7 +103,8 @@ class PokemonAdvisorTools():
             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."}
+        if len(self.ALL_PROD_NAMES) == 0:
+            return {"error": "Data not loaded."}
 
         match = process.extractOne(name_query, self.ALL_PROD_NAMES, scorer=fuzz.WRatio)
         if not match or match[1] < 70:
@@ -84,11 +114,10 @@ class PokemonAdvisorTools():
         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 {"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 
@@ -110,7 +139,7 @@ class PokemonAdvisorTools():
         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)
+        ).head(10)
         output_columns = [
             "prod_name",
             "used_price",
@@ -123,7 +152,7 @@ class PokemonAdvisorTools():
         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:
+    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).
         
@@ -140,8 +169,11 @@ class PokemonAdvisorTools():
         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 = self.knowledge_base_latest.sort_values(
+            by=f"{market_type}_trend_{interval}", 
+            ascending=(sort_by != "uptrend")
+        ).head(10)
+        output_columns = ["prod_name", "used_price", "graded_price", f"{market_type}_trend_{interval}"]
         market_move_data = market_move_data[output_columns]
         return market_move_data.to_dict(orient="records")
 
@@ -173,8 +205,8 @@ class PokemonAdvisorTools():
             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 "error" in card_info:
+            return card_info
 
         if interval not in [3, 6]:
             return {"error": f"Invalid interval requested: {interval}. Only 3 or 6 months are supported."}
@@ -198,13 +230,13 @@ class PokemonAdvisorTools():
             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 {"error": f"Volatility data missing for {card_name} at {interval} months."}
 
         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),
+            "used_volatility": used_volatility,
+            "graded_volatility": graded_volatility,
+            f"used_volatility_assessment_{interval}_months": self._calculate_risk_label(used_volatility, used_vol_low_threshold, used_vol_high_threshold),
+            f"graded_volatility_assessment_{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:
@@ -220,8 +252,9 @@ class PokemonAdvisorTools():
             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."}
+        if "error" in card_info:
+            return card_info
+            
         return {
             "used_price": card_info.get("used_price"),
             "used_return_3_months": card_info.get("used_return_3"),
@@ -255,7 +288,6 @@ class PokemonAdvisorTools():
             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'."}
 
@@ -266,11 +298,14 @@ class PokemonAdvisorTools():
         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'].
+            artist_name (str): The artist's name.
             
         Returns:
             dict: A list of cards by that artist, sorted by profitability.
         """
+        if len(self.ALL_ARTIST_NAMES) == 0:
+            return {"error": "Data not loaded."}
+            
         artist_match = process.extractOne(artist_name, self.ALL_ARTIST_NAMES, scorer=fuzz.WRatio)
 
         if not artist_match or artist_match[1] < 75:
@@ -289,7 +324,6 @@ class PokemonAdvisorTools():
             "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."}
 
@@ -306,23 +340,35 @@ class PokemonAdvisorTools():
         rather than specific cards.
         
         Args:
-            set_query (str): The name of the set (e.g., "Sun & Moon"). Fuzzy matched.
+            set_name (str): The name of the set (e.g., "Evolving Skies"). 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]
+        """
+        if len(self.ALL_SET_NAMES) == 0:
+            return {"error": "Data not loaded."}
+            
+        set_match = process.extractOne(set_name, self.ALL_SET_NAMES, scorer=fuzz.WRatio)
+        
+        if not set_match or set_match[1] < 70:
+            return {"error": f"Set '{set_name}' not found. Available sets: {list(self.ALL_SET_NAMES[:5])}"}
+        
+        set_name_match = set_match[0]
         set_card_data = self.knowledge_base_latest[self.knowledge_base_latest["set_name"] == set_name_match]
+        
+        if set_card_data.empty:
+            return {"error": f"No data found for set: {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("-", " "),
+            "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']
+            "chase_card_price": f"${chase_card_row['used_price']:.2f}"
         }