Download plotting.py from RazHadas/Quant_Connect_JSON_analysis: direct link, hf CLI and curl.
- Browser
- Download file 13.9 kB
-
https://huggingface.co/spaces/RazHadas/Quant_Connect_JSON_analysis/resolve/main/plotting.py
- Command line
-
hf download hf://spaces/RazHadas/Quant_Connect_JSON_analysis/plotting.py
-
curl -L -o plotting.py https://huggingface.co/spaces/RazHadas/Quant_Connect_JSON_analysis/resolve/main/plotting.py
13.9 kB
| # -*- coding: utf-8 -*- | |
| """plotting.py | |
| Automatically generated by Colab. | |
| Original file is located at | |
| https://colab.research.google.com/drive/1ILADgRrYqkAEj5jyymO50ZvzDzVdfD6g | |
| """ | |
| # plotting.py | |
| # Functions for generating Plotly figures from processed strategy data. | |
| import plotly.express as px | |
| import plotly.graph_objects as go | |
| import pandas as pd | |
| import numpy as np | |
| import traceback | |
| from utils import create_empty_figure # Import helper | |
| def generate_figures_for_strategy(strategy_results): | |
| """ | |
| Generates standard Plotly figures for a single strategy's results. | |
| Args: | |
| strategy_results: Dictionary containing processed data for one strategy, | |
| as returned by process_single_file. Expected keys include: | |
| 'filename', 'equity_df', 'drawdown_df', 'benchmark_df', | |
| 'trades_df', 'exposure_series', 'turnover_df'. | |
| Returns: | |
| A dictionary containing Plotly figure objects: | |
| 'equity_fig', 'drawdown_fig', 'benchmark_fig', 'pnl_hist_fig', | |
| 'duration_hist_fig', 'exposure_fig', 'turnover_fig'. | |
| Uses empty figures if data is missing or invalid. | |
| """ | |
| figures = { | |
| "equity_fig": create_empty_figure("Equity Curve"), | |
| "drawdown_fig": create_empty_figure("Drawdown Curve"), | |
| "benchmark_fig": create_empty_figure("Equity vs Benchmark"), | |
| "pnl_hist_fig": create_empty_figure("P/L Distribution"), | |
| "duration_hist_fig": create_empty_figure("Trade Duration Distribution"), | |
| "exposure_fig": create_empty_figure("Exposure"), | |
| "turnover_fig": create_empty_figure("Portfolio Turnover") | |
| } | |
| filename = strategy_results.get("filename", "Strategy") # Get filename for titles | |
| try: | |
| # --- Equity Curve --- | |
| equity_df = strategy_results.get("equity_df") | |
| if equity_df is not None and not equity_df.empty and 'Time' in equity_df.columns and 'Equity' in equity_df.columns: | |
| # Ensure Time is datetime | |
| equity_df['Time'] = pd.to_datetime(equity_df['Time']) | |
| fig = px.line(equity_df, x='Time', y='Equity', title=f'Equity Curve ({filename})') | |
| fig.update_layout(yaxis_title="Portfolio Value") | |
| figures["equity_fig"] = fig | |
| # --- Drawdown Curve --- | |
| drawdown_df = strategy_results.get("drawdown_df") | |
| if drawdown_df is not None and not drawdown_df.empty and 'Time' in drawdown_df.columns and 'Drawdown' in drawdown_df.columns: | |
| # Ensure Time is datetime | |
| drawdown_df['Time'] = pd.to_datetime(drawdown_df['Time']) | |
| # Convert drawdown to percentage for plotting | |
| drawdown_df['Drawdown_pct'] = drawdown_df['Drawdown'] * 100 | |
| fig = px.area(drawdown_df, x='Time', y='Drawdown_pct', title=f'Drawdown Curve (%) ({filename})', labels={'Drawdown_pct': 'Drawdown (%)'}) | |
| fig.update_layout(yaxis_title="Drawdown (%)") | |
| figures["drawdown_fig"] = fig | |
| # --- Equity vs Benchmark --- | |
| benchmark_df = strategy_results.get("benchmark_df") | |
| # Requires both equity and benchmark data | |
| if equity_df is not None and not equity_df.empty and 'Time' in equity_df.columns and 'Equity' in equity_df.columns and \ | |
| benchmark_df is not None and not benchmark_df.empty and 'Time' in benchmark_df.columns and 'Benchmark' in benchmark_df.columns: | |
| try: | |
| # Ensure Time columns are datetime | |
| equity_df['Time'] = pd.to_datetime(equity_df['Time']) | |
| benchmark_df['Time'] = pd.to_datetime(benchmark_df['Time']) | |
| # Merge on Time after setting as index | |
| equity_indexed = equity_df.set_index('Time')['Equity'] | |
| benchmark_indexed = benchmark_df.set_index('Time')['Benchmark'] | |
| # Combine, handling potential different start/end dates | |
| combined = pd.concat([equity_indexed, benchmark_indexed], axis=1, keys=['Equity', 'Benchmark'], join='outer') | |
| # Normalize to start at 1 (or 100) for comparison | |
| # Check if first row has NaN values after outer join | |
| first_valid_index = combined.first_valid_index() | |
| if first_valid_index is not None: | |
| # Normalize using the first non-NaN value for each column | |
| normalized_equity = (combined['Equity'] / combined['Equity'].loc[combined['Equity'].first_valid_index()])#.fillna(method='ffill') # Optional fill | |
| normalized_benchmark = (combined['Benchmark'] / combined['Benchmark'].loc[combined['Benchmark'].first_valid_index()])#.fillna(method='ffill') # Optional fill | |
| # Create figure and add traces | |
| fig = go.Figure() | |
| fig.add_trace(go.Scatter(x=normalized_equity.index, y=normalized_equity, mode='lines', name='Strategy Equity')) | |
| fig.add_trace(go.Scatter(x=normalized_benchmark.index, y=normalized_benchmark, mode='lines', name='Benchmark')) | |
| fig.update_layout(title=f'Normalized Equity vs Benchmark ({filename})', xaxis_title='Date', yaxis_title='Normalized Value (Start = 1)') | |
| figures["benchmark_fig"] = fig | |
| else: | |
| print("Could not normalize Equity vs Benchmark: No valid starting point found after merge.") | |
| figures["benchmark_fig"] = create_empty_figure(f"Equity vs Benchmark ({filename}) - Normalization Failed") | |
| except Exception as merge_err: | |
| print(f"Error merging/plotting Equity vs Benchmark: {merge_err}") | |
| figures["benchmark_fig"] = create_empty_figure(f"Equity vs Benchmark ({filename}) - Error") | |
| # --- Trade P/L Distribution --- | |
| trades_df = strategy_results.get("trades_df") | |
| if trades_df is not None and not trades_df.empty and 'profitLoss' in trades_df.columns: | |
| # Ensure profitLoss is numeric | |
| trades_df['profitLoss'] = pd.to_numeric(trades_df['profitLoss'], errors='coerce') | |
| valid_pnl = trades_df['profitLoss'].dropna() | |
| if not valid_pnl.empty: | |
| fig = px.histogram(valid_pnl, title=f'Trade Profit/Loss Distribution ({filename})', labels={'value': 'Profit/Loss'}) | |
| figures["pnl_hist_fig"] = fig | |
| # --- Trade Duration Distribution --- | |
| # Uses 'duration_days' calculated in processing.py | |
| if trades_df is not None and not trades_df.empty and 'duration_days' in trades_df.columns: | |
| # Ensure duration_days is numeric | |
| trades_df['duration_days'] = pd.to_numeric(trades_df['duration_days'], errors='coerce') | |
| valid_duration = trades_df['duration_days'].dropna() | |
| if not valid_duration.empty: | |
| fig = px.histogram(valid_duration, title=f'Trade Duration Distribution (Days) ({filename})', labels={'value': 'Duration (Days)'}) | |
| figures["duration_hist_fig"] = fig | |
| # --- Exposure Chart --- | |
| # Exposure data format varies; this is a basic example assuming a dict of series | |
| exposure_series_dict = strategy_results.get("exposure_series") | |
| if exposure_series_dict and isinstance(exposure_series_dict, dict): | |
| fig = go.Figure() | |
| exposure_plotted = False | |
| for series_name, series_data in exposure_series_dict.items(): | |
| if 'values' in series_data and isinstance(series_data['values'], list): | |
| # Process this specific series using the timeseries helper | |
| exposure_df = process_timeseries_chart(series_data['values'], series_name) | |
| if not exposure_df.empty: | |
| # Plot as area chart if 'Exposure' in name, else line | |
| plot_type = 'area' if 'Exposure' in series_name else 'scatter' | |
| fill_type = 'tozeroy' if plot_type == 'area' else None | |
| fig.add_trace(go.Scatter(x=exposure_df.index, y=exposure_df[series_name], | |
| mode='lines', name=series_name, fill=fill_type)) | |
| exposure_plotted = True | |
| if exposure_plotted: | |
| fig.update_layout(title=f'Exposure ({filename})', xaxis_title='Date', yaxis_title='Value / % Exposure') | |
| figures["exposure_fig"] = fig | |
| else: | |
| figures["exposure_fig"] = create_empty_figure(f"Exposure ({filename}) - No PlotData") | |
| else: | |
| figures["exposure_fig"] = create_empty_figure(f"Exposure ({filename}) - Data Missing/Invalid") | |
| # --- Portfolio Turnover --- | |
| turnover_df = strategy_results.get("turnover_df") | |
| if turnover_df is not None and not turnover_df.empty and 'Time' in turnover_df.columns and 'Turnover' in turnover_df.columns: | |
| # Ensure Time is datetime | |
| turnover_df['Time'] = pd.to_datetime(turnover_df['Time']) | |
| fig = px.line(turnover_df, x='Time', y='Turnover', title=f'Portfolio Turnover ({filename})') | |
| fig.update_layout(yaxis_title="Turnover") | |
| figures["turnover_fig"] = fig | |
| except Exception as e: | |
| print(f"Error generating figures for {filename}: {e}") | |
| traceback.print_exc() | |
| # Keep default empty figures on error | |
| return figures | |
| def generate_manual_risk_figures(analysis_results, filename="Strategy"): | |
| """ | |
| Generates Plotly figures from manually calculated risk analysis results. | |
| Args: | |
| analysis_results: Dictionary containing results from calculate_manual_risk_stats. | |
| Expected keys: 'monthly_returns_table_for_heatmap', 'rolling_vol_df'. | |
| filename: Name of the strategy for figure titles. | |
| Returns: | |
| A dictionary containing Plotly figure objects: | |
| 'monthly_heatmap_fig', 'rolling_vol_fig'. | |
| Uses empty figures if data is missing or invalid. | |
| """ | |
| figures = { | |
| "monthly_heatmap_fig": create_empty_figure(f"Monthly Returns Heatmap ({filename})"), | |
| "rolling_vol_fig": create_empty_figure(f"Rolling Volatility ({filename})") | |
| } | |
| try: | |
| # --- Monthly Returns Heatmap --- | |
| # Expects percentages (values * 100) from calculate_manual_risk_stats | |
| monthly_ret_table = analysis_results.get("monthly_returns_table_for_heatmap") | |
| if monthly_ret_table is not None and not monthly_ret_table.empty: | |
| z = monthly_ret_table.values # The percentage values | |
| x = monthly_ret_table.columns # Month names | |
| y = monthly_ret_table.index # Years | |
| # Create heatmap | |
| fig = go.Figure(data=go.Heatmap( | |
| z=z, x=x, y=y, | |
| colorscale='RdYlGn', # Red-Yellow-Green scale, good for returns | |
| zmid=0, # Center color scale around zero | |
| # Format text labels shown on the heatmap cells | |
| text=monthly_ret_table.applymap(lambda v: f'{v:.1f}%' if pd.notna(v) else '').values, | |
| texttemplate="%{text}", # Use the formatted text | |
| hoverongaps=False, # Don't show hover info for gaps | |
| colorbar=dict(title='Monthly Return (%)') # Add color bar title | |
| )) | |
| fig.update_layout( | |
| title=f'Monthly Returns (%) ({filename})', | |
| yaxis_nticks=len(y), # Ensure all years are shown as ticks | |
| yaxis_title="Year", | |
| yaxis_autorange='reversed' # Show earlier years at the top | |
| ) | |
| figures["monthly_heatmap_fig"] = fig | |
| # --- Rolling Volatility Plot --- | |
| rolling_vol_df = analysis_results.get("rolling_vol_df") | |
| # Check if DataFrame exists, is not empty, and has the 'Time' column | |
| if rolling_vol_df is not None and not rolling_vol_df.empty and 'Time' in rolling_vol_df.columns: | |
| # Ensure Time is datetime | |
| rolling_vol_df['Time'] = pd.to_datetime(rolling_vol_df['Time']) | |
| fig = go.Figure() | |
| colors = px.colors.qualitative.Plotly # Get a qualitative color sequence | |
| i = 0 # Color index | |
| vol_plotted = False | |
| # Iterate through columns starting with 'vol_' | |
| for col in rolling_vol_df.columns: | |
| if col.startswith('vol_'): | |
| window_label = col.split('_')[1] # Extract window label (e.g., '3M') | |
| # Plot volatility as percentage | |
| fig.add_trace(go.Scatter( | |
| x=rolling_vol_df['Time'], | |
| y=rolling_vol_df[col] * 100, # Convert to percentage | |
| mode='lines', | |
| name=f'Rolling Vol ({window_label})', | |
| line=dict(color=colors[i % len(colors)]) # Cycle through colors | |
| )) | |
| i += 1 | |
| vol_plotted = True | |
| # Update layout if at least one volatility series was plotted | |
| if vol_plotted: | |
| fig.update_layout( | |
| title=f'Annualized Rolling Volatility ({filename})', | |
| xaxis_title='Date', | |
| yaxis_title='Volatility (%)' # Y-axis label as percentage | |
| ) | |
| figures["rolling_vol_fig"] = fig | |
| else: | |
| figures["rolling_vol_fig"] = create_empty_figure(f"Rolling Volatility ({filename}) - No Plot Data") | |
| else: | |
| figures["rolling_vol_fig"] = create_empty_figure(f"Rolling Volatility ({filename}) - Data Missing/Invalid") | |
| except Exception as e: | |
| print(f"Error generating manual risk figures for {filename}: {e}") | |
| traceback.print_exc() | |
| # Keep default empty figures on error | |
| return figures |