app.py

import streamlit as st
import numpy as np
import pandas as pd
import plotly.graph_objects as go
from datetime import datetime, timedelta
import yfinance as yf

# Set wide page layout and page title
st.set_page_config(layout="wide", page_title="Self-Tuning SuperTrend and K-Means")

# App title and purpose explanation
st.title("Self-Tuning SuperTrend and K-Means")
st.write("This tool builds a self-tuning SuperTrend indicator using k-means clustering to adapt stop levels and generate buy/sell signals. It uses daily price data to compute volatility, score multiple configurations, and select the best one in real time.")

# Methodology expander (closed by default)
with st.expander("Methodology", expanded=False):
    #st.write("The tool self-tunes the SuperTrend indicator by testing multiple configurations and picking the best one.")
    
    #st.write("**Step 1: Data Preparation**")
    #st.write("Daily price data is downloaded. Key columns (High, Low, Close) are retained and prepped.")
    
    st.write("**Volatility Measurement**")
    st.write("An Average True Range (ATR) is computed using an Exponential Moving Average (EMA).")
    st.latex(r'\text{ATR} = \text{EMA}\left(\max\left\{High-Low,\ |High-\text{PrevClose}|,\ |Low-\text{PrevClose}|\right\}\right)')
    
    st.write("**Generating SuperTrend Variants**")
    st.write("Multiple SuperTrend signals are calculated. They use:")
    st.latex(r'\text{Upper Band} = hl2 + ATR \times \text{factor}')
    st.latex(r'\text{Lower Band} = hl2 - ATR \times \text{factor}')
    
    st.write("**Performance Scoring & Clustering**")
    st.write("Each variant is scored based on price movement. K-means (k=3) clusters these scores into Best, Average, and Worst groups.")
    
    st.write("**Final Signal Generation**")
    st.write("The indicator is recomputed using the average factor from the selected cluster. This gives a self-calibrated trading signal.")
    
    st.write("This process minimizes manual tuning and adapts with recent price action.")
    
    st.write("For more details, see the this article [here](https://entreprenerdly.com/trading-signals-with-adaptive-supertrend-and-k-means/).")
    
    # Sidebar inputs explanation
    st.write("#### Adjustable Inputs & Implications")
    st.write("""
    - **Ticker:** The stock symbol to analyze. Changing this lets you switch assets.
    - **Start Date & End Date:** Define the analysis window. End Date defaults to today plus one day.
    - **ATR Length:** Sets the period for ATR. Lower values react faster; higher values smooth out noise.
    - **Minimum/Maximum Multipliers & Step:** Define the range for SuperTrend sensitivity. Smaller steps improve resolution but increase compute time.
    - **Performance Alpha:** Determines the smoothness of the performance score. Lower makes the metric more reactive; higher favors stability.
    - **K-Means Options (From Cluster & Max Iterations):** Choose which cluster (Best, Average, Worst) to use and limit iterations for clustering. This controls how variants are grouped.
    """)

# Sidebar inputs
with st.sidebar:
    st.header("Input Parameters")
    
    # Data inputs expander
    with st.expander("Data Inputs", expanded=True):
        ticker = st.text_input(
            "Ticker", 
            value="ASML", 
            help="Enter the stock symbol to analyze. Example: AAPL, MSFT, NVDA. This determines which asset's data will be used."
        )
        start_date = st.date_input(
            "Start Date", 
            value=datetime(2022, 1, 1), 
            help="Start of the historical data window. Affects the amount of price history used to compute signals."
        )
        default_end_date = datetime.today() + timedelta(days=1)
        end_date = st.date_input(
            "End Date", 
            value=default_end_date, 
            help="End of the data window. Automatically set to today + 1 to include the most recent bar."
        )

    # Methodology parameters expander
    with st.expander("Methodology Parameters", expanded=True):
        atr_length = st.number_input(
            "ATR Length", 
            min_value=1, 
            value=7, 
            step=1, 
            help="ATR period controls how volatility is measured. Lower values make the stop more sensitive to short-term moves. Higher values smooth noise but may react slower."
        )
        min_mult = st.number_input(
            "Minimum Multiplier", 
            value=1.0, 
            step=0.1, 
            help="Defines the tightest SuperTrend stop. Lower values mean tighter stops, which react quickly but may whipsaw in choppy conditions."
        )
        max_mult = st.number_input(
            "Maximum Multiplier", 
            value=5.0, 
            step=0.1, 
            help="Defines the widest SuperTrend stop. Higher values give more breathing room but may delay trend changes."
        )
        step_mult = st.number_input(
            "Step", 
            value=0.5, 
            step=0.1, 
            help="Step size between multipliers. Smaller values give finer resolution but increase compute time."
        )
        perf_alpha = st.number_input(
            "Performance Alpha", 
            min_value=1, 
            value=8, 
            step=1, 
            help="Controls how quickly the performance metric responds to new price behavior. Lower = more reactive, higher = more stable but slower to adapt."
        )
        from_cluster = st.selectbox(
            "From Cluster", 
            options=["Best", "Average", "Worst"], 
            help="Selects which k-means cluster to use for final signal generation. 'Best' is typical for trend-following. 'Average' or 'Worst' can simulate conservative or contrarian behavior."
        )
        max_iter = st.number_input(
            "Max Iterations", 
            min_value=1, 
            value=1000, 
            step=1, 
            help="Upper limit on how long k-means clustering can run. Higher values allow more precise convergence but slow down the run time."
        )

    # Action button
    run_analysis = st.button("Run Analysis")

if run_analysis:
    # Validate date input
    if start_date >= end_date:
        st.error("Start Date must be before End Date.")
    else:
        with st.spinner("Running analysis..."):
            try:
                # Convert dates to string format
                start_date_str = start_date.strftime("%Y-%m-%d")
                end_date_str = end_date.strftime("%Y-%m-%d")
                
                # 1) Download data
                df = yf.download(ticker, start=start_date_str, end=end_date_str, interval="1d", auto_adjust=False)
                if df.empty:
                    st.error("No data returned from the data provider.")
                    st.stop()
                if isinstance(df.columns, pd.MultiIndex):
                    df.columns = df.columns.get_level_values(0)
                df.rename(columns={"Open": "Open", "High": "High", "Low": "Low", "Close": "Close", "Volume": "Volume"}, inplace=True)
                df.dropna(subset=["High", "Low", "Close"], inplace=True)
                df["hl2"] = (df["High"] + df["Low"]) / 2.0

                # 2) Compute ATR
                df["prev_close"] = df["Close"].shift(1)
                df["tr1"] = df["High"] - df["Low"]
                df["tr2"] = (df["High"] - df["prev_close"]).abs()
                df["tr3"] = (df["Low"] - df["prev_close"]).abs()
                df["tr"] = df[["tr1", "tr2", "tr3"]].max(axis=1)
                df["atr"] = df["tr"].ewm(alpha=2/(atr_length+1), adjust=False).mean()
                df.dropna(inplace=True)
                df.reset_index(drop=False, inplace=True)
                n = len(df)

                # Helper function: sign
                def sign(x):
                    return np.where(x > 0, 1, np.where(x < 0, -1, 0))

                # 3) Compute supertrend for each factor
                def compute_supertrend(df, factor, perf_alpha):
                    arr_close = df["Close"].values
                    arr_hl2 = df["hl2"].values
                    arr_atr = df["atr"].values

                    trend  = np.zeros(n, dtype=int)
                    upper  = np.zeros(n, dtype=float)
                    lower  = np.zeros(n, dtype=float)
                    output = np.zeros(n, dtype=float)
                    perf   = np.zeros(n, dtype=float)

                    trend[0]  = 1 if arr_close[0] > arr_hl2[0] else 0
                    upper[0]  = arr_hl2[0]
                    lower[0]  = arr_hl2[0]
                    output[0] = arr_hl2[0]
                    perf[0]   = 0.0

                    for i in range(1, n):
                        up = arr_hl2[i] + arr_atr[i] * factor
                        dn = arr_hl2[i] - arr_atr[i] * factor

                        if arr_close[i] > upper[i-1]:
                            trend[i] = 1
                        elif arr_close[i] < lower[i-1]:
                            trend[i] = 0
                        else:
                            trend[i] = trend[i-1]

                        if arr_close[i-1] < upper[i-1]:
                            upper[i] = min(up, upper[i-1])
                        else:
                            upper[i] = up

                        if arr_close[i-1] > lower[i-1]:
                            lower[i] = max(dn, lower[i-1])
                        else:
                            lower[i] = dn

                        diff_sign = sign(arr_close[i-1] - output[i-1])
                        perf[i] = perf[i-1] + 2/(perf_alpha+1)*((arr_close[i] - arr_close[i-1]) * diff_sign - perf[i-1])
                        output[i] = lower[i] if trend[i] == 1 else upper[i]

                    return {
                        "trend": trend,
                        "upper": upper,
                        "lower": lower,
                        "output": output,
                        "perf": perf,
                        "factor": factor
                    }

                factors = np.arange(min_mult, max_mult + 0.0001, step_mult)
                st_results = []
                for f in factors:
                    st_results.append(compute_supertrend(df, f, perf_alpha))

                perf_vals = np.array([res["perf"][-1] for res in st_results])
                fact_vals = np.array([res["factor"] for res in st_results])

                # 4) K-means clustering (k=3)
                def k_means(data, factors, k=3, max_iter=max_iter):
                    c1, c2, c3 = np.percentile(data, [25, 50, 75])
                    centroids = np.array([c1, c2, c3])
                    for _ in range(max_iter):
                        clusters = {0: [], 1: [], 2: []}
                        cluster_factors = {0: [], 1: [], 2: []}
                        for d, f in zip(data, factors):
                            dist = np.abs(d - centroids)
                            idx = dist.argmin()
                            clusters[idx].append(d)
                            cluster_factors[idx].append(f)
                        new_centroids = np.array([np.mean(clusters[i]) if len(clusters[i]) > 0 else centroids[i] for i in range(3)])
                        if np.allclose(new_centroids, centroids):
                            break
                        centroids = new_centroids
                    return clusters, cluster_factors, centroids

                clusters, cluster_factors, centroids = k_means(perf_vals, fact_vals, k=3, max_iter=max_iter)
                order = np.argsort(centroids)
                sorted_clusters = {i: clusters[j] for i, j in enumerate(order)}
                sorted_cluster_factors = {i: cluster_factors[j] for i, j in enumerate(order)}
                sorted_centroids = centroids[order]

                if from_cluster == "Best":
                    chosen_index = 2
                elif from_cluster == "Average":
                    chosen_index = 1
                else:
                    chosen_index = 0

                if len(sorted_cluster_factors[chosen_index]) > 0:
                    target_factor = np.mean(sorted_cluster_factors[chosen_index])
                else:
                    target_factor = factors[-1]

                if len(sorted_clusters[chosen_index]) > 0:
                    target_perf = np.mean(sorted_clusters[chosen_index])
                else:
                    target_perf = 0.0

                # 5) Recompute final supertrend with target_factor
                st_final = compute_supertrend(df, target_factor, perf_alpha)
                ts = st_final["output"]
                os_arr = np.zeros(n, dtype=int)
                os_arr[0] = 1 if df["Close"].iloc[0] > st_final["upper"][0] else 0

                for i in range(1, n):
                    c = df["Close"].iloc[i]
                    up = st_final["upper"][i]
                    dn = st_final["lower"][i]
                    if c > up:
                        os_arr[i] = 1
                    elif c < dn:
                        os_arr[i] = 0
                    else:
                        os_arr[i] = os_arr[i-1]

                # Build an adaptive MA for the trailing stop
                den_close_diff = (df["Close"] - df["Close"].shift(1)).abs()
                den = den_close_diff.ewm(alpha=2/(perf_alpha+1), adjust=False).mean()
                den_val = den.iloc[-1] if den.iloc[-1] != 0 else 1e-9
                perf_idx = max(target_perf, 0) / den_val

                perf_ama = np.zeros(n, dtype=float)
                perf_ama[0] = ts[0]
                for i in range(1, n):
                    perf_ama[i] = perf_ama[i-1] + perf_idx * (ts[i] - perf_ama[i-1])

                # 6) Build Plotly chart
                fig = go.Figure()
                fig.add_trace(go.Scatter(
                    x=df["Date"],
                    y=df["Close"],
                    mode="lines",
                    line=dict(color="silver", width=1.2),
                    name="Close Price"
                ))

                ts_bull = np.where(os_arr == 1, ts, np.nan)
                ts_bear = np.where(os_arr == 0, ts, np.nan)

                fig.add_trace(go.Scatter(
                    x=df["Date"],
                    y=ts_bull,
                    mode="lines",
                    line=dict(color="teal", width=1.2),
                    name="Bullish Stop"
                ))
                fig.add_trace(go.Scatter(
                    x=df["Date"],
                    y=ts_bear,
                    mode="lines",
                    line=dict(color="red", width=1.2),
                    name="Bearish Stop"
                ))
                fig.add_trace(go.Scatter(
                    x=df["Date"],
                    y=perf_ama,
                    mode="lines",
                    line=dict(color="orange", width=1.0),
                    opacity=0.7,
                    name="Trailing Stop AMA"
                ))

                for i in range(1, n):
                    if os_arr[i] != os_arr[i-1]:
                        if os_arr[i] == 1:
                            fig.add_trace(go.Scatter(
                                x=[df["Date"].iloc[i]],
                                y=[ts[i]],
                                mode="markers",
                                marker=dict(symbol="triangle-up", size=10, color="teal",
                                            line=dict(color="white", width=1)),
                                name="Bullish Signal",
                                showlegend=False
                            ))
                        else:
                            fig.add_trace(go.Scatter(
                                x=[df["Date"].iloc[i]],
                                y=[ts[i]],
                                mode="markers",
                                marker=dict(symbol="triangle-down", size=10, color="red",
                                            line=dict(color="white", width=1)),
                                name="Bearish Signal",
                                showlegend=False
                            ))

                fig.update_layout(
                    title=f"SuperTrend (Clustering) - {ticker}   [Factor ~ {target_factor:.2f}]",
                    xaxis_title="Date",
                    yaxis_title="Price",
                    template="plotly_dark",
                    width=1600,
                    height=800
                )
                fig.update_xaxes(tickformat='%Y-%m-%d')

                # Display results
                st.markdown("### Analysis Results")
                st.write("The plot below shows the closing price with the SuperTrend indicators and signals.")
                st.plotly_chart(fig, use_container_width=True)
            
            except Exception as e:
                st.error("An error occurred during the analysis.")
                st.error(str(e))

# Hide default Streamlit style
st.markdown(
    """
    <style>
    #MainMenu {visibility: hidden;}
    footer {visibility: hidden;}
    </style>
    """,
    unsafe_allow_html=True
)