Fix three correctness bugs in the Yahoo ingest

All three are silent: nothing raises, the numbers are just wrong.

Unadjusted prices. yahoo.auto_adjust defaulted to false and _normalize_ohlcv
selects only the required columns, discarding 'adj close'. So splits were not
adjusted for: NVDA's June 2024 10:1 reads as a -90% single-day return and AAPL's
2020 4:1 as -75%. Any strategy or risk control spanning a split saw a phantom
crash. Default is now true, opting out logs a warning, and _warn_if_unadjusted
names the dates of daily-or-slower bars moving more than 35% as a backstop when
adjustment is off.

Partial bars reported as final. _poll_once pulls period='5d' and
_ingest_new_bars took every row past the watermark, including the period Yahoo
is still building. It emitted that in-progress bar, advanced _last_bar_ts past
it, and so never re-emitted the finished version — the strategy acted on a close
that had not happened yet, then never saw the real one. _drop_incomplete now
withholds a bar until its interval has elapsed; yahoo.emit_incomplete_bars
re-enables the old behaviour explicitly.

No backoff on rate limiting. The poll loop caught exceptions and retried at a
fixed 60s, so once Yahoo throttled us we stayed throttled. _poll_once now
reports success, and _next_delay backs off exponentially to
yahoo.max_backoff_seconds with jitter so symbols do not resynchronise after an
outage. connect() seeds the counter from its first attempt.

17 new tests, all of which fail against the previous implementation, including a
split fixture modelled on NVDA. Existing tests unchanged and still pass.

agentic_ai_system/yahoo_data_stream.py

@@ -1,4 +1,5 @@
 import logging
+import random
 import threading
 import time
 from typing import Any, Callable, Dict, List, Optional
@@ -41,6 +42,26 @@ _MAX_LOOKBACK = {
     '3mo': None,
 }
 
+# How long each bar covers. Used to tell a finished bar from the one still
+# forming right now -- see _drop_incomplete.
+_INTERVAL_DURATION = {
+    '1m': pd.Timedelta(minutes=1),
+    '2m': pd.Timedelta(minutes=2),
+    '5m': pd.Timedelta(minutes=5),
+    '15m': pd.Timedelta(minutes=15),
+    '30m': pd.Timedelta(minutes=30),
+    '60m': pd.Timedelta(hours=1),
+    '90m': pd.Timedelta(minutes=90),
+    '1h': pd.Timedelta(hours=1),
+    '1d': pd.Timedelta(days=1),
+    '5d': pd.Timedelta(days=5),
+    '1wk': pd.Timedelta(weeks=1),
+}
+
+# A daily-or-slower bar that moves more than this is almost always an
+# unadjusted split rather than a real move (NVDA's 2024 10:1 shows up as -90%).
+_SPLIT_SUSPECT_MOVE = 0.35
+
 
 class YahooDataStream:
     """
@@ -62,8 +83,13 @@ class YahooDataStream:
             self.symbols = ['AAPL']
         yahoo_cfg = config.get('yahoo', {})
         self.poll_interval = int(yahoo_cfg.get('poll_interval_seconds', 60))
-        self.auto_adjust = bool(yahoo_cfg.get('auto_adjust', False))
+        # Adjusted by default. With auto_adjust off, Yahoo returns raw Close and
+        # every split reads as a crash: NVDA's June 2024 10:1 becomes a -90% bar.
+        self.auto_adjust = bool(yahoo_cfg.get('auto_adjust', True))
+        self.emit_incomplete_bars = bool(yahoo_cfg.get('emit_incomplete_bars', False))
+        self.max_backoff = int(yahoo_cfg.get('max_backoff_seconds', 900))
         self.interval = self._map_interval(config.get('trading', {}).get('timeframe', '1d'))
+        self._consecutive_failures = 0
         self.data_callbacks: List[Callable] = []
         self.is_connected = False
         self.data_buffer: Dict[str, Dict[str, Any]] = {}
@@ -80,11 +106,21 @@ class YahooDataStream:
                 'latest_bar': None,
             }
 
+        if not self.auto_adjust:
+            logger.warning(
+                "yahoo.auto_adjust is false: prices are NOT split- or dividend-adjusted. "
+                "Every split will appear as a large single-bar loss and any backtest "
+                "spanning one will be wrong."
+            )
+
         logger.info(
-            "Initialized YahooDataStream symbols=%s interval=%s poll_interval=%ss",
+            "Initialized YahooDataStream symbols=%s interval=%s poll_interval=%ss "
+            "auto_adjust=%s emit_incomplete_bars=%s",
             self.symbols,
             self.interval,
             self.poll_interval,
+            self.auto_adjust,
+            self.emit_incomplete_bars,
         )
 
     @staticmethod
@@ -102,7 +138,9 @@ class YahooDataStream:
             return
 
         self._stop_event.clear()
-        self._poll_once()
+        # Seed the backoff from the first attempt: if we are already being
+        # throttled, the loop should start backed off rather than hammering.
+        self._consecutive_failures = 0 if self._poll_once() else 1
         self._poll_thread = threading.Thread(target=self._poll_loop, name='yahoo-poll', daemon=True)
         self._poll_thread.start()
         self.is_connected = True
@@ -137,7 +175,8 @@ class YahooDataStream:
         start, end = self._clamp_window(start_date, end_date, self.interval)
         try:
             raw = self._download(symbol, start=start, end=end, interval=self.interval)
-            df = self._normalize_ohlcv(raw)
+            df = self._drop_incomplete(self._normalize_ohlcv(raw))
+            self._warn_if_unadjusted(symbol, df)
             if df.empty:
                 logger.warning("No Yahoo historical data for %s between %s and %s", symbol, start, end)
             else:
@@ -173,8 +212,6 @@ class YahooDataStream:
         }
 
     def generate_simulated_data(self, symbol: str) -> Dict[str, Any]:
-        import random
-
         latest_data = self.get_latest_data(symbol)
         base_price = 150.0
         if latest_data.get('latest_bar'):
@@ -197,13 +234,40 @@ class YahooDataStream:
         return simulated_bar
 
     def _poll_loop(self) -> None:
-        while not self._stop_event.wait(self.poll_interval):
+        delay = self.poll_interval
+        while not self._stop_event.wait(delay):
             try:
-                self._poll_once()
+                succeeded = self._poll_once()
             except Exception as e:
                 logger.error("Yahoo poll loop error: %s", e, exc_info=True)
-
-    def _poll_once(self) -> None:
+                succeeded = False
+            self._consecutive_failures = 0 if succeeded else self._consecutive_failures + 1
+            delay = self._next_delay()
+
+    def _next_delay(self) -> float:
+        """Poll interval, backed off exponentially while Yahoo is refusing us.
+
+        Yahoo rate-limits aggressively and an unofficial API gives no
+        Retry-After, so a fixed interval just keeps you throttled. Jitter stops
+        several symbols (or several deployments) resynchronising after an outage.
+        """
+        if self._consecutive_failures == 0:
+            base = float(self.poll_interval)
+        else:
+            base = min(
+                self.poll_interval * (2 ** self._consecutive_failures),
+                float(self.max_backoff),
+            )
+            logger.warning(
+                "Yahoo poll failed %s time(s) in a row; next attempt in ~%.0fs",
+                self._consecutive_failures,
+                base,
+            )
+        return max(1.0, base * random.uniform(0.8, 1.2))
+
+    def _poll_once(self) -> bool:
+        """Fetch and ingest one round of bars. Returns True if any symbol succeeded."""
+        any_success = False
         for symbol in self.symbols:
             try:
                 raw = self._download(symbol, period='5d', interval=self.interval)
@@ -212,14 +276,60 @@ class YahooDataStream:
                     logger.warning("Yahoo poll returned no bars for %s", symbol)
                     continue
                 self._ingest_new_bars(symbol, df)
+                any_success = True
             except Exception as e:
                 logger.error("Yahoo poll failed for %s: %s", symbol, e)
+        return any_success
+
+    def _warn_if_unadjusted(self, symbol: str, df: pd.DataFrame) -> int:
+        """Flag single-bar moves that look like unadjusted corporate actions.
+
+        This is a backstop rather than the fix -- the fix is auto_adjust. But a
+        split slipping through silently corrupts every downstream number, so it
+        is worth naming the dates rather than letting a strategy trade them.
+        Returns the number of suspicious bars found.
+        """
+        duration = _INTERVAL_DURATION.get(self.interval)
+        if df.empty or len(df) < 2 or duration is None or duration < pd.Timedelta(days=1):
+            return 0
+        moves = df['close'].pct_change()
+        suspects = df.loc[moves.abs() > _SPLIT_SUSPECT_MOVE, 'timestamp']
+        if len(suspects):
+            dates = ', '.join(str(pd.Timestamp(t).date()) for t in suspects.head(5))
+            logger.warning(
+                "%s has %s bar(s) moving more than %.0f%% (%s). On a liquid name that is "
+                "usually an unadjusted split, not a real move — check yahoo.auto_adjust.",
+                symbol,
+                len(suspects),
+                _SPLIT_SUSPECT_MOVE * 100,
+                dates,
+            )
+        return int(len(suspects))
+
+    def _drop_incomplete(self, df: pd.DataFrame) -> pd.DataFrame:
+        """Remove the bar that is still forming.
+
+        Yahoo returns the in-progress period as an ordinary row. Emitting it
+        would hand the strategy a close that has not happened yet, and because
+        the watermark advances past it, the finished version never arrives.
+        """
+        if self.emit_incomplete_bars or df.empty:
+            return df
+        duration = _INTERVAL_DURATION.get(self.interval)
+        if duration is None:
+            return df
+        now = pd.Timestamp.now(tz='UTC').tz_convert(None)
+        complete = df[df['timestamp'] + duration <= now]
+        dropped = len(df) - len(complete)
+        if dropped:
+            logger.debug("Dropped %s in-progress %s bar(s)", dropped, self.interval)
+        return complete
 
     def _ingest_new_bars(self, symbol: str, df: pd.DataFrame) -> None:
+        rows = self._drop_incomplete(df)
         last_ts = self._last_bar_ts.get(symbol)
-        rows = df
         if last_ts is not None:
-            rows = df[df['timestamp'] > last_ts]
+            rows = rows[rows['timestamp'] > last_ts]
         if rows.empty:
             return
 

config.yaml

@@ -33,7 +33,13 @@ alpaca:
 # Unofficial API, typically delayed; 1m lookback is ~7 days.
 yahoo:
   poll_interval_seconds: 60
-  auto_adjust: false
+  # Keep true. With auto_adjust off, Yahoo returns raw Close and every stock
+  # split reads as a crash (NVDA's June 2024 10:1 becomes a -90% bar).
+  auto_adjust: true
+  # Yahoo returns the still-forming period as an ordinary row. Emitting it would
+  # trade on a close that has not happened yet.
+  emit_incomplete_bars: false
+  max_backoff_seconds: 900
   start_date: '2024-01-01'
   end_date: '2026-12-31'
 

tests/test_yahoo_data_stream.py

@@ -1,3 +1,5 @@
+import logging
+
 import pandas as pd
 import pytest
 from unittest.mock import patch
@@ -32,6 +34,39 @@ def _sample_yahoo_frame():
     )
 
 
+def _split_frame():
+    """An unadjusted 10:1 split, as Yahoo returns it with auto_adjust=False.
+
+    Modelled on NVDA, 10 June 2024: the raw Close drops from ~1200 to ~120 and
+    a backtest reads it as a -90% day.
+    """
+    idx = pd.date_range('2024-06-06', periods=4, freq='D', tz='America/New_York')
+    close = [1200.0, 1208.0, 120.5, 121.0]
+    return pd.DataFrame(
+        {
+            'Open': close,
+            'High': [c * 1.01 for c in close],
+            'Low': [c * 0.99 for c in close],
+            'Close': close,
+            'Volume': [1_000_000] * 4,
+        },
+        index=idx,
+    )
+
+
+def _dated_frame(timestamps, close=100.0):
+    return pd.DataFrame(
+        {
+            'timestamp': [pd.Timestamp(t) for t in timestamps],
+            'open': close,
+            'high': close,
+            'low': close,
+            'close': close,
+            'volume': 1_000.0,
+        }
+    )
+
+
 class TestYahooDataStream:
     def test_initialization_from_symbol(self, yahoo_config):
         stream = YahooDataStream(yahoo_config)
@@ -58,3 +93,137 @@ class TestYahooDataStream:
             df = stream.get_historical_data('AAPL', '2024-01-01', '2024-12-31')
         assert len(df) == 3
         assert 'open' in df.columns
+
+
+class TestPriceAdjustment:
+    """Unadjusted prices turn every split into a phantom crash."""
+
+    def test_adjustment_is_on_by_default(self):
+        stream = YahooDataStream({'trading': {'symbol': 'AAPL', 'timeframe': '1d'}})
+        assert stream.auto_adjust is True
+
+    def test_auto_adjust_is_passed_through_to_yfinance(self):
+        stream = YahooDataStream({'trading': {'symbol': 'AAPL', 'timeframe': '1d'}})
+        with patch('yfinance.download', return_value=_sample_yahoo_frame()) as download:
+            stream._download('AAPL', period='5d', interval='1d')
+        assert download.call_args.kwargs['auto_adjust'] is True
+
+    def test_opting_out_of_adjustment_warns(self, caplog):
+        with caplog.at_level(logging.WARNING):
+            YahooDataStream({
+                'trading': {'symbol': 'AAPL', 'timeframe': '1d'},
+                'yahoo': {'auto_adjust': False},
+            })
+        assert any('not split' in r.message.lower() for r in caplog.records)
+
+    def test_split_sized_move_is_flagged(self, yahoo_config, caplog):
+        stream = YahooDataStream(yahoo_config)
+        df = stream._normalize_ohlcv(_split_frame())
+        with caplog.at_level(logging.WARNING):
+            found = stream._warn_if_unadjusted('NVDA', df)
+        assert found == 1
+        assert any('auto_adjust' in r.message for r in caplog.records)
+
+    def test_ordinary_moves_are_not_flagged(self, yahoo_config, caplog):
+        stream = YahooDataStream(yahoo_config)
+        df = stream._normalize_ohlcv(_sample_yahoo_frame())
+        with caplog.at_level(logging.WARNING):
+            assert stream._warn_if_unadjusted('AAPL', df) == 0
+
+    def test_intraday_bars_are_not_split_checked(self, yahoo_config):
+        """A 40% move in one minute is a halt or a fat finger, not a split."""
+        yahoo_config['trading']['timeframe'] = '1m'
+        stream = YahooDataStream(yahoo_config)
+        df = stream._normalize_ohlcv(_split_frame())
+        assert stream._warn_if_unadjusted('NVDA', df) == 0
+
+
+class TestIncompleteBars:
+    """The bar Yahoo is still building must not be reported as final."""
+
+    def test_forming_bar_is_dropped(self, yahoo_config):
+        stream = YahooDataStream(yahoo_config)
+        now = pd.Timestamp.now(tz='UTC').tz_convert(None).normalize()
+        df = _dated_frame([now - pd.Timedelta(days=2), now - pd.Timedelta(days=1), now])
+        kept = stream._drop_incomplete(df)
+        assert len(kept) == 2
+        assert kept['timestamp'].max() < now
+
+    def test_finished_bars_all_survive(self, yahoo_config):
+        stream = YahooDataStream(yahoo_config)
+        now = pd.Timestamp.now(tz='UTC').tz_convert(None).normalize()
+        df = _dated_frame([now - pd.Timedelta(days=5), now - pd.Timedelta(days=4)])
+        assert len(stream._drop_incomplete(df)) == 2
+
+    def test_opting_in_keeps_the_forming_bar(self, yahoo_config):
+        yahoo_config['yahoo']['emit_incomplete_bars'] = True
+        stream = YahooDataStream(yahoo_config)
+        now = pd.Timestamp.now(tz='UTC').tz_convert(None).normalize()
+        df = _dated_frame([now - pd.Timedelta(days=1), now])
+        assert len(stream._drop_incomplete(df)) == 2
+
+    def test_partial_bar_is_never_emitted_then_stranded(self, yahoo_config):
+        """The bug this guards: emitting the forming bar advanced the watermark,
+        so the finished version of that same bar never reached a callback."""
+        stream = YahooDataStream(yahoo_config)
+        received = []
+        stream.add_data_callback(lambda kind, bar: received.append(bar))
+
+        now = pd.Timestamp.now(tz='UTC').tz_convert(None).normalize()
+        yesterday, today = now - pd.Timedelta(days=1), now
+
+        stream._ingest_new_bars('AAPL', _dated_frame([yesterday, today], close=100.0))
+        assert len(received) == 1  # only yesterday's completed bar
+
+        # Next day: what was the forming bar is now final and must arrive.
+        with patch.object(stream, '_drop_incomplete', side_effect=lambda d: d):
+            stream._ingest_new_bars('AAPL', _dated_frame([yesterday, today], close=105.0))
+        assert len(received) == 2
+        assert received[-1]['close'] == 105.0
+
+
+class TestPollBackoff:
+    """Yahoo rate-limits hard, and a fixed interval keeps you throttled."""
+
+    def test_success_polls_at_the_configured_interval(self, yahoo_config):
+        yahoo_config['yahoo']['poll_interval_seconds'] = 60
+        stream = YahooDataStream(yahoo_config)
+        stream._consecutive_failures = 0
+        assert 48 <= stream._next_delay() <= 72  # 60s +/- jitter
+
+    def test_delay_grows_with_consecutive_failures(self, yahoo_config):
+        yahoo_config['yahoo']['poll_interval_seconds'] = 60
+        stream = YahooDataStream(yahoo_config)
+        delays = []
+        for failures in (1, 2, 3):
+            stream._consecutive_failures = failures
+            delays.append(stream._next_delay())
+        assert delays[0] < delays[1] < delays[2]
+
+    def test_backoff_is_capped(self, yahoo_config):
+        yahoo_config['yahoo']['poll_interval_seconds'] = 60
+        yahoo_config['yahoo']['max_backoff_seconds'] = 300
+        stream = YahooDataStream(yahoo_config)
+        stream._consecutive_failures = 20
+        assert stream._next_delay() <= 300 * 1.2
+
+    def test_jitter_desynchronises_retries(self, yahoo_config):
+        stream = YahooDataStream(yahoo_config)
+        stream._consecutive_failures = 3
+        assert len({stream._next_delay() for _ in range(20)}) > 1
+
+    def test_poll_reports_failure_when_every_symbol_fails(self, yahoo_config):
+        stream = YahooDataStream(yahoo_config)
+        with patch.object(stream, '_download', side_effect=RuntimeError('429 Too Many Requests')):
+            assert stream._poll_once() is False
+
+    def test_poll_reports_success_when_a_symbol_returns_bars(self, yahoo_config):
+        stream = YahooDataStream(yahoo_config)
+        with patch.object(stream, '_download', return_value=_sample_yahoo_frame()):
+            assert stream._poll_once() is True
+
+    def test_empty_response_counts_as_failure(self, yahoo_config):
+        """A rate-limited yfinance returns an empty frame rather than raising."""
+        stream = YahooDataStream(yahoo_config)
+        with patch.object(stream, '_download', return_value=pd.DataFrame()):
+            assert stream._poll_once() is False