File size: 9,259 Bytes
9d67baa 8a5d4f8 9d67baa 8a5d4f8 9d67baa | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 | import logging
import pandas as pd
import pytest
from unittest.mock import patch
from agentic_ai_system.yahoo_data_stream import YahooDataStream
@pytest.fixture
def yahoo_config():
return {
'data_source': {'type': 'yahoo'},
'yahoo': {'poll_interval_seconds': 1, 'auto_adjust': False},
'trading': {
'symbol': 'AAPL',
'timeframe': '1d',
},
'realtime_data': {'buffer_size': 10},
}
def _sample_yahoo_frame():
idx = pd.date_range('2024-06-03', periods=3, freq='D', tz='America/New_York')
return pd.DataFrame(
{
'Open': [190.0, 191.0, 192.0],
'High': [191.5, 192.5, 193.5],
'Low': [189.0, 190.0, 191.0],
'Close': [191.0, 192.0, 193.0],
'Volume': [1_000_000, 1_100_000, 1_200_000],
},
index=idx,
)
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)
assert stream.symbols == ['AAPL']
assert stream.interval == '1d'
def test_normalize_ohlcv(self, yahoo_config):
stream = YahooDataStream(yahoo_config)
df = stream._normalize_ohlcv(_sample_yahoo_frame())
assert list(df.columns) == ['timestamp', 'open', 'high', 'low', 'close', 'volume']
assert len(df) == 3
assert df['close'].iloc[-1] == 193.0
def test_clamp_intraday_lookback(self, yahoo_config):
yahoo_config['trading']['timeframe'] = '1m'
stream = YahooDataStream(yahoo_config)
start, end = stream._clamp_window('2020-01-01', '2026-01-01', '1m')
assert start > '2020-01-01'
assert end >= start
def test_get_historical_data(self, yahoo_config):
stream = YahooDataStream(yahoo_config)
with patch.object(stream, '_download', return_value=_sample_yahoo_frame()):
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
|