"""sentiment_feed.py — trader-sentiment feed for the Bird's Eye dashboard (v1.17). Polls StockTwits' public symbol streams for SPY, QQQ and the top 20 names on the shortlist and writes sentiment.json beside quotes.json. The receiver already serves this folder as static files, so the dashboard fetches it directly — no receiver change, no Claude usage, no TWS lines. What it measures, per symbol, from the latest 30 messages: bull / bear / untagged users' own Bullish/Bearish tags (StockTwits' on-page gauge is proprietary and is NOT this number — ours is the tagged split, which is what we can reproduce) bull_pct bull / (bull + bear); None when fewer than MIN_TAGGED posts are tagged (too thin) label Bullish >= 65%, Bearish <= 35%, else Mixed; "thin" when bull_pct is None vel messages per hour, from the time span the 30 messages cover base / buzz a slow moving average of vel per symbol (kept in sentiment_state.json across runs) and vel / base. BUZZ is flagged at >= BUZZ_X once the base has >= BASE_MIN samples; until then the name reads "learning", never a false buzz posts the newest few posts (trimmed), with their tag and a link Pacing: StockTwits' unauthenticated limit has historically been ~200 requests/hour per IP. Default is one request every 20 s (180/hour). SPY and QQQ are refreshed every INDEX_EVERY seconds; the shortlist names rotate in between, so a full sweep of 20 takes ~8 minutes. A 429 backs off 5 minutes; a 403 (a bot wall) backs off 10 and is written into sentiment.json so the board shows the fault instead of going quiet. Usage python sentiment_feed.py # run forever (start it beside birdseye_tvfeed.py) python sentiment_feed.py --once # one sweep of every symbol, then exit (first-run check) python sentiment_feed.py --pace 30 # slower, if 429s show up python sentiment_feed.py --symbols HOOD,TEM # override the shortlist python sentiment_feed.py --selftest # offline checks, no network """ import argparse, json, os, re, sys, time, urllib.request, urllib.error from datetime import datetime, timezone HERE = os.path.dirname(os.path.abspath(__file__)) OUT = os.path.join(HERE, "sentiment.json") STATE = os.path.join(HERE, "sentiment_state.json") SHORTLIST = os.path.join(os.path.dirname(HERE), "Bernie_s Shortlist", "shortlist.json") API = "https://api.stocktwits.com/api/2/streams/symbol/{}.json" UA = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/140.0 Safari/537.36" INDEXES = ["SPY", "QQQ"] TOP_N = 20 PACE = 20 # seconds between requests INDEX_EVERY = 180 # SPY/QQQ refresh interval, seconds MIN_TAGGED = 5 # below this many tagged posts, bull_pct is None (too thin to call) BULL_CUT, BEAR_CUT = 65, 35 BUZZ_X = 3.0 # vel / base at or above this = BUZZ BASE_MIN = 5 # samples before a base is trusted BASE_ALPHA = 0.15 # moving-average weight of each new sample MAX_POSTS = 4 def now_utc(): return datetime.now(timezone.utc) def iso(dt): return dt.replace(microsecond=0).isoformat() def parse_ts(s): return datetime.strptime(s, "%Y-%m-%dT%H:%M:%SZ").replace(tzinfo=timezone.utc) def clean(body, n=180): t = re.sub(r"\s+", " ", body or "").strip() return t if len(t) <= n else t[: n - 1].rstrip() + "…" def load_json(path, default): try: with open(path, encoding="utf-8") as f: return json.load(f) except (OSError, ValueError): return default def write_atomic(path, obj): tmp = path + ".tmp" with open(tmp, "w", encoding="utf-8") as f: json.dump(obj, f, indent=1) os.replace(tmp, path) # the dashboard polls; never let it read a half-written file def shortlist_top(path=SHORTLIST, n=TOP_N): """Top n shortlist names by priority. Accepts the three shapes the receiver accepts.""" d = load_json(path, None) if d is None: raise RuntimeError("shortlist.json not readable at " + path) if isinstance(d, list): return [str(s).upper() for s in d][:n] names = d.get("names", d.get("tickers")) if isinstance(names, list): return [(x if isinstance(x, str) else x.get("ticker") or x.get("s")).upper() for x in names][:n] if isinstance(names, dict): ranked = sorted(names.items(), key=lambda kv: (kv[1] or {}).get("priority") or 999) return [k.upper() for k, _ in ranked][:n] raise RuntimeError("shortlist.json has no names") def summarize(doc, prev_state, t_now): """One StockTwits stream response -> the per-symbol record, plus the updated state entry.""" msgs = doc.get("messages") or [] bull = bear = 0 posts = [] for m in msgs: tag = ((m.get("entities") or {}).get("sentiment") or {}).get("basic") if tag == "Bullish": bull += 1 elif tag == "Bearish": bear += 1 if len(posts) < MAX_POSTS and m.get("body"): u = (m.get("user") or {}).get("username") or "" posts.append({"t": m.get("created_at"), "user": u, "tag": tag, "text": clean(m.get("body")), "url": f"https://stocktwits.com/{u}/message/{m.get('id')}" if u else None}) tagged = bull + bear bull_pct = round(100.0 * bull / tagged) if tagged >= MIN_TAGGED else None label = ("thin" if bull_pct is None else "Bullish" if bull_pct >= BULL_CUT else "Bearish" if bull_pct <= BEAR_CUT else "Mixed") vel = None if len(msgs) >= 2: t0, t1 = parse_ts(msgs[-1]["created_at"]), parse_ts(msgs[0]["created_at"]) # span floor of 2 min so a burst of near-simultaneous posts can't read as infinite speed hrs = max((t1 - t0).total_seconds(), 120) / 3600.0 vel = round((len(msgs) - 1) / hrs, 1) # a stream whose newest post is old is quiet now, whatever its old burst rate was idle_h = (t_now - t1).total_seconds() / 3600.0 if idle_h > hrs: vel = round((len(msgs) - 1) / (idle_h + hrs), 1) st = dict(prev_state or {}) base, n = st.get("base"), st.get("n", 0) buzz = round(vel / base, 1) if (vel is not None and base and n >= BASE_MIN) else None if vel is not None: base = vel if base is None else (1 - BASE_ALPHA) * base + BASE_ALPHA * vel st = {"base": round(base, 2), "n": n + 1, "seen": iso(t_now)} rec = {"bull": bull, "bear": bear, "untagged": len(msgs) - tagged, "n": len(msgs), "bull_pct": bull_pct, "label": label, "tone": {"Bullish": "bull", "Bearish": "bear"}.get(label, "neutral"), "vel": vel, "base": st.get("base"), "base_n": st.get("n", 0), "buzz": buzz, "hot": bool(buzz is not None and buzz >= BUZZ_X), "watchers": (doc.get("symbol") or {}).get("watchlist_count"), "posts": posts, "asof": iso(t_now)} return rec, st def fetch(sym, timeout=15): req = urllib.request.Request(API.format(sym), headers={"User-Agent": UA, "Accept": "application/json"}) with urllib.request.urlopen(req, timeout=timeout) as r: return json.loads(r.read().decode("utf-8")) class Feed: def __init__(self, symbols_override=None, pace=PACE): self.override = symbols_override self.pace = pace self.state = load_json(STATE, {}) old = load_json(OUT, {}) self.names = old.get("names", {}) if isinstance(old, dict) else {} self.error = None self.top = [] self.rr = 0 self.last_index = 0.0 def refresh_top(self): if self.override: self.top = self.override return try: self.top = shortlist_top() if self.error and self.error.startswith("shortlist"): self.error = None # recovered except Exception as e: # keep the last list; say so on the board self.error = f"shortlist: {e}" def write(self): write_atomic(OUT, {"asof": iso(now_utc()), "source": "StockTwits public streams (user-tagged posts)", "indexes": INDEXES, "top": self.top, "pace_s": self.pace, "cuts": {"bull": BULL_CUT, "bear": BEAR_CUT, "buzz": BUZZ_X, "min_tagged": MIN_TAGGED}, "error": self.error, "names": self.names}) write_atomic(STATE, self.state) def one(self, sym): t = now_utc() try: doc = fetch(sym) except urllib.error.HTTPError as e: wait = 600 if e.code == 403 else 300 if e.code == 429 else 60 self.error = f"StockTwits HTTP {e.code} on {sym} at {t.strftime('%H:%M')}Z — backing off {wait // 60} min" print(self.error, flush=True) self.write() return wait except Exception as e: self.error = f"StockTwits unreachable ({type(e).__name__}) on {sym}" print(self.error, flush=True) self.write() return 60 rec, self.state[sym] = summarize(doc, self.state.get(sym), t) self.names[sym] = rec if self.error and self.error.startswith("StockTwits"): self.error = None # recovered self.write() b = f"{rec['bull_pct']}%" if rec["bull_pct"] is not None else "thin" print(f"{t.strftime('%H:%M:%S')}Z {sym:5} {rec['label']:7} {b:>5} vel {rec['vel']}/h" + (f" BUZZ {rec['buzz']}x" if rec["hot"] else ""), flush=True) return 0 def next_symbol(self): if time.time() - self.last_index >= INDEX_EVERY: self.last_index = time.time() self._pending_idx = list(INDEXES) if getattr(self, "_pending_idx", None): return self._pending_idx.pop(0) if not self.top: self.refresh_top() if not self.top: return None s = self.top[self.rr % len(self.top)] self.rr += 1 if self.rr % len(self.top) == 0: self.refresh_top() # pick up a new shortlist once per sweep return s def run(self, once=False): self.refresh_top() if once: for s in INDEXES + [x for x in self.top if x not in INDEXES]: self.one(s) time.sleep(self.pace) return while True: s = self.next_symbol() extra = self.one(s) if s else 60 time.sleep(self.pace + extra) def selftest(): ok = 0 def check(cond, what): nonlocal ok if not cond: raise AssertionError(what) ok += 1 t = parse_ts("2026-09-29T14:50:00Z") def msg(i, minute, tag=None, body="x"): return {"id": i, "body": body, "created_at": f"2026-09-29T14:{minute:02d}:00Z", "user": {"username": "u%d" % i}, "entities": {"sentiment": {"basic": tag} if tag else None}} # 30 posts over 10 minutes: 8 bull, 2 bear, 20 untagged msgs = [msg(i, 49 - i // 3, "Bullish" if i < 8 else "Bearish" if i < 10 else None, "$QQQ up\n\n now") for i in range(30)] rec, st = summarize({"messages": msgs, "symbol": {"watchlist_count": 5}}, None, t) check(rec["bull"] == 8 and rec["bear"] == 2 and rec["untagged"] == 20, "counts") check(rec["bull_pct"] == 80 and rec["label"] == "Bullish" and rec["tone"] == "bull", "bull label") check(rec["posts"][0]["text"] == "$QQQ up now", "whitespace cleaned") check(rec["posts"][0]["url"] == "https://stocktwits.com/u0/message/0", "post link") check(len(rec["posts"]) == MAX_POSTS, "post cap") check(rec["vel"] is not None and 150 < rec["vel"] < 250, f"velocity ~174/h, got {rec['vel']}") check(rec["buzz"] is None and not rec["hot"], "no buzz before a base exists") check(st["n"] == 1 and st["base"] == rec["vel"], "base seeded") # thin: only 4 tagged thin = [msg(i, 49 - i, "Bearish" if i < 4 else None) for i in range(30)] r2, _ = summarize({"messages": thin}, None, t) check(r2["bull_pct"] is None and r2["label"] == "thin", "thin under MIN_TAGGED") # bearish bear = [msg(i, 49 - i // 3, "Bearish" if i < 7 else "Bullish" if i < 9 else None) for i in range(30)] r3, _ = summarize({"messages": bear}, None, t) check(r3["bull_pct"] == 22 and r3["label"] == "Bearish", "bear label") # buzz: trusted base of 40/h, current ~174/h -> ~4.3x r4, st4 = summarize({"messages": msgs}, {"base": 40.0, "n": 6}, t) check(r4["hot"] and r4["buzz"] >= BUZZ_X, f"buzz fires, got {r4['buzz']}") check(st4["n"] == 7 and 40 < st4["base"] < r4["vel"], "base moves toward the new sample") # an untrusted base never fires r5, _ = summarize({"messages": msgs}, {"base": 10.0, "n": BASE_MIN - 1}, t) check(r5["buzz"] is None and not r5["hot"], "no buzz while learning") # idle stream: newest post 3h old reads slow even if its burst was fast old = [dict(m, created_at=m["created_at"].replace("T14:", "T11:")) for m in msgs] r6, _ = summarize({"messages": old}, None, t) check(r6["vel"] < 15, f"idle stream decays, got {r6['vel']}") # empty stream is not an error r7, _ = summarize({"messages": []}, None, t) check(r7["n"] == 0 and r7["vel"] is None and r7["label"] == "thin", "empty stream") # shortlist shapes import tempfile d = tempfile.mkdtemp() p = os.path.join(d, "s.json") for shape, want in [({"names": {"B": {"priority": 2}, "A": {"priority": 1}, "C": {"priority": 3}}}, ["A", "B"]), (["x", "y", "z"], ["X", "Y"]), ({"tickers": ["q", "r"]}, ["Q", "R"])]: with open(p, "w") as f: json.dump(shape, f) check(shortlist_top(p, 2) == want, f"shortlist shape {shape}") # the real file, if present if os.path.exists(SHORTLIST): top = shortlist_top() check(len(top) == min(TOP_N, len(top)) and all(s.isupper() for s in top), "real shortlist reads") print(f"selftest: {ok} assertions passed") if __name__ == "__main__": ap = argparse.ArgumentParser(description=__doc__.split("\n")[0]) ap.add_argument("--once", action="store_true") ap.add_argument("--pace", type=float, default=PACE) ap.add_argument("--symbols") ap.add_argument("--selftest", action="store_true") a = ap.parse_args() if a.selftest: selftest() sys.exit(0) syms = [s.strip().upper() for s in a.symbols.split(",") if s.strip()] if a.symbols else None print(f"sentiment_feed: writing {OUT}, one request every {a.pace:.0f}s. Ctrl+C to stop.", flush=True) try: Feed(syms, a.pace).run(once=a.once) except KeyboardInterrupt: pass