app.py

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)