| |
| """app.py |
| |
| Automatically generated by Colab. |
| |
| Original file is located at |
| https://colab.research.google.com/drive/18CPi10QPKtnp8wBs3Fd21JjaDxoHytAM |
| """ |
|
|
| |
| |
|
|
| import gradio as gr |
| import pandas as pd |
| import numpy as np |
| import traceback |
|
|
| |
| 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 |
|
|
| |
| DEFAULT_TRADES_COLS_DISPLAY = [ |
| 'symbol', 'entryTime', 'exitTime', 'direction', 'quantity', |
| 'entryPrice', 'exitPrice', 'profitLoss', 'totalFees', 'duration_days' |
| ] |
| MAX_TRADES_DISPLAY = 50 |
|
|
| |
|
|
| 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) |
| """ |
| |
| |
| 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() |
| 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() |
|
|
| |
| 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 |
| ] |
| |
| all_default_outputs = initial_outputs + correlation_outputs + manual_risk_outputs |
|
|
| |
| if not uploaded_files: |
| return ( |
| "Please upload one or more QuantConnect JSON files.", |
| gr.Dropdown(choices=[], value=None, visible=False), |
| {}, |
| *all_default_outputs |
| ) |
|
|
| |
| all_results = {} |
| status_messages = [] |
| processed_files_count = 0 |
|
|
| for file_obj in uploaded_files: |
| if file_obj is None: |
| continue |
| try: |
| file_path = file_obj.name |
| |
| strategy_result = process_single_file(file_path) |
| |
| all_results[strategy_result["filename"]] = strategy_result |
| |
| if strategy_result["error"]: |
| status_messages.append(strategy_result["error"]) |
| else: |
| processed_files_count += 1 |
| except Exception as e: |
| |
| error_msg = f"Failed to process an uploaded file object: {e}" |
| print(error_msg) |
| traceback.print_exc() |
| status_messages.append(error_msg) |
|
|
| |
| 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), |
| {}, |
| *all_default_outputs |
| ) |
|
|
| |
| try: |
| corr_matrix_df, corr_heatmap_fig, corr_status = calculate_correlation(all_results) |
| status_messages.append(corr_status) |
| except Exception as e: |
| print(f"Error during correlation calculation: {e}") |
| traceback.print_exc() |
| status_messages.append(f"Correlation Error: {e}") |
| |
| corr_matrix_df = default_corr_matrix |
| corr_heatmap_fig = default_corr_heatmap |
|
|
| |
| first_filename = list(all_results.keys())[0] |
| initial_strategy_results = all_results[first_filename] |
|
|
| |
| 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')} |
| status_messages.append(f"Plotting Error (Initial): {e}") |
|
|
|
|
| |
| 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']}") |
| |
| 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}") |
| |
| 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 |
| } |
|
|
| |
| initial_stats_df = initial_strategy_results.get("stats_df", default_stats_df) |
| initial_trades_df = initial_strategy_results.get("trades_df", pd.DataFrame()) |
|
|
| |
| if not initial_trades_df.empty: |
| |
| 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) |
| |
| if 'symbol' in initial_trades_df_display.columns: |
| |
| 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): |
| |
| 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 |
| ) |
| |
| 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 |
|
|
| |
| 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() |
| |
| df_display = df_display.applymap(lambda x: f'{x:.2f}%' if pd.notna(x) else '') |
| |
| formatted_monthly_table = df_display.reset_index() |
|
|
|
|
| |
| 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)." |
|
|
| |
| outputs_to_return = [ |
| final_status, |
| gr.Dropdown( |
| choices=list(all_results.keys()), |
| value=first_filename, |
| visible=True, |
| label="Select Strategy to View", |
| interactive=True |
| ), |
| all_results, |
| |
| 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, |
| |
| corr_matrix_df, |
| corr_heatmap_fig, |
| |
| formatted_monthly_table, |
| 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. |
| """ |
| |
| |
| 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() |
|
|
| |
| 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 |
|
|
| |
| 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.") |
| |
| return tuple(all_default_outputs) |
|
|
| |
| strategy_results = all_results_state[selected_filename] |
|
|
| |
| if strategy_results.get("error"): |
| print(f"Displaying error state for {selected_filename}: {strategy_results['error']}") |
| |
| error_df = pd.DataFrame([{"Metric": "Error", "Value": strategy_results['error']}]) |
| error_outputs = [error_df] + [ |
| create_empty_figure(f"{fig_name} - Error") for fig_name in [ |
| "Equity", "Drawdown", "Benchmark", "P/L", "Duration", "Exposure", "Turnover" |
| ] |
| ] + [default_trades_df_display] |
| error_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) |
|
|
|
|
| |
| |
| 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')} |
|
|
|
|
| |
| try: |
| manual_risk_analysis = calculate_manual_risk_stats(strategy_results.get("daily_returns")) |
| |
| 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() |
| |
| 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 |
| } |
|
|
|
|
| |
| stats_df = strategy_results.get("stats_df", default_stats_df) |
| trades_df = strategy_results.get("trades_df", pd.DataFrame()) |
|
|
| |
| 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 |
| ) |
| |
| 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 |
|
|
| |
| 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() |
|
|
|
|
| |
| |
| outputs_to_return = [ |
| |
| 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, |
| |
| formatted_monthly_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) |
|
|
|
|
| |
| 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.") |
|
|
| |
| all_results_state = gr.State({}) |
|
|
| |
| with gr.Row(): |
| file_input = gr.File( |
| label="Upload QuantConnect JSON File(s)", |
| file_count="multiple", |
| file_types=['.json'] |
| ) |
|
|
| |
| with gr.Row(): |
| status_output = gr.Textbox(label="Processing Status", interactive=False, lines=2) |
|
|
| |
| with gr.Row(): |
| strategy_dropdown = gr.Dropdown( |
| label="Select Strategy to View", |
| choices=[], |
| visible=False, |
| interactive=True |
| ) |
|
|
| |
| with gr.Tabs(): |
| |
| with gr.TabItem("π Overview"): |
| with gr.Column(): |
| gr.Markdown("## Key Performance Metrics") |
| stats_output = gr.DataFrame(label="Overall Statistics", interactive=False, wrap=True) |
|
|
| |
| 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)") |
|
|
| |
| 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})") |
| trades_output = gr.DataFrame(label="Closed Trades Sample", interactive=False, wrap=True) |
|
|
| |
| 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") |
|
|
| |
| with gr.TabItem("π Risk Analysis"): |
| with gr.Column(): |
| gr.Markdown("## Monthly Performance") |
| plot_monthly_heatmap = gr.Plot(label="Monthly Returns Heatmap") |
| |
| 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) |
|
|
| |
| with gr.TabItem("π€ Correlation"): |
| with gr.Column(): |
| gr.Markdown("## Strategy (+Benchmark) Correlation") |
| gr.Markdown("_Based on daily equity percentage change._") |
| corr_heatmap_output = gr.Plot(label="Correlation Heatmap") |
| corr_matrix_output = gr.DataFrame(label="Correlation Matrix", interactive=False, wrap=True) |
|
|
|
|
| |
| |
| 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 = [ |
| 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] |
|
|
| |
| file_upload_all_outputs = ( |
| file_processing_outputs + |
| individual_report_outputs + |
| correlation_tab_outputs + |
| manual_risk_tab_outputs |
| ) |
|
|
| |
| dropdown_outputs = individual_report_outputs + manual_risk_tab_outputs |
|
|
| |
| |
| file_input.change( |
| fn=process_files_and_update_ui, |
| inputs=[file_input], |
| outputs=file_upload_all_outputs |
| ) |
|
|
| |
| strategy_dropdown.change( |
| fn=display_selected_strategy, |
| inputs=[strategy_dropdown, all_results_state], |
| outputs=dropdown_outputs |
| ) |
|
|
| |
| if __name__ == '__main__': |
| |
| |
| iface.launch(debug=True, share=False) |