fix(dataflows): trim social sentiment sources to the analysis window

- StockTwits and Reddit were fetched with no date, so a historical run showed
  today's chatter as if it were from the as-of date
- pass the analysis window to both fetchers, filter to it, and emit a clear
  placeholder when nothing qualifies
- centralize the UTC half-open window rule in dataflows/date_window so news,
  StockTwits, and Reddit share one look-ahead-safe filter #1220
This commit is contained in:
Yijia-Xiao
2026-08-30 06:18:59 +00:00
parent 8b7ece8a3e
commit 9b98f09613
6 changed files with 234 additions and 28 deletions

View File

@@ -0,0 +1,121 @@
"""Historical social sentiment must not leak current data into a backtest (#1220).
StockTwits and Reddit fetchers pull only recent items, so for a historical run
they must be trimmed to the analysis window (and yield a clear placeholder when
nothing qualifies) rather than showing today's chatter as if it were from the
as-of date. All three sources share dataflows.date_window.in_window.
"""
from __future__ import annotations
import json
from datetime import datetime, timezone
import pytest
from tradingagents.dataflows import reddit, stocktwits
from tradingagents.dataflows.date_window import in_window
class _JsonResp:
"""Minimal urlopen() context-manager stub returning a JSON body."""
def __init__(self, payload):
self._body = json.dumps(payload).encode()
def __enter__(self):
return self
def __exit__(self, *a):
return False
def read(self):
return self._body
# --- shared window helper ---------------------------------------------------
@pytest.mark.unit
def test_in_window_bounds_and_exclusive_upper():
start = datetime(2026, 5, 1)
end = datetime(2026, 5, 9)
assert in_window(datetime(2026, 5, 5, tzinfo=timezone.utc), start, end) is True
assert in_window(datetime(2026, 5, 9, 23, 59, tzinfo=timezone.utc), start, end) is True
# exactly midnight after end -> excluded (no leak)
assert in_window(datetime(2026, 5, 10, 0, 0, tzinfo=timezone.utc), start, end) is False
# offset-aware converted, not truncated: 05-10T01:00+05:00 == 05-09T20:00Z
assert in_window(datetime.fromisoformat("2026-05-10T01:00:00+05:00"), start, end) is True
@pytest.mark.unit
def test_in_window_undated_excluded_in_backtest_kept_live():
old = datetime(2026, 5, 9)
assert in_window(None, datetime(2026, 5, 1), old) is False # historical
now = datetime.now(timezone.utc)
assert in_window(None, now, now) is True # live
# --- StockTwits -------------------------------------------------------------
def _msg(created_iso, sentiment=None):
return {
"created_at": created_iso,
"user": {"username": "u"},
"entities": {"sentiment": {"basic": sentiment}},
"body": "text",
}
@pytest.mark.unit
def test_stocktwits_historical_window_excludes_recent(monkeypatch):
# All messages are "today"; a run as-of a past week must show none of them.
recent = [_msg("2026-08-30T12:00:00Z", "Bullish"), _msg("2026-08-29T09:00:00Z")]
monkeypatch.setattr(stocktwits, "urlopen", lambda *a, **k: _JsonResp({"messages": recent}))
out = stocktwits.fetch_stocktwits_messages("AAPL", start_date="2026-05-01", end_date="2026-05-08")
assert "no StockTwits messages" in out
assert "2026-05-01..2026-05-08" in out
assert "Bullish: 1" not in out # the recent bullish message did not leak
@pytest.mark.unit
def test_stocktwits_live_window_keeps_in_range(monkeypatch):
msgs = [_msg("2026-05-05T12:00:00Z", "Bullish"), _msg("2026-05-07T09:00:00Z", "Bearish")]
monkeypatch.setattr(stocktwits, "urlopen", lambda *a, **k: _JsonResp({"messages": msgs}))
out = stocktwits.fetch_stocktwits_messages("AAPL", start_date="2026-05-01", end_date="2026-05-08")
assert "Total: 2" in out
@pytest.mark.unit
def test_stocktwits_no_window_is_unfiltered(monkeypatch):
msgs = [_msg("2026-08-30T12:00:00Z", "Bullish")]
monkeypatch.setattr(stocktwits, "urlopen", lambda *a, **k: _JsonResp({"messages": msgs}))
out = stocktwits.fetch_stocktwits_messages("AAPL") # live caller, no dates
assert "Total: 1" in out
# --- Reddit -----------------------------------------------------------------
def _epoch(date_str):
return int(datetime.strptime(date_str, "%Y-%m-%d").replace(tzinfo=timezone.utc).timestamp())
@pytest.mark.unit
def test_reddit_historical_window_excludes_recent(monkeypatch):
posts = [{"title": "NOW", "created_utc": _epoch("2026-08-30"), "source": "rss"}]
monkeypatch.setattr(reddit, "_fetch_subreddit", lambda *a, **k: posts)
out = reddit.fetch_reddit_posts(
"AAPL", subreddits=("stocks",), inter_request_delay=0,
start_date="2026-05-01", end_date="2026-05-08",
)
assert "NOW" not in out
assert "no posts" in out.lower() or "no reddit posts" in out.lower()
@pytest.mark.unit
def test_reddit_live_window_keeps_in_range(monkeypatch):
posts = [{"title": "INRANGE", "created_utc": _epoch("2026-05-05"), "source": "rss"}]
monkeypatch.setattr(reddit, "_fetch_subreddit", lambda *a, **k: posts)
out = reddit.fetch_reddit_posts(
"AAPL", subreddits=("stocks",), inter_request_delay=0,
start_date="2026-05-01", end_date="2026-05-08",
)
assert "INRANGE" in out

View File

@@ -68,8 +68,12 @@ def create_sentiment_analyst(llm):
# returns a string (no exceptions surface from here), so the LLM
# always sees something — either real data or a clear placeholder.
news_block = get_news.func(ticker, start_date, end_date)
stocktwits_block = fetch_stocktwits_messages(ticker, limit=30)
reddit_block = fetch_reddit_posts(ticker)
# Pass the analysis window so a historical run trims social posts to it
# instead of leaking today's chatter into a backtest (#1220).
stocktwits_block = fetch_stocktwits_messages(
ticker, limit=30, start_date=start_date, end_date=end_date
)
reddit_block = fetch_reddit_posts(ticker, start_date=start_date, end_date=end_date)
system_message = _build_system_message(
ticker=ticker,

View File

@@ -0,0 +1,30 @@
"""Shared look-ahead-safe date-window filtering for dated content.
News, StockTwits, and Reddit all pull recent items that must be trimmed to the
analysis window so a historical/backtest run never sees content published after
its as-of date. Centralizing the rule keeps every source consistent (#1126,
#1220): every timestamp is normalized to UTC, the upper bound is exclusive at
midnight after ``end`` (so an item stamped exactly then can't leak), and an
undated item is kept only when the window reaches the present (a live run), since
in a backtest we can't prove it isn't future.
"""
from __future__ import annotations
from datetime import datetime, timedelta, timezone
def to_utc(dt: datetime) -> datetime:
"""Normalize a datetime to UTC-aware; a naive value is assumed to be UTC."""
return dt.replace(tzinfo=timezone.utc) if dt.tzinfo is None else dt.astimezone(timezone.utc)
def in_window(pub_dt: datetime | None, start_dt: datetime, end_dt: datetime) -> bool:
"""Whether an item belongs in the half-open window ``[start, end + 1 day)``.
``pub_dt`` None means undated: kept only when the window reaches the present.
"""
end = to_utc(end_dt)
if pub_dt is not None:
return to_utc(start_dt) <= to_utc(pub_dt) < end + timedelta(days=1)
return end >= datetime.now(timezone.utc) - timedelta(days=1)

View File

@@ -25,15 +25,35 @@ import re
import time
import xml.etree.ElementTree as ET
from collections.abc import Iterable
from datetime import datetime
from datetime import datetime, timezone
from urllib.error import HTTPError
from urllib.parse import urlencode
from urllib.request import Request, urlopen
from .date_window import in_window
from .symbol_utils import crypto_base
logger = logging.getLogger(__name__)
def _within_window(posts, start_date, end_date):
"""Keep only posts published in [start_date, end_date] (look-ahead safe).
No window (both None) leaves the list untouched for live callers. A post with
no ``created_utc`` epoch is dropped in a historical window (#1220).
"""
if not (start_date and end_date):
return posts
start_dt = datetime.strptime(start_date, "%Y-%m-%d")
end_dt = datetime.strptime(end_date, "%Y-%m-%d")
kept = []
for p in posts:
ts = p.get("created_utc")
created = datetime.fromtimestamp(ts, tz=timezone.utc) if ts else None
if in_window(created, start_dt, end_dt):
kept.append(p)
return kept
_API = "https://www.reddit.com/r/{sub}/search.json?{qs}"
_RSS = "https://www.reddit.com/r/{sub}/search.rss?{qs}"
# A descriptive, identified User-Agent (per Reddit's API etiquette). Reddit
@@ -194,6 +214,8 @@ def fetch_reddit_posts(
limit_per_sub: int = 5,
timeout: float = 10.0,
inter_request_delay: float = 1.0,
start_date: str | None = None,
end_date: str | None = None,
) -> str:
"""Fetch recent Reddit posts mentioning ``ticker`` across finance
subreddits and return them as a formatted plaintext block.
@@ -201,6 +223,10 @@ def fetch_reddit_posts(
``inter_request_delay`` paces the (now RSS-only) per-subreddit requests to
stay under Reddit's public per-IP rate limit; combined with the RSS-first
path it makes 429s rare even when several analyses run back-to-back.
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).
"""
# 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.
@@ -210,7 +236,8 @@ def fetch_reddit_posts(
for i, sub in enumerate(subreddits):
if i > 0:
time.sleep(inter_request_delay)
posts = _fetch_subreddit(ticker, sub, limit_per_sub, timeout)
posts = _within_window(_fetch_subreddit(ticker, sub, limit_per_sub, timeout),
start_date, end_date)
total_posts += len(posts)
if not posts:
blocks.append(f"r/{sub}: <no posts found mentioning {ticker.upper()} in the past 7 days>")

View File

@@ -14,11 +14,14 @@ network call succeeded.
from __future__ import annotations
import contextlib
import http.client
import json
import logging
from datetime import datetime
from urllib.request import Request, urlopen
from .date_window import in_window
from .symbol_utils import crypto_base
logger = logging.getLogger(__name__)
@@ -27,6 +30,29 @@ _API = "https://api.stocktwits.com/api/2/streams/symbol/{ticker}.json"
_UA = "tradingagents/0.2 (+https://github.com/TauricResearch/TradingAgents)"
def _within_window(messages, start_date, end_date):
"""Keep only messages published in [start_date, end_date] (look-ahead safe).
No window (both None) leaves the list untouched for live callers. A message
whose ``created_at`` (ISO 8601) is unparseable is dropped in a historical
window, since we can't prove it isn't from after the as-of date (#1220).
"""
if not (start_date and end_date):
return messages
start_dt = datetime.strptime(start_date, "%Y-%m-%d")
end_dt = datetime.strptime(end_date, "%Y-%m-%d")
kept = []
for m in messages:
created = None
raw = m.get("created_at")
if raw:
with contextlib.suppress(ValueError, TypeError):
created = datetime.fromisoformat(str(raw).replace("Z", "+00:00"))
if in_window(created, start_dt, end_dt):
kept.append(m)
return kept
def _stocktwits_symbol(ticker: str) -> str:
"""Map a crypto pair to StockTwits' ``<BASE>.X`` convention.
@@ -38,10 +64,21 @@ def _stocktwits_symbol(ticker: str) -> str:
return f"{base}.X" if base else ticker.strip().upper()
def fetch_stocktwits_messages(ticker: str, limit: int = 30, timeout: float = 10.0) -> str:
def fetch_stocktwits_messages(
ticker: str,
limit: int = 30,
timeout: float = 10.0,
start_date: str | None = None,
end_date: str | None = None,
) -> str:
"""Fetch recent StockTwits messages for ``ticker`` and return them as a
formatted plaintext block ready for prompt injection.
When ``start_date``/``end_date`` (yyyy-mm-dd) are given, messages are trimmed
to that window. The StockTwits public stream only serves recent messages, so
for a historical run they all fall after the window and a clear placeholder
is returned rather than leaking today's chatter into a backtest (#1220).
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.
@@ -58,7 +95,13 @@ def fetch_stocktwits_messages(ticker: str, limit: int = 30, timeout: float = 10.
return f"<stocktwits unavailable: {type(exc).__name__}>"
messages = data.get("messages", []) if isinstance(data, dict) else []
messages = _within_window(messages, start_date, end_date)
if not messages:
if start_date and end_date:
return (
f"<no StockTwits messages for ${ticker.upper()} within "
f"{start_date}..{end_date} (public stream serves only recent messages)>"
)
return f"<no StockTwits messages found for ${ticker.upper()}>"
lines = []

View File

@@ -1,26 +1,17 @@
"""yfinance-based news data fetching functions."""
import contextlib
from datetime import datetime, timedelta, timezone
from datetime import datetime, timezone
import yfinance as yf
from dateutil.relativedelta import relativedelta
from .config import get_config
from .date_window import in_window
from .stockstats_utils import yf_retry
from .symbol_utils import normalize_symbol
def _as_utc(dt: datetime) -> datetime:
"""Normalize a datetime to UTC-aware; a naive value is assumed to be UTC.
Window bounds arrive naive (parsed from ``yyyy-mm-dd``) while article
timestamps may be offset-aware, so every operand is normalized before
comparison. Without this the filter depends on the host timezone (#1126).
"""
return dt.replace(tzinfo=timezone.utc) if dt.tzinfo is None else dt.astimezone(timezone.utc)
def _extract_article_data(article: dict) -> dict:
"""Extract article data from yfinance news format (handles nested 'content' structure)."""
# Handle nested content structure
@@ -70,18 +61,8 @@ def _extract_article_data(article: dict) -> dict:
def _in_news_window(pub_date, start_dt, end_dt) -> bool:
"""Whether an article belongs in the half-open window ``[start, end + 1 day)``.
Every operand is normalized to UTC, and the upper bound is exclusive so an
article stamped exactly at midnight after ``end_dt`` cannot leak into a
historical run (#1126). An undated article is kept only when the window
reaches the present (live run) — in a historical/backtest window it's
excluded, since we can't prove it isn't future news (#992/#1007).
"""
end = _as_utc(end_dt)
if pub_date is not None:
return _as_utc(start_dt) <= _as_utc(pub_date) < end + timedelta(days=1)
return end >= datetime.now(timezone.utc) - timedelta(days=1)
"""Look-ahead-safe article-window check; see dataflows.date_window.in_window."""
return in_window(pub_date, start_dt, end_dt)
def get_news_yfinance(