Spaces:
Running
Running
File size: 3,266 Bytes
1dece25 a0c219b 47d4895 a0c219b 47d4895 a0c219b 47d4895 a0c219b 1dece25 a0c219b a0a8d76 47d4895 1dece25 a0c219b 1dece25 a0c219b 1dece25 a0c219b 1dece25 a0c219b 1dece25 a0c219b 1dece25 a0c219b 1dece25 a0c219b 1dece25 a0c219b |
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 |
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)}"} |