# app.py - The Analyst Hub Dashboard for the V16 Trading Agent import gradio as gr import pandas as pd import plotly.express as px import os import json import firebase_admin from firebase_admin import credentials, db from datetime import datetime import numpy as np # --- Configuration --- # The Firebase DB reference name must match the one used in your agent (App3.txt) DB_REF_NAME = 'signals_v2' # --- Firebase Connection --- def initialize_firebase(): """Initializes the Firebase app, checking for existing initializations.""" try: # Get secrets from environment variables sa_key_json_str = os.environ.get('FIRESTORE_SA_KEY') db_url = os.environ.get('FIREBASE_DB_URL') if not all([sa_key_json_str, db_url]): print("ANALYST HUB: Firebase secrets not set. App will run in offline mode.") return None, "Missing Secrets" sa_key_dict = json.loads(sa_key_json_str) cred = credentials.Certificate(sa_key_dict) # Check if the app is already initialized to prevent errors on hot-reloading if not firebase_admin._apps: firebase_admin.initialize_app(cred, {'databaseURL': db_url}) print("ANALYST HUB: Firebase connection established.") return db.reference(DB_REF_NAME), "Connected" except Exception as e: print(f"ANALYST HUB: CRITICAL ERROR - Failed to initialize Firebase: {e}") return None, f"Connection Error: {e}" # --- Data Fetching and Processing --- def fetch_and_process_data(db_ref): """Fetches data from Firebase and processes it into Pandas DataFrames.""" if db_ref is None: # Return empty structures if DB is not connected return pd.DataFrame(), pd.DataFrame(), {} try: data = db_ref.get() if not data: print("ANALYST HUB: No data found in the database reference.") return pd.DataFrame(), pd.DataFrame(), {} # Convert the dictionary of signals into a list all_signals = [signal for signal in data.values()] df = pd.DataFrame(all_signals) # --- Data Cleaning and Type Conversion --- # Convert timestamps to datetime objects for sorting and plotting df['timestamp_entry'] = pd.to_datetime(df['timestamp_entry']) df['timestamp_exit'] = pd.to_datetime(df['timestamp_exit'], errors='coerce') # Convert numeric columns, coercing errors to NaN numeric_cols = ['entry_price', 'stop_loss', 'take_profit', 'exit_price', 'pnl', 'reward'] for col in numeric_cols: df[col] = pd.to_numeric(df[col], errors='coerce') # Format PnL for display if 'pnl' in df.columns: df['pnl_display'] = df['pnl'].map(lambda x: f"{x:.5f}" if pd.notna(x) else "N/A") # Separate into closed and open trades closed_trades = df[df['pnl'].notna()].sort_values(by='timestamp_entry').copy() open_trades = df[df['pnl'].isna()].sort_values(by='timestamp_entry').copy() # --- Calculate Key Performance Indicators (KPIs) --- kpis = {} if not closed_trades.empty: total_pnl = closed_trades['pnl'].sum() total_trades = len(closed_trades) winning_trades = (closed_trades['pnl'] > 0).sum() win_rate = (winning_trades / total_trades) * 100 if total_trades > 0 else 0 kpis = { "total_pnl": f"{total_pnl:+.5f}", "win_rate": f"{win_rate:.2f}%", "total_closed_trades": str(total_trades), "average_pnl": f"{closed_trades['pnl'].mean():+.5f}" } # Calculate cumulative PnL for the equity curve closed_trades['equity_curve'] = closed_trades['pnl'].cumsum() else: kpis = { "total_pnl": "0.00000", "win_rate": "0.00%", "total_closed_trades": "0", "average_pnl": "0.00000" } closed_trades['equity_curve'] = 0 return closed_trades, open_trades, kpis except Exception as e: print(f"ANALYST HUB: Error processing data: {e}") return pd.DataFrame(), pd.DataFrame(), {} # --- Plotting Functions --- def create_equity_curve(df): """Creates an equity curve chart using Plotly.""" if df.empty or 'equity_curve' not in df.columns: return px.line(title="Equity Curve (No Data)").update_layout(template="plotly_dark") fig = px.line( df, x='timestamp_entry', y='equity_curve', title="Agent Performance: Equity Curve", labels={'timestamp_entry': 'Date', 'equity_curve': 'Cumulative PnL (in pips/points)'} ) fig.update_layout(template="plotly_dark", title_x=0.5, font=dict(color="white")) fig.update_traces(line_color='#00BFFF', fill='tozeroy') # Deep sky blue with fill return fig def create_pnl_analytics_charts(df): """Creates bar charts for PnL breakdown by strategy and regime with distinct profit/loss colors.""" if df.empty: no_data_fig = px.bar(title="PnL by Strategy (No Data)").update_layout(template="plotly_dark") return no_data_fig, no_data_fig # Define the discrete colors for profit (green) and loss (red) color_map = {'Profit': 'darkgreen', 'Loss': 'darkred'} # --- PnL by Strategy Chart --- pnl_by_strategy = df.groupby('strategy_chosen')['pnl'].sum().reset_index() # Create a new column to categorize as 'Profit' or 'Loss' pnl_by_strategy['Outcome'] = np.where(pnl_by_strategy['pnl'] >= 0, 'Profit', 'Loss') fig_strategy = px.bar( pnl_by_strategy, x='strategy_chosen', y='pnl', color='Outcome', # Color bars based on the 'Outcome' column color_discrete_map=color_map, # Apply our specific green/red colors title="PnL by Strategy", labels={'strategy_chosen': 'Strategy', 'pnl': 'Total PnL'} ) # Clean up layout: dark theme, centered title, no redundant legend fig_strategy.update_layout(template="plotly_dark", title_x=0.5, font=dict(color="white"), showlegend=False) # --- PnL by Market Regime Chart --- pnl_by_regime = df.groupby('market_regime')['pnl'].sum().reset_index() # Create the same 'Outcome' column for this chart pnl_by_regime['Outcome'] = np.where(pnl_by_regime['pnl'] >= 0, 'Profit', 'Loss') fig_regime = px.bar( pnl_by_regime, x='market_regime', y='pnl', color='Outcome', # Color bars based on the 'Outcome' column color_discrete_map=color_map, # Apply our specific green/red colors title="PnL by Market Regime", labels={'market_regime': 'Market Regime', 'pnl': 'Total PnL'} ) # Clean up layout fig_regime.update_layout(template="plotly_dark", title_x=0.5, font=dict(color="white"), showlegend=False) return fig_strategy, fig_regime # --- Main Gradio Function --- def update_dashboard(): """Main function to refresh all UI components.""" db_ref, status = initialize_firebase() if status != "Connected": # Handle connection failure gracefully no_data_df = pd.DataFrame() kpis = { "total_pnl": "N/A", "win_rate": "N/A", "total_closed_trades": "N/A", "average_pnl": "N/A" } no_data_plot = px.line(title=f"Database Connection Failed: {status}").update_layout(template="plotly_dark") no_data_bar = px.bar(title=f"Database Connection Failed").update_layout(template="plotly_dark") return (kpis["total_pnl"], kpis["win_rate"], kpis["total_closed_trades"], kpis["average_pnl"], no_data_plot, no_data_df, no_data_df, no_data_bar, no_data_bar, f"DB Status: {status} at {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") closed_df, open_df, kpis = fetch_and_process_data(db_ref) # Create plots equity_plot = create_equity_curve(closed_df) strategy_plot, regime_plot = create_pnl_analytics_charts(closed_df) # Prepare dataframes for display open_trades_display = open_df[[ 'timestamp_entry', 'action', 'strategy_chosen', 'confidence', 'entry_price', 'stop_loss', 'take_profit', 'reasoning_initial' ]].copy() closed_trades_display = closed_df[[ 'timestamp_entry', 'action', 'strategy_chosen', 'outcome_reason', 'pnl_display', 'entry_price', 'exit_price', 'market_regime' ]].copy() # Format datetime for better readability open_trades_display['timestamp_entry'] = open_trades_display['timestamp_entry'].dt.strftime('%Y-%m-%d %H:%M') closed_trades_display['timestamp_entry'] = closed_trades_display['timestamp_entry'].dt.strftime('%Y-%m-%d %H:%M') return (kpis["total_pnl"], kpis["win_rate"], kpis["total_closed_trades"], kpis["average_pnl"], equity_plot, open_trades_display.iloc[::-1], # Show newest first closed_trades_display.iloc[::-1], # Show newest first strategy_plot, regime_plot, f"DB Status: {status} | Last Updated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") # --- Gradio UI Layout --- with gr.Blocks(theme=gr.themes.Glass(), title="Agent Analyst Hub") as demo: gr.Markdown("# 📈 V16 Agent Analyst Hub") gr.Markdown("Live performance monitoring and analysis of the GuardianAI trading agent. Data is fetched directly from the Firebase Realtime Database.") with gr.Row(): db_status_output = gr.Textbox(label="Database Status", interactive=False) refresh_btn = gr.Button("🔄 Refresh Data", variant="primary") gr.Markdown("## Key Performance Indicators (KPIs)") with gr.Row(): pnl_output = gr.Textbox(label="Total Net PnL", interactive=False) win_rate_output = gr.Textbox(label="Win Rate", interactive=False) total_trades_output = gr.Textbox(label="Total Closed Trades", interactive=False) avg_pnl_output = gr.Textbox(label="Average PnL / Trade", interactive=False) with gr.Tabs(): with gr.TabItem("📊 Dashboard"): equity_curve_plot = gr.Plot() with gr.TabItem("🔍 Performance Analytics"): gr.Markdown("### PnL Breakdown Analysis") with gr.Row(): pnl_by_strategy_plot = gr.Plot() pnl_by_regime_plot = gr.Plot() with gr.TabItem("📖 Open Trades"): open_trades_df = gr.DataFrame(headers=[ 'Entry Time', 'Action', 'Strategy', 'Confidence', 'Entry Price', 'Stop Loss', 'Take Profit', 'Initial Reasoning' ], interactive=False) with gr.TabItem("📚 Trade History"): closed_trades_df = gr.DataFrame(headers=[ 'Entry Time', 'Action', 'Strategy', 'Outcome', 'PnL', 'Entry Price', 'Exit Price', 'Market Regime' ], interactive=False) # --- Component Connections --- outputs_list = [ pnl_output, win_rate_output, total_trades_output, avg_pnl_output, equity_curve_plot, open_trades_df, closed_trades_df, pnl_by_strategy_plot, pnl_by_regime_plot, db_status_output ] # Refresh data on button click refresh_btn.click(fn=update_dashboard, inputs=[], outputs=outputs_list) # Load data when the app starts demo.load(fn=update_dashboard, inputs=[], outputs=outputs_list) if __name__ == "__main__": # Check if secrets are present before launching if not all([os.environ.get(k) for k in ['FIRESTORE_SA_KEY', 'FIREBASE_DB_URL']]): print("\nWARNING: Firebase secrets are not set as environment variables.") print("The dashboard will load but will not be able to connect to the database.") print("Please set FIRESTORE_SA_KEY and FIREBASE_DB_URL in your environment.\n") demo.launch()