Update crypto_tools.py

crypto_tools.py

@@ -1,130 +1,146 @@
-from langchain.tools import tool
-import requests
-import pandas as pd
-import numpy as np
+# crypto_tools.py - CrewAI Native Version
 import os
-from typing import Dict, Any
+import requests
+from crewai.tools import BaseTool
+from typing import Type
+from pydantic import BaseModel, Field
+
+class CryptoPriceInput(BaseModel):
+    """Input schema for CryptoPriceTool."""
+    crypto_name: str = Field(..., description="Name of the cryptocurrency to get price for")
 
-# Define tools as standalone functions, not methods
-@tool("Get Crypto Price")
-def get_crypto_price(crypto_name: str) -> str:
-    """Gets the current price of a cryptocurrency."""
-    try:
-        # Convert crypto name to CoinGecko ID format
-        crypto_id = crypto_name.lower().replace(' ', '-')
-        
-        url = "https://api.coingecko.com/api/v3/simple/price"
-        params = {
-            'ids': crypto_id,
-            'vs_currencies': 'usd',
-            'include_market_cap': 'true',
-            'include_24hr_change': 'true'
-        }
-        
-        response = requests.get(url, params=params, timeout=10)
-        response.raise_for_status()
-        
-        data = response.json()
-        
-        if crypto_id in data:
-            price = data[crypto_id]['usd']
-            change_24h = data[crypto_id].get('usd_24h_change', 0)
-            return f"${price:,.2f} (24h change: {change_24h:+.2f}%)"
-        else:
-            # Try common alternative names
-            alternatives = {
-                'btc': 'bitcoin',
-                'eth': 'ethereum',
-                'ada': 'cardano',
-                'dot': 'polkadot',
-                'sol': 'solana'
+class CryptoPriceTool(BaseTool):
+    name: str = "Get Crypto Price"
+    description: str = "Gets the current price of a cryptocurrency from CoinGecko API"
+    args_schema: Type[BaseModel] = CryptoPriceInput
+
+    def _run(self, crypto_name: str) -> str:
+        try:
+            crypto_id = crypto_name.lower().replace(' ', '-')
+            
+            url = "https://api.coingecko.com/api/v3/simple/price"
+            params = {
+                'ids': crypto_id,
+                'vs_currencies': 'usd',
+                'include_market_cap': 'true',
+                'include_24hr_change': 'true'
             }
             
-            if crypto_name.lower() in alternatives:
-                return get_crypto_price(alternatives[crypto_name.lower()])
+            response = requests.get(url, params=params, timeout=10)
+            response.raise_for_status()
             
-            return f"Cryptocurrency '{crypto_name}' not found. Try using full name like 'bitcoin' instead of 'btc'."
+            data = response.json()
             
-    except requests.exceptions.RequestException as e:
-        return f"Error fetching price data: {str(e)}"
-    except Exception as e:
-        return f"Unexpected error: {str(e)}"
+            if crypto_id in data:
+                price = data[crypto_id]['usd']
+                change_24h = data[crypto_id].get('usd_24h_change', 0)
+                return f"${price:,.2f} (24h change: {change_24h:+.2f}%)"
+            else:
+                # Try common alternatives
+                alternatives = {
+                    'btc': 'bitcoin',
+                    'eth': 'ethereum', 
+                    'ada': 'cardano',
+                    'dot': 'polkadot',
+                    'sol': 'solana'
+                }
+                
+                if crypto_name.lower() in alternatives:
+                    return self._run(alternatives[crypto_name.lower()])
+                
+                return f"Cryptocurrency '{crypto_name}' not found. Try using full name like 'bitcoin' instead of 'btc'."
+                
+        except Exception as e:
+            return f"Error fetching price data: {str(e)}"
 
-@tool("Get Market Cap")
-def get_market_cap(crypto_name: str) -> str:
-    """Gets the market capitalization of a cryptocurrency."""
-    try:
-        crypto_id = crypto_name.lower().replace(' ', '-')
-        
-        url = "https://api.coingecko.com/api/v3/simple/price"
-        params = {
-            'ids': crypto_id,
-            'vs_currencies': 'usd',
-            'include_market_cap': 'true'
-        }
-        
-        response = requests.get(url, params=params, timeout=10)
-        response.raise_for_status()
-        
-        data = response.json()
-        
-        if crypto_id in data:
-            market_cap = data[crypto_id].get('usd_market_cap', 0)
-            return f"${market_cap:,.0f}"
-        else:
-            return f"Market cap data not found for '{crypto_name}'"
-            
-    except Exception as e:
-        return f"Error fetching market cap: {str(e)}"
+class MarketCapInput(BaseModel):
+    """Input schema for MarketCapTool."""
+    crypto_name: str = Field(..., description="Name of the cryptocurrency to get market cap for")
+
+class MarketCapTool(BaseTool):
+    name: str = "Get Market Cap"
+    description: str = "Gets the market capitalization of a cryptocurrency"
+    args_schema: Type[BaseModel] = MarketCapInput
 
-@tool("Calculate RSI")
-def calculate_rsi(crypto_name: str) -> str:
-    """Calculates the RSI (Relative Strength Index) for a cryptocurrency."""
-    try:
-        # For demo purposes, we'll return a mock RSI
-        # In production, you'd fetch historical price data and calculate actual RSI
-        import random
-        mock_rsi = round(random.uniform(20, 80), 2)
-        
-        if mock_rsi > 70:
-            signal = "overbought"
-        elif mock_rsi < 30:
-            signal = "oversold"
-        else:
-            signal = "neutral"
+    def _run(self, crypto_name: str) -> str:
+        try:
+            crypto_id = crypto_name.lower().replace(' ', '-')
+            
+            url = "https://api.coingecko.com/api/v3/simple/price"
+            params = {
+                'ids': crypto_id,
+                'vs_currencies': 'usd',
+                'include_market_cap': 'true'
+            }
             
-        return f"RSI: {mock_rsi}, Signal: {signal}"
-        
-    except Exception as e:
-        return f"Error calculating RSI: {str(e)}"
+            response = requests.get(url, params=params, timeout=10)
+            response.raise_for_status()
+            
+            data = response.json()
+            
+            if crypto_id in data:
+                market_cap = data[crypto_id].get('usd_market_cap', 0)
+                return f"${market_cap:,.0f}"
+            else:
+                return f"Market cap data not found for '{crypto_name}'"
+                
+        except Exception as e:
+            return f"Error fetching market cap: {str(e)}"
+
+class RSIInput(BaseModel):
+    """Input schema for RSITool."""
+    crypto_name: str = Field(..., description="Name of the cryptocurrency to calculate RSI for")
+
+class RSITool(BaseTool):
+    name: str = "Calculate RSI"
+    description: str = "Calculates the RSI (Relative Strength Index) for a cryptocurrency"
+    args_schema: Type[BaseModel] = RSIInput
 
-@tool("Calculate Moving Average")
-def calculate_moving_average(crypto_name: str) -> str:
-    """Calculates the 7-day moving average for a cryptocurrency."""
-    try:
-        # Get current price first
-        price_result = get_crypto_price(crypto_name)
-        
-        # Extract price from result (simplified approach)
-        if "$" in price_result:
-            try:
-                current_price = float(price_result.split("$")[1].split(" ")[0].replace(",", ""))
-                # Mock 7-day MA (in production, calculate from historical data)
-                import random
-                ma_7 = current_price * random.uniform(0.95, 1.05)
-                return f"7-day MA: ${ma_7:,.2f}"
-            except:
-                return "Unable to calculate moving average"
-        else:
-            return "Unable to get price for moving average calculation"
+    def _run(self, crypto_name: str) -> str:
+        try:
+            # Mock RSI for demo (in production, you'd calculate from historical data)
+            import random
+            mock_rsi = round(random.uniform(20, 80), 2)
             
-    except Exception as e:
-        return f"Error calculating moving average: {str(e)}"
+            if mock_rsi > 70:
+                signal = "overbought"
+            elif mock_rsi < 30:
+                signal = "oversold"
+            else:
+                signal = "neutral"
+                
+            return f"RSI: {mock_rsi}, Signal: {signal}"
+            
+        except Exception as e:
+            return f"Error calculating RSI: {str(e)}"
+
+class MovingAverageInput(BaseModel):
+    """Input schema for MovingAverageTool."""
+    crypto_name: str = Field(..., description="Name of the cryptocurrency to calculate moving average for")
 
-class CryptoTools:
-    """Container class for crypto tools - for backward compatibility"""
-    def __init__(self):
-        self.get_crypto_price = get_crypto_price
-        self.get_market_cap = get_market_cap
-        self.calculate_rsi = calculate_rsi
-        self.calculate_moving_average = calculate_moving_average
+class MovingAverageTool(BaseTool):
+    name: str = "Calculate Moving Average"
+    description: str = "Calculates the 7-day moving average for a cryptocurrency"
+    args_schema: Type[BaseModel] = MovingAverageInput
+
+    def _run(self, crypto_name: str) -> str:
+        try:
+            # Get current price first
+            price_tool = CryptoPriceTool()
+            price_result = price_tool._run(crypto_name)
+            
+            # Extract price from result
+            if "$" in price_result:
+                try:
+                    current_price = float(price_result.split("$")[1].split(" ")[0].replace(",", ""))
+                    # Mock 7-day MA
+                    import random
+                    ma_7 = current_price * random.uniform(0.95, 1.05)
+                    return f"7-day MA: ${ma_7:,.2f}"
+                except:
+                    return "Unable to calculate moving average"
+            else:
+                return "Unable to get price for moving average calculation"
+                
+        except Exception as e:
+            return f"Error calculating moving average: {str(e)}"