Spaces:
Sleeping
Sleeping
| 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) | |