Update app.py

app.py

@@ -17,9 +17,6 @@ model = HfApiModel(
     custom_role_conversions=None,
 )
 
-# ✅ Import Tool from Hub
-image_generation_tool = load_tool("agents-course/text-to-image", trust_remote_code=True)
-
 # ✅ Load Prompt Templates
 with open("prompts.yaml", 'r') as stream:
     prompt_templates = yaml.safe_load(stream)
@@ -28,15 +25,9 @@ with open("prompts.yaml", 'r') as stream:
 # --- Utility Functions ---
 
 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.
-
-    :param url: The API endpoint URL.
-    :param params: Query parameters for the request.
-    :return: The JSON response from the API.
-    """
+    """Make GET requests and return JSON response."""
     try:
-        response = requests.get(url, params=params)
+        response = requests.get(url, params=params, timeout=10)
         response.raise_for_status()
         return response.json()
     except requests.exceptions.RequestException as e:
@@ -44,142 +35,90 @@ def fetch_data(url: str, params: Dict[str, Any] = None) -> Dict[str, Any]:
 
 
 def compute_moving_average(prices: List[float], window: int) -> float:
-    """
-    Computes the moving average for the given price data over a specified window.
-
-    :param prices: List of historical price values.
-    :param window: The number of data points to include in the average.
-    :return: The computed moving average.
-    """
+    """Compute moving average for given price data."""
     return np.mean(prices[-window:]) if len(prices) >= window else np.mean(prices)
 
 
 # --- Tools ---
 
-### 🔹 1. Fetch Latest BTC News ###
 @tool
 def fetch_btc_news() -> List[Dict[str, str]]:
-    """
-    Fetches the latest Bitcoin news from the CryptoPanic API.
-
-    :return: A list of dictionaries, each containing the 'title' and 'url'
-             of a Bitcoin news article.
-    """
+    """Fetch latest Bitcoin news from CryptoPanic API."""
     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. Get BTC Market Insights ###
 @tool
 def btc_market_insights() -> Dict[str, Any]:
-    """
-    Fetches BTC price, historical trends, moving averages, and the Fear & Greed Index.
-
-    :return: 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"
-    price_data = fetch_data(price_url, {"ids": "bitcoin", "vs_currencies": "usd", "include_24hr_change": "true"})
-    btc_price = price_data["bitcoin"]["usd"]
-    btc_change = price_data["bitcoin"]["usd_24h_change"]
-
-    # Fetch historical price data (1 year)
-    hist_url = "https://api.coingecko.com/api/v3/coins/bitcoin/market_chart"
-    hist_data = fetch_data(hist_url, {"vs_currency": "usd", "days": "365", "interval": "minute"})
-    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_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
-
-    # Fetch Fear & Greed Index
-    fgi_url = "https://api.alternative.me/fng/"
-    fgi_data = fetch_data(fgi_url)
-    fear_greed = fgi_data["data"][0]["value"]
-    fgi_text = fgi_data["data"][0]["value_classification"]
+    """Fetch BTC market data and technical indicators."""
+    # Current price data
+    price_data = fetch_data(
+        "https://api.coingecko.com/api/v3/simple/price",
+        {"ids": "bitcoin", "vs_currencies": "usd", "include_24hr_change": "true"}
+    )
+    if "error" in price_data:
+        return price_data
+
+    # Historical data (90 days with hourly intervals)
+    hist_data = fetch_data(
+        "https://api.coingecko.com/api/v3/coins/bitcoin/market_chart",
+        {"vs_currency": "usd", "days": "90", "interval": "hourly"}
+    )
+    if "error" in hist_data:
+        return hist_data
 
+    prices = [entry[1] for entry in hist_data.get("prices", [])]
+    
+    # Moving averages calculations (hours)
     return {
-        "price": btc_price,
-        "24h_change": btc_change,
-        "ma_5m": ma_5m,
-        "ma_1h": ma_1h,
-        "ma_1d": ma_1d,
-        "ma_7d": ma_7d,
-        "ma_1m": ma_1m,
-        "ma_1y": ma_1y,
-        "fear_greed": fear_greed,
-        "fgi_text": fgi_text
+        "price": price_data["bitcoin"]["usd"],
+        "24h_change": price_data["bitcoin"]["usd_24h_change"],
+        "ma_4h": compute_moving_average(prices, 4),
+        "ma_12h": compute_moving_average(prices, 12),
+        "ma_24h": compute_moving_average(prices, 24),
+        "ma_7d": compute_moving_average(prices, 168),
+        "ma_30d": compute_moving_average(prices, 720),
+        **get_fear_greed_index()
     }
 
 
-### 🔹 3. Make Buy/Sell/HODL Decision ###
 @tool
 def btc_decision(market_insights: Dict[str, Any]) -> Dict[str, str]:
-    """
-    Aggregates market insights to suggest a Buy, Sell, or HODL decision.
-
-    :param market_insights: A dictionary containing BTC market data. It must include:
-         - '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 text for the Fear & Greed index.
-    :return: A dictionary mapping timeframes (e.g., "5m-1H", "1D", "7D", "1M", "1Y") to a decision 
-             ("BUY 🚀", "SELL 📉", or "HODL 🤔").
-    """
-    btc_price = market_insights["price"]
-    fgi = int(market_insights["fear_greed"])
-
-    # Moving averages for different timeframes
-    ma_5m = market_insights["ma_5m"]
-    ma_1h = market_insights["ma_1h"]
-    ma_1d = market_insights["ma_1d"]
-    ma_7d = market_insights["ma_7d"]
-    ma_1m = market_insights["ma_1m"]
-    ma_1y = market_insights["ma_1y"]
-
-    # Initialize decision template
-    decision = {
-        "5m-1H": "HODL 🤔",
-        "1D": "HODL 🤔",
-        "7D": "HODL 🤔",
-        "1M": "HODL 🤔",
-        "1Y": "HODL 🤔"
+    """Generate trading decisions based on market insights."""
+    decisions = {}
+    price = market_insights["price"]
+    fgi = int(market_insights.get("fear_greed", 50))
+    
+    timeframes = {
+        "4H": market_insights["ma_4h"],
+        "12H": market_insights["ma_12h"],
+        "24H": market_insights["ma_24h"],
+        "7D": market_insights["ma_7d"],
+        "30D": market_insights["ma_30d"]
     }
 
-    def make_decision(timeframe: str, ma: float) -> str:
-        if btc_price > ma and fgi < 40:
-            decision[timeframe] = "BUY 🚀"
-        elif btc_price < ma and fgi > 60:
-            decision[timeframe] = "SELL 📉"
-        return decision[timeframe]
-
-    decision["5m-1H"] = make_decision("5m-1H", ma_5m)
-    decision["1D"] = make_decision("1D", ma_1d)
-    decision["7D"] = make_decision("7D", ma_7d)
-    decision["1M"] = make_decision("1M", ma_1m)
-
-    return decision
+    for tf, ma in timeframes.items():
+        if price > ma and fgi < 40:
+            decisions[tf] = "BUY 🚀"
+        elif price < ma and fgi > 60:
+            decisions[tf] = "SELL 📉"
+        else:
+            decisions[tf] = "HODL 🤔"
+    
+    return decisions
+
+
+def get_fear_greed_index() -> Dict[str, Any]:
+    """Fetch Fear & Greed Index data."""
+    fgi_data = fetch_data("https://api.alternative.me/fng/")
+    if "error" in fgi_data or not fgi_data.get("data"):
+        return {"fear_greed": 50, "fgi_text": "Neutral"}
+    
+    return {
+        "fear_greed": int(fgi_data["data"][0]["value"]),
+        "fgi_text": fgi_data["data"][0]["value_classification"]
+    }
 
 
 # --- Agent Setup ---
@@ -188,8 +127,12 @@ agent = CodeAgent(
     tools=[final_answer, fetch_btc_news, btc_market_insights, btc_decision],
     max_steps=6,
     verbosity_level=1,
-    prompt_templates=prompt_templates
+    prompt_templates=prompt_templates,
+    grammar=None,
+    planning_interval=None,
+    name=None,
+    description=None
 )
 
 # --- Launch UI ---
-GradioUI(agent).launch()
+GradioUI(agent).launch()