app.py

from smolagents import CodeAgent, DuckDuckGoSearchTool, HfApiModel, load_tool, tool
import datetime
import requests
import pytz
import yaml
import os
from tools.final_answer import FinalAnswerTool
from Gradio_UI import GradioUI
from typing import Optional, List
from datetime import datetime, timedelta

# Set up OpenWeather API key
OPENWEATHER_API_KEY = os.getenv('OPENWEATHER_API_KEY', 'your_api_key_here')

@tool
def get_weather_info(city: str) -> str:
    """A tool that fetches current weather information for a given city
    Args:
        city: The name of the city to get weather for
    """
    try:
        base_url = "http://api.openweathermap.org/data/2.5/weather"
        params = {
            'q': city,
            'appid': OPENWEATHER_API_KEY,
            'units': 'metric'
        }
        
        response = requests.get(base_url, params=params)
        response.raise_for_status()
        
        weather_data = response.json()
        
        temperature = weather_data['main']['temp']
        conditions = weather_data['weather'][0]['description']
        humidity = weather_data['main']['humidity']
        wind_speed = weather_data['wind']['speed']
        
        return (f"Current weather in {city}:\n"
                f"Temperature: {temperature}°C\n"
                f"Conditions: {conditions}\n"
                f"Humidity: {humidity}%\n"
                f"Wind Speed: {wind_speed} m/s")
    except Exception as e:
        return f"Error fetching weather for city '{city}': {str(e)}"

@tool
def crypto_tracker(action: str, crypto_id: Optional[str] = None) -> str:
    """Tracks cryptocurrency prices and changes using CoinGecko API
    Args:
        action: The action to perform ('price', 'top', 'detail', 'summary')
        crypto_id: The cryptocurrency ID (e.g., 'bitcoin', 'ethereum') - required for 'price' and 'detail' actions
    """
    try:
        base_url = "https://api.coingecko.com/api/v3"
        
        def format_price_change(change: float) -> str:
            """Format price change with color indicator"""
            if change > 0:
                return f"↑ {change:.2f}%"
            elif change < 0:
                return f"↓ {change:.2f}%"
            return f"{change:.2f}%"

        if action == "top":
            # Get top 10 cryptocurrencies by market cap
            response = requests.get(
                f"{base_url}/coins/markets",
                params={
                    "vs_currency": "usd",
                    "order": "market_cap_desc",
                    "per_page": 10,
                    "page": 1,
                    "sparkline": False,
                    "price_change_percentage": "24h,7d"
                }
            )
            response.raise_for_status()
            data = response.json()
            
            result = "Top 10 Cryptocurrencies:\n\n"
            for coin in data:
                result += (
                    f"📊 {coin['name']} ({coin['symbol'].upper()})\n"
                    f"   Price: ${coin['current_price']:,.2f}\n"
                    f"   24h Change: {format_price_change(coin['price_change_percentage_24h'])}\n"
                    f"   7d Change: {format_price_change(coin['price_change_percentage_7d'])}\n"
                    f"   Market Cap: ${coin['market_cap']:,.0f}\n\n"
                )
            return result

        elif action == "price" and crypto_id:
            # Get simple price data for a specific cryptocurrency
            response = requests.get(
                f"{base_url}/simple/price",
                params={
                    "ids": crypto_id,
                    "vs_currencies": "usd",
                    "include_24hr_change": True,
                    "include_market_cap": True
                }
            )
            response.raise_for_status()
            data = response.json()
            
            if crypto_id not in data:
                return f"Cryptocurrency '{crypto_id}' not found"
                
            coin_data = data[crypto_id]
            return (
                f"Current {crypto_id.title()} Price:\n"
                f"USD: ${coin_data['usd']:,.2f}\n"
                f"24h Change: {format_price_change(coin_data['usd_24h_change'])}\n"
                f"Market Cap: ${coin_data['usd_market_cap']:,.0f}"
            )

        elif action == "detail" and crypto_id:
            # Get detailed information about a specific cryptocurrency
            response = requests.get(f"{base_url}/coins/{crypto_id}")
            response.raise_for_status()
            data = response.json()
            
            market_data = data['market_data']
            return (
                f"📊 {data['name']} ({data['symbol'].upper()}) Detailed Report:\n\n"
                f"Current Price: ${market_data['current_price']['usd']:,.2f}\n"
                f"24h Change: {format_price_change(market_data['price_change_percentage_24h'])}\n"
                f"7d Change: {format_price_change(market_data['price_change_percentage_7d'])}\n"
                f"30d Change: {format_price_change(market_data['price_change_percentage_30d'])}\n"
                f"Market Cap Rank: #{data['market_cap_rank']}\n"
                f"Market Cap: ${market_data['market_cap']['usd']:,.0f}\n"
                f"24h Trading Volume: ${market_data['total_volume']['usd']:,.0f}\n"
                f"Circulating Supply: {market_data['circulating_supply']:,.0f} {data['symbol'].upper()}\n"
                f"All-Time High: ${market_data['ath']['usd']:,.2f} ({market_data['ath_date']['usd'][:10]})\n"
                f"All-Time Low: ${market_data['atl']['usd']:,.2f} ({market_data['atl_date']['usd'][:10]})\n\n"
                f"Description: {data['description']['en'][:300]}..."
            )
            
        elif action == "summary":
            # Get 24-hour summary for top 5 cryptocurrencies
            response = requests.get(
                f"{base_url}/coins/markets",
                params={
                    "vs_currency": "usd",
                    "order": "market_cap_desc",
                    "per_page": 5,
                    "page": 1,
                    "sparkline": False,
                    "price_change_percentage": "24h",
                    "include_24hr_vol": True
                }
            )
            response.raise_for_status()
            data = response.json()
            
            total_market_change = sum(coin['price_change_percentage_24h'] for coin in data) / len(data)
            total_volume = sum(coin['total_volume'] for coin in data)
            
            summary = "🚀 24-Hour Crypto Market Summary\n\n"
            summary += f"Overall Market Trend: {format_price_change(total_market_change)} (Top 5 Average)\n"
            summary += f"Total Trading Volume: ${total_volume:,.0f}\n\n"
            summary += "Top 5 Cryptocurrency Performance:\n\n"
            
            for coin in data:
                price_change = coin['price_change_percentage_24h']
                volume_change = (coin['total_volume'] / coin['market_cap']) * 100
                
                # Determine market sentiment
                if price_change > 5:
                    sentiment = "🔥 Strong Bullish"
                elif price_change > 2:
                    sentiment = "📈 Bullish"
                elif price_change < -5:
                    sentiment = "💧 Strong Bearish"
                elif price_change < -2:
                    sentiment = "📉 Bearish"
                else:
                    sentiment = "➡️ Stable"
                
                summary += (
                    f"{coin['name']} ({coin['symbol'].upper()}):\n"
                    f"   Current Price: ${coin['current_price']:,.2f}\n"
                    f"   24h Change: {format_price_change(price_change)}\n"
                    f"   24h Volume: ${coin['total_volume']:,.0f}\n"
                    f"   Volume/Market Cap: {volume_change:.2f}%\n"
                    f"   Market Sentiment: {sentiment}\n\n"
                )
            
            # Add significant events
            significant_events = []
            for coin in data:
                if abs(coin['price_change_percentage_24h']) > 10:
                    significant_events.append(
                        f"⚠️ {coin['name']} moved {format_price_change(coin['price_change_percentage_24h'])} in 24h"
                    )
                if coin['current_price'] > coin.get('ath', float('inf')):
                    significant_events.append(f"🎉 {coin['name']} reached a new all-time high!")
                    
            if significant_events:
                summary += "Significant Events:\n"
                summary += "\n".join(f"• {event}" for event in significant_events)
            
            return summary
            
        else:
            return ("Invalid action. Use:\n"
                   "'top' to see top 10 cryptocurrencies\n"
                   "'price <crypto_id>' to get current price (e.g., 'price bitcoin')\n"
                   "'detail <crypto_id>' to get detailed information\n"
                   "'summary' to get 24-hour summary of top 5 cryptocurrencies")
                   
    except requests.exceptions.RequestException as e:
        return f"Error fetching cryptocurrency data: {str(e)}"
    except KeyError as e:
        return f"Error processing cryptocurrency data: {str(e)}"
    except Exception as e:
        return f"Unexpected error: {str(e)}"

@tool
def get_current_time_in_timezone(timezone: str) -> str:
    """A tool that fetches the current local time in a specified timezone.
    Args:
        timezone: A string representing a valid timezone (e.g., 'America/New_York').
    """
    try:
        tz = pytz.timezone(timezone)
        local_time = 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=2096,
    temperature=0.5,
    model_id='Qwen/Qwen2.5-Coder-32B-Instruct',
    custom_role_conversions=None,
)

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)

search_tool = DuckDuckGoSearchTool()

agent = CodeAgent(
    model=model,
    tools=[
        final_answer,
        search_tool,
        image_generation_tool,
        get_current_time_in_timezone,
        get_weather_info,
        crypto_tracker  # Fixed the name here
    ],
    max_steps=6,
    verbosity_level=1,
    grammar=None,
    planning_interval=None,
    name="Multi-tool Assistant",
    description="An agent capable of searching, generating images, checking time zones, checking crypto prices and getting weather information.",
    prompt_templates=prompt_templates
)

GradioUI(agent).launch()