Spaces:
Running
Running
import warnings
warnings.filterwarnings("ignore")
import os
import re
import time
import requests
from datetime import time as dt_time
import streamlit as st
import yfinance as yf
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import matplotlib.gridspec as gridspec
import plotly.graph_objects as go
import plotly.express as px
from numpy.lib.stride_tricks import sliding_window_view
from scipy.stats import spearmanr, binom as binom_dist
from sklearn.ensemble import GradientBoostingClassifier
from huggingface_hub import HfApi, hf_hub_download
from scipy.signal import find_peaks
st.set_page_config(page_title="Market Dashboard", page_icon="π", layout="wide")
# =========================================================
# SHARED: PERSISTENT SIGNAL LOG (Hugging Face Dataset backend)
# =========================================================
HF_TOKEN = os.environ.get("HF_TOKEN")
DATASET_REPO = os.environ.get("HF_DATASET_REPO")
LOG_FILENAME = "signal_log.csv"
def _log_configured() -> bool:
return bool(HF_TOKEN and DATASET_REPO)
def load_signal_log() -> pd.DataFrame:
if not _log_configured():
return pd.DataFrame()
try:
local_path = hf_hub_download(
repo_id=DATASET_REPO, repo_type="dataset", filename=LOG_FILENAME, token=HF_TOKEN
)
df = pd.read_csv(local_path)
if "FlagDate" in df.columns:
df["FlagDate"] = pd.to_datetime(df["FlagDate"])
return df
except Exception:
return pd.DataFrame()
def save_signal_log(df: pd.DataFrame) -> bool:
if not _log_configured():
return False
try:
api = HfApi(token=HF_TOKEN)
csv_bytes = df.to_csv(index=False).encode("utf-8")
api.upload_file(
path_or_fileobj=csv_bytes,
path_in_repo=LOG_FILENAME,
repo_id=DATASET_REPO,
repo_type="dataset",
commit_message=f"Update signal log ({len(df)} total signals)",
)
return True
except Exception as e:
st.warning(f"Could not save signal log: {e}")
return False
def append_hits_to_log(hits_df: pd.DataFrame, settings: dict) -> bool:
if hits_df.empty or not _log_configured():
return False
existing = load_signal_log()
new_rows = hits_df.copy()
new_rows["FlagDate"] = pd.Timestamp.now().normalize()
new_rows["RSI_Threshold"] = settings["rsi_threshold"]
new_rows["Stretch_Threshold"] = settings["stretch_threshold"]
new_rows["Strict_Window"] = settings["strict_window"]
new_rows["Lookback"] = settings["lookback"]
new_rows["WinRate_Min"] = settings["win_rate_min"]
combined = pd.concat([existing, new_rows], ignore_index=True) if not existing.empty else new_rows
combined = combined.drop_duplicates(subset=["Ticker", "FlagDate"], keep="last")
return save_signal_log(combined)
# =========================================================
# TAB 1: WHALE MONITOR
# =========================================================
SIZE_COLORS = {
"WHALE": "#8e44ad", # purple
"INSTITUTIONAL": "#e67e22", # orange
"LARGE": "#2980b9", # blue
"MEDIUM": "#27ae60", # green
"RETAIL": "#7f8c8d", # grey
}
@st.cache_data(ttl=120, show_spinner="Fetching data...")
def fetch_whale_data(sym: str) -> pd.DataFrame:
try:
ticker = yf.Ticker(sym)
# 1-minute granularity for accurate size-tier classification;
# Yahoo only serves 1m bars for the trailing 7 days.
return ticker.history(period="7d", interval="1m")
except Exception:
return pd.DataFrame()
def classify_size(vol, median):
ratio = vol / median if median > 0 else 0
if ratio >= 10:
return "WHALE"
elif ratio >= 5:
return "INSTITUTIONAL"
elif ratio >= 2:
return "LARGE"
elif ratio >= 1:
return "MEDIUM"
return "RETAIL"
def get_sentiment(row):
if row["Price_Change"] > 0:
return row["Volume"], "BUY"
elif row["Price_Change"] < 0:
return -row["Volume"], "SELL"
return 0, "NEUTRAL"
def render_whale_monitor():
with st.sidebar:
st.header("Whale Monitor")
wm_mode = st.radio("Mode", ["Single Ticker", "Premarket Scan"], key="wm_mode")
st.markdown("---")
if wm_mode == "Single Ticker":
render_whale_single_ticker()
else:
render_premarket_scan()
@st.cache_data(ttl=24 * 3600, show_spinner="Loading curated ticker list...")
def get_clean_premarket_universe():
try:
local_path = hf_hub_download(
repo_id="Trebor1009/clean-premarket-data", repo_type="dataset",
filename="clean_premarket_data.csv", token=HF_TOKEN,
)
df = pd.read_csv(local_path, usecols=["Ticker"])
return sorted(df["Ticker"].dropna().unique().tolist())
except Exception as e:
st.warning(f"Could not load curated ticker list from Trebor1009/clean-premarket-data: {e}")
return []
@st.cache_data(ttl=3600, show_spinner="Fetching prior-close reference prices...")
def get_daily_reference_data(tickers_tuple):
ref = {}
batch_size = 10
total_batches = max(1, (len(tickers_tuple) + batch_size - 1) // batch_size)
progress = st.progress(0, text="Fetching reference prices...")
today_date = pd.Timestamp.now(tz="America/New_York").normalize().tz_localize(None)
for b, i in enumerate(range(0, len(tickers_tuple), batch_size)):
batch = tickers_tuple[i:i + batch_size]
try:
data = yf.download(list(batch), period="1mo", interval="1d", group_by="ticker",
progress=False, threads=True)
except Exception:
progress.progress((b + 1) / total_batches)
continue
for ticker in batch:
try:
try:
raw = data[ticker]
except Exception:
raw = data
closes = raw["Close"].dropna()
vol = raw["Volume"].dropna()
if closes.empty:
continue
idx = closes.index.tz_localize(None) if closes.index.tz is not None else closes.index
prior = closes[idx.normalize() < today_date]
if not prior.empty:
prev_close = float(prior.iloc[-1])
elif len(closes) >= 2:
prev_close = float(closes.iloc[-2])
else:
prev_close = float(closes.iloc[-1])
avg_vol = float(vol.mean()) if len(vol) >= 5 else None
ref[ticker] = {"prev_close": prev_close, "avg_vol": avg_vol}
except Exception:
continue
progress.progress((b + 1) / total_batches, text=f"Reference data: batch {b + 1}/{total_batches}...")
progress.empty()
return ref
@st.cache_data(ttl=300, show_spinner=False)
def run_premarket_scan(tickers_tuple, min_gap_pct, direction):
ref_data = get_daily_reference_data(tickers_tuple)
results = []
batch_size = 10
total_batches = max(1, (len(tickers_tuple) + batch_size - 1) // batch_size)
progress = st.progress(0, text="Scanning premarket activity...")
for b, i in enumerate(range(0, len(tickers_tuple), batch_size)):
batch = tickers_tuple[i:i + batch_size]
try:
data = yf.download(list(batch), period="1d", interval="1m", prepost=True,
group_by="ticker", progress=False, threads=True)
except Exception:
progress.progress((b + 1) / total_batches, text=f"Batch {b + 1}/{total_batches} failed, continuing...")
continue
for ticker in batch:
try:
try:
raw = data[ticker]
except Exception:
raw = data
df = raw.dropna(subset=["Close"])
if df.empty:
continue
if df.index.tz is not None:
df = df.copy()
df.index = df.index.tz_convert("America/New_York")
else:
df = df.copy()
df.index = df.index.tz_localize("America/New_York")
premarket = df[df.index.time < dt_time(9, 30)]
if premarket.empty:
continue
ref = ref_data.get(ticker)
if not ref or not ref.get("prev_close"):
continue
prev_close = ref["prev_close"]
last_price = float(premarket["Close"].iloc[-1])
if not prev_close:
continue
gap_pct = (last_price - prev_close) / prev_close * 100
if abs(gap_pct) < min_gap_pct:
continue
if direction == "Up only" and gap_pct <= 0:
continue
if direction == "Down only" and gap_pct >= 0:
continue
pm_volume = int(premarket["Volume"].sum())
results.append({
"Ticker": ticker,
"Prev Close": round(prev_close, 2),
"Premarket Price": round(last_price, 2),
"Gap %": round(gap_pct, 2),
"Premarket Volume": pm_volume,
"Avg Daily Vol": int(ref["avg_vol"]) if ref.get("avg_vol") else None,
})
except Exception:
continue
progress.progress((b + 1) / total_batches, text=f"Scanned batch {b + 1}/{total_batches}...")
progress.empty()
return pd.DataFrame(results)
def render_whale_single_ticker():
st.title("π Whale Monitor β Single Ticker")
st.caption(
"Flags unusual 1-minute volume on the public exchange tape over the trailing 7 days, "
"compared against each day's own volume baseline. "
"This is a lit-market 'unusual volume' heuristic, not off-exchange dark pool print data."
)
with st.sidebar:
st.subheader("Single Ticker Settings")
symbol = st.text_input("Ticker", value="AIIO", key="wm_ticker").strip().upper()
multiplier = st.slider("Sensitivity (x daily median volume)", 1.0, 10.0, 2.0, 0.5, key="wm_mult")
whale_threshold = st.number_input("Whale alert threshold (volume)", value=50000, step=5000, key="wm_thresh")
show_all = st.checkbox("Show all candles (ignore threshold)", value=False, key="wm_all")
exclude_open = st.checkbox("Exclude opening auction print (each day's first bar)", value=True, key="wm_exopen")
if st.button("π Refresh data", key="wm_refresh"):
fetch_whale_data.clear()
if not symbol:
st.info("Enter a ticker symbol in the sidebar to begin.")
return
df = fetch_whale_data(symbol)
if df.empty:
st.error(f"No data available for '{symbol}'. Check the ticker symbol and try again.")
return
df = df.copy()
df["_day"] = df.index.normalize()
first_day = df["_day"].min()
last_day = df["_day"].max()
if exclude_open:
bar_position_in_day = df.groupby("_day").cumcount()
df = df[bar_position_in_day > 0]
if df.empty:
st.error("No data for the selected 7-day window.")
return
df = df[df["Volume"] >= 0]
if (df["Volume"] > 0).sum() == 0:
st.error("Not enough volume data after filtering.")
return
vol_positive = df["Volume"] > 0
daily_median = df[vol_positive].groupby(df["_day"])["Volume"].median()
daily_mean = df[vol_positive].groupby(df["_day"])["Volume"].mean()
df["Daily_Median_Vol"] = df["_day"].map(daily_median)
df["Daily_Mean_Vol"] = df["_day"].map(daily_mean)
df["Price_Change"] = df["Close"].diff()
df["Pct_Change"] = df["Close"].pct_change() * 100
df["Vol_Ratio"] = df["Volume"] / df["Daily_Median_Vol"]
df["is_notable"] = df["Volume"] > (df["Daily_Median_Vol"] * multiplier) if not show_all else True
df[["Delta", "Sentiment"]] = df.apply(lambda r: pd.Series(get_sentiment(r)), axis=1)
df["Cum_Delta"] = df["Delta"].cumsum()
df["Size"] = df.apply(lambda r: classify_size(r["Volume"], r["Daily_Median_Vol"]), axis=1)
median_vol = daily_median.mean()
mean_vol = daily_mean.mean()
st.subheader(f"{symbol} β Last 7 Days: {first_day.date()} to {last_day.date()}")
c1, c2, c3, c4 = st.columns(4)
c1.metric("Avg of Daily Median Vol", f"{median_vol:,.0f}")
c2.metric("Avg of Daily Mean Vol", f"{mean_vol:,.0f}")
c3.metric("Flag Threshold (avg)", f"{median_vol * multiplier:,.0f}")
c4.metric(
"Last Price",
f"${df['Close'].iloc[-1]:.2f}",
f"{df['Pct_Change'].iloc[-1]:+.2f}%" if pd.notna(df["Pct_Change"].iloc[-1]) else None,
)
st.caption(
"Size-tier classification (WHALE/INSTITUTIONAL/etc.) and the Notable Activity table below use "
"each bar's **own day's** median volume, not the pooled 7-day figure shown above."
)
st.markdown("### Notable Activity")
notable = df[df["is_notable"]].copy()
if notable.empty:
st.info("No candles crossed the sensitivity threshold for this window.")
else:
notable["Whale Alert"] = notable["Volume"] >= whale_threshold
table = notable[["Volume", "Vol_Ratio", "Size", "Sentiment", "Close", "Pct_Change", "Whale Alert"]]
table.index.name = "Datetime"
st.dataframe(
table.style.format(
{"Volume": "{:,.0f}", "Vol_Ratio": "{:.1f}x", "Close": "${:.2f}", "Pct_Change": "{:+.2f}%"}
),
use_container_width=True, height=400,
)
st.markdown("### Cumulative Buy/Sell Delta")
delta_by_size = pd.DataFrame(index=df.index)
for size in ["WHALE", "INSTITUTIONAL", "LARGE", "MEDIUM", "RETAIL"]:
masked_delta = df["Delta"].where(df["Size"] == size, 0)
delta_by_size[size] = masked_delta.cumsum()
fig = go.Figure()
for size, color in SIZE_COLORS.items():
fig.add_trace(go.Scatter(
x=delta_by_size.index,
y=delta_by_size[size],
mode="lines",
name=size.title(),
line=dict(color=color, width=2),
))
fig.add_trace(go.Scatter(
x=df.index,
y=df["Cum_Delta"],
mode="lines",
name="Total (all tiers)",
line=dict(color="#2c3e50", width=1.5, dash="dot"),
))
fig.update_layout(
xaxis_title="Time",
yaxis_title="Cumulative Delta (shares)",
legend_title="Size Tier",
hovermode="x unified",
height=450,
margin=dict(l=10, r=10, t=10, b=10),
)
st.plotly_chart(fig, use_container_width=True, key=f"wm_cumdelta_{symbol}")
total_buy = df[df["Delta"] > 0]["Delta"].sum()
total_sell = abs(df[df["Delta"] < 0]["Delta"].sum())
net = total_buy - total_sell
whale_buys = df[(df["Size"] == "WHALE") & (df["Sentiment"] == "BUY")]["Volume"].sum()
whale_sells = df[(df["Size"] == "WHALE") & (df["Sentiment"] == "SELL")]["Volume"].sum()
inst_buys = df[(df["Size"].isin(["INSTITUTIONAL", "WHALE"])) & (df["Sentiment"] == "BUY")]["Volume"].sum()
inst_sells = df[(df["Size"].isin(["INSTITUTIONAL", "WHALE"])) & (df["Sentiment"] == "SELL")]["Volume"].sum()
st.markdown("### Session Analysis")
colA, colB, colC = st.columns(3)
with colA:
st.markdown("**Overall**")
st.write(f"Buy volume: {total_buy:,.0f}")
st.write(f"Sell volume: {total_sell:,.0f}")
st.write(f"Net delta: {net:,.0f}")
st.write(f"Buy/Sell ratio: {(total_buy / total_sell) if total_sell > 0 else 0:.2f}x")
with colB:
st.markdown("**Whale tier**")
st.write(f"Buy: {whale_buys:,.0f}")
st.write(f"Sell: {whale_sells:,.0f}")
st.write(f"Net: {whale_buys - whale_sells:,.0f}")
with colC:
st.markdown("**Institutional+**")
st.write(f"Buy: {inst_buys:,.0f}")
st.write(f"Sell: {inst_sells:,.0f}")
st.write(f"Net: {inst_buys - inst_sells:,.0f}")
status = "π’ ACCUMULATION (Bullish)" if net > 0 else "π΄ DISTRIBUTION (Bearish)"
st.markdown(f"## Market Sentiment: {status}")
st.markdown("### Volume by Size Category")
rows = []
total_vol = df["Volume"].sum()
for size in ["WHALE", "INSTITUTIONAL", "LARGE", "MEDIUM", "RETAIL"]:
vol = df[df["Size"] == size]["Volume"].sum()
count = len(df[df["Size"] == size])
pct = (vol / total_vol * 100) if total_vol > 0 else 0
rows.append({"Size": size, "Volume": vol, "% of Total": round(pct, 1), "Candles": count})
st.dataframe(pd.DataFrame(rows), use_container_width=True, hide_index=True)
st.caption(
"Data source: yfinance (Yahoo Finance), unofficial and delayed. 1-minute bars, 7-day window, "
"classified against each day's own volume baseline. "
"Cached for 2 minutes β use the Refresh button in the sidebar for the latest data."
)
st.markdown("### Notable Activity")
notable = df[df["is_notable"]].copy()
if notable.empty:
st.info("No candles crossed the sensitivity threshold for this window.")
else:
notable["Whale Alert"] = notable["Volume"] >= whale_threshold
table = notable[["Volume", "Vol_Ratio", "Size", "Sentiment", "Close", "Pct_Change", "Whale Alert"]]
table.index.name = "Datetime"
st.dataframe(
table.style.format(
{"Volume": "{:,.0f}", "Vol_Ratio": "{:.1f}x", "Close": "${:.2f}", "Pct_Change": "{:+.2f}%"}
),
use_container_width=True, height=400,
)
# ---------------------------------------------------------------
# Cumulative Buy/Sell Delta β per size-tier colored lines
# ---------------------------------------------------------------
st.markdown("### Cumulative Buy/Sell Delta")
delta_by_size = pd.DataFrame(index=df.index)
for size in ["WHALE", "INSTITUTIONAL", "LARGE", "MEDIUM", "RETAIL"]:
masked_delta = df["Delta"].where(df["Size"] == size, 0)
delta_by_size[size] = masked_delta.cumsum()
fig = go.Figure()
for size, color in SIZE_COLORS.items():
fig.add_trace(go.Scatter(
x=delta_by_size.index,
y=delta_by_size[size],
mode="lines",
name=size.title(),
line=dict(color=color, width=2),
))
# thin dotted total line for reference
fig.add_trace(go.Scatter(
x=df.index,
y=df["Cum_Delta"],
mode="lines",
name="Total (all tiers)",
line=dict(color="#2c3e50", width=1.5, dash="dot"),
))
fig.update_layout(
xaxis_title="Time",
yaxis_title="Cumulative Delta (shares)",
legend_title="Size Tier",
hovermode="x unified",
height=450,
margin=dict(l=10, r=10, t=10, b=10),
)
st.plotly_chart(fig, use_container_width=True)
total_buy = df[df["Delta"] > 0]["Delta"].sum()
total_sell = abs(df[df["Delta"] < 0]["Delta"].sum())
net = total_buy - total_sell
whale_buys = df[(df["Size"] == "WHALE") & (df["Sentiment"] == "BUY")]["Volume"].sum()
whale_sells = df[(df["Size"] == "WHALE") & (df["Sentiment"] == "SELL")]["Volume"].sum()
inst_buys = df[(df["Size"].isin(["INSTITUTIONAL", "WHALE"])) & (df["Sentiment"] == "BUY")]["Volume"].sum()
inst_sells = df[(df["Size"].isin(["INSTITUTIONAL", "WHALE"])) & (df["Sentiment"] == "SELL")]["Volume"].sum()
st.markdown("### Session Analysis")
colA, colB, colC = st.columns(3)
with colA:
st.markdown("**Overall**")
st.write(f"Buy volume: {total_buy:,.0f}")
st.write(f"Sell volume: {total_sell:,.0f}")
st.write(f"Net delta: {net:,.0f}")
st.write(f"Buy/Sell ratio: {(total_buy / total_sell) if total_sell > 0 else 0:.2f}x")
with colB:
st.markdown("**Whale tier**")
st.write(f"Buy: {whale_buys:,.0f}")
st.write(f"Sell: {whale_sells:,.0f}")
st.write(f"Net: {whale_buys - whale_sells:,.0f}")
with colC:
st.markdown("**Institutional+**")
st.write(f"Buy: {inst_buys:,.0f}")
st.write(f"Sell: {inst_sells:,.0f}")
st.write(f"Net: {inst_buys - inst_sells:,.0f}")
status = "π’ ACCUMULATION (Bullish)" if net > 0 else "π΄ DISTRIBUTION (Bearish)"
st.markdown(f"## Market Sentiment: {status}")
st.markdown("### Volume by Size Category")
rows = []
total_vol = df["Volume"].sum()
for size in ["WHALE", "INSTITUTIONAL", "LARGE", "MEDIUM", "RETAIL"]:
vol = df[df["Size"] == size]["Volume"].sum()
count = len(df[df["Size"] == size])
pct = (vol / total_vol * 100) if total_vol > 0 else 0
rows.append({"Size": size, "Volume": vol, "% of Total": round(pct, 1), "Candles": count})
st.dataframe(pd.DataFrame(rows), use_container_width=True, hide_index=True)
st.caption(
"Data source: yfinance (Yahoo Finance), unofficial and delayed. 1-minute bars, 7-day window, "
"classified against each day's own volume baseline. "
"Cached for 2 minutes β use the Refresh button in the sidebar for the latest data."
)
# =========================================================
# TAB 2: PEAK EXHAUSTION DETECTOR
# =========================================================
DEFAULT_PEAK_TICKERS = [
"SOL-USD", "SLMT", "BTC-USD", "GC=F", "GPUS", "LLY", "MSFT", "NVDA",
"AIIO", "^IXIC", "IAUP.L", "PLTR", "MSTR", "CVX", "BZ=F", "CRML", "ABTC", "^FTSE",
]
def get_clean_peak_data(ticker: str) -> pd.DataFrame:
df = yf.download(ticker, start="2018-01-01", progress=False)
if df.empty:
return df
if isinstance(df.columns, pd.MultiIndex):
df.columns = df.columns.get_level_values(0)
return df.reset_index()
def label_peaks(df: pd.DataFrame, target_drop: float, window: int) -> pd.DataFrame:
d = df.copy()
d["Peak"] = 0
w = 10
for i in range(w, len(d) - w):
if d["High"].iloc[i] == d["High"].iloc[i - w : i + w].max():
if d["Low"].iloc[i + 1 : i + 15].min() <= (1 - target_drop) * d["High"].iloc[i]:
d.at[i, "Peak"] = 1
d["Target"] = d["Peak"].rolling(window=window).max().shift(-window).fillna(0)
return d
def add_peak_features(df: pd.DataFrame, vol_sensitivity: float) -> pd.DataFrame:
d = df.copy()
delta = d["Close"].diff()
gain = (delta.where(delta > 0, 0)).rolling(14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(14).mean()
d["RSI"] = 100 - (100 / (1 + (gain / (loss + 1e-9))))
d["Div_Warning"] = ((d["High"] >= d["High"].shift(1)) & (d["RSI"] < d["RSI"].shift(1))).astype(int)
d["Vol_Rel"] = d["Volume"] / d["Volume"].rolling(20).mean()
d["Vol_Price_Inertia"] = (d["Vol_Rel"] * vol_sensitivity) / (abs(d["Close"].pct_change()) + 0.001)
d["EMA_20"] = d["Close"].ewm(span=20).mean()
d["Dist_EMA"] = (d["Close"] - d["EMA_20"]) / d["EMA_20"]
d["RSI_Blowoff_Band"] = d["Close"] * (1 + (80 - d["RSI"]) / 100)
return d.dropna()
@st.cache_data(ttl=6 * 3600, show_spinner="Training peak detection model...")
def run_peak_model(ticker: str, target_drop: float, window: int, vol_sensitivity: float):
df = get_clean_peak_data(ticker)
if df.empty:
return None
df = label_peaks(df, target_drop, window)
data = add_peak_features(df, vol_sensitivity)
if data.empty:
return None
features = ["RSI", "Dist_EMA", "Div_Warning", "Vol_Price_Inertia"]
X, y = data[features], data["Target"]
model = GradientBoostingClassifier(n_estimators=100, random_state=42)
model.fit(X, y)
data["Risk_Pct"] = model.predict_proba(X)[:, 1] * 100
return data
def render_peak_detector():
st.title("π Peak Exhaustion Detector")
st.caption(
"Scores potential local price peaks using RSI divergence, distance from EMA20, "
"and a volume/price-inertia signal, via a Gradient Boosting model."
)
st.warning(
"β οΈ This model is fit and scored on the same historical data β there's no held-out "
"test period. The Risk % shows how closely current conditions match past peaks in "
"its own training data, not a validated out-of-sample forecast. Treat it as structured "
"pattern-matching against history, not a calibrated probability."
)
with st.sidebar:
st.markdown("---")
st.header("Peak Detector Settings")
pd_ticker = st.selectbox("Ticker", DEFAULT_PEAK_TICKERS, index=0, key="pd_ticker")
pd_custom = st.text_input("...or enter a custom ticker", value="", key="pd_custom")
target_drop = st.slider("Drop % that counts as a peak", 0.03, 0.15, 0.06, 0.01, key="pd_drop")
window = st.slider("Forward warning window (days)", 3, 15, 5, 1, key="pd_window")
vol_sensitivity = st.slider("Volume sensitivity", 0.5, 3.0, 1.0, 0.1, key="pd_vol")
if st.button("π Retrain model", key="pd_refresh"):
run_peak_model.clear()
ticker = pd_custom.strip().upper() if pd_custom.strip() else pd_ticker
data = run_peak_model(ticker, target_drop, window, vol_sensitivity)
if data is None or data.empty:
st.error(f"No usable data for '{ticker}'.")
return
latest = data.iloc[-1]
status = "π΄ CRITICAL" if latest["Risk_Pct"] > 50 else "π’ STABLE"
st.subheader(f"{ticker}")
c1, c2, c3 = st.columns(3)
c1.metric("Status", status)
c2.metric("Risk %", f"{latest['Risk_Pct']:.1f}%")
c3.metric("RSI", f"{latest['RSI']:.1f}")
plot_df = data.iloc[-240:].copy()
fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(12, 8), sharex=True,
gridspec_kw={"height_ratios": [2.5, 1]})
ax1.plot(plot_df["Date"], plot_df["Close"], color="black", linewidth=1)
ax1.fill_between(plot_df["Date"], plot_df["Close"], plot_df["RSI_Blowoff_Band"], color="red", alpha=0.1)
risk_days = plot_df[plot_df["Risk_Pct"] > 50]
ax1.scatter(risk_days["Date"], risk_days["High"] * 1.02, color="red", marker="D", s=50)
for _, row in risk_days.iterrows():
ax1.text(row["Date"], row["High"] * 1.03, "!", color="red", fontweight="bold", ha="center")
ax1.set_title(f"{ticker} | Price vs Risk Heatmap")
ax2.fill_between(plot_df["Date"], 0, plot_df["Risk_Pct"], color="purple", alpha=0.4)
ax2.axhline(50, color="red", linestyle="--", alpha=0.3)
ax2.set_ylabel("Drop Risk %")
plt.tight_layout()
st.pyplot(fig)
plt.close(fig)
st.caption(
"Data source: yfinance (Yahoo Finance), daily bars since 2018. "
"Model retrains on parameter changes and is cached for 6 hours."
)
# =========================================================
# TAB 3: SMALL-CAP PUMP SCANNER v3 β Serial Pumper Framing
# =========================================================
#
# Rebuilt from the 2-year backtest report (Pump Scanner v3):
# "The stock matters more than the setup" β prior pump history +
# recency account for ~88% of the model's predictive power.
# Technical patterns (coil/surge) are a secondary gate, not the
# primary signal.
#
# Architecture:
# STEP 1 Serial Pumper Registry β tag tickers with β₯5 historical
# pumps as "known pumpers"
# STEP 2 Recency Check β last pump β€10 days? β "active pumper"
# STEP 3 Technical Gate (secondary) β T1A/T2A BB-width/price tier,
# coil or surge pattern
# STEP 4 Composite score β empirical pump-rate lookup from the
# Prior-Pumps Γ Days-Since-Pump crosstab, boosted by the
# technical gate when it fires
# STEP 5 Alert + log to Signal Tracker
#
# NOTE: the report's gradient-boosting model (AUC 0.979) was trained
# offline on the full 2-year/730-ticker backtest. It isn't shipped as
# a pickle here, so this scanner reproduces its decision logic directly
# from the empirical tables in the report (the crosstab below IS what
# the model mostly learned β Days-Since-Pump alone carried 88.5% of
# feature importance). If you export the trained model later, swap
# `lookup_pump_rate()` / `classify_composite()` for a `model.predict_proba()` call.
# Thresholds (calibrated from 2-year backtest, 730 tickers)
MAX_PRICE = 5.0
MIN_PRICE = 0.05
MIN_BARS = 30
MIN_VOL_DAYS = 15
COIL_VOL_THRESH = 0.4 # below = quiet day
COIL_MIN_DAYS = 5 # backtest: 5d quiet optimal
COIL_BREAK_MIN = 0.8
RSI_COIL_MAX = 47
BB_WIDTH_MAX = 80.0 # hard exclusion above this
BB_WIDTH_T1_MIN = 35.0 # Tier 1 needs BB β₯ 35%
SURGE_VOL_MIN = 3.0
SURGE_VOL_GRML = 5.0 # GRML-lesson threshold: surge re-enabled at vol>5x
SURGE_RANGE_POS = 0.35
RSI_SURGE_MAX = 52
RANGE_POS_MAX = 0.35
PRICE_T1_MAX = 0.20 # Tier 1 price ceiling
PRICE_T2_MAX = 0.50 # Tier 2 price ceiling
# Serial pumper registry
PUMP_MOVE_THRESH = 0.50 # +50% max-High within horizon = "a pump"
PUMP_HORIZON = 5 # trading days
PUMP_COOLDOWN = 10 # days to skip after a flagged pump (no double counting a continuation)
REGISTRY_PERIOD = "2y" # history length used to build the registry
# Months with above-baseline pump rates (directional, not statistically confirmed)
GOOD_MONTHS = {3, 4, 5, 8, 9} # Mar Apr May Aug Sep
WEAK_MONTHS = {6, 7} # Jun Jul
BASELINE_PUMP_RATE = 1.8 # random baseline, max-High β₯50% within 5 days
# ββ Tier classification (technical gate β secondary) βββββββββββββββββββββββ
def classify_pump_tier(price: float, bb_width: float) -> dict:
"""
Returns tier dict based on Price Γ BB Width cross-tab from the backtest.
BB Width β₯ 80% is excluded entirely (falling knife).
"""
if bb_width >= BB_WIDTH_MAX:
return {"tier": 0, "label": "Excluded (BBβ₯80%)",
"pump_rate": 0.0, "exclude": True}
if price < PRICE_T1_MAX:
if BB_WIDTH_T1_MIN <= bb_width < BB_WIDTH_MAX:
return {"tier": 1, "label": "Tier 1A β 19.7% pump rate",
"pump_rate": 19.7, "exclude": False}
else:
return {"tier": 2, "label": "Tier 2B β 10.1% pump rate",
"pump_rate": 10.1, "exclude": False}
elif price < PRICE_T2_MAX:
if BB_WIDTH_T1_MIN <= bb_width < BB_WIDTH_MAX:
return {"tier": 2, "label": "Tier 2A β ~13% pump rate",
"pump_rate": 13.0, "exclude": False}
else:
return {"tier": 3, "label": "Tier 3 β 7.0% pump rate",
"pump_rate": 7.0, "exclude": False}
else: # price >= 0.50
if BB_WIDTH_T1_MIN <= bb_width < BB_WIDTH_MAX:
return {"tier": 3, "label": "Tier 3 β 13% (small n, caution)",
"pump_rate": 13.0, "exclude": False}
else:
return {"tier": 0, "label": "Dead zone β 0.6% pump rate",
"pump_rate": 0.6, "exclude": True}
TIER_ICONS = {1: "π₯", 2: "β
", 3: "β οΈ", 0: "β"}
# ββ STEP 1+2: Serial Pumper Registry ββββββββββββββββββββββββββββββββββββββββ
def detect_pump_events(df: pd.DataFrame, threshold: float = PUMP_MOVE_THRESH,
horizon: int = PUMP_HORIZON, cooldown: int = PUMP_COOLDOWN) -> list:
"""
Backward-looking pump detector, usable live (no future data needed):
day t is flagged as a pump event if the day's High is β₯ `threshold`
above the lowest Close in the preceding `horizon` trading days.
A cooldown window prevents one big move being counted as several pumps.
"""
if df is None or len(df) < horizon + 2:
return []
if isinstance(df.columns, pd.MultiIndex):
df = df.copy()
df.columns = df.columns.get_level_values(0)
high = df['High'].astype(float).values
close = df['Close'].astype(float).values
n = len(df)
events = []
last_flag = -cooldown - 1
for t in range(horizon, n):
base = close[t - horizon:t].min()
if base <= 0:
continue
if high[t] >= base * (1 + threshold) and (t - last_flag) >= cooldown:
events.append(df.index[t])
last_flag = t
return events
def serial_pumper_stats(df: pd.DataFrame) -> dict:
"""Prior pump count + recency, as-of the last bar in df."""
events = detect_pump_events(df)
prior_pumps = len(events)
if events:
last_event = events[-1]
today = df.index[-1]
days_since = int((today - last_event).days)
else:
days_since = None
return {"prior_pumps": prior_pumps, "days_since_last": days_since, "events": events}
def bucket_prior_pumps(n: int) -> str:
if n == 0: return "0"
if n <= 2: return "1-2"
if n <= 4: return "3-4"
return "5+"
def bucket_days_since(d) -> str:
if d is None: return ">90d"
if d < 3: return "<3d"
if d < 10: return "3-10d"
if d < 30: return "10-30d"
if d < 90: return "30-90d"
return ">90d"
# Empirical crosstab: Pump5d rate% by Prior-Pump-Count Γ Days-Since-Last-Pump
# (this is the table behind "88% of model power in 2 features")
PUMP_RATE_TABLE = {
"0": {"<3d": 0.0, "3-10d": 0.0, "10-30d": 0.0, "30-90d": 0.0, ">90d": 0.0},
"1-2": {"<3d": 11.9, "3-10d": 3.8, "10-30d": 0.0, "30-90d": 0.0, ">90d": 0.0},
"3-4": {"<3d": 9.8, "3-10d": 7.3, "10-30d": 0.0, "30-90d": 0.3, ">90d": 0.1},
"5+": {"<3d": 37.2, "3-10d": 10.1, "10-30d": 0.5, "30-90d": 0.2, ">90d": 0.1},
}
def lookup_pump_rate(prior_pumps: int, days_since) -> float:
return PUMP_RATE_TABLE[bucket_prior_pumps(prior_pumps)][bucket_days_since(days_since)]
# ββ Feature engineering (technical gate inputs) βββββββββββββββββββββββββββββ
def compute_features(df: pd.DataFrame) -> pd.DataFrame | None:
if df is None or len(df) < MIN_BARS:
return None
if isinstance(df.columns, pd.MultiIndex):
df.columns = df.columns.get_level_values(0)
df = df[['Open', 'High', 'Low', 'Close', 'Volume']].copy().dropna()
if len(df) < MIN_BARS:
return None
close = df['Close'].astype(float)
high = df['High'].astype(float)
low = df['Low'].astype(float)
vol = df['Volume'].astype(float)
if (vol.iloc[-20:] > 0).sum() < MIN_VOL_DAYS:
return None
# RSI
delta = close.diff()
gain = delta.where(delta > 0, 0).ewm(alpha=1/14, adjust=False).mean()
loss = (-delta.where(delta < 0, 0)).ewm(alpha=1/14, adjust=False).mean()
df['RSI'] = 100 - 100 / (1 + gain / (loss + 1e-9))
# Volume
df['Vol_MA20'] = vol.rolling(20).mean()
df['Vol_Ratio'] = vol / (df['Vol_MA20'] + 1e-9)
df['Vol_MA5'] = vol.rolling(5).mean()
# Quiet streak
quiet = (df['Vol_Ratio'] < COIL_VOL_THRESH).astype(int)
streak = quiet * (quiet.groupby((quiet != quiet.shift()).cumsum()).cumcount() + 1)
df['Quiet_Streak'] = streak.where(quiet == 1, 0)
# Bollinger Bands
sma20 = close.rolling(20).mean()
std20 = close.rolling(20).std()
df['SMA20'] = sma20
df['BB_Width'] = (std20 * 4) / (sma20 + 1e-9) * 100 # 2-sigma bandwidth normalised
df['Dist_SMA'] = (close - sma20) / (sma20 + 1e-9) * 100
# 20-day range position
low20 = low.rolling(20).min()
high20 = high.rolling(20).max()
df['Range_Pos'] = (close - low20) / (high20 - low20 + 1e-9)
df['Day_Range_Pos'] = (close - low) / (high - low + 1e-9)
return df.dropna()
def check_signals(df: pd.DataFrame) -> dict:
if df is None or len(df) < 2:
return {"pattern_a": False, "pattern_b": False}
latest = df.iloc[-1]
prev = df.iloc[-2]
rsi = float(latest['RSI'])
vol_ratio = float(latest['Vol_Ratio'])
quiet_streak = float(prev['Quiet_Streak']) # yesterday's streak
range_pos = float(latest['Range_Pos'])
day_range = float(latest['Day_Range_Pos'])
bb_width = float(latest['BB_Width'])
close = float(latest['Close'])
# Hard exclusion
if bb_width >= BB_WIDTH_MAX:
return {"pattern_a": False, "pattern_b": False,
"rsi": round(rsi,1), "vol_ratio": round(vol_ratio,2),
"quiet_streak": int(quiet_streak), "range_pos": round(range_pos,3),
"day_range_pos": round(day_range,3), "bb_width": round(bb_width,1),
"dist_sma": round(float(latest['Dist_SMA']),1),
"close": round(close,4), "score": 0,
"vol_ma20": round(float(latest['Vol_MA20']),0),
"vol_today": round(float(latest['Volume']),0)}
# Pattern A: Coil break (5+ quiet days, first vol return, RSI<47, Range<0.35, BB<80%)
pattern_a = (
quiet_streak >= COIL_MIN_DAYS and
vol_ratio >= COIL_BREAK_MIN and
rsi < RSI_COIL_MAX and
range_pos < RANGE_POS_MAX and
bb_width < BB_WIDTH_MAX
)
# Pattern B: surge without a coil. GRML lesson from v3 backtest β re-enabled
# as a standalone signal (previously required pattern_a alongside it) but
# only above a higher vol threshold, and its true edge depends on whether
# the ticker is a registered serial pumper (10.6% generic vs 25%+ if serial).
pattern_b = (
vol_ratio >= SURGE_VOL_MIN and
day_range >= SURGE_RANGE_POS and
rsi < RSI_SURGE_MAX and
range_pos < RANGE_POS_MAX and
bb_width < BB_WIDTH_MAX
)
# Score: BB Width in 35-80% is rewarded (not penalised)
score = 0
score += max(0, (47 - rsi)) * 1.0
score += max(0, (RANGE_POS_MAX - range_pos)) * 60
bb_score = max(0, min(bb_width, BB_WIDTH_MAX) - BB_WIDTH_T1_MIN) / (BB_WIDTH_MAX - BB_WIDTH_T1_MIN) * 20
score += bb_score
if pattern_a:
score += min(25, quiet_streak * 3)
if pattern_b:
score += min(20, (vol_ratio - SURGE_VOL_MIN) * 4)
return {
"pattern_a": pattern_a,
"pattern_b": pattern_b,
"rsi": round(rsi, 1),
"vol_ratio": round(vol_ratio, 2),
"quiet_streak": int(quiet_streak),
"range_pos": round(range_pos, 3),
"day_range_pos":round(day_range, 3),
"bb_width": round(bb_width, 1),
"dist_sma": round(float(latest['Dist_SMA']), 1),
"close": round(close, 4),
"score": round(min(100, score), 1),
"vol_ma20": round(float(latest['Vol_MA20']), 0),
"vol_today": round(float(latest['Volume']), 0),
}
# ββ STEP 3+4: Technical gate + composite score ββββββββββββββββββββββββββββββ
COMPOSITE_ICONS = {
1: "π₯π₯π₯", 2: "π₯π₯", 3: "π₯", 4: "β‘β‘", 5: "β‘", 6: "π", 7: "β’",
}
def classify_composite(prior_pumps: int, days_since, tier_info: dict, sig: dict) -> dict:
"""
Reproduces the v3 decision cascade / "best combo" ranking:
1. Serial + Tier + Signal 25.1% (14x baseline)
2. Serial + Tier 22.4%
3. Serial pumper, active β€10d 18.7%
4. Pattern B surge >5x, Tier 14.2% (GRML re-enable)
5. Pattern B surge >3x, Tier 10.6%
6. Pattern A coil, Tier 8.3%
7. Registry-only / below combos (empirical lookup, floor = baseline)
"""
is_serial = prior_pumps >= 5
is_active = (days_since is not None) and (days_since <= 10)
tier_ok = (not tier_info["exclude"]) and (tier_info["tier"] in (1, 2))
pattern_a = sig.get("pattern_a", False)
pattern_b = sig.get("pattern_b", False)
vol_ratio = sig.get("vol_ratio", 0) or 0
if is_serial and is_active and tier_ok and (pattern_a or pattern_b):
p, rate, label = 1, 25.1, "Serial + Tier + Signal"
elif is_serial and is_active and tier_ok:
p, rate, label = 2, 22.4, "Serial + Tier"
elif is_serial and is_active:
p, rate, label = 3, 18.7, "Serial pumper, active β€10d"
elif pattern_b and tier_ok and vol_ratio > SURGE_VOL_GRML:
p, rate, label = 4, 14.2, "Pattern B surge >5x (GRML)"
elif pattern_b and tier_ok:
p, rate, label = 5, 10.6, "Pattern B surge >3x"
elif pattern_a and tier_ok:
p, rate, label = 6, 8.3, "Pattern A coil"
else:
rate = max(lookup_pump_rate(prior_pumps, days_since), BASELINE_PUMP_RATE)
p, label = 7, "Registry only"
return {"priority": p, "label": label, "icon": COMPOSITE_ICONS[p],
"expected_rate": rate, "is_serial": is_serial, "is_active": is_active}
# ββ Signal log helpers βββββββββββββββββββββββββββββββββββββββββββββββββββββ
def append_pump_signals_to_log(hits_df: pd.DataFrame) -> bool:
"""
Save pump scanner hits to the shared HF signal log.
Uses Scanner_Type='Pump' to distinguish from combined scanner entries.
"""
if hits_df.empty:
return False
if not (HF_TOKEN and DATASET_REPO):
return False
try:
existing = load_signal_log()
new_rows = hits_df.copy()
new_rows["FlagDate"] = pd.Timestamp.now().normalize()
new_rows["Scanner_Type"] = "Pump"
combined = (pd.concat([existing, new_rows], ignore_index=True)
if not existing.empty else new_rows)
combined = combined.drop_duplicates(
subset=["Ticker", "FlagDate", "Scanner_Type"], keep="last")
save_signal_log(combined)
return True
except Exception as e:
st.warning(f"Could not save pump signals: {e}")
return False
# ββ Data helpers βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
@st.cache_data(ttl=3 * 3600, show_spinner=False)
def fetch_penny_data(ticker: str, period: str = REGISTRY_PERIOD) -> pd.DataFrame | None:
try:
raw = yf.download(ticker, period=period, interval="1d",
progress=False, auto_adjust=True)
if isinstance(raw.columns, pd.MultiIndex):
raw.columns = raw.columns.get_level_values(0)
return raw if not raw.empty else None
except Exception:
return None
@st.cache_data(ttl=6 * 3600, show_spinner=False)
def run_penny_scan(
tickers_tuple: tuple,
min_price: float,
max_price: float,
require_exchange: bool,
min_priority: int,
period: str = REGISTRY_PERIOD,
) -> pd.DataFrame:
"""
Full-universe scan implementing the v3 cascade:
Step 1-2: build the serial-pumper registry from `period` of history
Step 3: technical gate (BB/price tier, coil/surge pattern)
Step 4: composite score β keep rows at or above min_priority
(1 = best combo, 7 = weakest / registry-only)
"""
rows = [{"ticker": t[0], "yf": t[1], "name": t[2],
"close": t[3], "exchange": t[4], "otc_f": t[5]}
for t in tickers_tuple]
results = []
batch = 10
n_batch = max(1, (len(rows) + batch - 1) // batch)
prog = st.progress(0, text="Scanningβ¦")
for b, i in enumerate(range(0, len(rows), batch)):
batch_t = rows[i:i + batch]
batch_yf = [r["yf"] for r in batch_t]
try:
data = yf.download(batch_yf, period=period, interval="1d",
group_by="ticker", progress=False,
threads=True, auto_adjust=True)
except Exception:
prog.progress((b+1)/n_batch); continue
for row in batch_t:
ticker = row["ticker"]
yf_tick = row["yf"]
price = row["close"]
exchange = row["exchange"]
is_otc_f = row["otc_f"]
if price < min_price or price > max_price:
continue
if require_exchange and exchange == "OTC":
continue
try:
df_raw = data[yf_tick] if len(batch_yf) > 1 else data
if df_raw is None or df_raw.dropna(how="all").empty:
continue
reg = serial_pumper_stats(df_raw)
features = compute_features(df_raw)
if features is None:
continue
sig = check_signals(features)
tier_info = classify_pump_tier(sig["close"], sig["bb_width"])
comp = classify_composite(reg["prior_pumps"], reg["days_since_last"],
tier_info, sig)
if comp["priority"] > min_priority:
continue
# Seasonal flag
month = pd.Timestamp.now().month
season_tag = ("GOOD month" if month in GOOD_MONTHS else
("WEAK month" if month in WEAK_MONTHS else ""))
pattern_label = ("Coil+Surge" if (sig["pattern_a"] and sig["pattern_b"])
else "Surge" if sig["pattern_b"]
else "Coil" if sig["pattern_a"] else "β")
days_since_disp = reg["days_since_last"] if reg["days_since_last"] is not None else "β"
results.append({
"Composite": comp["icon"] + " " + comp["label"],
"Exp. Rate%": comp["expected_rate"],
"Tier": TIER_ICONS.get(tier_info["tier"],"") + " " + tier_info["label"],
"Pattern": pattern_label,
"Ticker": ticker,
"Name": row["name"][:22],
"Exchange": exchange + (" β οΈ" if is_otc_f else ""),
"Price": sig["close"],
"Prior Pumps": reg["prior_pumps"],
"Days Since": days_since_disp,
"BB Width%": sig["bb_width"],
"RSI": sig["rsi"],
"Vol x MA20": sig["vol_ratio"],
"Range pos": sig["range_pos"],
"Season": season_tag,
"Vol MA20": int(sig["vol_ma20"]),
"Vol today": int(sig["vol_today"]),
# columns for signal log / styling
"_priority": comp["priority"],
"_is_otc_f": is_otc_f,
})
except Exception:
continue
prog.progress((b+1)/n_batch, text=f"Batch {b+1}/{n_batch}β¦")
prog.empty()
if not results:
return pd.DataFrame()
out = (pd.DataFrame(results)
.sort_values(["_priority", "Exp. Rate%"], ascending=[True, False])
.reset_index(drop=True))
return out
# ββ Chart ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
def plot_penny_chart(ticker: str, yf_ticker: str):
with st.spinner(f"Fetching {ticker}β¦"):
raw = fetch_penny_data(yf_ticker)
reg = serial_pumper_stats(raw) if raw is not None else {"prior_pumps": 0, "days_since_last": None, "events": []}
features = compute_features(raw)
if features is None or len(features) < 20:
st.error(f"Not enough data for {ticker}.")
return
sig = check_signals(features)
tier_info = classify_pump_tier(sig["close"], sig["bb_width"])
comp = classify_composite(reg["prior_pumps"], reg["days_since_last"], tier_info, sig)
recent = features.tail(90)
fig = plt.figure(figsize=(13, 12))
gs2 = gridspec.GridSpec(4, 1, figure=fig, hspace=0.55, height_ratios=[2.5, 0.6, 1, 1])
# Price + BB bands + historical pump markers
ax1 = fig.add_subplot(gs2[0])
sma = recent['SMA20']
std = recent['Close'].rolling(20).std().iloc[-len(recent):]
ax1.plot(recent.index, recent['Close'], color='black', lw=1.5, label='Price')
ax1.plot(recent.index, sma, color='blue', lw=0.8, ls='--', alpha=0.6, label='SMA20')
ax1.fill_between(recent.index, sma - 2*std, sma + 2*std,
alpha=0.09, color='purple', label='BB Β±2Ο')
ax1.axvline(recent.index[-1], color='red', lw=1.5, ls='--', alpha=0.7)
recent_events = [e for e in reg["events"] if e in recent.index]
if recent_events:
ev_y = recent.loc[recent_events, 'Close']
ax1.scatter(recent_events, ev_y, marker='*', s=160, color='#e67e22',
edgecolor='black', linewidth=0.5, zorder=5, label='Historical pump')
days_since_txt = f"{reg['days_since_last']}d ago" if reg["days_since_last"] is not None else "none in window"
ax1.set_title(
f"{ticker} ${sig['close']:.4f} | {comp['icon']} {comp['label']} "
f"(~{comp['expected_rate']:.1f}% exp. pump rate)\n"
f"Registry: {reg['prior_pumps']} prior pumps, last {days_since_txt} | "
f"{tier_info['label']} | Score:{sig['score']:.0f}",
fontweight='bold', fontsize=10)
ax1.legend(fontsize=8, loc='upper left'); ax1.grid(True, alpha=0.15)
# Serial pumper recency strip
ax0 = fig.add_subplot(gs2[1])
ax0.axis('off')
rate_table_row = PUMP_RATE_TABLE[bucket_prior_pumps(reg['prior_pumps'])]
cells = list(rate_table_row.items())
cur_bucket = bucket_days_since(reg['days_since_last'])
x0 = 0.0
for label, rate in cells:
w = 1.0 / len(cells)
is_cur = (label == cur_bucket)
ax0.axvspan(x0, x0 + w, ymin=0, ymax=1,
color='#2ca02c' if is_cur else '#dddddd',
alpha=0.85 if is_cur else 0.35)
ax0.text(x0 + w/2, 0.5, f"{label}\n{rate:.1f}%", ha='center', va='center',
fontsize=8, fontweight='bold' if is_cur else 'normal')
x0 += w
ax0.set_xlim(0, 1); ax0.set_ylim(0, 1)
ax0.set_title(f"Prior-pump bucket: {bucket_prior_pumps(reg['prior_pumps'])} "
f"β empirical pump-5d rate by days since last pump", fontsize=8, style='italic')
# BB Width panel
ax2 = fig.add_subplot(gs2[2])
ax2.plot(recent.index, recent['BB_Width'], color='navy', lw=1.5, label='BB Width%')
ax2.axhline(BB_WIDTH_T1_MIN, color='green', ls='--', lw=1,
alpha=0.7, label=f'Tier 1 min ({BB_WIDTH_T1_MIN:.0f}%)')
ax2.axhline(BB_WIDTH_MAX, color='red', ls='--', lw=1,
alpha=0.7, label=f'Hard cap ({BB_WIDTH_MAX:.0f}%)')
ax2.fill_between(recent.index, BB_WIDTH_T1_MIN, BB_WIDTH_MAX,
alpha=0.06, color='green', label='Sweet spot 35-80%')
ax2.set_title(f"BB Width% β current: {sig['bb_width']:.1f}%", fontsize=9)
ax2.legend(fontsize=7, loc='upper left'); ax2.grid(True, alpha=0.15)
# Volume
ax3 = fig.add_subplot(gs2[3])
vol_ratio = recent['Vol_Ratio']
colors_v = ['#e74c3c' if v >= 3 else ('#f39c12' if v >= 1 else '#95a5a6')
for v in vol_ratio]
ax3.bar(recent.index, recent['Volume'], color=colors_v, alpha=0.8, width=1)
ax3.axhline(recent['Vol_MA20'].iloc[-1], color='blue', lw=1.2, ls='--', alpha=0.7,
label=f"20D avg vol ({recent['Vol_MA20'].iloc[-1]:,.0f})")
ax3.set_title(f"Volume | Today: {sig['vol_ratio']:.1f}x MA20 | "
f"Quiet streak before: {sig['quiet_streak']}d | "
f"Range pos: {sig['range_pos']:.2f}", fontsize=9)
from matplotlib.patches import Patch
ax3.legend(handles=[
Patch(facecolor='#e74c3c', label='Vol β₯3x (surge)'),
Patch(facecolor='#f39c12', label='Vol 1-3x (normal)'),
Patch(facecolor='#95a5a6', label='Vol <1x (quiet)'),
] + [ax3.get_legend_handles_labels()[0][0]], fontsize=7, loc='upper left')
ax3.grid(True, alpha=0.15, axis='y')
plt.suptitle("Backtest: max-High β₯50% within 5 days | baseline 1.8% | best combo 25.1% (14x)",
fontsize=8, style='italic', y=1.005)
st.pyplot(fig)
plt.close(fig)
# ββ Universe loader ββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
@st.cache_data(ttl=24 * 3600, show_spinner="Loading penny stock universeβ¦")
def load_penny_universe() -> list[dict]:
try:
from huggingface_hub import hf_hub_download as _hf_dl
df = pd.read_csv(_hf_dl(
repo_id="Trebor1009/clean-premarket-data", repo_type="dataset",
filename="my_penny_stocks.csv", token=HF_TOKEN))
except Exception:
return []
df['exchange'] = df['ticker'].str.split(':').str[0]
df['yf_ticker'] = df['ticker'].str.split(':').str[-1].str.replace('.', '-', regex=False)
df['otc_f'] = (df['exchange'] == 'OTC') & (df['yf_ticker'].str.endswith('F'))
df['close'] = pd.to_numeric(df['close'], errors='coerce').fillna(0)
return df.to_dict('records')
# ββ MAIN RENDER ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
def render_penny_scanner():
st.title("π‘ Small-Cap Pump Scanner v3 β Serial Pumper Framing")
with st.expander("π Backtest findings (2-year, 730 tickers) β read this first", expanded=False):
st.markdown("""
**Headline finding:** *"The stock matters more than the setup."* Prior pump
history and recency (Days Since Last Pump) account for **88% of the model's
predictive power** (feature importance: Days Since Pump = 0.885; everything
else β RSI, BB Width, price, range position β together β 0.10). The old v2
scanner treated coil/surge patterns as the primary signal; v3 treats them as
a secondary gate applied *after* the serial-pumper check.
**Pump-5d rate by Prior Pumps Γ Days Since Last Pump** (the core empirical table):
| Prior pumps | <3d | 3-10d | 10-30d | 30-90d | >90d |
|---|---|---|---|---|---|
| 0 | 0.0% | 0.0% | 0.0% | 0.0% | 0.0% |
| 1-2 | 11.9% | 3.8% | 0.0% | 0.0% | 0.0% |
| 3-4 | 9.8% | 7.3% | 0.0% | 0.3% | 0.1% |
| **β₯5 (serial)** | **37.2%** | **10.1%** | 0.5% | 0.2% | 0.1% |
An "active serial pumper" (β₯5 prior pumps, last one β€10 days ago) hits **37%**
in the <3d bucket vs a **2.3%** all-ticker baseline (~1.8% at the stricter
scoring threshold used in the combo chart below).
**Best signal combinations (5-day horizon, max-High β₯50%):**
| Combo | Rate | vs baseline |
|---|---|---|
| Serial + Tier 1A/2A + technical signal | **25.1%** | 14x |
| Serial + Tier 1A/2A (no signal yet) | 22.4% | 12x |
| Serial pumper, active β€10d (registry only) | 18.7% | 10x |
| Pattern B surge >5x + Tier (GRML lesson) | 14.2% | 8x |
| Pattern B surge >3x + Tier | 10.6% | 6x |
| Pattern A coil + Tier | 8.3% | 5x |
| Baseline (random) | 1.8% | β |
Model AUC at 3-10 days since last pump is **0.85** β genuine prediction
beyond simple continuation (0-3 days is mostly continuation, AUC 0.65).
Full gradient-boosting model AUC: **0.979**; precision β20% at score
threshold 0.20. RSI and 20-day range position are *not* independently
predictive β kept only as pre-filters, not scoring inputs.
**GRML lesson:** a pure volume surge without a coil (Pattern B) was
previously excluded as noise. Re-enabled at vol >5x + Tier 1A/2A, but its
edge is conditional β check the registry: a non-serial ticker surging gets
~10-14%, a serial ticker (β₯5 prior pumps) surging gets 25%+.
""")
# ββ Sidebar βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
with st.sidebar:
st.markdown("---")
st.header("Pump Scanner v3")
max_price_f = st.slider("Max price ($)", 0.10, 5.00, 5.00, 0.10, key="ps3_max")
min_price_f = st.slider("Min price ($)", 0.01, 2.00, 0.05, 0.01, key="ps3_min")
min_priority_f = st.radio(
"Minimum combo strength", [3, 5, 6, 7],
format_func=lambda x: {
3: "π₯ Serial pumpers only (β₯18.7%)",
5: "β‘ + surge/GRML combos (β₯10.6%)",
6: "π + coil combos (β₯8.3%)",
7: "β’ Show everything (registry-only incl.)"}[x],
index=0, key="ps3_priority")
excl_otc = st.checkbox("Exclude OTC (exchange-listed only)", value=False, key="ps3_exch")
registry_period = st.selectbox(
"Registry lookback (for prior-pump count)",
["1y", "2y", "5y"], index=1, key="ps3_period",
help="Longer lookback finds more historical pumps but the scan is slower.")
st.markdown("---")
st.caption(
"Registry-first: a ticker with β₯5 prior pumps active β€10d beats "
"any technical pattern alone. Technical tier/pattern is now a "
"secondary boost, not the primary trigger.\n\n"
"OTC-F β οΈ = foreign OTC β volume data may be unreliable"
)
# ββ Single ticker βββββββββββββββββββββββββββββββββββββββββββββββββββββ
st.subheader("Analyze a single ticker")
col_a, col_b = st.columns([2, 1])
single = col_a.text_input("Ticker (Yahoo Finance symbol)", value="LEDS", key="ps3_single")
go_btn = col_b.button("Analyze", key="ps3_go")
if go_btn and single:
raw = fetch_penny_data(single.strip().upper(), period=registry_period)
reg = serial_pumper_stats(raw) if raw is not None else {"prior_pumps": 0, "days_since_last": None, "events": []}
features = compute_features(raw)
if features is None:
st.error(f"No usable data for {single}.")
else:
sig = check_signals(features)
tier_info = classify_pump_tier(sig["close"], sig["bb_width"])
comp = classify_composite(reg["prior_pumps"], reg["days_since_last"], tier_info, sig)
c1,c2,c3,c4,c5,c6 = st.columns(6)
c1.metric("Price", f"${sig['close']:.4f}")
c2.metric("Prior pumps", reg["prior_pumps"])
c3.metric("Days since", reg["days_since_last"] if reg["days_since_last"] is not None else "β")
c4.metric("BB Width", f"{sig['bb_width']:.1f}%")
c5.metric("RSI", f"{sig['rsi']}")
c6.metric("Vol ratio", f"{sig['vol_ratio']:.1f}x")
msg = (f"{comp['icon']} **{comp['label']}** | "
f"Expected pump rate: ~{comp['expected_rate']:.1f}% (baseline {BASELINE_PUMP_RATE}%)\n\n"
f"{tier_info['label']} | Pattern: "
f"{'Coil' if sig.get('pattern_a') else ''}"
f"{'+Surge' if sig.get('pattern_b') else ''}"
f"{' (none)' if not (sig.get('pattern_a') or sig.get('pattern_b')) else ''} | "
f"Score {sig['score']:.0f}/100")
if comp["priority"] <= 2:
st.success(msg)
elif comp["priority"] <= 4:
st.info(msg)
elif comp["priority"] <= 6:
st.warning(msg)
else:
st.info(msg)
plot_penny_chart(single.strip().upper(), single.strip().upper())
# ββ Full scan βββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
st.markdown("---")
st.subheader("Full watchlist scan")
universe = load_penny_universe()
filtered = [r for r in universe
if min_price_f <= r.get("close", 0) <= max_price_f
and (not excl_otc or r.get("exchange") != "OTC")]
month_now = pd.Timestamp.now().month
season_txt = ("Good month for pumps (Mar/Apr/May/Aug/Sep historically strongest)" if month_now in GOOD_MONTHS else
"Weak month β Jun/Jul historically below baseline" if month_now in WEAK_MONTHS else "")
if season_txt:
st.info(f"π
{season_txt}")
st.caption(
f"Universe: {len(filtered):,} tickers after price filter | "
f"Registry lookback: {registry_period} | Cached 6h | "
f"Longer lookback = slower first run"
)
if st.button("π Run Scan", key="ps3_scan"):
if not filtered:
st.error("No tickers match the filters.")
else:
scan_tuples = tuple(
(r.get("ticker", r.get("yf_ticker", "")),
r.get("yf_ticker", r.get("ticker", "").split(":")[-1]),
r.get("name", ""),
float(r.get("close", 0)),
str(r.get("exchange", "UNK")),
bool(r.get("otc_f", False)))
for r in filtered
)
results_df = run_penny_scan(scan_tuples, min_price_f, max_price_f,
excl_otc, min_priority_f, registry_period)
st.session_state["ps3_results"] = results_df
# Auto-log to signal tracker
if not results_df.empty and HF_TOKEN and DATASET_REPO:
log_df = results_df[[
"Ticker","Price","BB Width%","RSI","Vol x MA20",
"Prior Pumps","Days Since","Exp. Rate%","Tier","Pattern",
"Composite","Season"
]].copy()
log_df = log_df.rename(columns={
"BB Width%": "BB_Width",
"Vol x MA20": "Vol_Ratio",
"Prior Pumps": "Prior_Pumps",
"Days Since": "Days_Since_Pump",
"Exp. Rate%": "Est_Pump_Rate",
})
if append_pump_signals_to_log(log_df):
st.success(f"β
{len(log_df)} signal(s) saved to Signal Tracker.")
else:
st.caption("Signal log not configured β hits not saved.")
if "ps3_results" in st.session_state:
df_r = st.session_state["ps3_results"]
if df_r.empty:
st.info("No tickers matched. Try relaxing the combo-strength filter or price range.")
else:
serial_n = (df_r["_priority"] <= 3).sum()
signal_n = (df_r["_priority"] <= 1).sum()
st.success(f"{len(df_r)} hits β π₯ {serial_n} serial pumpers active | "
f"π₯π₯π₯ {signal_n} serial+tier+signal (best combo)")
st.caption("β οΈ = OTC-F (foreign OTC) β verify volume on chart before acting.")
display_cols = ["Composite","Exp. Rate%","Tier","Pattern","Ticker","Name","Exchange",
"Price","Prior Pumps","Days Since","BB Width%","RSI","Vol x MA20",
"Range pos","Season"]
display_cols = [c for c in display_cols if c in df_r.columns]
def row_style(row):
p = row.get("_priority", 9) if "_priority" in row.index else 9
if p == 1: return ["background-color: #a3d9a5"] * len(row)
if p <= 3: return ["background-color: #d4edda"] * len(row)
if p <= 5: return ["background-color: #fff3cd"] * len(row)
return [""] * len(row)
st.dataframe(
df_r[display_cols].style.apply(row_style, axis=1),
use_container_width=True, hide_index=True,
)
pick = st.selectbox("View chart:", df_r["Ticker"].tolist(), key="ps3_pick")
if pick:
match = [r for r in filtered
if r.get("ticker","") == pick or r.get("yf_ticker","") == pick]
yf_t = match[0].get("yf_ticker", pick.split(":")[-1]) if match else pick
plot_penny_chart(pick, yf_t)
# ββ Usage guide βββββββββββββββββββββββββββββββββββββββββββββββββββββββ
st.markdown("---")
with st.expander("π‘ How to use + position sizing"):
st.markdown(f"""
**Step 1-2 (primary): Serial Pumper Registry.**
Every ticker's price history ({registry_period}) is scanned for prior pump
events (max-High β₯{PUMP_MOVE_THRESH*100:.0f}% within {PUMP_HORIZON} days,
{PUMP_COOLDOWN}-day cooldown to avoid double-counting one move). β₯5 events
= "serial pumper." Last event β€10 days ago = "active." This alone (no
technical pattern required) is worth 18.7%, more than any pattern-only combo.
**Step 3 (secondary): Technical gate.**
Coil break: quiet β₯{COIL_MIN_DAYS}d (vol <{COIL_VOL_THRESH}x avg), vol returns
β₯{COIL_BREAK_MIN}x, RSI <{RSI_COIL_MAX}, bottom 35% of 20-day range, BB Width
{BB_WIDTH_T1_MIN}-{BB_WIDTH_MAX}%. Surge (GRML-enabled): vol β₯{SURGE_VOL_MIN}x
(β₯{SURGE_VOL_GRML}x for the strongest tier), close in top
{int(SURGE_RANGE_POS*100)}% of day's range, same RSI/range/BB filters.
**Step 4: Composite score.**
Registry status Γ technical gate β expected pump rate from the empirical
table, used to rank and color the results.
**Exit rule (from backtest):**
If stock reaches +50% above entry at any point in 20 days β sell 2 days after.
If no pump in 20 days β sell at day 10 close.
**Position sizing:**
These are lottery tickets. A handful of large winners drive all the profit β
size small enough that 10 consecutive losers doesn't change your behaviour.
The registry improves hit rate meaningfully (37% in the best bucket vs ~2%
baseline) but is still far from a coin flip β most flagged trades still lose.
**OTC-F β οΈ stocks:**
Foreign OTC tickers ending in 'F' often have sparse Yahoo volume data β
unreliable registry counts and false technical signals. Always check the
chart. Exchange-listed NASDAQ/NYSE signals are more reliable.
""")
# ββ Shared data helper (Signal Tracker + Analog Projection) βββββββββββββββββ
@st.cache_data(ttl=12 * 3600, show_spinner=False)
def get_sniper_ticker_data(ticker: str) -> pd.DataFrame:
"""Fetch max-period daily OHLCV β used by Signal Tracker and Analog Projection."""
df = yf.download(ticker, period="max", interval="1d", progress=False)
if df.empty:
return df
if isinstance(df.columns, pd.MultiIndex):
df.columns = df.columns.get_level_values(0)
return df.dropna()
# ANALOG PATH PROJECTION
# =========================================================
ANALOG_LOOKBACK = 60
ANALOG_FORWARD = 60
ANALOG_TOP_N = 5
ANALOG_WEIGHT_STRENGTH = 5
def _similarity_matches(close, end_idx, lookback, forward):
if end_idx - lookback < 0:
return []
recent = close.iloc[end_idx - lookback:end_idx]
recent_norm = (recent / recent.iloc[0]).values.flatten()
matches = []
max_search = end_idx - lookback - forward
for i in range(max_search):
cand = close.iloc[i:i + lookback]
cand_norm = (cand / cand.iloc[0]).values.flatten()
corr = spearmanr(recent_norm, cand_norm)[0]
if not np.isnan(corr):
matches.append((float(corr), i))
return matches
@st.cache_data(ttl=24 * 3600, show_spinner="Building analog path projection...")
def compute_analog_projection(ticker: str, flag_date_str: str, lookback=ANALOG_LOOKBACK,
forward=ANALOG_FORWARD, top_n=ANALOG_TOP_N,
weight_strength=ANALOG_WEIGHT_STRENGTH):
hist = get_sniper_ticker_data(ticker)
if hist.empty:
return None
close = hist["Close"]
flag_date = pd.Timestamp(flag_date_str)
end_idx = int(close.index.searchsorted(flag_date))
end_idx = min(end_idx, len(close) - 1)
if end_idx - lookback - forward < 5:
return None
matches = _similarity_matches(close, end_idx, lookback, forward)
if not matches:
return None
matches.sort(key=lambda x: x[0], reverse=True)
filtered, used = [], []
for corr, idx in matches:
if all(abs(idx - u) >= lookback for u in used):
filtered.append((corr, idx))
used.append(idx)
if len(filtered) >= top_n:
break
if not filtered:
return None
corrs = np.array([c for c, _ in filtered])
weights = np.exp(weight_strength * corrs)
weights /= weights.sum()
anchor_price = close.iloc[end_idx]
projections = []
for corr, idx in filtered:
fwd = close.iloc[idx + lookback: idx + lookback + forward]
fwd_norm = (fwd / fwd.iloc[0]).values.flatten()
projections.append(fwd_norm * anchor_price)
projections = np.vstack(projections)
weighted_mean = np.average(projections, axis=0, weights=weights)
weighted_var = np.average((projections - weighted_mean) ** 2, axis=0, weights=weights)
weighted_std = np.sqrt(weighted_var)
future_dates = pd.bdate_range(start=close.index[end_idx], periods=forward + 1)[1:]
return {
"future_dates": future_dates,
"weighted_mean": weighted_mean,
"upper": weighted_mean + weighted_std,
"lower": weighted_mean - weighted_std,
"robust_similarity": float(np.sqrt(np.sum(weights * corrs ** 2)) * 100),
"match_count": len(filtered),
}
# =========================================================
# TAB 4: SIGNAL TRACKER
# =========================================================
def render_signal_tracker():
st.title("ποΈ Signal Tracker")
st.caption(
"Every signal flagged by the Pump Scanner or Combined Scanner, "
"with predicted vs actual outcome. This is the real out-of-sample test."
)
if not _log_configured():
st.error(
"Signal log not configured. Add HF_TOKEN secret and HF_DATASET_REPO "
"variable in Space Settings, then restart."
)
return
log = load_signal_log()
if log.empty:
st.info("No signals logged yet. Run the Pump Scanner or Combined Scanner.")
return
log["FlagDate"] = pd.to_datetime(log["FlagDate"])
log["DaysSinceFlag"] = (pd.Timestamp.now().normalize() - log["FlagDate"]).dt.days
# Detect scanner type β add column if missing (backward compat)
if "Scanner_Type" not in log.columns:
log["Scanner_Type"] = "Combined"
pump_log = log[log["Scanner_Type"] == "Pump"].copy()
comb_log = log[log["Scanner_Type"] != "Pump"].copy()
st.markdown(
f"### {len(log)} signals logged β "
f"π‘ {len(pump_log)} Pump | π― {len(comb_log)} Combined"
)
tab_pump, tab_comb = st.tabs(["π‘ Pump Scanner", "π― Combined Scanner"])
# ββ PUMP SCANNER TAB ββββββββββββββββββββββββββββββββββββββββββββββββββ
with tab_pump:
if pump_log.empty:
st.info("No pump signals logged yet. Run the Pump Scanner tab.")
else:
st.caption(
"Exit rule: +50% pump in 20 days β sell 2 days after. "
"No pump β sell at day 10. Baseline pump rate: 8.8%."
)
pump_rows = []
for _, row in pump_log.iterrows():
ticker = row["Ticker"]
flag_price = float(row.get("Price", 0) or 0)
days = int(row["DaysSinceFlag"])
tier = str(row.get("Tier", "β"))
pattern = str(row.get("Pattern", "β"))
est_rate = row.get("Est_Pump_Rate", row.get("Pump Rate%", np.nan))
bb_width = row.get("BB_Width", np.nan)
season = str(row.get("Season", ""))
pumped = False
pump_ret = np.nan
day10_ret = np.nan
current_ret = np.nan
try:
hist = get_sniper_ticker_data(ticker)
if not hist.empty and flag_price > 0:
flag_dt = row["FlagDate"]
fwd = hist[hist.index > flag_dt]["Close"]
cur_px = hist["Close"].iloc[-1]
current_ret = (cur_px / flag_price - 1) * 100
# Check if +50% was hit in first 20 bars
fwd20 = fwd.iloc[:20]
if len(fwd20) > 0:
max_px = fwd20.max()
if max_px >= flag_price * 1.50:
pumped = True
pump_ret = (max_px / flag_price - 1) * 100
# Day-10 return
if len(fwd) >= 10:
day10_ret = (fwd.iloc[9] / flag_price - 1) * 100
except Exception:
pass
# Status
if days >= 20:
status = f"Complete ({days}d)"
else:
status = f"In window ({days}/20d)"
pump_rows.append({
"Ticker": ticker,
"Flag Date": row["FlagDate"].date(),
"Tier": tier,
"Pattern": pattern,
"Est Pump%": round(float(est_rate), 1) if pd.notna(est_rate) else None,
"BB Width%": round(float(bb_width), 1) if pd.notna(bb_width) else None,
"Season": season,
"Days": days,
"Status": status,
"Pumped >50%": "β
" if pumped else ("β¬" if days < 20 else "β"),
"Pump return%": round(pump_ret, 1) if pd.notna(pump_ret) else None,
"Day-10 ret%": round(day10_ret, 1) if pd.notna(day10_ret) else None,
"Current ret%": round(current_ret, 1) if pd.notna(current_ret) else None,
})
pump_df = pd.DataFrame(pump_rows).sort_values("Flag Date", ascending=False)
st.dataframe(pump_df, use_container_width=True, hide_index=True)
# Summary stats for completed signals
done = pump_df[pump_df["Status"].str.startswith("Complete")]
if not done.empty:
pump_hit = (done["Pumped >50%"] == "β
").mean() * 100
day10_wr = (done["Day-10 ret%"].dropna() > 0).mean() * 100 if done["Day-10 ret%"].notna().any() else np.nan
col1, col2, col3 = st.columns(3)
col1.metric("Pump detection rate (completed)",
f"{pump_hit:.1f}% (n={len(done)})",
delta=f"vs 8.8% baseline")
col2.metric("Day-10 win rate",
f"{day10_wr:.1f}%" if pd.notna(day10_wr) else "β")
col3.metric("Signals in window", f"{(pump_df['Status'].str.contains('window')).sum()}")
else:
st.caption("No signals have passed the 20-day window yet.")
# Download
st.download_button(
"β¬οΈ Download pump signal log",
data=pump_log.to_csv(index=False).encode("utf-8"),
file_name="pump_signals.csv", mime="text/csv",
)
# Chart picker
st.markdown("---")
st.subheader("Chart a pump signal")
pump_opts = [f"{r['Ticker']} β flagged {r['Flag Date']}"
for _, r in pump_df.iterrows()]
if pump_opts:
pick = st.selectbox("Pick a signal", pump_opts, key="st_pump_pick")
if pick:
tkr, fdate_str = pick.split(" β flagged ")
hist = get_sniper_ticker_data(tkr)
match_rows = pump_log[
(pump_log["Ticker"] == tkr) &
(pump_log["FlagDate"].dt.date.astype(str) == fdate_str)
]
if not hist.empty and not match_rows.empty:
mrow = match_rows.iloc[0]
flag_date = mrow["FlagDate"]
flag_px = float(mrow.get("Price", 0) or 0)
context = hist[hist.index >= flag_date - pd.Timedelta(days=10)]
fwd = hist[hist.index > flag_date].iloc[:25]
fig, ax = plt.subplots(figsize=(13, 5))
if not context.empty:
ax.plot(context.index, context["Close"],
color="black", lw=1.5, marker="o", ms=3, label="Price")
ax.axvline(flag_date, color="red", lw=1.5, ls="--",
alpha=0.7, label="Signal date")
if flag_px > 0:
ax.axhline(flag_px, color="gray", ls=":", alpha=0.6,
label=f"Entry ${flag_px:.4f}")
ax.axhline(flag_px * 1.5, color="green", ls="--",
alpha=0.6, lw=1.2, label="+50% pump target")
if not fwd.empty:
ax.plot(fwd.index, fwd["Close"],
color="orange", lw=1.5, ls="--", label="Forward price")
tier = str(mrow.get("Tier","β"))
pat = str(mrow.get("Pattern","β"))
ax.set_title(f"{tkr} β {pat} | {tier} | flagged {fdate_str}",
fontweight="bold")
ax.legend(fontsize=8); ax.grid(True, alpha=0.12)
st.pyplot(fig); plt.close(fig)
# ββ COMBINED SCANNER TAB ββββββββββββββββββββββββββββββββββββββββββββββ
with tab_comb:
if comb_log.empty:
st.info("No combined scanner signals logged yet.")
else:
comb_log2 = comb_log.copy()
comb_log2["DaysSinceFlag"] = (
pd.Timestamp.now().normalize() - comb_log2["FlagDate"]
).dt.days
summary_rows = []
for _, row in comb_log2.iterrows():
ticker = row["Ticker"]
flag_price = float(row.get("Price", 0) or 0)
days = int(row["DaysSinceFlag"])
actual_ret = np.nan
try:
hist = get_sniper_ticker_data(ticker)
if not hist.empty and flag_price > 0:
actual_ret = (hist["Close"].iloc[-1] / flag_price - 1) * 100
except Exception:
pass
status = "Complete (63D+)" if days >= 63 else f"In progress ({days}d/63d)"
summary_rows.append({
"Ticker": ticker,
"Flag Date": row["FlagDate"].date(),
"Days": days,
"Status": status,
"Tier": row.get("Tier","β"),
"Est Win%": row.get("Est. Win%", row.get("WinRate_3M", np.nan)),
"Actual ret%": round(actual_ret, 2) if pd.notna(actual_ret) else None,
"On Track": (actual_ret > 0) if pd.notna(actual_ret) else None,
})
comb_df = pd.DataFrame(summary_rows).sort_values("Flag Date", ascending=False)
st.dataframe(comb_df, use_container_width=True, hide_index=True)
done_c = comb_df[comb_df["Status"].str.startswith("Complete")]
if not done_c.empty:
hr = (done_c["On Track"] == True).mean() * 100
st.metric("Realized hit rate on 63D signals",
f"{hr:.0f}% (n={len(done_c)})", delta="vs 61.4% backtest")
else:
st.caption("No combined signals have reached 63 days yet.")
st.download_button(
"β¬οΈ Download combined signal log",
data=comb_log.to_csv(index=False).encode("utf-8"),
file_name="combined_signals.csv", mime="text/csv",
)
st.markdown("---")
st.subheader("Chart a combined signal")
comb_opts = [f"{r['Ticker']} β flagged {r['Flag Date']}"
for _, r in comb_df.iterrows()]
if comb_opts:
pick2 = st.selectbox("Pick a signal", comb_opts, key="st_comb_pick")
if pick2:
tkr2, fdate2 = pick2.split(" β flagged ")
match2 = comb_log[
(comb_log["Ticker"] == tkr2) &
(comb_log["FlagDate"].dt.date.astype(str) == fdate2)
]
if not match2.empty:
mrow2 = match2.iloc[0]
flag_date2 = mrow2["FlagDate"]
flag_px2 = float(mrow2.get("Price", 0) or 0)
hist2 = get_sniper_ticker_data(tkr2)
if not hist2.empty:
context2 = hist2[hist2.index >= flag_date2 - pd.Timedelta(days=45)]
fig2, ax2 = plt.subplots(figsize=(13, 5))
if not context2.empty:
ax2.plot(context2.index, context2["Close"],
color="black", lw=1.5, marker="o", ms=3, label="Price")
ax2.axvline(flag_date2, color="red", lw=1.5, ls="--",
alpha=0.7, label="Signal date")
if flag_px2 > 0:
ax2.axhline(flag_px2, color="gray", ls=":",
alpha=0.6, label=f"Entry ${flag_px2:.2f}")
proj = compute_analog_projection(tkr2, str(flag_date2.date()))
if proj:
ax2.plot(proj["future_dates"], proj["weighted_mean"],
color="green", ls="--", lw=2, label="Analog projection")
ax2.fill_between(proj["future_dates"],
proj["lower"], proj["upper"],
color="green", alpha=0.12)
ax2.set_title(f"{tkr2} β flagged {fdate2}",fontweight="bold")
ax2.legend(fontsize=8); ax2.grid(True, alpha=0.12)
st.pyplot(fig2); plt.close(fig2)
# =========================================================
# TAB 5: ROTATION MAP
# =========================================================
DEFAULT_ROTATION_TICKERS = ["XLK", "XLB", "XLF", "XLE", "XLV", "XLI", "XLU", "XLP", "XLRE", "XLY", "XLC"]
@st.cache_data(ttl=4 * 3600, show_spinner="Building rotation map...")
def compute_rotation_data(tickers_tuple, benchmark, lookback_years, window):
symbols = list(tickers_tuple) + [benchmark]
start_date = (pd.Timestamp.now() - pd.DateOffset(years=lookback_years)).strftime("%Y-%m-%d")
try:
data = yf.download(symbols, start=start_date, progress=False)["Close"]
except Exception:
return None, list(tickers_tuple)
if isinstance(data, pd.Series):
data = data.to_frame()
weekly = data.resample("W-FRI").last()
missing = [t for t in tickers_tuple if t not in weekly.columns or weekly[t].dropna().empty]
valid_tickers = [t for t in tickers_tuple if t not in missing]
if benchmark not in weekly.columns or weekly[benchmark].dropna().empty or not valid_tickers:
return None, missing
rs_df = weekly[valid_tickers].div(weekly[benchmark], axis=0)
rs_ratio = 100 + ((rs_df - rs_df.rolling(window).mean()) / rs_df.rolling(window).std()) * 5
rs_momentum = 100 + ((rs_ratio - rs_ratio.rolling(window).mean()) / rs_ratio.rolling(window).std()) * 5
rs_ratio = rs_ratio.dropna()
rs_momentum = rs_momentum.dropna()
return {"rs_ratio": rs_ratio, "rs_momentum": rs_momentum, "valid_tickers": valid_tickers}, missing
def render_rotation_map():
st.title("π§ Rotation Map")
st.caption(
"Plots relative strength and momentum versus a benchmark for a basket of tickers, showing "
"which quadrant each one currently sits in and how it's been moving over recent weeks."
)
st.info(
"π How to read it: this is RELATIVE positioning vs the benchmark, not absolute "
"performance. Quadrants: Leading (strong + improving), Weakening (strong but losing steam), "
"Lagging (weak + still losing ground), Improving (weak but starting to turn)."
)
st.warning(
"β οΈ 'Momentum' here is a z-score of the relative-strength ratio itself, not a true "
"rate-of-change. Small/thinly-traded names show noisier loops than large liquid ETFs β "
"a big swing on a small-cap isn't necessarily more meaningful than a small swing on XLK."
)
with st.sidebar:
st.markdown("---")
st.header("Rotation Map Settings")
tickers_input = st.text_area(
"Tickers (comma-separated)",
value=", ".join(DEFAULT_ROTATION_TICKERS), key="rm_tickers",
)
benchmark = st.text_input("Benchmark", value="RSP", key="rm_benchmark").strip().upper()
trail_weeks = st.slider("Trail length (weeks)", 2, 12, 6, 1, key="rm_trail")
lookback_years = st.slider("History to fetch (years)", 1, 5, 2, 1, key="rm_lookback")
window = st.slider("Z-score window (weeks)", 8, 52, 21, 1, key="rm_window")
if st.button("π Refresh data", key="rm_refresh"):
compute_rotation_data.clear()
tickers = tuple(t.strip().upper() for t in tickers_input.split(",") if t.strip())
if not tickers:
st.info("Enter at least one ticker in the sidebar.")
return
result, missing = compute_rotation_data(tickers, benchmark, lookback_years, window)
if missing:
st.warning(f"No usable data for: {', '.join(missing)} β excluded from the chart.")
if result is None:
st.error("Not enough data to build the rotation map. Check the benchmark and ticker symbols.")
return
rs_ratio = result["rs_ratio"]
rs_momentum = result["rs_momentum"]
valid_tickers = result["valid_tickers"]
tail_ratio = rs_ratio.tail(trail_weeks)
tail_mom = rs_momentum.tail(trail_weeks)
if tail_ratio.empty:
st.error("Not enough history after the rolling window β try a shorter window or longer lookback.")
return
fig = go.Figure()
colors = px.colors.qualitative.Dark24
for i, ticker in enumerate(valid_tickers):
if ticker not in tail_ratio.columns:
continue
x_data = tail_ratio[ticker].dropna().tolist()
y_data = tail_mom[ticker].dropna().tolist() if ticker in tail_mom.columns else []
n = min(len(x_data), len(y_data))
if n == 0:
continue
x_data, y_data = x_data[-n:], y_data[-n:]
color = colors[i % len(colors)]
fig.add_trace(go.Scatter(
x=x_data, y=y_data, mode="lines+markers+text", name=ticker,
text=[""] * (n - 1) + [ticker], textposition="top center",
textfont=dict(size=12, color=color),
marker=dict(size=[6] * (n - 1) + [16], color=color, symbol="circle"),
line=dict(width=2, color=color, shape="spline"),
))
fig.add_hline(y=100, line_dash="dash", line_color="gray", opacity=0.5)
fig.add_vline(x=100, line_dash="dash", line_color="gray", opacity=0.5)
for x, y, text, color in [(105,105,"LEADING","green"),(105,95,"WEAKENING","orange"),
(95,95,"LAGGING","red"),(95,105,"IMPROVING","blue")]:
fig.add_annotation(x=x, y=y, text=text, showarrow=False,
font=dict(size=20, color=color), xref="x", yref="y", opacity=0.15)
fig.update_layout(
title=f"Rotation Map vs {benchmark} (Last {trail_weeks} Weeks)",
xaxis_title="Relative Strength (RS-Ratio)",
yaxis_title="Momentum (RS-Momentum)",
height=700, template="plotly_white", hovermode="closest", showlegend=False,
)
st.plotly_chart(fig, use_container_width=True)
st.caption(
"Tip: a Bottom Scanner hit sitting deep in Lagging with its tail starting to curl toward "
"Improving is worth cross-referencing with the Combined Scanner (Tab 6) to check whether "
"the daily RRG turn has started yet."
)
# =========================================================
# TAB 6: COMBINED SCANNER (RRG + Bottom Scanner)
# =========================================================
# Quality / liquidity constants
PRICE_FLOOR = 5.0
MIN_DVOL = 1_000_000
LARGE_CAP_DVOL = 50_000_000
DEDUP_DAYS = 63
PRICE_FLOOR = 5.0
MIN_DVOL = 1_000_000 # $1M 20-day avg dollar volume β quality gate
LARGE_CAP_DVOL = 50_000_000 # $50M for large-cap classification
DEDUP_DAYS = 63
# ββ RRG maths βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
def _rrg(price: pd.Series, benchmark: pd.Series, window: int) -> pd.DataFrame:
rs = price / (benchmark + 1e-9)
rs_ratio = 100 + ((rs - rs.rolling(window).mean()) /
(rs.rolling(window).std() + 1e-9)) * 5
rs_mom = 100 + ((rs_ratio - rs_ratio.rolling(window).mean()) /
(rs_ratio.rolling(window).std() + 1e-9)) * 5
return pd.DataFrame({"RS_Ratio": rs_ratio, "RS_Momentum": rs_mom})
def _quadrant(ratio: float, mom: float) -> str:
if ratio >= 100 and mom >= 100: return "Leading"
elif ratio >= 100 and mom < 100: return "Weakening"
elif ratio < 100 and mom < 100: return "Lagging"
else: return "Improving"
QUAD_COLORS = {"Leading": "#27ae60", "Weakening": "#e67e22",
"Lagging": "#e74c3c", "Improving": "#2980b9"}
QUAD_ICONS = {"Leading": "π’", "Weakening": "π ",
"Lagging": "π΄", "Improving": "π΅"}
# ββ Signal tier classification (rebuilt from properly-deduped, out-of-sample-tested grid) ββ
#
# HISTORY: the original version of this function was calibrated from a raw
# (un-deduped) backtest. Investigation found that ~75% of raw daily hits were
# not independent signals β a single sustained RRG move can re-satisfy the
# 2-day-turn condition on many consecutive days, so winning moves (which last
# longer) were systematically overrepresented relative to losing moves (which
# fire once and die). This inflated several cells, most dramatically the
# 52-week-high-proximity adjustment (a 12pp spread that was mostly artifact)
# and the RSI 65-80 / stretch 7-15% cell (was rated tier 1 at ~68%, actually
# ~52% once deduped).
#
# This version is rebuilt on one row per underlying move (first hit of each
# cluster, 15-calendar-day gap = same move) and every cell below was required
# to hold up on an OUT-OF-SAMPLE split: grid built on 2016-2022 data, verified
# against 2023-2026 data it never saw. win_pct below is the OUT-OF-SAMPLE
# (2023-2026) win rate, since that's the honest forward-looking number β
# not the in-sample training number, which runs optimistic.
#
# Grid axes: RSI bucket x SMA20-stretch bucket x ATH-distance zone
# (NEAR = within 10% of 52w high, FAR = 10%+ below).
RSI_BUCKETS = ["<40", "40-50", "50-55", "55-60", "60-65", "65-70", "70-75", ">75"]
STRETCH_BUCKETS = ["<-10", "-10to-5", "-5to-2", "-2to0", "0to+3", "+3to+7", "+7to+15", ">+15"]
def _rsi_bucket(rsi: float) -> str:
if rsi < 40: return "<40"
if rsi < 50: return "40-50"
if rsi < 55: return "50-55"
if rsi < 60: return "55-60"
if rsi < 65: return "60-65"
if rsi < 70: return "65-70"
if rsi < 75: return "70-75"
return ">75"
def _stretch_bucket(stretch: float) -> str:
if stretch < -10: return "<-10"
if stretch < -5: return "-10to-5"
if stretch < -2: return "-5to-2"
if stretch < 0: return "-2to0"
if stretch < 3: return "0to+3"
if stretch < 7: return "+3to+7"
if stretch < 15: return "+7to+15"
return ">+15"
def _ath_zone(pct_from_52w: float) -> str:
# pct_from_52w is negative-or-zero (0 = at ATH, -20 = 20% below)
return "NEAR" if pct_from_52w > -10 else "FAR"
# (rsi_bucket, stretch_bucket, ath_zone) -> (tier, out-of-sample win%, verdict tag)
# Cells absent from this table had insufficient data in one or both periods to
# trust either way β they default to tier 3 with an "unverified" note, NOT to
# "assumed safe". See the rule-out lists for exactly what was excluded and why.
GRID_CALIBRATION = {
# ββ KEEP: stable edge, holds up out-of-sample ββββββββββββββββββββββββββ
("<40","-5to-2","FAR"): (1, 80.7, "KEEP β best cell in the grid, stable across both periods"),
("<40","-10to-5","FAR"): (1, 74.2, "KEEP β strong and stable"),
("70-75","0to+3","NEAR"): (1, 69.0, "KEEP β smaller sample but consistent"),
("40-50","-2to0","NEAR"): (1, 68.0, "KEEP β large sample, stable, best-populated tier-1 cell"),
(">75","+3to+7","NEAR"): (1, 60.0, "KEEP β modest but stable edge"),
("40-50","-2to0","FAR"): (1, 59.8, "KEEP β weaker than the NEAR-ATH version of this cell but still stable"),
("40-50","-5to-2","NEAR"): (1, 57.9, "KEEP β stable, near baseline+"),
("50-55","+3to+7","FAR"): (1, 56.7, "KEEP β stable"),
("60-65","+7to+15","FAR"): (1, 55.9, "KEEP β stable, modest edge"),
("40-50","-5to-2","FAR"): (1, 55.3, "KEEP β stable, modest edge"),
# ββ MARGINAL: near baseline both periods, not clearly harmful or helpful β
("50-55","-5to-2","NEAR"): (2, 58.7, "MARGINAL β near baseline, unclear edge"),
("50-55","0to+3","NEAR"): (2, 58.5, "MARGINAL β near baseline"),
("65-70","0to+3","NEAR"): (2, 58.1, "MARGINAL β near baseline"),
("55-60","0to+3","NEAR"): (2, 55.7, "MARGINAL β near baseline, large sample"),
("40-50","-10to-5","FAR"): (2, 55.2, "MARGINAL β flat both periods"),
("60-65","0to+3","NEAR"): (2, 55.0, "MARGINAL β near baseline"),
(">75","+7to+15","NEAR"): (2, 54.8, "MARGINAL β near baseline"),
# ββ CAUTION: real direction but swings 12-23pp between periods β don't trust the number β
("<40","-5to-2","NEAR"): (2, 75.8, "CAUTION β high but volatile (62% train -> 76% test), small n"),
("40-50","0to+3","FAR"): (2, 73.8, "CAUTION β volatile (55% train -> 74% test)"),
("70-75","+3to+7","NEAR"): (2, 63.3, "CAUTION β volatile (50% train -> 63% test)"),
("<40","<-10","FAR"): (2, 57.5, "CAUTION β was 80% in training, dropped to 58% out-of-sample"),
("70-75",">+15","NEAR"): (2, 56.8, "CAUTION β was 73% in training, dropped to 57% out-of-sample"),
# ββ RULE OUT: below baseline both periods β never worked ββββββββββββββββ
("55-60","-2to0","NEAR"): (3, 53.2, "RULE OUT β below baseline in both periods"),
("60-65","+3to+7","NEAR"): (3, 53.1, "RULE OUT β below baseline in both periods"),
("55-60","-2to0","FAR"): (3, 53.1, "RULE OUT β below baseline in both periods"),
("65-70","+3to+7","NEAR"): (3, 52.9, "RULE OUT β below baseline in both periods"),
("50-55","-5to-2","FAR"): (3, 52.4, "RULE OUT β below baseline in both periods"),
("50-55","-2to0","FAR"): (3, 52.0, "RULE OUT β below baseline in both periods"),
("55-60","+3to+7","FAR"): (3, 51.7, "RULE OUT β below baseline in both periods"),
("55-60","+7to+15","FAR"): (3, 51.2, "RULE OUT β below baseline in both periods"),
("40-50","<-10","FAR"): (3, 50.9, "RULE OUT β below baseline in both periods"),
("55-60","+3to+7","NEAR"): (3, 49.3, "RULE OUT β below baseline in both periods"),
("65-70",">+15","NEAR"): (3, 45.9, "RULE OUT β below baseline in both periods"),
("60-65","0to+3","FAR"): (3, 44.2, "RULE OUT β below baseline in both periods"),
("60-65","+7to+15","NEAR"): (3, 43.9, "RULE OUT β below baseline in both periods"),
("65-70","+3to+7","FAR"): (3, 41.5, "RULE OUT β below baseline in both periods"),
# ββ RULE OUT: looked good in training, reversed out-of-sample (cluster-inflation artifacts) β
("60-65","+3to+7","FAR"): (3, 54.5, "RULE OUT β was 63% in training, fell to 55% (still shown, but do not trust)"),
("50-55","-2to0","NEAR"): (3, 54.2, "RULE OUT β reversed out-of-sample"),
("50-55","0to+3","FAR"): (3, 52.6, "RULE OUT β was 58% in training, fell to 53%"),
("55-60","+7to+15","NEAR"): (3, 50.0, "RULE OUT β reversed out-of-sample"),
("55-60","0to+3","FAR"): (3, 47.5, "RULE OUT β was 61% in training, fell to 48%"),
("70-75","+7to+15","NEAR"): (3, 46.9, "RULE OUT β was 58% in training, fell to 47%"),
("65-70","+7to+15","FAR"): (3, 46.0, "RULE OUT β was 66% in training, fell to 46% β biggest reversal"),
("65-70","+7to+15","NEAR"): (3, 44.1, "RULE OUT β was 56% in training, fell to 44%"),
}
TIER_LABELS = {1: "π Tier 1", 2: "β
Tier 2", 3: "β οΈ Tier 3"}
def classify_tier(rsi: float, stretch: float, pct_from_52w: float) -> dict:
"""
Grid-lookup tier classification (RSI bucket x Stretch bucket x ATH zone),
rebuilt from a properly-deduped (one row per underlying move) backtest and
validated out-of-sample (grid built on 2016-2022, tested blind on 2023-2026).
pct_from_52w: percentage below 52-week high (negative = below, 0 = at ATH).
"""
rsi_b = _rsi_bucket(rsi)
st_b = _stretch_bucket(stretch)
zone = _ath_zone(pct_from_52w)
key = (rsi_b, st_b, zone)
if key in GRID_CALIBRATION:
tier, win_pct, verdict = GRID_CALIBRATION[key]
note = f"{rsi_b} RSI / {st_b}% stretch / {zone} ATH β {verdict}"
else:
tier, win_pct = 3, None
note = (f"{rsi_b} RSI / {st_b}% stretch / {zone} ATH β "
f"UNVERIFIED: insufficient backtest data in one or both periods, "
f"not the same as 'safe'; defaulted to Tier 3")
tier_label = TIER_LABELS.get(tier, "β")
ath_note = ("Within 10% of 52w high" if zone == "NEAR"
else f"{abs(pct_from_52w):.0f}% below 52w high")
return {
"tier": tier,
"tier_label": tier_label,
"estimated_win": win_pct, # None if unverified β don't fabricate a number
"rsi_note": note,
"ath_note": ath_note,
"grid_cell": key,
}
# ββ Cluster / re-trigger detection (informational only β NOT a filter) ββββββββ
#
# A single sustained RRG move can re-satisfy the Weakening+2d-turn condition on
# many consecutive days (it doesn't require a fresh inflection, just 2 more
# days of positive dx/dy). Backtesting confirmed this: winning moves average
# longer "clusters" of re-triggers than losing moves β but that's an outcome
# you only know in hindsight. Testing "wait for a 2nd trigger before buying"
# as a live filter showed NO improvement in forward returns (p=0.81 overall,
# p=0.67 within the best grid cell) β by the time a re-trigger confirms,
# you've already missed whatever return happened between the two triggers.
#
# So this is surfaced for your own situational awareness only (e.g. "have I
# already seen this ticker recently?") β it is deliberately NOT used to filter,
# rank, or adjust the tier/win% shown above.
CLUSTER_GAP_DAYS = 15 # calendar days; matches the backtest's cluster definition
def _signal_history_series(daily_df, benchmark_bm, weekly_window, daily_window, target_quads) -> pd.Series:
"""Full historical daily signal_met boolean series for a ticker (not just today),
used only to look backward for prior re-triggers."""
close = daily_df["Close"].astype(float).dropna()
volume = daily_df["Volume"].astype(float)
bm = benchmark_bm.reindex(close.index, method="ffill").astype(float)
if bm.isna().all() or len(close) < daily_window + 5:
return pd.Series(dtype=bool)
rrg_d = _rrg(close, bm, daily_window)
dx, dy = rrg_d["RS_Ratio"].diff(), rrg_d["RS_Momentum"].diff()
turn_2d_series = (dx > 0) & (dy > 0) & (dx.shift(1) > 0) & (dy.shift(1) > 0)
close_w = close.resample("W-FRI").last().dropna()
bm_w = bm.resample("W-FRI").last().reindex(close_w.index, method="ffill")
if len(close_w) < weekly_window + 2:
return pd.Series(dtype=bool)
rrg_w = _rrg(close_w, bm_w, weekly_window)
quad_w = pd.Series(
[_quadrant(r, m) for r, m in zip(rrg_w["RS_Ratio"], rrg_w["RS_Momentum"])],
index=rrg_w.index,
)
quad_daily = quad_w.reindex(close.index, method="ffill")
dvol_20 = (close * volume).rolling(20).mean()
quality_ok_series = (close >= PRICE_FLOOR) & (dvol_20 >= MIN_DVOL)
signal_met_series = quad_daily.isin(target_quads) & turn_2d_series & quality_ok_series
return signal_met_series.fillna(False)
def _cluster_info(signal_met_series: pd.Series, gap_days: int = CLUSTER_GAP_DAYS) -> dict:
"""Looks backward from the most recent True (assumed 'today') to determine
whether this is a fresh signal or a re-trigger, and which hit number it is."""
hit_dates = signal_met_series.index[signal_met_series.values]
if len(hit_dates) == 0:
return {"is_first_signal": True, "hit_number": 1, "days_since_last_hit": None}
chain = [hit_dates[-1]] # today, then walk backward
for d in hit_dates[-2::-1]:
if (chain[-1] - d).days <= gap_days:
chain.append(d)
else:
break
hit_number = len(chain)
return {
"is_first_signal": hit_number == 1,
"hit_number": hit_number,
"days_since_last_hit": (chain[0] - chain[1]).days if hit_number > 1 else None,
}
# ββ Per-ticker analysis ββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
def analyze_combined(
ticker: str,
daily_df: pd.DataFrame,
benchmark_bm: pd.Series,
weekly_window: int = 13,
daily_window: int = 63,
target_quads: list = None,
) -> dict | None:
"""
Returns signal dict or None if data insufficient.
target_quads: list of weekly quadrants to accept, default ['Weakening'].
"""
if target_quads is None:
target_quads = ["Weakening"]
if daily_df is None or len(daily_df) < max(weekly_window * 5, daily_window, 252) + 5:
return None
close = daily_df["Close"].astype(float).dropna()
volume = daily_df["Volume"].astype(float)
# ββ Quality gate βββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
current_price = float(close.iloc[-1])
dvol_20 = float(close.iloc[-20:].mean() * volume.iloc[-20:].mean()) if len(close) >= 20 else 0.0
quality_ok = current_price >= PRICE_FLOOR and dvol_20 >= MIN_DVOL
large_cap = dvol_20 >= LARGE_CAP_DVOL
# ββ 52-week high proximity ββββββββββββββββββββββββββββββββββββββββββββββββ
if len(close) >= 252:
high_52w = float(close.iloc[-252:].max())
elif len(close) >= 126:
high_52w = float(close.iloc[-126:].max())
else:
high_52w = float(close.max())
pct_from_52w = (current_price / high_52w - 1) * 100 # 0 = at ATH, -20 = 20% below
# ββ Benchmark alignment βββββββββββββββββββββββββββββββββββββββββββββββββββ
bm = benchmark_bm.reindex(close.index, method="ffill").astype(float)
if bm.isna().all():
return None
# ββ Daily RRG βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
rrg_d = _rrg(close, bm, daily_window)
dx = rrg_d["RS_Ratio"].diff()
dy = rrg_d["RS_Momentum"].diff()
turn_2d = bool(
len(dx) >= 2 and
dx.iloc[-1] > 0 and dy.iloc[-1] > 0 and
dx.iloc[-2] > 0 and dy.iloc[-2] > 0
)
# ββ Weekly RRG ββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
close_w = close.resample("W-FRI").last().dropna()
bm_w = bm.resample("W-FRI").last().reindex(close_w.index, method="ffill")
if len(close_w) < weekly_window + 2:
return None
rrg_w = _rrg(close_w, bm_w, weekly_window)
w_ratio = float(rrg_w["RS_Ratio"].iloc[-1])
w_mom = float(rrg_w["RS_Momentum"].iloc[-1])
weekly_quad = _quadrant(w_ratio, w_mom)
prev_weekly_quad = _quadrant(
float(rrg_w["RS_Ratio"].iloc[-2]),
float(rrg_w["RS_Momentum"].iloc[-2]),
)
# ββ RSI + Stretch βββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
sma20 = close.rolling(20).mean()
delta = close.diff()
gain = delta.where(delta > 0, 0).ewm(alpha=1/14, adjust=False).mean()
loss = (-delta.where(delta < 0, 0)).ewm(alpha=1/14, adjust=False).mean()
rsi = float(100 - 100 / (1 + gain.iloc[-1] / (loss.iloc[-1] + 1e-9)))
stretch = float((close.iloc[-1] - sma20.iloc[-1]) / (sma20.iloc[-1] + 1e-9) * 100)
# ββ Signal classification βββββββββββββββββββββββββββββββββββββββββββββββββ
rrg_entry = weekly_quad in target_quads and turn_2d
signal_met = rrg_entry and quality_ok
tier_info = classify_tier(rsi, stretch, pct_from_52w) if signal_met else None
# ββ Cluster / re-trigger info (informational only, see note above) βββββββ
cluster_info = None
if signal_met:
hist_series = _signal_history_series(daily_df, benchmark_bm, weekly_window,
daily_window, target_quads)
if len(hist_series):
cluster_info = _cluster_info(hist_series)
return {
"ticker": ticker,
"price": round(current_price, 2),
"dvol_20m": round(dvol_20 / 1e6, 1),
"quality_ok": quality_ok,
"large_cap": large_cap,
"rsi": round(rsi, 1),
"stretch": round(stretch, 2),
"pct_from_52w": round(pct_from_52w, 1),
"high_52w": round(high_52w, 2),
# weekly RRG
"weekly_quad": weekly_quad,
"prev_weekly_quad": prev_weekly_quad,
"w_ratio": round(w_ratio, 2),
"w_mom": round(w_mom, 2),
# daily RRG
"d_ratio": round(float(rrg_d["RS_Ratio"].iloc[-1]), 2),
"d_mom": round(float(rrg_d["RS_Momentum"].iloc[-1]), 2),
"dx_today": round(float(dx.iloc[-1]), 3),
"dy_today": round(float(dy.iloc[-1]), 3),
"turn_2d": turn_2d,
# signal
"rrg_entry": rrg_entry,
"signal_met": signal_met,
"tier_info": tier_info,
"cluster_info": cluster_info,
# chart data
"daily_trail": rrg_d.tail(10).copy(),
"weekly_trail": rrg_w.tail(7).copy(),
"rrg_w_full": rrg_w,
"rrg_d_full": rrg_d,
"close": close,
"sma20": sma20,
}
# ββ Mini RRG chart (Plotly) βββββββββββββββββββββββββββββββββββββββββββββββββββ
def mini_rrg_chart(result: dict, ticker: str) -> go.Figure:
wt = result["weekly_trail"]
dt = result["daily_trail"]
ti = result.get("tier_info") or {}
fig = go.Figure()
# Quadrant shading
for x0,x1,y0,y1,color,label in [
(100,112,100,112,"rgba(39,174,96,0.08)", "Leading"),
(100,112,88, 100,"rgba(230,126,34,0.08)", "Weakening"),
(88, 100,88, 100,"rgba(231,76,60,0.08)", "Lagging"),
(88, 100,100,112,"rgba(41,128,185,0.08)", "Improving"),
]:
fig.add_shape(type="rect",x0=x0,x1=x1,y0=y0,y1=y1,
fillcolor=color,line_width=0,layer="below")
fig.add_annotation(x=(x0+x1)/2,y=(y0+y1)/2,text=label,
font=dict(size=10,color=color.replace("0.08","0.4")),
showarrow=False)
# Weekly trail
fig.add_trace(go.Scatter(
x=wt["RS_Ratio"].tolist(), y=wt["RS_Momentum"].tolist(),
mode="lines+markers", name="Weekly (6w)",
line=dict(color="#1f77b4",width=2),
marker=dict(size=6,color="#1f77b4"),
text=[str(d.date()) for d in wt.index],
))
# Latest weekly point
wq_color = QUAD_COLORS.get(result["weekly_quad"], "gray")
fig.add_trace(go.Scatter(
x=[wt["RS_Ratio"].iloc[-1]], y=[wt["RS_Momentum"].iloc[-1]],
mode="markers+text",
name=f"Weekly now ({result['weekly_quad']})",
marker=dict(size=14, color=wq_color, symbol="diamond"),
text=[ticker], textposition="top center", textfont=dict(size=11),
))
# Daily trail
fig.add_trace(go.Scatter(
x=dt["RS_Ratio"].tolist(), y=dt["RS_Momentum"].tolist(),
mode="lines+markers", name="Daily (10d)",
line=dict(color="orange",width=1.5,dash="dot"),
marker=dict(size=5,color="orange"),
))
fig.add_hline(y=100,line_dash="dash",line_color="gray",opacity=0.4)
fig.add_vline(x=100,line_dash="dash",line_color="gray",opacity=0.4)
tier_label = ti.get("tier_label","")
est_win = ti.get("estimated_win","")
subtitle = f"{tier_label} β’ Est. {est_win}% win β’ {result['pct_from_52w']:.1f}% from 52w high" if ti else ""
fig.update_layout(
title=f"{ticker} β Dual-TF RRG (weekly: {result['weekly_quad']})<br>"
f"<sup>{subtitle}</sup>",
xaxis_title="RS-Ratio",
yaxis_title="RS-Momentum",
height=390,
template="plotly_white",
legend=dict(orientation="h",yanchor="bottom",y=1.02),
margin=dict(l=40,r=20,t=80,b=40),
)
return fig
# ββ Detail chart (Matplotlib) βββββββββββββββββββββββββββββββββββββββββββββββββ
def detail_chart(result: dict, ticker: str):
close = result["close"]
rrg_d = result["rrg_d_full"]
sma20 = result["sma20"]
dx = rrg_d["RS_Ratio"].diff()
dy = rrg_d["RS_Momentum"].diff()
recent = close.iloc[-120:]
sma_r = sma20.iloc[-120:]
fig, axes = plt.subplots(3,1,figsize=(13,10),sharex=False,
gridspec_kw={"height_ratios":[2.5,1,1]})
# Price
axes[0].plot(recent.index,recent.values,color="black",lw=1.5,label="Price")
axes[0].plot(recent.index,sma_r.values,color="blue",lw=0.9,alpha=0.6,label="SMA20")
if result.get("high_52w"):
axes[0].axhline(result["high_52w"],color="gold",ls="--",lw=1,
alpha=0.7,label=f"52w high ${result['high_52w']:.2f}")
if result["turn_2d"]:
axes[0].axvspan(recent.index[-2],recent.index[-1],
color="limegreen",alpha=0.18,label="2-day turn")
ti = result.get("tier_info") or {}
title_extra = f" | {ti.get('tier_label','')} Est. {ti.get('estimated_win','')}% win {result['pct_from_52w']:.1f}% from 52w high" if ti else ""
axes[0].set_title(f"{ticker}{title_extra}",fontweight="bold",fontsize=12)
axes[0].legend(fontsize=8); axes[0].grid(True,alpha=0.15)
# Daily RS-Ratio
rd = rrg_d["RS_Ratio"].iloc[-120:]
pos_d = dx.iloc[-120:] > 0
axes[1].plot(rd.index,rd.values,color="#1f77b4",lw=1.5)
axes[1].axhline(100,color="gray",ls="--",lw=0.8)
axes[1].fill_between(rd.index,rd.values,100,where=pos_d.values,
alpha=0.2,color="green",label="Gaining vs RSP")
axes[1].fill_between(rd.index,rd.values,100,where=~pos_d.values,
alpha=0.15,color="red",label="Losing vs RSP")
axes[1].set_title("Daily RS-Ratio vs RSP",fontsize=10)
axes[1].legend(fontsize=8); axes[1].grid(True,alpha=0.15)
# Daily RS-Momentum
rm = rrg_d["RS_Momentum"].iloc[-120:]
pos_m = dy.iloc[-120:] > 0
axes[2].plot(rm.index,rm.values,color="purple",lw=1.5)
axes[2].axhline(100,color="gray",ls="--",lw=0.8)
axes[2].fill_between(rm.index,rm.values,100,where=pos_m.values,
alpha=0.2,color="green",label="Momentum rising")
axes[2].fill_between(rm.index,rm.values,100,where=~pos_m.values,
alpha=0.15,color="red",label="Momentum falling")
axes[2].set_title("Daily RS-Momentum (entry = 2 consecutive green bars both panels)",
fontsize=10)
axes[2].legend(fontsize=8); axes[2].grid(True,alpha=0.15)
plt.tight_layout()
st.pyplot(fig)
plt.close(fig)
# ββ Data fetchers βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
@st.cache_data(ttl=4 * 3600, show_spinner=False)
def _fetch_benchmark(bm_ticker: str) -> pd.Series:
raw = yf.download(bm_ticker, period="5y", interval="1d", progress=False)
if isinstance(raw.columns, pd.MultiIndex):
raw.columns = raw.columns.get_level_values(0)
return raw["Close"].dropna().astype(float)
@st.cache_data(ttl=4 * 3600, show_spinner=False)
def _fetch_ticker(ticker: str) -> pd.DataFrame:
raw = yf.download(ticker, period="5y", interval="1d", progress=False)
if isinstance(raw.columns, pd.MultiIndex):
raw.columns = raw.columns.get_level_values(0)
return raw.dropna()
# ββ Full universe scan ββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
@st.cache_data(ttl=12 * 3600, show_spinner=False)
def run_combined_scan(
tickers_tuple: tuple,
benchmark_ticker: str,
weekly_window: int,
daily_window: int,
target_quads: tuple,
min_tier: int,
require_large_cap: bool,
) -> pd.DataFrame:
bm = _fetch_benchmark(benchmark_ticker)
results = []
seen = {}
batch = 10
n_batch = max(1,(len(tickers_tuple)+batch-1)//batch)
prog = st.progress(0, text="Scanningβ¦")
for b, i in enumerate(range(0, len(tickers_tuple), batch)):
batch_t = list(tickers_tuple[i:i+batch])
try:
data = yf.download(batch_t, period="5y", interval="1d",
group_by="ticker", progress=False, threads=True)
except Exception:
prog.progress((b+1)/n_batch); continue
for ticker in batch_t:
try:
raw = data[ticker] if isinstance(data.columns, pd.MultiIndex) else data
df = raw.dropna()
r = analyze_combined(ticker, df, bm, weekly_window,
daily_window, list(target_quads))
if r is None or not r["signal_met"]:
continue
if require_large_cap and not r["large_cap"]:
continue
ti = r["tier_info"]
if ti and ti["tier"] > min_tier:
continue
# dedup
today = pd.Timestamp.now().normalize()
last = seen.get(ticker)
if last and (today - last).days < DEDUP_DAYS:
continue
seen[ticker] = today
ci = r.get("cluster_info")
if ci is None:
cluster_str = "β"
elif ci["is_first_signal"]:
cluster_str = "π First signal"
else:
cluster_str = f"π Repeat (hit #{ci['hit_number']}, {ci['days_since_last_hit']}d ago)"
win_display = f"{ti['estimated_win']:.1f}" if (ti and ti["estimated_win"] is not None) else "unverified"
results.append({
"Tier": ti["tier_label"] if ti else "β",
"Est. Win%": win_display,
"Ticker": ticker,
"Cluster": cluster_str,
"Price": r["price"],
"$Vol 20D (M)": r["dvol_20m"],
"Large Cap": "β
" if r["large_cap"] else "β",
"RSI": r["rsi"],
"Stretch %": r["stretch"],
"% from 52w H": r["pct_from_52w"],
"Wk Quad": QUAD_ICONS.get(r["weekly_quad"],"") + " " + r["weekly_quad"],
"Prev Wk Quad": QUAD_ICONS.get(r["prev_weekly_quad"],"") + " " + r["prev_weekly_quad"],
"W RS-Ratio": r["w_ratio"],
"W RS-Mom": r["w_mom"],
"dx today": r["dx_today"],
"dy today": r["dy_today"],
"Grid note": ti["rsi_note"] if ti else "",
})
except Exception:
continue
prog.progress((b+1)/n_batch, text=f"Batch {b+1}/{n_batch}β¦")
prog.empty()
if not results:
return pd.DataFrame()
df_out = pd.DataFrame(results)
# Sort: tier asc, est win% desc
df_out["_tier_num"] = df_out["Tier"].map({"π Tier 1":1,"β
Tier 2":2,"β οΈ Tier 3":3}).fillna(9)
df_out = (df_out.sort_values(["_tier_num","Est. Win%"], ascending=[True,False])
.drop(columns=["_tier_num"])
.reset_index(drop=True))
return df_out
# ββ MAIN RENDER βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
def render_combined_scanner():
st.title("π― Combined Scanner v2 β RRG + Backtest-Calibrated Tiers")
# ββ Backtest findings callout βββββββββββββββββββββββββββββββββββββββββββββ
with st.expander("π Backtest findings β what actually drives the win rate", expanded=False):
st.markdown("""
**10-year backtest, 3,800 large-cap Weakening + 2d turn entries (2016β2026)**
The blanket "79.6% win rate" figure from v1 required the Bottom Scanner oversold filter
(RSI<35). Today's scan hits typically have RSI 55-65 β the correct win rate for those
is **61-65%**, not 79.6%. This version shows the calibrated figure per entry.
| Configuration | Win% | Mean 63D | xs_t | Notes |
|---|---|---|---|---|
| RSI <40, any stretch | **75.3%** | +10.2% | β | Strongest RSI bucket |
| Stretch <-10% | **80%** | +20.1% | β | n=50 only, treat carefully |
| Stretch -10 to -5% | **71%** | +8.0% | β | Extended below SMA20 |
| RSI 40-55 + stretch 3-10% | **67.1%** | +8.3% | -2.92 | Strong bucket |
| RSI 55-65 + stretch >10% | **67.6%** | +11.4% | -1.23 | High momentum |
| RSI 65-80 + stretch >10% | **67.9%** | +13.1% | **+1.54** | β
β
Only positive xs_t |
| RSI 55-65 + stretch 3-10% | **64.7%** | +7.7% | **-0.48** | β
Best alpha quality |
| Within 5% of 52w high | 57.3% | +2.5% | β | β οΈ Near-random, avoid |
| RSI 65-75 + stretch 3-10% | 53.9% | +1.2% | -5.04 | β Weakest bucket |
**52-week high proximity is the biggest single filter:**
- Within 5% of ATH β 57% win (near-random regardless of RSI/stretch)
- 10-20% below ATH β 66% win
- 20-35% below ATH β 69% win β sweet spot
**Default benchmark changed from SPY β RSP (equal-weight S&P 500).**
SPY's 35% mega-cap weighting inflates relative strength for non-tech names,
producing Weakening signals that are actually just "not as good as Nvidia."
RSP asks the correct question: is this stock beating the average company?
""")
# ββ Sidebar βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
with st.sidebar:
st.markdown("---")
st.header("Combined Scanner v2")
benchmark_sym = st.text_input("Benchmark", value="RSP", key="cs2_bm").strip().upper()
st.caption("RSP = equal-weight S&P 500 β recommended over SPY (see backtest notes)")
weekly_win = st.slider("Weekly z-score window (weeks)", 8, 26, 13, 1, key="cs2_ww")
daily_win = st.slider("Daily z-score window (days)", 21, 126, 63, 1, key="cs2_dw")
st.markdown("**Quadrant filter**")
weak_only = st.checkbox("Weakening only (excludes Lagging)", value=True, key="cs2_wk")
target_quads = ("Weakening",) if weak_only else ("Weakening","Lagging")
st.markdown("**Quality filters**")
min_tier = st.radio("Minimum tier", [1, 2, 3],
format_func=lambda x: {1:"π Tier 1 only (β₯67% est.)",
2:"β
Tier 1+2 (β₯60%)",
3:"β οΈ All tiers"}[x],
index=1, key="cs2_tier")
require_lc = st.checkbox("Large cap only (β₯$50M 20D $vol)", value=True, key="cs2_lc")
st.caption("Large-cap filter strongly recommended: all-cap xs_t = -5 vs large-cap xs_t closer to 0")
# ββ Single ticker βββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
st.subheader("Analyze a single ticker")
col_a, col_b = st.columns([2,1])
single = col_a.text_input("Ticker", value="AAPL", key="cs2_single").strip().upper()
go_btn = col_b.button("Analyze", key="cs2_analyze")
if go_btn and single:
with st.spinner(f"Fetching {single} + {benchmark_sym}β¦"):
bm = _fetch_benchmark(benchmark_sym)
df = _fetch_ticker(single)
r = analyze_combined(single, df, bm, weekly_win, daily_win, list(target_quads))
if r is None:
st.error("Not enough data (need β₯5y daily bars for 52w high calculation).")
else:
ti = r.get("tier_info")
# Metric row
c1,c2,c3,c4,c5,c6 = st.columns(6)
c1.metric("Price", f"${r['price']}")
c2.metric("RSI", f"{r['rsi']}")
c3.metric("Stretch", f"{r['stretch']:.1f}%")
c4.metric("vs 52w High", f"{r['pct_from_52w']:.1f}%")
c5.metric("Weekly Quad", QUAD_ICONS.get(r["weekly_quad"],"") + " " + r["weekly_quad"])
c6.metric("2-day Turn", "β
YES" if r["turn_2d"] else "β NO",
delta=f"dx={r['dx_today']:+.2f} dy={r['dy_today']:+.2f}")
if not r["quality_ok"]:
st.warning(f"β οΈ Quality gate failed β price ${r['price']} or liquidity "
f"(${r['dvol_20m']:.1f}M 20D $vol < ${MIN_DVOL/1e6:.0f}M min)")
if r["signal_met"] and ti:
col = {"π Tier 1":"success","β
Tier 2":"info","β οΈ Tier 3":"warning"}.get(ti["tier_label"],"info")
win_str = f"**{ti['estimated_win']:.1f}% out-of-sample win rate**" if ti["estimated_win"] is not None \
else "**unverified β insufficient backtest data for this exact grid cell**"
msg = f"{ti['tier_label']} | {win_str}\n\nGrid cell: {ti['rsi_note']}"
ci = r.get("cluster_info")
if ci:
if ci["is_first_signal"]:
msg += "\n\nπ First signal β no matching hit in the prior 15 days."
else:
msg += (f"\n\nπ Re-trigger β hit #{ci['hit_number']} of an ongoing cluster "
f"({ci['days_since_last_hit']}d since the last hit). "
f"For your awareness only β backtesting found waiting for a re-trigger "
f"does not improve forward returns, so this is not treated as a stronger signal.")
if col == "success":
st.success(msg)
elif col == "warning":
st.warning(msg)
else:
st.info(msg)
elif not r["rrg_entry"]:
reasons = []
if r["weekly_quad"] not in list(target_quads):
reasons.append(f"weekly quad is {r['weekly_quad']} (need {'/'.join(target_quads)})")
if not r["turn_2d"]:
reasons.append("no 2-day positive daily turn")
st.info("Signal not active: " + " | ".join(reasons))
elif not r["quality_ok"]:
st.warning("Signal fires but quality gate failed β treat with caution.")
st.plotly_chart(mini_rrg_chart(r, single), use_container_width=True)
with st.expander("π Detailed price + daily RRG panels"):
detail_chart(r, single)
# ββ Full scan βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
st.markdown("---")
st.subheader("Full universe scan")
universe = get_clean_premarket_universe()
quad_str = " or ".join(target_quads)
tier_str = {1:"Tier 1 only",2:"Tier 1+2",3:"All tiers"}.get(min_tier,"")
lc_str = "large-cap " if require_lc else ""
st.caption(
f"Universe: {len(universe)} tickers | Benchmark: {benchmark_sym} | "
f"Quadrant: {quad_str} | Filter: {lc_str}{tier_str} | Cached 12h"
)
if st.button("π Run Scan", key="cs2_scan"):
if not universe:
st.error("No tickers loaded.")
else:
df_r = run_combined_scan(
tuple(universe), benchmark_sym, weekly_win, daily_win,
target_quads, min_tier, require_lc,
)
st.session_state["cs2_results"] = df_r
if "cs2_results" in st.session_state:
df_r = st.session_state["cs2_results"]
if df_r.empty:
st.info("No tickers cleared all filters. Try relaxing Tier or large-cap settings.")
else:
st.success(f"{len(df_r)} ticker(s) β sorted by Tier then Est. Win%")
# Highlight tier 1 rows
def row_style(row):
if "Tier 1" in str(row.get("Tier","")):
return ["background-color: #d4edda"] * len(row)
elif "Tier 3" in str(row.get("Tier","")):
return ["background-color: #f8d7da"] * len(row)
return [""] * len(row)
display_cols = ["Tier","Est. Win%","Ticker","Price","RSI","Stretch %",
"% from 52w H","Wk Quad","Prev Wk Quad","Large Cap",
"$Vol 20D (M)","RSI note","ATH note"]
display_cols = [c for c in display_cols if c in df_r.columns]
st.dataframe(
df_r[display_cols].style.apply(row_style, axis=1),
use_container_width=True, hide_index=True,
)
pick = st.selectbox("View chart for:", df_r["Ticker"].tolist(), key="cs2_pick")
if pick:
with st.spinner(f"Fetching {pick}β¦"):
bm = _fetch_benchmark(benchmark_sym)
dfp = _fetch_ticker(pick)
rp = analyze_combined(pick, dfp, bm, weekly_win, daily_win, list(target_quads))
if rp:
ti = rp.get("tier_info")
if ti:
st.info(f"{ti['tier_label']} β’ Est. {ti['estimated_win']:.1f}% win\n\n"
f"{ti['rsi_note']}\n\n{ti['ath_note']}")
st.plotly_chart(mini_rrg_chart(rp, pick), use_container_width=True)
with st.expander("π Detailed panels"):
detail_chart(rp, pick)
# ββ Workflow guide ββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
st.markdown("---")
with st.expander("π‘ How to read the tiers + DCA workflow"):
st.markdown(f"""
**Tier 1 π** β Est. win rate β₯67%. Best historical outcomes within the signal.
Look for RSI<40, deep negative stretch, or RSI 65-80 with stretch >10%. Position is
10-35% below its 52-week high. These are the clearest mean-reversion setups.
**Tier 2 β
** β Est. win rate 60-66%. Solid signal, majority of scan hits will be here.
RSI 55-65 + stretch 3-10% is the sweet spot for alpha quality (xs_t closest to 0).
The six positions in the live test portfolio are Tier 2 entries.
**Tier 3 β οΈ** β Est. win rate <61%, near-random. RSI 65-75 with moderate stretch, or
within 5% of the 52-week high. The scanner flags these so you can see them but they
carry no statistically meaningful edge over just buying the benchmark.
**DCA workflow:**
1. Watchlist conviction first (macro thesis, fundamentals)
2. Combined Scanner fires β Tranche 1 (β
position) if Tier 1 or 2
3. If signal re-fires within 63 days β Tranche 2 (lowers average cost)
4. Peak Detector >50% risk β hold / re-evaluate exit
5. RRG rolls out of Weakening/Leading back to Lagging β consider exit
**Stop losses: the backtest says don't.** 91% of entries (including winners) touch -5%
intraperiod. Mean intraperiod drawdown is -14.3% even for winning trades.
The edge is mean-reversion β cutting at -5% destroys it.
""")
# =========================================================
# TAB 7: CATCH THE KNIFE (Large-Cap + EDGAR Pre-Signal Scan)
# =========================================================
#
# Combines three things developed and backtested separately:
# 1. The RSI+SMA20-stretch "catch the knife" signal (catch_the_knife_code.txt)
# 2. A large-cap liquidity gate, using TRAILING rolling dollar volume at each
# date -- not a static snapshot -- since a static snapshot let names that
# only became liquid recently (e.g. hype-driven small caps) qualify for
# signals years earlier. Backtesting (10-ticker sample, then full ~2,300-
# ticker universe) found this filter roughly halves tail-loss exposure
# (>-15% 3-month loss) vs an unfiltered universe: ~18-19% -> ~11%.
# 3. An EDGAR pre-signal filing scan, run automatically on any hit, checking
# for company disclosures (buybacks, debt paydown, dilution, credit
# agreement changes) in the ~15 trading days before the signal date.
# Validated against FICO/ORCL/FN/ARES/KOS/NLOP: real catalysts do show up
# this way, but this is text/keyword-based, not numeric -- it can't tell
# debt ISSUANCE from debt REPAYMENT, or judge whether a dollar figure is
# material at a given company's scale. Treat hits as "worth reading the
# actual filing," not as a validated trading signal on their own.
import re
import requests
# ---- Knife scanner constants ----
KNIFE_RSI_THRESHOLD = 45
KNIFE_STRETCH_THRESHOLD = -2.0
KNIFE_STRICT_WINDOW = 0.5
KNIFE_ADV_LOOKBACK = 60 # trailing days for the large-cap/liquidity proxy
KNIFE_FORWARD_1D = 1
KNIFE_FORWARD_5D = 5
KNIFE_FORWARD_3M = 63
KNIFE_MIN_HISTORY_DAYS = 300
KNIFE_DEFAULT_ADV = 100_000_000 # $100M/day -- matches the backtest-validated threshold
def _knife_rsi(series: pd.Series, period: int = 14) -> pd.Series:
delta = series.diff()
gain = delta.where(delta > 0, 0).ewm(alpha=1/period, adjust=False).mean()
loss = (-delta.where(delta < 0, 0)).ewm(alpha=1/period, adjust=False).mean()
rs = gain / (loss + 1e-9)
return 100 - (100 / (1 + rs))
def _knife_strict_stats(close_arr, rsi_arr, sma_arr, current_stretch,
strict_window=KNIFE_STRICT_WINDOW):
"""Historical analog matching: same logic as the original catch-the-knife
script's get_strict_stats_with_dates, returns None if <10 historical
matches (insufficient sample to trust the estimated win rate)."""
stretches = (close_arr - sma_arr) / sma_arr * 100
ret_1d = (pd.Series(close_arr).shift(-KNIFE_FORWARD_1D).values - close_arr) / close_arr
ret_5d = (pd.Series(close_arr).shift(-KNIFE_FORWARD_5D).values - close_arr) / close_arr
ret_3m = (pd.Series(close_arr).shift(-KNIFE_FORWARD_3M).values - close_arr) / close_arr
mask = ((stretches >= current_stretch - strict_window) &
(stretches <= current_stretch + strict_window) &
(rsi_arr < KNIFE_RSI_THRESHOLD))
valid_mask = mask & (~np.isnan(ret_3m))
n = int(valid_mask.sum())
if n < 10:
return None
return {
"count": n,
"avg_1d": float(np.mean(ret_1d[valid_mask])) * 100,
"avg_5d": float(np.mean(ret_5d[valid_mask])) * 100,
"avg_3m": float(np.mean(ret_3m[valid_mask])) * 100,
"win_3m": float((ret_3m[valid_mask] > 0).mean()) * 100,
}
def analyze_knife(ticker: str, df: pd.DataFrame, adv_threshold: float,
win_rate_min: float, require_triple_positive: bool):
"""Returns a hit dict if the ticker clears the knife signal + large-cap
gate + win-rate/triple-positive filters, else None."""
if df is None or len(df) < KNIFE_MIN_HISTORY_DAYS:
return None
close = df["Close"].astype(float).dropna()
volume = df["Volume"].astype(float)
if len(close) < KNIFE_MIN_HISTORY_DAYS:
return None
rsi = _knife_rsi(close)
sma20 = close.rolling(20).mean()
stretch_series = (close - sma20) / sma20 * 100
# Large-cap proxy: TRAILING rolling median dollar volume, evaluated at
# today's date specifically -- not a static/most-recent snapshot, to
# avoid crediting a ticker's current liquidity to a signal fired years
# ago when it may have been illiquid.
dollar_vol = close * volume
rolling_adv = dollar_vol.rolling(KNIFE_ADV_LOOKBACK).median()
current_rsi = float(rsi.iloc[-1])
current_stretch = float(stretch_series.iloc[-1])
current_adv = float(rolling_adv.iloc[-1]) if pd.notna(rolling_adv.iloc[-1]) else 0.0
large_cap_ok = current_adv >= adv_threshold
signal_met = (current_rsi < KNIFE_RSI_THRESHOLD and
current_stretch < KNIFE_STRETCH_THRESHOLD and
large_cap_ok)
if not signal_met:
return None
stats = _knife_strict_stats(close.values, rsi.values, sma20.values, current_stretch)
if stats is None:
return None
triple_positive = stats["avg_1d"] > 0 and stats["avg_5d"] > 0 and stats["avg_3m"] > 0
if require_triple_positive and not triple_positive:
return None
if stats["win_3m"] < win_rate_min:
return None
return {
"ticker": ticker,
"price": round(float(close.iloc[-1]), 2),
"rsi": round(current_rsi, 1),
"stretch": round(current_stretch, 2),
"adv_millions": round(current_adv / 1e6, 1),
"avg_1d": round(stats["avg_1d"], 2),
"avg_5d": round(stats["avg_5d"], 2),
"avg_3m": round(stats["avg_3m"], 2),
"win_3m": round(stats["win_3m"], 1),
"match_count": stats["count"],
"triple_positive": triple_positive,
}
@st.cache_data(ttl=12 * 3600, show_spinner=False)
def run_knife_scan(tickers_tuple: tuple, adv_threshold: float,
win_rate_min: float, require_triple_positive: bool) -> pd.DataFrame:
results = []
batch = 10
n_batch = max(1, (len(tickers_tuple) + batch - 1) // batch)
prog = st.progress(0, text="Scanning for knife signalsβ¦")
for b, i in enumerate(range(0, len(tickers_tuple), batch)):
batch_t = list(tickers_tuple[i:i + batch])
try:
data = yf.download(batch_t, period="5y", interval="1d",
group_by="ticker", progress=False, threads=True)
except Exception:
prog.progress((b + 1) / n_batch)
continue
for ticker in batch_t:
try:
raw = data[ticker] if isinstance(data.columns, pd.MultiIndex) else data
df = raw.dropna()
hit = analyze_knife(ticker, df, adv_threshold, win_rate_min, require_triple_positive)
if hit:
results.append(hit)
except Exception:
continue
prog.progress((b + 1) / n_batch, text=f"Scanning for knife signals⦠batch {b+1}/{n_batch}")
prog.empty()
if not results:
return pd.DataFrame()
return pd.DataFrame(results).sort_values("win_3m", ascending=False).reset_index(drop=True)
# ---------------------------------------------------------
# EDGAR pre-signal scan (ported from edgar_scanner.py)
# ---------------------------------------------------------
SEC_USER_AGENT_DEFAULT = os.environ.get("SEC_USER_AGENT", "")
EDGAR_LOOKBACK_TRADING_DAYS = 15
EDGAR_POSITIVE_KEYWORDS = [
"accelerated share repurchase", "new share repurchase authorization",
"increased share repurchase authorization", "increased our share repurchase",
"special cash dividend", "increased the quarterly dividend", "raised its quarterly dividend",
"completed the sale of", "definitive agreement to sell", "closed the sale of",
"used the proceeds to repay", "used the net proceeds to repay",
"extended and upsized", "upsized the credit facility", "amended and restated credit agreement",
"record backlog", "record revenue", "raised its full-year guidance", "raised full year guidance",
]
EDGAR_NEGATIVE_KEYWORDS = [
"goodwill impairment charge", "non-cash impairment charge", "recorded an impairment charge",
"material weakness in internal control", "identified a material weakness",
"restated its previously issued financial statements", "restatement of previously issued",
"notice of default", "in default under", "event of default has occurred",
"going concern", "substantial doubt about", "unable to timely file",
"delisting notice", "notice of noncompliance with", "receives sec subpoena",
"sec investigation", "dilutive issuance of", "registered direct offering",
"private placement of convertible notes", "at-the-market equity offering program",
"reduced its guidance", "lowered its guidance", "withdrew its guidance",
]
EDGAR_OWNERSHIP_FORM_TYPES = {"SCHEDULE 13D", "SCHEDULE 13D/A", "SCHEDULE 13G", "SCHEDULE 13G/A"}
AGREEMENT_EXHIBIT_PATTERN = re.compile(r'ex-?10[^0-9]|ex-?4[^0-9]|ex-?2[^0-9]|dex10|dex4|dex2', re.IGNORECASE)
def _edgar_headers():
ua = st.session_state.get("edgar_user_agent") or SEC_USER_AGENT_DEFAULT
return {"User-Agent": ua} if ua else {"User-Agent": "unset-contact@example.com"}
def _clean_html_text(raw_html: str) -> str:
text = re.sub(r'<[^>]+>', ' ', raw_html)
text = text.replace(' ', ' ').replace('&', '&')
text = re.sub(r'&#\d+;', ' ', text)
text = re.sub(r'\s+', ' ', text)
return text.lower()
@st.cache_data(ttl=24 * 3600, show_spinner=False)
def edgar_get_cik(ticker: str, _headers_key: str) -> str | None:
try:
r = requests.get("https://www.sec.gov/files/company_tickers.json",
headers=_edgar_headers(), timeout=15)
r.raise_for_status()
data = r.json()
for entry in data.values():
if entry["ticker"].upper() == ticker.upper():
return str(entry["cik_str"]).zfill(10)
except Exception:
return None
return None
def edgar_get_filings(cik: str, start_date, end_date) -> pd.DataFrame:
try:
r = requests.get(f"https://data.sec.gov/submissions/CIK{cik}.json",
headers=_edgar_headers(), timeout=15)
r.raise_for_status()
recent = r.json()["filings"]["recent"]
df = pd.DataFrame({
"form": recent["form"],
"filingDate": pd.to_datetime(recent["filingDate"]),
"accessionNumber": recent["accessionNumber"],
"primaryDocument": recent["primaryDocument"],
})
mask = (df["filingDate"] >= start_date) & (df["filingDate"] <= end_date)
return df[mask].sort_values("filingDate")
except Exception:
return pd.DataFrame()
def edgar_fetch_index(cik: str, accession_number: str) -> list:
accession_nodash = accession_number.replace("-", "")
cik_int = str(int(cik))
url = f"https://www.sec.gov/Archives/edgar/data/{cik_int}/{accession_nodash}/index.json"
try:
r = requests.get(url, headers=_edgar_headers(), timeout=15)
if r.status_code != 200:
return []
items = r.json().get("directory", {}).get("item", [])
return [it["name"] for it in items if it["name"].lower().endswith((".htm", ".html"))]
except Exception:
return []
def edgar_fetch_text(cik: str, accession_number: str, doc_name: str, max_chars=400000) -> str:
accession_nodash = accession_number.replace("-", "")
cik_int = str(int(cik))
url = f"https://www.sec.gov/Archives/edgar/data/{cik_int}/{accession_nodash}/{doc_name}"
try:
r = requests.get(url, headers=_edgar_headers(), timeout=15)
if r.status_code != 200 or not r.text.strip():
return ""
return _clean_html_text(r.text)[:max_chars]
except Exception:
return ""
def edgar_classify_text(text: str, source_label: str = ""):
is_legal_agreement = bool(AGREEMENT_EXHIBIT_PATTERN.search(source_label))
pos_hits = [kw for kw in EDGAR_POSITIVE_KEYWORDS if kw in text]
neg_hits = [kw for kw in EDGAR_NEGATIVE_KEYWORDS if kw in text]
# Negation check: impairment/goodwill phrases routinely sit inside
# sentences stating NO charge occurred (standard annual testing language).
def negated_nearby(phrase, window=40):
for m in re.finditer(re.escape(phrase), text):
start = max(0, m.start() - window)
preceding = text[start:m.start()]
if any(neg in preceding for neg in
("did not result in", "did not recognize", "no ", "not recognize any")):
return True
return False
neg_hits = [kw for kw in neg_hits
if not (('impairment charge' in kw or 'goodwill' in kw) and negated_nearby(kw))]
if is_legal_agreement:
# Material contracts structurally define "Event of Default" regardless
# of whether one occurred -- drop default-related hits from these.
neg_hits = [h for h in neg_hits if 'default' not in h]
def near(term_a, term_b, window=60):
for m in re.finditer(re.escape(term_a), text):
start = max(0, m.start() - window)
end = min(len(text), m.end() + window)
if term_b in text[start:end]:
return True
return False
if near("repurchas", "$") or near("buyback", "$"):
pos_hits.append("repurchase near dollar amount")
if (near("sold", "debt") or near("sale of", "debt") or near("divest", "debt")
or near("proceeds", "repay") or near("proceeds", "reduc")):
pos_hits.append("asset sale / debt reduction (proximity match)")
if near("default", "occurred") and not is_legal_agreement \
and "no event of default has occurred" not in text \
and "not in default" not in text:
neg_hits.append("default (proximity match -- verify manually, still noisy)")
return sorted(set(pos_hits)), sorted(set(neg_hits))
def run_edgar_scan_for_ticker(ticker: str, signal_date, lookback_days: int = EDGAR_LOOKBACK_TRADING_DAYS) -> dict:
"""Runs the full pre-signal EDGAR scan for one ticker. Requires internet
access (works on HF Spaces / Colab; will fail gracefully if SEC blocks
an unset User-Agent -- set one in the sidebar first)."""
signal_ts = pd.Timestamp(signal_date)
start_date = signal_ts - pd.Timedelta(days=int(lookback_days * 1.5))
cik = edgar_get_cik(ticker, _headers_key=st.session_state.get("edgar_user_agent", ""))
if cik is None:
return {"ticker": ticker, "status": "CIK not found", "n_filings": 0,
"form_types": [], "positive_hits": [], "negative_hits": [], "ownership_filings": []}
filings = edgar_get_filings(cik, start_date, signal_ts)
result = {
"ticker": ticker, "status": "ok",
"n_filings": len(filings),
"form_types": filings["form"].tolist() if not filings.empty else [],
"positive_hits": [], "negative_hits": [],
"ownership_filings": (filings[filings["form"].isin(EDGAR_OWNERSHIP_FORM_TYPES)]["form"].tolist()
if not filings.empty else []),
}
if filings.empty:
return result
equity_shelf = filings[filings["form"].isin({"424B5", "424B2", "S-3", "S-3/A", "S-1"})]
if not equity_shelf.empty:
result["negative_hits"].append(
f"equity shelf/ATM filing present ({', '.join(equity_shelf['form'].tolist())}) "
f"-- dilution capacity registered or drawn on"
)
for _, row in filings.iterrows():
form = row["form"]
if form in ("8-K", "8-K/A"):
doc_names = edgar_fetch_index(cik, row["accessionNumber"]) or [row["primaryDocument"]]
for doc_name in doc_names:
text = edgar_fetch_text(cik, row["accessionNumber"], doc_name)
pos, neg = edgar_classify_text(text, source_label=f"{ticker}/{doc_name}")
result["positive_hits"].extend(pos)
result["negative_hits"].extend(neg)
time.sleep(0.15)
elif form in ("10-K", "10-Q", "424B5", "424B2", "S-3"):
text = edgar_fetch_text(cik, row["accessionNumber"], row["primaryDocument"])
pos, neg = edgar_classify_text(text, source_label=f"{ticker}/{row['primaryDocument']}")
result["positive_hits"].extend(pos)
result["negative_hits"].extend(neg)
time.sleep(0.15)
result["positive_hits"] = sorted(set(result["positive_hits"]))
result["negative_hits"] = sorted(set(result["negative_hits"]))
return result
# ---------------------------------------------------------
# Render
# ---------------------------------------------------------
def render_catch_the_knife():
st.title("πͺ Catch the Knife β Large-Cap Filter + EDGAR Pre-Signal Scan")
with st.expander("π What this tab does and what it's validated on", expanded=False):
st.markdown("""
**Signal:** RSI(14) < 45 AND price < -2% below its 20-day SMA (the "catch the knife"
mean-reversion setup), gated by historical analog matching (needs β₯10 similar
historical setups on this ticker before showing a win rate).
**Large-cap filter:** TRAILING 60-day median dollar volume β₯ threshold, computed
at the signal date itself (not a static snapshot). Backtested on the full
~2,300-ticker universe: cuts tail-loss risk (3-month losses worse than -15%)
roughly in half vs. an unfiltered universe (~18-19% β ~11%), while modestly
improving median return and win rate.
**EDGAR scan:** runs automatically on every hit β pulls 8-K/10-K/10-Q/424B5/
13D/13G filings from the ~15 trading days before today, and flags buyback,
debt-paydown, credit-agreement, dilution, and ownership-change activity.
This is text/keyword-based: it can tell you a filing *exists* and roughly
what topic it touches, but it cannot judge direction (debt issued vs. repaid)
or materiality (a $93M buyback reads the same as a $1.5B one). **Treat any
hit β positive or negative β as "go read the actual filing," not as a
validated signal on its own.** (See the ORCL case: this exact scanner
initially misread aggressive debt issuance as "debt reduction" because both
mention "proceeds" near "repay.")
""")
with st.sidebar:
st.markdown("---")
st.header("Catch the Knife")
adv_threshold_m = st.slider("Large-cap filter: min $ volume/day (rolling 60d median, $M)",
10, 500, int(KNIFE_DEFAULT_ADV / 1e6), 10, key="knife_adv")
adv_threshold = adv_threshold_m * 1_000_000
win_rate_min = st.slider("Min historical 3-month win rate (%)", 50, 90, 70, 1, key="knife_winmin")
require_triple_positive = st.checkbox("Require triple-positive (1D, 5D, 3M all historically positive)",
value=True, key="knife_triplepos")
st.markdown("---")
st.subheader("EDGAR scan settings")
st.caption("SEC requires a real contact in the User-Agent header or it will reject requests.")
edgar_ua = st.text_input("SEC User-Agent (Your Name your_email@domain.com)",
value=SEC_USER_AGENT_DEFAULT, key="edgar_user_agent")
run_edgar_auto = st.checkbox("Automatically run EDGAR scan on every hit", value=True, key="knife_edgar_auto")
if not edgar_ua and run_edgar_auto:
st.warning("β οΈ No SEC User-Agent set β EDGAR requests will likely be rejected (403). "
"Set one in the sidebar (e.g. 'Robert research@yourdomain.com').")
universe = get_clean_premarket_universe()
st.caption(f"Universe: {len(universe)} tickers from Trebor1009/clean-premarket-data "
f"| Large-cap threshold: ${adv_threshold_m}M/day | Min win rate: {win_rate_min}%")
if st.button("π Run Full Universe Scan", key="knife_scan_btn"):
if not universe:
st.error("No tickers loaded β check the dataset is reachable.")
else:
df_hits = run_knife_scan(tuple(universe), adv_threshold, win_rate_min, require_triple_positive)
st.session_state["knife_hits"] = df_hits
st.session_state["knife_edgar_results"] = {} # reset EDGAR cache for new scan
if "knife_hits" in st.session_state:
df_hits = st.session_state["knife_hits"]
if df_hits.empty:
st.info("No tickers cleared the knife signal + large-cap + win-rate filters. "
"Try lowering the win-rate minimum or the large-cap threshold.")
else:
st.success(f"{len(df_hits)} hit(s) found. Running EDGAR pre-signal scan on eachβ¦"
if run_edgar_auto else f"{len(df_hits)} hit(s) found.")
edgar_results = st.session_state.get("knife_edgar_results", {})
if run_edgar_auto:
today = pd.Timestamp.now().normalize()
remaining = [t for t in df_hits["ticker"] if t not in edgar_results]
if remaining:
prog = st.progress(0, text="Running EDGAR scansβ¦")
for i, ticker in enumerate(remaining):
edgar_results[ticker] = run_edgar_scan_for_ticker(ticker, today)
prog.progress((i + 1) / len(remaining), text=f"EDGAR scan: {ticker} ({i+1}/{len(remaining)})")
prog.empty()
st.session_state["knife_edgar_results"] = edgar_results
display_rows = []
for _, row in df_hits.iterrows():
er = edgar_results.get(row["ticker"], {})
display_rows.append({
"Ticker": row["ticker"],
"Price": row["price"],
"RSI": row["rsi"],
"Stretch %": row["stretch"],
"$Vol 60D (M)": row["adv_millions"],
"Avg 1D%": row["avg_1d"],
"Avg 5D%": row["avg_5d"],
"Avg 3M%": row["avg_3m"],
"3M Win%": row["win_3m"],
"Matches": row["match_count"],
"EDGAR +": len(er.get("positive_hits", [])) if er else "β",
"EDGAR -": len(er.get("negative_hits", [])) if er else "β",
"Ownership filings": ", ".join(er.get("ownership_filings", [])) if er else "β",
})
display_df = pd.DataFrame(display_rows)
def _row_style(row):
neg = row.get("EDGAR -")
pos = row.get("EDGAR +")
if isinstance(neg, int) and neg > 0 and (not isinstance(pos, int) or pos == 0):
return ["background-color: #f8d7da"] * len(row)
if isinstance(pos, int) and pos > 0 and (not isinstance(neg, int) or neg == 0):
return ["background-color: #d4edda"] * len(row)
return [""] * len(row)
st.dataframe(display_df.style.apply(_row_style, axis=1),
use_container_width=True, hide_index=True)
st.markdown("---")
st.subheader("EDGAR detail for a specific hit")
pick = st.selectbox("Ticker", df_hits["ticker"].tolist(), key="knife_edgar_pick")
if pick:
er = edgar_results.get(pick)
if er is None:
if st.button(f"Run EDGAR scan for {pick}", key="knife_edgar_manual"):
er = run_edgar_scan_for_ticker(pick, pd.Timestamp.now().normalize())
edgar_results[pick] = er
st.session_state["knife_edgar_results"] = edgar_results
if er:
st.write(f"**Filings found ({er['n_filings']}):** {', '.join(er['form_types']) or 'β'}")
if er["positive_hits"]:
st.success("Positive hits: " + ", ".join(er["positive_hits"]))
if er["negative_hits"]:
st.error("Negative hits: " + ", ".join(er["negative_hits"]))
if er["ownership_filings"]:
st.info("Ownership-change filings (13D/13G): " + ", ".join(er["ownership_filings"]))
if not er["positive_hits"] and not er["negative_hits"] and not er["ownership_filings"]:
st.caption("No keyword or structural hits β either no filings, or nothing in "
"our keyword/proximity list matched. Worth a manual check before "
"treating silence as 'all clear.'")
# =========================================================
# TAB 8: CYCLE PROJECTOR (manual resolution)
# =========================================================
#
# Nested Daily/Weekly/Yearly pivot-low cycle detection, phase-modulated
# weekly model + transition-adaptive yearly model. Yearly cycle length is
# set MANUALLY per ticker (e.g. BTC needs ~6y or the 2018 low gets merged
# into the COVID low) -- automatic cycle-length discovery was tried and
# found to underperform manual tuning on every tested ticker, so this tab
# keeps the resolution as a direct user input rather than guessing it.
#
def cp_find_cycle_lows(price, expected_cycle_days, prominence_pct=0.05, trend_window_days=None):
price = price.dropna()
log_price = np.log(price)
# bars per calendar day: ~1.0 for crypto, ~0.69 for equities
bpd = len(price) / max((price.index[-1] - price.index[0]).days, 1)
if trend_window_days is None:
trend_window_days = expected_cycle_days * 2
trend_window = max(int(trend_window_days * bpd), 5)
trend = log_price.rolling(trend_window, center=True,
min_periods=max(int(trend_window * 0.25), 30)).mean()
detrended = (log_price - trend).dropna()
min_spacing = max(int(expected_cycle_days * 0.5 * bpd), 20)
troughs, _ = find_peaks(-detrended.values, distance=min_spacing, prominence=prominence_pct)
return detrended.index[troughs]
def cp_cycle_stats(lows: pd.DatetimeIndex) -> dict:
if len(lows) < 2:
return {"mean_days": np.nan, "std_days": np.nan, "n_cycles": 0, "spacing_days": []}
spacing = np.diff(lows).astype("timedelta64[D]").astype(int)
return {"mean_days": spacing.mean(), "std_days": spacing.std(),
"n_cycles": len(spacing), "spacing_days": spacing.tolist()}
def cp_project_next_low(lows: pd.DatetimeIndex, stats: dict) -> dict:
if len(lows) == 0 or np.isnan(stats["mean_days"]):
return {}
last_low = lows[-1]
eta = last_low + pd.Timedelta(days=stats["mean_days"])
std = stats["std_days"] if stats["n_cycles"] >= 4 else stats["mean_days"] * 0.35
return {"last_low": last_low, "eta": eta,
"window_start": eta - pd.Timedelta(days=std),
"window_end": eta + pd.Timedelta(days=std),
"low_reliability_warning": stats["n_cycles"] < 4, "n_cycles_used": stats["n_cycles"]}
def cp_classify_translation(price: pd.Series, low_start: pd.Timestamp, low_end: pd.Timestamp) -> dict:
cycle_slice = price[(price.index >= low_start) & (price.index <= low_end)]
if len(cycle_slice) < 5:
return {"translation": "insufficient data", "peak_date": None}
peak_date = cycle_slice.idxmax()
midpoint = low_start + (low_end - low_start) / 2
translation = "right (bullish structure)" if peak_date > midpoint else "left (bearish structure)"
pct_through = (peak_date - low_start) / (low_end - low_start) * 100
return {"translation": translation, "peak_date": peak_date, "pct_through_cycle": pct_through}
def cp_get_confirmed_and_candidate_low(price: pd.Series, lows: pd.DatetimeIndex,
confirmation_pct: float = 0.05) -> dict:
if len(lows) == 0:
return {"confirmed_low": None, "candidate_low": None}
confirmed_low, candidate_low = None, None
for low_date in lows:
low_price = price.loc[low_date]
after = price.loc[low_date:]
threshold = low_price * (1 + confirmation_pct)
if (after >= threshold).any():
confirmed_low = low_date
else:
candidate_low = low_date
today = price.index[-1]
result = {"confirmed_low": confirmed_low, "candidate_low": candidate_low}
if confirmed_low is not None:
result["days_since_confirmed"] = (today - confirmed_low).days
if candidate_low is not None:
result["days_since_candidate"] = (today - candidate_low).days
return result
def cp_macro_phase_at_date(date: pd.Timestamp, yearly_lows: pd.DatetimeIndex) -> float:
if len(yearly_lows) < 2:
return np.nan
yearly_lows = yearly_lows.sort_values()
for i in range(len(yearly_lows) - 1):
start, end = yearly_lows[i], yearly_lows[i + 1]
if start <= date <= end:
duration = (end - start).days
if duration <= 0:
return np.nan
return np.clip((date - start).days / duration, 0.0, 1.0)
if date > yearly_lows[-1]:
previous, latest = yearly_lows[-2], yearly_lows[-1]
cycle_days = (latest - previous).days
if cycle_days <= 0:
return np.nan
return ((date - latest).days / cycle_days) % 1.0
return np.nan
def cp_classify_cycle_lengths(spacing_days: np.ndarray):
q1, q2 = np.quantile(spacing_days, [1 / 3, 2 / 3])
def classify(x):
return "Short" if x <= q1 else ("Normal" if x <= q2 else "Long")
return q1, q2, np.array([classify(x) for x in spacing_days])
def cp_fit_phase_modulation(weekly_lows: pd.DatetimeIndex, yearly_lows: pd.DatetimeIndex) -> dict:
if len(weekly_lows) < 5 or len(yearly_lows) < 2:
return {"valid": False}
spacing = np.diff(weekly_lows).astype("timedelta64[D]").astype(int)
phases, durations = [], []
for i, duration in enumerate(spacing):
start, end = weekly_lows[i], weekly_lows[i + 1]
midpoint = start + (end - start) / 2
phase = cp_macro_phase_at_date(midpoint, yearly_lows)
if np.isfinite(phase):
phases.append(phase)
durations.append(duration)
if len(durations) < 8:
return {"valid": False}
phases, durations = np.asarray(phases, dtype=float), np.asarray(durations, dtype=float)
X = np.column_stack([np.ones(len(phases)), np.sin(2*np.pi*phases), np.cos(2*np.pi*phases)])
beta, *_ = np.linalg.lstsq(X, durations, rcond=None)
return {"valid": True, "beta": beta, "global_mean": float(durations.mean())}
def cp_transition_adjustment(spacing_days: np.ndarray) -> dict:
if len(spacing_days) < 9:
return {"valid": False}
q1, q2, categories = cp_classify_cycle_lengths(spacing_days)
transition_data = {"Short": [], "Normal": [], "Long": []}
for i in range(len(spacing_days) - 1):
transition_data[categories[i]].append(spacing_days[i + 1])
global_mean = spacing_days.mean()
prior_strength = 8
multipliers, n_per_category = {}, {}
for category, values in transition_data.items():
n_per_category[category] = len(values)
if len(values) == 0:
multipliers[category] = 1.0
continue
raw_mean = np.mean(values)
shrunk = (raw_mean * len(values) + global_mean * prior_strength) / (len(values) + prior_strength)
multipliers[category] = shrunk / global_mean
return {"valid": True, "q1": q1, "q2": q2, "multipliers": multipliers,
"n_per_category": n_per_category, "global_mean": global_mean}
def cp_build_phase_modulated_wcl_model(weekly_lows, yearly_lows) -> dict:
if len(weekly_lows) < 5:
return {"valid": False}
spacing = np.diff(weekly_lows).astype("timedelta64[D]").astype(int)
return {"valid": True, "spacing_days": spacing, "global_mean": float(np.mean(spacing)),
"phase_model": cp_fit_phase_modulation(weekly_lows, yearly_lows),
"transition_model": cp_transition_adjustment(spacing)}
def cp_predict_next_wcl_days(model: dict, last_weekly_low, yearly_lows) -> dict:
if not model.get("valid"):
return {}
global_mean = model["global_mean"]
current_phase = cp_macro_phase_at_date(last_weekly_low, yearly_lows)
phase_model = model["phase_model"]
if phase_model.get("valid") and np.isfinite(current_phase):
beta = phase_model["beta"]
phase_pred = beta[0] + beta[1]*np.sin(2*np.pi*current_phase) + beta[2]*np.cos(2*np.pi*current_phase)
phase_pred = np.clip(phase_pred, global_mean*0.60, global_mean*1.45)
else:
phase_pred = global_mean
transition_model = model["transition_model"]
if transition_model.get("valid"):
spacing = model["spacing_days"]
prev_duration = spacing[-1]
q1, q2 = transition_model["q1"], transition_model["q2"]
prev_cat = "Short" if prev_duration <= q1 else ("Normal" if prev_duration <= q2 else "Long")
mult = transition_model["multipliers"][prev_cat]
else:
mult = 1.0
predicted_days = np.clip(phase_pred * mult, global_mean*0.60, global_mean*1.50)
return {"predicted_days": float(predicted_days)}
def cp_build_transition_only_model(lows, stats) -> dict:
if len(lows) < 2:
return {"valid": False}
spacing = np.array(stats["spacing_days"], dtype=float)
global_mean = stats["mean_days"]
if len(spacing) == 0 or np.isnan(global_mean):
return {"valid": False}
tm = cp_transition_adjustment(spacing)
if not tm.get("valid"):
return {"valid": True, "predicted_days": float(global_mean), "previous_category": "n/a",
"n_obs_in_category": 0, "spacing_days": spacing, "used_transition": False}
q1, q2 = tm["q1"], tm["q2"]
prev_duration = spacing[-1]
prev_cat = "Short" if prev_duration <= q1 else ("Normal" if prev_duration <= q2 else "Long")
mult = tm["multipliers"][prev_cat]
n_obs = tm["n_per_category"][prev_cat]
predicted_days = float(np.clip(global_mean * mult, global_mean*0.6, global_mean*1.5))
return {"valid": True, "predicted_days": predicted_days, "previous_category": prev_cat,
"n_obs_in_category": n_obs, "spacing_days": spacing, "used_transition": True}
def cp_project_next_wcl_low(lows, predicted_days, model) -> dict:
if len(lows) == 0 or not np.isfinite(predicted_days):
return {}
last_low = lows[-1]
eta = last_low + pd.Timedelta(days=predicted_days)
spacing = np.asarray(model["spacing_days"], dtype=float)
median = np.median(spacing)
mad = np.median(np.abs(spacing - median))
robust_std = 1.4826 * mad
if not np.isfinite(robust_std) or robust_std <= 0:
robust_std = np.std(spacing)
window_days = max(robust_std, predicted_days * 0.15)
return {"last_low": last_low, "eta": eta,
"window_start": eta - pd.Timedelta(days=window_days),
"window_end": eta + pd.Timedelta(days=window_days),
"predicted_days": predicted_days, "n_cycles_used": len(spacing)}
def cp_phase_locked_wave(price, lows, predicted_low, view_start, y_frac=0.25, amplitude_frac=0.10):
if len(lows) < 2:
return None, None
wave_start = max(pd.Timestamp(view_start), lows[0])
dates, values = [], []
price_min, price_max = float(price.min()), float(price.max())
log_range = np.log(price_max) - np.log(price_min)
y_center = np.exp(np.log(price_min) + log_range * y_frac)
amplitude = y_center * amplitude_frac
def make_segment(start, end):
if end <= start:
return None, None
seg_dates = pd.date_range(start, end, freq="D")
duration = max((end - start).days, 1)
progress = (seg_dates - start).days / duration
phase = -np.pi / 2 + 2 * np.pi * progress
return seg_dates, y_center + amplitude * np.sin(phase)
for i in range(len(lows) - 1):
start, end = lows[i], lows[i + 1]
if end < wave_start:
continue
start = max(start, wave_start)
d, v = make_segment(start, end)
if d is None:
continue
if len(dates) > 0:
d, v = d[1:], v[1:]
dates.extend(d)
values.extend(v)
if predicted_low:
start = max(lows[-1], wave_start)
end = predicted_low["eta"]
d, v = make_segment(start, end)
if d is not None:
if len(dates) > 0:
d, v = d[1:], v[1:]
dates.extend(d)
values.extend(v)
if not dates:
return None, None
return pd.DatetimeIndex(dates), np.asarray(values)
def cp_build_nested_cycle_report(price: pd.Series, daily_days=40, weekly_days=200,
yearly_days=1460, prominence_daily=0.03,
prominence_weekly=0.05, prominence_yearly=0.08) -> dict:
daily_lows = cp_find_cycle_lows(price, daily_days, prominence_daily, trend_window_days=weekly_days*1.5)
weekly_lows = cp_find_cycle_lows(price, weekly_days, prominence_weekly, trend_window_days=weekly_days*2)
yearly_lows = cp_find_cycle_lows(price, yearly_days, prominence_yearly, trend_window_days=yearly_days*2)
daily_stats = cp_cycle_stats(daily_lows)
weekly_stats = cp_cycle_stats(weekly_lows)
yearly_stats = cp_cycle_stats(yearly_lows)
wcl_model = cp_build_phase_modulated_wcl_model(weekly_lows, yearly_lows)
if wcl_model.get("valid") and len(weekly_lows) > 0:
next_wcl = cp_predict_next_wcl_days(wcl_model, weekly_lows[-1], yearly_lows)
weekly_adaptive = cp_project_next_wcl_low(weekly_lows, next_wcl["predicted_days"], wcl_model)
else:
weekly_adaptive = {}
standard_weekly_projection = cp_project_next_low(weekly_lows, weekly_stats)
yearly_transition_model = cp_build_transition_only_model(yearly_lows, yearly_stats)
if yearly_transition_model.get("valid"):
yearly_adaptive = cp_project_next_wcl_low(yearly_lows, yearly_transition_model["predicted_days"],
yearly_transition_model)
else:
yearly_adaptive = {}
standard_yearly_projection = cp_project_next_low(yearly_lows, yearly_stats)
weekly_translations, yearly_translations = [], []
for i in range(len(weekly_lows) - 1):
t = cp_classify_translation(price, weekly_lows[i], weekly_lows[i+1])
t["cycle_start"], t["cycle_end"] = weekly_lows[i], weekly_lows[i+1]
weekly_translations.append(t)
for i in range(len(yearly_lows) - 1):
t = cp_classify_translation(price, yearly_lows[i], yearly_lows[i+1])
t["cycle_start"], t["cycle_end"] = yearly_lows[i], yearly_lows[i+1]
yearly_translations.append(t)
return {
"daily": {"lows": daily_lows, "stats": daily_stats,
"projection": cp_project_next_low(daily_lows, daily_stats)},
"weekly": {"lows": weekly_lows, "stats": weekly_stats,
"projection": standard_weekly_projection, "adaptive_projection": weekly_adaptive,
"wcl_model": wcl_model, "translations": weekly_translations},
"yearly": {"lows": yearly_lows, "stats": yearly_stats,
"projection": standard_yearly_projection, "adaptive_projection": yearly_adaptive,
"transition_model": yearly_transition_model, "translations": yearly_translations},
}
def cp_plot_nested_dashboard(price: pd.Series, nested: dict, asset_name: str,
view_start: str = "2000-01-01"):
view_start_ts = pd.Timestamp(view_start)
fig, ax = plt.subplots(figsize=(14, 7))
fig.patch.set_facecolor("#0d1321")
ax.set_facecolor("#0d1321")
ax.plot(price.index, price.values, color="#e8e8e8", lw=1.1, zorder=6)
ax.set_yscale("log")
for low in nested["daily"]["lows"]:
if low >= view_start_ts:
ax.plot(low, price.loc[low], marker=".", color="#7aa8d8", ms=4, alpha=0.6, zorder=3)
weekly_lows = nested["weekly"]["lows"]
for low in weekly_lows:
if low >= view_start_ts:
ax.plot(low, price.loc[low], marker="^", color="#4a90d9", ms=8, zorder=4)
for i in range(len(weekly_lows) - 1):
if weekly_lows[i+1] < view_start_ts:
continue
gap_weeks = (weekly_lows[i+1] - weekly_lows[i]).days / 7
mid = weekly_lows[i] + (weekly_lows[i+1] - weekly_lows[i]) / 2
y = price.loc[weekly_lows[i]:weekly_lows[i+1]].min() * 0.90
ax.annotate(f"{gap_weeks:.0f}W", (mid, y), color="#4a90d9", fontsize=7.5, ha="center", alpha=0.85)
yearly_lows = nested["yearly"]["lows"]
for low in yearly_lows:
if low >= view_start_ts:
ax.axvline(low, color="#c9302c", ls=":", lw=0.9, alpha=0.7)
ax.plot(low, price.loc[low], marker="D", color="#e8b84b", ms=11, zorder=5,
markeredgecolor="#7a5a30", markeredgewidth=0.8)
weekly_adaptive = nested["weekly"].get("adaptive_projection", {})
wcl_model = nested["weekly"].get("wcl_model", {})
weekly_wave_source = weekly_adaptive if wcl_model.get("valid") and weekly_adaptive else nested["weekly"]["projection"]
wd, wv = cp_phase_locked_wave(price, weekly_lows, weekly_wave_source, view_start_ts, y_frac=0.25, amplitude_frac=0.10)
if wd is not None:
ax.plot(wd, wv, color="#4a90d9", lw=1.8, alpha=0.85, zorder=2,
label="Weekly cycle rhythm" + (" (adaptive)" if wcl_model.get("valid") else ""))
yearly_adaptive = nested["yearly"].get("adaptive_projection", {})
yearly_transition_model = nested["yearly"].get("transition_model", {})
yearly_wave_source = yearly_adaptive if yearly_adaptive else nested["yearly"]["projection"]
yd, yv = cp_phase_locked_wave(price, yearly_lows, yearly_wave_source, view_start_ts, y_frac=0.75, amplitude_frac=0.14)
if yd is not None:
ax.plot(yd, yv, color="#e8b84b", lw=2, alpha=0.85, zorder=2,
label="Yearly cycle rhythm" + (" (adaptive)" if yearly_transition_model.get("used_transition") else ""))
wproj = weekly_wave_source
if wproj:
ax.axvspan(wproj["window_start"], wproj["window_end"], color="#4a90d9", alpha=0.15)
ax.axvline(wproj["eta"], color="#4a90d9", ls="--", lw=1.2, alpha=0.9)
yproj = yearly_wave_source
if yproj:
ax.axvspan(yproj["window_start"], yproj["window_end"], color="#e8b84b", alpha=0.15)
ax.axvline(yproj["eta"], color="#e8b84b", ls="--", lw=1.4, alpha=0.9)
ax.set_xlim(left=view_start_ts)
ax.set_title(f"{asset_name}: nested Daily / Weekly / Yearly cycle lows", color="white", fontsize=12)
ax.tick_params(colors="white")
for spine in ax.spines.values():
spine.set_color("#444")
legend_elems = [
plt.Line2D([0], [0], marker='.', color='w', markerfacecolor='#7aa8d8', ms=8,
label=f"Daily-tier low ({nested['daily']['stats']['n_cycles']} cycles)", ls=''),
plt.Line2D([0], [0], marker='^', color='w', markerfacecolor='#4a90d9', ms=9,
label=f"Weekly-tier low ({nested['weekly']['stats']['n_cycles']} cycles)", ls=''),
plt.Line2D([0], [0], marker='D', color='w', markerfacecolor='#e8b84b', ms=9,
label=f"Yearly-tier low ({nested['yearly']['stats']['n_cycles']} cycles)", ls=''),
]
if wproj:
legend_elems.append(plt.Line2D([0], [0], color='#4a90d9', ls='--', lw=1.5,
label=f"Next weekly low: {wproj['eta'].date()}"))
if yproj:
low_rel = nested["yearly"]["stats"]["n_cycles"] < 4
tm = nested["yearly"].get("transition_model", {})
n_obs = tm.get("n_obs_in_category")
obs_note = f", n={n_obs} in category" if tm.get("used_transition") and n_obs is not None else ""
legend_elems.append(plt.Line2D([0], [0], color='#e8b84b', ls='--', lw=1.5,
label=f"Next yearly low: {yproj['eta'].date()}"
+ (" [low reliability]" if low_rel else "") + obs_note))
ax.legend(handles=legend_elems, loc="upper left", facecolor="#1a2036",
labelcolor="white", framealpha=0.85, fontsize=8.5)
plt.tight_layout()
return fig
def render_cycle_projector_tab():
st.title("πͺ Cycle Projector")
st.caption(
"Nested Daily/Weekly/Yearly pivot-low detection with a phase-modulated weekly model "
"and a transition-adaptive yearly model (Short/Normal/Long-conditioned band, not a "
"point forecast). Yearly resolution is set MANUALLY per ticker β auto-detection was "
"tested and found to underperform manual tuning on BTC, gold, and Nasdaq alike, so "
"this tab keeps you in the loop on that call."
)
with st.sidebar:
st.markdown("---")
st.header("Cycle Projector")
cp_ticker = st.text_input("Ticker (yfinance format)", value="BTC-USD", key="cp_ticker")
cp_yearly_years = st.number_input(
"Yearly cycle resolution (years)", min_value=1.0, max_value=15.0,
value=6.0, step=0.25, key="cp_yearly_years",
help="BTC needed ~6y or the 2018 low merges into the COVID low. "
"Gold and Nasdaq will likely need different values β tune per ticker."
)
with st.expander("Advanced settings"):
cp_daily_days = st.number_input("Daily cycle length (days)", value=40, step=5, key="cp_daily_days")
cp_weekly_days = st.number_input("Weekly cycle length (days)", value=200, step=10, key="cp_weekly_days")
cp_prom_daily = st.slider("Daily prominence", 0.01, 0.20, 0.03, 0.01, key="cp_prom_daily")
cp_prom_weekly = st.slider("Weekly prominence", 0.01, 0.20, 0.05, 0.01, key="cp_prom_weekly")
cp_prom_yearly = st.slider("Yearly prominence", 0.01, 0.30, 0.08, 0.01, key="cp_prom_yearly")
cp_view_start = st.text_input("Chart view start (YYYY-MM-DD)", value="2000-01-01", key="cp_view_start")
cp_run = st.button("βΆοΈ Run", key="cp_run")
if not cp_run:
st.info("Set a ticker and yearly resolution in the sidebar, then click Run.")
return
with st.spinner(f"Fetching {cp_ticker}β¦"):
data = yf.download(cp_ticker, period="max", auto_adjust=True, progress=False)
if data.empty:
st.error(f"No data returned for '{cp_ticker}'.")
return
price = data["Close"]
if isinstance(price, pd.DataFrame):
price = price.iloc[:, 0]
nested = cp_build_nested_cycle_report(
price, daily_days=cp_daily_days, weekly_days=cp_weekly_days,
yearly_days=cp_yearly_years * 365.25,
prominence_daily=cp_prom_daily, prominence_weekly=cp_prom_weekly,
prominence_yearly=cp_prom_yearly,
)
if nested["weekly"]["stats"]["n_cycles"] == 0 and nested["yearly"]["stats"]["n_cycles"] == 0:
st.warning("No valid lows found at this resolution/prominence β try adjusting the sliders.")
return
wstats, ystats = nested["weekly"]["stats"], nested["yearly"]["stats"]
c1, c2, c3, c4 = st.columns(4)
c1.metric("Weekly cycle avg", f"{wstats['mean_days']/7:.0f}w" if wstats["n_cycles"] else "n/a")
c2.metric("Yearly cycle avg", f"{ystats['mean_days']/365.25:.2f}y" if ystats["n_cycles"] else "n/a")
c3.metric("Weekly cycles found", wstats["n_cycles"])
c4.metric("Yearly cycles found", ystats["n_cycles"])
if ystats["n_cycles"] < 4:
st.warning(f"Only {ystats['n_cycles']} completed yearly cycles β treat the yearly window as very rough.")
wproj = nested["weekly"]["adaptive_projection"] or nested["weekly"]["projection"]
yproj = nested["yearly"]["adaptive_projection"] or nested["yearly"]["projection"]
if wproj:
st.info(f"**Next weekly-tier low:** {wproj['eta'].strftime('%Y-%m-%d')} "
f"(window: {wproj['window_start'].strftime('%Y-%m-%d')} to {wproj['window_end'].strftime('%Y-%m-%d')})")
if yproj:
ytm = nested["yearly"]["transition_model"]
note = ""
if ytm.get("used_transition"):
note = f" (adaptive, {ytm['previous_category']} category, n={ytm['n_obs_in_category']} prior obs)"
st.info(f"**Next yearly-tier low{note}:** {yproj['eta'].strftime('%Y-%m-%d')} "
f"(window: {yproj['window_start'].strftime('%Y-%m-%d')} to {yproj['window_end'].strftime('%Y-%m-%d')})")
fig = cp_plot_nested_dashboard(price, nested, cp_ticker, view_start=cp_view_start)
st.pyplot(fig)
plt.close(fig)
conf = cp_get_confirmed_and_candidate_low(price, nested["weekly"]["lows"])
cA, cB = st.columns(2)
if conf.get("confirmed_low") is not None:
cA.success(f"β Confirmed weekly-tier low: {conf['confirmed_low'].date()} ({conf['days_since_confirmed']}d ago)")
else:
cA.info("β Confirmed low: none yet")
if conf.get("candidate_low") is not None:
cB.warning(f"β Unconfirmed candidate: {conf['candidate_low'].date()} ({conf['days_since_candidate']}d ago)")
else:
cB.info("β Unconfirmed candidate: none")
with st.expander("Weekly-tier translation per completed cycle"):
for t in nested["weekly"]["translations"]:
peak_str = t['peak_date'].date() if t['peak_date'] else 'n/a'
st.write(f"{t['cycle_start'].date()} β {t['cycle_end'].date()}: **{t['translation']}** "
f"(peak at {t['pct_through_cycle']:.0f}% through the cycle, {peak_str})")
# =========================================================
# MAIN LAYOUT
# =========================================================
tab1, tab2, tab3, tab4, tab5, tab6, tab7, tab8 = st.tabs([
"π Whale Monitor",
"π Peak Detector",
"π‘ Pump Scanner",
"ποΈ Signal Tracker",
"π§ Rotation Map",
"π― Combined Scanner",
"πͺ Catch the Knife",
"πͺ Cycle Projector",
])
with tab1:
render_whale_monitor()
with tab2:
render_peak_detector()
with tab3:
render_penny_scanner()
with tab4:
render_signal_tracker()
with tab5:
render_rotation_map()
with tab6:
render_combined_scanner()
with tab7:
render_catch_the_knife()
with tab8:
render_cycle_projector_tab()
import warnings
warnings.filterwarnings("ignore")
import os
import re
import time
import requests
from datetime import time as dt_time
import streamlit as st
import yfinance as yf
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
import matplotlib.gridspec as gridspec
import plotly.graph_objects as go
import plotly.express as px
from numpy.lib.stride_tricks import sliding_window_view
from scipy.stats import spearmanr, binom as binom_dist
from sklearn.ensemble import GradientBoostingClassifier
from huggingface_hub import HfApi, hf_hub_download
from scipy.signal import find_peaks
st.set_page_config(page_title="Market Dashboard", page_icon="π", layout="wide")
# =========================================================
# SHARED: PERSISTENT SIGNAL LOG (Hugging Face Dataset backend)
# =========================================================
HF_TOKEN = os.environ.get("HF_TOKEN")
DATASET_REPO = os.environ.get("HF_DATASET_REPO")
LOG_FILENAME = "signal_log.csv"
def _log_configured() -> bool:
return bool(HF_TOKEN and DATASET_REPO)
def load_signal_log() -> pd.DataFrame:
if not _log_configured():
return pd.DataFrame()
try:
local_path = hf_hub_download(
repo_id=DATASET_REPO, repo_type="dataset", filename=LOG_FILENAME, token=HF_TOKEN
)
df = pd.read_csv(local_path)
if "FlagDate" in df.columns:
df["FlagDate"] = pd.to_datetime(df["FlagDate"])
return df
except Exception:
return pd.DataFrame()
def save_signal_log(df: pd.DataFrame) -> bool:
if not _log_configured():
return False
try:
api = HfApi(token=HF_TOKEN)
csv_bytes = df.to_csv(index=False).encode("utf-8")
api.upload_file(
path_or_fileobj=csv_bytes,
path_in_repo=LOG_FILENAME,
repo_id=DATASET_REPO,
repo_type="dataset",
commit_message=f"Update signal log ({len(df)} total signals)",
)
return True
except Exception as e:
st.warning(f"Could not save signal log: {e}")
return False
def append_hits_to_log(hits_df: pd.DataFrame, settings: dict) -> bool:
if hits_df.empty or not _log_configured():
return False
existing = load_signal_log()
new_rows = hits_df.copy()
new_rows["FlagDate"] = pd.Timestamp.now().normalize()
new_rows["RSI_Threshold"] = settings["rsi_threshold"]
new_rows["Stretch_Threshold"] = settings["stretch_threshold"]
new_rows["Strict_Window"] = settings["strict_window"]
new_rows["Lookback"] = settings["lookback"]
new_rows["WinRate_Min"] = settings["win_rate_min"]
combined = pd.concat([existing, new_rows], ignore_index=True) if not existing.empty else new_rows
combined = combined.drop_duplicates(subset=["Ticker", "FlagDate"], keep="last")
return save_signal_log(combined)
# =========================================================
# TAB 1: WHALE MONITOR
# =========================================================
SIZE_COLORS = {
"WHALE": "#8e44ad", # purple
"INSTITUTIONAL": "#e67e22", # orange
"LARGE": "#2980b9", # blue
"MEDIUM": "#27ae60", # green
"RETAIL": "#7f8c8d", # grey
}