diff --git a/tests/test_fred.py b/tests/test_fred.py index 2a03f6456..a4ffbd213 100644 --- a/tests/test_fred.py +++ b/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): diff --git a/tests/test_reddit_fallback.py b/tests/test_reddit_fallback.py index dece01ca6..ec9a89fdd 100644 --- a/tests/test_reddit_fallback.py +++ b/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: diff --git a/tradingagents/agents/managers/portfolio_manager.py b/tradingagents/agents/managers/portfolio_manager.py index 00d5503cd..75508dca6 100644 --- a/tradingagents/agents/managers/portfolio_manager.py +++ b/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()}""" diff --git a/tradingagents/agents/managers/research_manager.py b/tradingagents/agents/managers/research_manager.py index 1aec89572..85c2d8e90 100644 --- a/tradingagents/agents/managers/research_manager.py +++ b/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. --- diff --git a/tradingagents/agents/schemas.py b/tradingagents/agents/schemas.py index 63ff221fd..961cde79a 100644 --- a/tradingagents/agents/schemas.py +++ b/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( diff --git a/tradingagents/dataflows/fred.py b/tradingagents/dataflows/fred.py index 38ab7d340..98f1827b6 100644 --- a/tradingagents/dataflows/fred.py +++ b/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] diff --git a/tradingagents/dataflows/reddit.py b/tradingagents/dataflows/reddit.py index 9595e621e..f8f0e1755 100644 --- a/tradingagents/dataflows/reddit.py +++ b/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)