From 7cc478ad07c3ace6dfdbbf3cffa2530cdb95db2b Mon Sep 17 00:00:00 2001 From: Yijia-Xiao Date: Mon, 7 Sep 2026 20:54:19 +0000 Subject: [PATCH] fix(dataflows): report a failed Reddit fetch as unavailable, not silence - a failed fetch and an empty search both returned [], so a 429 rendered as 'no posts found' and the sentiment analyst read throttling as real silence; when every subreddit was throttled the summary asserted it outright - a failed fetch now returns None and renders as unavailable, and the summary only claims silence for subreddits actually searched - raise the headerless-429 back-off to 60s, which is where a retry starts succeeding; pay it at most once per run so three throttled subreddits do not stall the analysis, and match the Retry-After cap to it #1295 --- tests/test_reddit_fallback.py | 77 +++++++++++++++++++++++++++---- tradingagents/dataflows/reddit.py | 75 ++++++++++++++++++++++++------ 2 files changed, 129 insertions(+), 23 deletions(-) diff --git a/tests/test_reddit_fallback.py b/tests/test_reddit_fallback.py index ec9a89fdd..a2bc2cc25 100644 --- a/tests/test_reddit_fallback.py +++ b/tests/test_reddit_fallback.py @@ -86,9 +86,9 @@ class TestRssParsing: assert posts[0]["created_utc"] > 0 assert "datacenter unit" in posts[0]["selftext"] - def test_malformed_xml_fails_open(self): + def test_malformed_xml_reports_unavailable(self): with patch.object(reddit, "urlopen", return_value=_resp(lambda: b"<>")): - assert reddit._fetch_subreddit_rss("NVDA", "stocks", 5, 5.0) == [] + assert reddit._fetch_subreddit_rss("NVDA", "stocks", 5, 5.0) is None @pytest.mark.unit @@ -139,7 +139,7 @@ class TestRss429Backoff: patch.object(reddit.time, "sleep"): posts = reddit._fetch_subreddit_rss("NVDA", "stocks", 5, 5.0) assert op.call_count == 2 # one retry, then gives up cleanly - assert posts == [] + assert posts is None def test_retry_after_header_is_honoured(self): err = HTTPError("url", 429, "Too Many Requests", {"Retry-After": "12"}, None) @@ -166,7 +166,7 @@ class TestRss429Backoff: reddit._fetch_subreddit_rss("NVDA", "stocks", 5, 5.0) slept.assert_called_once() (wait,), _ = slept.call_args - assert 4.0 <= wait <= 6.0 # 5s +/-20% jitter + assert 48.0 <= wait <= 72.0 # 60s +/-20% jitter @pytest.mark.unit @@ -174,9 +174,9 @@ class TestChunkedTransferErrorsHandled: """IncompleteRead/RemoteDisconnected come from http.client and are NOT OSErrors, so they were previously uncaught and crashed the pipeline (#1024).""" - def test_rss_incomplete_read_degrades_to_empty(self): + def test_rss_incomplete_read_reports_unavailable(self): with patch.object(reddit, "urlopen", return_value=_raise(http.client.IncompleteRead(b""))): - assert reddit._fetch_subreddit_rss("NVDA", "stocks", 5, 5.0) == [] + assert reddit._fetch_subreddit_rss("NVDA", "stocks", 5, 5.0) is None def test_json_incomplete_read_falls_back_to_rss(self): with patch.object(reddit, "urlopen", return_value=_raise(http.client.IncompleteRead(b""))), \ @@ -190,7 +190,7 @@ class TestChunkedTransferErrorsHandled: big = _resp(lambda: b"x" * 100) with patch.object(reddit, "_MAX_FEED_BYTES", 10), \ patch.object(reddit, "urlopen", return_value=big): - assert reddit._fetch_subreddit_rss("NVDA", "stocks", 5, 5.0) == [] + assert reddit._fetch_subreddit_rss("NVDA", "stocks", 5, 5.0) is None @pytest.mark.unit @@ -228,7 +228,7 @@ class TestCryptoSearchTerm: def _captured_ticker(self, ticker): seen = {} - def fake_fetch(t, sub, limit, timeout): + def fake_fetch(t, sub, limit, timeout, **kwargs): seen["ticker"] = t return [] @@ -241,3 +241,64 @@ class TestCryptoSearchTerm: def test_equity_passes_through(self): assert self._captured_ticker("NVDA") == "NVDA" + + +@pytest.mark.unit +class TestFailedFetchIsNotSilence: + """A throttled fetch must not be rendered as "no posts found" (#1295). + + Returning [] for both a failed request and a genuinely empty search made the + sentiment analyst read rate limiting as real silence ("r/stocks and + r/investing are silent"), which is a signal that was never observed. + """ + + _POST = { + "title": "NVDA pops", "score": None, "num_comments": None, + "created_utc": reddit._iso_to_timestamp("2026-05-20T14:30:00Z"), + "selftext": "", "source": "rss", + } + + def _run(self, results): + """Drive fetch_reddit_posts with a per-subreddit result sequence.""" + subs = tuple(f"s{i}" for i in range(len(results))) + with patch.object(reddit, "_fetch_subreddit", side_effect=list(results)): + return reddit.fetch_reddit_posts( + "NVDA", subreddits=subs, inter_request_delay=0 + ) + + def test_failed_subreddit_is_marked_unavailable_not_empty(self): + out = self._run([None, [self._POST]]) + assert "unavailable" in out + assert "no posts found" not in out.split("unavailable")[0] + + def test_all_sources_failing_does_not_claim_no_posts(self): + out = self._run([None, None]) + assert "Reddit unavailable" in out + assert "no Reddit posts found" not in out + + def test_mixed_failure_and_empty_only_claims_silence_for_searched_subs(self): + # s0 failed, s1 genuinely returned nothing: the "no posts" claim must + # cover only s1, with s0 reported separately as unavailable. + out = self._run([None, []]) + assert "r/s1" in out.split("unavailable (fetch failed)")[0] + assert "unavailable (fetch failed): r/s0" in out + + def test_genuine_empty_still_reports_no_posts(self): + out = self._run([[], []]) + assert "no Reddit posts found" in out + assert "unavailable" not in out + + def test_retry_is_not_spent_again_after_a_failure(self): + # The 60s back-off must be paid at most once per run, so subsequent + # subreddits are fetched with retry disabled rather than stalling. + seen = [] + + def record(t, sub, limit, timeout, _retry=True): + seen.append(_retry) + return None + + with patch.object(reddit, "_fetch_subreddit", side_effect=record): + reddit.fetch_reddit_posts( + "NVDA", subreddits=("a", "b", "c"), inter_request_delay=0 + ) + assert seen == [True, False, False] diff --git a/tradingagents/dataflows/reddit.py b/tradingagents/dataflows/reddit.py index f8f0e1755..502c84d68 100644 --- a/tradingagents/dataflows/reddit.py +++ b/tradingagents/dataflows/reddit.py @@ -10,6 +10,10 @@ off once (honouring ``Retry-After``). RSS lacks score / comment counts, so those posts are marked and the formatter omits the metrics rather than printing fake zeros. +A fetch that fails is reported as ````, never as "no posts found": +the two are different claims, and passing a rate-limited fetch off as silence +hands the sentiment analyst a signal that was never observed (#1295). + No API key required. Returns formatted plaintext blocks ready for prompt injection and degrades gracefully — returns a placeholder string rather than raising, so callers never special-case missing data. @@ -102,9 +106,12 @@ def _strip_html(content: str) -> str: return " ".join(html.unescape(text).split()) -# Headerless-429 backoff when Reddit gives no Retry-After. Jittered so several -# analyses sharing an IP don't retry in lockstep and re-collide on the limit. -_RETRY_FALLBACK_SECONDS = 5.0 +# Headerless-429 backoff when Reddit gives no Retry-After. Measured against +# /r/{sub}/search.rss, a retry still 429s at 8s, 10s and 30s of spacing and +# succeeds at 60s, so a shorter wait spends the one retry on a request that +# cannot succeed (#1295). Jittered so several analyses sharing an IP don't +# retry in lockstep and re-collide on the limit. +_RETRY_FALLBACK_SECONDS = 60.0 def _jitter(seconds: float, frac: float = 0.2) -> float: @@ -114,14 +121,18 @@ def _jitter(seconds: float, frac: float = 0.2) -> float: def _retry_after_seconds(exc: HTTPError) -> float | None: - """Seconds to wait from a 429's ``Retry-After`` header, capped at 30s. + """Seconds to wait from a 429's ``Retry-After`` header, capped at 60s. + + The cap matches ``_RETRY_FALLBACK_SECONDS``: honouring less than we would + wait on our own would spend the one retry on a request we already know is + too early. Returns ``None`` only when the header is absent or unparseable; a valid ``Retry-After: 0`` returns ``0.0`` (retry at once), not ``None``. """ try: val = exc.headers.get("Retry-After") if getattr(exc, "headers", None) else None - return min(float(val), 30.0) if val is not None else None + return min(float(val), 60.0) if val is not None else None except (ValueError, TypeError, AttributeError): return None @@ -149,13 +160,18 @@ def _fetch_subreddit_rss( limit: int, timeout: float, _retry: bool = True, -) -> list[dict]: +) -> list[dict] | None: """Default path: parse the public Atom search feed for a subreddit. Carries no score / comment counts, so those fields are left None and the post is tagged ``source="rss"`` for honest display. On a 429 (Reddit's per-IP rate limit) we back off once — honouring ``Retry-After`` when present — before giving up, so a transient burst doesn't blank the feed. + + Returns ``[]`` when the search ran and matched nothing, and ``None`` when + the fetch itself failed. The caller must keep these apart: rendering a + failed fetch as "no posts found" hands the sentiment analyst an absence of + discussion that was never observed (#1295). """ url = _RSS.format(sub=sub, qs=_search_qs(ticker, limit)) req = Request(url, headers={"User-Agent": _UA}) @@ -175,12 +191,12 @@ def _fetch_subreddit_rss( time.sleep(wait) return _fetch_subreddit_rss(ticker, sub, limit, timeout, _retry=False) logger.warning("Reddit RSS fetch failed for r/%s · %s: %s", sub, ticker, exc) - return [] + return None except (OSError, http.client.HTTPException, ET.ParseError) as exc: # OSError covers URLError/TimeoutError/connection resets; HTTPException # covers chunked-transfer errors (IncompleteRead/BadStatusLine, #1024). logger.warning("Reddit RSS fetch failed for r/%s · %s: %s", sub, ticker, exc) - return [] + return None posts = [] for entry in root.findall("atom:entry", _ATOM_NS)[:limit]: @@ -234,14 +250,15 @@ def _fetch_subreddit( sub: str, limit: int, timeout: float, -) -> list[dict]: - """Fetch one subreddit, RSS-first. + _retry: bool = True, +) -> list[dict] | None: + """Fetch one subreddit, RSS-first. ``None`` means the fetch failed. The JSON search endpoint is reliably WAF-blocked (403) for public clients, so we go straight to the RSS feed — which serves our identified User-Agent reliably — halving our request volume against Reddit's per-IP rate limit. """ - return _fetch_subreddit_rss(ticker, sub, limit, timeout) + return _fetch_subreddit_rss(ticker, sub, limit, timeout, _retry=_retry) def fetch_reddit_posts( @@ -267,13 +284,26 @@ def fetch_reddit_posts( # 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. ticker = crypto_base(ticker) or ticker + subreddits = list(subreddits) blocks = [] total_posts = 0 + unavailable = [] + allow_retry = True for i, sub in enumerate(subreddits): if i > 0 and inter_request_delay: time.sleep(_jitter(inter_request_delay)) - posts = _within_window(_fetch_subreddit(ticker, sub, limit_per_sub, timeout), - start_date, end_date) + fetched = _fetch_subreddit(ticker, sub, limit_per_sub, timeout, _retry=allow_retry) + if fetched is None: + # A failed fetch is not an absence of discussion, so it must not be + # rendered as "no posts found" (#1295). One failure also means the + # per-IP budget is likely gone, so skip the (now 60s) back-off on + # the remaining subreddits rather than stalling the run on retries + # that cannot succeed; #1286 tracks coordinating this properly. + allow_retry = False + unavailable.append(sub) + blocks.append(f"r/{sub}: ") + continue + posts = _within_window(fetched, start_date, end_date) total_posts += len(posts) if not posts: blocks.append(f"r/{sub}: ") @@ -306,8 +336,23 @@ def fetch_reddit_posts( blocks.append("\n".join(lines)) if total_posts == 0: - return ( + searched = [s for s in subreddits if s not in unavailable] + if not searched: + # Every source failed: claiming "no posts" here would assert a + # silence we never observed. + return ( + f"" + ) + summary = ( f"" + f"{', '.join(f'r/{s}' for s in searched)} in the past 7 days>" ) + if unavailable: + summary += ( + f"\n" + ) + return summary return "\n\n".join(blocks)