Update app.py

app.py

@@ -1,5 +1,5 @@
 import requests
-from typing import List, Dict
+from typing import List, Dict, Any
 import numpy as np
 import yaml
 from smolagents import CodeAgent, HfApiModel, load_tool, tool
@@ -10,7 +10,7 @@ from Gradio_UI import GradioUI
 # ✅ Final Answer Tool
 final_answer = FinalAnswerTool()
 
-# ✅ AI Model
+# ✅ AI Model Setup
 model = HfApiModel(
     max_tokens=2096,
     temperature=0.5,
@@ -28,9 +28,16 @@ with open("prompts.yaml", 'r') as stream:
 
 # --- Utility Functions ---
 
-def fetch_data(url, params=None):
+def fetch_data(url: str, params: Dict[str, Any] = None) -> Dict[str, Any]:
     """
     Utility function to make GET requests and return the response as JSON.
+
+    Arguments:
+        url (str): The API endpoint URL.
+        params (Dict[str, Any], optional): Query parameters for the request.
+
+    Returns:
+        Dict[str, Any]: The JSON response from the API.
     """
     try:
         response = requests.get(url, params=params)
@@ -40,9 +47,16 @@ def fetch_data(url, params=None):
         return {"error": str(e)}
 
 
-def compute_moving_average(prices, window):
+def compute_moving_average(prices: List[float], window: int) -> float:
     """
     Computes the moving average for the given price data over a specified window.
+
+    Arguments:
+        prices (List[float]): List of historical price values.
+        window (int): The number of data points to include in the average.
+
+    Returns:
+        float: The computed moving average.
     """
     return np.mean(prices[-window:]) if len(prices) >= window else np.mean(prices)
 
@@ -51,50 +65,65 @@ def compute_moving_average(prices, window):
 
 ### 🔹 1. Fetch Latest BTC News ###
 @tool
-def analyze_btc_news(articles: List[Dict[str, str]]) -> Dict[str, any]:
+def fetch_btc_news() -> List[Dict[str, str]]:
     """
-    Analyzes sentiment of Bitcoin news articles using TextBlob.
-    
-    Arguments:
-    articles (List[Dict[str, str]]): A list of dictionaries representing Bitcoin news articles.
-                                        Each dictionary contains the 'title' and 'url' of the article.
+    Fetches the latest Bitcoin news from the CryptoPanic API.
     
     Returns:
-    Dict[str, any]: A dictionary containing the sentiment score, sentiment label, and keywords.
+        List[Dict[str, str]]: A list of dictionaries, each containing the 'title' and 'url'
+                              of a Bitcoin news article.
     """
-    sentiments = [TextBlob(article["title"]).sentiment.polarity for article in articles]
-    avg_sentiment = np.mean(sentiments) if sentiments else 0
-    sentiment_label = "Positive 😊" if avg_sentiment > 0 else "Negative 😡" if avg_sentiment < 0 else "Neutral 😐"
-    keywords = list(set([word for article in articles for word in article["title"].split()[:5]]))
-    
-    return {
-        "sentiment_score": avg_sentiment, 
-        "sentiment_label": sentiment_label, 
-        "keywords": keywords
-    }
-    
+    url = "https://cryptopanic.com/api/v1/posts/?auth_token=ccbeafb47d266a78086f76c3f547546f6b747e88&currencies=BTC"
+    data = fetch_data(url)
+    return [{"title": post["title"], "url": post["url"]} for post in data.get("results", [])[:5]]
+
+
 ### 🔹 2. Analyze News Sentiment ###
 @tool
-def analyze_btc_news(articles: List[Dict[str, str]]) -> Dict[str, any]:
+def analyze_btc_news(articles: List[Dict[str, str]]) -> Dict[str, Any]:
     """
     Analyzes sentiment of Bitcoin news articles using TextBlob.
+    
+    Arguments:
+        articles (List[Dict[str, str]]): A list of dictionaries representing Bitcoin news articles.
+                                         Each dictionary should contain 'title' and 'url'.
+    
+    Returns:
+        Dict[str, Any]: A dictionary containing:
+                        - 'sentiment_score': The average sentiment polarity of the article titles.
+                        - 'sentiment_label': A label ('Positive 😊', 'Negative 😡', or 'Neutral 😐')
+                                             based on the sentiment score.
+                        - 'keywords': A list of keywords extracted from the article titles.
     """
     sentiments = [TextBlob(article["title"]).sentiment.polarity for article in articles]
     avg_sentiment = np.mean(sentiments) if sentiments else 0
     sentiment_label = "Positive 😊" if avg_sentiment > 0 else "Negative 😡" if avg_sentiment < 0 else "Neutral 😐"
-    keywords = list(set([word for article in articles for word in article["title"].split()[:5]]))
-    
+    keywords = list(set(word for article in articles for word in article["title"].split()[:5]))
     return {
-        "sentiment_score": avg_sentiment, 
-        "sentiment_label": sentiment_label, 
+        "sentiment_score": avg_sentiment,
+        "sentiment_label": sentiment_label,
         "keywords": keywords
     }
 
+
 ### 🔹 3. Get BTC Market Insights ###
 @tool
-def btc_market_insights() -> Dict[str, any]:
+def btc_market_insights() -> Dict[str, Any]:
     """
-    Fetch BTC price, historical trends, moving averages, and Fear & Greed Index.
+    Fetches BTC price, historical trends, moving averages, and the Fear & Greed Index.
+    
+    Returns:
+        Dict[str, Any]: A dictionary containing:
+                        - 'price': Current BTC price in USD.
+                        - '24h_change': 24-hour price change percentage.
+                        - 'ma_5m': 5-minute moving average.
+                        - 'ma_1h': 1-hour moving average.
+                        - 'ma_1d': 1-day moving average.
+                        - 'ma_7d': 7-day moving average.
+                        - 'ma_1m': 1-month moving average.
+                        - 'ma_1y': 1-year moving average.
+                        - 'fear_greed': Current Fear & Greed index value.
+                        - 'fgi_text': Classification of the Fear & Greed index.
     """
     # Fetch current BTC price
     price_url = "https://api.coingecko.com/api/v3/simple/price"
@@ -108,12 +137,12 @@ def btc_market_insights() -> Dict[str, any]:
     prices = [entry[1] for entry in hist_data.get("prices", [])]
 
     # Compute moving averages for different timeframes
-    ma_5m = compute_moving_average(prices, 1)  # 5-minute
+    ma_5m = compute_moving_average(prices, 1)   # 5-minute
     ma_1h = compute_moving_average(prices, 12)  # 1-hour
     ma_1d = compute_moving_average(prices, 24)  # 1-day
-    ma_7d = compute_moving_average(prices, 168)  # 7-day
-    ma_1m = compute_moving_average(prices, 720)  # 1-month
-    ma_1y = compute_moving_average(prices, 8760)  # 1-year
+    ma_7d = compute_moving_average(prices, 168) # 7-day
+    ma_1m = compute_moving_average(prices, 720) # 1-month
+    ma_1y = compute_moving_average(prices, 8760) # 1-year
 
     # Fetch Fear & Greed Index
     fgi_url = "https://api.alternative.me/fng/"
@@ -137,9 +166,16 @@ def btc_market_insights() -> Dict[str, any]:
 
 ### 🔹 4. Make Buy/Sell/HODL Decision ###
 @tool
-def btc_decision(news_analysis: Dict[str, any], market_insights: Dict[str, any]) -> Dict[str, str]:
+def btc_decision(news_analysis: Dict[str, Any], market_insights: Dict[str, Any]) -> Dict[str, str]:
     """
-    Aggregates news sentiment & market insights to suggest Buy, Sell, or HODL.
+    Aggregates news sentiment and market insights to suggest a Buy, Sell, or HODL decision.
+    
+    Arguments:
+        news_analysis (Dict[str, Any]): Output from the news sentiment analysis tool.
+        market_insights (Dict[str, Any]): BTC market data from the market insights tool.
+    
+    Returns:
+        Dict[str, str]: A dictionary mapping timeframes to a decision ("BUY 🚀", "SELL 📉", or "HODL 🤔").
     """
     sentiment_score = news_analysis["sentiment_score"]
     fgi = int(market_insights["fear_greed"])
@@ -153,7 +189,7 @@ def btc_decision(news_analysis: Dict[str, any], market_insights: Dict[str, any])
     ma_1m = market_insights["ma_1m"]
     ma_1y = market_insights["ma_1y"]
 
-    # Decision template
+    # Initialize decision template
     decision = {
         "5m-1H": "HODL 🤔", 
         "1D": "HODL 🤔", 
@@ -162,8 +198,8 @@ def btc_decision(news_analysis: Dict[str, any], market_insights: Dict[str, any])
         "1Y": "HODL 🤔"
     }
 
-    # Decision rules for each timeframe
-    def make_decision(timeframe, ma, price, sentiment, fear_greed):
+    # Internal helper to update decision based on parameters
+    def make_decision(timeframe: str, ma: float, price: float, sentiment: float, fear_greed: int) -> str:
         if price > ma and fear_greed < 40 and sentiment > 0:
             decision[timeframe] = "BUY 🚀"
         elif price < ma and fear_greed > 60 and sentiment < 0:
@@ -181,7 +217,7 @@ def btc_decision(news_analysis: Dict[str, any], market_insights: Dict[str, any])
 # --- Agent Setup ---
 agent = CodeAgent(
     model=model,
-    tools=[final_answer, fetch_btc_news, analyze_btc_news, btc_market_insights, btc_decision],  
+    tools=[final_answer, fetch_btc_news, analyze_btc_news, btc_market_insights, btc_decision],
     max_steps=6,
     verbosity_level=1,
     prompt_templates=prompt_templates