RazHadas's picture
Upload 6 files
76317bb verified
# -*- 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