Spaces:
Running
Running
File size: 7,600 Bytes
1597e57 79c3a6c 1597e57 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 |
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)
|