File: //opt/af/src/app.py
"""
Main application file for ActionForex AI Agent.
"""
import streamlit as st
from dotenv import load_dotenv
import os
import logging
import pandas as pd
import json
from streamlit.delta_generator import DeltaGenerator
from datetime import datetime
import time
import stripe
import uuid
import webbrowser
import traceback
import hashlib
import random
import string
import re
import pickle
import os.path
from pathlib import Path
# Kommentiere die st_paywall-Integration aus, da sie Probleme verursacht
# from st_paywall import add_auth
# add_auth(required=True)
# Load environment variables (early, important for Stripe configuration)
load_dotenv()
# Configure logging mit ausführlicherem Format
logging.basicConfig(level=logging.DEBUG, format='%(asctime)s - %(levelname)s - %(message)s')
log = logging.getLogger(__name__)
# Stripe Konfiguration und Hilfsfunktionen
stripe_api_key = st.secrets.get("stripe_api_key")
stripe_product_id = st.secrets.get("stripe_product_id")
redirect_url = st.secrets.get("redirect_url")
# Pfad für Benutzer-Datenbank definieren
USERS_DB_PATH = Path("data/users_db.pickle")
TOKENS_DB_PATH = Path("data/tokens_db.pickle")
# Stelle sicher, dass das Datenverzeichnis existiert
os.makedirs(USERS_DB_PATH.parent, exist_ok=True)
# Hilfsfunktion zum Laden der Benutzer-Datenbank
def load_users_db():
if USERS_DB_PATH.exists():
with open(USERS_DB_PATH, 'rb') as f:
return pickle.load(f)
return {}
# Hilfsfunktion zum Speichern der Benutzer-Datenbank
def save_users_db(users_db):
with open(USERS_DB_PATH, 'wb') as f:
pickle.dump(users_db, f)
# Hilfsfunktion zum Laden der Tokens-Datenbank
def load_tokens_db():
if TOKENS_DB_PATH.exists():
with open(TOKENS_DB_PATH, 'rb') as f:
return pickle.load(f)
return {}
# Hilfsfunktion zum Speichern der Tokens-Datenbank
def save_tokens_db(tokens_db):
with open(TOKENS_DB_PATH, 'wb') as f:
pickle.dump(tokens_db, f)
# Hilfsfunktion zum Hashen eines Passworts
def hash_password(password):
return hashlib.sha256(password.encode()).hexdigest()
# Hilfsfunktion zum Überprüfen eines Passworts
def verify_password(stored_hash, password):
return stored_hash == hash_password(password)
# Hilfsfunktion zum Generieren eines zufälligen Tokens
def generate_token():
return ''.join(random.choices(string.ascii_letters + string.digits, k=32))
# Hilfsfunktion zum Senden einer E-Mail mit Token (Simulation)
def send_verification_email(email, token):
log.info(f"Sende Bestätigungs-E-Mail an {email} mit Token {token}")
# In einer echten Anwendung würde hier eine E-Mail gesendet
# Für Testzwecke geben wir den Token einfach aus
return True
# Hilfsfunktion zur E-Mail-Validierung
def is_valid_email(email):
# Einfache Validierung, ob die E-Mail ein @ und einen Punkt enthält
return re.match(r"[^@]+@[^@]+\.[^@]+", email) is not None
# Logge die Konfigurationsdaten (teilweise maskiert für Sicherheit)
if stripe_api_key:
masked_key = stripe_api_key[:4] + "*" * (len(stripe_api_key) - 8) + stripe_api_key[-4:]
log.info(f"Stripe API-Schlüssel konfiguriert: {masked_key}")
else:
log.error("Stripe API-Schlüssel fehlt in secrets.toml")
log.info(f"Stripe Produkt-ID: {stripe_product_id}")
log.info(f"Redirect-URL: {redirect_url}")
if not stripe_api_key or not stripe_product_id:
log.error("Stripe API-Schlüssel oder Produkt-ID fehlt in der secrets.toml")
try:
# Setze den API-Schlüssel und logge die Stripe-Bibliotheksversion
stripe.api_key = stripe_api_key
log.info(f"Stripe-Bibliotheksversion: {stripe.api_version}")
# Teste die Verbindung mit einem einfachen API-Aufruf
account = stripe.Account.retrieve()
log.info(f"Stripe-Verbindung erfolgreich hergestellt. Account-ID: {account.id}")
except Exception as e:
log.error(f"Fehler beim Initialisieren der Stripe-Verbindung: {str(e)}")
log.error(traceback.format_exc())
def debug_stripe_api_call(func_name, params=None, error=None, response=None):
"""Hilfsfunktion zum Loggen von Stripe-API-Aufrufen."""
log.info(f"STRIPE API CALL: {func_name}")
if params:
# Parameter loggen, sensible Daten maskieren
safe_params = params.copy() if isinstance(params, dict) else {"non_dict_params": str(params)}
if safe_params.get("customer_email"):
safe_params["customer_email"] = safe_params["customer_email"][:3] + "***" + safe_params["customer_email"].split("@")[0][-2:] + "@" + safe_params["customer_email"].split("@")[1]
log.info(f"STRIPE PARAMS: {json.dumps(safe_params, default=str)}")
if error:
log.error(f"STRIPE ERROR: {str(error)}")
log.error(traceback.format_exc())
if response:
# Antwort loggen, sensible Daten maskieren
try:
if hasattr(response, "to_dict"):
response_dict = response.to_dict()
elif isinstance(response, dict):
response_dict = response
else:
response_dict = {"response": str(response)}
# Maskiere sensible Daten
if "url" in response_dict and response_dict["url"]:
log.info(f"STRIPE RESPONSE URL: {response_dict['url']}")
log.info(f"STRIPE RESPONSE: {json.dumps(response_dict, default=str)[:1000]}...")
except Exception as e:
log.error(f"Fehler beim Loggen der Stripe-Antwort: {str(e)}")
def create_checkout_session(customer_email):
"""Erstellt eine Stripe Checkout-Session für ein Abonnement."""
try:
log.info(f"Erstelle Checkout-Session für E-Mail: {customer_email}")
# Struktur der line_items entsprechend der Stripe API-Dokumentation korrigieren
# https://stripe.com/docs/api/checkout/sessions/create
params = {
"payment_method_types": ['card'],
"line_items": [
{
'price': stripe_product_id,
'quantity': 1,
},
],
"mode": 'subscription',
"success_url": f'{redirect_url}?session_id={{CHECKOUT_SESSION_ID}}',
"cancel_url": f'{redirect_url}?canceled=true',
}
# Füge die E-Mail nur hinzu, wenn sie nicht leer ist
if customer_email and '@' in customer_email:
params["customer_email"] = customer_email
# Logge die API-Parameter
debug_stripe_api_call("checkout.Session.create", params)
# Teste zuerst, ob das Produkt existiert
try:
price = stripe.Price.retrieve(stripe_product_id)
log.info(f"Produkt gefunden: {price.id}, aktiv: {price.active}")
except Exception as e:
log.error(f"Produkt nicht gefunden oder nicht aktiv: {stripe_product_id}, Fehler: {e}")
return None
# Stripe-Anfrage durchführen
checkout_session = stripe.checkout.Session.create(**params)
# Logge die Antwort
debug_stripe_api_call("checkout.Session.create", params=None, response=checkout_session)
log.info(f"Checkout-Session erstellt: {checkout_session.url}")
return checkout_session
except Exception as e:
debug_stripe_api_call("checkout.Session.create", params, error=e)
log.error(f"Stripe Checkout-Fehler: {e}")
log.error(traceback.format_exc())
return None
def is_active_subscriber(email):
"""Überprüft, ob die E-Mail-Adresse einem aktiven Abonnenten gehört."""
try:
log.info(f"Überprüfe Abonnement für E-Mail: {email}")
# Kommentiere diese Zeile aus, um die tatsächliche Stripe-Prüfung durchzuführen
# log.info("Entwicklungsmodus: Gebe False zurück für Testzwecke")
# return False
# Suche nach dem Kunden anhand der E-Mail-Adresse
debug_stripe_api_call("Customer.list", {"email": email})
customers = stripe.Customer.list(email=email)
debug_stripe_api_call("Customer.list", {"email": email}, response=customers)
if not customers.data:
log.info(f"Kein Kunde mit E-Mail {email} gefunden")
return False
# Überprüfe für jeden gefundenen Kunden, ob aktive Abonnements vorhanden sind
for customer in customers.data:
debug_stripe_api_call("Subscription.list", {"customer": customer.id, "status": "active"})
subscriptions = stripe.Subscription.list(customer=customer.id, status='active')
debug_stripe_api_call("Subscription.list", {"customer": customer.id}, response=subscriptions)
if subscriptions.data:
log.info(f"Aktives Abonnement für Kunde {customer.id} gefunden")
return True
log.info(f"Kein aktives Abonnement für E-Mail {email} gefunden")
return False
except Exception as e:
debug_stripe_api_call("is_active_subscriber", {"email": email}, error=e)
log.error(f"Stripe Abonnementprüfung fehlgeschlagen: {e}")
# Bei einem Fehler zeigen wir eine detaillierte Fehlermeldung an
st.error(f"Fehler bei der Überprüfung des Abonnements: {e}")
return False
def show_stripe_payment_page(email):
"""Zeigt die Stripe-Zahlungsseite an oder eine simulierte Zahlungsseite, wenn Stripe nicht verfügbar ist."""
log.info(f"Zeige Zahlungsseite für E-Mail: {email}")
st.title("Premium Subscription Required")
st.markdown("""
### A premium subscription is required for full access to ActionForex AI.
With a premium subscription, you'll get:
- Real-time market data analysis
- Access to historical data and trends
- Trading recommendations and technical indicators
- Personalized notifications for important market events
""")
try:
# Versuche eine Checkout-Session zu erstellen
checkout_session = create_checkout_session(email)
if checkout_session:
# Zeige den Abonnieren-Button an
st.markdown(f"""
<a href="{checkout_session.url}" target="_blank">
<button style="
background-color: #6772E5;
color: white;
padding: 12px 24px;
border: none;
border-radius: 4px;
cursor: pointer;
font-size: 16px;
margin-top: 20px;
width: 100%;
">
Subscribe now for only €19.99/month
</button>
</a>
""", unsafe_allow_html=True)
else:
# Die checkout_session konnte nicht erstellt werden
if stripe_api_key and stripe_api_key.startswith("sk_live_") and stripe_product_id and stripe_product_id.startswith("price_"):
# Wahrscheinlich ein Mismatch zwischen Live-Schlüssel und Test-Produkt
st.error("""
**Configuration error detected:**
There appears to be a mismatch between the API key and the product ID:
- A live mode API key is being used (starts with 'sk_live_')
- The product ID likely exists only in test mode
Please contact the administrator to resolve this issue.
""")
else:
# Allgemeine Fehlermeldung
st.error("There was a problem connecting to our payment provider. Please try again later.")
except Exception as e:
log.error(f"Fehler bei der Stripe-Integration: {e}")
# SIMULIERTE ZAHLUNG - Nur für Testzwecke
st.warning("**Stripe integration is unavailable**")
st.info("For testing purposes, we're using a simulated payment option.")
col1, col2 = st.columns(2)
with col1:
# Simulierter Zahlungsbutton
if st.button("Simulate payment (test mode)", key="simulate_payment"):
st.session_state.payment_processing = True
st.rerun()
with col2:
# Debug-Option zum Überspringen
if st.button("Skip (developers only)", key="skip_payment"):
st.session_state.subscription_verified = True
st.rerun()
# Simulierte Zahlungsverarbeitung
if st.session_state.get("payment_processing", False):
with st.spinner("Processing payment..."):
# Simuliere Verzögerung
time.sleep(2)
# Zeige Erfolgsmeldung
st.success("Payment successful! Your premium subscription is now active.")
# Weiterleitung zur Hauptseite nach kurzer Verzögerung
time.sleep(1)
st.session_state.subscription_verified = True
st.session_state.payment_processing = False
st.rerun()
# Stoppt die weitere Ausführung
st.stop()
# Erweiterte Authentifizierung mit Registrierung und E-Mail-Bestätigung
def auth_system():
"""Zeigt entweder die Login- oder Registrierungsseite an und verwaltet den Authentifizierungsfluss."""
# Initialisiere Session-State-Variablen, falls sie nicht existieren
if 'authenticated' not in st.session_state:
st.session_state.authenticated = False
if 'subscription_verified' not in st.session_state:
st.session_state.subscription_verified = False
if 'auth_page' not in st.session_state:
st.session_state.auth_page = 'login' # 'login', 'register', 'verify', 'reset_password'
# Token in der URL verarbeiten (für E-Mail-Bestätigung)
query_params = st.query_params
if 'token' in query_params and query_params['token']:
token = query_params['token']
tokens_db = load_tokens_db()
if token in tokens_db:
email = tokens_db[token]['email']
action = tokens_db[token]['action']
if action == 'verify_email':
# E-Mail bestätigen
users_db = load_users_db()
if email in users_db:
users_db[email]['email_verified'] = True
save_users_db(users_db)
# Token entfernen
del tokens_db[token]
save_tokens_db(tokens_db)
st.success(f"E-Mail-Adresse {email} erfolgreich bestätigt! Sie können sich jetzt anmelden.")
st.session_state.auth_page = 'login'
else:
st.error("Ungültiger Token oder Benutzer existiert nicht.")
elif action == 'reset_password':
# Hier könnte eine Passwort-Reset-Logik implementiert werden
st.warning("Passwort-Reset-Funktion ist noch nicht implementiert.")
# Query-Parameter entfernen
st.query_params.clear()
# Zeige Login-Formular
if st.session_state.auth_page == 'login':
display_login_form()
# Zeige Registrierungsformular
elif st.session_state.auth_page == 'register':
display_register_form()
# Zeige Verifizierungshinweis
elif st.session_state.auth_page == 'verify':
display_verification_notice()
# Nach erfolgreicher Anmeldung Abonnement prüfen
if st.session_state.authenticated and not st.session_state.subscription_verified:
user_email = st.session_state.user_email
log.info(f"Überprüfe Abonnement nach erfolgreicher Anmeldung für: {user_email}")
# Überprüfe, ob der Benutzer ein aktives Abonnement hat
has_subscription = is_active_subscriber(user_email)
if has_subscription:
log.info(f"Aktives Abonnement gefunden für: {user_email}")
st.session_state.subscription_verified = True
st.rerun()
else:
log.info(f"Kein aktives Abonnement gefunden für: {user_email}")
# Zeige die Zahlungsseite
show_stripe_payment_page(user_email)
def display_login_form():
"""Zeigt das Login-Formular an."""
st.title("ActionForex Login")
col1, col2 = st.columns([2, 1])
with col1:
email = st.text_input("Email")
password = st.text_input("Password", type="password")
if st.button("Sign In"):
if validate_login(email, password):
st.session_state.authenticated = True
st.session_state.user_email = email
st.rerun()
else:
st.error("Invalid credentials or email not verified.")
with col2:
st.markdown("### Don't have an account?")
if st.button("Register"):
st.session_state.auth_page = 'register'
st.rerun()
if st.button("Forgot Password"):
# Hier würde eine Passwort-Reset-Funktion implementiert
st.warning("Password reset functionality is not yet implemented.")
# Debug-Login für Entwicklung (In Produktion entfernen!)
with st.expander("Debug Login (development only)"):
if st.button("Debug Login"):
st.session_state.authenticated = True
st.session_state.user_email = "debug@example.com"
st.session_state.subscription_verified = True
st.rerun()
def display_register_form():
"""Zeigt das Registrierungsformular an."""
st.title("ActionForex Registration")
col1, col2 = st.columns([2, 1])
with col1:
email = st.text_input("Email")
password = st.text_input("Password", type="password")
password_confirm = st.text_input("Confirm Password", type="password")
if st.button("Register"):
if register_user(email, password, password_confirm):
st.session_state.auth_page = 'verify'
st.rerun()
with col2:
st.markdown("### Already registered?")
if st.button("Go to Login"):
st.session_state.auth_page = 'login'
st.rerun()
def display_verification_notice():
"""Zeigt einen Hinweis an, dass eine Bestätigungs-E-Mail gesendet wurde."""
st.title("Verification Email Sent")
st.info("We've sent a verification email to your address. Please click on the link in the email to complete your registration.")
# In einer Testumgebung: Zeige den Token direkt an
tokens_db = load_tokens_db()
user_email = st.session_state.get('registered_email', '')
if user_email:
for token, data in tokens_db.items():
if data['email'] == user_email and data['action'] == 'verify_email':
st.code(f"{redirect_url}?token={token}")
if st.button("Go to Login"):
st.session_state.auth_page = 'login'
st.rerun()
def validate_login(email, password):
"""Überprüft, ob die Anmeldedaten gültig sind."""
if not email or not password:
return False
users_db = load_users_db()
if email in users_db:
user = users_db[email]
if verify_password(user['password_hash'], password):
if user.get('email_verified', False):
return True
else:
st.warning("Your email address has not been verified yet. Please check your inbox.")
return False
return False
def register_user(email, password, password_confirm):
"""Registriert einen neuen Benutzer."""
# Validierung
if not email or not password or not password_confirm:
st.error("Please fill in all fields.")
return False
if not is_valid_email(email):
st.error("Please enter a valid email address.")
return False
if password != password_confirm:
st.error("Passwords don't match.")
return False
if len(password) < 8:
st.error("Password must be at least 8 characters long.")
return False
users_db = load_users_db()
if email in users_db:
st.error("A user with this email address already exists.")
return False
# Neuen Benutzer erstellen
users_db[email] = {
'password_hash': hash_password(password),
'created_at': datetime.now().isoformat(),
'email_verified': False
}
save_users_db(users_db)
# Bestätigungs-Token generieren und speichern
token = generate_token()
tokens_db = load_tokens_db()
tokens_db[token] = {
'email': email,
'action': 'verify_email',
'created_at': datetime.now().isoformat()
}
save_tokens_db(tokens_db)
# Bestätigungs-E-Mail senden
send_verification_email(email, token)
# E-Mail für die Anzeige des Tokens speichern
st.session_state.registered_email = email
return True
# Configure the Streamlit page - THIS MUST REMAIN THE FIRST STREAMLIT COMMAND
st.set_page_config(
page_title="ActionForex AI Agent",
page_icon="📊",
layout="wide"
)
# Authentifizierungsprüfung durchführen
auth_system()
# Own modules
from src.ai.agent import AIAgent # Import our agent
from src.utils.sftp_downloader import download_sftp_data # Keep the download utility
from src.ui.visualizations import display_data # New import
from src.calendar_agent.routes import render_calendar_page # Calendar Agent import
# Import the Historical Data Agent
from metric_agents import metric_team, historical_data_agent
# Import the UI modules
from src.ui.notifications import get_notification_system
from src.utils.config import MARKET_INTELLIGENCE_CONFIG
# Import the new UI modules for market alerts if available
try:
from src.ui.market_alerts_sidebar import initialize_sidebar_alerts
except ImportError:
log.warning("Market alerts sidebar module not available")
# --- Agent initialization in Session State ---
if 'agent' not in st.session_state:
log.info("Initializing AIAgent...")
st.session_state.agent = AIAgent()
log.info("AIAgent initialized and stored in Session State.")
# Initialize the Market Intelligence Suite
if 'metric_team_agent' not in st.session_state:
log.info("Initializing Market Intelligence Suite...")
# Print existing team members to debug
if hasattr(metric_team, 'members'):
existing_members = [m.name if hasattr(m, 'name') else str(m) for m in metric_team.members]
log.info(f"Market Intelligence Suite existing team members: {existing_members}")
# Store the team instance directly
st.session_state.metric_team_agent = metric_team
log.info("Market Intelligence Suite initialized and stored in Session State.")
# Import and log forex agents, but don't add them directly here since they're added in metric_agents.py
try:
from src.ai.forex_agents import (
FXSeerLSTMAgent,
TrendSightTAAgent,
MomentumMoverAgent,
MeanReversionScoutAgent
)
log.info("✅ Successfully imported forex trading agents")
# Create agents for logging purposes
try:
# Create the agents - they directly inherit from Agent class
fxseer_agent = FXSeerLSTMAgent()
log.info(f"✅ Created FXSeer LSTM Agent: {fxseer_agent.name if hasattr(fxseer_agent, 'name') else 'No name'}")
trendsight_agent = TrendSightTAAgent()
log.info(f"✅ Created TrendSight TA Agent: {trendsight_agent.name if hasattr(trendsight_agent, 'name') else 'No name'}")
momentum_agent = MomentumMoverAgent()
log.info(f"✅ Created Momentum Mover Agent: {momentum_agent.name if hasattr(momentum_agent, 'name') else 'No name'}")
mean_reversion_agent = MeanReversionScoutAgent()
log.info(f"✅ Created Mean Reversion Scout Agent: {mean_reversion_agent.name if hasattr(mean_reversion_agent, 'name') else 'No name'}")
# Note: We don't add them to metric_team.members here since they're already added in metric_agents.py
# Print team members for debugging
if hasattr(metric_team, 'members'):
member_names = [m.name if hasattr(m, 'name') else str(m) for m in metric_team.members]
log.info(f"Market Intelligence Suite team members: {member_names}")
# Print coordinator instructions
if hasattr(metric_team, 'instructions'):
log.info(f"Coordinator instructions excerpt: {metric_team.instructions[:3] if metric_team.instructions else 'None'}")
except Exception as e:
log.error(f"❌ Error creating forex trading agents: {e}", exc_info=True)
except ImportError as e:
log.error(f"❌ Could not import forex trading agents: {e}", exc_info=True)
# Initialize the notification system
notification_system = get_notification_system()
# Initialize customizable market alerts if available
try:
from src.ai.customizable_market_alerts import get_customizable_alerts
customizable_alerts = get_customizable_alerts()
# Initialize the market alerts sidebar
if 'ai_agent' in st.session_state and 'market_alerts_agent' not in st.session_state:
try:
# Get the market alerts agent from the AI agent to connect to the customizable alerts
market_alerts_agent = st.session_state.ai_agent.get_specialized_agent("market_alerts")
if market_alerts_agent and hasattr(market_alerts_agent, 'movement_detector'):
# Store the agent for reference
st.session_state.market_alerts_agent = market_alerts_agent
# Connect the agent's data to our customizable alerts
# This allows the customizable alerts to use the AI agent's data sources
if hasattr(market_alerts_agent, 'data_store'):
customizable_alerts.data_store = market_alerts_agent.data_store
# Set up the AI agent's detector to show notifications
from src.ui.notifications import add_market_alert_notification
market_alerts_agent.movement_detector.set_notification_callback(
lambda alert_type, alert_data: add_market_alert_notification(alert_data)
)
log.info("Market alerts agent initialized for sidebar integration")
except Exception as e:
log.error(f"Error initializing market alerts for sidebar: {e}")
# Initialize partial sidebar for market alerts
if 'initialize_sidebar_alerts' in locals():
# Don't use the full initialize_sidebar_alerts() function
if "sidebar_alert_manager" not in st.session_state:
from src.ui.market_alerts_sidebar import SidebarAlertManager
st.session_state.sidebar_alert_manager = SidebarAlertManager()
# Create a simple sidebar structure with only what we want
# 1. Render the alert notifications
if "sidebar_alert_manager" in st.session_state:
st.session_state.sidebar_alert_manager.render_alerts()
# 2. Add monitor controls (which includes currency pair monitoring)
if "sidebar_alert_manager" in st.session_state:
st.session_state.sidebar_alert_manager.render_monitor_controls()
except ImportError:
log.warning("Customizable market alerts module not available")
# Add the notification center to the sidebar
notification_system.render_notification_center()
# Chat history for the Market Intelligence Suite
if 'chat_history_team' not in st.session_state:
st.session_state.chat_history_team = []
# Initialize chat history for Historical Data Agent
if 'chat_history_historical' not in st.session_state:
st.session_state.chat_history_historical = []
# Function to create suggestion cards
def render_suggestion_cards():
"""Create clickable suggestion cards organized into rows"""
# Create suggestions based on the available agents
market_news_suggestions = [
"What are the latest news affecting EUR/USD?",
"Summarize today's forex market news and impact",
"Analyze recent central bank statements and their effect on forex",
"What are today's key forex market developments?"
]
market_alerts_suggestions = [
"Show me recent EUR/USD price breakouts",
"What significant forex movements have happened today?",
"Has there been unusual volatility in USD/JPY in the past 24 hours?",
"Which forex pairs have approached key support/resistance levels today?"
]
historical_data_suggestions = [
"What was the EUR/USD exchange rate on January 20, 2023?",
"Generate a chart of USD/JPY for the first quarter of 2023",
"Compare EUR/USD performance against GBP/USD in 2022",
"What major forex events impacted the markets last month?"
]
analytics_suggestions = [
"Analyze the correlation between EUR/USD and gold prices",
"Identify seasonal patterns in EUR/USD",
"How have interest rates affected currency pairs in 2023?",
"Compare volatility between major and minor currency pairs"
]
# New suggestion categories for forex trading agents
forex_prediction_suggestions = [
"Predict EUR/USD price movement for the next 24 hours",
"What's the 3-day forecast for GBP/JPY?",
"Which major currency pairs will likely strengthen against USD this week?",
"Generate a price prediction for USD/CAD for tomorrow"
]
technical_analysis_suggestions = [
"Analyze USD/JPY with RSI, MACD and moving averages",
"What do technical indicators show for EUR/USD right now?",
"Perform a comprehensive technical analysis of GBP/USD",
"Identify key support and resistance levels for AUD/USD"
]
momentum_trading_suggestions = [
"Which currency pairs have the strongest bullish momentum today?",
"Identify emerging breakout opportunities across major forex pairs",
"Rank the top 3 currency pairs showing unusually strong directional movement",
"Which currency pair is showing the most promising upside momentum right now?"
]
mean_reversion_suggestions = [
"Is USD/JPY currently overbought or oversold?",
"Find mean reversion opportunities in the major currency pairs",
"Which currency pairs are furthest from their moving averages?",
"Identify pairs that have deviated significantly from historical means"
]
st.markdown("### Suggested Questions")
st.markdown("Choose a suggestion to use in your next query")
# Function to render a category of suggestions
def render_suggestion_category(title, suggestions, col):
with col:
st.markdown(f"**{title}**")
for suggestion in suggestions:
cols = st.columns([4, 1])
# Copy button (larger)
with cols[0]:
if st.button(
suggestion,
key=f"copy_{title}_{suggestion}",
use_container_width=True
):
st.session_state.selected_suggestion = suggestion
st.session_state.execute_suggestion = False
# Execute button (smaller)
with cols[1]:
if st.button(
"▶",
key=f"run_{title}_{suggestion}",
use_container_width=True,
help="Run this question immediately"
):
st.session_state.selected_suggestion = suggestion
st.session_state.execute_suggestion = True
# First row of suggestions
col1, col2, col3, col4 = st.columns(4)
render_suggestion_category("Market News", market_news_suggestions, col1)
render_suggestion_category("Market Alerts", market_alerts_suggestions, col2)
render_suggestion_category("Historical Data", historical_data_suggestions, col3)
render_suggestion_category("Analytics", analytics_suggestions, col4)
# Second row of suggestions for forex trading agents
st.markdown("---")
st.markdown("### Forex Trading Intelligence")
col5, col6, col7, col8 = st.columns(4)
render_suggestion_category("Forex Predictions", forex_prediction_suggestions, col5)
render_suggestion_category("Technical Analysis", technical_analysis_suggestions, col6)
render_suggestion_category("Momentum Trading", momentum_trading_suggestions, col7)
render_suggestion_category("Mean Reversion", mean_reversion_suggestions, col8)
# --- UI Rendering ---
# Display active notifications at the top of the app, not in sidebar
notification_system.render_notifications()
st.title("ActionForex Data Solution 🤖")
st.markdown("Ask questions about market data, historical information, and specialized metrics.")
st.markdown("---")
# Tabs for different functions - making Market Intelligence Suite the first tab
tab1, tab2, tab3 = st.tabs([
"📈 Market Intelligence Suite",
"📅 Calendar",
"📜 Historical Data"
])
# TAB 1: MARKET INTELLIGENCE SUITE
with tab1:
st.header("Market Intelligence Suite")
# Container order: description at top, chat display and input above suggestions
description_container = st.container()
chat_container = st.container()
input_container = st.container()
# 1. Description (at the top)
with description_container:
st.markdown("This premium analytics suite brings you real-time market insights by connecting specialized financial experts to analyze trends, volatility, pivot points, and technical indicators.")
# IMPORTANT: Make sure that the team_response_data object does NOT exist in the global scope
# Definition of a dedicated key for the current response variable in session_state
# With each page refresh, this value is explicitly reset
if 'current_team_response' in st.session_state:
# Delete the variable to prevent unwanted debug display
st.session_state.current_team_response = None
# Initialize selected suggestion in session state if not present
if 'selected_suggestion' not in st.session_state:
st.session_state.selected_suggestion = ""
# 2. Display chat history
with chat_container:
# Display chat history for the Market Intelligence Suite
for message in st.session_state.chat_history_team:
with st.chat_message(message["role"]):
# Use display_data to show content
display_data(message["content"])
# 3. Chat input
with input_container:
# Initialize execute flag if it doesn't exist
if 'execute_suggestion' not in st.session_state:
st.session_state.execute_suggestion = False
# Handle the case where a suggestion has been selected
if st.session_state.selected_suggestion:
# Check if we should execute immediately (double-click case)
if st.session_state.execute_suggestion:
team_prompt = st.session_state.selected_suggestion
st.session_state.selected_suggestion = ""
st.session_state.execute_suggestion = False
else:
# Single click - just copy to input
st.success(f"Selected: {st.session_state.selected_suggestion}")
# Display the suggestion above the input field
st.info(f"Question: {st.session_state.selected_suggestion}")
# Add a button to use the suggestion directly
if st.button("Send this question", key="use_suggestion"):
team_prompt = st.session_state.selected_suggestion
st.session_state.selected_suggestion = ""
else:
# Since chat_input doesn't support value parameter, just show the input field
team_prompt = st.chat_input("Your question for the Market Intelligence Suite...", key="team_input")
else:
team_prompt = st.chat_input("Your question for the Market Intelligence Suite...", key="team_input")
# Process the user prompt and display the response
if team_prompt:
# Add user message to team history and display
st.session_state.chat_history_team.append({"role": "user", "content": team_prompt})
with chat_container.chat_message("user"):
st.markdown(team_prompt)
# Generate and display suite response
with chat_container.chat_message("assistant"):
team_full_response_str = ""
try:
with st.spinner("Market Intelligence Suite analyzing..."):
log.info(f"Processing user input for Market Intelligence Suite: {team_prompt}")
# Make sure the coordinator is initialized
if 'metric_team_agent' not in st.session_state or st.session_state.metric_team_agent is None:
team_full_response_str = "Error: Market Intelligence Suite was not properly initialized."
log.error("Market Intelligence Suite not initialized.")
content_to_display = team_full_response_str
else:
# Call the coordinator to handle routing based on its instructions
log.info("Routing request to Market Intelligence Suite coordinator")
response_data = st.session_state.metric_team_agent.run(team_prompt)
# Extract the content directly
if hasattr(response_data, 'content'):
content_to_display = response_data.content
team_full_response_str = str(response_data.content)
else:
log.warning("TeamRunResponse has no .content attribute.")
content_to_display = str(response_data)
team_full_response_str = content_to_display
# Display the answer directly with display_data
display_data(content_to_display)
# Important: Make sure no reference to the TeamRunResponse object remains
response_data = None
except Exception as e:
log.error(f"Error in Market Intelligence Suite tab while processing request: {e}", exc_info=True)
team_full_response_str = f"Sorry, an internal error occurred: {str(e)}"
st.error(team_full_response_str)
# Add response to history
if team_full_response_str:
st.session_state.chat_history_team.append({"role": "assistant", "content": team_full_response_str})
# 4. Suggestion cards (below the chat interaction section)
suggestion_container = st.container()
with suggestion_container:
st.markdown("---") # Add a separator
render_suggestion_cards()
# TAB 2: CALENDAR (moved from third position)
with tab2:
st.header("Calendar Agent")
# Render the calendar page in the second tab
render_calendar_page()
# TAB 3: HISTORICAL DATA (moved from second position)
with tab3:
st.header("Historical Data Agent")
st.markdown("This agent can search and analyze historical data, including past forex rates, events, and market analyses.")
# Use the current system date
current_date = datetime.now().strftime("%B %d, %Y")
st.info(f"Current date: {current_date}")
# Initialize historical suggestion in session state if not present
if 'historical_suggestion' not in st.session_state:
st.session_state.historical_suggestion = ""
# Display selected suggestion if available
if st.session_state.historical_suggestion:
st.success(f"Selected suggestion: {st.session_state.historical_suggestion}")
# Add a button to use the suggestion directly
if st.button("Ask this question", key="use_historical_suggestion"):
prompt = st.session_state.historical_suggestion
st.session_state.historical_suggestion = ""
else:
prompt = st.chat_input("Your question for the Historical Data Agent...")
else:
# Chat input
prompt = st.chat_input("Your question for the Historical Data Agent...")
# Display chat history
for message in st.session_state.chat_history_historical:
with st.chat_message(message["role"]):
display_data(message["content"])
if prompt:
# Add user message to history and display
st.session_state.chat_history_historical.append({"role": "user", "content": prompt})
with st.chat_message("user"):
st.markdown(prompt)
# Generate and display agent response
with st.chat_message("assistant"):
full_response_str = "" # For history
try:
with st.spinner("Historical Data Agent searching..."):
log.info(f"Processing user input for Historical Data Agent: {prompt}")
if historical_data_agent is None:
full_response_str = "Error: Historical Data Agent was not properly initialized."
log.error("Historical Data Agent not initialized.")
agent_response_data = full_response_str # Show error
else:
# Add current date context to the prompt
enhanced_prompt = f"Today's date is {current_date}. Answer based on this current date. {prompt}"
log.info(f"Enhanced prompt with current date: {enhanced_prompt}")
# Use the historical_data_agent directly with the enhanced prompt
response = historical_data_agent.run(enhanced_prompt)
# Extract the content
if hasattr(response, 'content'):
agent_response_data = response.content
full_response_str = str(response.content)
else:
log.warning("AgentRunResponse has no .content attribute.")
agent_response_data = str(response)
full_response_str = agent_response_data
# Display the answer directly with display_data
display_data(agent_response_data)
except Exception as e:
log.error(f"Error in Historical Data Tab while processing request: {e}", exc_info=True)
full_response_str = f"Sorry, an internal error occurred: {str(e)}"
st.error(full_response_str) # Show error directly
agent_response_data = None # Reset so nothing is added to history
# Add agent response to history (string representation)
if full_response_str:
st.session_state.chat_history_historical.append({"role": "assistant", "content": full_response_str})
# Add suggestion cards for Historical Data Agent
st.markdown("### Historical Data Suggestions")
st.markdown("Click any suggestion to explore forex historical data")
# Create categories of suggestions
forex_rate_suggestions = [
"What was the EUR/USD exchange rate on January 15, 2023?",
"Show USD/JPY rates for the first week of March 2023",
"What were the high and low for GBP/USD last month?",
"Compare EUR/USD rates before and after last ECB meeting"
]
trend_analysis_suggestions = [
"Show me EUR/USD price trend for Q1 2023",
"Analyze USD/JPY volatility over the past 6 months",
"Compare EUR/USD and GBP/USD trends in 2022",
"Generate quarterly performance chart for major forex pairs in 2023"
]
correlation_suggestions = [
"How did EUR/USD correlate with gold prices in 2022?",
"Analyze the relationship between USD/JPY and US interest rates last year",
"Compare USD/CHF movement with market volatility during 2022",
"Show correlation between EUR/USD and US stock market in past 6 months"
]
event_impact_suggestions = [
"How did ECB rate decisions affect EUR/USD in 2023?",
"Analyze USD/JPY reaction to Bank of Japan policy changes last year",
"What was GBP/USD impact from UK economic data releases in Q1 2023?",
"Show forex market reaction to major Fed announcements in 2022"
]
# Display in columns
col1, col2 = st.columns(2)
with col1:
st.markdown("**Historical Rates**")
for suggestion in forex_rate_suggestions:
if st.button(
suggestion,
key=f"hist_rate_{suggestion}",
use_container_width=True
):
st.session_state.historical_suggestion = suggestion
st.markdown("**Correlation Analysis**")
for suggestion in correlation_suggestions:
if st.button(
suggestion,
key=f"hist_corr_{suggestion}",
use_container_width=True
):
st.session_state.historical_suggestion = suggestion
with col2:
st.markdown("**Trend Analysis**")
for suggestion in trend_analysis_suggestions:
if st.button(
suggestion,
key=f"hist_trend_{suggestion}",
use_container_width=True
):
st.session_state.historical_suggestion = suggestion
st.markdown("**Event Impact**")
for suggestion in event_impact_suggestions:
if st.button(
suggestion,
key=f"hist_event_{suggestion}",
use_container_width=True
):
st.session_state.historical_suggestion = suggestion
# Simplified main function
def main():
"""Main application entry point."""
# Initialize or retrieve the AI agent
if 'ai_agent' not in st.session_state:
with st.spinner("Initializing AI agent and tools..."):
st.session_state.ai_agent = AIAgent()
# Don't add the title to the sidebar
# st.sidebar.title("ActionForex AI Agent")
# Run the application
if __name__ == "__main__":
main()