harden(dataflows): bound the Reddit feed read before parsing

- ElementTree does not resolve external entities, so the reported XXE flag
  doesn't apply; the real residual is an unbounded read of untrusted network XML
- cap both the RSS and JSON reads at 5 MiB; overflow degrades to empty / RSS
  fallback through the existing failure paths #1206 #1276

tests/test_fred.py

@@ -151,22 +151,46 @@ class FredFormattingTests(unittest.TestCase):
         self.assertEqual(obs_params["observation_start"], "2025-07-02")  # 90d back
 
     def test_requests_pin_the_data_vintage(self):
-        # #1275: both the metadata and observations requests must set
-        # realtime_start=realtime_end=curr_date, or FRED serves the latest
-        # revision and revision-prone series leak future information.
+        # #1275: both the metadata and observations requests must pin the vintage
+        # to curr_date (clamped to FRED's today), or FRED serves the latest
+        # revision and revision-prone series leak future information. A past
+        # curr_date sits below FRED's today, so it pins through unchanged.
         captured = {}
 
         def _capture(path, params):
             captured[path] = params
             return _META if path == "series" else _OBS
 
-        with mock.patch.object(fred, "_request", side_effect=_capture):
+        with mock.patch.object(fred, "_fred_today", return_value="2026-01-01"), \
+                mock.patch.object(fred, "_request", side_effect=_capture):
             fred.get_macro_data("cpi", "2025-09-30", 90)
 
         for path in ("series", "series/observations"):
             self.assertEqual(captured[path]["realtime_start"], "2025-09-30", path)
             self.assertEqual(captured[path]["realtime_end"], "2025-09-30", path)
 
+    def test_future_curr_date_clamps_vintage_to_fred_today(self):
+        # #1275 regression: on a live run curr_date is the caller's LOCAL date,
+        # which can be a day ahead of FRED's US-Central clock. Pinning the vintage
+        # to that future date 400s, and the routing layer then drops macro data
+        # silently. The pin must clamp to FRED's today; the observation window
+        # (future bars can't exist yet) stays at curr_date.
+        captured = {}
+
+        def _capture(path, params):
+            captured[path] = params
+            return _META if path == "series" else _OBS
+
+        with mock.patch.object(fred, "_fred_today", return_value="2026-08-31"), \
+                mock.patch.object(fred, "_request", side_effect=_capture):
+            fred.get_macro_data("cpi", "2026-09-01", 90)  # local a day ahead of Chicago
+
+        for path in ("series", "series/observations"):
+            self.assertEqual(captured[path]["realtime_start"], "2026-08-31", path)
+            self.assertEqual(captured[path]["realtime_end"], "2026-08-31", path)
+        # the observation window still tracks curr_date, not the clamped vintage
+        self.assertEqual(captured["series/observations"]["observation_end"], "2026-09-01")
+
 
 @pytest.mark.unit
 class FredRoutingTests(unittest.TestCase):

tests/test_reddit_fallback.py

@@ -36,8 +36,9 @@ def _resp(read_fn):
         def __exit__(self_inner, *a):
             return False
 
-        def read(self_inner):
-            return read_fn()
+        def read(self_inner, size=-1):
+            data = read_fn()
+            return data if size is None or size < 0 else data[:size]
     return _Resp()
 
 
@@ -147,6 +148,26 @@ class TestRss429Backoff:
             reddit._fetch_subreddit_rss("NVDA", "stocks", 5, 5.0)
         slept.assert_called_once_with(12.0)
 
+    def test_retry_after_zero_is_honoured_not_treated_as_absent(self):
+        # A valid "Retry-After: 0" means retry at once; it must not fall through
+        # to the fallback wait (the earlier `or 5.0` bug turned 0 into 5s).
+        err = HTTPError("url", 429, "Too Many Requests", {"Retry-After": "0"}, None)
+        with patch.object(reddit, "urlopen", side_effect=[err, _atom_resp()]), \
+             patch.object(reddit.time, "sleep") as slept:
+            reddit._fetch_subreddit_rss("NVDA", "stocks", 5, 5.0)
+        slept.assert_called_once_with(0.0)
+
+    def test_headerless_429_fallback_is_jittered(self):
+        # No Retry-After -> our own ~5s fallback, jittered so concurrent runs
+        # don't retry in lockstep (kept within a tight band).
+        err = HTTPError("url", 429, "Too Many Requests", {}, None)
+        with patch.object(reddit, "urlopen", side_effect=[err, _atom_resp()]), \
+             patch.object(reddit.time, "sleep") as slept:
+            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
+
 
 @pytest.mark.unit
 class TestChunkedTransferErrorsHandled:
@@ -163,6 +184,14 @@ class TestChunkedTransferErrorsHandled:
             reddit._fetch_subreddit_json("NVDA", "stocks", 5, 5.0)
         rss.assert_called_once()
 
+    def test_oversized_rss_feed_is_refused_not_parsed(self):
+        # A hostile/misbehaving endpoint streaming an unbounded body must not be
+        # read into memory before parsing; overflow degrades to an empty feed.
+        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) == []
+
 
 @pytest.mark.unit
 class TestFormatterHandlesRssPosts:

tradingagents/agents/managers/portfolio_manager.py

@@ -62,7 +62,7 @@ def create_portfolio_manager(llm):
 
 ---
 
-Be decisive and ground every conclusion in specific evidence from the analysts.
+Ground every conclusion in specific evidence from the analysts. Commit to a directional call only when the evidence clearly supports one; choose Hold when the case is balanced, materially conflicting, ambiguous, or insufficient to justify changing exposure, rather than forcing a direction to appear decisive. Weigh the analysts on their merits, independent of speaking order.
 
 {NO_EXTERNAL_TOOLS}{get_language_instruction()}"""
 

tradingagents/agents/managers/research_manager.py

@@ -36,7 +36,7 @@ def create_research_manager(llm):
 - **Underweight**: Cautious view; recommend trimming exposure
 - **Sell**: Strong conviction in the bear thesis; recommend exiting or avoiding the position
 
-Commit to a clear stance whenever the debate's strongest arguments warrant one; reserve Hold for situations where the evidence on both sides is genuinely balanced.
+Commit to a directional stance only when the debate's strongest arguments clearly warrant one. Choose Hold when the evidence is balanced, materially conflicting, ambiguous, or insufficient to justify changing exposure; do not manufacture a direction merely to appear decisive. Weigh the bull and bear cases on their merits, independent of which side spoke first or last.
 
 ---
 

tradingagents/agents/schemas.py

@@ -82,9 +82,11 @@ class ResearchPlan(BaseModel):
     recommendation: PortfolioRating = Field(
         description=(
             "The investment recommendation. Exactly one of Buy / Overweight / "
-            "Hold / Underweight / Sell. Reserve Hold for situations where the "
-            "evidence on both sides is genuinely balanced; otherwise commit to "
-            "the side with the stronger arguments."
+            "Hold / Underweight / Sell. Choose Hold when the evidence is "
+            "balanced, materially conflicting, ambiguous, or insufficient to "
+            "justify changing exposure; otherwise commit to the side with the "
+            "clearly stronger arguments. Do not pick a direction merely to be "
+            "decisive."
         ),
     )
     rationale: str = Field(
@@ -197,7 +199,10 @@ class PortfolioDecision(BaseModel):
     rating: PortfolioRating = Field(
         description=(
             "The final position rating. Exactly one of Buy / Overweight / Hold / "
-            "Underweight / Sell, picked based on the analysts' debate."
+            "Underweight / Sell, picked based on the analysts' debate. Choose "
+            "Hold when the case is balanced, materially conflicting, ambiguous, "
+            "or insufficient to justify changing exposure, rather than forcing a "
+            "direction to appear decisive."
         ),
     )
     executive_summary: str = Field(

tradingagents/dataflows/fred.py

@@ -12,6 +12,7 @@ import logging
 import os
 from datetime import datetime, timedelta
 
+import pytz
 import requests
 
 from .errors import VendorNotConfiguredError
@@ -20,6 +21,12 @@ logger = logging.getLogger(__name__)
 
 FRED_API_BASE = "https://api.stlouisfed.org/fred"
 
+# FRED's realtime clock runs on US Central (St. Louis Fed). It rejects a
+# realtime date in its own future with a 400, so the vintage pin is clamped to
+# this rather than the caller's local date (#1275). pytz (already a dependency)
+# bundles its own tz database, so this works where system tzdata is absent.
+FRED_TZ = pytz.timezone("America/Chicago")
+
 # Network timeout (seconds) so a stalled request can't hang the agents,
 # mirroring the Alpha Vantage client.
 REQUEST_TIMEOUT = 30
@@ -115,6 +122,16 @@ def _resolve_series_id(indicator: str) -> str:
     return candidate
 
 
+def _fred_today() -> str:
+    """FRED's current calendar date (US Central) as ``yyyy-mm-dd``.
+
+    The vintage pin is clamped to this: FRED rejects a ``realtime_start`` after
+    its own today with a 400, and ``curr_date`` on a live run comes from the
+    caller's local clock, which can already be tomorrow in Chicago.
+    """
+    return datetime.now(FRED_TZ).strftime("%Y-%m-%d")
+
+
 def _request(path: str, params: dict) -> dict:
     """GET a FRED endpoint, surfacing FRED's JSON error body on a bad request."""
     api_params = {**params, "api_key": get_api_key(), "file_type": "json"}
@@ -144,11 +161,11 @@ def get_macro_data(
         indicator: A friendly alias (e.g. "cpi", "unemployment", "10y_treasury")
             or a raw FRED series ID (e.g. "CPIAUCSL", "DGS10").
         curr_date: The as-of date (yyyy-mm-dd). It bounds the observation window
-            AND pins the data vintage: FRED is queried with
-            ``realtime_start = realtime_end = curr_date`` so a historical run sees
-            the values that were actually published by that date, not later
-            revisions. Without this, revision-prone series (CPI, GDP, ...) would
-            leak future information into a backtest (#1275).
+            AND pins the data vintage: FRED is queried with the realtime bounds
+            set to ``curr_date`` (clamped to FRED's own today) so a historical
+            run sees the values that were actually published by that date, not
+            later revisions. Without this, revision-prone series (CPI, GDP, ...)
+            would leak future information into a backtest (#1275).
         look_back_days: Trailing window length; ``None`` uses DEFAULT_LOOKBACK_DAYS.
 
     Returns:
@@ -161,11 +178,16 @@ def get_macro_data(
     end_dt = datetime.strptime(curr_date, "%Y-%m-%d")
     start_date = (end_dt - timedelta(days=look_back_days)).strftime("%Y-%m-%d")
 
-    # Pin the data vintage to curr_date. FRED defaults both realtime bounds to
-    # today, which serves the LATEST revision of every observation; a single-day
-    # realtime interval asks for the values known as of curr_date instead. This
-    # is applied to both the metadata and observations requests (#1275).
-    realtime = {"realtime_start": curr_date, "realtime_end": curr_date}
+    # Pin the data vintage. FRED defaults both realtime bounds to today, serving
+    # the LATEST revision of every observation; a single-day realtime interval
+    # asks for the values known as of the pin instead, on both the metadata and
+    # observations requests (#1275). Clamp to FRED's today: on a live run
+    # curr_date is the caller's local date, which can be a day ahead of Chicago,
+    # and a realtime date in FRED's future 400s -> the routing layer would then
+    # drop macro data silently. A past curr_date is unaffected, so historical
+    # point-in-time behaviour is preserved.
+    pit = min(curr_date, _fred_today())
+    realtime = {"realtime_start": pit, "realtime_end": pit}
 
     # Invalid LLM-supplied indicator: return guidance rather than raising, so a
     # bad argument doesn't abort the run (the routing layer also degrades macro
@@ -215,8 +237,10 @@ def get_macro_data(
 
     if not points:
         return header + (
-            f"\nNo observations for {series_id} in this window. The series may "
-            f"report less frequently than the window length; widen look_back_days."
+            f"\nNo observations for {series_id} in this window at the {pit} "
+            f"vintage. The series may report less frequently than the window "
+            f"(try a longer look_back_days), or have no vintage published by "
+            f"then (unpublished as of {pit}, or before ALFRED coverage begins)."
         )
 
     first_date, first_val = points[0]

tradingagents/dataflows/reddit.py

@@ -21,6 +21,7 @@ import html
 import http.client
 import json
 import logging
+import random
 import re
 import time
 import xml.etree.ElementTree as ET
@@ -101,15 +102,47 @@ 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
+
+
+def _jitter(seconds: float, frac: float = 0.2) -> float:
+    """Return ``seconds`` with +/-``frac`` random jitter, to desynchronize
+    concurrent runs pacing against the same per-IP limit."""
+    return seconds * (1.0 + random.uniform(-frac, frac))
+
+
 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 30s.
+
+    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 else None
+        return min(float(val), 30.0) if val is not None else None
     except (ValueError, TypeError, AttributeError):
         return None
 
 
+# Reddit search feeds are small (a page of results); cap the read so a
+# compromised or misbehaving endpoint can't stream an unbounded body into
+# memory before we parse it. Overflow raises http.client.HTTPException, which
+# both fetch paths already treat as a failed fetch (degrade to empty / RSS).
+_MAX_FEED_BYTES = 5 * 1024 * 1024
+
+
+def _read_capped(resp) -> bytes:
+    """Read a response body bounded to ``_MAX_FEED_BYTES``, raising on overflow."""
+    data = resp.read(_MAX_FEED_BYTES + 1)
+    if len(data) > _MAX_FEED_BYTES:
+        raise http.client.HTTPException(
+            f"Reddit feed exceeded {_MAX_FEED_BYTES} bytes; refusing to parse"
+        )
+    return data
+
+
 def _fetch_subreddit_rss(
     ticker: str,
     sub: str,
@@ -128,10 +161,13 @@ def _fetch_subreddit_rss(
     req = Request(url, headers={"User-Agent": _UA})
     try:
         with urlopen(req, timeout=timeout) as resp:
-            root = ET.fromstring(resp.read())
+            root = ET.fromstring(_read_capped(resp))
     except HTTPError as exc:
         if exc.code == 429 and _retry:
-            wait = _retry_after_seconds(exc) or 5.0
+            # Honour a server-supplied Retry-After exactly (including 0); jitter
+            # only our own fallback so concurrent runs don't retry in lockstep.
+            retry_after = _retry_after_seconds(exc)
+            wait = retry_after if retry_after is not None else _jitter(_RETRY_FALLBACK_SECONDS)
             logger.warning(
                 "Reddit RSS 429 for r/%s · %s — backing off %.1fs then retrying once",
                 sub, ticker, wait,
@@ -182,7 +218,7 @@ def _fetch_subreddit_json(
     req = Request(url, headers={"User-Agent": _UA, "Accept": "application/json"})
     try:
         with urlopen(req, timeout=timeout) as resp:
-            payload = json.loads(resp.read())
+            payload = json.loads(_read_capped(resp))
         children = (payload.get("data") or {}).get("children") or []
         return [c.get("data", {}) for c in children if isinstance(c, dict)]
     except (OSError, http.client.HTTPException, json.JSONDecodeError) as exc:
@@ -234,8 +270,8 @@ def fetch_reddit_posts(
     blocks = []
     total_posts = 0
     for i, sub in enumerate(subreddits):
-        if i > 0:
-            time.sleep(inter_request_delay)
+        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)
         total_posts += len(posts)