Update src/streamlit_app.py

src/streamlit_app.py

@@ -7,16 +7,17 @@ from sklearn.svm import SVR
 from sklearn.preprocessing import StandardScaler
 import plotly.graph_objects as go
 import plotly.express as px
-from datetime import datetime, timedelta, date
+from datetime import datetime, date
+
 # --- Config ---
-st.set_page_config(page_title="Hybrid HMM-SVR Strategy Backtester", layout="wide")
+st.set_page_config(page_title="HMM-SVR Leverage Sniper", layout="wide")
 
 # --- Helper Functions ---
 
 @st.cache_data(ttl=3600)
 def fetch_data(ticker, start_date, end_date):
     """
-    Robust data fetching with caching, error handling, and string conversion.
+    Robust data fetching with caching.
     """
     ticker = ticker.strip().upper()
     
@@ -49,7 +50,7 @@ def calculate_metrics(df, strategy_col='Strategy_Value', benchmark_col='Buy_Hold
     """Calculates CAG, Sharpe, Drawdown, etc."""
     stats = {}
     
-    for col, name in [(strategy_col, 'Hybrid Strategy'), (benchmark_col, 'Buy & Hold')]:
+    for col, name in [(strategy_col, 'Smart Leverage Strategy'), (benchmark_col, 'Buy & Hold')]:
         initial = df[col].iloc[0]
         final = df[col].iloc[-1]
         total_return = (final - initial) / initial
@@ -71,7 +72,7 @@ def calculate_metrics(df, strategy_col='Strategy_Value', benchmark_col='Buy_Hold
     return pd.DataFrame(stats)
 
 def train_hmm_model(train_df, n_states):
-    """Trains HMM on historical data (In-Sample)."""
+    """Trains HMM on historical data and sorts states by volatility (0=Low, n=High)."""
     X_train = train_df[['Log_Returns', 'Volatility']].values * 100
     
     model = GaussianHMM(n_components=n_states, covariance_type="full", n_iter=100, random_state=42)
@@ -82,6 +83,8 @@ def train_hmm_model(train_df, n_states):
     for i in range(n_states):
         avg_vol = X_train[hidden_states == i, 1].mean()
         state_vol.append((i, avg_vol))
+    
+    # Sort states: State 0 = Lowest Volatility (Safe), State N = Highest Volatility (Crash)
     state_vol.sort(key=lambda x: x[1])
     
     mapping = {old: new for new, (old, _) in enumerate(state_vol)}
@@ -105,61 +108,60 @@ def train_svr_model(train_df):
     return model, scaler
 
 def generate_trade_log(df):
-    """
-    Scans the backtest dataframe to identify individual trade cycles.
-    A 'Trade' is defined as a period where Position Size > 0.
-    """
+    """Generates a log of trades including leverage used."""
     trades = []
     in_trade = False
     entry_date = None
     entry_price = 0
     trade_returns = []
+    avg_leverage = []
     
-    # We iterate through the dataframe
     for date, row in df.iterrows():
         pos = row['Final_Position']
         close_price = row['Close']
+        lev = row['Position_Size'] # Capture leverage used
         
-        # Check for Entry (Position goes from 0 to > 0)
         if pos > 0 and not in_trade:
             in_trade = True
             entry_date = date
-            entry_price = close_price # Approximation for log visualization
-            trade_returns = [row['Strategy_Returns']] # Start tracking returns for this specific trade
+            entry_price = close_price
+            trade_returns = [row['Strategy_Returns']]
+            avg_leverage = [lev]
             
-        # Check for adjustments while in trade
         elif pos > 0 and in_trade:
             trade_returns.append(row['Strategy_Returns'])
+            avg_leverage.append(lev)
             
-        # Check for Exit (Position goes to 0 while we were in a trade)
         elif pos == 0 and in_trade:
             in_trade = False
             exit_date = date
             exit_price = close_price
             
-            # Calculate compounded return for this specific trade period
-            # (1+r1)*(1+r2)... - 1
             cum_trade_ret = np.prod([1 + r for r in trade_returns]) - 1
+            mean_lev = np.mean(avg_leverage)
             
             trades.append({
                 'Entry Date': entry_date,
                 'Exit Date': exit_date,
-                'Entry Price (Approx)': entry_price,
+                'Entry Price': entry_price,
                 'Exit Price': exit_price,
-                'Duration (Days)': len(trade_returns),
+                'Duration': len(trade_returns),
+                'Avg Leverage': f"{mean_lev:.1f}x",
                 'Trade PnL': cum_trade_ret
             })
             trade_returns = []
+            avg_leverage = []
 
-    # Handle case where trade is still open at end of data
     if in_trade:
         cum_trade_ret = np.prod([1 + r for r in trade_returns]) - 1
+        mean_lev = np.mean(avg_leverage)
         trades.append({
             'Entry Date': entry_date,
             'Exit Date': df.index[-1],
-            'Entry Price (Approx)': entry_price,
+            'Entry Price': entry_price,
             'Exit Price': df.iloc[-1]['Close'],
-            'Duration (Days)': len(trade_returns),
+            'Duration': len(trade_returns),
+            'Avg Leverage': f"{mean_lev:.1f}x",
             'Trade PnL': cum_trade_ret
         })
 
@@ -167,57 +169,51 @@ def generate_trade_log(df):
 
 # --- Main Logic ---
 
-st.title("🧠 Hybrid HMM-SVR Strategy Backtester")
+st.title("⚡ HMM-SVR  Leverage Backtester")
 st.markdown("""
-**The Hybrid Strategy:**
-1.  **Driver:** EMA Crossover (Fast > Slow = Bullish).
-2.  **Filter (HMM):** If Regime is "High Vol/Crash", **Block Trade** (Size = 0).
-3.  **Sizing (SVR):** If Regime is Safe, adjust size based on predicted risk. 
-    * *If SVR predicts higher risk -> Reduce Position Size.*
-    * *If SVR predicts lower risk -> Increase Position Size.*
+**The "Strict Rules" Strategy:**
+1.  **Baseline:** Buy when Fast EMA > Slow EMA.
+2.  **Safety (HMM):** If Regime = **High Volatility (Crash)** -> **Exit (0x)**.
+3.  **Leverage Boost (SVR + HMM):** * IF Regime is **Lowest Volatility (State 0)**
+    * AND SVR predicts volatility **< 50% of average** (Risk Ratio < 0.5)
+    * THEN **Leverage = 3x**.
 """)
 
 # Sidebar Inputs
 with st.sidebar:
     st.header("Settings")
     
-    # Added key='ticker_select'
     ticker = st.selectbox(
         "Ticker", 
-        ["BTC-USD", "BNB-USD"," ETH-USD","SOL-USD"],
+        ["BNB-USD", "ETH-USD", "SOL-USD", "BTC-USD"],
         key="ticker_select" 
     )
     
-    # Added key='start_date'
     backtest_start = st.date_input(
         "Backtest Start Date", 
         date(2022, 1, 1),
         key="start_date"
     )
 
-    # Added key='end_date'
     backtest_end = st.date_input(
         "Backtest End Date", 
         datetime.now(),
         key="end_date"
     )
     
-    st.caption("Note: Models will automatically train on the **4 years** of data prior to your selected Start Date.")
-    
     st.divider()
     
-    # Added keys for these as well just to be safe
-    short_window = st.number_input("Fast EMA", 12, key="fast_ema")
-    long_window = st.number_input("Slow EMA", 26, key="slow_ema")
-    n_states = st.slider("HMM States", 2, 4, 3, key="hmm_slider")
+    st.subheader("Leverage Rules")
+    leverage_mult = st.number_input("Boost Leverage (Certainty Multiplier)", value=3.0, step=0.5)
+    risk_threshold = st.slider("Certainty Threshold (Risk Ratio < X)", 0.1, 1.0, 0.5, help="Lower = Stricter. Only boost leverage when predicted risk is extremely low.")
 
-if st.button("Run Hybrid Backtest"):
+if st.button("Run Leverage Backtest"):
     train_start_date = pd.Timestamp(backtest_start) - pd.DateOffset(years=4)
     
     df = fetch_data(ticker, train_start_date, backtest_end)
     
     if df is None or len(df) < 200:
-        st.error(f"Not enough data found for {ticker}. Ensure the ticker existed 4 years prior to {backtest_start}.")
+        st.error(f"Not enough data found for {ticker}.")
     else:
         # 1. Feature Engineering
         df['Log_Returns'] = np.log(df['Close'] / df['Close'].shift(1))
@@ -226,8 +222,9 @@ if st.button("Run Hybrid Backtest"):
         df['Downside_Returns'] = df['Log_Returns'].apply(lambda x: x if x < 0 else 0)
         df['Downside_Vol'] = df['Downside_Returns'].rolling(window=10).std()
         
-        df['EMA_Short'] = df['Close'].ewm(span=short_window, adjust=False).mean()
-        df['EMA_Long'] = df['Close'].ewm(span=long_window, adjust=False).mean()
+        # 12/26 EMA standard
+        df['EMA_Short'] = df['Close'].ewm(span=12, adjust=False).mean()
+        df['EMA_Long'] = df['Close'].ewm(span=26, adjust=False).mean()
         
         df['Target_Next_Vol'] = df['Volatility'].shift(-1)
         
@@ -238,133 +235,126 @@ if st.button("Run Hybrid Backtest"):
         test_df = df[df.index >= pd.Timestamp(backtest_start)].copy()
         
         if len(train_df) < 365:
-            st.warning(f"Warning: Only {len(train_df)} days found for training. HMM performs best with >2 years of data.")
+            st.warning(f"Warning: Only {len(train_df)} days found for training.")
         
         if len(test_df) < 10:
              st.error("Not enough data for backtesting range.")
         else:
-            st.info(f"Training on {len(train_df)} days ({train_df.index[0].date()} to {train_df.index[-1].date()}). Backtesting on {len(test_df)} days.")
+            n_states = 3 # Fixed 3 states: Low, Neutral, High
             
-            with st.spinner("Training HMM (Regime Detection)..."):
+            with st.spinner("Training AI Models (HMM & SVR)..."):
+                # Train HMM
                 hmm_model, state_map = train_hmm_model(train_df, n_states)
                 
+                # Get HMM Regimes for Train set (needed for SVR training)
                 X_train_hmm = train_df[['Log_Returns', 'Volatility']].values * 100
                 train_raw_states = hmm_model.predict(X_train_hmm)
                 train_df['Regime'] = [state_map.get(s, s) for s in train_raw_states]
                 
-            with st.spinner("Training SVR (Volatility Forecasting)..."):
+                # Train SVR
                 svr_model, svr_scaler = train_svr_model(train_df)
                 
-            with st.spinner("Running Backtest Loop..."):
                 # --- OUT OF SAMPLE BACKTEST ---
                 
+                # Predict Regimes
                 X_test_hmm = test_df[['Log_Returns', 'Volatility']].values * 100
                 test_raw_states = hmm_model.predict(X_test_hmm)
                 test_df['Regime'] = [state_map.get(s, s) for s in test_raw_states]
                 
+                # Predict Next Day Volatility
                 X_test_svr = test_df[['Log_Returns', 'Volatility', 'Downside_Vol', 'Regime']].values
                 X_test_svr_scaled = svr_scaler.transform(X_test_svr)
                 test_df['Predicted_Vol'] = svr_model.predict(X_test_svr_scaled)
                 
-                high_vol_state = n_states - 1
+                # --- STRICT LEVERAGE LOGIC ---
                 
+                # 1. Base Signal (Trend)
                 test_df['Signal'] = np.where(test_df['EMA_Short'] > test_df['EMA_Long'], 1, 0)
                 
+                # 2. Calculate Confidence
                 avg_train_vol = train_df['Volatility'].mean()
-                
                 test_df['Risk_Ratio'] = test_df['Predicted_Vol'] / avg_train_vol
-                test_df['Position_Size'] = (1.0 / test_df['Risk_Ratio']).clip(upper=1.0, lower=0.0)
+                
+                # 3. Apply "The Rules"
+                
+                # Rule A: Default Size
+                test_df['Position_Size'] = 1.0
+                
+                # Rule B: The "Certainty" Boost
+                # If Regime is lowest volatility (State 0) AND Risk Ratio is low (< threshold)
+                # Then apply leverage
+                condition_safe_regime = (test_df['Regime'] == 0)
+                condition_low_risk_prediction = (test_df['Risk_Ratio'] < risk_threshold)
                 
                 test_df['Position_Size'] = np.where(
-                    test_df['Regime'] == high_vol_state ,
-                    0.0, 
+                    condition_safe_regime & condition_low_risk_prediction,
+                    leverage_mult, # User selected leverage (e.g., 2.0x)
                     test_df['Position_Size']
                 )
                 
+                # Rule C: The "Danger" Cut
+                # If Regime is Highest Volatility (State n-1) -> Go to 0
+                condition_crash_regime = (test_df['Regime'] == (n_states - 1))
+                
+                test_df['Position_Size'] = np.where(
+                    condition_crash_regime,
+                    0.0,
+                    test_df['Position_Size']
+                )
+                
+                # Final Position Calculation
+                # Shift by 1 because we act on Today's close for Tomorrow's return
                 test_df['Final_Position'] = (test_df['Signal'] * test_df['Position_Size']).shift(1)
                 
+                # Returns Calculation
                 test_df['Simple_Returns'] = test_df['Close'].pct_change()
                 test_df['Strategy_Returns'] = test_df['Final_Position'] * test_df['Simple_Returns']
-                test_df['Buy_Hold_Returns'] = test_df['Simple_Returns']
                 
-                test_df['Strategy_Value'] = (1 + test_df['Strategy_Returns'].fillna(0)).cumprod()
-                test_df['Buy_Hold_Value'] = (1 + test_df['Buy_Hold_Returns'].fillna(0)).cumprod()
+                # --- METRICS & VISUALS ---
                 
+                test_df['Strategy_Value'] = (1 + test_df['Strategy_Returns'].fillna(0)).cumprod()
+                test_df['Buy_Hold_Value'] = (1 + test_df['Simple_Returns'].fillna(0)).cumprod()
                 test_df.dropna(inplace=True)
                 
-                # --- EXTRACT TRADES ---
-                trade_log = generate_trade_log(test_df)
-                
-                # --- RESULTS ---
-                
                 metrics_df = calculate_metrics(test_df)
-                st.subheader("Performance Metrics")
+                
+                st.subheader("Performance vs Benchmark")
                 st.table(metrics_df)
                 
-                # Charts
-                col1, col2 = st.columns([2, 1])
+                # Plot 1: Equity Curve
+                st.subheader("Equity Curve")
+                fig = go.Figure()
+                fig.add_trace(go.Scatter(x=test_df.index, y=test_df['Buy_Hold_Value'], name='Buy & Hold', line=dict(color='gray', dash='dot')))
+                fig.add_trace(go.Scatter(x=test_df.index, y=test_df['Strategy_Value'], name='Smart Leverage Strategy', line=dict(color='#00CC96', width=2)))
+                st.plotly_chart(fig, use_container_width=True)
                 
-                with col1:
-                    st.subheader("Equity Curve & Trade Executions")
-                    fig = go.Figure()
-                    
-                    # 1. Equity Curves
-                    fig.add_trace(go.Scatter(x=test_df.index, y=test_df['Buy_Hold_Value'], name='Buy & Hold', line=dict(color='gray', dash='dot')))
-                    fig.add_trace(go.Scatter(x=test_df.index, y=test_df['Strategy_Value'], name='Hybrid Strategy', line=dict(color='#00CC96', width=2)))
-                    
-                    # 2. Add Trade Markers
-                    # Filter Entry Points (Buy)
-                    if not trade_log.empty:
-                        # Map dates to Strategy Value for Y-axis placement
-                        buy_points = trade_log.set_index('Entry Date')
-                        buy_vals = test_df.loc[buy_points.index]['Strategy_Value']
-                        
-                        sell_points = trade_log.set_index('Exit Date')
-                        sell_vals = test_df.loc[sell_points.index]['Strategy_Value']
-
-                        fig.add_trace(go.Scatter(
-                            x=buy_points.index, 
-                            y=buy_vals,
-                            mode='markers',
-                            name='Buy Signal',
-                            marker=dict(symbol='triangle-up', size=10, color='lime')
-                        ))
-                        
-                        fig.add_trace(go.Scatter(
-                            x=sell_points.index, 
-                            y=sell_vals,
-                            mode='markers',
-                            name='Sell Signal',
-                            marker=dict(symbol='triangle-down', size=10, color='red')
-                        ))
-
-                    st.plotly_chart(fig, use_container_width=True)
-                    
-                with col2:
-                    st.subheader("Position Sizing (SVR Effect)")
-                    st.caption("How SVR adjusted trade size over time (0.0 to 1.0)")
-                    fig_size = px.area(test_df, x=test_df.index, y='Position_Size', title="Dynamic Exposure")
-                    st.plotly_chart(fig_size, use_container_width=True)
+                # Plot 2: Leverage Deployment
+                st.subheader("Leverage Deployment (0x, 1x, or 2x)")
+                st.caption("Notice how it shifts to 2x (Green Fill) only during smooth uptrends.")
                 
-                # --- NEW: Trade Log Table ---
+                fig_lev = go.Figure()
+                fig_lev.add_trace(go.Scatter(
+                    x=test_df.index, 
+                    y=test_df['Position_Size'], 
+                    mode='lines',
+                    fill='tozeroy',
+                    name='Leverage Used',
+                    line=dict(color='#636EFA')
+                ))
+                st.plotly_chart(fig_lev, use_container_width=True)
+                
+                # Trade Log
                 st.divider()
-                st.subheader("📝 Detailed Trade Log")
+                trade_log = generate_trade_log(test_df)
+                st.subheader("📝 Leverage Trade Log")
                 if not trade_log.empty:
-                    # Formatting for cleaner display
+                    # Formatting
                     display_log = trade_log.copy()
                     display_log['Entry Date'] = display_log['Entry Date'].dt.date
                     display_log['Exit Date'] = display_log['Exit Date'].dt.date
                     display_log['Trade PnL'] = display_log['Trade PnL'].map('{:.2%}'.format)
-                    display_log['Entry Price (Approx)'] = display_log['Entry Price (Approx)'].map('{:.2f}'.format)
+                    display_log['Entry Price'] = display_log['Entry Price'].map('{:.2f}'.format)
                     display_log['Exit Price'] = display_log['Exit Price'].map('{:.2f}'.format)
-                    
                     st.dataframe(display_log, use_container_width=True)
                 else:
-                    st.write("No trades executed in this period.")
-
-                st.subheader("SVR Prediction Accuracy (Test Set)")
-                fig_svr = go.Figure()
-                slice_df = test_df.iloc[-100:] 
-                fig_svr.add_trace(go.Scatter(x=slice_df.index, y=slice_df['Target_Next_Vol'], name='Actual Volatility'))
-                fig_svr.add_trace(go.Scatter(x=slice_df.index, y=slice_df['Predicted_Vol'], name='SVR Prediction', line=dict(dash='dot')))
-                st.plotly_chart(fig_svr, use_container_width=True)
+                    st.write("No trades generated.")