Create app.py

app.py

@@ -0,0 +1,361 @@
+import streamlit as st
+import numpy as np
+import pandas as pd
+import yfinance as yf
+import plotly.graph_objects as go
+import datetime
+
+# Set wide page layout
+st.set_page_config(
+    page_title="Pattern Recognition with KNN and Lorentzian Distance",
+    layout="wide"
+)
+
+# --- Sidebar Inputs ---
+st.sidebar.title("Input Parameters")
+with st.sidebar.expander("Data Parameters", expanded=True):
+    ticker = st.text_input("Ticker", value="ASML.AS", help="Enter the ticker symbol.")
+    start_date = st.date_input("Start Date", value=datetime.date(2022, 1, 1), help="Select start date for daily data.")
+    end_date = st.date_input("End Date", value=datetime.date.today() + datetime.timedelta(days=1), help="Select end date for daily data.")
+    
+with st.sidebar.expander("Model Parameters", expanded=True):
+    neighborsCount = st.number_input(
+        "KNN Neighbors Count",
+        value=100,
+        min_value=1,
+        step=1,
+        help="Higher = smoother signals, lower = more reactive."
+    )
+    maxBarsBack = st.number_input(
+        "Lookback Bars",
+        value=500,
+        min_value=1,
+        step=1,
+        help="How far back to search for similar patterns. Longer = more data, shorter = more recent context."
+    )
+    window_length = st.number_input(
+        "Window Length",
+        value=5,
+        min_value=1,
+        step=1,
+        help="Number of bars per pattern. Longer = more structure, shorter = more sensitivity."
+    )
+    barLookahead = st.number_input(
+        "Bar Lookahead",
+        value=4,
+        min_value=1,
+        step=1,
+        help="How far ahead to judge outcomes. Longer = trend focus, shorter = short-term bias."
+    )
+    
+run_button = st.sidebar.button("Run Analysis")
+
+# --- Title and Theory ---
+st.title("Pattern Recognition via Unsupervised KNN")
+st.markdown("#### **Compare recent market behavior to similar historical setups.**")
+
+st.write("This tool leverages historical price patterns to generate trading signals via self-supervised pattern recall. Instead of training a model, it compares the current market state to similar past conditions using a custom KNN approach with Lorentzian distance.")
+
+with st.expander("Methodology", expanded=False):
+    st.write(
+"""
+
+**Market State Representation**  
+Each trading day is represented by a feature vector built from 5 technical indicators:
+- **RSI:** Measures momentum.
+- **Wave Trend Oscillator:** Smooths price data.
+- **CCI:** Quantifies deviation from a moving average.
+- **ADX:** Assesses trend strength.
+- **Short-Term RSI:** Captures faster momentum.
+
+A sliding window (default: 5 bars) forms a flattened feature vector (5 indicators × 5 bars = 25 values) that captures short-term behavior.
+
+**Lorentzian Distance**  
+Similarity between market states is measured using Lorentzian distance:
+$$
+d(a, b) = \\sum_{i=1}^{N} \\log\\Big(1 + \\left|a_i - b_i\\right|\\Big)
+$$
+This function reduces the impact of extreme differences, making it robust to outliers.
+
+**KNN-Based Signal Generation**  
+For each new market state, the tool:
+1. Compares its feature vector to past states within a user-defined lookback window.
+2. Selects the \(k\) nearest neighbors (default: 100) using Lorentzian distance.
+3. Retrieves future price movement labels (over the next 4 bars by default):
+   - \(+1\) if the price rises.
+   - \(-1\) if the price falls.
+   - \(0\) if the price remains flat.
+4. Sums these labels to create a directional score:
+   - A positive sum indicates a long signal.
+   - A negative sum indicates a short signal.
+   - A zero sum retains the previous signal.
+
+**User Adjustable Variables**  
+You can adjust the following parameters in the sidebar. Each one controls how the pattern recognition behaves:
+
+- **KNN Neighbors Count:**  
+  Sets how many similar past patterns to compare against.  
+  - Higher values smooth the signal and reduce noise.  
+  - Lower values make it more reactive but may introduce false signals.
+
+- **Lookback Bars:**  
+  Defines how far back in history to search for similar patterns.  
+  - A longer lookback gives more pattern variety but may include outdated behavior.  
+  - A shorter lookback limits comparisons to recent market regimes.
+
+- **Window Length:**  
+  Determines how many consecutive bars are used to form each pattern (i.e., feature vector).  
+  - Longer windows capture broader structure but reduce signal frequency.  
+  - Shorter windows react faster but may miss context.
+
+- **Bar Lookahead:**  
+  Controls how far ahead the tool checks to define “what happened” after each past setup.  
+  - A longer lookahead focuses on trend outcomes.  
+  - A shorter lookahead favors short-term price moves.
+"""
+    )
+
+
+if run_button:
+    try:
+        with st.spinner("Running analysis..."):
+            # --- Download Data ---
+            df = yf.download(ticker, start=start_date, end=end_date, interval="1d")
+            if df.empty:
+                st.error("No data returned. Please check your inputs.")
+                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=["Close", "High", "Low"], inplace=True)
+            df["Date"] = df.index
+            df.reset_index(drop=True, inplace=True)
+            n = len(df)
+            
+            # --- Indicator Functions ---
+            def rsi(series, length=14):
+                delta = series.diff()
+                gain = delta.clip(lower=0)
+                loss = -delta.clip(upper=0)
+                avg_gain = gain.ewm(alpha=1/length, adjust=False).mean()
+                avg_loss = loss.ewm(alpha=1/length, adjust=False).mean()
+                rs = avg_gain / avg_loss
+                return 100 - (100 / (1 + rs))
+            
+            def wave_trend(hlc3, n1=10, n2=11):
+                esa = hlc3.ewm(span=n1, adjust=False).mean()
+                d = abs(hlc3 - esa).ewm(span=n1, adjust=False).mean()
+                ci = (hlc3 - esa) / (0.015 * d)
+                wt = ci.ewm(span=n2, adjust=False).mean()
+                return wt
+            
+            def cci(series, length=20):
+                ma = series.rolling(length).mean()
+                md = (series - ma).abs().rolling(length).mean()
+                return (series - ma) / (0.015 * md)
+            
+            def adx(df, length=14):
+                high = df["High"]
+                low = df["Low"]
+                close = df["Close"]
+                plus_dm = (high - high.shift(1)).clip(lower=0)
+                minus_dm = (low.shift(1) - low).clip(lower=0)
+                plus_dm[plus_dm < minus_dm] = 0
+                minus_dm[minus_dm <= plus_dm] = 0
+
+                tr1 = df["High"] - df["Low"]
+                tr2 = abs(df["High"] - close.shift(1))
+                tr3 = abs(df["Low"] - close.shift(1))
+                tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)
+                atr = tr.ewm(alpha=1/length, adjust=False).mean()
+                plus_di = 100 * (plus_dm.ewm(alpha=1/length, adjust=False).mean() / atr)
+                minus_di = 100 * (minus_dm.ewm(alpha=1/length, adjust=False).mean() / atr)
+                dx = 100 * abs(plus_di - minus_di) / (plus_di + minus_di)
+                return dx.ewm(alpha=1/length, adjust=False).mean()
+            
+            # --- Build Features ---
+            df["hlc3"] = (df["High"] + df["Low"] + df["Close"]) / 3.0
+            df["feat1"] = rsi(df["Close"], 14)
+            df["feat2"] = wave_trend(df["hlc3"], 10, 11)
+            df["feat3"] = cci(df["Close"], 20)
+            df["feat4"] = adx(df, 14)
+            df["feat5"] = rsi(df["Close"], 9)
+            
+            features = df[["feat1", "feat2", "feat3", "feat4", "feat5"]].to_numpy()
+            features_windowed = np.array([
+                features[i - window_length + 1: i + 1].flatten()
+                for i in range(window_length - 1, n)
+            ])
+            n_window = features_windowed.shape[0]
+            
+            # --- Lorentzian Distance & KNN ---
+            def lorentzian_distance(a, b):
+                return np.sum(np.log1p(np.abs(a - b)))
+            
+            y_train = np.zeros(n, dtype=int)
+            for i in range(n - barLookahead):
+                if df["Close"].iloc[i + barLookahead] > df["Close"].iloc[i]:
+                    y_train[i] = 1
+                elif df["Close"].iloc[i + barLookahead] < df["Close"].iloc[i]:
+                    y_train[i] = -1
+                else:
+                    y_train[i] = 0
+            
+            prediction_arr = np.zeros(n_window, dtype=float)
+            for idx in range(n_window):
+                global_idx = idx + window_length - 1
+                if global_idx < maxBarsBack:
+                    prediction_arr[idx] = 0
+                    continue
+                start_idx = max(0, idx - maxBarsBack)
+                dist_list = []
+                idx_list = []
+                for j in range(start_idx, idx):
+                    d = lorentzian_distance(features_windowed[idx], features_windowed[j])
+                    dist_list.append(d)
+                    idx_list.append(j)
+                dist_list = np.array(dist_list)
+                idx_list = np.array(idx_list)
+                if len(dist_list) > 0:
+                    k = min(neighborsCount, len(dist_list))
+                    nearest = np.argpartition(dist_list, k)[:k]
+                    neighbor_labels = y_train[idx_list[nearest] + window_length - 1]
+                    prediction_arr[idx] = neighbor_labels.sum()
+                else:
+                    prediction_arr[idx] = 0
+            
+            # --- Signal Logic ---
+            signal = np.zeros(n_window, dtype=int)
+            for idx in range(1, n_window):
+                if prediction_arr[idx] > 0:
+                    signal[idx] = 1
+                elif prediction_arr[idx] < 0:
+                    signal[idx] = -1
+                else:
+                    signal[idx] = signal[idx - 1]
+            
+            startLong = np.zeros(n_window, dtype=bool)
+            startShort = np.zeros(n_window, dtype=bool)
+            for idx in range(1, n_window):
+                startLong[idx] = (signal[idx] == 1) and (signal[idx - 1] != 1)
+                startShort[idx] = (signal[idx] == -1) and (signal[idx - 1] != -1)
+            
+            n_long_signals = int(np.count_nonzero(startLong))
+            n_short_signals = int(np.count_nonzero(startShort))
+            
+            # --- 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"
+            ))
+            
+            pos_x, pos_y = [], []
+            neg_x, neg_y = [], []
+            neu_x, neu_y = [], []
+            for idx in range(n_window):
+                global_idx = idx + window_length - 1
+                x_date = df["Date"].iloc[global_idx]
+                y_low = df["Low"].iloc[global_idx]
+                y_high = df["High"].iloc[global_idx]
+                if prediction_arr[idx] > 0:
+                    pos_x.extend([x_date, x_date, None])
+                    pos_y.extend([y_low, y_high, None])
+                elif prediction_arr[idx] < 0:
+                    neg_x.extend([x_date, x_date, None])
+                    neg_y.extend([y_low, y_high, None])
+                else:
+                    neu_x.extend([x_date, x_date, None])
+                    neu_y.extend([y_low, y_high, None])
+            
+            if pos_x:
+                fig.add_trace(go.Scatter(
+                    x=pos_x,
+                    y=pos_y,
+                    mode="lines",
+                    line=dict(color="rgba(0,204,0,0.5)", width=1.0),
+                    name="Positive predictions"
+                ))
+            if neg_x:
+                fig.add_trace(go.Scatter(
+                    x=neg_x,
+                    y=neg_y,
+                    mode="lines",
+                    line=dict(color="rgba(204,0,0,0.5)", width=1.0),
+                    name="Negative predictions"
+                ))
+            if neu_x:
+                fig.add_trace(go.Scatter(
+                    x=neu_x,
+                    y=neu_y,
+                    mode="lines",
+                    line=dict(color="rgba(179,179,179,0.3)", width=1.0),
+                    name="Neutral predictions"
+                ))
+            
+            long_x, long_y = [], []
+            short_x, short_y = [], []
+            for idx in range(1, n_window):
+                global_idx = idx + window_length - 1
+                if startLong[idx]:
+                    long_x.append(df["Date"].iloc[global_idx])
+                    long_y.append(df["Low"].iloc[global_idx] * 0.99)
+                elif startShort[idx]:
+                    short_x.append(df["Date"].iloc[global_idx])
+                    short_y.append(df["High"].iloc[global_idx] * 1.01)
+            
+            if long_x:
+                fig.add_trace(go.Scatter(
+                    x=long_x,
+                    y=long_y,
+                    mode='markers',
+                    marker=dict(symbol="triangle-up", size=10, color="lime", line=dict(color="white", width=1)),
+                    name="Long Entry"
+                ))
+            if short_x:
+                fig.add_trace(go.Scatter(
+                    x=short_x,
+                    y=short_y,
+                    mode='markers',
+                    marker=dict(symbol="triangle-down", size=10, color="red", line=dict(color="white", width=1)),
+                    name="Short Entry"
+                ))
+            
+            fig.update_layout(
+                template="plotly_dark",
+                title=dict(text=f"{ticker} — KNN Signals via Lorentzian Distance ({start_date} to {end_date})", font=dict(color="white")),
+                xaxis=dict(title="Date", tickformat="%Y-%m-%d", titlefont=dict(color="white"), tickfont=dict(color="white")),
+                yaxis=dict(title="Price", titlefont=dict(color="white"), tickfont=dict(color="white")),
+                legend=dict(font=dict(color="white"))
+            )
+            fig.update_xaxes(showgrid=True, gridcolor="grey")
+            fig.update_yaxes(showgrid=True, gridcolor="grey")
+        
+        # --- Output Only the Chart ---
+        st.markdown("### Price and Signal Annotations")
+        st.markdown(
+            f"""
+        The chart below shows **{ticker}** close price along with signals derived from historical pattern similarity. **Gray bars** mark the initial lookback window with no predictions.  
+
+        """
+        )
+        st.write(f"Long signals: {n_long_signals}, Short signals: {n_short_signals}.")
+        st.plotly_chart(fig, use_container_width=True)
+        
+    except Exception:
+        st.error("An error occurred during the analysis.")
+
+# Hide default Streamlit style
+st.markdown(
+    """
+    <style>
+    #MainMenu {visibility: hidden;}
+    footer {visibility: hidden;}
+    </style>
+    """,
+    unsafe_allow_html=True
+)