Spaces:
Running
Running
File size: 5,548 Bytes
b43f281 7e56b00 b43f281 3c429c7 b43f281 7e56b00 b43f281 7e56b00 b43f281 7e56b00 b43f281 3c429c7 b43f281 b4a502e b43f281 3c429c7 b43f281 3c429c7 b43f281 7e56b00 b43f281 7e56b00 b43f281 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 |
# crypto_tools.py - CrewAI Native Version
import os
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")
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'
}
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 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)}"
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
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'
}
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
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)
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 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)}" |