diff --git a/tests/test_post_screen.py b/tests/test_post_screen.py new file mode 100644 index 000000000..11c11fe42 --- /dev/null +++ b/tests/test_post_screen.py @@ -0,0 +1,288 @@ +"""Jev post screening, against TypeSafe's documented request and response shapes.""" + +import pytest +import requests + +from tradingagents.agents import post_screen as typesafe + +QUESTIONS = {"is_urgent": {"type": "noul", "instructions": "Does this convey urgency?"}} +ANSWERS = {"is_urgent": {"type": "noul", "noul": 0.95}} + + +class _Response: + def __init__(self, status, payload=None, headers=None): + self.status_code = status + self._payload = payload + self.headers = headers or {} + + def json(self): + if self._payload is None: + raise ValueError("no JSON") + return self._payload + + +class _Calls(list): + def __init__(self): + super().__init__() + self.queue = [] + self.sleeps = [] + + +@pytest.fixture +def post(monkeypatch): + """Queue responses on ``.queue``; the list records each call to requests.post.""" + calls = _Calls() + queue = calls.queue + + def fake_post(url, **kwargs): + calls.append((url, kwargs)) + item = queue.pop(0) + if isinstance(item, Exception): + raise item + return item + + monkeypatch.setattr(typesafe.requests, "post", fake_post) + monkeypatch.setattr(typesafe.time, "sleep", calls.sleeps.append) + monkeypatch.setenv("TYPESAFE_API_KEY", "ts-test") + monkeypatch.delenv("TYPESAFE_DEFAULT_MODEL", raising=False) + return calls + + +def _ok(): + return _Response(200, {"model": "jev-1.13.0", "answers": ANSWERS, + "usage": {"input_tokens": 296, "output_tokens": 20}}) + + +@pytest.mark.unit +def test_sends_the_documented_request_and_returns_the_answers(post): + post.queue.append(_ok()) + + assert typesafe.system_one("Help! My payouts have been failing.", QUESTIONS) == ANSWERS + + url, kwargs = post[0] + assert url == "https://api.typesafe.ai/v1/systemone" + assert kwargs["headers"]["Authorization"] == "Bearer ts-test" + assert kwargs["json"] == {"state": "Help! My payouts have been failing.", + "model": "jev-latest", "questions": QUESTIONS} + + +@pytest.mark.unit +def test_the_model_follows_the_sdk_environment(post, monkeypatch): + monkeypatch.setenv("TYPESAFE_DEFAULT_MODEL", "jev-1.13.0") + post.queue.append(_ok()) + + typesafe.system_one("s", QUESTIONS) + + assert post[0][1]["json"]["model"] == "jev-1.13.0" + + +@pytest.mark.unit +@pytest.mark.parametrize("transient", [ + _Response(429), _Response(529), requests.ConnectionError(), requests.Timeout(), + requests.exceptions.ChunkedEncodingError(), +]) +def test_rate_limits_overload_and_dropped_connections_are_retried(post, transient): + post.queue.extend([transient, _ok()]) + + assert typesafe.system_one("s", QUESTIONS) == ANSWERS + assert len(post) == 2 + + +@pytest.mark.unit +def test_a_retry_after_header_sets_the_wait(post): + post.queue.extend([_Response(429, headers={"retry-after": "7"}), _ok()]) + + typesafe.system_one("s", QUESTIONS) + + assert post.sleeps == [7.0] + + +@pytest.mark.unit +def test_a_long_retry_after_is_capped(post): + post.queue.extend([_Response(529, headers={"retry-after": "600"}), _ok()]) + + typesafe.system_one("s", QUESTIONS) + + assert post.sleeps == [typesafe._MAX_WAIT] + + +@pytest.mark.unit +def test_retries_are_bounded(post): + post.queue.extend([_Response(529)] * 3) + + with pytest.raises(typesafe.TypeSafeError, match="HTTP 529"): + typesafe.system_one("s", QUESTIONS) + assert len(post) == 3 + + +@pytest.mark.unit +@pytest.mark.parametrize("error", [requests.exceptions.InvalidHeader(), requests.exceptions.TooManyRedirects()]) +def test_other_request_errors_are_screening_failures_without_retry(post, error): + post.queue.append(error) + + with pytest.raises(typesafe.TypeSafeError, match=type(error).__name__): + typesafe.system_one("s", QUESTIONS) + assert len(post) == 1 + + +@pytest.mark.unit +@pytest.mark.parametrize("status", [401, 422, 500]) +def test_other_failures_raise_without_retry(post, status): + post.queue.append(_Response(status)) + + with pytest.raises(typesafe.TypeSafeError, match=f"HTTP {status}"): + typesafe.system_one("s", QUESTIONS) + assert len(post) == 1 + + +@pytest.mark.unit +@pytest.mark.parametrize("payload", [ + None, {"model": "jev"}, {"answers": {"other": {}}}, + {"answers": {"is_urgent": {"type": "choice", "choice": "yes"}}}, +]) +def test_a_response_without_every_answer_is_an_error(post, payload): + post.queue.append(_Response(200, payload)) + + with pytest.raises(typesafe.TypeSafeError, match="malformed"): + typesafe.system_one("s", QUESTIONS) + + + +def _post_answers(about: float, stance: str = "bullish", confidence: float = 0.9): + return _Response(200, {"model": "jev-1.13.0", "answers": { + "about": {"type": "noul", "noul": about}, + "stance": {"type": "choice", "choice": stance, "confidence": confidence, + "probabilities": {stance: 1.0}}, + }}) + + +@pytest.fixture +def jev(post, monkeypatch): + """Answer each post by its text: ``post`` maps text -> response.""" + answers = {} + + def fake_post(url, **kwargs): + post.append((url, kwargs)) + return answers[kwargs["json"]["state"]["post"]] + + monkeypatch.setattr(typesafe.requests, "post", fake_post) + monkeypatch.setattr(typesafe, "resolve_instrument_identity", + lambda t: {"company_name": "NVIDIA Corporation"}) + return answers + + +@pytest.mark.unit +def test_no_key_no_screen(monkeypatch): + monkeypatch.delenv("TYPESAFE_API_KEY", raising=False) + assert typesafe.jev_screen("NVDA") is None + + +@pytest.mark.unit +def test_each_post_is_its_own_state_under_the_fixed_questions(jev, post): + jev["NVDA to 200"] = _post_answers(0.9) + + typesafe.jev_screen("NVDA")(["NVDA to 200"]) + + body = post[0][1]["json"] + assert body["state"] == {"instrument": "NVIDIA Corporation (NVDA)", "post": "NVDA to 200"} + assert body["questions"] == typesafe.QUESTIONS + + +@pytest.mark.unit +def test_screen_drops_clear_off_topic_posts_and_counts_confident_stances(jev): + jev.update({ + "long NVDA": _post_answers(0.95, "bullish"), + "NVDA puts": _post_answers(0.9, "bearish"), + "maybe NVDA": _post_answers(0.4, "neutral"), # uncertain relevance: kept + "NVDA?": _post_answers(0.8, "bullish", 0.3), # uncertain stance: unclear + "NVDA!": _post_answers(0.8, "sideways"), # not an option: unclear + "$AAPL $MSFT $NVDA pump": _post_answers(0.1, "bullish"), + }) + + keep, note = typesafe.jev_screen("NVDA")(list(jev)) + + assert keep == [True, True, True, True, True, False] + assert note == ("Screened by Jev: 5 of the 6 posts fetched are about NVIDIA Corporation (NVDA); " + "their stance on its stock: 1 bullish, 1 bearish, 1 neutral, 2 unclear.") + + +@pytest.mark.unit +def test_one_failed_request_leaves_every_post_unscreened(jev): + jev.update({"a": _post_answers(0.1), "b": _Response(401)}) + + keep, note = typesafe.jev_screen("NVDA")(["a", "b"]) + + assert keep == [True, True] + assert note == "" + + +@pytest.mark.unit +def test_an_answer_missing_its_fields_reads_as_malformed(jev): + jev["a"] = _Response(200, {"answers": {"about": {"type": "noul"}, + "stance": {"type": "choice"}}}) + + keep, note = typesafe.jev_screen("NVDA")(["a"]) + + assert keep == [True] + assert note == "" + + +@pytest.mark.unit +def test_the_first_failure_cancels_the_requests_not_yet_sent(jev, post, monkeypatch): + monkeypatch.setattr(typesafe, "_WORKERS", 1) + jev.update({"a": _Response(401), **{f"p{i}": _post_answers(0.9) for i in range(20)}}) + + typesafe.jev_screen("NVDA")(list(jev)) + + assert len(post) < 21 + + +@pytest.mark.unit +def test_the_sentiment_analyst_hands_the_screen_to_both_social_fetchers(monkeypatch): + from langchain_core.messages import AIMessage + + from tradingagents.agents.analysts import sentiment_analyst + + screen = object() + seen = [] + monkeypatch.setattr(sentiment_analyst, "jev_screen", lambda ticker: screen) + monkeypatch.setattr(sentiment_analyst.get_news, "func", lambda *a: "news") + for name in ("fetch_stocktwits_messages", "fetch_reddit_posts"): + monkeypatch.setattr(sentiment_analyst, name, lambda *a, screen=None, **k: seen.append(screen) or "") + + class _LLM: + def with_structured_output(self, *a, **k): + raise NotImplementedError + + def invoke(self, messages): + return AIMessage(content="report") + + node = sentiment_analyst.create_sentiment_analyst(_LLM()) + node({"company_of_interest": "NVDA", "trade_date": "2026-01-09", "messages": []}) + + assert seen == [screen, screen] + + +@pytest.mark.unit +def test_a_failure_does_not_wait_for_requests_still_in_flight(jev, monkeypatch): + import threading + import time + + release = threading.Event() + + class _Slow: + status_code = 200 + headers = {} + + def json(self): + release.wait(5) + return _post_answers(0.9).json() + + jev.update({"slow": _Slow(), "bad": _Response(401)}) + started = time.monotonic() + keep, note = typesafe.jev_screen("NVDA")(["slow", "bad"]) + elapsed = time.monotonic() - started + release.set() + + assert keep == [True, True] and "unavailable" in note + assert elapsed < 2 diff --git a/tests/test_reddit_fallback.py b/tests/test_reddit_fallback.py index 77d50c1b8..2bd630fd2 100644 --- a/tests/test_reddit_fallback.py +++ b/tests/test_reddit_fallback.py @@ -271,3 +271,43 @@ def test_empty_subreddit_on_a_full_page_is_not_called_empty(): out = reddit.fetch_reddit_posts("NVDA", subreddits=("a", "b")) assert "r/b: " in out + + +@pytest.mark.unit +def test_an_unavailable_screen_keeps_every_post_and_says_so(): + posts = [{"title": "A1", "created_utc": None, "selftext": "", "subreddit": "a"}] + + def unavailable(texts): + return [True] * len(texts), "" + + with patch.object(reddit, "_fetch_subreddit_rss", return_value=posts): + screened = reddit.fetch_reddit_posts("NVDA", subreddits=("a", "b"), screen=unavailable) + plain = reddit.fetch_reddit_posts("NVDA", subreddits=("a", "b")) + assert screened == "\n\n" + plain diff --git a/tests/test_stocktwits_resilience.py b/tests/test_stocktwits_resilience.py index 429c0a138..6975b68b7 100644 --- a/tests/test_stocktwits_resilience.py +++ b/tests/test_stocktwits_resilience.py @@ -8,6 +8,7 @@ transport error must degrade to a placeholder rather than raise. from __future__ import annotations import http.client +import json from unittest.mock import patch from urllib.error import HTTPError @@ -75,3 +76,41 @@ class TestStockTwitsCryptoSymbols: with patch.object(stocktwits, "urlopen", side_effect=fake_urlopen): stocktwits.fetch_stocktwits_messages("BTC-USD") assert "/symbol/BTC.X.json" in seen["url"] + + +def _stream(*bodies): + payload = {"messages": [ + {"body": b, "created_at": "2026-01-09T15:00:00Z", "user": {"username": "u"}, + "entities": {"sentiment": {"basic": "Bullish"}}} + for b in bodies + ]} + + class _Resp: + def __enter__(self): + return self + + def __exit__(self, *a): + return False + + def read(self): + return json.dumps(payload).encode() + return _Resp() + + +def _drop_spam(texts): + return [not t.startswith("SPAM") for t in texts], "Screened: note" + + +@pytest.mark.unit +class TestStockTwitsScreening: + def test_screened_out_messages_leave_the_block_and_its_counts(self): + with patch.object(stocktwits, "urlopen", return_value=_stream("SPAM", "long NVDA")): + out = stocktwits.fetch_stocktwits_messages("NVDA", screen=_drop_spam) + assert out.startswith("Screened: note") + assert "long NVDA" in out and "SPAM" not in out + assert "Total: 1 most-recent" in out + + def test_all_screened_out_is_not_called_empty(self): + with patch.object(stocktwits, "urlopen", return_value=_stream("SPAM")): + out = stocktwits.fetch_stocktwits_messages("NVDA", screen=_drop_spam) + assert "none of the 1 StockTwits messages is about $NVDA" in out diff --git a/tradingagents/agents/analysts/sentiment_analyst.py b/tradingagents/agents/analysts/sentiment_analyst.py index 82f31f65a..ff2cf3623 100644 --- a/tradingagents/agents/analysts/sentiment_analyst.py +++ b/tradingagents/agents/analysts/sentiment_analyst.py @@ -7,7 +7,8 @@ prompt, so the model reports on data it was given rather than inventing posts: 2. StockTwits messages: the cashtag stream, with Bullish/Bearish tags 3. Reddit posts: r/wallstreetbets, r/stocks, r/investing -Each source is trimmed to the analysis window. These feeds serve recent items +Each source is trimmed to the analysis window. With a TypeSafe key, the social +posts are screened by Jev first (see post_screen). These feeds serve recent items and are not archived, so a historical run's sentiment inputs are not point-in-time. @@ -22,6 +23,7 @@ from langchain_core.messages import AIMessage from langchain_core.prompts import ChatPromptTemplate, MessagesPlaceholder from tradingagents.agents.context import get_instrument_context_from_state, get_language_instruction +from tradingagents.agents.post_screen import jev_screen from tradingagents.agents.schemas import SentimentReport, render_sentiment_report from tradingagents.agents.structured import ( NO_EXTERNAL_TOOLS, @@ -59,10 +61,11 @@ def create_sentiment_analyst(llm): news_block = get_news.func(ticker, start_date, end_date) # Pass the analysis window so a historical run trims social posts to it # instead of leaking today's chatter into a backtest (#1220). + screen = jev_screen(ticker) stocktwits_block = fetch_stocktwits_messages( - ticker, limit=30, start_date=start_date, end_date=end_date + ticker, limit=30, start_date=start_date, end_date=end_date, screen=screen ) - reddit_block = fetch_reddit_posts(ticker, start_date=start_date, end_date=end_date) + reddit_block = fetch_reddit_posts(ticker, start_date=start_date, end_date=end_date, screen=screen) system_message = _build_system_message( ticker=ticker, @@ -152,7 +155,7 @@ Community discussion, without vote or comment counts. Subreddit character matter ## How to analyze this data (best practices) -1. **Read the StockTwits Bullish/Bearish ratio as a leading retail-sentiment signal.** A 70/30 bullish/bearish split is moderately bullish; ≥90/10 may indicate over-extension and contrarian risk; 50/50 is uncertainty. Sample size matters — base rates on the actual message count, not percentages alone. +1. **Read the StockTwits Bullish/Bearish ratio as a leading retail-sentiment signal.** A 70/30 bullish/bearish split is moderately bullish; ≥90/10 may indicate over-extension and contrarian risk; 50/50 is uncertainty. Sample size matters — base rates on the actual message count, not percentages alone. A block headed "Screened by Jev" has had off-topic posts removed; its stance count is a classifier's read of every on-topic post fetched, labelled or not, of which the posts listed are a sample. Read it alongside the user tags. 2. **Look for cross-source divergences.** If news framing is bearish but StockTwits is overwhelmingly bullish, that mismatch is itself a signal — it can mean retail is leaning into a thesis the news flow hasn't caught up to (or vice versa, that retail is chasing while institutions are cautious). diff --git a/tradingagents/agents/post_screen.py b/tradingagents/agents/post_screen.py new file mode 100644 index 000000000..271c0009a --- /dev/null +++ b/tradingagents/agents/post_screen.py @@ -0,0 +1,174 @@ +"""Social-post screening with TypeSafe's Jev, when ``TYPESAFE_API_KEY`` is set. + +Jev answers typed questions with calibrated probabilities. Each StockTwits or +Reddit post is asked two: is it about the instrument, and which way does it +lean on the instrument's stock. Code turns the answers into what the Sentiment +Analyst reads: posts that are clearly about something else are dropped, and a +stance count over the rest heads the source's block. + +Configured by TypeSafe's own SDK variables, ``TYPESAFE_API_KEY`` and +``TYPESAFE_DEFAULT_MODEL``. Without a key nothing here runs; if any request fails, the source's posts are kept unscreened and the +block says screening was unavailable. +""" + +import logging +import os +import random +import time +from concurrent.futures import ThreadPoolExecutor, as_completed + +import requests + +from tradingagents.agents.context import resolve_instrument_identity + +logger = logging.getLogger(__name__) + +_URL = "https://api.typesafe.ai/v1/systemone" +_DEFAULT_MODEL = "jev-latest" +_RETRY_STATUSES = (429, 529) # rate limited, overloaded: back off and retry +_TRANSIENT = (requests.ConnectionError, requests.Timeout, requests.exceptions.ChunkedEncodingError) +_ATTEMPTS = 3 +_MAX_WAIT = 30.0 +_TIMEOUT = 15.0 +_WORKERS = 16 # well inside the documented 1,200 requests per minute + +# A post is dropped only on a clear "not about it"; the uncertain middle stays. +_OFF_TOPIC_BELOW = 0.3 +# A stance counts only when Jev is not genuinely uncertain about it. +_STANCE_CONFIDENCE = 0.5 + +QUESTIONS = { + "about": { + "type": "noul", + "instructions": "Is `post` about `instrument`: the company, its stock, its products or its outlook?", + "criteria": { + "true": "`post` discusses `instrument` itself.", + "false": "`post` names `instrument` only in passing or in a list of tickers, is spam or " + "promotion, or is about a different company.", + }, + }, + "stance": { + "type": "choice", + "instructions": "What does the author of `post` expect for the stock price of `instrument`?", + "criteria": { + "bullish": "The author expects `instrument`'s stock to rise, or is buying or holding it long.", + "bearish": "The author expects `instrument`'s stock to fall, or is selling or shorting it.", + "neutral": "The author gives no view of their own on `instrument`'s stock: a question, " + "news without opinion, or someone else's view quoted.", + }, + }, +} + + +class TypeSafeError(Exception): + """A System One request that did not produce answers.""" + + +def system_one(state, questions: dict) -> dict[str, dict]: + """Ask ``questions`` about ``state``; return the answers keyed by question name. + + Rate-limit and overload responses and dropped connections are retried with + backoff, honouring ``Retry-After``; any other failure raises + ``TypeSafeError`` at once. + """ + body = { + "state": state, + "model": os.environ.get("TYPESAFE_DEFAULT_MODEL") or _DEFAULT_MODEL, + "questions": questions, + } + headers = {"Authorization": f"Bearer {os.environ.get('TYPESAFE_API_KEY', '')}"} + backoff, retry_after = 1.0, None + for attempt in range(_ATTEMPTS): + if attempt: + time.sleep(retry_after if retry_after is not None else backoff * random.uniform(0.8, 1.2)) + backoff *= 2 + try: + response = requests.post(_URL, json=body, headers=headers, timeout=_TIMEOUT) + except requests.RequestException as exc: + failure, retry_after = type(exc).__name__, None + if isinstance(exc, _TRANSIENT): + continue + break + if response.status_code == 200: + return _answers(response, questions) + failure, retry_after = f"HTTP {response.status_code}", _retry_after(response) + if response.status_code not in _RETRY_STATUSES: + break + raise TypeSafeError(failure) + + +def _retry_after(response) -> float | None: + try: + return min(max(0.0, float(response.headers.get("retry-after"))), _MAX_WAIT) + except (TypeError, ValueError): + return None + + +def _answers(response, questions: dict) -> dict[str, dict]: + try: + payload = response.json() + answers = payload["answers"] + if all(answers[q]["type"] == spec["type"] for q, spec in questions.items()): + logger.debug("TypeSafe answered with %s", payload.get("model")) + return answers + except (ValueError, KeyError, TypeError): + pass + raise TypeSafeError("malformed response") + + +def _stance(answer: dict) -> str: + choice = answer["choice"] + if choice not in QUESTIONS["stance"]["criteria"] or answer["confidence"] < _STANCE_CONFIDENCE: + return "unclear" + return choice + + +def jev_screen(ticker: str): + """A post screen for the social fetchers, or None without a TypeSafe key. + + The screen takes the post texts and returns one keep flag per post and a + note line for the top of the source's block. + """ + if not os.environ.get("TYPESAFE_API_KEY"): + return None + name = resolve_instrument_identity(ticker).get("company_name") + instrument = f"{name} ({ticker})" if name else ticker + + def screen(posts: list[str]) -> tuple[list[bool], str]: + try: + answers = _ask_each(instrument, posts) + keep = [a["about"]["noul"] >= _OFF_TOPIC_BELOW for a in answers] + stances = [_stance(a["stance"]) for a, kept in zip(answers, keep, strict=True) if kept] + except (KeyError, TypeError): + return _unscreened(instrument, posts, "malformed response") + except TypeSafeError as exc: + return _unscreened(instrument, posts, str(exc)) + counts = ", ".join(f"{stances.count(s)} {s}" for s in ("bullish", "bearish", "neutral", "unclear")) + return keep, ( + f"Screened by Jev: {len(stances)} of the {len(posts)} posts fetched are about " + f"{instrument}; their stance on its stock: {counts}." + ) + + return screen + + +def _unscreened(instrument: str, posts: list[str], failure: str) -> tuple[list[bool], str]: + logger.warning("Jev screening failed for %s: %s", instrument, failure) + return [True] * len(posts), f"" + + +def _ask_each(instrument: str, posts: list[str]) -> list[dict]: + """One request per post. The first failure raises at once: requests not yet + sent are cancelled, and those in flight finish in the background unread.""" + answers: list = [None] * len(posts) + pool = ThreadPoolExecutor(max_workers=_WORKERS) + try: + futures = { + pool.submit(system_one, {"instrument": instrument, "post": post}, QUESTIONS): i + for i, post in enumerate(posts) + } + for future in as_completed(futures): + answers[futures[future]] = future.result() + finally: + pool.shutdown(wait=False, cancel_futures=True) + return answers diff --git a/tradingagents/dataflows/vendors/reddit.py b/tradingagents/dataflows/vendors/reddit.py index db8397bfe..c490ebabe 100644 --- a/tradingagents/dataflows/vendors/reddit.py +++ b/tradingagents/dataflows/vendors/reddit.py @@ -82,6 +82,7 @@ DEFAULT_SUBREDDITS = ("wallstreetbets", "stocks", "investing") # subreddits fits well inside one page, which keeps a high-volume subreddit from # crowding the others out of a combined search. _FEED_PAGE = 100 +_SCREEN_CHARS = 1000 # of a post's title and body sent for screening _SEARCH_LOOKBACK = timedelta(days=7) # matches t=week below @@ -238,6 +239,7 @@ def fetch_reddit_posts( timeout: float = 10.0, start_date: str | None = None, end_date: str | None = None, + screen=None, ) -> str: """Fetch recent Reddit posts mentioning ``ticker`` across finance subreddits and return them as a formatted plaintext block. @@ -250,6 +252,10 @@ def fetch_reddit_posts( When ``start_date``/``end_date`` (yyyy-mm-dd) are given, posts are trimmed to that window so a historical run does not leak current discussion into a backtest (#1220). + + ``screen``, when given, takes each post's title and body and returns a keep + flag per post and a note line that heads the block. It runs before the + per-subreddit cut, so the posts it keeps fill the slots. """ # Crypto reaches us as a Yahoo pair (BTC-USD); search Reddit for the base # ("BTC") so the query actually matches discussion instead of near-nothing. @@ -270,22 +276,34 @@ def fetch_reddit_posts( period = f"within {start_date}..{end_date}" if window else "in the past 7 days" return gap or f"" + def sub_of(p): + return p.get("subreddit") or (subreddits[0] if len(subreddits) == 1 else "unknown") + + note, screened_out = "", set() + if screen: + keep, note = screen([f"{p.get('title') or ''}\n{p.get('selftext') or ''}"[:_SCREEN_CHARS] + for p in posts]) + screened_out = {sub_of(p).lower() for p, kept in zip(posts, keep, strict=True) if not kept} + posts = [p for p, kept in zip(posts, keep, strict=True) if kept] + # Group by the subreddit each entry names, in the requested order. Nothing # is dropped: an unlabelled post from a one-subreddit request belongs to it, # and any other name gets its own block. by_sub = {s.lower(): (s, []) for s in subreddits} for p in posts: - name = p.get("subreddit") or (subreddits[0] if len(subreddits) == 1 else "unknown") - by_sub.setdefault(name.lower(), (name, []))[1].append(p) + by_sub.setdefault(sub_of(p).lower(), (sub_of(p), []))[1].append(p) page_full = len(fetched) >= _FEED_PAGE blocks = [] for sub, sub_posts in by_sub.values(): if not sub_posts: - blocks.append( - f"r/{sub}: " - if page_full else f"r/{sub}: " - ) + if sub.lower() in screened_out: + blocks.append(f"r/{sub}: ") + else: + blocks.append( + f"r/{sub}: " + if page_full else f"r/{sub}: " + ) continue sub_posts = sub_posts[:limit_per_sub] # the feed is newest-first lines = [f"r/{sub} — {len(sub_posts)} recent posts mentioning {ticker.upper()}:"] @@ -301,4 +319,4 @@ def fetch_reddit_posts( + (f"\n body excerpt: {selftext}" if selftext else "") ) blocks.append("\n".join(lines)) - return "\n\n".join(blocks) + return "\n\n".join(([note] if note else []) + blocks) diff --git a/tradingagents/dataflows/vendors/stocktwits.py b/tradingagents/dataflows/vendors/stocktwits.py index 987a279d6..b6985fc89 100644 --- a/tradingagents/dataflows/vendors/stocktwits.py +++ b/tradingagents/dataflows/vendors/stocktwits.py @@ -71,6 +71,7 @@ def fetch_stocktwits_messages( timeout: float = 10.0, start_date: str | None = None, end_date: str | None = None, + screen=None, ) -> str: """Fetch recent StockTwits messages for ``ticker`` and return them as a formatted plaintext block ready for prompt injection. @@ -80,6 +81,9 @@ def fetch_stocktwits_messages( public stream only serves recent messages, so a window it cannot reach is reported as unavailable rather than as silence. + ``screen``, when given, takes the message bodies and returns a keep flag per + message and a note line that heads the block. + Returns a placeholder string when the endpoint is unreachable, the symbol has no messages, or the response shape is unexpected — the caller never has to special-case None or exceptions. @@ -109,6 +113,14 @@ def fetch_stocktwits_messages( ) return f"" + note = "" + if screen: + keep, note = screen([m.get("body") or "" for m in messages]) + screened = len(messages) + messages = [m for m, kept in zip(messages, keep, strict=True) if kept] + if not messages: + return f"{note}\n\n" + lines = [] bullish = bearish = unlabeled = 0 for m in messages[:limit]: @@ -141,4 +153,4 @@ def fetch_stocktwits_messages( f"Unlabeled: {unlabeled} · " f"Total: {total} most-recent messages" ) - return summary + "\n\n" + "\n".join(lines) + return (f"{note}\n\n" if note else "") + summary + "\n\n" + "\n".join(lines)