Update utils/forex_signals.py

utils/forex_signals.py

@@ -1,63 +1,81 @@
 import requests
-import random
-from datetime import datetime, timedelta
+import pandas as pd
+import numpy as np
 
 API_KEY = "89SEdLScHxHk6j8J9OoH4sLFS3Mri4oW"
-BASE_URL = "https://financialmodelingprep.com/api/v3/forex"
 
-def fetch_forex_data(pair):
-    """Fetch historical forex data for a given currency pair."""
+CURRENCY_PAIRS = [
+    "EUR/USD", "GBP/USD", "USD/JPY", "AUD/USD",
+    "CAD/JPY", "NZD/USD", "CHF/JPY", "AUD/JPY",
+    "GBP/CHF", "EUR/GBP"
+]
+
+
+def fetch_forex_data(currency_pair):
     try:
-        url = f"{BASE_URL}/{pair}?apikey={API_KEY}"
+        # Convert the pair into the format required by the API (EURUSD, GBPUSD, etc.)
+        formatted_pair = currency_pair.replace("/", "")
+        url = f"https://financialmodelingprep.com/api/v3/forex/{formatted_pair}?apikey={API_KEY}"
         response = requests.get(url)
+        response.raise_for_status()
+
+        data = response.json()
 
-        if response.status_code == 200:
-            data = response.json()
-            if not data or "historical" not in data:
-                return None  # No data available for this pair
-            return data["historical"]
+        if "historical" in data and len(data["historical"]) > 0:
+            return pd.DataFrame(data["historical"])
         else:
-            print(f"Error fetching data for {pair}: {response.status_code} - {response.text}")
+            print(f"No data available for {currency_pair}.")
             return None
     except Exception as e:
-        print(f"Exception while fetching data for {pair}: {str(e)}")
+        print(f"Error fetching data for {currency_pair}: {e}")
+        return None
+
+
+def calculate_indicators(df):
+    if df is None or df.empty:
+        return None
+
+    try:
+        df["SMA_50"] = df["close"].rolling(window=50).mean()
+        df["SMA_200"] = df["close"].rolling(window=200).mean()
+        df["RSI"] = compute_rsi(df["close"])
+        return df
+    except Exception as e:
+        print(f"Error calculating indicators: {e}")
         return None
 
-def generate_forex_signals(trading_capital, market_risk, user_timezone, additional_pairs):
-    """Generate forex trading signals."""
+
+def compute_rsi(series, period=14):
+    delta = series.diff(1)
+    gain = delta.where(delta > 0, 0)
+    loss = -delta.where(delta < 0, 0)
+
+    avg_gain = gain.rolling(window=period).mean()
+    avg_loss = loss.rolling(window=period).mean()
+
+    rs = avg_gain / avg_loss
+    rsi = 100 - (100 / (1 + rs))
+    return rsi
+
+
+def generate_forex_signals():
     signals = []
-    for pair in additional_pairs:
-        print(f"Fetching data for {pair}...")
-        data = fetch_forex_data(pair)
-
-        if data is None:
-            print(f"Skipping {pair} due to missing data.")
-            continue  # Skip to the next pair if data is missing
-
-        # Simulate processing data for signal generation
-        entry_time = datetime.now()
-        exit_time = entry_time + timedelta(hours=random.randint(1, 5))  # Simulated exit time
-        roi = round(random.uniform(0.5, 10.0), 2)  # Simulated ROI
-        signal_strength = round(random.uniform(50, 100), 2)  # Simulated signal strength
-
-        signals.append({
-            "currency_pair": pair,
-            "entry_time": entry_time.strftime("%Y-%m-%d %H:%M:%S"),
-            "exit_time": exit_time.strftime("%Y-%m-%d %H:%M:%S"),
-            "roi": roi,
-            "signal_strength": signal_strength,
-        })
-
-    # Sort signals by ROI (descending) to recommend the best signal
-    signals.sort(key=lambda x: x["roi"], reverse=True)
-
-    if signals:
-        return {
-            "best_signal": signals[0],
-            "all_signals": signals,
-        }
-    else:
-        return {
-            "best_signal": None,
-            "all_signals": [],
-        }
+
+    for pair in CURRENCY_PAIRS:
+        print(f"Processing currency pair: {pair}")
+
+        df = fetch_forex_data(pair)
+        if df is None:
+            continue
+
+        df = calculate_indicators(df)
+        if df is None or df.empty:
+            continue
+
+        # Check for crossover signals
+        if df["SMA_50"].iloc[-1] > df["SMA_200"].iloc[-1] and df["RSI"].iloc[-1] < 70:
+            signals.append({"pair": pair, "signal": "Buy"})
+        elif df["SMA_50"].iloc[-1] < df["SMA_200"].iloc[-1] and df["RSI"].iloc[-1] > 30:
+            signals.append({"pair": pair, "signal": "Sell"})
+
+    return signals