dewdew

# Function to determine regime characteristics with hysteresis thresholds def determine_regime_characteristics_with_hysteresis(df, regime_probabilities, tenor, bullish_threshold=0.6, bearish_threshold=0.4): """ Determine regime characteristics using hysteresis thresholds to reduce noise. Parameters: - bullish_threshold: Probability threshold to enter bullish regime (default 0.6) - bearish_threshold: Probability threshold to enter bearish regime (default 0.4) """ # 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] # Add price data regime_df[tenor] = df[tenor] # First, determine which regime is bullish/bearish using the original method # This is needed to know which probability column represents bullish regime # Calculate price trends in each regime using the simple method first regime_df['Simple_Regime'] = regime_df[['Prob_Regime_0', 'Prob_Regime_1']].idxmax(axis=1) regime_df['Simple_Regime'] = regime_df['Simple_Regime'].str.replace('Prob_Regime_', '') # Calculate average price changes by simple regime to identify bullish/bearish regime_df['Price_Changes'] = regime_df[tenor].diff() avg_change_0 = regime_df[regime_df['Simple_Regime'] == '0']['Price_Changes'].mean() avg_change_1 = regime_df[regime_df['Simple_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_id = '0' bearish_regime_id = '1' else: bullish_regime_id = '1' bearish_regime_id = '0' regime_characteristics = { 'bullish_regime': bullish_regime_id, 'bearish_regime': bearish_regime_id } print(f"{tenor} - Bullish regime identified as: Regime {bullish_regime_id}") print(f"{tenor} - Using thresholds: Bullish >{bullish_threshold*100}%, Bearish <{bearish_threshold*100}%") # Now implement hysteresis logic bullish_prob_col = f'Prob_Regime_{bullish_regime_id}' # Initialize the regime classification with hysteresis regime_df['Market_Condition'] = None regime_df['Regime_Change'] = False # Start with the first regime based on initial probability initial_bullish_prob = regime_df.iloc[0][bullish_prob_col] if initial_bullish_prob > bullish_threshold: current_regime = 'Bullish' elif initial_bullish_prob < bearish_threshold: current_regime = 'Bearish' else: # If starting in the dead zone, use simple majority rule for first regime current_regime = 'Bullish' if initial_bullish_prob > 0.5 else 'Bearish' regime_df.iloc[0, regime_df.columns.get_loc('Market_Condition')] = current_regime # Apply hysteresis logic for the rest of the data for i in range(1, len(regime_df)): bullish_prob = regime_df.iloc[i][bullish_prob_col] previous_regime = regime_df.iloc[i-1]['Market_Condition'] # Apply hysteresis thresholds if previous_regime == 'Bearish': # Currently bearish, only switch to bullish if prob > bullish_threshold if bullish_prob > bullish_threshold: new_regime = 'Bullish' regime_change = True else: new_regime = 'Bearish' regime_change = False else: # previous_regime == 'Bullish' # Currently bullish, only switch to bearish if prob < bearish_threshold if bullish_prob < bearish_threshold: new_regime = 'Bearish' regime_change = True else: new_regime = 'Bullish' regime_change = False regime_df.iloc[i, regime_df.columns.get_loc('Market_Condition')] = new_regime regime_df.iloc[i, regime_df.columns.get_loc('Regime_Change')] = regime_change # Update the Regime column to match Market_Condition for consistency regime_df['Regime'] = regime_df['Market_Condition'].apply( lambda x: bullish_regime_id if x == 'Bullish' else bearish_regime_id ) return regime_df, regime_characteristics # Modified visualization function that shows the hysteresis thresholds def visualize_tenor_results_with_hysteresis(df, regime_df, regime_characteristics, tenor, lookback_days=365, bullish_threshold=0.6, bearish_threshold=0.4): print(f"Generating visualization for {tenor} with hysteresis thresholds...") 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=(14, 10)) # Set up grid for three plots with different heights gs = plt.GridSpec(3, 1, height_ratios=[3, 1, 1], figure=fig) # Create three axes ax1 = fig.add_subplot(gs[0]) # Price with regime classification ax2 = fig.add_subplot(gs[1]) # Regime probability with thresholds ax3 = fig.add_subplot(gs[2]) # Regime state over time # ===== 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 = plot_regime_df.loc[change_date, 'Market_Condition'] # 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 new_regime = plot_regime_df.loc[change_date, 'Market_Condition'] 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 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') 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 Hysteresis Regime Classification (Thresholds: {bearish_threshold*100}%-{bullish_threshold*100}%)', fontsize=14) ax1.set_ylabel(ticker_name, fontsize=12) ax1.grid(True, alpha=0.3) # ===== Middle Plot: Regime Probabilities with Thresholds ===== 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 horizontal threshold lines ax2.axhline(y=bullish_threshold, color='green', linestyle='-', linewidth=2, alpha=0.8, label=f'Bullish Threshold ({bullish_threshold*100}%)') ax2.axhline(y=bearish_threshold, color='red', linestyle='-', linewidth=2, alpha=0.8, label=f'Bearish Threshold ({bearish_threshold*100}%)') ax2.axhline(y=0.5, color='gray', linestyle=':', alpha=0.5, label='50% Line') # Shade the dead zone ax2.axhspan(bearish_threshold, bullish_threshold, facecolor='yellow', alpha=0.1, label='Dead Zone (No Change)') # Add vertical lines at regime changes 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 with Hysteresis Thresholds', fontsize=14) ax2.set_ylabel('Probability', fontsize=12) ax2.set_ylim(0, 1) ax2.legend(loc='upper left', fontsize=10) ax2.grid(True, alpha=0.3) # ===== Bottom Plot: Regime State Over Time ===== # Create a binary representation of regimes for clear visualization regime_binary = plot_regime_df['Market_Condition'].map({'Bullish': 1, 'Bearish': 0}) # Plot as a step function ax3.step(plot_regime_df.index, regime_binary, where='post', linewidth=3, color='darkblue', label='Regime State') # Fill the area ax3.fill_between(plot_regime_df.index, 0, regime_binary, step='post', alpha=0.3, color='lightblue') # Add vertical lines at regime changes for change_date in regime_changes: ax3.axvline(x=change_date, color='blue', linestyle='--', linewidth=1.5) # Customize the plot ax3.set_ylim(-0.1, 1.1) ax3.set_yticks([0, 1]) ax3.set_yticklabels(['Bearish', 'Bullish']) ax3.set_title('Regime State Timeline', fontsize=14) ax3.set_ylabel('Regime', fontsize=12) ax3.set_xlabel('Date', fontsize=12) ax3.grid(True, alpha=0.3) # Format x-axis for all plots for ax in [ax1, ax2, ax3]: ax.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d')) ax.xaxis.set_major_locator(mdates.MonthLocator(interval=1)) plt.setp(ax.get_xticklabels(), rotation=45, ha='right') # Add current regime information 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: {latest_date.strftime('%Y-%m-%d')}\n" summary_text += f"Regime: {latest_regime}\n" summary_text += f"Bullish Prob: {latest_prob:.2f}\n" summary_text += f"Thresholds: {bearish_threshold:.0%}-{bullish_threshold:.0%}" # Add text box with latest regime info 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 branding 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_hysteresis_{today_str}.png", dpi=300, bbox_inches='tight') plt.show() return fig except Exception as e: print(f"Error visualizing results for {tenor}: {str(e)}") import traceback traceback.print_exc() return None # Modified function to run the final model with hysteresis def run_final_model_with_hysteresis(df, window_size, bullish_threshold=0.6, bearish_threshold=0.4): """ Run the final model with hysteresis thresholds for regime switching """ # Create a copy to avoid modifying the original df_temp = df.copy() # Get the tenor name from the dataframe columns tenor = None for col in df_temp.columns: if col in ASW_TENORS: tenor = col break if tenor is None: print("Error: Could not identify tenor column") return None, None, None, None # Compute rolling average with the optimal window size df_temp['Close_Change'] = df_temp[tenor].pct_change() * 100 df_temp['Trend_Signal'] = df_temp['Close_Change'].rolling(window=int(window_size)).mean() df_temp = df_temp.dropna(subset=['Trend_Signal']) # Initialize backtesting parameters start_index = -619 # Start backtest regime_history = [] probability_history = [] # Rolling backtest loop for i in range(start_index, 0): df_subset = df_temp.iloc[:i].copy() if len(df_subset) > 10: try: model = MarkovRegression( df_subset['Trend_Signal'], k_regimes=2, trend='c', switching_variance=True ) result = model.fit(maxiter=200, disp=False, tol=1e-3) # Store results regime_history.append({ "Date": df_temp.index[i], "Regime": result.smoothed_marginal_probabilities.idxmax(axis=1).iloc[-1], tenor: df_temp[tenor].iloc[i] }) probability_history.append(result.smoothed_marginal_probabilities.iloc[-1].values) except Exception: continue # Convert results to DataFrame if not regime_history: return None, None, None, None regime_df = pd.DataFrame(regime_history) regime_df.set_index('Date', inplace=True) probability_df = pd.DataFrame( probability_history, columns=['Regime_0_Prob', 'Regime_1_Prob'], index=regime_df.index ) # Apply hysteresis logic regime_df, regime_characteristics = determine_regime_characteristics_with_hysteresis( regime_df, [probability_df['Regime_0_Prob'], probability_df['Regime_1_Prob']], tenor, bullish_threshold, bearish_threshold ) # Calculate trading performance with hysteresis trades = [] in_position = False position_type = None entry_price = None entry_date = None regime_changes = regime_df[regime_df['Regime_Change'] == True].index.tolist() for change_date in regime_changes: # Close existing position if any if in_position: exit_price = regime_df.loc[change_date, tenor] if position_type == 'Long': trade_pnl = exit_price - entry_price else: # Short trade_pnl = entry_price - exit_price trades.append({ 'Entry_Date': entry_date, 'Exit_Date': change_date, 'Position': position_type, 'Entry_Price': entry_price, 'Exit_Price': exit_price, 'PnL': trade_pnl }) # Open new position in_position = True new_regime = regime_df.loc[change_date, 'Market_Condition'] if new_regime == 'Bullish': position_type = 'Long' else: position_type = 'Short' entry_price = regime_df.loc[change_date, tenor] entry_date = change_date # Calculate performance metrics trades_df = pd.DataFrame(trades) performance = {} if len(trades_df) > 0: trades_df['Cumulative_PnL'] = trades_df['PnL'].cumsum() performance['win_rate'] = (trades_df['PnL'] >= 0).mean() performance['avg_win'] = trades_df[trades_df['PnL'] > 0]['PnL'].mean() if len(trades_df[trades_df['PnL'] > 0]) > 0 else 0 performance['avg_loss'] = trades_df[trades_df['PnL'] < 0]['PnL'].mean() if len(trades_df[trades_df['PnL'] < 0]) > 0 else 0 performance['total_pnl'] = trades_df['PnL'].sum() performance['num_trades'] = len(trades_df) performance['bullish_threshold'] = bullish_threshold performance['bearish_threshold'] = bearish_threshold return regime_df, probability_df, trades_df, performance # Example usage function showing how to test different threshold combinations def test_hysteresis_thresholds(df_daily_original, tenor='USSFCT05'): """ Test different hysteresis threshold combinations to find optimal settings """ print(f"Testing different hysteresis thresholds for {tenor}...") # Test different threshold combinations threshold_combinations = [ (0.55, 0.45), # Narrow band (0.6, 0.4), # Medium band (recommended) (0.65, 0.35), # Wide band (0.7, 0.3), # Very wide band ] results = [] for bullish_thresh, bearish_thresh in threshold_combinations: print(f"\nTesting thresholds: Bullish >{bullish_thresh*100}%, Bearish <{bearish_thresh*100}%") # Create a subset of data for this tenor tenor_data = df_daily_original[[tenor]].copy() try: regime_df, prob_df, trades_df, performance = run_final_model_with_hysteresis( tenor_data, window_size=40, bullish_threshold=bullish_thresh, bearish_threshold=bearish_thresh ) if performance and len(trades_df) > 0: results.append({ 'bullish_threshold': bullish_thresh, 'bearish_threshold': bearish_thresh, 'num_trades': performance['num_trades'], 'win_rate': performance['win_rate'], 'total_pnl': performance['total_pnl'], 'avg_trade_duration': len(regime_df) / max(1, performance['num_trades']) }) print(f" Number of trades: {performance['num_trades']}") print(f" Win rate: {performance['win_rate']:.2f}") print(f" Total PnL: {performance['total_pnl']:.4f}") else: print(" No valid results for this threshold combination") except Exception as e: print(f" Error with thresholds {bullish_thresh}-{bearish_thresh}: {str(e)}") # Display results summary if results: results_df = pd.DataFrame(results) print("\n" + "="*80) print("HYSTERESIS THRESHOLD COMPARISON RESULTS") print("="*80) print(results_df.to_string(index=False, float_format='%.3f')) # Recommend best threshold based on a composite score results_df['score'] = results_df['total_pnl'] * results_df['win_rate'] / results_df['num_trades'] best_idx = results_df['score'].idxmax() best_result = results_df.iloc[best_idx] print(f"\nRECOMMENDED THRESHOLDS:") print(f"Bullish threshold: {best_result['bullish_threshold']*100}%") print(f"Bearish threshold: {best_result['bearish_threshold']*100}%") print(f"This combination provides {best_result['num_trades']} trades with {best_result['win_rate']:.1%} win rate") return results # Print instructions for using the hysteresis model print(""" HYSTERESIS REGIME DETECTION MODEL ================================== This modified version implements hysteresis thresholds to reduce regime switching noise: Key Features: - Bullish regime entry: Only when probability > 60% (configurable) - Bearish regime entry: Only when probability < 40% (configurable) - Dead zone (40%-60%): No regime changes occur in this range - Reduces false signals and whipsaws Usage Examples: 1. Test different threshold combinations: results = test_hysteresis_thresholds(df_daily_original, 'USSFCT05') 2. Run model with custom thresholds: regime_df, prob_df, trades_df, performance = run_final_model_with_hysteresis( tenor_data, window_size=40, bullish_threshold=0.65, bearish_threshold=0.35 ) 3. Visualize results with thresholds: visualize_tenor_results_with_hysteresis( tenor_data, regime_df, regime_characteristics, 'USSFCT05', bullish_threshold=0.6, bearish_threshold=0.4 ) Benefits: - Fewer trades (reduced transaction costs) - More stable regime identification - Eliminates noise around 50% probability level - Customizable threshold levels per tenor """)

# ============================================================================= # MODIFIED FUNCTIONS FOR HYSTERESIS INTEGRATION # Replace these functions in your original script # ============================================================================= # 1. ADD HYSTERESIS THRESHOLDS TO YOUR ASW_TENORS CONFIGURATION # Add these lines to your ASW_TENORS dictionary: ASW_TENORS = { 'USSFCT02': { 'window_size': 40, 'include': True, 'display_name': 'ASW 2Y', 'bullish_threshold': 0.6, # ADD THIS LINE 'bearish_threshold': 0.4 # ADD THIS LINE }, 'USSFCT05': { 'window_size': 50, 'include': True, 'display_name': 'ASW 5Y', 'bullish_threshold': 0.6, # ADD THIS LINE 'bearish_threshold': 0.4 # ADD THIS LINE }, 'USSFCT07': { 'window_size': 60, 'include': True, 'display_name': 'ASW 7Y', 'bullish_threshold': 0.65, # ADD THIS LINE 'bearish_threshold': 0.35 # ADD THIS LINE }, 'USSFCT10': { 'window_size': 70, 'include': True, 'display_name': 'ASW 10Y', 'bullish_threshold': 0.65, # ADD THIS LINE 'bearish_threshold': 0.35 # ADD THIS LINE } } # ============================================================================= # 2. REPLACE: determine_regime_characteristics function # ============================================================================= def determine_regime_characteristics(df, regime_probabilities, tenor): """ Determine regime characteristics using hysteresis thresholds to reduce noise. This replaces your original function with hysteresis logic. """ # Get hysteresis thresholds for this tenor bullish_threshold = ASW_TENORS[tenor]['bullish_threshold'] bearish_threshold = ASW_TENORS[tenor]['bearish_threshold'] # 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 (standard approach first) 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 } else: # Fallback if we couldn't calculate slopes 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 } print(f"{tenor} - Bullish regime identified as: Regime {bullish_regime}") print(f"{tenor} - Using hysteresis thresholds: Bullish >{bullish_threshold*100}%, Bearish <{bearish_threshold*100}%") # ===== HYSTERESIS LOGIC IMPLEMENTATION ===== # Now implement hysteresis logic using the identified regimes bullish_prob_col = f'Prob_Regime_{bullish_regime}' # Initialize the regime classification with hysteresis regime_df['Market_Condition'] = None regime_df['Regime_Change'] = False # Start with the first regime based on initial probability initial_bullish_prob = regime_df.iloc[0][bullish_prob_col] if initial_bullish_prob > bullish_threshold: current_regime = 'Bullish' elif initial_bullish_prob < bearish_threshold: current_regime = 'Bearish' else: # If starting in the dead zone, use simple majority rule for first regime current_regime = 'Bullish' if initial_bullish_prob > 0.5 else 'Bearish' regime_df.iloc[0, regime_df.columns.get_loc('Market_Condition')] = current_regime # Apply hysteresis logic for the rest of the data for i in range(1, len(regime_df)): bullish_prob = regime_df.iloc[i][bullish_prob_col] previous_regime = regime_df.iloc[i-1]['Market_Condition'] # Apply hysteresis thresholds if previous_regime == 'Bearish': # Currently bearish, only switch to bullish if prob > bullish_threshold if bullish_prob > bullish_threshold: new_regime = 'Bullish' regime_change = True else: new_regime = 'Bearish' regime_change = False else: # previous_regime == 'Bullish' # Currently bullish, only switch to bearish if prob < bearish_threshold if bullish_prob < bearish_threshold: new_regime = 'Bearish' regime_change = True else: new_regime = 'Bullish' regime_change = False regime_df.iloc[i, regime_df.columns.get_loc('Market_Condition')] = new_regime regime_df.iloc[i, regime_df.columns.get_loc('Regime_Change')] = regime_change # Update the Regime column to match Market_Condition for consistency regime_df['Regime'] = regime_df['Market_Condition'].apply( lambda x: bullish_regime if x == 'Bullish' else bearish_regime ) return regime_df, regime_characteristics # ============================================================================= # 3. REPLACE: print_tenor_regime_summary function # ============================================================================= def print_tenor_regime_summary(regime_df, regime_characteristics, tenor): """Print tenor's regime summary with hysteresis info""" 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 # Get hysteresis thresholds bullish_thresh = ASW_TENORS[tenor]['bullish_threshold'] bearish_thresh = ASW_TENORS[tenor]['bearish_threshold'] print("\n" + "="*60) print(f"REGIME SUMMARY FOR {tenor} ON {latest_date.strftime('%Y-%m-%d')}") print("="*60) print(f"CURRENT MARKET REGIME: {latest_regime}") print(f"Bullish Probability: {bullish_prob:.4f}") print(f"Bearish Probability: {bearish_prob:.4f}") print(f"Hysteresis Thresholds: Bullish >{bullish_thresh*100}%, Bearish <{bearish_thresh*100}%") # Determine regime stability if latest_regime == 'Bullish': distance_to_switch = bullish_prob - bearish_thresh print(f"Distance to Bearish switch: {distance_to_switch:.4f} ({distance_to_switch*100:.1f}%)") else: distance_to_switch = bullish_thresh - bullish_prob print(f"Distance to Bullish switch: {distance_to_switch:.4f} ({distance_to_switch*100:.1f}%)") # Check for recent regime changes if len(regime_df) > 1: recent_changes = regime_df[regime_df['Regime_Change'] == True].tail(1) if len(recent_changes) > 0: last_change_date = recent_changes.index[0] days_since_change = (latest_date - last_change_date).days print(f"\nLast regime change: {last_change_date.strftime('%Y-%m-%d')} ({days_since_change} days ago)") print("="*60) except Exception as e: print(f"Error in print_tenor_regime_summary for {tenor}: {str(e)}") import traceback traceback.print_exc() # ============================================================================= # 4. REPLACE: visualize_tenor_results function # ============================================================================= def visualize_tenor_results(df, regime_df, regime_characteristics, tenor, lookback_days=365): """Enhanced visualization with hysteresis thresholds""" print(f"Generating visualization for {tenor} with hysteresis thresholds...") try: # Extract regime characteristics bullish_regime = regime_characteristics['bullish_regime'] bearish_regime = regime_characteristics['bearish_regime'] bullish_threshold = ASW_TENORS[tenor]['bullish_threshold'] bearish_threshold = ASW_TENORS[tenor]['bearish_threshold'] # 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=(14, 12)) # Set up grid for three plots with different heights gs = plt.GridSpec(3, 1, height_ratios=[3, 1.5, 1], figure=fig) # Create three axes ax1 = fig.add_subplot(gs[0]) # Price with regime classification ax2 = fig.add_subplot(gs[1]) # Regime probability with thresholds ax3 = fig.add_subplot(gs[2]) # Regime state over time # ===== 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 = plot_regime_df.loc[change_date, 'Market_Condition'] # 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 new_regime = plot_regime_df.loc[change_date, 'Market_Condition'] 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 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') 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 Hysteresis Regime Classification (Thresholds: {bearish_threshold*100}%-{bullish_threshold*100}%)', fontsize=14) ax1.set_ylabel(ticker_name, fontsize=12) ax1.grid(True, alpha=0.3) # ===== Middle Plot: Regime Probabilities with Thresholds ===== 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 horizontal threshold lines ax2.axhline(y=bullish_threshold, color='green', linestyle='-', linewidth=2, alpha=0.8, label=f'Bullish Threshold ({bullish_threshold*100}%)') ax2.axhline(y=bearish_threshold, color='red', linestyle='-', linewidth=2, alpha=0.8, label=f'Bearish Threshold ({bearish_threshold*100}%)') ax2.axhline(y=0.5, color='gray', linestyle=':', alpha=0.5, label='50% Line') # Shade the dead zone ax2.axhspan(bearish_threshold, bullish_threshold, facecolor='yellow', alpha=0.1, label='Dead Zone (No Change)') # Add vertical lines at regime changes for change_date in regime_changes: ax2.axvline(x=change_date, color='blue', linestyle='--', linewidth=1.5) ax2.set_title('Bullish Regime Probability with Hysteresis Thresholds', fontsize=14) ax2.set_ylabel('Probability', fontsize=12) ax2.set_ylim(0, 1) ax2.legend(loc='upper left', fontsize=10) ax2.grid(True, alpha=0.3) # ===== Bottom Plot: Regime State Over Time ===== # Create a binary representation of regimes for clear visualization regime_binary = plot_regime_df['Market_Condition'].map({'Bullish': 1, 'Bearish': 0}) # Plot as a step function ax3.step(plot_regime_df.index, regime_binary, where='post', linewidth=3, color='darkblue', label='Regime State') # Fill the area ax3.fill_between(plot_regime_df.index, 0, regime_binary, step='post', alpha=0.3, color='lightblue') # Add vertical lines at regime changes for change_date in regime_changes: ax3.axvline(x=change_date, color='blue', linestyle='--', linewidth=1.5) # Customize the plot ax3.set_ylim(-0.1, 1.1) ax3.set_yticks([0, 1]) ax3.set_yticklabels(['Bearish', 'Bullish']) ax3.set_title('Regime State Timeline', fontsize=14) ax3.set_ylabel('Regime', fontsize=12) ax3.set_xlabel('Date', fontsize=12) ax3.grid(True, alpha=0.3) # Format x-axis for all plots for ax in [ax1, ax2, ax3]: ax.xaxis.set_major_formatter(mdates.DateFormatter('%Y-%m-%d')) ax.xaxis.set_major_locator(mdates.MonthLocator(interval=1)) plt.setp(ax.get_xticklabels(), rotation=45, ha='right') # Add current regime information 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: {latest_date.strftime('%Y-%m-%d')}\n" summary_text += f"Regime: {latest_regime}\n" summary_text += f"Bullish Prob: {latest_prob:.2f}\n" summary_text += f"Thresholds: {bearish_threshold:.0%}-{bullish_threshold:.0%}" # Add text box with latest regime info 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 branding 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_hysteresis_{today_str}.png", dpi=300, bbox_inches='tight') # Also save the regime data regime_df.to_csv(f"{RESULTS_DIR}/{tenor}_regime_data_hysteresis_{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 # ============================================================================= # 5. MODIFY: create_summary_dashboard function (add hysteresis info to titles) # ============================================================================= def create_summary_dashboard(tenor_results, lookback_days=90): print("Creating summary dashboard for all tenors with hysteresis...") 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=(14, 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 and thresholds bullish_regime = regime_char['bullish_regime'] bullish_threshold = ASW_TENORS[tenor]['bullish_threshold'] bearish_threshold = ASW_TENORS[tenor]['bearish_threshold'] # 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 with hysteresis info display_name = ASW_TENORS[tenor]['display_name'] window_size = ASW_TENORS[tenor]['window_size'] title = (f"{display_name} - Current: {current_regime} (Prob: {bullish_prob:.2f}) | " f"Thresholds: {bearish_threshold*100}%-{bullish_threshold*100}% | Window: {window_size}") ax.set_title(title, 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 with Hysteresis', 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_hysteresis_{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 # ============================================================================= # USAGE INSTRUCTIONS: # ============================================================================= print(""" INTEGRATION INSTRUCTIONS: ========================= 1. ADD hysteresis thresholds to your ASW_TENORS configuration (top of script) 2. REPLACE these 4 functions in your original script: - determine_regime_characteristics() - print_tenor_regime_summary() - visualize_tenor_results() - create_summary_dashboard() 3. NO OTHER CHANGES neede