Add tools folder with market, sentiment, historical and analytical tools

tools/analytics_tool.py

@@ -0,0 +1,54 @@
+from crewai_tools import RagTool
+import json
+
+class AnalyticsTool(RagTool):
+    name: str = "analytics_tool"
+    description: str = (
+        "Processes and aggregates outputs from market, historical and sentiment agents"
+         "to generate structured indicators and performance metrics, giving a holistic view of the cryptocurrency's condition."
+    )
+
+    def _run(self, market_data: dict, historical_data: dict, sentiment_data: dict):
+        # combine results from the other agents into structured numeric metrics
+
+        try:
+            #Normalize strings -> dicts
+            if isinstance(market_data, str):
+                market_data = json.loads(market_data)
+            if isinstance(historical_data, str):
+                historical_data = json.loads(historical_data)
+            if isinstance(sentiment_data, str):
+                sentiment_data = json.loads(sentiment_data)
+
+            #Extract info safely
+            current_price = market_data.get("price") or market_data.get("latest_price")
+            pct_change = historical_data.get("pct_change", 0)
+            volatility = historical_data.get("volatility_pct", 0)
+            trend = historical_data.get("trend", "unknown")
+            sentiment = sentiment_data.get("final_sentiment_classification", "neutral")
+
+            #Compute basic consistency logic
+            aligned = (
+                (trend == "upward" and "bullish" in sentiment.lower()) or
+                (trend == "downward" and "bearish" in sentiment.lower())
+            )
+
+            score = (
+                (pct_change/10) + (0.2 if aligned else -0.2)
+                + (0.1 if "bullish" in sentiment.lower() else -0.1 if "bearish" in sentiment.lower() else 0)
+            )
+
+            score = round(max(-1, min(1, score)), 2)
+
+            return {
+                "price": current_price,
+                "pct_change": pct_change,
+                "volatility": volatility,
+                "trend": trend,
+                "sentiment": sentiment,
+                "alignment": "aligned" if aligned else "divergent",
+                "composite_score": score,
+                "summary": f"Trend={trend}, Sentiment={sentiment}, Alignment={'aligned' if aligned else 'divergent'}, Score = {score}"
+            }
+        except Exception as e:
+            return {"error": f"AnalyticsTool failed: {str(e)}"}

tools/historical_data_tool.py

@@ -0,0 +1,60 @@
+# tools/historical_data_tool.py
+from crewai_tools import RagTool
+import requests
+from datetime import datetime
+import statistics
+
+class HistoricalDataTool(RagTool):
+    name: str = "get_historical_data"
+    description: str = (
+        "Fetches and analyzes historical cryptocurrency market data "
+        "from the CoinGecko API for trend insights."
+    )
+
+    def _run(self, symbol: str = "bitcoin", currency: str = "usd", days: int = 30) -> dict:
+        """
+        Fetch historical data for the specified cryptocurrency over a given number of days.
+        Computes basic trend and volatility statistics.
+        """
+        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 data found for {symbol}."}
+
+            # Extract date and price
+            history = [
+                {"date": datetime.utcfromtimestamp(p[0] / 1000).strftime("%Y-%m-%d"), "price": p[1]}
+                for p in prices
+            ]
+
+            # Calculate simple stats
+            first, last = history[0]["price"], history[-1]["price"]
+            pct_change = ((last - first) / first) * 100
+            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 = "upward" if pct_change > 0 else "downward" if pct_change < 0 else "sideways"
+
+            return {
+                "symbol": symbol,
+                "currency": currency,
+                "days": days,
+                "start_price": round(first, 2),
+                "end_price": round(last, 2),
+                "pct_change": round(pct_change, 2),
+                "volatility_pct": round(volatility, 2),
+                "trend": trend,
+            }
+
+        except Exception as e:
+            return {"error": str(e)}

tools/market_data.py

@@ -0,0 +1,31 @@
+import os
+from crewai_tools import RagTool
+import requests
+
+class MarketDataTool(RagTool):
+    name: str ="get_market_data"
+    description: str ="Fetches the current market price of a cryptocurrency in a chosen fiat currency using the CoinGecko API."
+
+    def _run(self, symbol="bitcoin", currency="usd"):
+        url = f"https://api.coingecko.com/api/v3/simple/price?ids={symbol}&vs_currencies={currency}"
+        try:
+            response = requests.get(url, timeout=10)
+            response.raise_for_status()
+            data = response.json()
+            price = data.get(symbol, {}).get(currency)
+            if price is None:
+                return {
+                    "symbol": symbol,
+                    "currency": currency,
+                    "error": f"Could not find data for {symbol.upper()} in {currency.upper()}."
+                }
+            return {
+                "symbol": symbol,
+                "currency": currency,
+                "latest_price": price,
+                "formatted": f"{symbol.capitalize()} price: {price} {currency.upper()}"
+            }
+        except Exception as e:
+            return {"error": str(e)}
+
+

tools/sentiment_tool.py

@@ -0,0 +1,65 @@
+import os
+import requests
+from crewai_tools import RagTool
+from openai import OpenAI
+from pathlib import Path
+from dotenv import load_dotenv
+
+env_path = Path("/Users/clydecossey/Documents/Python/Crypto_Analysis_Agent/Open_AI.env")
+load_dotenv(dotenv_path=env_path)
+OPENAI_API_KEY = os.getenv(key = "OPENAI_API_KEY")
+
+client = OpenAI(api_key=OPENAI_API_KEY)
+
+serper_env_path = Path("Users/clydecossey/Documents/Python/Crypto_Analysis_Agent/Serper.env")
+load_dotenv(dotenv_path=serper_env_path)
+SERPER_API_KEY = os.getenv(key="SERPER_API_KEY")
+
+class SentimentTool(RagTool):
+    name: str = "get_crypto_sentiment"
+    description: str = (
+        "Fetches recent cryptocurrency news and Reddit discussions,"
+        "then analyses sentiment as bullish, bearish or neutral."
+    )
+
+    def _run(self, query: str = "bitcoin") -> str:
+        try:
+            #Google news sentiment
+            news_url = "https://google.serper.dev/news"
+            headers = {"X-API-KEY": SERPER_API_KEY, "Content-Type": "application/json"}
+            payload = {"q": f"{query} crypto", "num": 5}
+            news_response = requests.post(news_url, headers=headers, json=payload, timout=10)
+            news = news_response.json().get("news", [])
+            headlines = [n["title"] for n in news[:5]]
+
+            #Reddit sentiment
+            reddit_url = f"https://www.reddit.com/search.json?q={query}&sort=new&limit=5"
+            reddit_headers = {"User-Agent": "CryptoAgent/1.0"}
+            reddit_response = requests.get(reddit_url,headers=reddit_headers, timeout=10)
+            reddit_posts = reddit_response.json().get("data", {}).get("children", [])
+            reddit_titles = [p["data"]["title"] for p in reddit_posts[:5]]
+
+            combined_text = "News: " + " | ".join(headlines) + "\nReddit: " + " | ".join(reddit_titles)
+
+            #analyze sentiment
+            sentiment_prompt = f"""
+You are a crypto sentiment analyst. Based on the following headlines and Reddit posts,
+ intuitively determine if the overall sentiment towards "{query}" is bullish, bearish, or neutral.
+ Respond with a short summary and a one-word sentiment classification.
+
+ {combined_text}
+ """
+            completion = client.chat.completions.create(
+                model="gpt-5-mini",
+                messages=[
+                    {"role": "system", "content": "You are a precise sentiment classifier."},
+                    {"role": "user", "content": sentiment_prompt}
+                ],
+                temperature=0.2
+            )
+            sentiment_result = completion.choices[0].message.content.strip()
+            return sentiment_result
+        
+        except Exception as e:
+            return f"Error fetching sentiment data: {str(e)}"
+