Update app.py

app.py

@@ -1,165 +1,69 @@
+from smolagents import CodeAgent, DuckDuckGoSearchTool, HfApiModel, load_tool, tool
+import datetime
+import json
 import requests
-from typing import List, Dict, Any
-import numpy as np
+import pytz
 import yaml
-from smolagents import CodeAgent, HfApiModel, load_tool, tool
 from tools.final_answer import FinalAnswerTool
 from Gradio_UI import GradioUI
 
-# ✅ Final Answer Tool
-final_answer = FinalAnswerTool()
-
-# ✅ AI Model Setup
-model = HfApiModel(
-    max_tokens=2096,
-    temperature=0.5,
-    model_id='Qwen/Qwen2.5-Coder-32B-Instruct',
-    custom_role_conversions=None,
-)
-
-# ✅ Load Prompt Templates
-with open("prompts.yaml", 'r') as stream:
-    prompt_templates = yaml.safe_load(stream)
-
-
-# --- Utility Functions ---
-
-def fetch_data(url: str, params: Dict[str, Any] = None) -> Dict[str, Any]:
-    """Make GET requests and return JSON response."""
-    try:
-        response = requests.get(url, params=params, timeout=10)
-        response.raise_for_status()
-        return response.json()
-    except requests.exceptions.RequestException as e:
-        return {"error": str(e)}
-
-
-def compute_moving_average(prices: List[float], window: int) -> float:
-    """Compute moving average for given price data."""
-    return np.mean(prices[-window:]) if len(prices) >= window else np.mean(prices)
-
-
-# --- Tools ---
-
 @tool
-def fetch_btc_news() -> List[Dict[str, str]]:
-    """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]]
-
+def get_btc_price() -> str:
+    """Fetches the current Bitcoin price in USD using the CoinGecko API."""
+    url = "https://api.coingecko.com/api/v3/simple/price?ids=bitcoin&vs_currencies=usd"
+    response = requests.get(url)
+    if response.status_code != 200:
+        return '{"error": "Failed to fetch Bitcoin price"}'
+    data = response.json()
+    price = data.get('bitcoin', {}).get('usd')
+    if price is None:
+        return '{"error": "Bitcoin price not found"}'
+    return f"The current Bitcoin price is ${price:.2f} USD."
 
 @tool
-def btc_market_insights() -> Dict[str, Any]:
-    """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": 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()
-    }
+def get_btc_news() -> str:
+    """Fetches the latest Bitcoin news using DuckDuckGo search."""
+    search_query = "latest Bitcoin news"
+    search_results = DuckDuckGoSearchTool().search(search_query)
+    if not search_results:
+        return '{"error": "No news found"}'
+    news = [{"title": result['title'], "url": result['url']} for result in search_results]
+    return json.dumps(news, indent=2)
 
 @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 🤔"
-    }
-
-    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
+def get_current_time_in_timezone(timezone: str) -> str:
+    """Fetches the current local time in a specified timezone."""
+    try:
+        tz = pytz.timezone(timezone)
+        local_time = datetime.datetime.now(tz).strftime("%Y-%m-%d %H:%M:%S")
+        return f"The current local time in {timezone} is: {local_time}"
+    except Exception as e:
+        return f"Error fetching time for timezone '{timezone}': {str(e)}"
 
+final_answer = FinalAnswerTool()
+model = HfApiModel(
+    max_tokens=1048,
+    temperature=0.5,
+    model_id='deepseek-ai/DeepSeek-R1-Distill-Qwen-32B',
+    custom_role_conversions=None,
+)
 
-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"]
-    }
+# Import tool from Hub
+image_generation_tool = load_tool("agents-course/text-to-image", trust_remote_code=True)
 
+with open("prompts.yaml", 'r') as stream:
+    prompt_templates = yaml.safe_load(stream)
 
-# --- Agent Setup ---
 agent = CodeAgent(
     model=model,
-    tools=[final_answer, fetch_btc_news, btc_market_insights, btc_decision],
+    tools=[get_btc_price, get_btc_news, DuckDuckGoSearchTool(), image_generation_tool, final_answer],
     max_steps=6,
     verbosity_level=1,
-    prompt_templates=prompt_templates,
     grammar=None,
     planning_interval=None,
     name=None,
-    description=None
+    description=None,
+    prompt_templates=prompt_templates
 )
 
-# --- Launch UI ---
 GradioUI(agent).launch()