import streamlit as st import yfinance as yf import pandas as pd import numpy as np import plotly.graph_objs as go from itertools import product from datetime import datetime, timedelta from plotly.subplots import make_subplots # Set Streamlit page configuration st.set_page_config(page_title="Triple Moving Average Crossover Strategy", layout="wide") # Title and description st.title("Triple Moving Average Crossover Strategy") st.write(""" This tool allows users to backtest a 3-way moving average crossover strategy across different time horizons (short-term, medium-term, and long-term). The strategy uses three different moving averages to generate buy/sell signals when shorter-term averages cross above or below longer-term averages. By adjusting parameters like the length of each moving average and the signal threshold, you can further customize how strict or lenient the crossover signals are. """) # 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, BTC-USD) and date range for historical data. 2. **Run Strategy**: Click "Run Strategy" to perform optimization and backtesting of the strategy using the default parameters for the selected horizon. 3. **Adjust Parameters**: After running the strategy, use the sliders to adjust the moving average windows and threshold, and see the results update live. 4. **Threshold Parameter**: Controls how strict the buy/sell signals are when moving averages cross. Lower thresholds generate more signals; higher thresholds generate fewer, stricter signals. """) # Sidebar: Navigation st.sidebar.markdown("### Page Navigation") page = st.sidebar.radio("Select Strategy Horizon", options=["Short-Term", "Medium-Term", "Long-Term"]) # Sidebar: Select Ticker and Date Range with st.sidebar.expander("Asset Settings", expanded=True): ticker = st.text_input("Asset Symbol", value="AAPL", help="Ticker symbol (Indicate the stock ticker or Cryptocurrency Pair (e.g., AAPL, BTC-USD))") start_date = st.date_input("Start Date", value=datetime(2015, 1, 1), help="Select the start date for historical data.") end_date = st.date_input("End Date", value=datetime.today() + timedelta(days=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 moving averages def calculate_moving_averages(data, short_window, medium_window, long_window): data['short_ma'] = data['Adj Close'].rolling(window=short_window).mean() data['medium_ma'] = data['Adj Close'].rolling(window=medium_window).mean() data['long_ma'] = data['Adj Close'].rolling(window=long_window).mean() return data # Function to generate trading signals with a percentage-based threshold def generate_signals(data, threshold=0.01): data['signal'] = 0 data['signal'][(data['short_ma'] > data['medium_ma'] * (1 + threshold)) & (data['medium_ma'] > data['long_ma'] * (1 + threshold))] = 1 data['signal'][(data['short_ma'] < data['medium_ma'] * (1 - threshold)) & (data['medium_ma'] < data['long_ma'] * (1 - threshold))] = -1 data['positions'] = data['signal'].diff() return data # Function to calculate equity curve def calculate_equity_curve(data): data['returns'] = data['Adj Close'].pct_change() data['strategy_returns'] = data['returns'] * data['signal'].shift(1) data['equity_curve'] = (1 + data['strategy_returns']).cumprod() return data # Function to optimize parameters for different trading terms def optimize_parameters(data, short_window_range, medium_window_range, long_window_range): best_params = None best_equity = 0 for short, medium, long in product(short_window_range, medium_window_range, long_window_range): if short < medium < long: df = calculate_moving_averages(data.copy(), short, medium, long) df = generate_signals(df) df = calculate_equity_curve(df) final_equity = df['equity_curve'].iloc[-1] if final_equity > best_equity: best_equity = final_equity best_params = (short, medium, long) return best_params, best_equity # Function to execute and plot the strategy def execute_strategy(data, short_window, medium_window, long_window, threshold, title_suffix): data = calculate_moving_averages(data, short_window, medium_window, long_window) data = generate_signals(data, threshold) data = calculate_equity_curve(data) return data # Function to plot results with subplots for better alignment def plot_results(data, params, title_suffix): # Create subplots: 2 rows (Price + MA, and Equity Curve), shared x-axis for alignment fig = make_subplots( rows=2, cols=1, shared_xaxes=True, vertical_spacing=0.1, # Increased vertical spacing between plots subplot_titles=(f'{title_suffix} Price and Moving Averages', 'Equity Curve') ) # Price and Moving Averages plot fig.add_trace(go.Scatter(x=data.index, y=data['Adj Close'], mode='lines', name='Price', line=dict(color='white')), row=1, col=1) fig.add_trace(go.Scatter(x=data.index, y=data['short_ma'], mode='lines', name=f'Short MA ({params[0]})', line=dict(color='blue')), row=1, col=1) fig.add_trace(go.Scatter(x=data.index, y=data['medium_ma'], mode='lines', name=f'Medium MA ({params[1]})', line=dict(color='orange')), row=1, col=1) fig.add_trace(go.Scatter(x=data.index, y=data['long_ma'], mode='lines', name=f'Long MA ({params[2]})', line=dict(color='green')), row=1, col=1) # Buy/Sell Signals with markers buy_signals = data[data['positions'] == 1] sell_signals = data[data['positions'] == -1] fig.add_trace(go.Scatter(x=buy_signals.index, y=buy_signals['short_ma'], mode='markers', name='Buy Signal', marker=dict(color='green', symbol='triangle-up', size=10)), row=1, col=1) fig.add_trace(go.Scatter(x=sell_signals.index, y=sell_signals['short_ma'], mode='markers', name='Sell Signal', marker=dict(color='red', symbol='triangle-down', size=10)), row=1, col=1) # Equity Curve plot fig.add_trace(go.Scatter(x=data.index, y=data['equity_curve'], mode='lines', name='Equity Curve', line=dict(color='blue')), row=2, col=1) # Update layout for better clarity and spacing fig.update_layout( height=800, # Increased height for better visualization title_text=f'{title_suffix} 3-Way Moving Average Crossover', xaxis_title='Date', yaxis_title='Price', legend=dict(orientation="h", yanchor="bottom", y=1.15, xanchor="center", x=0.5), margin=dict(t=30, b=30), # Adjust top and bottom margins for spacing font=dict(size=12), showlegend=True ) # Display the chart in Streamlit st.plotly_chart(fig, use_container_width=True) # Load and cache data data = download_data(ticker, start_date, end_date) # Short, Medium, Long-Term settings horizons = { "Short-Term": {"short_range": range(2, 10, 1), "medium_range": range(10, 20, 1), "long_range": range(20, 50, 2)}, "Medium-Term": {"short_range": range(10, 30, 2), "medium_range": range(30, 60, 3), "long_range": range(60, 100, 5)}, "Long-Term": {"short_range": range(30, 60, 5), "medium_range": range(60, 120, 10), "long_range": range(120, 200, 10)} } # Cache the results for each horizon so they persist when switching between pages if "results_cache" not in st.session_state: st.session_state["results_cache"] = {} # Initialize or update the MA parameters and threshold based on the selected page if page in st.session_state["results_cache"]: params = st.session_state["results_cache"][page]["params"] threshold_value = st.session_state["results_cache"][page]["threshold"] else: params = None threshold_value = 0.01 # Default value for threshold # Run Strategy Button run_strategy = st.sidebar.button(f"Run Strategy for {page}") run_with_adjusted_params = False # If Run Strategy is clicked, run optimization and reset parameters if run_strategy: horizon_settings = horizons.get(page) # Re-run optimization and reset to best parameters best_params, best_equity = optimize_parameters( data, short_window_range=horizon_settings["short_range"], medium_window_range=horizon_settings["medium_range"], long_window_range=horizon_settings["long_range"] ) # Cache the best parameters and reset adjusted parameters to best params st.session_state["results_cache"][page] = { "best_params": best_params, "best_equity": best_equity, "threshold": threshold_value, # Store the default threshold initially "params": best_params, # Reset to best params after optimization } # Reset sliders to best parameters after optimization params = best_params run_with_adjusted_params = True # If user-adjusted parameters (after the initial run) if params: horizon_settings = horizons.get(page) short_window = st.sidebar.slider( f"Short MA Window ({page})", min_value=horizon_settings["short_range"].start, max_value=horizon_settings["short_range"].stop - 1, value=params[0], help="Defines the window for the shortest moving average. Increasing this value smooths the moving average and reduces its sensitivity to price changes." ) medium_window = st.sidebar.slider( f"Medium MA Window ({page})", min_value=horizon_settings["medium_range"].start, max_value=horizon_settings["medium_range"].stop - 1, value=params[1], help="Defines the window for the medium moving average. A larger window increases smoothing and lags price changes more than the short MA." ) long_window = st.sidebar.slider( f"Long MA Window ({page})", min_value=horizon_settings["long_range"].start, max_value=horizon_settings["long_range"].stop - 1, value=params[2], help="Defines the window for the long moving average. A larger window results in a much slower-moving average that tracks long-term trends." ) threshold = st.sidebar.slider( f"Threshold ({page})", 0.0, 0.05, threshold_value, 0.01, help="Adjusts the strictness of the crossover signals. A higher threshold generates fewer, stricter signals." ) # If any adjustments are made to the parameters, mark the run as "adjusted" run_with_adjusted_params = True # Execute the strategy using user-adjusted parameters result_data = execute_strategy(data.copy(), short_window, medium_window, long_window, threshold, page) # Cache updated parameters and threshold without overwriting the best params st.session_state["results_cache"][page]["params"] = (short_window, medium_window, long_window) st.session_state["results_cache"][page]["threshold"] = threshold st.session_state["results_cache"][page]["data"] = result_data # If results are cached, display them if page in st.session_state["results_cache"]: cached_result = st.session_state["results_cache"][page] # Display best parameters in JSON (always show the optimized "best" params, not the adjusted ones) st.json({ "Best Parameters": { "Short MA": cached_result["best_params"][0], "Medium MA": cached_result["best_params"][1], "Long MA": cached_result["best_params"][2], "Threshold": cached_result["threshold"], "Final Equity": cached_result["best_equity"] } }) # Plot results with either optimized or adjusted parameters if "data" in cached_result: plot_results(cached_result["data"], cached_result["params"], page) hide_streamlit_style = """ """ st.markdown(hide_streamlit_style, unsafe_allow_html=True)