crypto_tools.py

from langchain.tools import tool
import requests
import pandas as pd
import numpy as np

class CryptoTools:
    @tool("Get Cryptocurrency Price")
    def get_crypto_price(crypto):
        """Fetches the current price of a given cryptocurrency."""
        url = f"https://api.coingecko.com/api/v3/simple/price?ids={crypto}&vs_currencies=usd"
        response = requests.get(url)
        if response.status_code == 200:
            data = response.json()
            return data[crypto]['usd']
        else:
            return "Error fetching price data"

    @tool("Get Market Cap")
    def get_market_cap(crypto):
        """Fetches the current market cap of a given cryptocurrency."""
        url = f"https://api.coingecko.com/api/v3/coins/{crypto}"
        response = requests.get(url)
        if response.status_code == 200:
            data = response.json()
            return data['market_data']['market_cap']['usd']
        else:
            return "Error fetching market cap data"

    @tool("Calculate RSI")
    def calculate_rsi(crypto, period=14):
        """Calculates the Relative Strength Index (RSI) for a given cryptocurrency."""
        url = f"https://api.coingecko.com/api/v3/coins/{crypto}/market_chart?vs_currency=usd&days={period}"
        response = requests.get(url)
        if response.status_code == 200:
            data = response.json()
            prices = [price[1] for price in data['prices']]
            df = pd.DataFrame(prices, columns=['price'])
            delta = df['price'].diff()
            gain = (delta.where(delta > 0, 0)).rolling(window=period).mean()
            loss = (-delta.where(delta < 0, 0)).rolling(window=period).mean()
            rs = gain / loss
            rsi = 100 - (100 / (1 + rs))
            return rsi.iloc[-1]
        else:
            return "Error calculating RSI"

    @tool("Calculate Moving Average")
    def calculate_moving_average(crypto, days):
        """Calculates the moving average for a given cryptocurrency over a specified number of days."""
        url = f"https://api.coingecko.com/api/v3/coins/{crypto}/market_chart?vs_currency=usd&days={days}"
        response = requests.get(url)
        if response.status_code == 200:
            data = response.json()
            prices = [price[1] for price in data['prices']]
            return np.mean(prices)
        else:
            return "Error calculating moving average"