app.py

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

# Set Streamlit page configuration
st.set_page_config(page_title="Pivot Point-RSI Volatility Trading Strategy", layout="wide")

# App title
st.title("Pivot Point-RSI Volatility Trading Strategy")

# Description of the app
st.markdown('''
This tool executes a trading strategy that combines Pivot Points, RSI, and ATR to identify buy and sell signals. 
It works by calculating support and resistance levels using Pivot Points, determining market momentum through RSI, and adjusting 
stop-loss and target levels based on market volatility using ATR. 
The strategy finds the best possible parameters through optimization. You can also manually input your own settings. 
''')

# Sidebar: How to use the app
with st.sidebar.expander("How to Use", expanded=False):
    st.write('''
    1. Select the asset and date range for backtesting.
    2. Click "Run Strategy" to generate results.
    3. View the equity curve, buy/sell signals, and performance metrics.
    4. Adjust the parameters post-run to fine-tune results.
    ''')

# Sidebar: Select Ticker and Date Range
with st.sidebar.expander("Asset Settings", expanded=True):
    ticker = st.text_input("Asset Symbol", value="ASML.AS", help="Ticker Symbol or Cryptocurrency Pair (e.g., AAPL, BTC-USD)")
    start_date = st.date_input("Start Date", value=datetime(2020, 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.")

# Download data
@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 retrieved for {ticker}")
    if len(data) < 24:  # Minimum needed for 24-period Pivot Point calculation
        raise ValueError(f"Insufficient data points for {ticker}. Need at least 24 days.")
    return data

# Function to execute and cache strategy results
@st.cache_data
def execute_strategy(data, strategy_params):
    data_with_signals = trading_strategy(data.copy(), **strategy_params)
    profit, equity_curve, accuracy_rate = backtest_strategy(data_with_signals)
    return data_with_signals, profit, equity_curve, accuracy_rate, strategy_params

# Trading strategy function
def trading_strategy(data, rsi_period, rsi_oversold, rsi_overbought, atr_period, stop_loss_multiplier, target_multiplier, tolerance):
    data['high24'] = data['High'].rolling(window=24).max()
    data['low24'] = data['Low'].rolling(window=24).min()
    data['close24'] = data['Close'].rolling(window=24).mean()

    data['pivot'] = (data['high24'] + data['low24'] + data['close24']) / 3
    data['support'] = (2 * data['pivot'].rolling(window=12).min()) - data['high24']
    data['resistance'] = (2 * data['pivot'].rolling(window=12).max()) - data['low24']

    delta = data['Close'].diff()
    up, down = delta.copy(), delta.copy()
    up[up < 0] = 0
    down[down > 0] = 0

    roll_up1 = up.ewm(span=rsi_period).mean()
    roll_down1 = down.abs().ewm(span=rsi_period).mean()
    RS1 = roll_up1 / roll_down1
    data['rsi'] = 100.0 - (100.0 / (1.0 + RS1))

    data['HL'] = data['High'] - data['Low']
    data['absHC'] = abs(data['High'] - data['Close'].shift())
    data['absLC'] = abs(data['Low'] - data['Close'].shift())
    data['TR'] = data[['HL', 'absHC', 'absLC']].max(axis=1)
    data['atr'] = data['TR'].rolling(window=atr_period).mean()

    data['signal'] = np.where(
        (data['Close'] >= data['support'] * (1 - tolerance)) & (data['Close'] <= data['support'] * (1 + tolerance)) & (data['rsi'] < rsi_oversold), 'Buy',
        np.where((data['Close'] >= data['resistance'] * (1 - tolerance)) & (data['Close'] <= data['resistance'] * (1 + tolerance)) & (data['rsi'] > rsi_overbought), 'Sell', 'Hold')
    )

    data['stop_loss'] = data['Close'] - stop_loss_multiplier * data['atr']
    data['target'] = data['Close'] + target_multiplier * data['atr']

    return data

# Backtest function
def backtest_strategy(data):
    initial_capital = 100000  
    position = 0  
    capital = initial_capital
    equity_curve = []
    correct_signals = 0
    total_signals = 0

    for index, row in data.iterrows():
        if row['signal'] == 'Buy' and position == 0:
            position = 1
            entry_price = row['Close']
            stop_loss = row['stop_loss']
            target = row['target']
                
        elif row['signal'] == 'Sell' and position == 0:
            position = -1
            entry_price = row['Close']
            stop_loss = row['stop_loss']
            target = row['target']
                
        if position == 1:
            if row['Low'] <= stop_loss:  
                capital += (stop_loss - entry_price) * 100  
                position = 0
            elif row['High'] >= target:  
                capital += (target - entry_price) * 100
                position = 0
            elif row['signal'] == 'Sell':  
                capital += (row['Close'] - entry_price) * 100
                position = 0
                
        elif position == -1:
            if row['High'] >= stop_loss:  
                capital += (entry_price - stop_loss) * 100  
                position = 0
            elif row['Low'] <= target:  
                capital += (entry_price - target) * 100
                position = 0
            elif row['signal'] == 'Buy':  
                capital += (entry_price - row['Close']) * 100
                position = 0
                
        equity_curve.append(capital)

    final_profit = equity_curve[-1] - initial_capital
    return final_profit, equity_curve, 0

# Optimized parameters (fixed)
optimized_params = {
    'rsi_period': 13,
    'rsi_oversold': 30,
    'rsi_overbought': 70,
    'atr_period': 14,
    'stop_loss_multiplier': 2.25,
    'target_multiplier': 4.5,
    'tolerance': 0.01
}

# Run button to initiate the strategy
run_button = st.sidebar.button("Run Strategy")

# Running the strategy
if run_button:
    try:
        # Download data
        data = download_data(ticker, start_date, end_date)

        # Execute the strategy with optimized parameters
        data_with_signals, profit, equity_curve, _, best_params = execute_strategy(data, optimized_params)

        # Cache results in session state
        st.session_state['data_with_signals'] = data_with_signals
        st.session_state['equity_curve'] = equity_curve
        st.session_state['best_params'] = best_params

        # Display optimized parameters
        st.json(best_params)
    except Exception as e:
        st.error(f"An error occurred while running the analysis: {e}")

# If session state contains data, allow for post-run parameter adjustments
if 'data_with_signals' in st.session_state:
    st.sidebar.markdown("### Adjust Parameters Post-Run")

    adjusted_rsi_period = st.sidebar.slider("RSI Period", 5, 30, st.session_state['best_params']['rsi_period'], step=1, help="RSI period defines the sensitivity of the RSI. Lower values make the RSI more sensitive to price changes, generating more signals. Higher values smooth out price movements, reducing the number of signals.")
    adjusted_stop_loss_multiplier = st.sidebar.slider("Stop Loss Multiplier", 1.0, 3.0, st.session_state['best_params']['stop_loss_multiplier'], step=0.1, help="Stop Loss Multiplier adjusts how far the stop-loss is set from the entry price. Lower values keep the stop-loss closer, reducing risk but increasing the chance of being stopped out. Higher values place the stop-loss further away, allowing more flexibility but with higher potential risk.")
    adjusted_target_multiplier = st.sidebar.slider("Target Multiplier", 1.0, 5.0, st.session_state['best_params']['target_multiplier'], step=0.1, help="Target Multiplier adjusts how far the profit target is set from the entry price. Lower values take profits earlier but reduce potential gains. Higher values aim for larger profits but increase the chance of price reversal before hitting the target.")
    adjusted_tolerance = st.sidebar.slider("Signal Tolerance", 0.0, 0.05, st.session_state['best_params']['tolerance'], step=0.01, help="Signal Tolerance adjusts how strictly the strategy follows support and resistance levels. Lower values make signals more precise but reduce their frequency. Higher values increase the frequency of signals by being more lenient, which can lead to false signals.")

    # Recalculate the strategy with adjusted parameters
    updated_params = {
        'rsi_period': adjusted_rsi_period,
        'rsi_oversold': 30,
        'rsi_overbought': 70,
        'atr_period': 14,
        'stop_loss_multiplier': adjusted_stop_loss_multiplier,
        'target_multiplier': adjusted_target_multiplier,
        'tolerance': adjusted_tolerance
    }

    updated_data = trading_strategy(st.session_state['data_with_signals'].copy(), **updated_params)

    # Plotting with adjustments for easier comparison of x-axis
    fig = make_subplots(rows=3, cols=1, shared_xaxes=True,
                        subplot_titles=("Price and Bollinger Bands", "RSI", "Equity Curve"),
                        vertical_spacing=0.20)

    # Price and signal plot
    fig.add_trace(go.Scatter(x=updated_data.index, y=updated_data['Close'], mode='lines', name='Close Price'))
    fig.add_trace(go.Scatter(x=updated_data.index, y=updated_data['support'], mode='lines', name='Support Level', line=dict(dash='dash')))
    fig.add_trace(go.Scatter(x=updated_data.index, y=updated_data['resistance'], mode='lines', name='Resistance Level', line=dict(dash='dash')))

    # Buy/Sell Signals with increased marker size
    buy_signals = updated_data[updated_data['signal'] == 'Buy']
    sell_signals = updated_data[updated_data['signal'] == 'Sell']
    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=12)))
    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=12)))

    # RSI Plot with black line, red upper threshold, and green lower threshold
    fig.add_trace(go.Scatter(x=updated_data.index, y=updated_data['rsi'], mode='lines', name='RSI', line=dict(color='white')), row=2, col=1)
    fig.add_shape(type="line", x0=updated_data.index[0], x1=updated_data.index[-1], y0=30, y1=30, 
                  line=dict(dash='dash', color='green'), row=2, col=1)
    fig.add_shape(type="line", x0=updated_data.index[0], x1=updated_data.index[-1], y0=70, y1=70, 
                  line=dict(dash='dash', color='red'), row=2, col=1)

    # Equity Curve Plot
    fig.add_trace(go.Scatter(x=updated_data.index, y=st.session_state['equity_curve'], mode='lines', name='Equity Curve'), row=3, col=1)

    # Move the legend outside the plot and increase gap
    fig.update_layout(
        title=f'{ticker} Strategy with Adjusted Parameters',
        xaxis_title='Date',
        yaxis_title='Price',
        legend=dict(orientation="h", yanchor="bottom", y=1.15, xanchor="center", x=0.5, traceorder='normal', valign='top', borderwidth=0),
        height=1000,
        margin=dict(t=30, b=30),
        font=dict(size=12),
        annotations=[dict(font=dict(color="white", size=14)) for _ in range(3)]
    )

    # Display the chart
    st.plotly_chart(fig, use_container_width=True)

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