# -*- coding: utf-8 -*- """app.py Automatically generated by Colab. Original file is located at https://colab.research.google.com/drive/18CPi10QPKtnp8wBs3Fd21JjaDxoHytAM """ # app.py # Main Gradio application script for QuantConnect Report Enhancer. import gradio as gr import pandas as pd import numpy as np import traceback # Import functions from other modules from utils import create_empty_figure from processing import process_single_file from risk_analysis import calculate_correlation, calculate_manual_risk_stats from plotting import generate_figures_for_strategy, generate_manual_risk_figures # --- Constants for UI --- DEFAULT_TRADES_COLS_DISPLAY = [ 'symbol', 'entryTime', 'exitTime', 'direction', 'quantity', 'entryPrice', 'exitPrice', 'profitLoss', 'totalFees', 'duration_days' ] MAX_TRADES_DISPLAY = 50 # Limit number of trades shown in the table # --- Gradio Interface Callbacks --- def process_files_and_update_ui(uploaded_files): """ Callback function triggered when files are uploaded. Processes each file, calculates overall metrics (like correlation), updates the application state, and populates the UI with the first strategy's details. Args: uploaded_files: A list of file objects uploaded via the Gradio interface. Returns: A tuple containing updated values for all relevant Gradio components: - Status message (Textbox) - Strategy dropdown (Dropdown) - updated choices, value, visibility - Application state (State) - dictionary holding all processed results - Outputs for individual strategy tabs (DataFrames, Plots) - Outputs for correlation tab (DataFrame, Plot) - Outputs for manual risk analysis tab (DataFrames, Plots) """ # --- Initialize Default/Empty Outputs --- # Create empty figures and dataframes to return if processing fails or no files uploaded default_stats_df = pd.DataFrame(columns=['Metric', 'Value']) default_trades_df_display = pd.DataFrame() default_equity_fig = create_empty_figure("Equity Curve") default_drawdown_fig = create_empty_figure("Drawdown Curve") default_benchmark_fig = create_empty_figure("Equity vs Benchmark") default_pnl_hist_fig = create_empty_figure("P/L Distribution") default_duration_hist_fig = create_empty_figure("Trade Duration Distribution") default_exposure_fig = create_empty_figure("Exposure") default_turnover_fig = create_empty_figure("Portfolio Turnover") default_corr_matrix = pd.DataFrame() default_corr_heatmap = create_empty_figure("Correlation Heatmap") default_monthly_table_display = pd.DataFrame() # For the formatted table in UI default_monthly_stats = pd.DataFrame(columns=['Metric', 'Value']) default_monthly_heatmap = create_empty_figure("Monthly Returns Heatmap") default_rolling_vol_stats = pd.DataFrame(columns=['Window', 'Min Vol', 'Max Vol', 'Mean Vol']) default_rolling_vol_plot = create_empty_figure("Rolling Volatility") default_drawdown_table = pd.DataFrame() # Structure default outputs for return statement clarity initial_outputs = [ default_stats_df, default_equity_fig, default_drawdown_fig, default_benchmark_fig, default_pnl_hist_fig, default_duration_hist_fig, default_exposure_fig, default_turnover_fig, default_trades_df_display ] correlation_outputs = [default_corr_matrix, default_corr_heatmap] manual_risk_outputs = [ default_monthly_table_display, default_monthly_stats, default_monthly_heatmap, default_rolling_vol_plot, default_rolling_vol_stats, default_drawdown_table ] # Combine all output lists for the final return all_default_outputs = initial_outputs + correlation_outputs + manual_risk_outputs # --- Handle No Files Uploaded --- if not uploaded_files: return ( "Please upload one or more QuantConnect JSON files.", # Status message gr.Dropdown(choices=[], value=None, visible=False), # Hide dropdown {}, # Empty state *all_default_outputs # Return all default outputs ) # --- Process Uploaded Files --- all_results = {} # Dictionary to store results for each processed file {filename: results_dict} status_messages = [] # List to collect status/error messages processed_files_count = 0 for file_obj in uploaded_files: if file_obj is None: # Skip if file object is somehow None continue try: file_path = file_obj.name # Get the temporary file path from Gradio # Process the single file using the function from processing.py strategy_result = process_single_file(file_path) # Store the result using the filename as the key all_results[strategy_result["filename"]] = strategy_result # Log errors or increment success count if strategy_result["error"]: status_messages.append(strategy_result["error"]) else: processed_files_count += 1 except Exception as e: # Catch unexpected errors during the file processing loop error_msg = f"Failed to process an uploaded file object: {e}" print(error_msg) traceback.print_exc() status_messages.append(error_msg) # --- Handle No Valid Files Processed --- if not all_results or processed_files_count == 0: status = "\n".join(status_messages) if status_messages else "No valid QuantConnect JSON files processed." return ( status, gr.Dropdown(choices=[], value=None, visible=False), # Hide dropdown {}, # Empty state *all_default_outputs ) # --- Calculate Correlation (Across All Processed Files) --- try: corr_matrix_df, corr_heatmap_fig, corr_status = calculate_correlation(all_results) status_messages.append(corr_status) # Add correlation status to messages except Exception as e: print(f"Error during correlation calculation: {e}") traceback.print_exc() status_messages.append(f"Correlation Error: {e}") # Use default correlation outputs on error corr_matrix_df = default_corr_matrix corr_heatmap_fig = default_corr_heatmap # --- Prepare Initial UI Display (Using the First Processed Strategy) --- first_filename = list(all_results.keys())[0] initial_strategy_results = all_results[first_filename] # Generate standard plots for the first strategy try: initial_figures = generate_figures_for_strategy(initial_strategy_results) except Exception as e: print(f"Error generating initial figures for {first_filename}: {e}") initial_figures = {k: create_empty_figure(f"{k.replace('_fig','')} - Error") for k in initial_outputs_map.keys() if k.endswith('_fig')} # Create error figures status_messages.append(f"Plotting Error (Initial): {e}") # Perform manual risk analysis for the first strategy try: initial_manual_risk_analysis = calculate_manual_risk_stats(initial_strategy_results.get("daily_returns")) status_messages.append(f"Risk Analysis ({first_filename}): {initial_manual_risk_analysis['status']}") # Generate risk plots based on the analysis results initial_manual_risk_figures = generate_manual_risk_figures(initial_manual_risk_analysis, first_filename) except Exception as e: print(f"Error during initial manual risk analysis or plotting for {first_filename}: {e}") traceback.print_exc() status_messages.append(f"Risk Analysis/Plot Error (Initial): {e}") # Use default risk outputs on error initial_manual_risk_analysis = { "monthly_returns_table_for_heatmap": None, "monthly_perf_stats": default_monthly_stats, "rolling_vol_df": None, "rolling_vol_stats": default_rolling_vol_stats, "drawdown_table": default_drawdown_table } initial_manual_risk_figures = { "monthly_heatmap_fig": default_monthly_heatmap, "rolling_vol_fig": default_rolling_vol_plot } # --- Prepare DataFrames for Initial Display --- initial_stats_df = initial_strategy_results.get("stats_df", default_stats_df) initial_trades_df = initial_strategy_results.get("trades_df", pd.DataFrame()) # Select and format trades table for display if not initial_trades_df.empty: # Filter columns to display existing_display_cols = [col for col in DEFAULT_TRADES_COLS_DISPLAY if col in initial_trades_df.columns] initial_trades_df_display = initial_trades_df[existing_display_cols].head(MAX_TRADES_DISPLAY) # Handle complex 'symbol' column (often a dictionary in QC output) if 'symbol' in initial_trades_df_display.columns: # Check if the first non-null symbol is a dict first_symbol = initial_trades_df_display['symbol'].dropna().iloc[0] if not initial_trades_df_display['symbol'].dropna().empty else None if isinstance(first_symbol, dict): # Apply function to extract 'value' or 'ticker' if it's a dict, otherwise keep original initial_trades_df_display.loc[:, 'symbol'] = initial_trades_df_display['symbol'].apply( lambda x: x.get('value', x.get('ticker', str(x))) if isinstance(x, dict) else x ) # Convert datetime columns to string for display if needed (Gradio often handles it) for col in ['entryTime', 'exitTime']: if col in initial_trades_df_display.columns and pd.api.types.is_datetime64_any_dtype(initial_trades_df_display[col]): initial_trades_df_display[col] = initial_trades_df_display[col].dt.strftime('%Y-%m-%d %H:%M:%S') else: initial_trades_df_display = default_trades_df_display # Prepare formatted monthly returns table for UI display formatted_monthly_table = default_monthly_table_display heatmap_data = initial_manual_risk_analysis.get("monthly_returns_table_for_heatmap") if heatmap_data is not None and not heatmap_data.empty: df_display = heatmap_data.copy() # Work on a copy # Format values as percentages (e.g., "1.23%") df_display = df_display.applymap(lambda x: f'{x:.2f}%' if pd.notna(x) else '') # Reset index to make 'Year' a regular column for Gradio DataFrame display formatted_monthly_table = df_display.reset_index() # --- Consolidate Status Message --- final_status = "\n".join(s for s in status_messages if s).strip() if not final_status: final_status = f"Successfully processed {processed_files_count} file(s)." # --- Assemble Final Outputs --- outputs_to_return = [ final_status, # Status Textbox gr.Dropdown( # Strategy Dropdown choices=list(all_results.keys()), # Update choices value=first_filename, # Set initial value visible=True, # Make visible label="Select Strategy to View", interactive=True ), all_results, # Update the hidden state # --- Individual Strategy Tab Outputs --- initial_stats_df, initial_figures.get("equity_fig", default_equity_fig), initial_figures.get("drawdown_fig", default_drawdown_fig), initial_figures.get("benchmark_fig", default_benchmark_fig), initial_figures.get("pnl_hist_fig", default_pnl_hist_fig), initial_figures.get("duration_hist_fig", default_duration_hist_fig), initial_figures.get("exposure_fig", default_exposure_fig), initial_figures.get("turnover_fig", default_turnover_fig), initial_trades_df_display, # --- Correlation Tab Outputs --- corr_matrix_df, corr_heatmap_fig, # --- Manual Risk Tab Outputs --- formatted_monthly_table, # Use the formatted table for display initial_manual_risk_analysis.get("monthly_perf_stats", default_monthly_stats), initial_manual_risk_figures.get("monthly_heatmap_fig", default_monthly_heatmap), initial_manual_risk_figures.get("rolling_vol_fig", default_rolling_vol_plot), initial_manual_risk_analysis.get("rolling_vol_stats", default_rolling_vol_stats), initial_manual_risk_analysis.get("drawdown_table", default_drawdown_table) ] return tuple(outputs_to_return) def display_selected_strategy(selected_filename, all_results_state): """ Callback function triggered when a strategy is selected from the dropdown. Retrieves the data for the selected strategy from the state and updates the individual strategy tabs and the manual risk analysis tab accordingly. Args: selected_filename: The filename of the strategy selected in the dropdown. all_results_state: The current state dictionary containing all processed results. Returns: A tuple containing updated values for the Gradio components related to the selected strategy's details (Overview, Performance, Trade Analysis, Other Charts, Risk Analysis tabs). Correlation tab is not updated here. """ # --- Initialize Default/Empty Outputs --- # (Same defaults as in process_files_and_update_ui for the relevant outputs) default_stats_df = pd.DataFrame(columns=['Metric', 'Value']) default_trades_df_display = pd.DataFrame() default_equity_fig = create_empty_figure("Equity Curve") default_drawdown_fig = create_empty_figure("Drawdown Curve") default_benchmark_fig = create_empty_figure("Equity vs Benchmark") default_pnl_hist_fig = create_empty_figure("P/L Distribution") default_duration_hist_fig = create_empty_figure("Trade Duration Distribution") default_exposure_fig = create_empty_figure("Exposure") default_turnover_fig = create_empty_figure("Portfolio Turnover") default_monthly_table_display = pd.DataFrame() default_monthly_stats = pd.DataFrame(columns=['Metric', 'Value']) default_monthly_heatmap = create_empty_figure("Monthly Returns Heatmap") default_rolling_vol_stats = pd.DataFrame(columns=['Window', 'Min Vol', 'Max Vol', 'Mean Vol']) default_rolling_vol_plot = create_empty_figure("Rolling Volatility") default_drawdown_table = pd.DataFrame() # Structure default outputs for return statement clarity initial_outputs = [ default_stats_df, default_equity_fig, default_drawdown_fig, default_benchmark_fig, default_pnl_hist_fig, default_duration_hist_fig, default_exposure_fig, default_turnover_fig, default_trades_df_display ] manual_risk_outputs = [ default_monthly_table_display, default_monthly_stats, default_monthly_heatmap, default_rolling_vol_plot, default_rolling_vol_stats, default_drawdown_table ] all_default_outputs = initial_outputs + manual_risk_outputs # --- Validate Selection and State --- if not selected_filename or not all_results_state or selected_filename not in all_results_state: print(f"Warning: Invalid selection ('{selected_filename}') or state. Returning defaults.") # Potentially add a status message update here if you have a dedicated status output for selection changes return tuple(all_default_outputs) # --- Retrieve Selected Strategy Data --- strategy_results = all_results_state[selected_filename] # --- Handle Case Where Selected Strategy Had Processing Errors --- if strategy_results.get("error"): print(f"Displaying error state for {selected_filename}: {strategy_results['error']}") # Show the error in the statistics table and clear other plots/tables error_df = pd.DataFrame([{"Metric": "Error", "Value": strategy_results['error']}]) error_outputs = [error_df] + [ # Use error df for stats table create_empty_figure(f"{fig_name} - Error") for fig_name in [ # Create empty error figures "Equity", "Drawdown", "Benchmark", "P/L", "Duration", "Exposure", "Turnover" ] ] + [default_trades_df_display] # Empty trades table error_risk_outputs = [ # Empty risk outputs default_monthly_table_display, default_monthly_stats, create_empty_figure("Monthly Heatmap - Error"), create_empty_figure("Rolling Vol - Error"), default_rolling_vol_stats, default_drawdown_table ] return tuple(error_outputs + error_risk_outputs) # --- Generate Figures and Analysis for Selected Strategy --- # Generate standard plots try: figures = generate_figures_for_strategy(strategy_results) except Exception as e: print(f"Error generating figures for {selected_filename}: {e}") figures = {k: create_empty_figure(f"{k.replace('_fig','')} - Error") for k in initial_outputs_map.keys() if k.endswith('_fig')} # Perform manual risk analysis try: manual_risk_analysis = calculate_manual_risk_stats(strategy_results.get("daily_returns")) # Generate risk plots manual_risk_figures = generate_manual_risk_figures(manual_risk_analysis, selected_filename) except Exception as e: print(f"Error during manual risk analysis or plotting for {selected_filename}: {e}") traceback.print_exc() # Use default risk outputs on error manual_risk_analysis = { "monthly_returns_table_for_heatmap": None, "monthly_perf_stats": default_monthly_stats, "rolling_vol_df": None, "rolling_vol_stats": default_rolling_vol_stats, "drawdown_table": default_drawdown_table } manual_risk_figures = { "monthly_heatmap_fig": default_monthly_heatmap, "rolling_vol_fig": default_rolling_vol_plot } # --- Prepare DataFrames for Display --- stats_df = strategy_results.get("stats_df", default_stats_df) trades_df = strategy_results.get("trades_df", pd.DataFrame()) # Select and format trades table if not trades_df.empty: existing_display_cols = [col for col in DEFAULT_TRADES_COLS_DISPLAY if col in trades_df.columns] trades_df_display = trades_df[existing_display_cols].head(MAX_TRADES_DISPLAY) if 'symbol' in trades_df_display.columns: first_symbol = trades_df_display['symbol'].dropna().iloc[0] if not trades_df_display['symbol'].dropna().empty else None if isinstance(first_symbol, dict): trades_df_display.loc[:, 'symbol'] = trades_df_display['symbol'].apply( lambda x: x.get('value', x.get('ticker', str(x))) if isinstance(x, dict) else x ) # Convert datetime columns to string for display for col in ['entryTime', 'exitTime']: if col in trades_df_display.columns and pd.api.types.is_datetime64_any_dtype(trades_df_display[col]): trades_df_display[col] = trades_df_display[col].dt.strftime('%Y-%m-%d %H:%M:%S') else: trades_df_display = default_trades_df_display # Prepare formatted monthly returns table formatted_monthly_table = default_monthly_table_display heatmap_data = manual_risk_analysis.get("monthly_returns_table_for_heatmap") if heatmap_data is not None and not heatmap_data.empty: df_display = heatmap_data.copy() df_display = df_display.applymap(lambda x: f'{x:.2f}%' if pd.notna(x) else '') formatted_monthly_table = df_display.reset_index() # --- Assemble Outputs for Return --- # Return components for the tabs updated by the dropdown selection outputs_to_return = [ # --- Individual Strategy Tab Outputs --- stats_df, figures.get("equity_fig", default_equity_fig), figures.get("drawdown_fig", default_drawdown_fig), figures.get("benchmark_fig", default_benchmark_fig), figures.get("pnl_hist_fig", default_pnl_hist_fig), figures.get("duration_hist_fig", default_duration_hist_fig), figures.get("exposure_fig", default_exposure_fig), figures.get("turnover_fig", default_turnover_fig), trades_df_display, # --- Manual Risk Tab Outputs --- formatted_monthly_table, # Use formatted table manual_risk_analysis.get("monthly_perf_stats", default_monthly_stats), manual_risk_figures.get("monthly_heatmap_fig", default_monthly_heatmap), manual_risk_figures.get("rolling_vol_fig", default_rolling_vol_plot), manual_risk_analysis.get("rolling_vol_stats", default_rolling_vol_stats), manual_risk_analysis.get("drawdown_table", default_drawdown_table) ] return tuple(outputs_to_return) # --- Build Gradio Interface --- with gr.Blocks(theme=gr.themes.Soft()) as iface: gr.Markdown("# Trading Platform Report Enhancer") gr.Markdown("Upload one or more QuantConnect backtest JSON files to generate analysis reports and compare strategies.") # Hidden state to store all processed results between interactions all_results_state = gr.State({}) # --- Row 1: File Upload --- with gr.Row(): file_input = gr.File( label="Upload QuantConnect JSON File(s)", file_count="multiple", # Allow multiple files file_types=['.json'] # Restrict to JSON files ) # --- Row 2: Status Output --- with gr.Row(): status_output = gr.Textbox(label="Processing Status", interactive=False, lines=2) # Reduced lines # --- Row 3: Strategy Selection Dropdown --- with gr.Row(): strategy_dropdown = gr.Dropdown( label="Select Strategy to View", choices=[], # Initially empty, populated after file processing visible=False, # Initially hidden interactive=True # User can interact with it ) # --- Tabs for Different Analysis Views --- with gr.Tabs(): # --- Tab 1: Overview --- with gr.TabItem("📊 Overview"): with gr.Column(): gr.Markdown("## Key Performance Metrics") stats_output = gr.DataFrame(label="Overall Statistics", interactive=False, wrap=True) # --- Tab 2: Performance Charts --- with gr.TabItem("📈 Performance Charts"): with gr.Column(): gr.Markdown("## Equity & Drawdown") with gr.Row(): plot_equity = gr.Plot(label="Equity Curve") plot_drawdown = gr.Plot(label="Drawdown Curve") gr.Markdown("## Benchmark Comparison") plot_benchmark = gr.Plot(label="Equity vs Benchmark (Normalized)") # Clarified title # --- Tab 3: Trade Analysis --- with gr.TabItem("💹 Trade Analysis"): with gr.Column(): gr.Markdown("## Profit/Loss and Duration") with gr.Row(): plot_pnl_hist = gr.Plot(label="P/L Distribution") plot_duration_hist = gr.Plot(label="Trade Duration Distribution (Days)") gr.Markdown(f"## Closed Trades (Sample - First {MAX_TRADES_DISPLAY})") # Dynamic title trades_output = gr.DataFrame(label="Closed Trades Sample", interactive=False, wrap=True) # --- Tab 4: Other Charts --- with gr.TabItem("⚙️ Other Charts"): with gr.Column(): gr.Markdown("## Exposure & Turnover") with gr.Row(): plot_exposure = gr.Plot(label="Exposure") plot_turnover = gr.Plot(label="Portfolio Turnover") # --- Tab 5: Risk Analysis (Manual Calculations) --- with gr.TabItem("🔎 Risk Analysis"): with gr.Column(): gr.Markdown("## Monthly Performance") plot_monthly_heatmap = gr.Plot(label="Monthly Returns Heatmap") # Use specific names matching callback outputs monthly_returns_table_output = gr.DataFrame(label="Monthly Returns (%) Table", interactive=False, wrap=True) monthly_perf_stats_output = gr.DataFrame(label="Monthly Performance Stats", interactive=False, wrap=True) gr.Markdown("## Rolling Volatility") plot_rolling_vol = gr.Plot(label="Annualized Rolling Volatility") rolling_vol_stats_output = gr.DataFrame(label="Rolling Volatility Stats", interactive=False, wrap=True) gr.Markdown("## Drawdown Analysis") drawdown_table_output = gr.DataFrame(label=f"Top {5} Drawdown Periods", interactive=False, wrap=True) # Can make 'top' dynamic if needed # --- Tab 6: Correlation --- with gr.TabItem("🤝 Correlation"): with gr.Column(): gr.Markdown("## Strategy (+Benchmark) Correlation") gr.Markdown("_Based on daily equity percentage change._") # Subtitle explanation corr_heatmap_output = gr.Plot(label="Correlation Heatmap") corr_matrix_output = gr.DataFrame(label="Correlation Matrix", interactive=False, wrap=True) # --- Define Output Lists for Callbacks --- # Outputs updated by file upload (all tabs + state + dropdown) individual_report_outputs = [ stats_output, plot_equity, plot_drawdown, plot_benchmark, plot_pnl_hist, plot_duration_hist, plot_exposure, plot_turnover, trades_output ] manual_risk_tab_outputs = [ # Renamed for clarity monthly_returns_table_output, monthly_perf_stats_output, plot_monthly_heatmap, plot_rolling_vol, rolling_vol_stats_output, drawdown_table_output ] correlation_tab_outputs = [corr_matrix_output, corr_heatmap_output] file_processing_outputs = [status_output, strategy_dropdown, all_results_state] # Combine ALL outputs for the file upload callback trigger file_upload_all_outputs = ( file_processing_outputs + individual_report_outputs + correlation_tab_outputs + manual_risk_tab_outputs ) # Outputs updated by dropdown selection (individual strategy tabs + risk tab) dropdown_outputs = individual_report_outputs + manual_risk_tab_outputs # --- Connect Callbacks to Events --- # When files are uploaded (or cleared), trigger file processing file_input.change( fn=process_files_and_update_ui, inputs=[file_input], outputs=file_upload_all_outputs # Pass the combined list ) # When the dropdown value changes, trigger display update strategy_dropdown.change( fn=display_selected_strategy, inputs=[strategy_dropdown, all_results_state], outputs=dropdown_outputs # Pass the relevant outputs list ) # --- Launch the Gradio App --- if __name__ == '__main__': # share=True creates a public link (useful for HF Spaces) # debug=True provides detailed error logs in the console iface.launch(debug=True, share=False) # Set share=True for Hugging Face deployment if needed