Spaces:
Sleeping
Sleeping
| import requests | |
| import statistics | |
| from datetime import datetime | |
| from crewai.tools import BaseTool | |
| from pydantic import BaseModel, Field | |
| from typing import Type | |
| # ---------- Input Schema ---------- | |
| class HistoricalInput(BaseModel): | |
| symbol: str = Field(default="bitcoin", description="Cryptocurrency ID used by CoinGecko, e.g. 'bitcoin'") | |
| currency: str = Field(default="usd", description="Fiat currency, e.g., 'usd'") | |
| days: int = Field(default=30, description="Number of past days to retrieve (e.g., 30, 90, 365)") | |
| # ---------- Tool ---------- | |
| class HistoricalDataTool(BaseTool): | |
| name: str = "get_historical_data" | |
| description: str = ( | |
| "Fetches historical cryptocurrency market data from CoinGecko and computes " | |
| "trend, percent change, and volatility. Returns structured JSON." | |
| ) | |
| args_schema: Type[BaseModel] = HistoricalInput | |
| def _run(self, symbol: str = "bitcoin", currency: str = "usd", days: int = 30) -> dict: | |
| url = f"https://api.coingecko.com/api/v3/coins/{symbol}/market_chart" | |
| params = {"vs_currency": currency, "days": days} | |
| try: | |
| response = requests.get(url, params=params, timeout=10) | |
| response.raise_for_status() | |
| data = response.json() | |
| prices = data.get("prices", []) | |
| if not prices: | |
| return {"error": f"No historical data found for '{symbol}'."} | |
| # -------- Extract date + price history -------- | |
| history = [ | |
| { | |
| "date": datetime.utcfromtimestamp(p[0] / 1000).strftime("%Y-%m-%d"), | |
| "price": float(p[1]) | |
| } | |
| for p in prices | |
| ] | |
| if len(history) < 2: | |
| return {"error": "Insufficient historical data for volatility or trend analysis."} | |
| start_price = history[0]["price"] | |
| end_price = history[-1]["price"] | |
| # -------- Percent Change -------- | |
| pct_change = ((end_price - start_price) / start_price) * 100 | |
| # -------- Daily Returns + Volatility -------- | |
| daily_returns = [ | |
| (history[i + 1]["price"] - history[i]["price"]) / history[i]["price"] | |
| for i in range(len(history) - 1) | |
| ] | |
| volatility = ( | |
| statistics.stdev(daily_returns) * 100 | |
| if len(daily_returns) > 1 | |
| else 0 | |
| ) | |
| # -------- Trend Classification -------- | |
| if pct_change > 1.5: | |
| trend = "upward" | |
| elif pct_change < -1.5: | |
| trend = "downward" | |
| else: | |
| trend = "sideways" | |
| return { | |
| "symbol": symbol, | |
| "currency": currency, | |
| "days": days, | |
| "start_price": round(start_price, 2), | |
| "end_price": round(end_price, 2), | |
| "pct_change": round(pct_change, 2), | |
| "volatility_pct": round(volatility, 2), | |
| "trend": trend, | |
| "price_history": history, # optional but useful for advanced analytics | |
| } | |
| except Exception as e: | |
| return {"error": f"HistoricalDataTool failed: {str(e)}"} |