app.py

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

# Set Streamlit page configuration
st.set_page_config(page_title="EMA Envelope Strategy Optimization", layout="wide")

# Title and Description
st.title("EMA Envelope Strategy Optimization")
st.write("""
This tool backtests and optimizes an 'EMA Envelope' strategy.
It generates buy signals when the price drops below the lower envelope and sell signals when it rises above the upper envelope.
Key parameters include the ATR lookback, envelope width, and a sensitivity threshold.
You can optimize parameters and adjust them post-run to see their impact on signals and the equity curve.
""")

# Sidebar: How to use the app
with st.sidebar.expander("How to Use", expanded=False):
    st.write("""
    1. **Select Ticker**: Choose the asset ticker symbol (e.g., AAPL, TSLA) and date range for historical data.
    2. **Run Strategy**: Click "Run Strategy" to perform optimization.
    3. **Adjust Parameters**: After running the strategy, you can adjust the moving average windows, envelope settings, and threshold to see updated results live.
    """)

st.sidebar.title("Input Parameters")

# Sidebar: Select Ticker and Date Range
with st.sidebar.expander("Asset Settings", expanded=True):
    ticker = st.text_input("Asset Symbol", value="GOOGL", help="Ticker symbol (e.g., GOOGL, AAPL)")
    start_date = st.date_input("Start Date", value=pd.to_datetime("2020-01-01"), help="Select the start date for historical data.")
    end_date = st.date_input("End Date", value=pd.to_datetime("today") + pd.DateOffset(1), help="Select the end date for historical data.")

# Function to download data with yfinance adjustments
@st.cache_data
def download_data(ticker, start, end):
    data = yf.download(ticker, start=start, end=end, auto_adjust=False)
    if isinstance(data.columns, pd.MultiIndex):
        data.columns = data.columns.get_level_values(0)
    if data.empty:
        raise ValueError(f"No data fetched for {ticker} from {start} to {end}.")
    return data

# Function to calculate cumulative return with the strategy
def calculate_cumulative_return(data, atr_lookback, envelope_pct, atr_smoothing=14, threshold_multiplier=1, signal_threshold=0):
    data['TR'] = np.maximum(data['High'] - data['Low'],
                            np.maximum(np.abs(data['High'] - data['Close'].shift(1)),
                                       np.abs(data['Low'] - data['Close'].shift(1))))
    data['ATR'] = data['TR'].ewm(span=atr_smoothing, adjust=False).mean()

    data['Volatility'] = data['Close'].pct_change().rolling(window=atr_lookback).std()
    data['Lookback'] = (data['ATR'] / data['ATR'].max() * atr_lookback * data['Volatility']).apply(np.ceil)

    data = data.dropna()

    data['EMA'] = data['Close'].ewm(span=atr_lookback, adjust=False).mean()

    data['Upper_Envelope'] = data['EMA'] * (1 + envelope_pct * threshold_multiplier)
    data['Lower_Envelope'] = data['EMA'] * (1 - envelope_pct * threshold_multiplier)

    data['Signal'] = 0
    data.loc[data['Close'] >= data['Upper_Envelope'] * (1 - abs(signal_threshold)), 'Signal'] = -1
    data.loc[data['Close'] <= data['Lower_Envelope'] * (1 + abs(signal_threshold)), 'Signal'] = 1

    data['Return'] = data['Close'].pct_change()
    data['Strategy_Return'] = data['Return'] * data['Signal'].shift(1)
    data['Cumulative_Return'] = (1 + data['Strategy_Return']).cumprod()

    return data

# Grid search and optimization
def optimize_parameters(data, atr_lookback_range, envelope_pct_range, threshold_multiplier_range):
    best_return = -np.inf
    best_params = (None, None, None)

    for atr_lookback in atr_lookback_range:
        for envelope_pct in envelope_pct_range:
            for threshold_multiplier in threshold_multiplier_range:
                temp_data = calculate_cumulative_return(data.copy(), atr_lookback, envelope_pct, threshold_multiplier=threshold_multiplier)
                cumulative_return = temp_data['Cumulative_Return'].iloc[-1]
                if cumulative_return > best_return:
                    best_return = cumulative_return
                    best_params = (atr_lookback, envelope_pct, threshold_multiplier)

    return best_params

# Run Button to Fetch Data and Run Strategy
run_button = st.sidebar.button("Run Strategy")

if run_button:
    data = download_data(ticker, start_date, end_date)

    # Default ranges for optimization
    atr_lookback_range = range(10, 21)
    envelope_pct_range = np.linspace(0.01, 0.05, 9)
    threshold_multiplier_range = np.linspace(0.5, 1.5, 5)

    # Optimize parameters
    best_params = optimize_parameters(data, atr_lookback_range, envelope_pct_range, threshold_multiplier_range)

    # Store best params in session state
    st.session_state["best_params"] = best_params
    st.session_state["data"] = data
    st.session_state["signal_threshold"] = 0.00  # Default threshold

# If best_params are stored in session state, allow adjustments post-run
if "best_params" in st.session_state:
    best_params = st.session_state["best_params"]

    # Allow adjustments for ATR Lookback, Envelope Percentage, Threshold Multiplier, and Signal Threshold
    atr_lookback = st.sidebar.slider(
        "ATR Lookback", 10, 20, best_params[0], 1,
        help="ATR Lookback determines how far back in time the system looks to calculate the Average True Range (ATR), a measure of market volatility. Increasing this value will smooth volatility and lead to slower, more stable envelope adjustments."
    )
    envelope_pct = st.sidebar.slider(
        "Envelope Percentage", 0.01, 0.05, best_params[1], 0.001,
        help="Envelope Percentage defines the distance between the EMA and the upper and lower envelopes. A higher percentage widens the envelope, leading to fewer signals, while a lower percentage tightens the envelope, leading to more frequent signals."
    )
    threshold_multiplier = st.sidebar.slider(
        "Threshold Multiplier", 0.5, 1.5, best_params[2], 0.1,
        help="Threshold Multiplier adjusts how sensitive the envelope is to price changes. Increasing this multiplier makes the envelope more lenient, reducing the number of signals, while decreasing it makes the envelope more restrictive."
    )
    signal_threshold = st.sidebar.slider(
        "Signal Threshold", -0.05, 0.05, st.session_state["signal_threshold"], 0.01,
        help=("Signal Threshold adjusts the sensitivity of buy/sell signals relative to the EMA envelopes. "
              "Lower (negative) values make the strategy more lenient by generating more signals with smaller price deviations. "
              "Higher (positive) values make the strategy more restrictive, requiring larger price movements to trigger fewer but potentially more reliable signals.")
    )
    st.session_state["signal_threshold"] = signal_threshold

    # Calculate strategy with adjusted parameters
    optimized_data = calculate_cumulative_return(st.session_state["data"], atr_lookback, envelope_pct, 
                                                 threshold_multiplier=threshold_multiplier, signal_threshold=signal_threshold)

    # Display best parameters in JSON format
    st.json({
        "Best Parameters": {
            "ATR Lookback": best_params[0],
            "Envelope Percentage": best_params[1],
            "Threshold Multiplier": best_params[2]
        }
    })

    # Plot the stock price, EMA, and envelopes with adjusted parameters
    fig = go.Figure()

    # Add price trace
    fig.add_trace(go.Scatter(x=optimized_data.index, y=optimized_data['Close'], mode='lines', name='Close Price'))

    # Add volume as a secondary Y-axis (transparent fill)
    fig.add_trace(go.Bar(x=optimized_data.index, y=st.session_state["data"]['Volume'], name='Volume', yaxis='y2', opacity=0.3, marker_color='gray'))

    # Add EMA and Envelopes
    fig.add_trace(go.Scatter(x=optimized_data.index, y=optimized_data['EMA'], mode='lines', name='EMA', line=dict(dash='dash')))
    fig.add_trace(go.Scatter(x=optimized_data.index, y=optimized_data['Upper_Envelope'], mode='lines', name='Upper Envelope', line=dict(dash='dash')))
    fig.add_trace(go.Scatter(x=optimized_data.index, y=optimized_data['Lower_Envelope'], mode='lines', name='Lower Envelope', line=dict(dash='dash')))

    # Buy/Sell Signals
    buy_signals = optimized_data[optimized_data['Signal'] == 1]
    sell_signals = optimized_data[optimized_data['Signal'] == -1]
    fig.add_trace(go.Scatter(x=buy_signals.index, y=buy_signals['Close'], mode='markers', name='Buy Signal', marker=dict(color='green', symbol='triangle-up', size=15)))
    fig.add_trace(go.Scatter(x=sell_signals.index, y=sell_signals['Close'], mode='markers', name='Sell Signal', marker=dict(color='red', symbol='triangle-down', size=15)))

    # Update layout to include secondary Y-axis for volume
    fig.update_layout(
        title=f'{ticker} Exponential Moving Average Envelope Strategy',
        xaxis_title='Date',
        yaxis_title='Price',
        yaxis2=dict(title='Volume', overlaying='y', side='right', showgrid=False),
        legend=dict(orientation="h", yanchor="bottom", y=1.20, xanchor="center", x=0.5, traceorder='normal'),
        height=600,
        margin=dict(t=30, b=30)
    )

    st.plotly_chart(fig, use_container_width=True)

    # Plot the equity curve
    fig_equity = go.Figure()
    fig_equity.add_trace(go.Scatter(x=optimized_data.index, y=optimized_data['Cumulative_Return'], mode='lines', name='Equity Curve'))
    fig_equity.update_layout(
        title='Equity Curve',
        xaxis_title='Date',
        yaxis_title='Cumulative Return',
        legend=dict(orientation="h", yanchor="bottom", y=1.05, xanchor="center", x=0.5),
        height=400
    )
    st.plotly_chart(fig_equity, use_container_width=True)

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