app.py

from smolagents import CodeAgent,DuckDuckGoSearchTool, HfApiModel,load_tool,tool
import datetime
import requests
import pytz
import yaml
from tools.final_answer import FinalAnswerTool

from Gradio_UI import GradioUI
import requests
import pandas as pd
from datetime import datetime, timedelta

# Below is an example of a tool that does nothing. Amaze us with your creativity !
@tool
def my_custom_tool(arg1:str, arg2:int)-> str: #it's import to specify the return type
    #Keep this format for the description / args / args description but feel free to modify the tool
    """A tool that does nothing yet 
    Args:
        arg1: the first argument
        arg2: the second argument
    """
    return "What magic will you build ?"

@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:
        # Create timezone object
        tz = pytz.timezone(timezone)
        # Get current time in that 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)}"

@tool
def predict_crypto_price_binance(crypto_symbol: str = "BTC", vs_currency: str = "USDT") -> str:
    """
    Prédit l'évolution du prix d'une cryptomonnaie en se basant sur la SMA sur 9 et 20 jours et le RSI,
    en utilisant les données historiques issues de Binance.
    
    Args:
        crypto_symbol: Le symbole de la cryptomonnaie (ex: 'BTC').
        vs_currency: La devise de cotation (ex: 'USDT'). Pour 'USD', Binance utilise généralement 'USDT'.
    
    Returns:
        Un message indiquant le signal d'achat, de vente ou une analyse basée sur le RSI.
    """
    try:
        # Si l'utilisateur fournit 'usd', on le convertit en 'USDT' pour correspondre à Binance
        if vs_currency.lower() == "usd":
            vs_currency = "USDT"
        symbol = crypto_symbol.upper() + vs_currency.upper()
        
        # Récupération des données de 30 jours (bougies journalières) via l'API de Binance
        url = "https://api.binance.com/api/v3/klines"
        params = {
            "symbol": symbol,
            "interval": "1d",
            "limit": 30  # 30 jours de données
        }
        response = requests.get(url, params=params, timeout=10)
        response.raise_for_status()
        data = response.json()
        
        # Extraction des prix de clôture et des timestamps
        # Chaque kline (bougie) est une liste dont les éléments importants sont :
        # index 0: temps d'ouverture (en ms)
        # index 4: prix de clôture
        rows = []
        for kline in data:
            timestamp = kline[0]
            close_price = float(kline[4])
            rows.append([timestamp, close_price])
            
        df = pd.DataFrame(rows, columns=['timestamp', 'price'])
        df['date'] = pd.to_datetime(df['timestamp'], unit='ms')
        df.set_index('date', inplace=True)
        df.drop(columns=['timestamp'], inplace=True)
        
        # Calcul des moyennes mobiles simples sur 9 jours et 20 jours
        df['SMA_9'] = df['price'].rolling(window=9).mean()
        df['SMA_20'] = df['price'].rolling(window=20).mean()
        
        # Calcul du RSI sur une fenêtre de 14 jours
        delta = df['price'].diff()
        gain = delta.where(delta > 0, 0).rolling(window=14).mean()
        loss = -delta.where(delta < 0, 0).rolling(window=14).mean()
        rs = gain / loss
        df['RSI'] = 100 - (100 / (1 + rs))
        
        # Détection des signaux de croisement SMA
        if df['SMA_9'].iloc[-1] > df['SMA_20'].iloc[-1] and df['SMA_9'].iloc[-2] <= df['SMA_20'].iloc[-2]:
            return f"Signal d'achat : la SMA sur 9 jours a croisé au-dessus de la SMA sur 20 jours pour {symbol}."
        elif df['SMA_9'].iloc[-1] < df['SMA_20'].iloc[-1] and df['SMA_9'].iloc[-2] >= df['SMA_20'].iloc[-2]:
            return f"Signal de vente : la SMA sur 9 jours a croisé en dessous de la SMA sur 20 jours pour {symbol}."
        else:
            # Analyse du RSI si aucun croisement clair n'est détecté
            rsi_latest = df['RSI'].iloc[-1]
            if rsi_latest > 70:
                return f"Aucun croisement clair. Toutefois, le RSI est {rsi_latest:.2f}, indiquant des conditions de surachat (risque de correction)."
            elif rsi_latest < 30:
                return f"Aucun croisement clair. Toutefois, le RSI est {rsi_latest:.2f}, indiquant des conditions de survente (potentiel rebond)."
            else:
                return f"Aucun croisement clair. Le RSI est de {rsi_latest:.2f}, indiquant des conditions neutres."
    
    except Exception as e:
        return f"Une erreur s'est produite : {str(e)}"


final_answer = FinalAnswerTool()

# If the agent does not answer, the model is overloaded, please use another model or the following Hugging Face Endpoint that also contains qwen2.5 coder:
# model_id='https://pflgm2locj2t89co.us-east-1.aws.endpoints.huggingface.cloud' 

model = HfApiModel(
max_tokens=2096,
temperature=0.5,
model_id='Qwen/Qwen2.5-Coder-32B-Instruct',# it is possible that this model may be overloaded
custom_role_conversions=None,
)


# 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 = CodeAgent(
    model=model,
    tools=[final_answer], ## add your tools here (don't remove final answer)
    max_steps=6,
    verbosity_level=1,
    grammar=None,
    planning_interval=None,
    name=None,
    description=None,
    prompt_templates=prompt_templates
)


GradioUI(agent).launch()