Spaces:
Sleeping
Sleeping
| import streamlit as st | |
| import yfinance as yf | |
| import pandas as pd | |
| import numpy as np | |
| import plotly.express as px | |
| import plotly.graph_objects as go | |
| import seaborn as sns | |
| import matplotlib.pyplot as plt | |
| import matplotlib.colors as mcolors | |
| # Streamlit app setup | |
| st.set_page_config(layout="wide") | |
| st.title("Conditional Probability of Stock Price Comovements") | |
| st.markdown( | |
| """ | |
| This tool helps you explore how likely a price change is to follow another price change, based on past data. | |
| You can choose a stock or cryptocurrency, set the time periods you want to analyze, and define the percentage change you consider significant. | |
| The analysis shows you the chances of a price moving up or down after an initial movement. | |
| """ | |
| ) | |
| st.sidebar.title("Input Parameters") | |
| # Sidebar for "How to Use" instructions inside an expander, closed by default | |
| with st.sidebar.expander("How to Use", expanded=False): | |
| st.write(""" | |
| **How to use this app:** | |
| 1. Select the stock or crypto symbol. | |
| 2. Choose the analysis periods. | |
| 3. Set the percentage change thresholds. | |
| 4. Click 'Run Analysis' to see the results. | |
| """) | |
| # Input parameters inside an expander, open by default | |
| with st.sidebar.expander("Input Parameters", expanded=True): | |
| symbol = st.text_input("Asset Symbol", value="ASML.AS", help="Enter the ticker symbol for the stock or cryptocurrency you want to analyze.") | |
| start_date = st.date_input("Start Date", value=pd.to_datetime("2000-01-01"), help="Select the start date for the data range.") | |
| end_date = st.date_input("End Date", value=pd.to_datetime(pd.Timestamp.now().date() + pd.Timedelta(days=1)), help="Select the end date for the data range.") | |
| # Parameters for the selected method inside an expander, open by default | |
| with st.sidebar.expander("Analysis Parameters", expanded=True): | |
| first_move_days = st.number_input("First Move Days", min_value=1, value=1, help="Set the number of days for the first move.") | |
| second_move_days = st.number_input("Second Move Days", min_value=1, value=2, help="Set the number of days for the second move.") | |
| first_threshold_percentage = st.number_input("First Threshold (%)", min_value=0.0, value=10.3, help="Set the threshold for the first move as a percentage.") / 100 | |
| second_threshold_percentage = st.number_input("Second Threshold (%)", min_value=0.0, value=8.0, help="Set the threshold for the second move as a percentage.") / 100 | |
| # Run Analysis button in the sidebar | |
| run_button = st.sidebar.button("Run Analysis") | |
| if run_button: | |
| try: | |
| # Download historical data for the selected stock or crypto | |
| data = yf.download(symbol, start=start_date, end=end_date, auto_adjust=False) | |
| if isinstance(data.columns, pd.MultiIndex): # Flatten multi-index | |
| data.columns = data.columns.get_level_values(0) | |
| if data.empty: | |
| raise ValueError("No data retrieved. Please check the symbol and date range.") | |
| if len(data) < max(first_move_days, second_move_days) + 1: | |
| raise ValueError(f"Insufficient data points for {symbol}. Need at least {max(first_move_days, second_move_days) + 1} days.") | |
| # Calculate the first move percentage change | |
| data[f'{first_move_days}d_pct_change_first'] = data['Adj Close'].pct_change(first_move_days) | |
| # Calculate the second move percentage change only if the first move change is above the threshold | |
| data.loc[data[f'{first_move_days}d_pct_change_first'].abs() >= first_threshold_percentage, f'{second_move_days}d_pct_change_second'] = data['Adj Close'].shift(-second_move_days).pct_change(second_move_days) | |
| # Calculate the frequency of the conditional threshold percentage change (increase or decrease) for the second move | |
| total_periods = len(data) | |
| if total_periods == 0: | |
| st.warning("No periods available for analysis after filtering. Adjust your parameters or check the data.") | |
| else: | |
| down_periods = len(data[data[f'{second_move_days}d_pct_change_second'] <= -second_threshold_percentage]) | |
| up_periods = len(data[data[f'{second_move_days}d_pct_change_second'] >= second_threshold_percentage]) | |
| down_frequency = down_periods / total_periods | |
| up_frequency = up_periods / total_periods | |
| st.markdown( | |
| """ | |
| #### Distribution of Second Move Percentage Changes | |
| This plot shows the distribution of the second move percentage changes. It highlights the probabilities of the second move being above or below the threshold given the first move exceeded its threshold. | |
| """ | |
| ) | |
| # Plot the distribution of second move percentage changes | |
| fig1 = px.histogram(data, x=f'{second_move_days}d_pct_change_second', nbins=100, title=f'Distribution of {second_move_days}-day {second_threshold_percentage * 100:.2f}% changes for {symbol} after a {first_move_days}-day {first_threshold_percentage * 100:.2f}% change') | |
| fig1.add_vline(x=-second_threshold_percentage, line_dash="dash", line_color="red", annotation_text=f"Probability of {second_move_days}-day {second_threshold_percentage * 100:.2f}% decrease after a {first_move_days}-day {first_threshold_percentage * 100:.2f}% change: {down_frequency * 100:.2f}%") | |
| fig1.add_vline(x=second_threshold_percentage, line_dash="dash", line_color="blue", annotation_text=f"Probability of {second_move_days}-day {second_threshold_percentage * 100:.2f}% increase after a {first_move_days}-day {first_threshold_percentage * 100:.2f}% change: {up_frequency * 100:.2f}%") | |
| fig1.update_xaxes(title_text='Percentage change') | |
| fig1.update_yaxes(title_text='Frequency') | |
| # Plot the stock price | |
| st.markdown( | |
| """ | |
| #### Stock Price and Significant Moves | |
| This plot shows the adjusted closing price of the stock over time. Markers indicate significant moves based on the defined thresholds. | |
| """ | |
| ) | |
| fig2 = go.Figure() | |
| fig2.add_trace(go.Scatter(x=data.index, y=data['Adj Close'], mode='lines', name='Adjusted Close Price')) | |
| up_instances = data[data[f'{second_move_days}d_pct_change_second'] >= second_threshold_percentage] | |
| down_instances = data[data[f'{second_move_days}d_pct_change_second'] <= -second_threshold_percentage] | |
| fig2.add_trace(go.Scatter(x=up_instances.index, y=up_instances['Adj Close'], mode='markers', marker=dict(color='blue', symbol='triangle-up', size=10), name=f"{second_move_days}-day {second_threshold_percentage * 100:.2f}% increase after {first_move_days}-day {first_threshold_percentage * 100:.2f}% change")) | |
| fig2.add_trace(go.Scatter(x=down_instances.index, y=down_instances['Adj Close'], mode='markers', marker=dict(color='red', symbol='triangle-down', size=10), name=f"{second_move_days}-day {second_threshold_percentage * 100:.2f}% decrease after {first_move_days}-day {first_threshold_percentage * 100:.2f}% change")) | |
| fig2.update_layout(title=f'Stock Price for {symbol} with {second_move_days}-day {second_threshold_percentage * 100:.2f}% changes after a {first_move_days}-day {first_threshold_percentage * 100:.2f}% change markers', xaxis_title='Date', yaxis_title='Adjusted Close Price') | |
| st.plotly_chart(fig1, use_container_width=True) | |
| st.plotly_chart(fig2, use_container_width=True) | |
| # Rolling window analysis with different window sizes | |
| st.markdown( | |
| """ | |
| #### Rolling Window Analysis | |
| This analysis examines the conditional probabilities of significant price movements over different rolling windows (e.g., 6 months, 1 year, 2 years). It helps to understand how these probabilities change over time. | |
| """ | |
| ) | |
| rolling_windows = [126, 252, 504] # 6 months, 1 year, and 2 years | |
| def get_shade(base_color, factor): | |
| return tuple(min(1, base + factor) for base in base_color) | |
| base_red = (1, 0, 0) | |
| base_blue = (0, 0, 1) | |
| colors_down = [mcolors.to_hex(get_shade(base_red, i * 0.35)) for i in range(len(rolling_windows))] | |
| colors_up = [mcolors.to_hex(get_shade(base_blue, i * 0.35)) for i in range(len(rolling_windows))] | |
| fig3 = go.Figure() | |
| for i, window in enumerate(rolling_windows): | |
| rolling_down = data['Adj Close'].rolling(window=window).apply( | |
| lambda x: np.nanmean((x.pct_change(first_move_days).abs() >= first_threshold_percentage) & | |
| (x.pct_change(second_move_days).shift(-second_move_days) <= -second_threshold_percentage)) | |
| ) | |
| rolling_up = data['Adj Close'].rolling(window=window).apply( | |
| lambda x: np.nanmean((x.pct_change(first_move_days).abs() >= first_threshold_percentage) & | |
| (x.pct_change(second_move_days).shift(-second_move_days) >= second_threshold_percentage)) | |
| ) | |
| fig3.add_trace(go.Scatter(x=data.index, y=rolling_down, mode='lines', name=f'Down Movements (Window: {window} days)', line=dict(color=colors_down[i]))) | |
| fig3.add_trace(go.Scatter(x=data.index, y=rolling_up, mode='lines', name=f'Up Movements (Window: {window} days)', line=dict(color=colors_up[i]))) | |
| fig3.update_layout( | |
| title=f'Rolling Window Analysis of {second_move_days}-day {second_threshold_percentage * 100:.2f}% Conditional Price Movements\nconditional on {first_move_days}-day {first_threshold_percentage * 100:.2f}% change', | |
| xaxis_title='Date', | |
| yaxis_title='Conditional Probability', | |
| legend_title_text='Movements' | |
| ) | |
| st.plotly_chart(fig3, use_container_width=True) | |
| # Heatmap analysis | |
| st.markdown( | |
| """ | |
| #### Heatmap Analysis | |
| These heatmaps show the conditional probabilities of second move percentage changes across a range of first move thresholds. Each heatmap represents a different scenario, such as up movements, down movements, or mixed conditions. | |
| """ | |
| ) | |
| thresholds = np.linspace(0.01, 0.15, 11) | |
| data['30d_volatility'] = data['Adj Close'].pct_change().rolling(window=30).std() | |
| prob_matrix = np.zeros((len(thresholds), len(thresholds)), dtype=object) | |
| prob_matrix_downs = np.zeros((len(thresholds), len(thresholds)), dtype=object) | |
| prob_matrix_ups = np.zeros((len(thresholds), len(thresholds)), dtype=object) | |
| prob_matrix_up_after_down = np.zeros((len(thresholds), len(thresholds)), dtype=object) | |
| prob_matrix_down_after_up = np.zeros((len(thresholds), len(thresholds)), dtype=object) | |
| for i, first_threshold in enumerate(thresholds): | |
| for j, second_threshold in enumerate(thresholds): | |
| first_condition = data['Adj Close'].pct_change(first_move_days).abs() >= first_threshold | |
| second_condition = data['Adj Close'].shift(-second_move_days).pct_change(second_move_days).abs() >= second_threshold | |
| combined_condition = first_condition & second_condition | |
| prob_matrix[i, j] = ( | |
| np.nanmean(combined_condition), | |
| np.nanmean(data['30d_volatility'][combined_condition]) | |
| ) | |
| first_condition_down = data['Adj Close'].pct_change(first_move_days) <= -first_threshold | |
| second_condition_down = data['Adj Close'].shift(-second_move_days).pct_change(second_move_days) <= -second_threshold | |
| combined_condition_downs = first_condition_down & second_condition_down | |
| prob_matrix_downs[i, j] = ( | |
| np.nanmean(combined_condition_downs), | |
| np.nanmean(data['30d_volatility'][combined_condition_downs]) | |
| ) | |
| first_condition_up = data['Adj Close'].pct_change(first_move_days) >= first_threshold | |
| second_condition_up = data['Adj Close'].shift(-second_move_days).pct_change(second_move_days) >= second_threshold | |
| combined_condition_ups = first_condition_up & second_condition_up | |
| prob_matrix_ups[i, j] = ( | |
| np.nanmean(combined_condition_ups), | |
| np.nanmean(data['30d_volatility'][combined_condition_ups]) | |
| ) | |
| second_condition_up_after_down = data['Adj Close'].shift(-second_move_days).pct_change(second_move_days) >= second_threshold | |
| combined_condition_up_after_down = first_condition_down & second_condition_up_after_down | |
| prob_matrix_up_after_down[i, j] = ( | |
| np.nanmean(combined_condition_up_after_down), | |
| np.nanmean(data['30d_volatility'][combined_condition_up_after_down]) | |
| ) | |
| second_condition_down_after_up = data['Adj Close'].shift(-second_move_days).pct_change(second_move_days) <= -second_threshold | |
| combined_condition_down_after_up = first_condition_up & second_condition_down_after_up | |
| prob_matrix_down_after_up[i, j] = ( | |
| np.nanmean(combined_condition_down_after_up), | |
| np.nanmean(data['30d_volatility'][combined_condition_down_after_up]) | |
| ) | |
| def format_annotation(value): | |
| return f"{value[0]:.3f}\n({value[1]:.3f})" if value[0] > 0 else "N/A" | |
| annot_matrix = np.vectorize(format_annotation)(prob_matrix) | |
| annot_matrix_downs = np.vectorize(format_annotation)(prob_matrix_downs) | |
| annot_matrix_ups = np.vectorize(format_annotation)(prob_matrix_ups) | |
| annot_matrix_up_after_down = np.vectorize(format_annotation)(prob_matrix_up_after_down) | |
| annot_matrix_down_after_up = np.vectorize(format_annotation)(prob_matrix_down_after_up) | |
| plt.rcParams['axes.facecolor'] = 'black' | |
| plt.rcParams['figure.facecolor'] = 'black' | |
| plt.rcParams['text.color'] = 'white' | |
| plt.rcParams['xtick.color'] = 'white' | |
| plt.rcParams['ytick.color'] = 'white' | |
| plt.rcParams['axes.labelcolor'] = 'white' | |
| plt.rcParams['axes.edgecolor'] = 'white' | |
| fig, ax = plt.subplots(figsize=(12, 6)) | |
| sns.heatmap(np.array([[v[0] for v in row] for row in prob_matrix]), xticklabels=np.round(thresholds, 2), yticklabels=np.round(thresholds, 2), cmap='coolwarm', annot=annot_matrix, fmt="", cbar_kws={'label': 'Conditional Probability'}, ax=ax) | |
| ax.set_xlabel(f'Second Move Threshold') | |
| ax.set_ylabel(f'First Move Threshold') | |
| ax.set_title(f'Conditional Probabilities of {second_move_days}-day Absolute Price Movements for {symbol}\nconditional on {first_move_days}-day Absolute Movements (30d Volatility in Parentheses)') | |
| st.pyplot(fig) | |
| fig, ax = plt.subplots(figsize=(12, 6)) | |
| sns.heatmap(np.array([[v[0] for v in row] for row in prob_matrix_downs]), xticklabels=np.round(thresholds, 2), yticklabels=np.round(thresholds, 2), cmap='coolwarm', annot=annot_matrix_downs, fmt="", cbar_kws={'label': 'Conditional Probability'}, ax=ax) | |
| ax.set_xlabel(f'Second Move Down Threshold') | |
| ax.set_ylabel(f'First Move Down Threshold') | |
| ax.set_title(f'Conditional Probabilities of {second_move_days}-day Down Price Movements for {symbol}\nconditional on {first_move_days}-day Down Movements (30d Volatility in Parentheses)') | |
| st.pyplot(fig) | |
| fig, ax = plt.subplots(figsize=(12, 6)) | |
| sns.heatmap(np.array([[v[0] for v in row] for row in prob_matrix_ups]), xticklabels=np.round(thresholds, 2), yticklabels=np.round(thresholds, 2), cmap='coolwarm', annot=annot_matrix_ups, fmt="", cbar_kws={'label': 'Conditional Probability'}, ax=ax) | |
| ax.set_xlabel(f'Second Move Up Threshold') | |
| ax.set_ylabel(f'First Move Up Threshold') | |
| ax.set_title(f'Conditional Probabilities of {second_move_days}-day Up Price Movements for {symbol}\nconditional on {first_move_days}-day Up Movements (30d Volatility in Parentheses)') | |
| st.pyplot(fig) | |
| fig, ax = plt.subplots(figsize=(12, 6)) | |
| sns.heatmap(np.array([[v[0] for v in row] for row in prob_matrix_up_after_down]), xticklabels=np.round(thresholds, 2), yticklabels=np.round(thresholds, 2), cmap='coolwarm', annot=annot_matrix_up_after_down, fmt="", cbar_kws={'label': 'Conditional Probability'}, ax=ax) | |
| ax.set_xlabel(f'Second Move Up Threshold') | |
| ax.set_ylabel(f'First Move Down Threshold') | |
| ax.set_title(f'Conditional Probabilities of {second_move_days}-day Up Price Movements for {symbol}\nconditional on {first_move_days}-day Down Movements (30d Volatility in Parentheses)') | |
| st.pyplot(fig) | |
| fig, ax = plt.subplots(figsize=(12, 6)) | |
| sns.heatmap(np.array([[v[0] for v in row] for row in prob_matrix_down_after_up]), xticklabels=np.round(thresholds, 2), yticklabels=np.round(thresholds, 2), cmap='coolwarm', annot=annot_matrix_down_after_up, fmt="", cbar_kws={'label': 'Conditional Probability'}, ax=ax) | |
| ax.set_xlabel(f'Second Move Down Threshold') | |
| ax.set_ylabel(f'First Move Up Threshold') | |
| ax.set_title(f'Conditional Probabilities of {second_move_days}-day Down Price Movements for {symbol}\nconditional on {first_move_days}-day Up Movements (30d Volatility in Parentheses)') | |
| st.pyplot(fig) | |
| except Exception as e: | |
| st.error(f"An error occurred while running the analysis: {e}") | |
| # Hide the default Streamlit menu and footer | |
| hide_streamlit_style = """ | |
| <style> | |
| #MainMenu {visibility: hidden;} | |
| footer {visibility: hidden;} | |
| </style> | |
| """ | |
| st.markdown(hide_streamlit_style, unsafe_allow_html=True) |