Hzksj

# Copy and paste this fixed code into your script # Replace your main function with this version def main(): # Parse command line arguments (if any) args = parse_arguments() lookback_days = args.lookback_days if hasattr(args, 'lookback_days') else 365 # Update data path from arguments if provided data_path = args.data_path if hasattr(args, 'data_path') else DATA_PATH # Check if PDF generation is disabled generate_pdf = not (hasattr(args, 'no_pdf') and args.no_pdf) pdf_filename = args.pdf_filename if hasattr(args, 'pdf_filename') else PDF_FILENAME print("\n*** HMM REGIME DETECTION - MULTI-TENOR PRODUCTION RUN ***") print(f"Run date: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") # Print active tenors and their window sizes active_tenors = [f"{tenor} (Window: {config['window_size']})" for tenor, config in ASW_TENORS.items() if config['include']] print(f"Active tenors: {', '.join(active_tenors)}") # Load and preprocess data df = load_data(data_path) if df is None: print("Error: Could not load data. Exiting.") return # Store results for each tenor tenor_results = {} # Process each tenor for tenor in list(ASW_TENORS.keys()): # Check if tenor exists in the dataframe and should be included if tenor in df.columns and ASW_TENORS[tenor]['include']: print(f"\n{'-'*70}") print(f"Processing {tenor} with window size {ASW_TENORS[tenor]['window_size']}...") print(f"{'-'*70}") try: # Prepare tenor-specific data tenor_df = prepare_tenor_data(df, tenor) # Fit HMM model model_result = fit_hmm_model(tenor_df, tenor) if model_result is None: print(f"Error: Could not fit model for {tenor}. Skipping.") tenor_results[tenor] = { 'regime_df': None, 'regime_characteristics': None } continue # Get smoothed probabilities smoothed_probs = model_result.smoothed_marginal_probabilities # Determine regime characteristics regime_df, regime_characteristics = determine_regime_characteristics(tenor_df, smoothed_probs, tenor) # Store results tenor_results[tenor] = { 'regime_df': regime_df, 'regime_characteristics': regime_characteristics } # Print tenor's regime summary print_tenor_regime_summary(regime_df, regime_characteristics, tenor) # Visualize tenor results visualize_tenor_results(tenor_df, regime_df, regime_characteristics, tenor, lookback_days) except Exception as e: print(f"Error processing {tenor}: {str(e)}") import traceback traceback.print_exc() tenor_results[tenor] = { 'regime_df': None, 'regime_characteristics': None } else: # Skip this tenor - it's either not in the data or was excluded print(f"Skipping {tenor} - not found in data or excluded") tenor_results[tenor] = { 'regime_df': None, 'regime_characteristics': None } # Create summary dashboard for all tenors summary_fig = create_summary_dashboard(tenor_results, lookback_days=90) # Generate PDF report if enabled if generate_pdf: pdf_path = generate_pdf_report(tenor_results, summary_fig, data_path) if pdf_path: print(f"PDF report saved to: {pdf_path}") else: print("PDF report generation skipped.") print("\nProcess completed successfully.")

def generate_pdf_report(tenor_results, summary_fig, data_path): """Generate a comprehensive PDF report with all visualizations and results""" print("\nGenerating PDF report...") today_str = datetime.now().strftime('%Y%m%d') pdf_filename = f"{RESULTS_DIR}/{PDF_FILENAME.replace('.pdf', '')}_{today_str}.pdf" try: with PdfPages(pdf_filename) as pdf: # Title page fig = plt.figure(figsize=(11.7, 8.3)) # A4 size in inches plt.axis('off') # Title and subtitle plt.text(0.5, 0.8, "ASW Regime Detection Analysis", fontsize=24, ha='center', weight='bold') plt.text(0.5, 0.7, f"Generated on {datetime.now().strftime('%Y-%m-%d %H:%M')}", fontsize=14, ha='center') plt.text(0.5, 0.6, f"Data source: {os.path.basename(data_path)}", fontsize=14, ha='center') # Active tenors info active_tenors = [f"{tenor} (Window: {ASW_TENORS[tenor]['window_size']})" for tenor, data in tenor_results.items() if data['regime_df'] is not None] plt.text(0.5, 0.5, "Analyzed Tenors:", fontsize=16, ha='center') for i, tenor_text in enumerate(active_tenors): plt.text(0.5, 0.45 - i*0.05, tenor_text, fontsize=12, ha='center') # BNP Paribas footer plt.text(0.5, 0.05, "BNP PARIBAS\nCONFIDENTIAL", fontsize=10, ha='center', weight='bold') pdf.savefig(fig) plt.close(fig) # Table of Contents page fig = plt.figure(figsize=(11.7, 8.3)) # A4 size in inches plt.axis('off') plt.text(0.5, 0.9, "Table of Contents", fontsize=20, ha='center', weight='bold') content_items = [ ("1. Summary Dashboard", 0.8), ("2. Individual Tenor Analysis", 0.75) ] # Add each tenor to the TOC for i, tenor in enumerate([t for t in tenor_results if tenor_results[t]['regime_df'] is not None]): display_name = ASW_TENORS[tenor]['display_name'] content_items.append((f" 2.{i+1}. {display_name}", 0.7 - i*0.04)) for text, y_pos in content_items: plt.text(0.3, y_pos, text, fontsize=14, ha='left') pdf.savefig(fig) plt.close(fig) # Add summary dashboard if summary_fig is not None: pdf.savefig(summary_fig) plt.close(summary_fig) # Add individual tenor analysis pages for tenor, data in tenor_results.items(): if data['regime_df'] is None: continue regime_df = data['regime_df'] regime_char = data['regime_characteristics'] # Create a page with multiple visualizations for each tenor fig = plt.figure(figsize=(11.7, 8.3)) # A4 size gs = gridspec.GridSpec(3, 2, height_ratios=[1, 2, 2]) # 1. Tenor info section ax_info = plt.subplot(gs[0, :]) ax_info.axis('off') display_name = ASW_TENORS[tenor]['display_name'] window_size = ASW_TENORS[tenor]['window_size'] bullish_regime = regime_char['bullish_regime'] latest_date = regime_df.index[-1] latest_regime = regime_df.loc[latest_date, 'Market_Condition'] bullish_prob = regime_df.loc[latest_date, f'Prob_Regime_{bullish_regime}'] bearish_prob = 1 - bullish_prob regime_color = 'green' if latest_regime == 'Bullish' else 'red' # Create info text info_title = plt.text(0.5, 0.8, f"{display_name} Regime Analysis", fontsize=18, ha='center', weight='bold') info_text = ( f"Window Size: {window_size}\n" f"Current Regime: {latest_regime} (as of {latest_date.strftime('%Y-%m-%d')})\n" f"Bullish Probability: {bullish_prob:.2f} | Bearish Probability: {bearish_prob:.2f}" ) plt.text(0.5, 0.4, info_text, fontsize=12, ha='center', bbox=dict(facecolor='white', alpha=0.8, boxstyle='round'), color=regime_color) # 2. Full history plot ax_history = plt.subplot(gs[1, :]) ax_history.plot(regime_df.index, regime_df[tenor], 'k-', linewidth=1) # Color background based on regime for i in range(len(regime_df)-1): date = regime_df.index[i] next_date = regime_df.index[i+1] regime = regime_df.iloc[i]['Market_Condition'] color = 'green' if regime == 'Bullish' else 'red' ax_history.axvspan(date, next_date, facecolor=color, alpha=0.2) ax_history.set_title(f"Full History: {display_name}", fontsize=14) ax_history.grid(True, alpha=0.3) # 3. Recent history & regime probability plots # Get last year of data recent_start = latest_date - timedelta(days=365) recent_data = regime_df.loc[recent_start:].copy() ax_recent = plt.subplot(gs[2, 0]) ax_recent.plot(recent_data.index, recent_data[tenor], 'k-', linewidth=1.5) # Color background for recent data changes = recent_data[recent_data['Regime_Change'] == True].index.tolist() prev_regime = recent_data.iloc[0]['Market_Condition'] start = recent_data.index[0] for change_date in changes: color = 'green' if prev_regime == 'Bullish' else 'red' ax_recent.axvspan(start, change_date, facecolor=color, alpha=0.2) start = change_date prev_regime = 'Bullish' if prev_regime == 'Bearish' else 'Bearish' # Last segment color = 'green' if prev_regime == 'Bullish' else 'red' ax_recent.axvspan(start, recent_data.index[-1], facecolor=color, alpha=0.2) # Add regime change lines for change_date in changes: ax_recent.axvline(x=change_date, color='blue', linestyle='--', linewidth=1) # Format recent plot ax_recent.set_title(f"Last 12 Months: {display_name}", fontsize=12) ax_recent.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d')) ax_recent.xaxis.set_major_locator(mdates.MonthLocator(interval=2)) plt.setp(ax_recent.get_xticklabels(), rotation=45, ha='right', fontsize=8) ax_recent.grid(True, alpha=0.3) # 4. Probability plot ax_prob = plt.subplot(gs[2, 1]) bullish_prob_col = f'Prob_Regime_{bullish_regime}' ax_prob.plot(recent_data.index, recent_data[bullish_prob_col], color='blue', linewidth=2, label='Bullish Probability') ax_prob.fill_between(recent_data.index, 0, recent_data[bullish_prob_col], color='lightblue', alpha=0.4) # Add horizontal line at 0.5 ax_prob.axhline(y=0.5, color='gray', linestyle='--', alpha=0.7) # Add vertical lines at regime changes for change_date in changes: ax_prob.axvline(x=change_date, color='blue', linestyle='--', linewidth=1) # Format probability plot ax_prob.set_title("Bullish Regime Probability", fontsize=12) ax_prob.set_ylim(0, 1) ax_prob.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d')) ax_prob.xaxis.set_major_locator(mdates.MonthLocator(interval=2)) plt.setp(ax_prob.get_xticklabels(), rotation=45, ha='right', fontsize=8) ax_prob.grid(True, alpha=0.3) # Overall formatting plt.tight_layout(rect=[0, 0.03, 1, 0.95]) # Footer plt.figtext(0.01, 0.01, f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}", fontsize=8, style='italic') plt.figtext(0.99, 0.01, "BNP PARIBAS\nConfidential", ha='right', fontsize=8, style='italic', alpha=0.7) # Add to PDF pdf.savefig(fig) plt.close(fig) # Add a page with recent regime changes and statistics # Create a statistics and transitions page fig = plt.figure(figsize=(11.7, 8.3)) plt.axis('off') plt.text(0.5, 0.95, f"{display_name} Regime Statistics", fontsize=18, ha='center', weight='bold') # Calculate regime statistics total_days = len(regime_df) bullish_days = (regime_df['Market_Condition'] == 'Bullish').sum() bearish_days = (regime_df['Market_Condition'] == 'Bearish').sum() bullish_pct = (bullish_days / total_days) * 100 bearish_pct = (bearish_days / total_days) * 100 transitions = (regime_df['Market_Condition'] != regime_df['Market_Condition'].shift(1)).sum() avg_regime_duration = total_days / max(1, transitions) regime_shifts = regime_df[regime_df['Regime_Change'] == True].copy() # Statistics text stats_text = ( f"Total Analysis Period: {regime_df.index[0].strftime('%Y-%m-%d')} to {regime_df.index[-1].strftime('%Y-%m-%d')}\n\n" f"Time in Bullish Regime: {bullish_days} days ({bullish_pct:.1f}%)\n" f"Time in Bearish Regime: {bearish_days} days ({bearish_pct:.1f}%)\n\n" f"Number of Regime Transitions: {transitions}\n" f"Average Regime Duration: {avg_regime_duration:.1f} days" ) plt.text(0.5, 0.8, stats_text, fontsize=14, ha='center', bbox=dict(facecolor='white', edgecolor='gray', boxstyle='round', alpha=0.9)) # Recent regime changes table if len(regime_shifts) > 0: # Get last 15 regime changes recent_shifts = regime_shifts.tail(15).copy() recent_shifts['Previous Regime'] = recent_shifts['Market_Condition'].shift(1) # The first row will have NaN for previous regime first_regime = 'Bearish' if recent_shifts.iloc[0]['Market_Condition'] == 'Bullish' else 'Bullish' recent_shifts.loc[recent_shifts.index[0], 'Previous Regime'] = first_regime plt.text(0.5, 0.6, "Recent Regime Transitions", fontsize=16, ha='center', weight='bold') # Table header header = ['Date', 'From Regime', 'To Regime', f'{tenor} Value'] col_width = 0.2 # Table positions y_start = 0.55 y_step = 0.03 # Draw header for i, head in enumerate(header): x_pos = 0.2 + i * col_width plt.text(x_pos, y_start, head, fontsize=12, ha='left', weight='bold') # Draw lines plt.plot([0.18, 0.82], [y_start - 0.01, y_start - 0.01], 'k-', alpha=0.3) # Draw data rows for j, (idx, row) in enumerate(recent_shifts.iterrows()): y_pos = y_start - (j+1) * y_step # Date plt.text(0.2, y_pos, idx.strftime('%Y-%m-%d'), fontsize=11, ha='left') # From regime with color from_color = 'green' if row['Previous Regime'] == 'Bullish' else 'red' plt.text(0.4, y_pos, row['Previous Regime'], fontsize=11, ha='left', color=from_color) # To regime with color to_color = 'green' if row['Market_Condition'] == 'Bullish' else 'red' plt.text(0.6, y_pos, row['Market_Condition'], fontsize=11, ha='left', color=to_color) # Tenor value plt.text(0.8, y_pos, f"{row[tenor]:.3f}", fontsize=11, ha='left') else: plt.text(0.5, 0.5, "No regime transitions detected in the data period.", fontsize=14, ha='center', style='italic') # Footer plt.figtext(0.01, 0.01, f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}", fontsize=8, style='italic') plt.figtext(0.99, 0.01, "BNP PARIBAS\nConfidential", ha='right', fontsize=8, style='italic', alpha=0.7) pdf.savefig(fig) plt.close(fig) # Final page with methodology explanation fig = plt.figure(figsize=(11.7, 8.3)) plt.axis('off') plt.text(0.5, 0.95, "Methodology & Technical Notes", fontsize=18, ha='center', weight='bold') methodology_text = """ The ASW Regime Detection Model uses a Hidden Markov Model (HMM) with Markov Switching Regression to identify distinct market regimes based on the behavior of Asset Swap (ASW) spreads. Key aspects of the methodology: 1. Data Processing: • The raw ASW spread data is first differenced to make it stationary • A rolling average with an optimized window size is applied to smooth out noise 2. Regime Identification: • A two-state Markov Switching Model is fitted to the processed data • The model identifies distinct regimes based on the statistical properties of the series • Regimes are classified as 'Bullish' or 'Bearish' based on the average price movement 3. Optimization: • Each ASW tenor uses its own optimized window size for best performance • The window size affects the sensitivity of regime detection 4. Trading Signals: • Regime transitions naturally generate trading signals: - Transitions to Bullish regime → Buy Signal - Transitions to Bearish regime → Sell Signal 5. Probability Tracking: • The model provides probability estimates for each regime • This allows for monitoring regime strength and anticipating potential transitions This model is intended to be run daily to provide updated regime classifications and probability scores. """ plt.text(0.1, 0.8, methodology_text, fontsize=12, ha='left', va='top', bbox=dict(facecolor='white', edgecolor='lightgray', boxstyle='round', alpha=0.9)) # Usage notes usage_notes = """ Usage Notes: • Window Sizes: The window size for each tenor can be customized to adjust sensitivity • Lookback Period: The visualization lookback period can be adjusted (default: 365 days) • Exclusions: Individual tenors can be excluded from analysis if desired The PDF report is automatically generated with all relevant analysis and can be distributed to stakeholders as needed. """ plt.text(0.1, 0.35, usage_notes, fontsize=12, ha='left', va='top', bbox=dict(facecolor='#f0f0f0', edgecolor='lightgray', boxstyle='round', alpha=0.9)) # Footer plt.figtext(0.01, 0.01, f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}", fontsize=8, style='italic') plt.figtext(0.99, 0.01, "BNP PARIBAS\nConfidential", ha='right', fontsize=8, style='italic', alpha=0.7) pdf.savefig(fig) plt.close(fig) print(f"PDF report successfully generated: {pdf_filename}") return pdf_filename except Exception as e: print(f"Error generating PDF report: {str(e)}") import traceback traceback.print_exc() return None

# HMM Regime Detection - Multi-Tenor Production Script # This script runs a Markov Switching Model to identify market regimes # across multiple ASW tenors and visualizes the results import pandas as pd import numpy as np import matplotlib.pyplot as plt import matplotlib.dates as mdates from statsmodels.tsa.regime_switching.markov_regression import MarkovRegression import warnings from datetime import datetime, timedelta import os import matplotlib.patches as patches import argparse import sys # Add this import for sys.modules check from matplotlib.backends.backend_pdf import PdfPages # For PDF generation import io from PIL import Image # For handling images in the PDF import matplotlib.gridspec as gridspec # Suppress warnings for cleaner output warnings.filterwarnings('ignore') # Configuration parameters DATA_PATH = 'ASW_ALL.xlsx' # Update to your data path containing multiple ASW tenors SAVE_RESULTS = True RESULTS_DIR = 'regime_results' GENERATE_PDF = True # Set to generate PDF report PDF_FILENAME = 'ASW_Regime_Analysis_Report.pdf' # Default PDF filename # ASW tenor configuration ASW_TENORS = { 'USSFCT02': {'window_size': 40, 'include': True, 'display_name': 'ASW 2Y'}, 'USSFCT05': {'window_size': 50, 'include': True, 'display_name': 'ASW 5Y'}, 'USSFCT07': {'window_size': 60, 'include': True, 'display_name': 'ASW 7Y'}, 'USSFCT10': {'window_size': 70, 'include': True, 'display_name': 'ASW 10Y'} } # Create results directory if it doesn't exist if SAVE_RESULTS and not os.path.exists(RESULTS_DIR): os.makedirs(RESULTS_DIR) def parse_arguments(): """Parse command line arguments for customizing analysis""" parser = argparse.ArgumentParser(description='HMM Regime Detection for Multiple ASW Tenors') # Add arguments for each tenor's window size for tenor, config in ASW_TENORS.items(): parser.add_argument(f'--window_{tenor.lower()}', type=int, default=config['window_size'], help=f'Rolling window size for {tenor}') parser.add_argument(f'--exclude_{tenor.lower()}', action='store_true', help=f'Exclude {tenor} from the analysis') # Add common arguments parser.add_argument('--lookback_days', type=int, default=365, help='Number of days to look back for visualization') parser.add_argument('--data_path', type=str, default=DATA_PATH, help='Path to the data file') parser.add_argument('--no_pdf', action='store_true', help='Skip PDF report generation') parser.add_argument('--pdf_filename', type=str, default=PDF_FILENAME, help='Custom filename for the PDF report') # For Jupyter environment, don't parse sys.argv which might cause errors try: # Check if running in Jupyter if 'ipykernel' in sys.modules: args = parser.parse_args([]) # Empty list means don't parse any args else: args = parser.parse_args() except Exception as e: print(f"Warning: Argument parsing error: {str(e)}") print("Using default parameters instead.") args = parser.parse_args([]) # Use empty list for defaults # Update the global ASW_TENORS dictionary with parsed arguments for tenor in ASW_TENORS: window_arg = f'window_{tenor.lower()}' exclude_arg = f'exclude_{tenor.lower()}' if hasattr(args, window_arg): ASW_TENORS[tenor]['window_size'] = getattr(args, window_arg) if hasattr(args, exclude_arg) and getattr(args, exclude_arg): ASW_TENORS[tenor]['include'] = False return args # Function to load and preprocess data def load_data(file_path): print(f"Loading data from {file_path}...") try: # Import data df = pd.read_excel(file_path, index_col=0, keep_default_na=False, na_values=['N/A']) # Check if 'Date' is already the index if not isinstance(df.index, pd.DatetimeIndex): # If 'Date' is a column, set it as index if 'Date' in df.columns: df = df.set_index('Date') else: # Check if the first column might be the date column try: # Try to convert the first column to datetime first_col = df.reset_index().iloc[:, 0] first_col_dt = pd.to_datetime(first_col) # If successful, reset and set properly df = df.reset_index() df['Date'] = first_col_dt df = df.set_index('Date') except: # If that fails, convert the existing index to datetime if possible try: df.index = pd.to_datetime(df.index) except: print("Warning: Could not identify or convert date column. Please ensure your data has a proper date column.") # Drop missing values (if any) df = df.dropna() # Check which configured tenors exist in the dataframe available_tenors = [] for tenor in ASW_TENORS: if tenor in df.columns and ASW_TENORS[tenor]['include']: available_tenors.append(tenor) if not available_tenors: print("Error: No configured ASW tenors found in the data file.") print(f"Available columns in the data file: {', '.join(df.columns)}") print(f"Looking for tenors: {', '.join(ASW_TENORS.keys())}") return None print(f"Available ASW tenors for analysis: {', '.join(available_tenors)}") print(f"Data loaded successfully. Shape: {df.shape}") return df except Exception as e: print(f"Error loading data: {str(e)}") import traceback traceback.print_exc() return None # Function to prepare data for a specific tenor def prepare_tenor_data(df, tenor): """Prepare data for a specific ASW tenor, applying its optimal window size""" window_size = ASW_TENORS[tenor]['window_size'] print(f"Preparing {tenor} data with window size {window_size}...") # Make a copy to avoid modifying the original dataframe tenor_df = df.copy() # Compute first difference (rate of change) of interest rates tenor_df[f'{tenor}_Change'] = tenor_df[tenor].diff() # Compute rolling average with the tenor-specific window size tenor_df[f'{tenor}_Trend'] = tenor_df[f'{tenor}_Change'].rolling(window=window_size).mean() # Drop rows with NaN values after creating features tenor_df = tenor_df.dropna() return tenor_df # Function to fit HMM model for a specific tenor def fit_hmm_model(df, tenor): print(f"Fitting Markov Switching Model for {tenor}...") try: # Fit Markov Switching Model model = MarkovRegression( df[f'{tenor}_Trend'], k_regimes=2, trend='c', switching_variance=True ) result = model.fit(maxiter=200, disp=False) print(f"Model for {tenor} fitted successfully.") return result except Exception as e: print(f"Error fitting model for {tenor}: {str(e)}") return None # Function to determine regime characteristics for a specific tenor def determine_regime_characteristics(df, regime_probabilities, tenor): # Create a dataframe with smoothed probabilities regime_df = pd.DataFrame(index=df.index) # Add regime probabilities regime_df['Prob_Regime_0'] = regime_probabilities[0] regime_df['Prob_Regime_1'] = regime_probabilities[1] # Determine the most likely regime for each day regime_df['Regime'] = regime_df[['Prob_Regime_0', 'Prob_Regime_1']].idxmax(axis=1) regime_df['Regime'] = regime_df['Regime'].str.replace('Prob_Regime_', '') # Add price data regime_df[tenor] = df[tenor] # Create a more robust method to identify bullish and bearish regimes # Use linear regression slope for each regime section from scipy import stats # Initialize lists to store slopes slopes_regime_0 = [] slopes_regime_1 = [] # Get consecutive regimes current_regime = None start_idx = None for i, row in regime_df.iterrows(): if current_regime is None: current_regime = row['Regime'] start_idx = i elif row['Regime'] != current_regime: # Regime changed, calculate slope of previous segment segment = regime_df.loc[start_idx:i, tenor] if len(segment) >= 5: # Only calculate if enough data points x = np.arange(len(segment)) slope, _, _, _, _ = stats.linregress(x, segment.values) if current_regime == '0': slopes_regime_0.append(slope) else: slopes_regime_1.append(slope) # Reset for next segment current_regime = row['Regime'] start_idx = i # Don't forget the last segment if start_idx is not None and current_regime is not None: segment = regime_df.loc[start_idx:, tenor] if len(segment) >= 5: x = np.arange(len(segment)) slope, _, _, _, _ = stats.linregress(x, segment.values) if current_regime == '0': slopes_regime_0.append(slope) else: slopes_regime_1.append(slope) # Determine which regime is bullish/bearish based on average slope regime_characteristics = {} if slopes_regime_0 and slopes_regime_1: avg_slope_0 = np.mean(slopes_regime_0) avg_slope_1 = np.mean(slopes_regime_1) print(f"{tenor} - Average slope for Regime 0: {avg_slope_0:.6f}") print(f"{tenor} - Average slope for Regime 1: {avg_slope_1:.6f}") # For credit spreads, negative slope (decreasing) = bullish if avg_slope_0 < avg_slope_1: bullish_regime = '0' bearish_regime = '1' else: bullish_regime = '1' bearish_regime = '0' regime_characteristics = { 'bullish_regime': bullish_regime, 'bearish_regime': bearish_regime } # Label market conditions regime_df['Market_Condition'] = regime_df['Regime'].apply( lambda x: 'Bullish' if x == bullish_regime else 'Bearish' ) else: # Fallback if we couldn't calculate slopes # Use simpler approach based on price differences # First create the column for price changes regime_df['Price_Changes'] = regime_df[tenor].diff() # Calculate average price change by regime avg_change_0 = regime_df[regime_df['Regime'] == '0']['Price_Changes'].mean() avg_change_1 = regime_df[regime_df['Regime'] == '1']['Price_Changes'].mean() print(f"{tenor} - Average price change for Regime 0: {avg_change_0:.6f}") print(f"{tenor} - Average price change for Regime 1: {avg_change_1:.6f}") # For credit spreads, negative change (decreasing) = bullish if avg_change_0 < avg_change_1: bullish_regime = '0' bearish_regime = '1' else: bullish_regime = '1' bearish_regime = '0' regime_characteristics = { 'bullish_regime': bullish_regime, 'bearish_regime': bearish_regime } # Label market conditions regime_df['Market_Condition'] = regime_df['Regime'].apply( lambda x: 'Bullish' if x == bullish_regime else 'Bearish' ) # Detect regime changes regime_df['Regime_Change'] = regime_df['Market_Condition'].ne(regime_df['Market_Condition'].shift(1)) return regime_df, regime_characteristics # Function to print tenor's regime summary def print_tenor_regime_summary(regime_df, regime_characteristics, tenor): try: bullish_regime = regime_characteristics.get('bullish_regime') if not bullish_regime: print(f"Error: Could not determine bullish regime for {tenor}.") return latest_date = regime_df.index[-1] latest_regime = regime_df.loc[latest_date, 'Market_Condition'] # Handle the case where the bullish probability column might not exist try: bullish_prob = regime_df.loc[latest_date, f'Prob_Regime_{bullish_regime}'] bearish_prob = 1 - bullish_prob except KeyError: # If we can't get the specific probability, use the regime determination bullish_prob = 1.0 if latest_regime == 'Bullish' else 0.0 bearish_prob = 1.0 if latest_regime == 'Bearish' else 0.0 print("\n" + "="*50) print(f"REGIME SUMMARY FOR {tenor} ON {latest_date.strftime('%Y-%m-%d')}") print("="*50) print(f"CURRENT MARKET REGIME: {latest_regime}") print(f"Bullish Probability: {bullish_prob:.4f}") print(f"Bearish Probability: {bearish_prob:.4f}") # Check for regime change if len(regime_df) > 1: previous_date = regime_df.index[-2] previous_regime = regime_df.loc[previous_date, 'Market_Condition'] if previous_regime != latest_regime: print("\n*** REGIME CHANGE DETECTED ***") print(f"Previous regime ({previous_date.strftime('%Y-%m-%d')}): {previous_regime}") print(f"New regime ({latest_date.strftime('%Y-%m-%d')}): {latest_regime}") print("="*50) except Exception as e: print(f"Error in print_tenor_regime_summary for {tenor}: {str(e)}") import traceback traceback.print_exc() # Function to visualize results for a specific tenor def visualize_tenor_results(df, regime_df, regime_characteristics, tenor, lookback_days=365): print(f"Generating visualization for {tenor}...") try: # Extract regime characteristics bullish_regime = regime_characteristics['bullish_regime'] bearish_regime = regime_characteristics['bearish_regime'] # Get data for visualization if lookback_days is None or lookback_days <= 0: plot_df = df.copy() plot_regime_df = regime_df.copy() else: # Get data for the lookback period end_date = df.index[-1] start_date = end_date - timedelta(days=int(lookback_days)) plot_df = df.loc[start_date:end_date].copy() plot_regime_df = regime_df.loc[start_date:end_date].copy() # Create figure with specific size fig = plt.figure(figsize=(12, 8)) # Set up grid for two plots with different heights gs = plt.GridSpec(2, 1, height_ratios=[3, 1], figure=fig) # Create top and bottom axes ax1 = fig.add_subplot(gs[0]) ax2 = fig.add_subplot(gs[1]) # ===== Top Plot: Price with Regime Classification ===== # Find regime change points regime_changes = plot_regime_df[plot_regime_df['Regime_Change'] == True].index.tolist() # Get all date ranges for each regime regime_periods = [] if len(plot_regime_df) > 0: # Start with the first regime start_date = plot_regime_df.index[0] current_regime = plot_regime_df.iloc[0]['Market_Condition'] for change_date in regime_changes: # Add the period that just ended regime_periods.append({ 'start': start_date, 'end': change_date, 'regime': current_regime }) # Start new period start_date = change_date # Get the new regime (after the change) current_regime = 'Bullish' if current_regime == 'Bearish' else 'Bearish' # Add the final period regime_periods.append({ 'start': start_date, 'end': plot_regime_df.index[-1], 'regime': current_regime }) # Color the background based on regime periods for period in regime_periods: color = 'green' if period['regime'] == 'Bullish' else 'red' ax1.axvspan(period['start'], period['end'], facecolor=color, alpha=0.2, edgecolor=None) # Plot the price line on top ax1.plot(plot_regime_df.index, plot_regime_df[tenor], 'k-', linewidth=1.5, label=tenor) # Add vertical lines at regime changes with labels for change_date in regime_changes: ax1.axvline(x=change_date, color='blue', linestyle='--', linewidth=1.5) # Get the new regime after the change date (might be the next day) # Use the next day's regime to determine if it's a buy or sell signal next_dates = plot_regime_df.loc[change_date:].index if len(next_dates) > 1: next_date = next_dates[1] # Get the date after the change new_regime = plot_regime_df.loc[next_date, 'Market_Condition'] # Add regime change annotation y_pos = plot_regime_df.loc[change_date, tenor] if new_regime == 'Bullish': marker = '^' # up triangle for buy marker_color = 'green' else: marker = 'v' # down triangle for sell marker_color = 'red' # Add marker at regime change ax1.scatter(change_date, y_pos, marker=marker, color=marker_color, s=120, zorder=5, edgecolors='black') # Add Buy/Sell signals in legend buy_patch = patches.Patch(color='green', alpha=0.2, label='Bullish Regime') sell_patch = patches.Patch(color='red', alpha=0.2, label='Bearish Regime') buy_signal = plt.Line2D([0], [0], marker='^', color='w', markerfacecolor='green', markersize=10, label='Buy Signal') sell_signal = plt.Line2D([0], [0], marker='v', color='w', markerfacecolor='red', markersize=10, label='Sell Signal') # Add legend ax1.legend(handles=[buy_patch, sell_patch, buy_signal, sell_signal], loc='upper right', framealpha=0.7) # Set title and labels ticker_name = ASW_TENORS[tenor]['display_name'] ax1.set_title(f'{ticker_name} with Regime Classification', fontsize=14) ax1.set_ylabel(ticker_name, fontsize=12) ax1.grid(True, alpha=0.3) # Format x-axis ax1.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d')) ax1.xaxis.set_major_locator(mdates.MonthLocator(interval=1)) plt.setp(ax1.get_xticklabels(), rotation=45, ha='right') # ===== Bottom Plot: Regime Probabilities ===== # Plot only the Bullish probability with a clean blue line bullish_prob_col = f'Prob_Regime_{bullish_regime}' ax2.plot(plot_regime_df.index, plot_regime_df[bullish_prob_col], color='blue', linewidth=2, label='Bullish Probability') # Add light blue filled area under the line ax2.fill_between(plot_regime_df.index, 0, plot_regime_df[bullish_prob_col], color='lightblue', alpha=0.4) # Add a horizontal reference line at 0.5 ax2.axhline(y=0.5, color='gray', linestyle='--', alpha=0.7) # Add vertical lines at the same regime change points for change_date in regime_changes: ax2.axvline(x=change_date, color='blue', linestyle='--', linewidth=1.5) # Set title and labels ax2.set_title('Bullish Regime Probability Over Time', fontsize=14) ax2.set_ylabel('Probability', fontsize=12) ax2.set_xlabel('Date', fontsize=12) # Set y-axis limits ax2.set_ylim(0, 1) # Format x-axis to match top plot ax2.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d')) ax2.xaxis.set_major_locator(mdates.MonthLocator(interval=1)) plt.setp(ax2.get_xticklabels(), rotation=45, ha='right') # Add grid ax2.grid(True, alpha=0.3) # Add current regime information to figure latest_date = regime_df.index[-1] latest_regime = regime_df.loc[latest_date, 'Market_Condition'] latest_prob = regime_df.loc[latest_date, bullish_prob_col] regime_color = 'green' if latest_regime == 'Bullish' else 'red' # Create summary text summary_text = f"Latest Date: {latest_date.strftime('%Y-%m-%d')}\n" summary_text += f"Current Regime: {latest_regime}\n" summary_text += f"Bullish Probability: {latest_prob:.2f}" # Add a text box with the latest regime info (as a figure annotation) plt.figtext(0.02, 0.02, summary_text, fontsize=12, bbox=dict(facecolor='white', alpha=0.8, boxstyle='round'), color=regime_color) # Adjust layout plt.tight_layout() # Add subtle BNP Paribas branding in bottom right corner plt.figtext(0.98, 0.01, "BNP PARIBAS\nConfidential", ha='right', fontsize=8, style='italic', alpha=0.7) # Save results if requested if SAVE_RESULTS: today_str = datetime.now().strftime('%Y%m%d') fig.savefig(f"{RESULTS_DIR}/{tenor}_regime_analysis_{today_str}.png", dpi=300, bbox_inches='tight') # Also save the regime data regime_df.to_csv(f"{RESULTS_DIR}/{tenor}_regime_data_{today_str}.csv") plt.show() return fig except Exception as e: print(f"Error visualizing results for {tenor}: {str(e)}") import traceback traceback.print_exc() return None # Function to create summary dashboard for all tenors def create_summary_dashboard(tenor_results, lookback_days=90): print("Creating summary dashboard for all tenors...") try: # Count how many tenors we're dealing with active_tenors = [tenor for tenor, data in tenor_results.items() if data['regime_df'] is not None] num_tenors = len(active_tenors) if num_tenors == 0: print("No tenor data available for summary dashboard.") return None # Create a figure with subplots - one row per tenor fig, axes = plt.subplots(num_tenors, 1, figsize=(12, 4*num_tenors), sharex=True) # Handle the case where there's only one tenor if num_tenors == 1: axes = [axes] # End date will be the same for all tenors (today) end_date = tenor_results[active_tenors[0]]['regime_df'].index[-1] start_date = end_date - timedelta(days=lookback_days) # Plot each tenor for i, tenor in enumerate(active_tenors): ax = axes[i] regime_df = tenor_results[tenor]['regime_df'] regime_char = tenor_results[tenor]['regime_characteristics'] # Filter data for the lookback period plot_data = regime_df.loc[start_date:end_date].copy() # Get bullish regime identifier bullish_regime = regime_char['bullish_regime'] # Plot the price line ax.plot(plot_data.index, plot_data[tenor], 'k-', linewidth=1.5) # Color the background prev_date = plot_data.index[0] prev_regime = plot_data.iloc[0]['Market_Condition'] # Find regime changes changes = plot_data[plot_data['Regime_Change'] == True].index.tolist() # Add the first period start = plot_data.index[0] for change_date in changes: color = 'green' if prev_regime == 'Bullish' else 'red' ax.axvspan(start, change_date, facecolor=color, alpha=0.2) # Update for next period start = change_date prev_regime = 'Bullish' if prev_regime == 'Bearish' else 'Bearish' # Add the final period color = 'green' if prev_regime == 'Bullish' else 'red' ax.axvspan(start, plot_data.index[-1], facecolor=color, alpha=0.2) # Add vertical lines at regime changes for change_date in changes: ax.axvline(x=change_date, color='blue', linestyle='--', linewidth=1) # Current regime info current_regime = plot_data.iloc[-1]['Market_Condition'] bullish_prob = plot_data.iloc[-1][f'Prob_Regime_{bullish_regime}'] # Set title and labels display_name = ASW_TENORS[tenor]['display_name'] window_size = ASW_TENORS[tenor]['window_size'] ax.set_title(f"{display_name} - Current: {current_regime} (Prob: {bullish_prob:.2f}, Window: {window_size})", fontsize=10, loc='left') ax.set_ylabel(display_name, fontsize=9) ax.grid(True, alpha=0.3) ax.tick_params(axis='y', labelsize=8) # Format x-axis (only for bottom plot) axes[-1].xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d')) axes[-1].xaxis.set_major_locator(mdates.WeekdayLocator(interval=2)) axes[-1].tick_params(axis='x', rotation=45) axes[-1].set_xlabel('Date', fontsize=10) # Add title to the entire figure fig.suptitle('ASW Regime Analysis Summary Dashboard', fontsize=16, y=0.98) # Legend for the entire figure buy_patch = patches.Patch(color='green', alpha=0.2, label='Bullish Regime') sell_patch = patches.Patch(color='red', alpha=0.2, label='Bearish Regime') regime_change = plt.Line2D([0], [0], color='blue', linestyle='--', label='Regime Change') fig.legend(handles=[buy_patch, sell_patch, regime_change], loc='upper right', bbox_to_anchor=(0.99, 0.96), fontsize=9) # Add report generation date plt.figtext(0.01, 0.01, f"Report generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}", fontsize=8, style='italic') # BNP Paribas branding plt.figtext(0.99, 0.01, "BNP PARIBAS\nConfidential", ha='right', fontsize=8, style='italic', alpha=0.7) plt.tight_layout(rect=[0, 0.03, 1, 0.95]) # Save results if SAVE_RESULTS: today_str = datetime.now().strftime('%Y%m%d') fig.savefig(f"{RESULTS_DIR}/ASW_summary_dashboard_{today_str}.png", dpi=300, bbox_inches='tight') return fig except Exception as e: print(f"Error creating summary dashboard: {str(e)}") import traceback traceback.print_exc() return Noneset_major_formatter(mdates.DateFormatter('%Y-%m-%d')) axes[-1].xaxis.set_major_locator(mdates.WeekdayLocator(interval=2)) axes[-1].tick_params(axis='x', rotation=45) axes[-1].set_xlabel('Date', fontsize=10) # Add title to the entire figure fig.suptitle('ASW Regime Analysis Summary Dashboard', fontsize=16, y=0.98) # Legend for the entire figure buy_patch = patches.Patch(color='green', alpha=0.2, label='Bullish Regime') sell_patch = patches.Patch(color='red', alpha=0.2, label='Bearish Regime') regime_change = plt.Line2D([0], [0], color='blue', linestyle='--', label='Regime Change') fig.legend(handles=[buy_patch, sell_patch, regime_change], loc='upper right', bbox_to_anchor=(0.99, 0.96), fontsize=9) # Add report generation date plt.figtext(0.01, 0.01, f"Report generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}", fontsize=8, style='italic') # BNP Paribas branding plt.figtext(0.99, 0.01, "BNP PARIBAS\nConfidential", ha='right', fontsize=8, style='italic', alpha=0.7) plt.tight_layout(rect=[0, 0.03, 1, 0.95]) # Save results if SAVE_RESULTS: today_str = datetime.now().strftime('%Y%m%d') fig.savefig(f"{RESULTS_DIR}/ASW_summary_dashboard_{today_str}.png", dpi=300, bbox_inches='tight') return fig except Exception as e: print(f"Error creating summary dashboard: {str(e)}") import traceback traceback.print_exc() return None # Function to generate a PDF report of all results def generate_pdf_report(tenor_results, summary_fig, data_path): """Generate a comprehensive PDF report with all visualizations and results""" print("\nGenerating PDF report...") today_str = datetime.now().strftime('%Y%m%d') pdf_filename = f"{RESULTS_DIR}/{PDF_FILENAME.replace('.pdf', '')}_{today_str}.pdf" try: with PdfPages(pdf_filename) as pdf: # Title page fig = plt.figure(figsize=(11.7, 8.3)) # A4 size in inches plt.axis('off') # Title and subtitle plt.text(0.5, 0.8, "ASW Regime Detection Analysis", fontsize=24, ha='center', weight='bold') plt.text(0.5, 0.7, f"Generated on {datetime.now().strftime('%Y-%m-%d %H:%M')}", fontsize=14, ha='center') plt.text(0.5, 0.6, f"Data source: {os.path.basename(data_path)}", fontsize=14, ha='center') # Active tenors info active_tenors = [f"{tenor} (Window: {ASW_TENORS[tenor]['window_size']})" for tenor, data in tenor_results.items() if data['regime_df'] is not None] plt.text(0.5, 0.5, "Analyzed Tenors:", fontsize=16, ha='center') for i, tenor_text in enumerate(active_tenors): plt.text(0.5, 0.45 - i*0.05, tenor_text, fontsize=12, ha='center') # BNP Paribas footer plt.text(0.5, 0.05, "BNP PARIBAS\nCONFIDENTIAL", fontsize=10, ha='center', weight='bold') pdf.savefig(fig) plt.close(fig) # Table of Contents page fig = plt.figure(figsize=(11.7, 8.3)) # A4 size in inches plt.axis('off') plt.text(0.5, 0.9, "Table of Contents", fontsize=20, ha='center', weight='bold') content_items = [ ("1. Summary Dashboard", 0.8), ("2. Individual Tenor Analysis", 0.75) ] # Add each tenor to the TOC for i, tenor in enumerate([t for t in tenor_results if tenor_results[t]['regime_df'] is not None]): display_name = ASW_TENORS[tenor]['display_name'] content_items.append((f" 2.{i+1}. {display_name}", 0.7 - i*0.04)) for text, y_pos in content_items: plt.text(0.3, y_pos, text, fontsize=14, ha='left') pdf.savefig(fig) plt.close(fig) # Add summary dashboard if summary_fig is not None: pdf.savefig(summary_fig) plt.close(summary_fig) # Add individual tenor analysis pages for tenor, data in tenor_results.items(): if data['regime_df'] is None: continue regime_df = data['regime_df'] regime_char = data['regime_characteristics'] # Create a page with multiple visualizations for each tenor fig = plt.figure(figsize=(11.7, 8.3)) # A4 size gs = gridspec.GridSpec(3, 2, height_ratios=[1, 2, 2]) # 1. Tenor info section ax_info = plt.subplot(gs[0, :]) ax_info.axis('off') display_name = ASW_TENORS[tenor]['display_name'] window_size = ASW_TENORS[tenor]['window_size'] bullish_regime = regime_char['bullish_regime'] latest_date = regime_df.index[-1] latest_regime = regime_df.loc[latest_date, 'Market_Condition'] bullish_prob = regime_df.loc[latest_date, f'Prob_Regime_{bullish_regime}'] bearish_prob = 1 - bullish_prob regime_color = 'green' if latest_regime == 'Bullish' else 'red' # Create info text info_title = plt.text(0.5, 0.8, f"{display_name} Regime Analysis", fontsize=18, ha='center', weight='bold') info_text = ( f"Window Size: {window_size}\n" f"Current Regime: {latest_regime} (as of {latest_date.strftime('%Y-%m-%d')})\n" f"Bullish Probability: {bullish_prob:.2f} | Bearish Probability: {bearish_prob:.2f}" ) plt.text(0.5, 0.4, info_text, fontsize=12, ha='center', bbox=dict(facecolor='white', alpha=0.8, boxstyle='round'), color=regime_color) # 2. Full history plot ax_history = plt.subplot(gs[1, :]) ax_history.plot(regime_df.index, regime_df[tenor], 'k-', linewidth=1) # Color background based on regime for i in range(len(regime_df)-1): date = regime_df.index[i] next_date = regime_df.index[i+1] regime = regime_df.iloc[i]['Market_Condition'] color = 'green' if regime == 'Bullish' else 'red' ax_history.axvspan(date, next_date, facecolor=color, alpha=0.2) ax_history.set_title(f"Full History: {display_name}", fontsize=14) ax_history.grid(True, alpha=0.3) # 3. Recent history & regime probability plots # Get last year of data recent_start = latest_date - timedelta(days=365) recent_data = regime_df.loc[recent_start:].copy() ax_recent = plt.subplot(gs[2, 0]) ax_recent.plot(recent_data.index, recent_data[tenor], 'k-', linewidth=1.5) # Color background for recent data changes = recent_data[recent_data['Regime_Change'] == True].index.tolist() prev_regime = recent_data.iloc[0]['Market_Condition'] start = recent_data.index[0] for change_date in changes: color = 'green' if prev_regime == 'Bullish' else 'red' ax_recent.axvspan(start, change_date, facecolor=color, alpha=0.2) start = change_date prev_regime = 'Bullish' if prev_regime == 'Bearish' else 'Bearish' # Last segment color = 'green' if prev_regime == 'Bullish' else 'red' ax_recent.axvspan(start, recent_data.index[-1], facecolor=color, alpha=0.2) # Add regime change lines for change_date in changes: ax_recent.axvline(x=change_date, color='blue', linestyle='--', linewidth=1) # Format recent plot ax_recent.set_title(f"Last 12 Months: {display_name}", fontsize=12) ax_recent.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d')) ax_recent.xaxis.set_major_locator(mdates.MonthLocator(interval=2)) plt.setp(ax_recent.get_xticklabels(), rotation=45, ha='right', fontsize=8) ax_recent.grid(True, alpha=0.3) # 4. Probability plot ax_prob = plt.subplot(gs[2, 1]) bullish_prob_col = f'Prob_Regime_{bullish_regime}' ax_prob.plot(recent_data.index, recent_data[bullish_prob_col], color='blue', linewidth=2, label='Bullish Probability') ax_prob.fill_between(recent_data.index, 0, recent_data[bullish_prob_col], color='lightblue', alpha=0.4) # Add horizontal line at 0.5 ax_prob.axhline(y=0.5, color='gray', linestyle='--', alpha=0.7) # Add vertical lines at regime changes for change_date in changes: ax_prob.axvline(x=change_date, color='blue', linestyle='--', linewidth=1) # Format probability plot ax_prob.set_title("Bullish Regime Probability", fontsize=12) ax_prob.set_ylim(0, 1) ax_prob.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d')) ax_prob.xaxis.set_major_locator(mdates.MonthLocator(interval=2)) plt.setp(ax_prob.get_xticklabels(), rotation=45, ha='right', fontsize=8) ax_prob.grid(True, alpha=0.3) # Overall formatting plt.tight_layout(rect=[0, 0.03, 1, 0.95]) # Footer plt.figtext(0.01, 0.01, f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}", fontsize=8, style='italic') plt.figtext(0.99, 0.01, "BNP PARIBAS\nConfidential", ha='right', fontsize=8, style='italic', alpha=0.7) # Add to PDF pdf.savefig(fig) plt.close(fig) # Add a page with recent regime changes and statistics # Create a statistics and transitions page fig = plt.figure(figsize=(11.7, 8.3)) plt.axis('off') plt.text(0.5, 0.95, f"{display_name} Regime Statistics", fontsize=18, ha='center', weight='bold') # Calculate regime statistics total_days = len(regime_df) bullish_days = (regime_df['Market_Condition'] == 'Bullish').sum() bearish_days = (regime_df['Market_Condition'] == 'Bearish').sum() bullish_pct = (bullish_days / total_days) * 100 bearish_pct = (bearish_days / total_days) * 100 transitions = (regime_df['Market_Condition'] != regime_df['Market_Condition'].shift(1)).sum() avg_regime_duration = total_days / max(1, transitions) regime_shifts = regime_df[regime_df['Regime_Change'] == True].copy() # Statistics text stats_text = ( f"Total Analysis Period: {regime_df.index[0].strftime('%Y-%m-%d')} to {regime_df.index[-1].strftime('%Y-%m-%d')}\n\n" f"Time in Bullish Regime: {bullish_days} days ({bullish_pct:.1f}%)\n" f"Time in Bearish Regime: {bearish_days} days ({bearish_pct:.1f}%)\n\n" f"Number of Regime Transitions: {transitions}\n" f"Average Regime Duration: {avg_regime_duration:.1f} days" ) plt.text(0.5, 0.8, stats_text, fontsize=14, ha='center', bbox=dict(facecolor='white', edgecolor='gray', boxstyle='round', alpha=0.9)) # Recent regime changes table if len(regime_shifts) > 0: # Get last 15 regime changes recent_shifts = regime_shifts.tail(15).copy() recent_shifts['Previous Regime'] = recent_shifts['Market_Condition'].shift(1) # The first row will have NaN for previous regime first_regime = 'Bearish' if recent_shifts.iloc[0]['Market_Condition'] == 'Bullish' else 'Bullish' recent_shifts.loc[recent_shifts.index[0], 'Previous Regime'] = first_regime plt.text(0.5, 0.6, "Recent Regime Transitions", fontsize=16, ha='center', weight='bold') # Table header header = ['Date', 'From Regime', 'To Regime', f'{tenor} Value'] col_width = 0.2 # Table positions y_start = 0.55 y_step = 0.03 # Draw header for i, head in enumerate(header): x_pos = 0.2 + i * col_width plt.text(x_pos, y_start, head, fontsize=12, ha='left', weight='bold') # Draw lines plt.plot([0.18, 0.82], [y_start - 0.01, y_start - 0.01], 'k-', alpha=0.3) # Draw data rows for j, (idx, row) in enumerate(recent_shifts.iterrows()): y_pos = y_start - (j+1) * y_step # Date plt.text(0.2, y_pos, idx.strftime('%Y-%m-%d'), fontsize=11, ha='left') # From regime with color from_color = 'green' if row['Previous Regime'] == 'Bullish' else 'red' plt.text(0.4, y_pos, row['Previous Regime'], fontsize=11, ha='left', color=from_color) # To regime with color to_color = 'green' if row['Market_Condition'] == 'Bullish' else 'red' plt.text(0.6, y_pos, row['Market_Condition'], fontsize=11, ha='left', color=to_color) # Tenor value plt.text(0.8, y_pos, f"{row[tenor]:.3f}", fontsize=11, ha='left') else: plt.text(0.5, 0.5, "No regime transitions detected in the data period.", fontsize=14, ha='center', style='italic') # Footer plt.figtext(0.01, 0.01, f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}", fontsize=8, style='italic') plt.figtext(0.99, 0.01, "BNP PARIBAS\nConfidential", ha='right', fontsize=8, style='italic', alpha=0.7) pdf.savefig(fig) plt.close(fig) # Final page with methodology explanation fig = plt.figure(figsize=(11.7, 8.3)) plt.axis('off') plt.text(0.5, 0.95, "Methodology & Technical Notes", fontsize=18, ha='center', weight='bold') methodology_text = """ The ASW Regime Detection Model uses a Hidden Markov Model (HMM) with Markov Switching Regression to identify distinct market regimes based on the behavior of Asset Swap (ASW) spreads. Key aspects of the methodology: 1. Data Processing: • The raw ASW spread data is first differenced to make it stationary • A rolling average with an optimized window size is applied to smooth out noise 2. Regime Identification: • A two-state Markov Switching Model is fitted to the processed data • The model identifies distinct regimes based on the statistical properties of the series • Regimes are classified as 'Bullish' or 'Bearish' based on the average price movement 3. Optimization: • Each ASW tenor uses its own optimized window size for best performance • The window size affects the sensitivity of regime detection 4. Trading Signals: • Regime transitions naturally generate trading signals: - Transitions to Bullish regime → Buy Signal - Transitions to Bearish regime → Sell Signal 5. Probability Tracking: • The model provides probability estimates for each regime • This allows for monitoring regime strength and anticipating potential transitions This model is intended to be run daily to provide updated regime classifications and probability scores. """ plt.text(0.1, 0.8, methodology_text, fontsize=12, ha='left', va='top', bbox=dict(facecolor='white', edgecolor='lightgray', boxstyle='round', alpha=0.9)) # Usage notes usage_notes = """ Usage Notes: • Window Sizes: The window size for each tenor can be customized to adjust sensitivity • Lookback Period: The visualization lookback period can be adjusted (default: 365 days) • Exclusions: Individual tenors can be excluded from analysis if desired The PDF report is automatically generated with all relevant analysis and can be distributed to stakeholders as needed. """ plt.text(0.1, 0.35, usage_notes, fontsize=12, ha='left', va='top', bbox=dict(facecolor='#f0f0f0', edgecolor='lightgray', boxstyle='round', alpha=0.9)) # Footer plt.figtext(0.01, 0.01, f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}", fontsize=8, style='italic') plt.figtext(0.99, 0.01, "BNP PARIBAS\nConfidential", ha='right', fontsize=8, style='italic', alpha=0.7) pdf.savefig(fig) plt.close(fig) print(f"PDF report successfully generated: {pdf_filename}") return pdf_filename except Exception as e: print(f"Error generating PDF report: {str(e)}") import traceback traceback.print_exc() return None_info = plt.subplot(gs[0, :]) ax_info.axis('off') display_name = ASW_TENORS[tenor]['display_name'] window_size = ASW_TENORS[tenor]['window_size'] bullish_regime = regime_char['bullish_regime'] latest_date = regime_df.index[-1] latest_regime = regime_df.loc[latest_date, 'Market_Condition'] bullish_prob = regime_df.loc[latest_date, f'Prob_Regime_{bullish_regime}'] bearish_prob = 1 - bullish_prob regime_color = 'green' if latest_regime == 'Bullish' else 'red' # Create info text info_title = plt.text(0.5, 0.8, f"{display_name} Regime Analysis", fontsize=18, ha='center', weight='bold') info_text = ( f"Window Size: {window_size}\n" f"Current Regime: {latest_regime} (as of {latest_date.strftime('%Y-%m-%d')})\n" f"Bullish Probability: {bullish_prob:.2f} | Bearish Probability: {bearish_prob:.2f}" ) plt.text(0.5, 0.4, info_text, fontsize=12, ha='center', bbox=dict(facecolor='white', alpha=0.8, boxstyle='round'), color=regime_color) # 2. Full history plot ax_history = plt.subplot(gs[1, :]) ax_history.plot(regime_df.index, regime_df[tenor], 'k-', linewidth=1) # Color background based on regime for i in range(len(regime_df)-1): date = regime_df.index[i] next_date = regime_df.index[i+1] regime = regime_df.iloc[i]['Market_Condition'] color = 'green' if regime == 'Bullish' else 'red' ax_history.axvspan(date, next_date, facecolor=color, alpha=0.2) ax_history.set_title(f"Full History: {display_name}", fontsize=14) ax_history.grid(True, alpha=0.3) # 3. Recent history & regime probability plots # Get last year of data recent_start = latest_date - timedelta(days=365) recent_data = regime_df.loc[recent_start:].copy() ax_recent = plt.subplot(gs[2, 0]) ax_recent.plot(recent_data.index, recent_data[tenor], 'k-', linewidth=1.5) # Color background for recent data changes = recent_data[recent_data['Regime_Change'] == True].index.tolist() prev_regime = recent_data.iloc[0]['Market_Condition'] start = recent_data.index[0] for change_date in changes: color = 'green' if prev_regime == 'Bullish' else 'red' ax_recent.axvspan(start, change_date, facecolor=color, alpha=0.2) start = change_date prev_regime = 'Bullish' if prev_regime == 'Bearish' else 'Bearish' # Last segment color = 'green' if prev_regime == 'Bullish' else 'red' ax_recent.axvspan(start, recent_data.index[-1], facecolor=color, alpha=0.2) # Add regime change lines for change_date in changes: ax_recent.axvline(x=change_date, color='blue', linestyle='--', linewidth=1) # Format recent plot ax_recent.set_title(f"Last 12 Months: {display_name}", fontsize=12) ax_recent.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d')) ax_recent.xaxis.set_major_locator(mdates.MonthLocator(interval=2)) plt.setp(ax_recent.get_xticklabels(), rotation=45, ha='right', fontsize=8) ax_recent.grid(True, alpha=0.3) # 4. Probability plot ax_prob = plt.subplot(gs[2, 1]) bullish_prob_col = f'Prob_Regime_{bullish_regime}' ax_prob.plot(recent_data.index, recent_data[bullish_prob_col], color='blue', linewidth=2, label='Bullish Probability') ax_prob.fill_between(recent_data.index, 0, recent_data[bullish_prob_col], color='lightblue', alpha=0.4) # Add horizontal line at 0.5 ax_prob.axhline(y=0.5, color='gray', linestyle='--', alpha=0.7) # Add vertical lines at regime changes for change_date in changes: ax_prob.axvline(x=change_date, color='blue', linestyle='--', linewidth=1) # Format probability plot ax_prob.set_title("Bullish Regime Probability", fontsize=12) ax_prob.set_ylim(0, 1) ax_prob.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d')) ax_prob.xaxis.set_major_locator(mdates.MonthLocator(interval=2)) plt.setp(ax_prob.get_xticklabels(), rotation=45, ha='right', fontsize=8) ax_prob.grid(True, alpha=0.3) # Overall formatting plt.tight_layout(rect=[0, 0.03, 1, 0.95]) # Footer plt.figtext(0.01, 0.01, f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}", fontsize=8, style='italic') plt.figtext(0.99, 0.01, "BNP PARIBAS\nConfidential", ha='right', fontsize=8, style='italic', alpha=0.7) # Add to PDF pdf.savefig(fig) plt.close(fig) # Add a page with recent regime changes and statistics # Create a statistics and transitions page fig = plt.figure(figsize=(11.7, 8.3)) plt.axis('off') plt.text(0.5, 0.95, f"{display_name} Regime Statistics", fontsize=18, ha='center', weight='bold') # Calculate regime statistics total_days = len(regime_df) bullish_days = (regime_df['Market_Condition'] == 'Bullish').sum() bearish_days = (regime_df['Market_Condition'] == 'Bearish').sum() bullish_pct = (bullish_days / total_days) * 100 bearish_pct = (bearish_days / total_days) * 100 transitions = (regime_df['Market_Condition'] != regime_df['Market_Condition'].shift(1)).sum() avg_regime_duration = total_days / max(1, transitions) regime_shifts = regime_df[regime_df['Regime_Change'] == True].copy() # Statistics text stats_text = ( f"Total Analysis Period: {regime_df.index[0].strftime('%Y-%m-%d')} to {regime_df.index[-1].strftime('%Y-%m-%d')}\n\n" f"Time in Bullish Regime: {bullish_days} days ({bullish_pct:.1f}%)\n" f"Time in Bearish Regime: {bearish_days} days ({bearish_pct:.1f}%)\n\n" f"Number of Regime Transitions: {transitions}\n" f"Average Regime Duration: {avg_regime_duration:.1f} days" ) plt.text(0.5, 0.8, stats_text, fontsize=14, ha='center', bbox=dict(facecolor='white', edgecolor='gray', boxstyle='round', alpha=0.9)) # Recent regime changes table if len(regime_shifts) > 0: # Get last 15 regime changes recent_shifts = regime_shifts.tail(15).copy() recent_shifts['Previous Regime'] = recent_shifts['Market_Condition'].shift(1) # The first row will have NaN for previous regime first_regime = 'Bearish' if recent_shifts.iloc[0]['Market_Condition'] == 'Bullish' else 'Bullish' recent_shifts.loc[recent_shifts.index[0], 'Previous Regime'] = first_regime plt.text(0.5, 0.6, "Recent Regime Transitions", fontsize=16, ha='center', weight='bold') # Table header header = ['Date', 'From Regime', 'To Regime', f'{tenor} Value'] col_width = 0.2 # Table positions y_start = 0.55 y_step = 0.03 # Draw header for i, head in enumerate(header): x_pos = 0.2 + i * col_width plt.text(x_pos, y_start, head, fontsize=12, ha='left', weight='bold') # Draw lines plt.plot([0.18, 0.82], [y_start - 0.01, y_start - 0.01], 'k-', alpha=0.3) # Draw data rows for j, (idx, row) in enumerate(recent_shifts.iterrows()): y_pos = y_start - (j+1) * y_step # Date plt.text(0.2, y_pos, idx.strftime('%Y-%m-%d'), fontsize=11, ha='left') # From regime with color from_color = 'green' if row['Previous Regime'] == 'Bullish' else 'red' plt.text(0.4, y_pos, row['Previous Regime'], fontsize=11, ha='left', color=from_color) # To regime with color to_color = 'green' if row['Market_Condition'] == 'Bullish' else 'red' plt.text(0.6, y_pos, row['Market_Condition'], fontsize=11, ha='left', color=to_color) # Tenor value plt.text(0.8, y_pos, f"{row[tenor]:.3f}", fontsize=11, ha='left') else: plt.text(0.5, 0.5, "No regime transitions detected in the data period.", fontsize=14, ha='center', style='italic') # Footer plt.figtext(0.01, 0.01, f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}", fontsize=8, style='italic') plt.figtext(0.99, 0.01, "BNP PARIBAS\nConfidential", ha='right', fontsize=8, style='italic', alpha=0.7) pdf.savefig(fig) plt.close(fig) # Final page with methodology explanation fig = plt.figure(figsize=(11.7, 8.3)) plt.axis('off') plt.text(0.5, 0.95, "Methodology & Technical Notes", fontsize=18, ha='center', weight='bold') methodology_text = """ The ASW Regime Detection Model uses a Hidden Markov Model (HMM) with Markov Switching Regression to identify distinct market regimes based on the behavior of Asset Swap (ASW) spreads. Key aspects of the methodology: 1. Data Processing: • The raw ASW spread data is first differenced to make it stationary • A rolling average with an optimized window size is applied to smooth out noise 2. Regime Identification: • A two-state Markov Switching Model is fitted to the processed data • The model identifies distinct regimes based on the statistical properties of the series • Regimes are classified as 'Bullish' or 'Bearish' based on the average price movement 3. Optimization: • Each ASW tenor uses its own optimized window size for best performance • The window size affects the sensitivity of regime detection 4. Trading Signals: • Regime transitions naturally generate trading signals: - Transitions to Bullish regime → Buy Signal - Transitions to Bearish regime → Sell Signal 5. Probability Tracking: • The model provides probability estimates for each regime • This allows for monitoring regime strength and anticipating potential transitions This model is intended to be run daily to provide updated regime classifications and probability scores. """ plt.text(0.1, 0.8, methodology_text, fontsize=12, ha='left', va='top', bbox=dict(facecolor='white', edgecolor='lightgray', boxstyle='round', alpha=0.9)) # Usage notes usage_notes = """ Usage Notes: • Window Sizes: The window size for each tenor can be customized to adjust sensitivity • Lookback Period: The visualization lookback period can be adjusted (default: 365 days) • Exclusions: Individual tenors can be excluded from analysis if desired The PDF report is automatically generated with all relevant analysis and can be distributed to stakeholders as needed. """ plt.text(0.1, 0.35, usage_notes, fontsize=12, ha='left', va='top', bbox=dict(facecolor='#f0f0f0', edgecolor='lightgray', boxstyle='round', alpha=0.9)) # Footer plt.figtext(0.01, 0.01, f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}", fontsize=8, style='italic') plt.figtext(0.99, 0.01, "BNP PARIBAS\nConfidential", ha='right', fontsize=8, style='italic', alpha=0.7) pdf.savefig(fig) plt.close(fig) print(f"PDF report successfully generated: {pdf_filename}") return pdf_filename except Exception as e: print(f"Error generating PDF report: {str(e)}") import traceback traceback.print_exc() return Noneset_major_formatter(mdates.DateFormatter('%Y-%m-%d')) axes[-1].xaxis.set_major_locator(mdates.WeekdayLocator(interval=2)) axes[-1].tick_params(axis='x', rotation=45) axes[-1].set_xlabel('Date', fontsize=10) # Add title to the entire figure fig.suptitle('ASW Regime Analysis Summary Dashboard', fontsize=16, y=0.98) # Legend for the entire figure buy_patch = patches.Patch(color='green', alpha=0.2, label='Bullish Regime') sell_patch = patches.Patch(color='red', alpha=0.2, label='Bearish Regime') regime_change = plt.Line2D([0], [0], color='blue', linestyle='--', label='Regime Change') fig.legend(handles=[buy_patch, sell_patch, regime_change], loc='upper right', bbox_to_anchor=(0.99, 0.96), fontsize=9) # Add report generation date plt.figtext(0.01, 0.01, f"Report generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}", fontsize=8, style='italic') # BNP Paribas branding plt.figtext(0.99, 0.01, "BNP PARIBAS\nConfidential", ha='right', fontsize=8, style='italic', alpha=0.7) plt.tight_layout(rect=[0, 0.03, 1, 0.95]) # Save results if SAVE_RESULTS: today_str = datetime.now().strftime('%Y%m%d') fig.savefig(f"{RESULTS_DIR}/ASW_summary_dashboard_{today_str}.png", dpi=300, bbox_inches='tight') plt.show() return fig except Exception as e: print(f"Error creating summary dashboard: {str(e)}") import traceback traceback.print_exc() return None # Main function def main(): # Parse command line arguments (if any) args = parse_arguments() lookback_days = args.lookback_days if hasattr(args, 'lookback_days') else 365 # Update data path from arguments if provided data_path = args.data_path if hasattr(args, 'data_path') else DATA_PATH # Check if PDF generation is disabled generate_pdf = not (hasattr(args, 'no_pdf') and args.no_pdf) pdf_filename = args.pdf_filename if hasattr(args, 'pdf_filename') else PDF_FILENAME print("\n*** HMM REGIME DETECTION - MULTI-TENOR PRODUCTION RUN ***") print(f"Run date: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") # Print active tenors and their window sizes active_tenors = [f"{tenor} (Window: {config['window_size']})" for tenor, config in ASW_TENORS.items() if config['include']] print(f"Active tenors: {', '.join(active_tenors)}") # Load and preprocess data df = load_data(data_path) if df is None: print("Error: Could not load data. Exiting.") return # Store results for each tenor tenor_results = {} # Make df available at the module level for other functions to access global_df = df # Process each tenor for tenor in ASW_TENORS: if tenor in df.columns and ASW_TENORS[tenor]['include']: print(f"\n{'-'*70}") print(f"Processing {tenor} with window size {ASW_TENORS[tenor]['window_size']}...") print(f"{'-'*70}") try: # Prepare tenor-specific data tenor_df = prepare_tenor_data(df, tenor) # Fit HMM model model_result = fit_hmm_model(tenor_df, tenor) if model_result is None: print(f"Error: Could not fit model for {tenor}. Skipping.") tenor_results[tenor] = { 'regime_df': None, 'regime_characteristics': None } continue # Get smoothed probabilities smoothed_probs = model_result.smoothed_marginal_probabilities # Determine regime characteristics regime_df, regime_characteristics = determine_regime_characteristics(tenor_df, smoothed_probs, tenor) # Store results tenor_results[tenor] = { 'regime_df': regime_df, 'regime_characteristics': regime_characteristics } # Print tenor's regime summary print_tenor_regime_summary(regime_df, regime_characteristics, tenor) # Visualize tenor results visualize_tenor_results(tenor_df, regime_df, regime_characteristics, tenor, lookback_days) except Exception as e: print(f"Error processing {tenor}: {str(e)}") import traceback traceback.print_exc() tenor_results[tenor] = { 'regime_df': None, 'regime_characteristics': None } # Create summary dashboard for all tenors summary_fig = create_summary_dashboard(tenor_results, lookback_days=90) # Generate PDF report if enabled if generate_pdf: pdf_path = generate_pdf_report(tenor_results, summary_fig, data_path) if pdf_path: print(f"PDF report saved to: {pdf_path}") else: print("PDF report generation skipped.") print("\nProcess completed successfully.") # Run the script if __name__ == "__main__": main() # Store results for each tenor tenor_results = {} # Process each tenor for tenor in ASW_TENORS: if tenor in df.columns and ASW_TENORS[tenor]['include']: print(f"\n{'-'*70}") print(f"Processing {tenor} with window size {ASW_TENORS[tenor]['window_size']}...") print(f"{'-'*70}") try: # Prepare tenor-specific data tenor_df = prepare_tenor_data(df, tenor) # Fit HMM model model_result = fit_hmm_model(tenor_df, tenor) if model_result is None: print(f"Error: Could not fit model for {tenor}. Skipping.") tenor_results[tenor] = { 'regime_df': None, 'regime_characteristics': None } continue # Get smoothed probabilities smoothed_probs = model_result.smoothed_marginal_probabilities # Determine regime characteristics regime_df, regime_characteristics = determine_regime_characteristics(tenor_df, smoothed_probs, tenor) # Store results tenor_results[tenor] = { 'regime_df': regime_df, 'regime_characteristics': regime_characteristics } # Print tenor's regime summary print_tenor_regime_summary(regime_df, regime_characteristics, tenor) # Visualize tenor results visualize_tenor_results(tenor_df, regime_df, regime_characteristics, tenor, lookback_days) except Exception as e: print(f"Error processing {tenor}: {str(e)}") import traceback traceback.print_exc() tenor_results[tenor] = { 'regime_df': None, 'regime_characteristics': None } # Create summary dashboard for all tenors create_summary_dashboard(tenor_results, lookback_days=90) print("\nProcess completed successfully.") # Run the script if __name__ == "__main__": main()

# HMM Regime Detection - Multi-Tenor Production Script # This script runs a Markov Switching Model to identify market regimes # across multiple ASW tenors and visualizes the results import pandas as pd import numpy as np import matplotlib.pyplot as plt import matplotlib.dates as mdates from statsmodels.tsa.regime_switching.markov_regression import MarkovRegression import warnings from datetime import datetime, timedelta import os import matplotlib.patches as patches import argparse import sys # Add this import for sys.modules check from matplotlib.backends.backend_pdf import PdfPages # For PDF generation import io from PIL import Image # For handling images in the PDF import matplotlib.gridspec as gridspec # Suppress warnings for cleaner output warnings.filterwarnings('ignore') # Configuration parameters DATA_PATH = 'ASW_ALL.xlsx' # Update to your data path containing multiple ASW tenors SAVE_RESULTS = True RESULTS_DIR = 'regime_results' GENERATE_PDF = True # Set to generate PDF report PDF_FILENAME = 'ASW_Regime_Analysis_Report.pdf' # Default PDF filename # ASW tenor configuration ASW_TENORS = { 'USSFCT02': {'window_size': 40, 'include': True, 'display_name': 'ASW 2Y'}, 'USSFCT05': {'window_size': 50, 'include': True, 'display_name': 'ASW 5Y'}, 'USSFCT07': {'window_size': 60, 'include': True, 'display_name': 'ASW 7Y'}, 'USSFCT10': {'window_size': 70, 'include': True, 'display_name': 'ASW 10Y'} } # Create results directory if it doesn't exist if SAVE_RESULTS and not os.path.exists(RESULTS_DIR): os.makedirs(RESULTS_DIR) def parse_arguments(): """Parse command line arguments for customizing analysis""" parser = argparse.ArgumentParser(description='HMM Regime Detection for Multiple ASW Tenors') # Add arguments for each tenor's window size for tenor, config in ASW_TENORS.items(): parser.add_argument(f'--window_{tenor.lower()}', type=int, default=config['window_size'], help=f'Rolling window size for {tenor}') parser.add_argument(f'--exclude_{tenor.lower()}', action='store_true', help=f'Exclude {tenor} from the analysis') # Add common arguments parser.add_argument('--lookback_days', type=int, default=365, help='Number of days to look back for visualization') parser.add_argument('--data_path', type=str, default=DATA_PATH, help='Path to the data file') parser.add_argument('--no_pdf', action='store_true', help='Skip PDF report generation') parser.add_argument('--pdf_filename', type=str, default=PDF_FILENAME, help='Custom filename for the PDF report') # For Jupyter environment, don't parse sys.argv which might cause errors try: # Check if running in Jupyter if 'ipykernel' in sys.modules: args = parser.parse_args([]) # Empty list means don't parse any args else: args = parser.parse_args() except Exception as e: print(f"Warning: Argument parsing error: {str(e)}") print("Using default parameters instead.") args = parser.parse_args([]) # Use empty list for defaults # Update the global ASW_TENORS dictionary with parsed arguments for tenor in ASW_TENORS: window_arg = f'window_{tenor.lower()}' exclude_arg = f'exclude_{tenor.lower()}' if hasattr(args, window_arg): ASW_TENORS[tenor]['window_size'] = getattr(args, window_arg) if hasattr(args, exclude_arg) and getattr(args, exclude_arg): ASW_TENORS[tenor]['include'] = False return args # Function to load and preprocess data def load_data(file_path): print(f"Loading data from {file_path}...") try: # Import data df = pd.read_excel(file_path, index_col=0, keep_default_na=False, na_values=['N/A']) # Check if 'Date' is already the index if not isinstance(df.index, pd.DatetimeIndex): # If 'Date' is a column, set it as index if 'Date' in df.columns: df = df.set_index('Date') else: # Check if the first column might be the date column try: # Try to convert the first column to datetime first_col = df.reset_index().iloc[:, 0] first_col_dt = pd.to_datetime(first_col) # If successful, reset and set properly df = df.reset_index() df['Date'] = first_col_dt df = df.set_index('Date') except: # If that fails, convert the existing index to datetime if possible try: df.index = pd.to_datetime(df.index) except: print("Warning: Could not identify or convert date column. Please ensure your data has a proper date column.") # Drop missing values (if any) df = df.dropna() # Check which configured tenors exist in the dataframe available_tenors = [] for tenor in ASW_TENORS: if tenor in df.columns and ASW_TENORS[tenor]['include']: available_tenors.append(tenor) if not available_tenors: print("Error: No configured ASW tenors found in the data file.") print(f"Available columns in the data file: {', '.join(df.columns)}") print(f"Looking for tenors: {', '.join(ASW_TENORS.keys())}") return None print(f"Available ASW tenors for analysis: {', '.join(available_tenors)}") print(f"Data loaded successfully. Shape: {df.shape}") return df except Exception as e: print(f"Error loading data: {str(e)}") import traceback traceback.print_exc() return None # Function to prepare data for a specific tenor def prepare_tenor_data(df, tenor): """Prepare data for a specific ASW tenor, applying its optimal window size""" window_size = ASW_TENORS[tenor]['window_size'] print(f"Preparing {tenor} data with window size {window_size}...") # Make a copy to avoid modifying the original dataframe tenor_df = df.copy() # Compute first difference (rate of change) of interest rates tenor_df[f'{tenor}_Change'] = tenor_df[tenor].diff() # Compute rolling average with the tenor-specific window size tenor_df[f'{tenor}_Trend'] = tenor_df[f'{tenor}_Change'].rolling(window=window_size).mean() # Drop rows with NaN values after creating features tenor_df = tenor_df.dropna() return tenor_df # Function to fit HMM model for a specific tenor def fit_hmm_model(df, tenor): print(f"Fitting Markov Switching Model for {tenor}...") try: # Fit Markov Switching Model model = MarkovRegression( df[f'{tenor}_Trend'], k_regimes=2, trend='c', switching_variance=True ) result = model.fit(maxiter=200, disp=False) print(f"Model for {tenor} fitted successfully.") return result except Exception as e: print(f"Error fitting model for {tenor}: {str(e)}") return None # Function to determine regime characteristics for a specific tenor def determine_regime_characteristics(df, regime_probabilities, tenor): # Create a dataframe with smoothed probabilities regime_df = pd.DataFrame(index=df.index) # Add regime probabilities regime_df['Prob_Regime_0'] = regime_probabilities[0] regime_df['Prob_Regime_1'] = regime_probabilities[1] # Determine the most likely regime for each day regime_df['Regime'] = regime_df[['Prob_Regime_0', 'Prob_Regime_1']].idxmax(axis=1) regime_df['Regime'] = regime_df['Regime'].str.replace('Prob_Regime_', '') # Add price data regime_df[tenor] = df[tenor] # Create a more robust method to identify bullish and bearish regimes # Use linear regression slope for each regime section from scipy import stats # Initialize lists to store slopes slopes_regime_0 = [] slopes_regime_1 = [] # Get consecutive regimes current_regime = None start_idx = None for i, row in regime_df.iterrows(): if current_regime is None: current_regime = row['Regime'] start_idx = i elif row['Regime'] != current_regime: # Regime changed, calculate slope of previous segment segment = regime_df.loc[start_idx:i, tenor] if len(segment) >= 5: # Only calculate if enough data points x = np.arange(len(segment)) slope, _, _, _, _ = stats.linregress(x, segment.values) if current_regime == '0': slopes_regime_0.append(slope) else: slopes_regime_1.append(slope) # Reset for next segment current_regime = row['Regime'] start_idx = i # Don't forget the last segment if start_idx is not None and current_regime is not None: segment = regime_df.loc[start_idx:, tenor] if len(segment) >= 5: x = np.arange(len(segment)) slope, _, _, _, _ = stats.linregress(x, segment.values) if current_regime == '0': slopes_regime_0.append(slope) else: slopes_regime_1.append(slope) # Determine which regime is bullish/bearish based on average slope regime_characteristics = {} if slopes_regime_0 and slopes_regime_1: avg_slope_0 = np.mean(slopes_regime_0) avg_slope_1 = np.mean(slopes_regime_1) print(f"{tenor} - Average slope for Regime 0: {avg_slope_0:.6f}") print(f"{tenor} - Average slope for Regime 1: {avg_slope_1:.6f}") # For credit spreads, negative slope (decreasing) = bullish if avg_slope_0 < avg_slope_1: bullish_regime = '0' bearish_regime = '1' else: bullish_regime = '1' bearish_regime = '0' regime_characteristics = { 'bullish_regime': bullish_regime, 'bearish_regime': bearish_regime } # Label market conditions regime_df['Market_Condition'] = regime_df['Regime'].apply( lambda x: 'Bullish' if x == bullish_regime else 'Bearish' ) else: # Fallback if we couldn't calculate slopes # Use simpler approach based on price differences # First create the column for price changes regime_df['Price_Changes'] = regime_df[tenor].diff() # Calculate average price change by regime avg_change_0 = regime_df[regime_df['Regime'] == '0']['Price_Changes'].mean() avg_change_1 = regime_df[regime_df['Regime'] == '1']['Price_Changes'].mean() print(f"{tenor} - Average price change for Regime 0: {avg_change_0:.6f}") print(f"{tenor} - Average price change for Regime 1: {avg_change_1:.6f}") # For credit spreads, negative change (decreasing) = bullish if avg_change_0 < avg_change_1: bullish_regime = '0' bearish_regime = '1' else: bullish_regime = '1' bearish_regime = '0' regime_characteristics = { 'bullish_regime': bullish_regime, 'bearish_regime': bearish_regime } # Label market conditions regime_df['Market_Condition'] = regime_df['Regime'].apply( lambda x: 'Bullish' if x == bullish_regime else 'Bearish' ) # Detect regime changes regime_df['Regime_Change'] = regime_df['Market_Condition'].ne(regime_df['Market_Condition'].shift(1)) return regime_df, regime_characteristics # Function to print tenor's regime summary def print_tenor_regime_summary(regime_df, regime_characteristics, tenor): try: bullish_regime = regime_characteristics.get('bullish_regime') if not bullish_regime: print(f"Error: Could not determine bullish regime for {tenor}.") return latest_date = regime_df.index[-1] latest_regime = regime_df.loc[latest_date, 'Market_Condition'] # Handle the case where the bullish probability column might not exist try: bullish_prob = regime_df.loc[latest_date, f'Prob_Regime_{bullish_regime}'] bearish_prob = 1 - bullish_prob except KeyError: # If we can't get the specific probability, use the regime determination bullish_prob = 1.0 if latest_regime == 'Bullish' else 0.0 bearish_prob = 1.0 if latest_regime == 'Bearish' else 0.0 print("\n" + "="*50) print(f"REGIME SUMMARY FOR {tenor} ON {latest_date.strftime('%Y-%m-%d')}") print("="*50) print(f"CURRENT MARKET REGIME: {latest_regime}") print(f"Bullish Probability: {bullish_prob:.4f}") print(f"Bearish Probability: {bearish_prob:.4f}") # Check for regime change if len(regime_df) > 1: previous_date = regime_df.index[-2] previous_regime = regime_df.loc[previous_date, 'Market_Condition'] if previous_regime != latest_regime: print("\n*** REGIME CHANGE DETECTED ***") print(f"Previous regime ({previous_date.strftime('%Y-%m-%d')}): {previous_regime}") print(f"New regime ({latest_date.strftime('%Y-%m-%d')}): {latest_regime}") print("="*50) except Exception as e: print(f"Error in print_tenor_regime_summary for {tenor}: {str(e)}") import traceback traceback.print_exc() # Function to visualize results for a specific tenor def visualize_tenor_results(df, regime_df, regime_characteristics, tenor, lookback_days=365): print(f"Generating visualization for {tenor}...") try: # Extract regime characteristics bullish_regime = regime_characteristics['bullish_regime'] bearish_regime = regime_characteristics['bearish_regime'] # Get data for visualization if lookback_days is None or lookback_days <= 0: plot_df = df.copy() plot_regime_df = regime_df.copy() else: # Get data for the lookback period end_date = df.index[-1] start_date = end_date - timedelta(days=int(lookback_days)) plot_df = df.loc[start_date:end_date].copy() plot_regime_df = regime_df.loc[start_date:end_date].copy() # Create figure with specific size fig = plt.figure(figsize=(12, 8)) # Set up grid for two plots with different heights gs = plt.GridSpec(2, 1, height_ratios=[3, 1], figure=fig) # Create top and bottom axes ax1 = fig.add_subplot(gs[0]) ax2 = fig.add_subplot(gs[1]) # ===== Top Plot: Price with Regime Classification ===== # Find regime change points regime_changes = plot_regime_df[plot_regime_df['Regime_Change'] == True].index.tolist() # Get all date ranges for each regime regime_periods = [] if len(plot_regime_df) > 0: # Start with the first regime start_date = plot_regime_df.index[0] current_regime = plot_regime_df.iloc[0]['Market_Condition'] for change_date in regime_changes: # Add the period that just ended regime_periods.append({ 'start': start_date, 'end': change_date, 'regime': current_regime }) # Start new period start_date = change_date # Get the new regime (after the change) current_regime = 'Bullish' if current_regime == 'Bearish' else 'Bearish' # Add the final period regime_periods.append({ 'start': start_date, 'end': plot_regime_df.index[-1], 'regime': current_regime }) # Color the background based on regime periods for period in regime_periods: color = 'green' if period['regime'] == 'Bullish' else 'red' ax1.axvspan(period['start'], period['end'], facecolor=color, alpha=0.2, edgecolor=None) # Plot the price line on top ax1.plot(plot_regime_df.index, plot_regime_df[tenor], 'k-', linewidth=1.5, label=tenor) # Add vertical lines at regime changes with labels for change_date in regime_changes: ax1.axvline(x=change_date, color='blue', linestyle='--', linewidth=1.5) # Get the new regime after the change date (might be the next day) # Use the next day's regime to determine if it's a buy or sell signal next_dates = plot_regime_df.loc[change_date:].index if len(next_dates) > 1: next_date = next_dates[1] # Get the date after the change new_regime = plot_regime_df.loc[next_date, 'Market_Condition'] # Add regime change annotation y_pos = plot_regime_df.loc[change_date, tenor] if new_regime == 'Bullish': marker = '^' # up triangle for buy marker_color = 'green' else: marker = 'v' # down triangle for sell marker_color = 'red' # Add marker at regime change ax1.scatter(change_date, y_pos, marker=marker, color=marker_color, s=120, zorder=5, edgecolors='black') # Add Buy/Sell signals in legend buy_patch = patches.Patch(color='green', alpha=0.2, label='Bullish Regime') sell_patch = patches.Patch(color='red', alpha=0.2, label='Bearish Regime') buy_signal = plt.Line2D([0], [0], marker='^', color='w', markerfacecolor='green', markersize=10, label='Buy Signal') sell_signal = plt.Line2D([0], [0], marker='v', color='w', markerfacecolor='red', markersize=10, label='Sell Signal') # Add legend ax1.legend(handles=[buy_patch, sell_patch, buy_signal, sell_signal], loc='upper right', framealpha=0.7) # Set title and labels ticker_name = ASW_TENORS[tenor]['display_name'] ax1.set_title(f'{ticker_name} with Regime Classification', fontsize=14) ax1.set_ylabel(ticker_name, fontsize=12) ax1.grid(True, alpha=0.3) # Format x-axis ax1.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d')) ax1.xaxis.set_major_locator(mdates.MonthLocator(interval=1)) plt.setp(ax1.get_xticklabels(), rotation=45, ha='right') # ===== Bottom Plot: Regime Probabilities ===== # Plot only the Bullish probability with a clean blue line bullish_prob_col = f'Prob_Regime_{bullish_regime}' ax2.plot(plot_regime_df.index, plot_regime_df[bullish_prob_col], color='blue', linewidth=2, label='Bullish Probability') # Add light blue filled area under the line ax2.fill_between(plot_regime_df.index, 0, plot_regime_df[bullish_prob_col], color='lightblue', alpha=0.4) # Add a horizontal reference line at 0.5 ax2.axhline(y=0.5, color='gray', linestyle='--', alpha=0.7) # Add vertical lines at the same regime change points for change_date in regime_changes: ax2.axvline(x=change_date, color='blue', linestyle='--', linewidth=1.5) # Set title and labels ax2.set_title('Bullish Regime Probability Over Time', fontsize=14) ax2.set_ylabel('Probability', fontsize=12) ax2.set_xlabel('Date', fontsize=12) # Set y-axis limits ax2.set_ylim(0, 1) # Format x-axis to match top plot ax2.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d')) ax2.xaxis.set_major_locator(mdates.MonthLocator(interval=1)) plt.setp(ax2.get_xticklabels(), rotation=45, ha='right') # Add grid ax2.grid(True, alpha=0.3) # Add current regime information to figure latest_date = regime_df.index[-1] latest_regime = regime_df.loc[latest_date, 'Market_Condition'] latest_prob = regime_df.loc[latest_date, bullish_prob_col] regime_color = 'green' if latest_regime == 'Bullish' else 'red' # Create summary text summary_text = f"Latest Date: {latest_date.strftime('%Y-%m-%d')}\n" summary_text += f"Current Regime: {latest_regime}\n" summary_text += f"Bullish Probability: {latest_prob:.2f}" # Add a text box with the latest regime info (as a figure annotation) plt.figtext(0.02, 0.02, summary_text, fontsize=12, bbox=dict(facecolor='white', alpha=0.8, boxstyle='round'), color=regime_color) # Adjust layout plt.tight_layout() # Add subtle BNP Paribas branding in bottom right corner plt.figtext(0.98, 0.01, "BNP PARIBAS\nConfidential", ha='right', fontsize=8, style='italic', alpha=0.7) # Save results if requested if SAVE_RESULTS: today_str = datetime.now().strftime('%Y%m%d') fig.savefig(f"{RESULTS_DIR}/{tenor}_regime_analysis_{today_str}.png", dpi=300, bbox_inches='tight') # Also save the regime data regime_df.to_csv(f"{RESULTS_DIR}/{tenor}_regime_data_{today_str}.csv") plt.show() return fig except Exception as e: print(f"Error visualizing results for {tenor}: {str(e)}") import traceback traceback.print_exc() return None # Function to create summary dashboard for all tenors def create_summary_dashboard(tenor_results, lookback_days=90): print("Creating summary dashboard for all tenors...") try: # Count how many tenors we're dealing with active_tenors = [tenor for tenor, data in tenor_results.items() if data['regime_df'] is not None] num_tenors = len(active_tenors) if num_tenors == 0: print("No tenor data available for summary dashboard.") return None # Create a figure with subplots - one row per tenor fig, axes = plt.subplots(num_tenors, 1, figsize=(12, 4*num_tenors), sharex=True) # Handle the case where there's only one tenor if num_tenors == 1: axes = [axes] # End date will be the same for all tenors (today) end_date = tenor_results[active_tenors[0]]['regime_df'].index[-1] start_date = end_date - timedelta(days=lookback_days) # Plot each tenor for i, tenor in enumerate(active_tenors): ax = axes[i] regime_df = tenor_results[tenor]['regime_df'] regime_char = tenor_results[tenor]['regime_characteristics'] # Filter data for the lookback period plot_data = regime_df.loc[start_date:end_date].copy() # Get bullish regime identifier bullish_regime = regime_char['bullish_regime'] # Plot the price line ax.plot(plot_data.index, plot_data[tenor], 'k-', linewidth=1.5) # Color the background prev_date = plot_data.index[0] prev_regime = plot_data.iloc[0]['Market_Condition'] # Find regime changes changes = plot_data[plot_data['Regime_Change'] == True].index.tolist() # Add the first period start = plot_data.index[0] for change_date in changes: color = 'green' if prev_regime == 'Bullish' else 'red' ax.axvspan(start, change_date, facecolor=color, alpha=0.2) # Update for next period start = change_date prev_regime = 'Bullish' if prev_regime == 'Bearish' else 'Bearish' # Add the final period color = 'green' if prev_regime == 'Bullish' else 'red' ax.axvspan(start, plot_data.index[-1], facecolor=color, alpha=0.2) # Add vertical lines at regime changes for change_date in changes: ax.axvline(x=change_date, color='blue', linestyle='--', linewidth=1) # Current regime info current_regime = plot_data.iloc[-1]['Market_Condition'] bullish_prob = plot_data.iloc[-1][f'Prob_Regime_{bullish_regime}'] # Set title and labels display_name = ASW_TENORS[tenor]['display_name'] window_size = ASW_TENORS[tenor]['window_size'] ax.set_title(f"{display_name} - Current: {current_regime} (Prob: {bullish_prob:.2f}, Window: {window_size})", fontsize=10, loc='left') ax.set_ylabel(display_name, fontsize=9) ax.grid(True, alpha=0.3) ax.tick_params(axis='y', labelsize=8) # Format x-axis (only for bottom plot) axes[-1].xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d')) axes[-1].xaxis.set_major_locator(mdates.WeekdayLocator(interval=2)) axes[-1].tick_params(axis='x', rotation=45) axes[-1].set_xlabel('Date', fontsize=10) # Add title to the entire figure fig.suptitle('ASW Regime Analysis Summary Dashboard', fontsize=16, y=0.98) # Legend for the entire figure buy_patch = patches.Patch(color='green', alpha=0.2, label='Bullish Regime') sell_patch = patches.Patch(color='red', alpha=0.2, label='Bearish Regime') regime_change = plt.Line2D([0], [0], color='blue', linestyle='--', label='Regime Change') fig.legend(handles=[buy_patch, sell_patch, regime_change], loc='upper right', bbox_to_anchor=(0.99, 0.96), fontsize=9) # Add report generation date plt.figtext(0.01, 0.01, f"Report generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}", fontsize=8, style='italic') # BNP Paribas branding plt.figtext(0.99, 0.01, "BNP PARIBAS\nConfidential", ha='right', fontsize=8, style='italic', alpha=0.7) plt.tight_layout(rect=[0, 0.03, 1, 0.95]) # Save results if SAVE_RESULTS: today_str = datetime.now().strftime('%Y%m%d') fig.savefig(f"{RESULTS_DIR}/ASW_summary_dashboard_{today_str}.png", dpi=300, bbox_inches='tight') return fig except Exception as e: print(f"Error creating summary dashboard: {str(e)}") import traceback traceback.print_exc() return None # Function to generate a PDF report of all results def generate_pdf_report(tenor_results, summary_fig, data_path): """Generate a comprehensive PDF report with all visualizations and results""" print("\nGenerating PDF report...") today_str = datetime.now().strftime('%Y%m%d') pdf_filename = f"{RESULTS_DIR}/{PDF_FILENAME.replace('.pdf', '')}_{today_str}.pdf" try: with PdfPages(pdf_filename) as pdf: # Title page fig = plt.figure(figsize=(11.7, 8.3)) # A4 size in inches plt.axis('off') # Title and subtitle plt.text(0.5, 0.8, "ASW Regime Detection Analysis", fontsize=24, ha='center', weight='bold') plt.text(0.5, 0.7, f"Generated on {datetime.now().strftime('%Y-%m-%d %H:%M')}", fontsize=14, ha='center') plt.text(0.5, 0.6, f"Data source: {os.path.basename(data_path)}", fontsize=14, ha='center') # Active tenors info active_tenors = [f"{tenor} (Window: {ASW_TENORS[tenor]['window_size']})" for tenor, data in tenor_results.items() if data['regime_df'] is not None] plt.text(0.5, 0.5, "Analyzed Tenors:", fontsize=16, ha='center') for i, tenor_text in enumerate(active_tenors): plt.text(0.5, 0.45 - i*0.05, tenor_text, fontsize=12, ha='center') # BNP Paribas footer plt.text(0.5, 0.05, "BNP PARIBAS\nCONFIDENTIAL", fontsize=10, ha='center', weight='bold') pdf.savefig(fig) plt.close(fig) # Table of Contents page fig = plt.figure(figsize=(11.7, 8.3)) # A4 size in inches plt.axis('off') plt.text(0.5, 0.9, "Table of Contents", fontsize=20, ha='center', weight='bold') content_items = [ ("1. Summary Dashboard", 0.8), ("2. Individual Tenor Analysis", 0.75) ] # Add each tenor to the TOC for i, tenor in enumerate([t for t in tenor_results if tenor_results[t]['regime_df'] is not None]): display_name = ASW_TENORS[tenor]['display_name'] content_items.append((f" 2.{i+1}. {display_name}", 0.7 - i*0.04)) for text, y_pos in content_items: plt.text(0.3, y_pos, text, fontsize=14, ha='left') pdf.savefig(fig) plt.close(fig) # Add summary dashboard if summary_fig is not None: pdf.savefig(summary_fig) plt.close(summary_fig) # Add individual tenor analysis pages for tenor, data in tenor_results.items(): if data['regime_df'] is None: continue regime_df = data['regime_df'] regime_char = data['regime_characteristics'] # Create a page with multiple visualizations for each tenor fig = plt.figure(figsize=(11.7, 8.3)) # A4 size gs = gridspec.GridSpec(3, 2, height_ratios=[1, 2, 2]) # 1. Tenor info section ax_info = plt.subplot(gs[0, :]) ax_info.axis('off') display_name = ASW_TENORS[tenor]['display_name'] window_size = ASW_TENORS[tenor]['window_size'] bullish_regime = regime_char['bullish_regime'] latest_date = regime_df.index[-1] latest_regime = regime_df.loc[latest_date, 'Market_Condition'] bullish_prob = regime_df.loc[latest_date, f'Prob_Regime_{bullish_regime}'] bearish_prob = 1 - bullish_prob regime_color = 'green' if latest_regime == 'Bullish' else 'red' # Create info text info_title = plt.text(0.5, 0.8, f"{display_name} Regime Analysis", fontsize=18, ha='center', weight='bold') info_text = ( f"Window Size: {window_size}\n" f"Current Regime: {latest_regime} (as of {latest_date.strftime('%Y-%m-%d')})\n" f"Bullish Probability: {bullish_prob:.2f} | Bearish Probability: {bearish_prob:.2f}" ) plt.text(0.5, 0.4, info_text, fontsize=12, ha='center', bbox=dict(facecolor='white', alpha=0.8, boxstyle='round'), color=regime_color) # 2. Full history plot ax_history = plt.subplot(gs[1, :]) ax_history.plot(regime_df.index, regime_df[tenor], 'k-', linewidth=1) # Color background based on regime for i in range(len(regime_df)-1): date = regime_df.index[i] next_date = regime_df.index[i+1] regime = regime_df.iloc[i]['Market_Condition'] color = 'green' if regime == 'Bullish' else 'red' ax_history.axvspan(date, next_date, facecolor=color, alpha=0.2) ax_history.set_title(f"Full History: {display_name}", fontsize=14) ax_history.grid(True, alpha=0.3) # 3. Recent history & regime probability plots # Get last year of data recent_start = latest_date - timedelta(days=365) recent_data = regime_df.loc[recent_start:].copy() ax_recent = plt.subplot(gs[2, 0]) ax_recent.plot(recent_data.index, recent_data[tenor], 'k-', linewidth=1.5) # Color background for recent data changes = recent_data[recent_data['Regime_Change'] == True].index.tolist() prev_regime = recent_data.iloc[0]['Market_Condition'] start = recent_data.index[0] for change_date in changes: color = 'green' if prev_regime == 'Bullish' else 'red' ax_recent.axvspan(start, change_date, facecolor=color, alpha=0.2) start = change_date prev_regime = 'Bullish' if prev_regime == 'Bearish' else 'Bearish' # Last segment color = 'green' if prev_regime == 'Bullish' else 'red' ax_recent.axvspan(start, recent_data.index[-1], facecolor=color, alpha=0.2) # Add regime change lines for change_date in changes: ax_recent.axvline(x=change_date, color='blue', linestyle='--', linewidth=1) # Format recent plot ax_recent.set_title(f"Last 12 Months: {display_name}", fontsize=12) ax_recent.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d')) ax_recent.xaxis.set_major_locator(mdates.MonthLocator(interval=2)) plt.setp(ax_recent.get_xticklabels(), rotation=45, ha='right', fontsize=8) ax_recent.grid(True, alpha=0.3) # 4. Probability plot ax_prob = plt.subplot(gs[2, 1]) bullish_prob_col = f'Prob_Regime_{bullish_regime}' ax_prob.plot(recent_data.index, recent_data[bullish_prob_col], color='blue', linewidth=2, label='Bullish Probability') ax_prob.fill_between(recent_data.index, 0, recent_data[bullish_prob_col], color='lightblue', alpha=0.4) # Add horizontal line at 0.5 ax_prob.axhline(y=0.5, color='gray', linestyle='--', alpha=0.7) # Add vertical lines at regime changes for change_date in changes: ax_prob.axvline(x=change_date, color='blue', linestyle='--', linewidth=1) # Format probability plot ax_prob.set_title("Bullish Regime Probability", fontsize=12) ax_prob.set_ylim(0, 1) ax_prob.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d')) ax_prob.xaxis.set_major_locator(mdates.MonthLocator(interval=2)) plt.setp(ax_prob.get_xticklabels(), rotation=45, ha='right', fontsize=8) ax_prob.grid(True, alpha=0.3) # Overall formatting plt.tight_layout(rect=[0, 0.03, 1, 0.95]) # Footer plt.figtext(0.01, 0.01, f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}", fontsize=8, style='italic') plt.figtext(0.99, 0.01, "BNP PARIBAS\nConfidential", ha='right', fontsize=8, style='italic', alpha=0.7) # Add to PDF pdf.savefig(fig) plt.close(fig) # Add a page with recent regime changes and statistics # Create a statistics and transitions page fig = plt.figure(figsize=(11.7, 8.3)) plt.axis('off') plt.text(0.5, 0.95, f"{display_name} Regime Statistics", fontsize=18, ha='center', weight='bold') # Calculate regime statistics total_days = len(regime_df) bullish_days = (regime_df['Market_Condition'] == 'Bullish').sum() bearish_days = (regime_df['Market_Condition'] == 'Bearish').sum() bullish_pct = (bullish_days / total_days) * 100 bearish_pct = (bearish_days / total_days) * 100 transitions = (regime_df['Market_Condition'] != regime_df['Market_Condition'].shift(1)).sum() avg_regime_duration = total_days / max(1, transitions) regime_shifts = regime_df[regime_df['Regime_Change'] == True].copy() # Statistics text stats_text = ( f"Total Analysis Period: {regime_df.index[0].strftime('%Y-%m-%d')} to {regime_df.index[-1].strftime('%Y-%m-%d')}\n\n" f"Time in Bullish Regime: {bullish_days} days ({bullish_pct:.1f}%)\n" f"Time in Bearish Regime: {bearish_days} days ({bearish_pct:.1f}%)\n\n" f"Number of Regime Transitions: {transitions}\n" f"Average Regime Duration: {avg_regime_duration:.1f} days" ) plt.text(0.5, 0.8, stats_text, fontsize=14, ha='center', bbox=dict(facecolor='white', edgecolor='gray', boxstyle='round', alpha=0.9)) # Recent regime changes table if len(regime_shifts) > 0: # Get last 15 regime changes recent_shifts = regime_shifts.tail(15).copy() recent_shifts['Previous Regime'] = recent_shifts['Market_Condition'].shift(1) # The first row will have NaN for previous regime first_regime = 'Bearish' if recent_shifts.iloc[0]['Market_Condition'] == 'Bullish' else 'Bullish' recent_shifts.loc[recent_shifts.index[0], 'Previous Regime'] = first_regime plt.text(0.5, 0.6, "Recent Regime Transitions", fontsize=16, ha='center', weight='bold') # Table header header = ['Date', 'From Regime', 'To Regime', f'{tenor} Value'] col_width = 0.2 # Table positions y_start = 0.55 y_step = 0.03 # Draw header for i, head in enumerate(header): x_pos = 0.2 + i * col_width plt.text(x_pos, y_start, head, fontsize=12, ha='left', weight='bold') # Draw lines plt.plot([0.18, 0.82], [y_start - 0.01, y_start - 0.01], 'k-', alpha=0.3) # Draw data rows for j, (idx, row) in enumerate(recent_shifts.iterrows()): y_pos = y_start - (j+1) * y_step # Date plt.text(0.2, y_pos, idx.strftime('%Y-%m-%d'), fontsize=11, ha='left') # From regime with color from_color = 'green' if row['Previous Regime'] == 'Bullish' else 'red' plt.text(0.4, y_pos, row['Previous Regime'], fontsize=11, ha='left', color=from_color) # To regime with color to_color = 'green' if row['Market_Condition'] == 'Bullish' else 'red' plt.text(0.6, y_pos, row['Market_Condition'], fontsize=11, ha='left', color=to_color) # Tenor value plt.text(0.8, y_pos, f"{row[tenor]:.3f}", fontsize=11, ha='left') else: plt.text(0.5, 0.5, "No regime transitions detected in the data period.", fontsize=14, ha='center', style='italic') # Footer plt.figtext(0.01, 0.01, f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}", fontsize=8, style='italic') plt.figtext(0.99, 0.01, "BNP PARIBAS\nConfidential", ha='right', fontsize=8, style='italic', alpha=0.7) pdf.savefig(fig) plt.close(fig) # Final page with methodology explanation fig = plt.figure(figsize=(11.7, 8.3)) plt.axis('off') plt.text(0.5, 0.95, "Methodology & Technical Notes", fontsize=18, ha='center', weight='bold') methodology_text = """ The ASW Regime Detection Model uses a Hidden Markov Model (HMM) with Markov Switching Regression to identify distinct market regimes based on the behavior of Asset Swap (ASW) spreads. Key aspects of the methodology: 1. Data Processing: • The raw ASW spread data is first differenced to make it stationary • A rolling average with an optimized window size is applied to smooth out noise 2. Regime Identification: • A two-state Markov Switching Model is fitted to the processed data • The model identifies distinct regimes based on the statistical properties of the series • Regimes are classified as 'Bullish' or 'Bearish' based on the average price movement 3. Optimization: • Each ASW tenor uses its own optimized window size for best performance • The window size affects the sensitivity of regime detection 4. Trading Signals: • Regime transitions naturally generate trading signals: - Transitions to Bullish regime → Buy Signal - Transitions to Bearish regime → Sell Signal 5. Probability Tracking: • The model provides probability estimates for each regime • This allows for monitoring regime strength and anticipating potential transitions This model is intended to be run daily to provide updated regime classifications and probability scores. """ plt.text(0.1, 0.8, methodology_text, fontsize=12, ha='left', va='top', bbox=dict(facecolor='white', edgecolor='lightgray', boxstyle='round', alpha=0.9)) # Usage notes usage_notes = """ Usage Notes: • Window Sizes: The window size for each tenor can be customized to adjust sensitivity • Lookback Period: The visualization lookback period can be adjusted (default: 365 days) • Exclusions: Individual tenors can be excluded from analysis if desired The PDF report is automatically generated with all relevant analysis and can be distributed to stakeholders as needed. """ plt.text(0.1, 0.35, usage_notes, fontsize=12, ha='left', va='top', bbox=dict(facecolor='#f0f0f0', edgecolor='lightgray', boxstyle='round', alpha=0.9)) # Footer plt.figtext(0.01, 0.01, f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}", fontsize=8, style='italic') plt.figtext(0.99, 0.01, "BNP PARIBAS\nConfidential", ha='right', fontsize=8, style='italic', alpha=0.7) pdf.savefig(fig) plt.close(fig) print(f"PDF report successfully generated: {pdf_filename}") return pdf_filename except Exception as e: print(f"Error generating PDF report: {str(e)}") import traceback traceback.print_exc() return Noneset_major_formatter(mdates.DateFormatter('%Y-%m-%d')) axes[-1].xaxis.set_major_locator(mdates.WeekdayLocator(interval=2)) axes[-1].tick_params(axis='x', rotation=45) axes[-1].set_xlabel('Date', fontsize=10) # Add title to the entire figure fig.suptitle('ASW Regime Analysis Summary Dashboard', fontsize=16, y=0.98) # Legend for the entire figure buy_patch = patches.Patch(color='green', alpha=0.2, label='Bullish Regime') sell_patch = patches.Patch(color='red', alpha=0.2, label='Bearish Regime') regime_change = plt.Line2D([0], [0], color='blue', linestyle='--', label='Regime Change') fig.legend(handles=[buy_patch, sell_patch, regime_change], loc='upper right', bbox_to_anchor=(0.99, 0.96), fontsize=9) # Add report generation date plt.figtext(0.01, 0.01, f"Report generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}", fontsize=8, style='italic') # BNP Paribas branding plt.figtext(0.99, 0.01, "BNP PARIBAS\nConfidential", ha='right', fontsize=8, style='italic', alpha=0.7) plt.tight_layout(rect=[0, 0.03, 1, 0.95]) # Save results if SAVE_RESULTS: today_str = datetime.now().strftime('%Y%m%d') fig.savefig(f"{RESULTS_DIR}/ASW_summary_dashboard_{today_str}.png", dpi=300, bbox_inches='tight') plt.show() return fig except Exception as e: print(f"Error creating summary dashboard: {str(e)}") import traceback traceback.print_exc() return None # Main function def main(): # Parse command line arguments (if any) args = parse_arguments() lookback_days = args.lookback_days if hasattr(args, 'lookback_days') else 365 # Update data path from arguments if provided data_path = args.data_path if hasattr(args, 'data_path') else DATA_PATH # Check if PDF generation is disabled generate_pdf = not (hasattr(args, 'no_pdf') and args.no_pdf) pdf_filename = args.pdf_filename if hasattr(args, 'pdf_filename') else PDF_FILENAME print("\n*** HMM REGIME DETECTION - MULTI-TENOR PRODUCTION RUN ***") print(f"Run date: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") # Print active tenors and their window sizes active_tenors = [f"{tenor} (Window: {config['window_size']})" for tenor, config in ASW_TENORS.items() if config['include']] print(f"Active tenors: {', '.join(active_tenors)}") # Load and preprocess data df = load_data(data_path) if df is None: print("Error: Could not load data. Exiting.") return # Store results for each tenor tenor_results = {} # Process each tenor for tenor in ASW_TENORS: if tenor in df.columns and ASW_TENORS[tenor]['include']: print(f"\n{'-'*70}") print(f"Processing {tenor} with window size {ASW_TENORS[tenor]['window_size']}...") print(f"{'-'*70}") try: # Prepare tenor-specific data tenor_df = prepare_tenor_data(df, tenor) # Fit HMM model model_result = fit_hmm_model(tenor_df, tenor) if model_result is None: print(f"Error: Could not fit model for {tenor}. Skipping.") tenor_results[tenor] = { 'regime_df': None, 'regime_characteristics': None } continue # Get smoothed probabilities smoothed_probs = model_result.smoothed_marginal_probabilities # Determine regime characteristics regime_df, regime_characteristics = determine_regime_characteristics(tenor_df, smoothed_probs, tenor) # Store results tenor_results[tenor] = { 'regime_df': regime_df, 'regime_characteristics': regime_characteristics } # Print tenor's regime summary print_tenor_regime_summary(regime_df, regime_characteristics, tenor) # Visualize tenor results visualize_tenor_results(tenor_df, regime_df, regime_characteristics, tenor, lookback_days) except Exception as e: print(f"Error processing {tenor}: {str(e)}") import traceback traceback.print_exc() tenor_results[tenor] = { 'regime_df': None, 'regime_characteristics': None } # Create summary dashboard for all tenors summary_fig = create_summary_dashboard(tenor_results, lookback_days=90) # Generate PDF report if enabled if generate_pdf: pdf_path = generate_pdf_report(tenor_results, summary_fig, data_path) if pdf_path: print(f"PDF report saved to: {pdf_path}") else: print("PDF report generation skipped.") print("\nProcess completed successfully.") # Run the script if __name__ == "__main__": main() # Store results for each tenor tenor_results = {} # Process each tenor for tenor in ASW_TENORS: if tenor in df.columns and ASW_TENORS[tenor]['include']: print(f"\n{'-'*70}") print(f"Processing {tenor} with window size {ASW_TENORS[tenor]['window_size']}...") print(f"{'-'*70}") try: # Prepare tenor-specific data tenor_df = prepare_tenor_data(df, tenor) # Fit HMM model model_result = fit_hmm_model(tenor_df, tenor) if model_result is None: print(f"Error: Could not fit model for {tenor}. Skipping.") tenor_results[tenor] = { 'regime_df': None, 'regime_characteristics': None } continue # Get smoothed probabilities smoothed_probs = model_result.smoothed_marginal_probabilities # Determine regime characteristics regime_df, regime_characteristics = determine_regime_characteristics(tenor_df, smoothed_probs, tenor) # Store results tenor_results[tenor] = { 'regime_df': regime_df, 'regime_characteristics': regime_characteristics } # Print tenor's regime summary print_tenor_regime_summary(regime_df, regime_characteristics, tenor) # Visualize tenor results visualize_tenor_results(tenor_df, regime_df, regime_characteristics, tenor, lookback_days) except Exception as e: print(f"Error processing {tenor}: {str(e)}") import traceback traceback.print_exc() tenor_results[tenor] = { 'regime_df': None, 'regime_characteristics': None } # Create summary dashboard for all tenors create_summary_dashboard(tenor_results, lookback_days=90) print("\nProcess completed successfully.") # Run the script if __name__ == "__main__": main()

# Function to generate a PDF report of all results def generate_pdf_report(tenor_results, summary_fig, data_path): """Generate a comprehensive PDF report with all visualizations and results""" print("\nGenerating PDF report...") today_str = datetime.now().strftime('%Y%m%d') pdf_filename = f"{RESULTS_DIR}/{PDF_FILENAME.replace('.pdf', '')}_{today_str}.pdf" try: with PdfPages(pdf_filename) as pdf: # Title page fig = plt.figure(figsize=(11.7, 8.3)) # A4 size in inches plt.axis('off') # Title and subtitle plt.text(0.5, 0.8, "ASW Regime Detection Analysis", fontsize=24, ha='center', weight='bold') plt.text(0.5, 0.7, f"Generated on {datetime.now().strftime('%Y-%m-%d %H:%M')}", fontsize=14, ha='center') plt.text(0.5, 0.6, f"Data source: {os.path.basename(data_path)}", fontsize=14, ha='center') # Active tenors info active_tenors = [f"{tenor} (Window: {ASW_TENORS[tenor]['window_size']})" for tenor, data in tenor_results.items() if data['regime_df'] is not None] plt.text(0.5, 0.5, "Analyzed Tenors:", fontsize=16, ha='center') for i, tenor_text in enumerate(active_tenors): plt.text(0.5, 0.45 - i*0.05, tenor_text, fontsize=12, ha='center') # BNP Paribas footer plt.text(0.5, 0.05, "BNP PARIBAS\nCONFIDENTIAL", fontsize=10, ha='center', weight='bold') pdf.savefig(fig) plt.close(fig) # Table of Contents page fig = plt.figure(figsize=(11.7, 8.3)) # A4 size in inches plt.axis('off') plt.text(0.5, 0.9, "Table of Contents", fontsize=20, ha='center', weight='bold') content_items = [ ("1. Summary Dashboard", 0.8), ("2. Individual Tenor Analysis", 0.75) ] # Add each tenor to the TOC for i, tenor in enumerate([t for t in tenor_results if tenor_results[t]['regime_df'] is not None]): display_name = ASW_TENORS[tenor]['display_name'] content_items.append((f" 2.{i+1}. {display_name}", 0.7 - i*0.04)) for text, y_pos in content_items: plt.text(0.3, y_pos, text, fontsize=14, ha='left') pdf.savefig(fig) plt.close(fig) # Add summary dashboard if summary_fig is not None: pdf.savefig(summary_fig) plt.close(summary_fig) # Add individual tenor analysis pages for tenor, data in tenor_results.items(): if data['regime_df'] is None: continue regime_df = data['regime_df'] regime_char = data['regime_characteristics'] # Create a page with multiple visualizations for each tenor fig = plt.figure(figsize=(11.7, 8.3)) # A4 size gs = gridspec.GridSpec(3, 2, height_ratios=[1, 2, 2]) # 1. Tenor info section ax_info = plt.subplot(gs[0, :]) ax_info.axis('off') display_name = ASW_TENORS[tenor]['display_name'] window_size = ASW_TENORS[tenor]['window_size'] bullish_regime = regime_char['bullish_regime'] latest_date = regime_df.index[-1] latest_regime = regime_df.loc[latest_date, 'Market_Condition'] bullish_prob = regime_df.loc[latest_date, f'Prob_Regime_{bullish_regime}'] bearish_prob = 1 - bullish_prob regime_color = 'green' if latest_regime == 'Bullish' else 'red' # Create info text info_title = plt.text(0.5, 0.8, f"{display_name} Regime Analysis", fontsize=18, ha='center', weight='bold') info_text = ( f"Window Size: {window_size}\n" f"Current Regime: {latest_regime} (as of {latest_date.strftime('%Y-%m-%d')})\n" f"Bullish Probability: {bullish_prob:.2f} | Bearish Probability: {bearish_prob:.2f}" ) plt.text(0.5, 0.4, info_text, fontsize=12, ha='center', bbox=dict(facecolor='white', alpha=0.8, boxstyle='round'), color=regime_color) # 2. Full history plot ax_history = plt.subplot(gs[1, :]) ax_history.plot(regime_df.index, regime_df[tenor], 'k-', linewidth=1) # Color background based on regime for i in range(len(regime_df)-1): date = regime_df.index[i] next_date = regime_df.index[i+1] regime = regime_df.iloc[i]['Market_Condition'] color = 'green' if regime == 'Bullish' else 'red' ax_history.axvspan(date, next_date, facecolor=color, alpha=0.2) ax_history.set_title(f"Full History: {display_name}", fontsize=14) ax_history.grid(True, alpha=0.3) # 3. Recent history & regime probability plots # Get last year of data recent_start = latest_date - timedelta(days=365) recent_data = regime_df.loc[recent_start:].copy() ax_recent = plt.subplot(gs[2, 0]) ax_recent.plot(recent_data.index, recent_data[tenor], 'k-', linewidth=1.5) # Color background for recent data changes = recent_data[recent_data['Regime_Change'] == True].index.tolist() prev_regime = recent_data.iloc[0]['Market_Condition'] start = recent_data.index[0] for change_date in changes: color = 'green' if prev_regime == 'Bullish' else 'red' ax_recent.axvspan(start, change_date, facecolor=color, alpha=0.2) start = change_date prev_regime = 'Bullish' if prev_regime == 'Bearish' else 'Bearish' # Last segment color = 'green' if prev_regime == 'Bullish' else 'red' ax_recent.axvspan(start, recent_data.index[-1], facecolor=color, alpha=0.2) # Add regime change lines for change_date in changes: ax_recent.axvline(x=change_date, color='blue', linestyle='--', linewidth=1) # Format recent plot ax_recent.set_title(f"Last 12 Months: {display_name}", fontsize=12) ax_recent.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d')) ax_recent.xaxis.set_major_locator(mdates.MonthLocator(interval=2)) plt.setp(ax_recent.get_xticklabels(), rotation=45, ha='right', fontsize=8) ax_recent.grid(True, alpha=0.3) # 4. Probability plot ax_prob = plt.subplot(gs[2, 1]) bullish_prob_col = f'Prob_Regime_{bullish_regime}' ax_prob.plot(recent_data.index, recent_data[bullish_prob_col], color='blue', linewidth=2, label='Bullish Probability') ax_prob.fill_between(recent_data.index, 0, recent_data[bullish_prob_col], color='lightblue', alpha=0.4) # Add horizontal line at 0.5 ax_prob.axhline(y=0.5, color='gray', linestyle='--', alpha=0.7) # Add vertical lines at regime changes for change_date in changes: ax_prob.axvline(x=change_date, color='blue', linestyle='--', linewidth=1) # Format probability plot ax_prob.set_title("Bullish Regime Probability", fontsize=12) ax_prob.set_ylim(0, 1) ax_prob.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d')) ax_prob.xaxis.set_major_locator(mdates.MonthLocator(interval=2)) plt.setp(ax_prob.get_xticklabels(), rotation=45, ha='right', fontsize=8) ax_prob.grid(True, alpha=0.3) # Overall formatting plt.tight_layout(rect=[0, 0.03, 1, 0.95]) # Footer plt.figtext(0.01, 0.01, f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}", fontsize=8, style='italic') plt.figtext(0.99, 0.01, "BNP PARIBAS\nConfidential", ha='right', fontsize=8, style='italic', alpha=0.7) # Add to PDF pdf.savefig(fig) plt.close(fig) # Add a page with recent regime changes and statistics # Create a statistics and transitions page fig = plt.figure(figsize=(11.7, 8.3)) plt.axis('off') plt.text(0.5, 0.95, f"{display_name} Regime Statistics", fontsize=18, ha='center', weight='bold') # Calculate regime statistics total_days = len(regime_df) bullish_days = (regime_df['Market_Condition'] == 'Bullish').sum() bearish_days = (regime_df['Market_Condition'] == 'Bearish').sum() bullish_pct = (bullish_days / total_days) * 100 bearish_pct = (bearish_days / total_days) * 100 transitions = (regime_df['Market_Condition'] != regime_df['Market_Condition'].shift(1)).sum() avg_regime_duration = total_days / max(1, transitions) regime_shifts = regime_df[regime_df['Regime_Change'] == True].copy() # Statistics text stats_text = ( f"Total Analysis Period: {regime_df.index[0].strftime('%Y-%m-%d')} to {regime_df.index[-1].strftime('%Y-%m-%d')}\n\n" f"Time in Bullish Regime: {bullish_days} days ({bullish_pct:.1f}%)\n" f"Time in Bearish Regime: {bearish_days} days ({bearish_pct:.1f}%)\n\n" f"Number of Regime Transitions: {transitions}\n" f"Average Regime Duration: {avg_regime_duration:.1f} days" ) plt.text(0.5, 0.8, stats_text, fontsize=14, ha='center', bbox=dict(facecolor='white', edgecolor='gray', boxstyle='round', alpha=0.9)) # Recent regime changes table if len(regime_shifts) > 0: # Get last 15 regime changes recent_shifts = regime_shifts.tail(15).copy() recent_shifts['Previous Regime'] = recent_shifts['Market_Condition'].shift(1) # The first row will have NaN for previous regime first_regime = 'Bearish' if recent_shifts.iloc[0]['Market_Condition'] == 'Bullish' else 'Bullish' recent_shifts.loc[recent_shifts.index[0], 'Previous Regime'] = first_regime plt.text(0.5, 0.6, "Recent Regime Transitions", fontsize=16, ha='center', weight='bold') # Table header header = ['Date', 'From Regime', 'To Regime', f'{tenor} Value'] col_width = 0.2 # Table positions y_start = 0.55 y_step = 0.03 # Draw header for i, head in enumerate(header): x_pos = 0.2 + i * col_width plt.text(x_pos, y_start, head, fontsize=12, ha='left', weight='bold') # Draw lines plt.plot([0.18, 0.82], [y_start - 0.01, y_start - 0.01], 'k-', alpha=0.3) # Draw data rows for j, (idx, row) in enumerate(recent_shifts.iterrows()): y_pos = y_start - (j+1) * y_step # Date plt.text(0.2, y_pos, idx.strftime('%Y-%m-%d'), fontsize=11, ha='left') # From regime with color from_color = 'green' if row['Previous Regime'] == 'Bullish' else 'red' plt.text(0.4, y_pos, row['Previous Regime'], fontsize=11, ha='left', color=from_color) # To regime with color to_color = 'green' if row['Market_Condition'] == 'Bullish' else 'red' plt.text(0.6, y_pos, row['Market_Condition'], fontsize=11, ha='left', color=to_color) # Tenor value plt.text(0.8, y_pos, f"{row[tenor]:.3f}", fontsize=11, ha='left') else: plt.text(0.5, 0.5, "No regime transitions detected in the data period.", fontsize=14, ha='center', style='italic') # Footer plt.figtext(0.01, 0.01, f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}", fontsize=8, style='italic') plt.figtext(0.99, 0.01, "BNP PARIBAS\nConfidential", ha='right', fontsize=8, style='italic', alpha=0.7) pdf.savefig(fig) plt.close(fig) # Final page with methodology explanation fig = plt.figure(figsize=(11.7, 8.3)) plt.axis('off') plt.text(0.5, 0.95, "Methodology & Technical Notes", fontsize=18, ha='center', weight='bold') methodology_text = """ The ASW Regime Detection Model uses a Hidden Markov Model (HMM) with Markov Switching Regression to identify distinct market regimes based on the behavior of Asset Swap (ASW) spreads. Key aspects of the methodology: 1. Data Processing: • The raw ASW spread data is first differenced to make it stationary • A rolling average with an optimized window size is applied to smooth out noise 2. Regime Identification: • A two-state Markov Switching Model is fitted to the processed data • The model identifies distinct regimes based on the statistical properties of the series • Regimes are classified as 'Bullish' or 'Bearish' based on the average price movement 3. Optimization: • Each ASW tenor uses its own optimized window size for best performance • The window size affects the sensitivity of regime detection 4. Trading Signals: • Regime transitions naturally generate trading signals: - Transitions to Bullish regime → Buy Signal - Transitions to Bearish regime → Sell Signal 5. Probability Tracking: • The model provides probability estimates for each regime • This allows for monitoring regime strength and anticipating potential transitions This model is intended to be run daily to provide updated regime classifications and probability scores. """ plt.text(0.1, 0.8, methodology_text, fontsize=12, ha='left', va='top', bbox=dict(facecolor='white', edgecolor='lightgray', boxstyle='round', alpha=0.9)) # Usage notes usage_notes = """ Usage Notes: • Window Sizes: The window size for each tenor can be customized to adjust sensitivity • Lookback Period: The visualization lookback period can be adjusted (default: 365 days) • Exclusions: Individual tenors can be excluded from analysis if desired The PDF report is automatically generated with all relevant analysis and can be distributed to stakeholders as needed. """ plt.text(0.1, 0.35, usage_notes, fontsize=12, ha='left', va='top', bbox=dict(facecolor='#f0f0f0', edgecolor='lightgray', boxstyle='round', alpha=0.9)) # Footer plt.figtext(0.01, 0.01, f"Generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}", fontsize=8, style='italic') plt.figtext(0.99, 0.01, "BNP PARIBAS\nConfidential", ha='right', fontsize=8, style='italic', alpha=0.7) pdf.savefig(fig) plt.close(fig) print(f"PDF report successfully generated: {pdf_filename}") return pdf_filename except Exception as e: print(f"Error generating PDF report: {str(e)}") import traceback traceback.print_exc() return Noneset_major_formatter(mdates.DateFormatter('%Y-%m-%d')) axes[-1].xaxis.set_major_locator(mdates.WeekdayLocator(interval=2)) axes[-1].tick_params(axis='x', rotation=45) axes[-1].set_xlabel('Date', fontsize=10) # Add title to the entire figure fig.suptitle('ASW Regime Analysis Summary Dashboard', fontsize=16, y=0.98) # Legend for the entire figure buy_patch = patches.Patch(color='green', alpha=0.2, label='Bullish Regime') sell_patch = patches.Patch(color='red', alpha=0.2, label='Bearish Regime') regime_change = plt.Line2D([0], [0], color='blue', linestyle='--', label='Regime Change') fig.legend(handles=[buy_patch, sell_patch, regime_change], loc='upper right', bbox_to_anchor=(0.99, 0.96), fontsize=9) # Add report generation date plt.figtext(0.01, 0.01, f"Report generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}", fontsize=8, style='italic') # BNP Paribas branding plt.figtext(0.99, 0.01, "BNP PARIBAS\nConfidential", ha='right', fontsize=8, style='italic', alpha=0.7) plt.tight_layout(rect=[0, 0.03, 1, 0.95]) # Save results if SAVE_RESULTS: today_str = datetime.now().strftime('%Y%m%d') fig.savefig(f"{RESULTS_DIR}/ASW_summary_dashboard_{today_str}.png", dpi=300, bbox_inches='tight') plt.show() return fig except Exception as e: print(f"Error creating summary dashboard: {str(e)}") import traceback traceback.print_exc() return None # Main function def main(): # Parse command line arguments (if any) args = parse_arguments() lookback_days = args.lookback_days if hasattr(args, 'lookback_days') else 365 # Update data path from arguments if provided data_path = args.data_path if hasattr(args, 'data_path') else DATA_PATH # Check if PDF generation is disabled generate_pdf = not (hasattr(args, 'no_pdf') and args.no_pdf) pdf_filename = args.pdf_filename if hasattr(args, 'pdf_filename') else PDF_FILENAME print("\n*** HMM REGIME DETECTION - MULTI-TENOR PRODUCTION RUN ***") print(f"Run date: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}") # Print active tenors and their window sizes active_tenors = [f"{tenor} (Window: {config['window_size']})" for tenor, config in ASW_TENORS.items() if config['include']] print(f"Active tenors: {', '.join(active_tenors)}") # Load and preprocess data df = load_data(data_path) if df is None: print("Error: Could not load data. Exiting.") return # Store results for each tenor tenor_results = {} # Process each tenor for tenor in ASW_TENORS: if tenor in df.columns and ASW_TENORS[tenor]['include']: print(f"\n{'-'*70}") print(f"Processing {tenor} with window size {ASW_TENORS[tenor]['window_size']}...") print(f"{'-'*70}") try: # Prepare tenor-specific data tenor_df = prepare_tenor_data(df, tenor) # Fit HMM model model_result = fit_hmm_model(tenor_df, tenor) if model_result is None: print(f"Error: Could not fit model for {tenor}. Skipping.") tenor_results[tenor] = { 'regime_df': None, 'regime_characteristics': None } continue # Get smoothed probabilities smoothed_probs = model_result.smoothed_marginal_probabilities # Determine regime characteristics regime_df, regime_characteristics = determine_regime_characteristics(tenor_df, smoothed_probs, tenor) # Store results tenor_results[tenor] = { 'regime_df': regime_df, 'regime_characteristics': regime_characteristics } # Print tenor's regime summary print_tenor_regime_summary(regime_df, regime_characteristics, tenor) # Visualize tenor results visualize_tenor_results(tenor_df, regime_df, regime_characteristics, tenor, lookback_days) except Exception as e: print(f"Error processing {tenor}: {str(e)}") import traceback traceback.print_exc() tenor_results[tenor] = { 'regime_df': None, 'regime_characteristics': None } # Create summary dashboard for all tenors summary_fig = create_summary_dashboard(tenor_results, lookback_days=90) # Generate PDF report if enabled if generate_pdf: pdf_path = generate_pdf_report(tenor_results, summary_fig, data_path) if pdf_path: print(f"PDF report saved to: {pdf_path}") else: print("PDF report generation skipped.") print("\nProcess completed successfully.") # Run the script if __name__ == "__main__": main() # Store results for each tenor tenor_results = {} # Process each tenor for tenor in ASW_TENORS: if tenor in df.columns and ASW_TENORS[tenor]['include']: print(f"\n{'-'*70}") print(f"Processing {tenor} with window size {ASW_TENORS[tenor]['window_size']}...") print(f"{'-'*70}") try: # Prepare tenor-specific data tenor_df = prepare_tenor_data(df, tenor) # Fit HMM model model_result = fit_hmm_model(tenor_df, tenor) if model_result is None: print(f"Error: Could not fit model for {tenor}. Skipping.") tenor_results[tenor] = { 'regime_df': None, 'regime_characteristics': None } continue # Get smoothed probabilities smoothed_probs = model_result.smoothed_marginal_probabilities # Determine regime characteristics regime_df, regime_characteristics = determine_regime_characteristics(tenor_df, smoothed_probs, tenor) # Store results tenor_results[tenor] = { 'regime_df': regime_df, 'regime_characteristics': regime_characteristics } # Print tenor's regime summary print_tenor_regime_summary(regime_df, regime_characteristics, tenor) # Visualize tenor results visualize_tenor_results(tenor_df, regime_df, regime_characteristics, tenor, lookback_days) except Exception as e: print(f"Error processing {tenor}: {str(e)}") import traceback traceback.print_exc() tenor_results[tenor] = { 'regime_df': None, 'regime_characteristics': None } # Create summary dashboard for all tenors create_summary_dashboard(tenor_results, lookback_days=90) print("\nProcess completed successfully.") # Run the script if __name__ == "__main__": main()

# HMM Regime Detection - Multi-Tenor Production Script # This script runs a Markov Switching Model to identify market regimes # across multiple ASW tenors and visualizes the results import pandas as pd import numpy as np import matplotlib.pyplot as plt import matplotlib.dates as mdates from statsmodels.tsa.regime_switching.markov_regression import MarkovRegression import warnings from datetime import datetime, timedelta import os import matplotlib.patches as patches import argparse import sys # Add this import for sys.modules check from matplotlib.backends.backend_pdf import PdfPages # For PDF generation import io from PIL import Image # For handling images in the PDF import matplotlib.gridspec as gridspec # Suppress warnings for cleaner output warnings.filterwarnings('ignore') # Configuration parameters DATA_PATH = 'ASW_ALL.xlsx' # Update to your data path containing multiple ASW tenors SAVE_RESULTS = True RESULTS_DIR = 'regime_results' GENERATE_PDF = True # Set to generate PDF report PDF_FILENAME = 'ASW_Regime_Analysis_Report.pdf' # Default PDF filename # ASW tenor configuration ASW_TENORS = { 'USSFCT02': {'window_size': 40, 'include': True, 'display_name': 'ASW 2Y'}, 'USSFCT05': {'window_size': 50, 'include': True, 'display_name': 'ASW 5Y'}, 'USSFCT07': {'window_size': 60, 'include': True, 'display_name': 'ASW 7Y'}, 'USSFCT10': {'window_size': 70, 'include': True, 'display_name': 'ASW 10Y'} } # Create results directory if it doesn't exist if SAVE_RESULTS and not os.path.exists(RESULTS_DIR): os.makedirs(RESULTS_DIR) def parse_arguments(): """Parse command line arguments for customizing analysis""" parser = argparse.ArgumentParser(description='HMM Regime Detection for Multiple ASW Tenors') # Add arguments for each tenor's window size for tenor, config in ASW_TENORS.items(): parser.add_argument(f'--window_{tenor.lower()}', type=int, default=config['window_size'], help=f'Rolling window size for {tenor}') parser.add_argument(f'--exclude_{tenor.lower()}', action='store_true', help=f'Exclude {tenor} from the analysis') # Add common arguments parser.add_argument('--lookback_days', type=int, default=365, help='Number of days to look back for visualization') parser.add_argument('--data_path', type=str, default=DATA_PATH, help='Path to the data file') parser.add_argument('--no_pdf', action='store_true', help='Skip PDF report generation') parser.add_argument('--pdf_filename', type=str, default=PDF_FILENAME, help='Custom filename for the PDF report') # For Jupyter environment, don't parse sys.argv which might cause errors try: # Check if running in Jupyter if 'ipykernel' in sys.modules: args = parser.parse_args([]) # Empty list means don't parse any args else: args = parser.parse_args() except Exception as e: print(f"Warning: Argument parsing error: {str(e)}") print("Using default parameters instead.") args = parser.parse_args([]) # Use empty list for defaults # Update the global ASW_TENORS dictionary with parsed arguments for tenor in ASW_TENORS: window_arg = f'window_{tenor.lower()}' exclude_arg = f'exclude_{tenor.lower()}' if hasattr(args, window_arg): ASW_TENORS[tenor]['window_size'] = getattr(args, window_arg) if hasattr(args, exclude_arg) and getattr(args, exclude_arg): ASW_TENORS[tenor]['include'] = False return args # Function to load and preprocess data def load_data(file_path): print(f"Loading data from {file_path}...") try: # Import data df = pd.read_excel(file_path, index_col=0, keep_default_na=False, na_values=['N/A']) # Check if 'Date' is already the index if not isinstance(df.index, pd.DatetimeIndex): # If 'Date' is a column, set it as index if 'Date' in df.columns: df = df.set_index('Date') else: # Check if the first column might be the date column try: # Try to convert the first column to datetime first_col = df.reset_index().iloc[:, 0] first_col_dt = pd.to_datetime(first_col) # If successful, reset and set properly df = df.reset_index() df['Date'] = first_col_dt df = df.set_index('Date') except: # If that fails, convert the existing index to datetime if possible try: df.index = pd.to_datetime(df.index) except: print("Warning: Could not identify or convert date column. Please ensure your data has a proper date column.") # Drop missing values (if any) df = df.dropna() # Check which configured tenors exist in the dataframe available_tenors = [] for tenor in ASW_TENORS: if tenor in df.columns and ASW_TENORS[tenor]['include']: available_tenors.append(tenor) if not available_tenors: print("Error: No configured ASW tenors found in the data file.") print(f"Available columns in the data file: {', '.join(df.columns)}") print(f"Looking for tenors: {', '.join(ASW_TENORS.keys())}") return None print(f"Available ASW tenors for analysis: {', '.join(available_tenors)}") print(f"Data loaded successfully. Shape: {df.shape}") return df except Exception as e: print(f"Error loading data: {str(e)}") import traceback traceback.print_exc() return None # Function to prepare data for a specific tenor def prepare_tenor_data(df, tenor): """Prepare data for a specific ASW tenor, applying its optimal window size""" window_size = ASW_TENORS[tenor]['window_size'] print(f"Preparing {tenor} data with window size {window_size}...") # Make a copy to avoid modifying the original dataframe tenor_df = df.copy() # Compute first difference (rate of change) of interest rates tenor_df[f'{tenor}_Change'] = tenor_df[tenor].diff() # Compute rolling average with the tenor-specific window size tenor_df[f'{tenor}_Trend'] = tenor_df[f'{tenor}_Change'].rolling(window=window_size).mean() # Drop rows with NaN values after creating features tenor_df = tenor_df.dropna() return tenor_df # Function to fit HMM model for a specific tenor def fit_hmm_model(df, tenor): print(f"Fitting Markov Switching Model for {tenor}...") try: # Fit Markov Switching Model model = MarkovRegression( df[f'{tenor}_Trend'], k_regimes=2, trend='c', switching_variance=True ) result = model.fit(maxiter=200, disp=False) print(f"Model for {tenor} fitted successfully.") return result except Exception as e: print(f"Error fitting model for {tenor}: {str(e)}") return None # Function to determine regime characteristics for a specific tenor def determine_regime_characteristics(df, regime_probabilities, tenor): # Create a dataframe with smoothed probabilities regime_df = pd.DataFrame(index=df.index) # Add regime probabilities regime_df['Prob_Regime_0'] = regime_probabilities[0] regime_df['Prob_Regime_1'] = regime_probabilities[1] # Determine the most likely regime for each day regime_df['Regime'] = regime_df[['Prob_Regime_0', 'Prob_Regime_1']].idxmax(axis=1) regime_df['Regime'] = regime_df['Regime'].str.replace('Prob_Regime_', '') # Add price data regime_df[tenor] = df[tenor] # Create a more robust method to identify bullish and bearish regimes # Use linear regression slope for each regime section from scipy import stats # Initialize lists to store slopes slopes_regime_0 = [] slopes_regime_1 = [] # Get consecutive regimes current_regime = None start_idx = None for i, row in regime_df.iterrows(): if current_regime is None: current_regime = row['Regime'] start_idx = i elif row['Regime'] != current_regime: # Regime changed, calculate slope of previous segment segment = regime_df.loc[start_idx:i, tenor] if len(segment) >= 5: # Only calculate if enough data points x = np.arange(len(segment)) slope, _, _, _, _ = stats.linregress(x, segment.values) if current_regime == '0': slopes_regime_0.append(slope) else: slopes_regime_1.append(slope) # Reset for next segment current_regime = row['Regime'] start_idx = i # Don't forget the last segment if start_idx is not None and current_regime is not None: segment = regime_df.loc[start_idx:, tenor] if len(segment) >= 5: x = np.arange(len(segment)) slope, _, _, _, _ = stats.linregress(x, segment.values) if current_regime == '0': slopes_regime_0.append(slope) else: slopes_regime_1.append(slope) # Determine which regime is bullish/bearish based on average slope regime_characteristics = {} if slopes_regime_0 and slopes_regime_1: avg_slope_0 = np.mean(slopes_regime_0) avg_slope_1 = np.mean(slopes_regime_1) print(f"{tenor} - Average slope for Regime 0: {avg_slope_0:.6f}") print(f"{tenor} - Average slope for Regime 1: {avg_slope_1:.6f}") # For credit spreads, negative slope (decreasing) = bullish if avg_slope_0 < avg_slope_1: bullish_regime = '0' bearish_regime = '1' else: bullish_regime = '1' bearish_regime = '0' regime_characteristics = { 'bullish_regime': bullish_regime, 'bearish_regime': bearish_regime } # Label market conditions regime_df['Market_Condition'] = regime_df['Regime'].apply( lambda x: 'Bullish' if x == bullish_regime else 'Bearish' ) else: # Fallback if we couldn't calculate slopes # Use simpler approach based on price differences # First create the column for price changes regime_df['Price_Changes'] = regime_df[tenor].diff() # Calculate average price change by regime avg_change_0 = regime_df[regime_df['Regime'] == '0']['Price_Changes'].mean() avg_change_1 = regime_df[regime_df['Regime'] == '1']['Price_Changes'].mean() print(f"{tenor} - Average price change for Regime 0: {avg_change_0:.6f}") print(f"{tenor} - Average price change for Regime 1: {avg_change_1:.6f}") # For credit spreads, negative change (decreasing) = bullish if avg_change_0 < avg_change_1: bullish_regime = '0' bearish_regime = '1' else: bullish_regime = '1' bearish_regime = '0' regime_characteristics = { 'bullish_regime': bullish_regime, 'bearish_regime': bearish_regime } # Label market conditions regime_df['Market_Condition'] = regime_df['Regime'].apply( lambda x: 'Bullish' if x == bullish_regime else 'Bearish' ) # Detect regime changes regime_df['Regime_Change'] = regime_df['Market_Condition'].ne(regime_df['Market_Condition'].shift(1)) return regime_df, regime_characteristics # Function to print tenor's regime summary def print_tenor_regime_summary(regime_df, regime_characteristics, tenor): try: bullish_regime = regime_characteristics.get('bullish_regime') if not bullish_regime: print(f"Error: Could not determine bullish regime for {tenor}.") return latest_date = regime_df.index[-1] latest_regime = regime_df.loc[latest_date, 'Market_Condition'] # Handle the case where the bullish probability column might not exist try: bullish_prob = regime_df.loc[latest_date, f'Prob_Regime_{bullish_regime}'] bearish_prob = 1 - bullish_prob except KeyError: # If we can't get the specific probability, use the regime determination bullish_prob = 1.0 if latest_regime == 'Bullish' else 0.0 bearish_prob = 1.0 if latest_regime == 'Bearish' else 0.0 print("\n" + "="*50) print(f"REGIME SUMMARY FOR {tenor} ON {latest_date.strftime('%Y-%m-%d')}") print("="*50) print(f"CURRENT MARKET REGIME: {latest_regime}") print(f"Bullish Probability: {bullish_prob:.4f}") print(f"Bearish Probability: {bearish_prob:.4f}") # Check for regime change if len(regime_df) > 1: previous_date = regime_df.index[-2] previous_regime = regime_df.loc[previous_date, 'Market_Condition'] if previous_regime != latest_regime: print("\n*** REGIME CHANGE DETECTED ***") print(f"Previous regime ({previous_date.strftime('%Y-%m-%d')}): {previous_regime}") print(f"New regime ({latest_date.strftime('%Y-%m-%d')}): {latest_regime}") print("="*50) except Exception as e: print(f"Error in print_tenor_regime_summary for {tenor}: {str(e)}") import traceback traceback.print_exc() # Function to visualize results for a specific tenor def visualize_tenor_results(df, regime_df, regime_characteristics, tenor, lookback_days=365): print(f"Generating visualization for {tenor}...") try: # Extract regime characteristics bullish_regime = regime_characteristics['bullish_regime'] bearish_regime = regime_characteristics['bearish_regime'] # Get data for visualization if lookback_days is None or lookback_days <= 0: plot_df = df.copy() plot_regime_df = regime_df.copy() else: # Get data for the lookback period end_date = df.index[-1] start_date = end_date - timedelta(days=int(lookback_days)) plot_df = df.loc[start_date:end_date].copy() plot_regime_df = regime_df.loc[start_date:end_date].copy() # Create figure with specific size fig = plt.figure(figsize=(12, 8)) # Set up grid for two plots with different heights gs = plt.GridSpec(2, 1, height_ratios=[3, 1], figure=fig) # Create top and bottom axes ax1 = fig.add_subplot(gs[0]) ax2 = fig.add_subplot(gs[1]) # ===== Top Plot: Price with Regime Classification ===== # Find regime change points regime_changes = plot_regime_df[plot_regime_df['Regime_Change'] == True].index.tolist() # Get all date ranges for each regime regime_periods = [] if len(plot_regime_df) > 0: # Start with the first regime start_date = plot_regime_df.index[0] current_regime = plot_regime_df.iloc[0]['Market_Condition'] for change_date in regime_changes: # Add the period that just ended regime_periods.append({ 'start': start_date, 'end': change_date, 'regime': current_regime }) # Start new period start_date = change_date # Get the new regime (after the change) current_regime = 'Bullish' if current_regime == 'Bearish' else 'Bearish' # Add the final period regime_periods.append({ 'start': start_date, 'end': plot_regime_df.index[-1], 'regime': current_regime }) # Color the background based on regime periods for period in regime_periods: color = 'green' if period['regime'] == 'Bullish' else 'red' ax1.axvspan(period['start'], period['end'], facecolor=color, alpha=0.2, edgecolor=None) # Plot the price line on top ax1.plot(plot_regime_df.index, plot_regime_df[tenor], 'k-', linewidth=1.5, label=tenor) # Add vertical lines at regime changes with labels for change_date in regime_changes: ax1.axvline(x=change_date, color='blue', linestyle='--', linewidth=1.5) # Get the new regime after the change date (might be the next day) # Use the next day's regime to determine if it's a buy or sell signal next_dates = plot_regime_df.loc[change_date:].index if len(next_dates) > 1: next_date = next_dates[1] # Get the date after the change new_regime = plot_regime_df.loc[next_date, 'Market_Condition'] # Add regime change annotation y_pos = plot_regime_df.loc[change_date, tenor] if new_regime == 'Bullish': marker = '^' # up triangle for buy marker_color = 'green' else: marker = 'v' # down triangle for sell marker_color = 'red' # Add marker at regime change ax1.scatter(change_date, y_pos, marker=marker, color=marker_color, s=120, zorder=5, edgecolors='black') # Add Buy/Sell signals in legend buy_patch = patches.Patch(color='green', alpha=0.2, label='Bullish Regime') sell_patch = patches.Patch(color='red', alpha=0.2, label='Bearish Regime') buy_signal = plt.Line2D([0], [0], marker='^', color='w', markerfacecolor='green', markersize=10, label='Buy Signal') sell_signal = plt.Line2D([0], [0], marker='v', color='w', markerfacecolor='red', markersize=10, label='Sell Signal') # Add legend ax1.legend(handles=[buy_patch, sell_patch, buy_signal, sell_signal], loc='upper right', framealpha=0.7) # Set title and labels ticker_name = ASW_TENORS[tenor]['display_name'] ax1.set_title(f'{ticker_name} with Regime Classification', fontsize=14) ax1.set_ylabel(ticker_name, fontsize=12) ax1.grid(True, alpha=0.3) # Format x-axis ax1.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d')) ax1.xaxis.set_major_locator(mdates.MonthLocator(interval=1)) plt.setp(ax1.get_xticklabels(), rotation=45, ha='right') # ===== Bottom Plot: Regime Probabilities ===== # Plot only the Bullish probability with a clean blue line bullish_prob_col = f'Prob_Regime_{bullish_regime}' ax2.plot(plot_regime_df.index, plot_regime_df[bullish_prob_col], color='blue', linewidth=2, label='Bullish Probability') # Add light blue filled area under the line ax2.fill_between(plot_regime_df.index, 0, plot_regime_df[bullish_prob_col], color='lightblue', alpha=0.4) # Add a horizontal reference line at 0.5 ax2.axhline(y=0.5, color='gray', linestyle='--', alpha=0.7) # Add vertical lines at the same regime change points for change_date in regime_changes: ax2.axvline(x=change_date, color='blue', linestyle='--', linewidth=1.5) # Set title and labels ax2.set_title('Bullish Regime Probability Over Time', fontsize=14) ax2.set_ylabel('Probability', fontsize=12) ax2.set_xlabel('Date', fontsize=12) # Set y-axis limits ax2.set_ylim(0, 1) # Format x-axis to match top plot ax2.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d')) ax2.xaxis.set_major_locator(mdates.MonthLocator(interval=1)) plt.setp(ax2.get_xticklabels(), rotation=45, ha='right') # Add grid ax2.grid(True, alpha=0.3) # Add current regime information to figure latest_date = regime_df.index[-1] latest_regime = regime_df.loc[latest_date, 'Market_Condition'] latest_prob = regime_df.loc[latest_date, bullish_prob_col] regime_color = 'green' if latest_regime == 'Bullish' else 'red' # Create summary text summary_text = f"Latest Date: {latest_date.strftime('%Y-%m-%d')}\n" summary_text += f"Current Regime: {latest_regime}\n" summary_text += f"Bullish Probability: {latest_prob:.2f}" # Add a text box with the latest regime info (as a figure annotation) plt.figtext(0.02, 0.02, summary_text, fontsize=12, bbox=dict(facecolor='white', alpha=0.8, boxstyle='round'), color=regime_color) # Adjust layout plt.tight_layout() # Add subtle BNP Paribas branding in bottom right corner plt.figtext(0.98, 0.01, "BNP PARIBAS\nConfidential", ha='right', fontsize=8, style='italic', alpha=0.7) # Save results if requested if SAVE_RESULTS: today_str = datetime.now().strftime('%Y%m%d') fig.savefig(f"{RESULTS_DIR}/{tenor}_regime_analysis_{today_str}.png", dpi=300, bbox_inches='tight') # Also save the regime data regime_df.to_csv(f"{RESULTS_DIR}/{tenor}_regime_data_{today_str}.csv") plt.show() return fig except Exception as e: print(f"Error visualizing results for {tenor}: {str(e)}") import traceback traceback.print_exc() return None # Function to create summary dashboard for all tenors def create_summary_dashboard(tenor_results, lookback_days=90): print("Creating summary dashboard for all tenors...") try: # Count how many tenors we're dealing with active_tenors = [tenor for tenor, data in tenor_results.items() if data['regime_df'] is not None] num_tenors = len(active_tenors) if num_tenors == 0: print("No tenor data available for summary dashboard.") return None # Create a figure with subplots - one row per tenor fig, axes = plt.subplots(num_tenors, 1, figsize=(12, 4*num_tenors), sharex=True) # Handle the case where there's only one tenor if num_tenors == 1: axes = [axes] # End date will be the same for all tenors (today) end_date = tenor_results[active_tenors[0]]['regime_df'].index[-1] start_date = end_date - timedelta(days=lookback_days) # Plot each tenor for i, tenor in enumerate(active_tenors): ax = axes[i] regime_df = tenor_results[tenor]['regime_df'] regime_char = tenor_results[tenor]['regime_characteristics'] # Filter data for the lookback period plot_data = regime_df.loc[start_date:end_date].copy() # Get bullish regime identifier bullish_regime = regime_char['bullish_regime'] # Plot the price line ax.plot(plot_data.index, plot_data[tenor], 'k-', linewidth=1.5) # Color the background prev_date = plot_data.index[0] prev_regime = plot_data.iloc[0]['Market_Condition'] # Find regime changes changes = plot_data[plot_data['Regime_Change'] == True].index.tolist() # Add the first period start = plot_data.index[0] for change_date in changes: color = 'green' if prev_regime == 'Bullish' else 'red' ax.axvspan(start, change_date, facecolor=color, alpha=0.2) # Update for next period start = change_date prev_regime = 'Bullish' if prev_regime == 'Bearish' else 'Bearish' # Add the final period color = 'green' if prev_regime == 'Bullish' else 'red' ax.axvspan(start, plot_data.index[-1], facecolor=color, alpha=0.2) # Add vertical lines at regime changes for change_date in changes: ax.axvline(x=change_date, color='blue', linestyle='--', linewidth=1) # Current regime info current_regime = plot_data.iloc[-1]['Market_Condition'] bullish_prob = plot_data.iloc[-1][f'Prob_Regime_{bullish_regime}'] # Set title and labels display_name = ASW_TENORS[tenor]['display_name'] window_size = ASW_TENORS[tenor]['window_size'] ax.set_title(f"{display_name} - Current: {current_regime} (Prob: {bullish_prob:.2f}, Window: {window_size})", fontsize=10, loc='left') ax.set_ylabel(display_name, fontsize=9) ax.grid(True, alpha=0.3) ax.tick_params(axis='y', labelsize=8) # Format x-axis (only for bottom plot) axes[-1].xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d')) axes[-1].xaxis.set_major_locator(mdates.WeekdayLocator(interval=2)) axes[-1].tick_params(axis='x', rotation=45) axes[-1].set_xlabel('Date', fontsize=10) # Add title to the entire figure fig.suptitle('ASW Regime Analysis Summary Dashboard', fontsize=16, y=0.98) # Legend for the entire figure buy_patch = patches.Patch(color='green', alpha=0.2, label='Bullish Regime') sell_patch = patches.Patch(color='red', alpha=0.2, label='Bearish Regime') regime_change = plt.Line2D([0], [0], color='blue', linestyle='--', label='Regime Change') fig.legend(handles=[buy_patch, sell_patch, regime_change], loc='upper right', bbox_to_anchor=(0.99, 0.96), fontsize=9) # Add report generation date plt.figtext(0.01, 0.01, f"Report generated: {datetime.now().strftime('%Y-%m-%d %H:%M')}", fontsize=8, style='italic') # BNP Paribas branding plt.figtext(0.99, 0.01, "BNP PARIBAS\nConfidential", ha='right', fontsize=8, style='italic', alpha=0.7) plt.tight_layout(rect=[0, 0.03, 1, 0.95]) # Save results if SAVE_RESULTS: today_str = datetime.now().strftime('%Y%m%d') fig.savefig(f"{RESULTS_DIR}/ASW_summary_dashboard_{today_str}.png", dpi=300, bbox_inches='tight') return fig except Exception as e: print(f"Error creating summary dashboard: {str(e)}") import traceback traceback.print_exc() return None