| 1 | """StockTwits source for last30days — ticker/crypto topics only. |
| 2 | |
| 3 | StockTwits is a cashtag-native social network for traders. Every message can |
| 4 | carry a self-reported Bullish/Bearish tag, which makes it uniquely good at one |
| 5 | thing the other sources can't quantify: a sentiment *ratio* and retail *volume* |
| 6 | on a specific symbol. |
| 7 | |
| 8 | GATING: this source is only meaningful for financial topics. `detect_symbols()` |
| 9 | and `is_financial_topic()` are the gate — the pipeline must NOT register |
| 10 | stocktwits for a non-ticker topic (see INTEGRATION.md, step 3). Treat the output |
| 11 | as a direction/volume signal, never as analysis: StockTwits skews retail and |
| 12 | promotional, sentiment tags are self-reported, and bot/pump noise is common. |
| 13 | |
| 14 | API: public, no auth. Symbol stream + symbol search endpoints. Unauthenticated |
| 15 | quota is ~200 requests/hour and is rate-limited per IP — keep pagination small. |
| 16 | Respect StockTwits' API terms if this is ever shipped beyond personal use. |
| 17 | |
| 18 | NOTE: uses raw urllib rather than the shared `from . import http` helper. |
| 19 | Every call is wrapped in try/except and degrades to a partial/empty result, but |
| 20 | switching to the shared helper (429 Retry-After, retry budget, backoff) is a |
| 21 | known follow-up to match siblings like hackernews.py. |
| 22 | """ |
| 23 | |
| 24 | from __future__ import annotations |
| 25 | |
| 26 | import datetime |
| 27 | import json |
| 28 | import re |
| 29 | import sys |
| 30 | import time |
| 31 | import urllib.parse |
| 32 | import urllib.request |
| 33 | from typing import Any |
| 34 | |
| 35 | _UA = "Mozilla/5.0 (last30days stocktwits source)" |
| 36 | _STREAM_URL = "https://api.stocktwits.com/api/2/streams/symbol/{symbol}.json" |
| 37 | _SEARCH_URL = "https://api.stocktwits.com/api/2/search/symbols.json" |
| 38 | |
| 39 | # Topic must look financial before we even try to resolve a symbol. This is the |
| 40 | # coarse gate; symbol resolution is the fine gate. |
| 41 | _FINANCE_HINTS = re.compile( |
| 42 | # Unambiguous finance vocabulary only. Bare "share/token/coin/bull/bear" |
| 43 | # were removed: they misfire on general topics ("share files", "token |
| 44 | # limits", "coin collecting", "bear attacks") and would inject stock chatter |
| 45 | # into non-financial runs. |
| 46 | r"\b(stock|stocks|ticker|cashtag|equit(?:y|ies)|price target|" |
| 47 | r"earnings|premarket|pre-?market|after\s?hours|dividend|valuation|" |
| 48 | r"crypto|altcoin|defi|market cap|bullish|bearish|" |
| 49 | # Unambiguous crypto names so "bitcoin price" gates without a cashtag. |
| 50 | # Short aliases (eth, sol, ada, doge, ripple) stay OUT of the gate: they |
| 51 | # collide with everyday topics (ETH Zurich, ADA compliance, doge memes); |
| 52 | # they still resolve via _CRYPTO_ALIASES once the gate fires another way. |
| 53 | r"bitcoin|btc|ethereum|solana|dogecoin|cardano|xrp|" |
| 54 | r"\$[A-Za-z]{1,5}(?:\.[A-Z])?)\b", |
| 55 | re.IGNORECASE, |
| 56 | ) |
| 57 | _CASHTAG = re.compile(r"\$([A-Za-z]{1,5}(?:\.[A-Z])?)\b") |
| 58 | |
| 59 | # A small built-in crypto map so we don't burn a symbol-search call on the |
| 60 | # obvious ones. StockTwits uses the `.X` suffix for crypto symbols. |
| 61 | _CRYPTO_ALIASES = { |
| 62 | "bitcoin": "BTC.X", "btc": "BTC.X", |
| 63 | "ethereum": "ETH.X", "eth": "ETH.X", |
| 64 | "solana": "SOL.X", "sol": "SOL.X", |
| 65 | "dogecoin": "DOGE.X", "doge": "DOGE.X", |
| 66 | "ripple": "XRP.X", "xrp": "XRP.X", |
| 67 | "cardano": "ADA.X", "ada": "ADA.X", |
| 68 | } |
| 69 | |
| 70 | |
| 71 | def _log(msg: str) -> None: |
| 72 | try: |
| 73 | from . import log as _enginelog |
| 74 | _enginelog.source_log("StockTwits", msg, tty_only=False) |
| 75 | except Exception: # standalone / outside package |
| 76 | print(f"[StockTwits] {msg}", file=sys.stderr) |
| 77 | |
| 78 | |
| 79 | def _get_json(url: str, timeout: int = 20) -> dict[str, Any]: |
| 80 | req = urllib.request.Request(url, headers={"User-Agent": _UA}) |
| 81 | with urllib.request.urlopen(req, timeout=timeout) as resp: |
| 82 | return json.load(resp) |
| 83 | |
| 84 | |
| 85 | # --------------------------------------------------------------------------- # |
| 86 | # Gating + symbol resolution # |
| 87 | # --------------------------------------------------------------------------- # |
| 88 | |
| 89 | def is_financial_topic(topic: str) -> bool: |
| 90 | """Coarse gate: does the topic look like it's about a tradeable asset?""" |
| 91 | return bool(_CASHTAG.search(topic) or _FINANCE_HINTS.search(topic)) |
| 92 | |
| 93 | |
| 94 | def detect_symbols(topic: str, *, resolve: bool = True, max_symbols: int = 2) -> list[str]: |
| 95 | """Resolve a topic to StockTwits symbols. Returns [] for non-financial topics. |
| 96 | |
| 97 | Order of resolution: |
| 98 | 1. Explicit cashtags in the topic ($NOW, $BTC.X) — trusted as-is. |
| 99 | 2. Crypto name aliases (bitcoin -> BTC.X). |
| 100 | 3. StockTwits symbol-search API for a company/product name, but ONLY if the |
| 101 | topic also tripped the finance gate (so "Apple pie" never resolves AAPL). |
| 102 | |
| 103 | `resolve=False` skips the network call (useful for the cheap gate check). |
| 104 | """ |
| 105 | found: list[str] = [] |
| 106 | |
| 107 | for m in _CASHTAG.finditer(topic): |
| 108 | sym = m.group(1).upper() |
| 109 | if sym not in found: |
| 110 | found.append(sym) |
| 111 | |
| 112 | lowered = topic.lower() |
| 113 | for alias, sym in _CRYPTO_ALIASES.items(): |
| 114 | if re.search(rf"\b{re.escape(alias)}\b", lowered) and sym not in found: |
| 115 | found.append(sym) |
| 116 | |
| 117 | if found: |
| 118 | return found[:max_symbols] |
| 119 | |
| 120 | # No explicit symbol. Only hit the network if the topic looks financial. |
| 121 | if not resolve or not is_financial_topic(topic): |
| 122 | return [] |
| 123 | |
| 124 | # Strip finance noise words so the search query is just the entity name. |
| 125 | name = re.sub( |
| 126 | r"\b(stock|stocks|shares?|price|ticker|earnings|forecast|crypto|" |
| 127 | r"token|coin|news|today|now)\b", |
| 128 | "", topic, flags=re.IGNORECASE, |
| 129 | ).strip() |
| 130 | if not name: |
| 131 | return [] |
| 132 | try: |
| 133 | url = _SEARCH_URL + "?" + urllib.parse.urlencode({"q": name}) |
| 134 | data = _get_json(url) |
| 135 | for result in data.get("results", []): |
| 136 | sym = (result.get("symbol") or "").upper() |
| 137 | if sym and sym not in found: |
| 138 | found.append(sym) |
| 139 | if len(found) >= max_symbols: |
| 140 | break |
| 141 | if found: |
| 142 | _log(f"Resolved '{name}' -> {found}") |
| 143 | except Exception as e: # noqa: BLE001 — network/parse, degrade gracefully |
| 144 | _log(f"symbol search failed for '{name}': {e}") |
| 145 | return found[:max_symbols] |
| 146 | |
| 147 | |
| 148 | # --------------------------------------------------------------------------- # |
| 149 | # Fetch + parse # |
| 150 | # --------------------------------------------------------------------------- # |
| 151 | |
| 152 | _DEPTH = {"quick": 30, "default": 60, "deep": 120} |
| 153 | |
| 154 | |
| 155 | def search_stocktwits( |
| 156 | topic_or_symbol: str, |
| 157 | from_date: str | None = None, |
| 158 | to_date: str | None = None, |
| 159 | *, |
| 160 | depth: str = "default", |
| 161 | ) -> dict[str, Any]: |
| 162 | """Fetch the symbol stream for the topic. Paginates by cursor up to `depth`. |
| 163 | |
| 164 | Accepts either a raw topic (resolved via detect_symbols) or an explicit |
| 165 | symbol. Returns {"messages": [...], "symbols": [...], "watchlist": int}. |
| 166 | """ |
| 167 | symbols = ( |
| 168 | [topic_or_symbol.lstrip("$").upper()] |
| 169 | if _CASHTAG.fullmatch("$" + topic_or_symbol.lstrip("$")) |
| 170 | else detect_symbols(topic_or_symbol) |
| 171 | ) |
| 172 | if not symbols: |
| 173 | return {"messages": [], "symbols": [], "error": "no symbol resolved"} |
| 174 | |
| 175 | symbol = symbols[0] # primary symbol drives the stream |
| 176 | target = _DEPTH.get(depth, 60) |
| 177 | messages: list[dict[str, Any]] = [] |
| 178 | watchlist = None |
| 179 | cursor_max = None |
| 180 | try: |
| 181 | while len(messages) < target: |
| 182 | url = _STREAM_URL.format(symbol=urllib.parse.quote(symbol)) |
| 183 | if cursor_max: |
| 184 | url += f"?max={cursor_max}" |
| 185 | data = _get_json(url) |
| 186 | if watchlist is None: |
| 187 | watchlist = (data.get("symbol") or {}).get("watchlist_count") |
| 188 | batch = data.get("messages", []) |
| 189 | if not batch: |
| 190 | break |
| 191 | messages.extend(batch) |
| 192 | cursor = data.get("cursor", {}) |
| 193 | if not cursor.get("more") or not cursor.get("max"): |
| 194 | break |
| 195 | cursor_max = cursor["max"] |
| 196 | time.sleep(0.8) # be polite to the unauth quota |
| 197 | except Exception as e: # noqa: BLE001 |
| 198 | _log(f"stream fetch failed for {symbol}: {e}") |
| 199 | messages = _filter_by_date(messages, from_date, to_date) |
| 200 | return { |
| 201 | "messages": messages, |
| 202 | "symbols": symbols, |
| 203 | "error": str(e), |
| 204 | "freshness_window": { |
| 205 | "depth": depth, |
| 206 | "from_date": from_date, |
| 207 | "to_date": to_date, |
| 208 | }, |
| 209 | } |
| 210 | |
| 211 | messages = _filter_by_date(messages, from_date, to_date) |
| 212 | _log(f"{symbol}: {len(messages)} messages (watchlist {watchlist})") |
| 213 | return { |
| 214 | "messages": messages, |
| 215 | "symbols": symbols, |
| 216 | "watchlist": watchlist, |
| 217 | "freshness_window": { |
| 218 | "depth": depth, |
| 219 | "from_date": from_date, |
| 220 | "to_date": to_date, |
| 221 | }, |
| 222 | } |
| 223 | |
| 224 | |
| 225 | def _filter_by_date(messages: list[dict], from_date: str | None, to_date: str | None) -> list[dict]: |
| 226 | if not (from_date or to_date): |
| 227 | return messages |
| 228 | out = [] |
| 229 | for m in messages: |
| 230 | d = (m.get("created_at") or "")[:10] |
| 231 | if from_date and d and d < from_date: |
| 232 | continue |
| 233 | if to_date and d and d > to_date: |
| 234 | continue |
| 235 | out.append(m) |
| 236 | return out |
| 237 | |
| 238 | |
| 239 | def parse_stocktwits_response(response: dict[str, Any], query: str = "") -> list[dict[str, Any]]: |
| 240 | """Normalize the stream into engine-style item dicts (same keys as HN/Reddit). |
| 241 | |
| 242 | Each item carries metadata.sentiment in {"Bullish","Bearish",None} and the |
| 243 | symbol-level bull/bear aggregate so synthesis can cite the ratio. |
| 244 | """ |
| 245 | messages = response.get("messages", []) |
| 246 | symbols = response.get("symbols", []) |
| 247 | agg = aggregate_sentiment(messages) |
| 248 | items: list[dict[str, Any]] = [] |
| 249 | for i, m in enumerate(messages): |
| 250 | user = m.get("user") or {} |
| 251 | username = user.get("username") or "unknown" |
| 252 | body = (m.get("body") or "").strip() |
| 253 | sentiment = ((m.get("entities") or {}).get("sentiment") or {}).get("basic") |
| 254 | likes = (m.get("likes") or {}).get("total", 0) or 0 |
| 255 | reshares = (m.get("reshares") or {}).get("reshared_count", 0) or 0 |
| 256 | followers = user.get("followers", 0) or 0 |
| 257 | # Relevance: cashtag-native source, so on-symbol is a near-given. Nudge |
| 258 | # by author reach + a tagged-sentiment bonus (tagged posts are higher |
| 259 | # intent than chatter). |
| 260 | relevance = min(1.0, 0.7 + (0.1 if sentiment else 0.0) + min(0.2, followers / 50000)) |
| 261 | items.append({ |
| 262 | "id": str(m.get("id") or f"ST{i+1}"), |
| 263 | "title": body[:120] or f"${symbols[0] if symbols else ''} post", |
| 264 | "url": f"https://stocktwits.com/{username}/message/{m.get('id')}", |
| 265 | "author": username, |
| 266 | "date": (m.get("created_at") or "")[:10] or None, |
| 267 | "engagement": {"likes": likes, "reshares": reshares, "followers": followers}, |
| 268 | "relevance": round(relevance, 2), |
| 269 | "why_relevant": f"StockTwits ${symbols[0] if symbols else ''} post" |
| 270 | + (f" tagged {sentiment}" if sentiment else ""), |
| 271 | "snippet": body[:400], |
| 272 | "metadata": { |
| 273 | "sentiment": sentiment, |
| 274 | "symbol": symbols[0] if symbols else None, |
| 275 | "sentiment_aggregate": agg, # same dict on every item; cheap, lets synthesis cite it |
| 276 | "watchlist": response.get("watchlist"), |
| 277 | "freshness_window": response.get("freshness_window"), |
| 278 | }, |
| 279 | }) |
| 280 | return items |
| 281 | |
| 282 | |
| 283 | def aggregate_sentiment(messages: list[dict[str, Any]]) -> dict[str, Any]: |
| 284 | """Bull/bear counts + ratio over the sentiment-tagged subset.""" |
| 285 | bull = bear = 0 |
| 286 | for m in messages: |
| 287 | s = ((m.get("entities") or {}).get("sentiment") or {}).get("basic") |
| 288 | if s == "Bullish": |
| 289 | bull += 1 |
| 290 | elif s == "Bearish": |
| 291 | bear += 1 |
| 292 | tagged = bull + bear |
| 293 | return { |
| 294 | "bullish": bull, |
| 295 | "bearish": bear, |
| 296 | "untagged": len(messages) - tagged, |
| 297 | "pct_bullish": round(100 * bull / tagged) if tagged else None, |
| 298 | "sample": len(messages), |
| 299 | } |
| 300 | |
| 301 | |
| 302 | def refetch_datum(item: Any, datum_key: str) -> dict[str, Any]: |
| 303 | """Re-fetch the same paginated, date-filtered symbol-stream population.""" |
| 304 | from . import http |
| 305 | |
| 306 | if datum_key != "pct_bullish": |
| 307 | raise KeyError(f"Unsupported StockTwits datum: {datum_key}") |
| 308 | symbol = str(item.metadata.get("symbol") or item.container or "").strip().upper() |
| 309 | if not symbol: |
| 310 | raise ValueError("StockTwits item has no symbol") |
| 311 | url = _STREAM_URL.format(symbol=urllib.parse.quote(symbol)) |
| 312 | window = item.metadata.get("freshness_window") or {} |
| 313 | depth = str(window.get("depth") or "default") |
| 314 | target = _DEPTH.get(depth, _DEPTH["default"]) |
| 315 | messages: list[dict[str, Any]] = [] |
| 316 | cursor_max = None |
| 317 | while len(messages) < target: |
| 318 | request_kwargs: dict[str, Any] = {"timeout": 10, "retries": 2} |
| 319 | if cursor_max: |
| 320 | request_kwargs["params"] = {"max": cursor_max} |
| 321 | data = http.request("GET", url, **request_kwargs) |
| 322 | if not isinstance(data, dict) or not isinstance(data.get("messages"), list): |
| 323 | raise KeyError("StockTwits symbol stream was not returned") |
| 324 | batch = data["messages"] |
| 325 | if not batch: |
| 326 | break |
| 327 | messages.extend(batch) |
| 328 | cursor = data.get("cursor") or {} |
| 329 | if not cursor.get("more") or not cursor.get("max"): |
| 330 | break |
| 331 | cursor_max = cursor["max"] |
| 332 | messages = _filter_by_date( |
| 333 | messages, |
| 334 | window.get("from_date"), |
| 335 | window.get("to_date"), |
| 336 | ) |
| 337 | aggregate = aggregate_sentiment(messages) |
| 338 | value = aggregate.get("pct_bullish") |
| 339 | if value is None: |
| 340 | raise KeyError("StockTwits stream has no tagged sentiment") |
| 341 | newest = max( |
| 342 | (str(message.get("created_at") or "") for message in messages), |
| 343 | default="", |
| 344 | ) |
| 345 | return { |
| 346 | "value": value, |
| 347 | "values": {"pct_bullish": value}, |
| 348 | "url": item.url, |
| 349 | "timestamp": newest or None, |
| 350 | } |
| 351 | |
| 352 | |
| 353 | # --------------------------------------------------------------------------- # |
| 354 | # Standalone CLI (ad-hoc use today, before any engine wiring) # |
| 355 | # python3 stocktwits.py "ServiceNow stock" # |
| 356 | # --------------------------------------------------------------------------- # |
| 357 | if __name__ == "__main__": |
| 358 | topic = " ".join(sys.argv[1:]) or "$NOW" |
| 359 | if not is_financial_topic(topic) and not detect_symbols(topic, resolve=False): |
| 360 | print(f"Not a ticker/crypto topic — skipping StockTwits: {topic!r}") |
| 361 | raise SystemExit(0) |
| 362 | today = datetime.date.today() |
| 363 | since = (today - datetime.timedelta(days=30)).isoformat() |
| 364 | resp = search_stocktwits(topic, from_date=since, depth="default") |
| 365 | if resp.get("error") and not resp.get("messages"): |
| 366 | print("error:", resp["error"]); raise SystemExit(1) |
| 367 | items = parse_stocktwits_response(resp, query=topic) |
| 368 | agg = aggregate_sentiment(resp["messages"]) |
| 369 | print(f"symbol(s): {resp.get('symbols')} | watchlist {resp.get('watchlist')}") |
| 370 | print(f"sentiment: {agg['bullish']} bull / {agg['bearish']} bear " |
| 371 | f"({agg['pct_bullish']}% bullish of tagged) over {agg['sample']} msgs") |
| 372 | for it in sorted(items, key=lambda x: x["engagement"]["likes"], reverse=True)[:8]: |
| 373 | s = it["metadata"]["sentiment"] or "-" |
| 374 | print(f" [{it['engagement']['likes']}♥ {s}] @{it['author']}: {it['snippet'][:120]}") |
| 375 |