app.py

import os
import sys
import requests
from openai import OpenAI
import pandas as pd
import pandas_ta as ta
import yfinance as yf
from dotenv import load_dotenv
import gradio as gr

# Load environment variables
load_dotenv()


class IntelligentForexAnalyzer:
    def __init__(self):
        print(f"Python version: {sys.version}")
        print("IntelligentForexAnalyzer initialized.\n")

    def get_forex_data(self, pair, period="1y", interval="1h"):
        """
        Fetches Forex data from Yahoo Finance and calculates RSI, MACD, and Bollinger Bands.
        """
        try:
            print(f"Fetching data for {pair}...")
            symbol = f"{pair.upper()}=X"
            df = yf.download(symbol, period=period, interval=interval, progress=False)

            if df.empty:
                print(f"No data returned for {symbol}. Check the pair or interval.")
                return None

            print(df.head())  # Debug fetched data

            # Calculate indicators
            df["RSI"] = ta.rsi(df["Close"], length=14)

            macd = ta.macd(df["Close"])
            if macd is not None:
                df["MACD"] = macd["MACD_12_26_9"]
                df["MACD_signal"] = macd["MACDs_12_26_9"]
                df["MACD_hist"] = macd["MACDh_12_26_9"]
            else:
                df["MACD"], df["MACD_signal"], df["MACD_hist"] = None, None, None

            bb = ta.bbands(df["Close"], length=20, std=2)
            if bb is not None:
                df["BB_lower"], df["BB_mid"], df["BB_upper"] = (
                    bb["BBL_20_2.0"],
                    bb["BBM_20_2.0"],
                    bb["BBU_20_2.0"],
                )
            else:
                df["BB_lower"], df["BB_mid"], df["BB_upper"] = None, None, None

            return df
        except Exception as e:
            print(f"Error fetching data for {pair}: {e}")
            return None

    def analyze_pairs(self, pairs):
        """
        Analyzes Forex pairs and summarizes indicators for trading decisions.
        """
        results = {}
        for pair in pairs:
            df = self.get_forex_data(pair)
            if df is None:
                results[pair] = "No data available"
                continue

            close_price = df["Close"].iloc[-1]
            rsi = df["RSI"].iloc[-1]
            macd = df["MACD"].iloc[-1]
            macd_signal = df["MACD_signal"].iloc[-1]
            macd_hist = df["MACD_hist"].iloc[-1]
            bb_lower = df["BB_lower"].iloc[-1]
            bb_upper = df["BB_upper"].iloc[-1]

            indicators = {
                "Close Price": close_price,
                "RSI": f"{rsi:.2f}" if not pd.isna(rsi) else "Not available",
                "MACD": f"{macd:.2f}" if not pd.isna(macd) else "Not available",
                "MACD Signal": (
                    f"{macd_signal:.2f}"
                    if not pd.isna(macd_signal)
                    else "Not available"
                ),
                "MACD Hist": (
                    f"{macd_hist:.2f}" if not pd.isna(macd_hist) else "Not available"
                ),
                "Bollinger Bands": (
                    f"{bb_lower:.2f} - {bb_upper:.2f}"
                    if not pd.isna(bb_lower) and not pd.isna(bb_upper)
                    else "Not available"
                ),
            }

            decision = self.make_trading_decision(indicators)
            results[pair] = {"Indicators": indicators, "Decision": decision}

        return results

    def make_trading_decision(self, indicators):
        """
        Makes a basic trading decision based on indicators.
        """
        rsi = float(indicators["RSI"]) if indicators["RSI"] != "Not available" else None
        macd = (
            float(indicators["MACD"]) if indicators["MACD"] != "Not available" else None
        )
        macd_signal = (
            float(indicators["MACD Signal"])
            if indicators["MACD Signal"] != "Not available"
            else None
        )
        close_price = indicators["Close Price"]

        if rsi is not None:
            if rsi < 30:
                return f"LONG: RSI={rsi} indicates oversold conditions."
            elif rsi > 70:
                return f"SHORT: RSI={rsi} indicates overbought conditions."

        if macd is not None and macd_signal is not None:
            if macd > macd_signal:
                return (
                    f"LONG: MACD={macd} is above Signal={macd_signal} (bullish trend)."
                )
            elif macd < macd_signal:
                return (
                    f"SHORT: MACD={macd} is below Signal={macd_signal} (bearish trend)."
                )

        return f"HOLD: Price={close_price} within Bollinger range."

    def analyze_with_llm(self, analysis_results):
        """
        Uses OpenAI GPT for deeper insights using the updated ChatCompletion API.
        """
        client = OpenAI(api_key=os.getenv("OPENAI_API_KEY"))

        messages = [
            {
                "role": "system",
                "content": "You are a Forex trading assistant. Analyze the following data:",
            }
        ]
        for pair, data in analysis_results.items():
            if isinstance(data, str):
                messages.append(
                    {"role": "user", "content": f"PAIR: {pair}\nError: {data}"}
                )
            else:
                messages.append(
                    {
                        "role": "user",
                        "content": f"PAIR: {pair}\nIndicators: {data['Indicators']}\nDecision: {data['Decision']}",
                    }
                )

        try:
            response = client.chat.completions.create(
                model="gpt-3.5-turbo", messages=messages, temperature=0.5
            )
            return response.choices[0].message.content.strip()
        except Exception as e:
            return f"Error during LLM analysis: {e}"


def analyze_forex(forex_pairs):
    pairs = forex_pairs.split(",")
    analyzer = IntelligentForexAnalyzer()
    analysis = analyzer.analyze_pairs([pair.strip() for pair in pairs])
    llm_result = analyzer.analyze_with_llm(analysis)
    result = "\n\n".join([f"{pair}: {data}" for pair, data in analysis.items()])
    return result, llm_result


interface = gr.Interface(
    fn=analyze_forex,
    inputs=gr.Textbox(
        label="Enter Forex Pairs (comma-separated)",
        placeholder="e.g., EURUSD, GBPUSD, USDJPY",
    ),
    outputs=[gr.Textbox(label="Analysis Results"), gr.Textbox(label="LLM analysis")],
    title="Intelligent Forex Analyzer",
    description="Enter Forex pairs to analyze their RSI, MACD, and Bollinger Bands and get trading recommendations.",
)


def auth_function(username, password):
    valid_users = {"admin": "demo4nicolas"}
    return username in valid_users and valid_users[username] == password


if __name__ == "__main__":
    interface.launch(auth=auth_function, share=True, ssr_mode=False)
# if __name__ == "__main__":
#     analyzer = IntelligentForexAnalyzer()
#     forex_pairs = ["EURUSD", "GBPUSD", "USDJPY"]
#     analysis = analyzer.analyze_pairs(forex_pairs)

#     print("\n=== Forex Analysis ===")
#     for pair, data in analysis.items():
#         print(f"{pair}: {data}")

#     llm_output = analyzer.analyze_with_llm(analysis)
#     print("\n=== LLM Analysis ===")
#     print(llm_output)