"""
finviz_alert.py — Finviz Elite screener alert bot
Polls multiple presets every 5 minutes, sends Telegram alerts for new tickers.
Runs during market hours (ET), sleeps and auto-wakes at 4 AM ET.
"""
import csv
import io
import logging
import os
import time
from datetime import datetime, timedelta
import pytz
import requests
from dotenv import load_dotenv
load_dotenv()
# ── Config ────────────────────────────────────────────────────────────────────
TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
TELEGRAM_CHAT_ID = os.getenv("TELEGRAM_CHAT_ID")
FINVIZ_EMAIL = os.getenv("FINVIZ_EMAIL")
FINVIZ_PASSWORD = os.getenv("FINVIZ_PASSWORD")
POLL_INTERVAL = 300 # seconds between scans (5 min)
EASTERN = pytz.timezone("America/New_York")
# Session windows (ET)
PREMARKET_START = (4, 0) # 04:00 AM ET
MARKET_OPEN = (9, 30) # 09:30 AM ET
MARKET_CLOSE = (16, 0) # 04:00 PM ET
# ── Screener presets ──────────────────────────────────────────────────────────
# Each entry becomes one independent Finviz screener call.
# Finviz filter codes: https://finviz.com/screener.ashx filter reference
# sh_price_o1 = price > $1
# sh_price_u20 = price < $20
# sh_relvol_o2 = relative volume > 2x
# sh_relvol_o3 = relative volume > 3x
# sh_avgvol_o500 = avg volume > 500K
# sh_float_u10 = float < 10M shares
# sh_float_u20 = float < 20M shares
# ta_change_u10 = change > 10% ("u" = "up" in change/gap codes)
# ta_change_u20 = change > 20%
# ta_gap_u10 = gap up > 10%
PRESETS = [
{
"name": "RVOL>2 Float<10M Change>10%",
"filters": "sh_price_o1,sh_price_u20,sh_relvol_o2,sh_avgvol_o500,sh_float_u10,ta_change_u10",
},
{
"name": "RVOL>3 Float<20M Change>20%",
"filters": "sh_price_o1,sh_price_u20,sh_relvol_o3,sh_avgvol_o500,sh_float_u20,ta_change_u20",
},
{
"name": "RVOL>2 Float<10M Gap>10%",
"filters": "sh_price_o1,sh_price_u20,sh_relvol_o2,sh_avgvol_o500,sh_float_u10,ta_gap_u10",
},
{
"name": "RVOL>3 Float<10M Change>20%",
"filters": "sh_price_o1,sh_price_u20,sh_relvol_o3,sh_avgvol_o500,sh_float_u10,ta_change_u20",
},
{
"name": "RVOL>2 Float<20M Gap>10%",
"filters": "sh_price_o1,sh_price_u20,sh_relvol_o2,sh_avgvol_o500,sh_float_u20,ta_gap_u10",
},
]
# ── Logging ───────────────────────────────────────────────────────────────────
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s %(levelname)-7s %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
handlers=[
logging.StreamHandler(),
logging.FileHandler("finviz_alert.log"),
],
)
log = logging.getLogger(__name__)
# ── Finviz Elite session ──────────────────────────────────────────────────────
class FinvizSession:
LOGIN_URL = "https://finviz.com/login_submit.ashx"
EXPORT_URL = "https://elite.finviz.com/export.ashx"
def __init__(self, email: str, password: str):
self.email = email
self.password = password
self.session = requests.Session()
self.session.headers.update({
"User-Agent": (
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/124.0.0.0 Safari/537.36"
),
"Referer": "https://finviz.com/",
})
self._logged_in = False
def login(self):
"""Authenticate and store session cookie."""
log.info("Logging in to Finviz Elite …")
resp = self.session.post(
self.LOGIN_URL,
data={"email": self.email, "password": self.password, "remember": "on"},
timeout=15,
)
resp.raise_for_status()
if ".ASPXAUTH" not in self.session.cookies and "remember_token" not in self.session.cookies and "user" not in self.session.cookies:
# Finviz returns 200 even on bad credentials; check for auth cookie
raise RuntimeError(
"Login failed — no auth cookie set. "
"Check FINVIZ_EMAIL / FINVIZ_PASSWORD in .env."
)
self._logged_in = True
log.info("Finviz login OK.")
def fetch_screener(self, filters: str) -> list[dict]:
"""
Fetch screener CSV export for *filters* string.
Returns list of row dicts with normalised keys.
Re-authenticates once on 401/403.
"""
if not self._logged_in:
self.login()
for attempt in range(2):
resp = self.session.get(
self.EXPORT_URL,
params={"v": "152", "f": filters, "ft": "4", "c": "0,1,2,65,67"},
timeout=20,
)
if resp.status_code in (401, 403) and attempt == 0:
log.warning("Session expired — re-authenticating …")
self._logged_in = False
self.login()
continue
resp.raise_for_status()
break
return _parse_csv(resp.text)
def _parse_csv(raw: str) -> list[dict]:
"""Parse Finviz export CSV, return list of normalised row dicts."""
reader = csv.DictReader(io.StringIO(raw))
rows = []
for row in reader:
# Normalise common header name variants
ticker = (
row.get("Ticker") or row.get("ticker") or row.get("Symbol") or ""
).strip().upper()
if not ticker or ticker == "TICKER":
continue
price_str = row.get("Price", row.get("price", "0")).strip()
try:
price_val = float(price_str.replace(",", ""))
except ValueError:
price_val = 0.0
if not (1.0 <= price_val <= 20.0):
continue
rows.append({
"ticker": ticker,
"price": price_str,
"change": row.get("Change", row.get("change", "N/A")).strip(),
"volume": row.get("Volume", row.get("volume", "N/A")).strip(),
"rvol": row.get("Rel Volume", row.get("Relative Volume", "N/A")).strip(),
})
return rows
# ── Telegram ──────────────────────────────────────────────────────────────────
def send_telegram(token: str, chat_id: str, text: str):
url = f"https://api.telegram.org/bot{token}/sendMessage"
resp = requests.post(
url,
json={"chat_id": chat_id, "text": text, "parse_mode": "HTML"},
timeout=10,
)
resp.raise_for_status()
def session_label() -> str:
t = (now_et().hour, now_et().minute)
if PREMARKET_START <= t < MARKET_OPEN:
return "🌅 PRE-MARKET"
return "📈 MARKET"
def _esc(s: str) -> str:
return s.replace("&", "&").replace("<", "<").replace(">", ">")
def format_alert(preset_name: str, ticker: str, data: dict) -> str:
return (
f"🚨 NEW HIT — {ticker}\n"
f"{session_label()}\n"
f"📋 Preset: {_esc(preset_name)}\n"
f"💰 Price: ${data['price']}\n"
f"📈 Change: {data['change']}\n"
f"📊 Volume: {data['volume']}\n"
f"⚡ RVOL: {data['rvol']}\n"
f"🔗 Finviz chart"
)
# ── Market hours ──────────────────────────────────────────────────────────────
def now_et() -> datetime:
return datetime.now(EASTERN)
def is_active_session() -> bool:
"""True if ET time is within pre-market or market hours on a weekday."""
now = now_et()
if now.weekday() >= 5:
return False
t = (now.hour, now.minute)
return PREMARKET_START <= t < MARKET_CLOSE
def seconds_until_next_open() -> float:
"""Return seconds until next 04:00 AM ET pre-market open."""
now = now_et()
target = now.replace(hour=PREMARKET_START[0], minute=PREMARKET_START[1],
second=0, microsecond=0)
if now >= target:
target += timedelta(days=1)
while target.weekday() >= 5:
target += timedelta(days=1)
return (target - now).total_seconds()
# ── Main loop ─────────────────────────────────────────────────────────────────
def run():
if not all([TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID]):
raise EnvironmentError("TELEGRAM_BOT_TOKEN and TELEGRAM_CHAT_ID must be set in .env")
if not all([FINVIZ_EMAIL, FINVIZ_PASSWORD]):
raise EnvironmentError("FINVIZ_EMAIL and FINVIZ_PASSWORD must be set in .env")
fv = FinvizSession(FINVIZ_EMAIL, FINVIZ_PASSWORD)
# seen[preset_name] = set of tickers found in previous scan
seen: dict[str, set[str]] = {p["name"]: set() for p in PRESETS}
# global set — no ticker alerted more than once per day regardless of preset
alerted_today: set[str] = set()
# Reset seen tickers each calendar day
last_reset_date: datetime | None = None
last_login_alert: datetime | None = None # throttle login-failure alerts to 1/hr
# morning window candidates: ticker → best data seen (highest RVOL wins)
morning_candidates: dict[str, dict] = {}
pick_sent_today: bool = False # ensure recommendation fires only once per day
log.info("Finviz alert bot started. Pre-market %02d:%02d | Market open %02d:%02d | Close %02d:%02d ET.",
*PREMARKET_START, *MARKET_OPEN, *MARKET_CLOSE)
while True:
if not is_active_session():
secs = seconds_until_next_open()
wake = now_et() + timedelta(seconds=secs)
log.info("Outside session — sleeping %.0f min. Next open ~%s ET.",
secs / 60, wake.strftime("%Y-%m-%d %H:%M"))
time.sleep(secs)
continue
# Reset seen tickers at the start of each new trading day
today = now_et().date()
if last_reset_date != today:
log.info("New trading day — clearing seen-ticker cache.")
seen = {p["name"]: set() for p in PRESETS}
alerted_today = set()
morning_candidates = {}
pick_sent_today = False
last_reset_date = today
scan_start = time.monotonic()
total_new = 0
for preset in PRESETS:
name = preset["name"]
filters = preset["filters"]
try:
rows = fv.fetch_screener(filters)
except Exception as exc:
log.error("Screener fetch failed [%s]: %s", name, exc)
if "Login failed" in str(exc):
now = now_et()
if last_login_alert is None or (now - last_login_alert).total_seconds() > 3600:
last_login_alert = now
try:
send_telegram(
TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID,
"⚠️ Finviz login broken\nBot can't authenticate — no alerts will fire until fixed.\nCheck .env credentials or Finviz may have changed their login.",
)
except Exception:
pass
continue
current_tickers = {r["ticker"] for r in rows}
new_tickers = (current_tickers - seen[name]) - alerted_today
now_t = now_et()
in_morning_window = (now_t.hour, now_t.minute) < MARKET_OPEN or (
(now_t.hour == 9 and now_t.minute >= 30) or
(now_t.hour == 9 and now_t.minute < 60)
)
in_morning_window = (
(now_t.hour == 9 and now_t.minute >= 30) or now_t.hour < 10
) and now_t.hour >= 4
if new_tickers:
log.info("[%s] %d new ticker(s): %s",
name, len(new_tickers), ", ".join(sorted(new_tickers)))
data_by_ticker = {r["ticker"]: r for r in rows}
for ticker in sorted(new_tickers):
data = data_by_ticker[ticker]
msg = format_alert(name, ticker, data)
try:
send_telegram(TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID, msg)
log.info(" ✓ Alert sent: %s", ticker)
alerted_today.add(ticker)
except Exception as exc:
log.error(" ✗ Telegram send failed for %s: %s", ticker, exc)
# Accumulate for morning recommendation (9:30–10:00 ET)
if in_morning_window:
try:
rvol_val = float(data.get("rvol", "0").replace("x", "") or 0)
except ValueError:
rvol_val = 0.0
existing = morning_candidates.get(ticker)
if existing is None or rvol_val > existing.get("_rvol", 0):
morning_candidates[ticker] = {**data, "_rvol": rvol_val}
total_new += len(new_tickers)
else:
log.info("[%s] %d ticker(s) — no new hits.", name, len(current_tickers))
seen[name] = current_tickers
# ── 10:00 AM ET pick: top-2 by RVOL from morning window ─────────────
now_t = now_et()
past_10am = now_t.hour >= 10
if past_10am and not pick_sent_today and morning_candidates:
ranked = sorted(
morning_candidates.items(),
key=lambda kv: kv[1].get("_rvol", 0),
reverse=True,
)
top2 = ranked[:2]
lines = []
for i, (ticker, data) in enumerate(top2, 1):
lines.append(
f"{i}. {ticker} "
f"💰 ${data.get('price','?')} "
f"📈 {data.get('change','?')} "
f"⚡ RVOL {data.get('rvol','?')}"
)
msg = (
"🏆 10 AM Top Picks\n"
"Best setups from the opening 30 min:\n\n"
+ "\n".join(lines)
)
try:
send_telegram(TELEGRAM_BOT_TOKEN, TELEGRAM_CHAT_ID, msg)
pick_sent_today = True
log.info("10 AM picks sent: %s", ", ".join(t for t, _ in top2))
except Exception as exc:
log.error("Failed to send 10 AM picks: %s", exc)
elapsed = time.monotonic() - scan_start
sleep_for = max(0, POLL_INTERVAL - elapsed)
log.info("Scan done in %.1fs. %d new total. Sleeping %.0fs …",
elapsed, total_new, sleep_for)
time.sleep(sleep_for)
if __name__ == "__main__":
while True:
try:
run()
except KeyboardInterrupt:
log.info("Interrupted. Bye.")
break
except Exception as exc:
log.exception("Unhandled error — restarting in 60s: %s", exc)
time.sleep(60)