charieswei
/
TradingAgents
mirror of https://github.com/TauricResearch/TradingAgents.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

feat: add scanner tests, global demo key in conftest, remove 48 inline key patches 

Co-authored-by: aguzererler <6199053+aguzererler@users.noreply.github.com>
This commit is contained in:
copilot-swe-agent[bot] 2026-03-17 16:01:52 +00:00
parent 4155b1754b
commit 9174ebe763
7 changed files with 993 additions and 115 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

32
tests/conftest.py
View File

@ -4,18 +4,40 @@ import os
import pytest
_DEMO_KEY = "demo"
def pytest_configure(config):
config.addinivalue_line("markers", "integration: tests that hit real external APIs")
config.addinivalue_line("markers", "slow: tests that take a long time to run")
@pytest.fixture(autouse=True)
def _set_alpha_vantage_demo_key(monkeypatch):
"""Ensure ALPHA_VANTAGE_API_KEY is always set to 'demo' unless the test
overrides it. This means no test needs its own patch.dict for the key."""
if not os.environ.get("ALPHA_VANTAGE_API_KEY"):
monkeypatch.setenv("ALPHA_VANTAGE_API_KEY", _DEMO_KEY)
@pytest.fixture
def av_api_key():
"""Return the Alpha Vantage API key or skip the test."""
key = os.environ.get("ALPHA_VANTAGE_API_KEY")
if not key:
pytest.skip("ALPHA_VANTAGE_API_KEY not set")
return key
"""Return the Alpha Vantage API key ('demo' by default).
Skips the test automatically when the Alpha Vantage API endpoint is not
reachable (e.g. sandboxed CI without outbound network access).
"""
import socket
try:
socket.setdefaulttimeout(3)
socket.socket(socket.AF_INET, socket.SOCK_STREAM).connect(
("www.alphavantage.co", 443)
)
except (socket.error, OSError):
pytest.skip("Alpha Vantage API not reachable — skipping live API test")
return os.environ.get("ALPHA_VANTAGE_API_KEY", _DEMO_KEY)
@pytest.fixture

15
tests/test_alpha_vantage_exceptions.py
View File

@ -57,14 +57,13 @@ class TestMakeApiRequestErrors:
def test_timeout_raises_timeout_error(self):
"""A timeout should raise ThirdPartyTimeoutError."""
with patch.dict(os.environ, {"ALPHA_VANTAGE_API_KEY": "demo"}):
with pytest.raises(ThirdPartyTimeoutError):
# Use an impossibly short timeout
_make_api_request(
"TIME_SERIES_DAILY",
{"symbol": "IBM"},
timeout=0.001,
)
with pytest.raises(ThirdPartyTimeoutError):
# Use an impossibly short timeout
_make_api_request(
"TIME_SERIES_DAILY",
{"symbol": "IBM"},
timeout=0.001,
)
def test_valid_request_succeeds(self, av_api_key):
"""A valid request with a real key should return data."""

90
tests/test_alpha_vantage_integration.py
View File

@ -97,9 +97,8 @@ class TestMakeApiRequest:
with patch("tradingagents.dataflows.alpha_vantage_common.requests.get",
return_value=_mock_response(CSV_DAILY_ADJUSTED)):
with patch.dict("os.environ", {"ALPHA_VANTAGE_API_KEY": "demo"}):
result = _make_api_request("TIME_SERIES_DAILY_ADJUSTED",
{"symbol": "AAPL", "datatype": "csv"})
result = _make_api_request("TIME_SERIES_DAILY_ADJUSTED",
{"symbol": "AAPL", "datatype": "csv"})
assert "timestamp" in result
assert "186.00" in result
@ -112,9 +111,8 @@ class TestMakeApiRequest:
with patch("tradingagents.dataflows.alpha_vantage_common.requests.get",
return_value=_mock_response(RATE_LIMIT_JSON)):
with patch.dict("os.environ", {"ALPHA_VANTAGE_API_KEY": "demo"}):
with pytest.raises(AlphaVantageRateLimitError):
_make_api_request("TIME_SERIES_DAILY_ADJUSTED", {"symbol": "AAPL"})
with pytest.raises(AlphaVantageRateLimitError):
_make_api_request("TIME_SERIES_DAILY_ADJUSTED", {"symbol": "AAPL"})
def test_raises_api_key_error_for_invalid_api_key(self):
"""An 'Invalid API key' Information response raises an API-key-related error.
@ -145,9 +143,8 @@ class TestMakeApiRequest:
with patch("tradingagents.dataflows.alpha_vantage_common.requests.get",
side_effect=TimeoutError("connection timed out")):
with patch.dict("os.environ", {"ALPHA_VANTAGE_API_KEY": "demo"}):
with pytest.raises(TimeoutError):
_make_api_request("OVERVIEW", {"symbol": "AAPL"})
with pytest.raises(TimeoutError):
_make_api_request("OVERVIEW", {"symbol": "AAPL"})
def test_http_error_propagates_on_non_200_status(self):
"""HTTP 4xx/5xx responses raise an error.
@ -164,9 +161,8 @@ class TestMakeApiRequest:
with patch("tradingagents.dataflows.alpha_vantage_common.requests.get",
return_value=bad_resp):
with patch.dict("os.environ", {"ALPHA_VANTAGE_API_KEY": "demo"}):
with pytest.raises(Exception):
_make_api_request("OVERVIEW", {"symbol": "AAPL"})
with pytest.raises(Exception):
_make_api_request("OVERVIEW", {"symbol": "AAPL"})
# ---------------------------------------------------------------------------
@ -259,8 +255,7 @@ class TestAlphaVantageGetStock:
with patch("tradingagents.dataflows.alpha_vantage_common.requests.get",
return_value=_mock_response(CSV_DAILY_ADJUSTED)):
with patch.dict("os.environ", {"ALPHA_VANTAGE_API_KEY": "demo"}):
result = get_stock("AAPL", "2024-01-01", "2024-01-05")
result = get_stock("AAPL", "2024-01-01", "2024-01-05")
assert isinstance(result, str)
@ -276,8 +271,7 @@ class TestAlphaVantageGetStock:
with patch("tradingagents.dataflows.alpha_vantage_common.requests.get",
side_effect=capture_request):
with patch.dict("os.environ", {"ALPHA_VANTAGE_API_KEY": "demo"}):
get_stock("AAPL", "2020-01-01", "2020-01-05")
get_stock("AAPL", "2020-01-01", "2020-01-05")
assert captured_params.get("outputsize") == "full"
@ -287,9 +281,8 @@ class TestAlphaVantageGetStock:
with patch("tradingagents.dataflows.alpha_vantage_common.requests.get",
return_value=_mock_response(RATE_LIMIT_JSON)):
with patch.dict("os.environ", {"ALPHA_VANTAGE_API_KEY": "demo"}):
with pytest.raises(AlphaVantageRateLimitError):
get_stock("AAPL", "2024-01-01", "2024-01-05")
with pytest.raises(AlphaVantageRateLimitError):
get_stock("AAPL", "2024-01-01", "2024-01-05")
# ---------------------------------------------------------------------------
@ -305,8 +298,7 @@ class TestAlphaVantageGetFundamentals:
with patch("tradingagents.dataflows.alpha_vantage_common.requests.get",
return_value=_mock_response(OVERVIEW_JSON)):
with patch.dict("os.environ", {"ALPHA_VANTAGE_API_KEY": "demo"}):
result = get_fundamentals("AAPL")
result = get_fundamentals("AAPL")
assert "Apple Inc" in result
assert "TECHNOLOGY" in result
@ -317,9 +309,8 @@ class TestAlphaVantageGetFundamentals:
with patch("tradingagents.dataflows.alpha_vantage_common.requests.get",
return_value=_mock_response(RATE_LIMIT_JSON)):
with patch.dict("os.environ", {"ALPHA_VANTAGE_API_KEY": "demo"}):
with pytest.raises(AlphaVantageRateLimitError):
get_fundamentals("AAPL")
with pytest.raises(AlphaVantageRateLimitError):
get_fundamentals("AAPL")
class TestAlphaVantageGetBalanceSheet:
@ -329,8 +320,7 @@ class TestAlphaVantageGetBalanceSheet:
payload = json.dumps({"symbol": "AAPL", "annualReports": [], "quarterlyReports": []})
with patch("tradingagents.dataflows.alpha_vantage_common.requests.get",
return_value=_mock_response(payload)):
with patch.dict("os.environ", {"ALPHA_VANTAGE_API_KEY": "demo"}):
result = get_balance_sheet("AAPL")
result = get_balance_sheet("AAPL")
assert "AAPL" in result
@ -342,8 +332,7 @@ class TestAlphaVantageGetCashflow:
payload = json.dumps({"symbol": "AAPL", "annualReports": [], "quarterlyReports": []})
with patch("tradingagents.dataflows.alpha_vantage_common.requests.get",
return_value=_mock_response(payload)):
with patch.dict("os.environ", {"ALPHA_VANTAGE_API_KEY": "demo"}):
result = get_cashflow("AAPL")
result = get_cashflow("AAPL")
assert "AAPL" in result
@ -355,8 +344,7 @@ class TestAlphaVantageGetIncomeStatement:
payload = json.dumps({"symbol": "AAPL", "annualReports": [], "quarterlyReports": []})
with patch("tradingagents.dataflows.alpha_vantage_common.requests.get",
return_value=_mock_response(payload)):
with patch.dict("os.environ", {"ALPHA_VANTAGE_API_KEY": "demo"}):
result = get_income_statement("AAPL")
result = get_income_statement("AAPL")
assert "AAPL" in result
@ -397,8 +385,7 @@ class TestAlphaVantageGetNews:
with patch("tradingagents.dataflows.alpha_vantage_common.requests.get",
return_value=_mock_response(NEWS_JSON)):
with patch.dict("os.environ", {"ALPHA_VANTAGE_API_KEY": "demo"}):
result = get_news("AAPL", "2024-01-01", "2024-01-05")
result = get_news("AAPL", "2024-01-01", "2024-01-05")
assert "Apple Hits Record High" in result
@ -408,9 +395,8 @@ class TestAlphaVantageGetNews:
with patch("tradingagents.dataflows.alpha_vantage_common.requests.get",
return_value=_mock_response(RATE_LIMIT_JSON)):
with patch.dict("os.environ", {"ALPHA_VANTAGE_API_KEY": "demo"}):
with pytest.raises(AlphaVantageRateLimitError):
get_news("AAPL", "2024-01-01", "2024-01-05")
with pytest.raises(AlphaVantageRateLimitError):
get_news("AAPL", "2024-01-01", "2024-01-05")
class TestAlphaVantageGetGlobalNews:
@ -419,8 +405,7 @@ class TestAlphaVantageGetGlobalNews:
with patch("tradingagents.dataflows.alpha_vantage_common.requests.get",
return_value=_mock_response(NEWS_JSON)):
with patch.dict("os.environ", {"ALPHA_VANTAGE_API_KEY": "demo"}):
result = get_global_news("2024-01-15", look_back_days=7)
result = get_global_news("2024-01-15", look_back_days=7)
assert isinstance(result, str)
@ -436,8 +421,7 @@ class TestAlphaVantageGetGlobalNews:
with patch("tradingagents.dataflows.alpha_vantage_common.requests.get",
side_effect=capture):
with patch.dict("os.environ", {"ALPHA_VANTAGE_API_KEY": "demo"}):
get_global_news("2024-01-15", look_back_days=7)
get_global_news("2024-01-15", look_back_days=7)
# time_from should be 7 days before 2024-01-15 → 2024-01-08
assert "20240108T0000" in captured_params.get("time_from", "")
@ -449,8 +433,7 @@ class TestAlphaVantageGetInsiderTransactions:
with patch("tradingagents.dataflows.alpha_vantage_common.requests.get",
return_value=_mock_response(INSIDER_JSON)):
with patch.dict("os.environ", {"ALPHA_VANTAGE_API_KEY": "demo"}):
result = get_insider_transactions("AAPL")
result = get_insider_transactions("AAPL")
assert "Tim Cook" in result
@ -460,9 +443,8 @@ class TestAlphaVantageGetInsiderTransactions:
with patch("tradingagents.dataflows.alpha_vantage_common.requests.get",
return_value=_mock_response(RATE_LIMIT_JSON)):
with patch.dict("os.environ", {"ALPHA_VANTAGE_API_KEY": "demo"}):
with pytest.raises(AlphaVantageRateLimitError):
get_insider_transactions("AAPL")
with pytest.raises(AlphaVantageRateLimitError):
get_insider_transactions("AAPL")
# ---------------------------------------------------------------------------
@ -477,10 +459,9 @@ class TestAlphaVantageGetIndicator:
with patch("tradingagents.dataflows.alpha_vantage_common.requests.get",
return_value=_mock_response(CSV_RSI)):
with patch.dict("os.environ", {"ALPHA_VANTAGE_API_KEY": "demo"}):
result = get_indicator(
"AAPL", "rsi", "2024-01-05", look_back_days=5
)
result = get_indicator(
"AAPL", "rsi", "2024-01-05", look_back_days=5
)
assert isinstance(result, str)
assert "RSI" in result.upper()
@ -490,10 +471,9 @@ class TestAlphaVantageGetIndicator:
with patch("tradingagents.dataflows.alpha_vantage_common.requests.get",
return_value=_mock_response(CSV_SMA)):
with patch.dict("os.environ", {"ALPHA_VANTAGE_API_KEY": "demo"}):
result = get_indicator(
"AAPL", "close_50_sma", "2024-01-05", look_back_days=5
)
result = get_indicator(
"AAPL", "close_50_sma", "2024-01-05", look_back_days=5
)
assert isinstance(result, str)
assert "SMA" in result.upper()
@ -510,8 +490,7 @@ class TestAlphaVantageGetIndicator:
with patch("tradingagents.dataflows.alpha_vantage_common.requests.get",
return_value=_mock_response(RATE_LIMIT_JSON)):
with patch.dict("os.environ", {"ALPHA_VANTAGE_API_KEY": "demo"}):
result = get_indicator("AAPL", "rsi", "2024-01-05", look_back_days=5)
result = get_indicator("AAPL", "rsi", "2024-01-05", look_back_days=5)
assert "Error" in result or "rate limit" in result.lower()
@ -519,8 +498,7 @@ class TestAlphaVantageGetIndicator:
"""VWMA is not directly available; a descriptive message is returned."""
from tradingagents.dataflows.alpha_vantage_indicator import get_indicator
with patch.dict("os.environ", {"ALPHA_VANTAGE_API_KEY": "demo"}):
result = get_indicator("AAPL", "vwma", "2024-01-05", look_back_days=5)
result = get_indicator("AAPL", "vwma", "2024-01-05", look_back_days=5)
assert "VWMA" in result
assert "not directly available" in result.lower() or "Volume Weighted" in result

46
tests/test_e2e_api_integration.py
View File

@ -100,8 +100,7 @@ class TestRouteToVendor:
with patch("tradingagents.dataflows.interface.get_config", return_value=patched_config):
with patch("tradingagents.dataflows.alpha_vantage_common.requests.get",
return_value=_mock_av_response(_OHLCV_CSV_AV)):
with patch.dict("os.environ", {"ALPHA_VANTAGE_API_KEY": "demo"}):
result = route_to_vendor("get_stock_data", "AAPL", "2024-01-04", "2024-01-05")
result = route_to_vendor("get_stock_data", "AAPL", "2024-01-04", "2024-01-05")
assert isinstance(result, str)
@ -124,13 +123,12 @@ class TestRouteToVendor:
# AV returns a rate-limit response
with patch("tradingagents.dataflows.alpha_vantage_common.requests.get",
return_value=_mock_av_response(_RATE_LIMIT_JSON)):
with patch.dict("os.environ", {"ALPHA_VANTAGE_API_KEY": "demo"}):
# yfinance is the fallback
with patch("tradingagents.dataflows.y_finance.yf.Ticker",
return_value=mock_ticker):
result = route_to_vendor(
"get_stock_data", "AAPL", "2024-01-04", "2024-01-05"
)
# yfinance is the fallback
with patch("tradingagents.dataflows.y_finance.yf.Ticker",
return_value=mock_ticker):
result = route_to_vendor(
"get_stock_data", "AAPL", "2024-01-04", "2024-01-05"
)
assert isinstance(result, str)
assert "AAPL" in result
@ -150,13 +148,12 @@ class TestRouteToVendor:
with patch("tradingagents.dataflows.interface.get_config", return_value=patched_config):
with patch("tradingagents.dataflows.alpha_vantage_common.requests.get",
return_value=_mock_av_response(_RATE_LIMIT_JSON)):
with patch.dict("os.environ", {"ALPHA_VANTAGE_API_KEY": "demo"}):
with patch(
"tradingagents.dataflows.y_finance.yf.Ticker",
side_effect=ConnectionError("network unavailable"),
):
with pytest.raises(RuntimeError, match="No available vendor"):
route_to_vendor("get_stock_data", "AAPL", "2024-01-04", "2024-01-05")
with patch(
"tradingagents.dataflows.y_finance.yf.Ticker",
side_effect=ConnectionError("network unavailable"),
):
with pytest.raises(RuntimeError, match="No available vendor"):
route_to_vendor("get_stock_data", "AAPL", "2024-01-04", "2024-01-05")
def test_unknown_method_raises_value_error(self):
from tradingagents.dataflows.interface import route_to_vendor
@ -195,8 +192,7 @@ class TestFullPipeline:
with patch("tradingagents.dataflows.alpha_vantage_common.requests.get",
return_value=_mock_av_response(_OHLCV_CSV_AV)):
with patch.dict("os.environ", {"ALPHA_VANTAGE_API_KEY": "demo"}):
result = get_stock("AAPL", "2024-01-04", "2024-01-05")
result = get_stock("AAPL", "2024-01-04", "2024-01-05")
assert isinstance(result, str)
# pandas may reformat "185.00" → "185.0"; check for the numeric value
@ -230,8 +226,7 @@ class TestFullPipeline:
with patch("tradingagents.dataflows.alpha_vantage_common.requests.get",
return_value=_mock_av_response(_OVERVIEW_JSON)):
with patch.dict("os.environ", {"ALPHA_VANTAGE_API_KEY": "demo"}):
result = get_fundamentals("AAPL")
result = get_fundamentals("AAPL")
assert "Apple Inc" in result
assert "TECHNOLOGY" in result
@ -262,8 +257,7 @@ class TestFullPipeline:
with patch("tradingagents.dataflows.alpha_vantage_common.requests.get",
return_value=_mock_av_response(_NEWS_JSON)):
with patch.dict("os.environ", {"ALPHA_VANTAGE_API_KEY": "demo"}):
result = get_news("AAPL", "2024-01-01", "2024-01-05")
result = get_news("AAPL", "2024-01-01", "2024-01-05")
assert "Apple Hits Record High" in result
@ -288,8 +282,7 @@ class TestFullPipeline:
# --- Step 2: Alpha Vantage fundamentals ---
with patch("tradingagents.dataflows.alpha_vantage_common.requests.get",
return_value=_mock_av_response(_OVERVIEW_JSON)):
with patch.dict("os.environ", {"ALPHA_VANTAGE_API_KEY": "demo"}):
fundamentals = get_fundamentals("AAPL")
fundamentals = get_fundamentals("AAPL")
# --- Assertions ---
assert isinstance(price_data, str)
@ -330,9 +323,8 @@ class TestFullPipeline:
# Alpha Vantage rate-limits
with patch("tradingagents.dataflows.alpha_vantage_common.requests.get",
return_value=_mock_av_response(_RATE_LIMIT_JSON)):
with patch.dict("os.environ", {"ALPHA_VANTAGE_API_KEY": "demo"}):
with pytest.raises(AlphaVantageRateLimitError):
get_fundamentals("AAPL")
with pytest.raises(AlphaVantageRateLimitError):
get_fundamentals("AAPL")
# ---------------------------------------------------------------------------

34
tests/test_scanner_fallback.py
View File

@ -83,33 +83,29 @@ class TestAlphaVantageFailoverRaise:
def test_sector_perf_raises_on_total_failure(self):
"""When every GLOBAL_QUOTE call fails, the function should raise."""
with patch.dict(os.environ, {"ALPHA_VANTAGE_API_KEY": "demo"}):
with pytest.raises(AlphaVantageError, match="All .* sector queries failed"):
get_sector_performance_alpha_vantage()
with pytest.raises(AlphaVantageError, match="All .* sector queries failed"):
get_sector_performance_alpha_vantage()
def test_industry_perf_raises_on_total_failure(self):
"""When every ticker quote fails, the function should raise."""
with patch.dict(os.environ, {"ALPHA_VANTAGE_API_KEY": "demo"}):
with pytest.raises(AlphaVantageError, match="All .* ticker queries failed"):
get_industry_performance_alpha_vantage("technology")
with pytest.raises(AlphaVantageError, match="All .* ticker queries failed"):
get_industry_performance_alpha_vantage("technology")
class TestRouteToVendorFallback:
"""Verify route_to_vendor falls back from AV to yfinance."""
def test_sector_perf_falls_back_to_yfinance(self):
with patch.dict(os.environ, {"ALPHA_VANTAGE_API_KEY": "demo"}):
from tradingagents.dataflows.interface import route_to_vendor
result = route_to_vendor("get_sector_performance")
# Should get yfinance data (no "Alpha Vantage" in header)
assert "Sector Performance Overview" in result
# Should have actual percentage data, not all errors
assert "Error:" not in result or result.count("Error:") < 3
from tradingagents.dataflows.interface import route_to_vendor
result = route_to_vendor("get_sector_performance")
# Should get yfinance data (no "Alpha Vantage" in header)
assert "Sector Performance Overview" in result
# Should have actual percentage data, not all errors
assert "Error:" not in result or result.count("Error:") < 3
def test_industry_perf_falls_back_to_yfinance(self):
with patch.dict(os.environ, {"ALPHA_VANTAGE_API_KEY": "demo"}):
from tradingagents.dataflows.interface import route_to_vendor
result = route_to_vendor("get_industry_performance", "technology")
assert "Industry Performance" in result
# Should contain real ticker symbols
assert "N/A" not in result or result.count("N/A") < 5
from tradingagents.dataflows.interface import route_to_vendor
result = route_to_vendor("get_industry_performance", "technology")
assert "Industry Performance" in result
# Should contain real ticker symbols
assert "N/A" not in result or result.count("N/A") < 5

729
tests/test_scanner_mocked.py Normal file
View File

@ -0,0 +1,729 @@
"""Offline mocked tests for the market-wide scanner layer.
Covers both yfinance and Alpha Vantage scanner functions, plus the
route_to_vendor scanner routing. All external calls are mocked so
these tests run without a network connection or API key.
"""
import json
import pandas as pd
import pytest
from datetime import date, datetime
from unittest.mock import patch, MagicMock
# ---------------------------------------------------------------------------
# Helpers — mock data factories
# ---------------------------------------------------------------------------
def _av_response(payload: dict | str) -> MagicMock:
"""Build a mock requests.Response wrapping a JSON dict or raw string."""
resp = MagicMock()
resp.status_code = 200
resp.text = json.dumps(payload) if isinstance(payload, dict) else payload
resp.raise_for_status = MagicMock()
return resp
def _global_quote(symbol: str, price: float = 480.0, change: float = 2.5,
change_pct: str = "0.52%") -> dict:
return {
"Global Quote": {
"01. symbol": symbol,
"05. price": str(price),
"09. change": str(change),
"10. change percent": change_pct,
}
}
def _time_series_daily(symbol: str) -> dict:
"""Return a minimal TIME_SERIES_DAILY JSON payload."""
return {
"Meta Data": {"2. Symbol": symbol},
"Time Series (Daily)": {
"2024-01-08": {"4. close": "482.00"},
"2024-01-05": {"4. close": "480.00"},
"2024-01-04": {"4. close": "475.00"},
},
}
_TOP_GAINERS_LOSERS = {
"top_gainers": [
{"ticker": "NVDA", "price": "620.00", "change_percentage": "5.10%", "volume": "45000000"},
{"ticker": "AMD", "price": "175.00", "change_percentage": "3.20%", "volume": "32000000"},
],
"top_losers": [
{"ticker": "INTC", "price": "31.00", "change_percentage": "-4.50%", "volume": "28000000"},
],
"most_actively_traded": [
{"ticker": "TSLA", "price": "240.00", "change_percentage": "1.80%", "volume": "90000000"},
],
}
_NEWS_SENTIMENT = {
"feed": [
{
"title": "AI Stocks Rally on Positive Earnings",
"summary": "Tech stocks continued their upward climb.",
"source": "Reuters",
"url": "https://example.com/news/1",
"time_published": "20240108T130000",
"overall_sentiment_score": 0.35,
}
]
}
# ---------------------------------------------------------------------------
# yfinance scanner — get_market_movers_yfinance
# ---------------------------------------------------------------------------
class TestYfinanceScannerMarketMovers:
"""Offline tests for get_market_movers_yfinance."""
def _screener_data(self, category: str = "day_gainers") -> dict:
return {
"quotes": [
{
"symbol": "NVDA",
"shortName": "NVIDIA Corp",
"regularMarketPrice": 620.00,
"regularMarketChangePercent": 5.10,
"regularMarketVolume": 45_000_000,
"marketCap": 1_500_000_000_000,
},
{
"symbol": "AMD",
"shortName": "Advanced Micro Devices",
"regularMarketPrice": 175.00,
"regularMarketChangePercent": 3.20,
"regularMarketVolume": 32_000_000,
"marketCap": 280_000_000_000,
},
]
}
def test_returns_markdown_table_for_day_gainers(self):
from tradingagents.dataflows.yfinance_scanner import get_market_movers_yfinance
with patch("tradingagents.dataflows.yfinance_scanner.yf.screener.screen",
return_value=self._screener_data()):
result = get_market_movers_yfinance("day_gainers")
assert isinstance(result, str)
assert "Market Movers" in result
assert "NVDA" in result
assert "5.10%" in result
assert "|" in result # markdown table
def test_returns_markdown_table_for_day_losers(self):
from tradingagents.dataflows.yfinance_scanner import get_market_movers_yfinance
data = {"quotes": [{"symbol": "INTC", "shortName": "Intel", "regularMarketPrice": 31.00,
"regularMarketChangePercent": -4.5, "regularMarketVolume": 28_000_000,
"marketCap": 130_000_000_000}]}
with patch("tradingagents.dataflows.yfinance_scanner.yf.screener.screen",
return_value=data):
result = get_market_movers_yfinance("day_losers")
assert "Market Movers" in result
assert "INTC" in result
def test_invalid_category_returns_error_string(self):
from tradingagents.dataflows.yfinance_scanner import get_market_movers_yfinance
result = get_market_movers_yfinance("not_a_category")
assert "Invalid category" in result
def test_empty_quotes_returns_no_data_message(self):
from tradingagents.dataflows.yfinance_scanner import get_market_movers_yfinance
with patch("tradingagents.dataflows.yfinance_scanner.yf.screener.screen",
return_value={"quotes": []}):
result = get_market_movers_yfinance("day_gainers")
assert "No quotes found" in result
def test_api_error_returns_error_string(self):
from tradingagents.dataflows.yfinance_scanner import get_market_movers_yfinance
with patch("tradingagents.dataflows.yfinance_scanner.yf.screener.screen",
side_effect=Exception("network failure")):
result = get_market_movers_yfinance("day_gainers")
assert "Error" in result
# ---------------------------------------------------------------------------
# yfinance scanner — get_market_indices_yfinance
# ---------------------------------------------------------------------------
class TestYfinanceScannerMarketIndices:
"""Offline tests for get_market_indices_yfinance."""
def _make_multi_etf_df(self) -> pd.DataFrame:
"""Build a minimal multi-ticker Close DataFrame as yf.download returns."""
symbols = ["^GSPC", "^DJI", "^IXIC", "^VIX", "^RUT"]
idx = pd.date_range("2024-01-04", periods=3, freq="B", tz="UTC")
closes = pd.DataFrame(
{s: [4800.0 + i * 10, 4810.0 + i * 10, 4820.0 + i * 10] for i, s in enumerate(symbols)},
index=idx,
)
return pd.DataFrame({"Close": closes})
def test_returns_markdown_table_with_indices(self):
from tradingagents.dataflows.yfinance_scanner import get_market_indices_yfinance
# Multi-symbol download returns a MultiIndex DataFrame
symbols = ["^GSPC", "^DJI", "^IXIC", "^VIX", "^RUT"]
idx = pd.date_range("2024-01-04", periods=5, freq="B")
close_data = {s: [4800.0 + i for i in range(5)] for s in symbols}
# yf.download with multiple symbols returns DataFrame with MultiIndex columns
multi_df = pd.DataFrame(close_data, index=idx)
multi_df.columns = pd.MultiIndex.from_product([["Close"], symbols])
with patch("tradingagents.dataflows.yfinance_scanner.yf.download",
return_value=multi_df):
result = get_market_indices_yfinance()
assert isinstance(result, str)
assert "Market Indices" in result or "Index" in result.split("\n")[0]
def test_returns_string_on_download_error(self):
from tradingagents.dataflows.yfinance_scanner import get_market_indices_yfinance
with patch("tradingagents.dataflows.yfinance_scanner.yf.download",
side_effect=Exception("network error")):
result = get_market_indices_yfinance()
assert isinstance(result, str)
# ---------------------------------------------------------------------------
# yfinance scanner — get_sector_performance_yfinance
# ---------------------------------------------------------------------------
class TestYfinanceScannerSectorPerformance:
"""Offline tests for get_sector_performance_yfinance."""
def _make_sector_df(self) -> pd.DataFrame:
"""Multi-symbol ETF DataFrame covering 6 months of daily closes."""
etfs = ["XLK", "XLV", "XLF", "XLE", "XLY", "XLP", "XLI", "XLB", "XLRE", "XLU", "XLC"]
# 130 trading days ~ 6 months
idx = pd.date_range("2023-07-01", periods=130, freq="B")
data = {e: [100.0 + i * 0.01 for i in range(130)] for e in etfs}
df = pd.DataFrame(data, index=idx)
df.columns = pd.MultiIndex.from_product([["Close"], etfs])
return df
def test_returns_sector_performance_table(self):
from tradingagents.dataflows.yfinance_scanner import get_sector_performance_yfinance
with patch("tradingagents.dataflows.yfinance_scanner.yf.download",
return_value=self._make_sector_df()):
result = get_sector_performance_yfinance()
assert isinstance(result, str)
assert "Sector Performance Overview" in result
assert "|" in result
def test_contains_all_sectors(self):
from tradingagents.dataflows.yfinance_scanner import get_sector_performance_yfinance
with patch("tradingagents.dataflows.yfinance_scanner.yf.download",
return_value=self._make_sector_df()):
result = get_sector_performance_yfinance()
# 11 GICS sectors should all appear
for sector in ["Technology", "Healthcare", "Financials", "Energy"]:
assert sector in result
def test_download_error_returns_error_string(self):
from tradingagents.dataflows.yfinance_scanner import get_sector_performance_yfinance
with patch("tradingagents.dataflows.yfinance_scanner.yf.download",
side_effect=Exception("connection refused")):
result = get_sector_performance_yfinance()
assert "Error" in result
# ---------------------------------------------------------------------------
# yfinance scanner — get_industry_performance_yfinance
# ---------------------------------------------------------------------------
class TestYfinanceScannerIndustryPerformance:
"""Offline tests for get_industry_performance_yfinance."""
def _mock_sector_with_companies(self) -> MagicMock:
top_companies = pd.DataFrame(
{
"name": ["Apple Inc.", "Microsoft Corp", "NVIDIA Corp"],
"rating": [4.5, 4.8, 4.2],
"market weight": [0.072, 0.065, 0.051],
},
index=pd.Index(["AAPL", "MSFT", "NVDA"], name="symbol"),
)
mock_sector = MagicMock()
mock_sector.top_companies = top_companies
return mock_sector
def test_returns_industry_table_for_valid_sector(self):
from tradingagents.dataflows.yfinance_scanner import get_industry_performance_yfinance
with patch("tradingagents.dataflows.yfinance_scanner.yf.Sector",
return_value=self._mock_sector_with_companies()):
result = get_industry_performance_yfinance("technology")
assert isinstance(result, str)
assert "Industry Performance" in result
assert "AAPL" in result
assert "Apple Inc." in result
def test_empty_top_companies_returns_no_data_message(self):
from tradingagents.dataflows.yfinance_scanner import get_industry_performance_yfinance
mock_sector = MagicMock()
mock_sector.top_companies = pd.DataFrame()
with patch("tradingagents.dataflows.yfinance_scanner.yf.Sector",
return_value=mock_sector):
result = get_industry_performance_yfinance("technology")
assert "No industry data found" in result
def test_none_top_companies_returns_no_data_message(self):
from tradingagents.dataflows.yfinance_scanner import get_industry_performance_yfinance
mock_sector = MagicMock()
mock_sector.top_companies = None
with patch("tradingagents.dataflows.yfinance_scanner.yf.Sector",
return_value=mock_sector):
result = get_industry_performance_yfinance("healthcare")
assert "No industry data found" in result
def test_sector_error_returns_error_string(self):
from tradingagents.dataflows.yfinance_scanner import get_industry_performance_yfinance
with patch("tradingagents.dataflows.yfinance_scanner.yf.Sector",
side_effect=Exception("yfinance unavailable")):
result = get_industry_performance_yfinance("technology")
assert "Error" in result
# ---------------------------------------------------------------------------
# yfinance scanner — get_topic_news_yfinance
# ---------------------------------------------------------------------------
class TestYfinanceScannerTopicNews:
"""Offline tests for get_topic_news_yfinance."""
def _mock_search(self, title: str = "AI Revolution in Tech") -> MagicMock:
mock_search = MagicMock()
mock_search.news = [
{
"title": title,
"publisher": "TechCrunch",
"link": "https://techcrunch.com/story",
"summary": "Artificial intelligence is transforming the industry.",
}
]
return mock_search
def test_returns_formatted_news_for_topic(self):
from tradingagents.dataflows.yfinance_scanner import get_topic_news_yfinance
with patch("tradingagents.dataflows.yfinance_scanner.yf.Search",
return_value=self._mock_search()):
result = get_topic_news_yfinance("artificial intelligence")
assert isinstance(result, str)
assert "AI Revolution in Tech" in result
assert "News for Topic" in result
def test_no_results_returns_no_news_message(self):
from tradingagents.dataflows.yfinance_scanner import get_topic_news_yfinance
mock_search = MagicMock()
mock_search.news = []
with patch("tradingagents.dataflows.yfinance_scanner.yf.Search",
return_value=mock_search):
result = get_topic_news_yfinance("obscure_topic")
assert "No news found" in result
def test_handles_nested_content_structure(self):
from tradingagents.dataflows.yfinance_scanner import get_topic_news_yfinance
mock_search = MagicMock()
mock_search.news = [
{
"content": {
"title": "Semiconductor Demand Surges",
"summary": "Chip makers report record orders.",
"provider": {"displayName": "Bloomberg"},
"canonicalUrl": {"url": "https://bloomberg.com/chips"},
}
}
]
with patch("tradingagents.dataflows.yfinance_scanner.yf.Search",
return_value=mock_search):
result = get_topic_news_yfinance("semiconductors")
assert "Semiconductor Demand Surges" in result
# ---------------------------------------------------------------------------
# Alpha Vantage scanner — get_market_movers_alpha_vantage
# ---------------------------------------------------------------------------
class TestAVScannerMarketMovers:
"""Offline mocked tests for get_market_movers_alpha_vantage."""
def test_day_gainers_returns_markdown_table(self):
from tradingagents.dataflows.alpha_vantage_scanner import get_market_movers_alpha_vantage
with patch("tradingagents.dataflows.alpha_vantage_scanner._rate_limited_request",
return_value=json.dumps(_TOP_GAINERS_LOSERS)):
result = get_market_movers_alpha_vantage("day_gainers")
assert "Market Movers" in result
assert "NVDA" in result
assert "5.10%" in result
assert "|" in result
def test_day_losers_returns_markdown_table(self):
from tradingagents.dataflows.alpha_vantage_scanner import get_market_movers_alpha_vantage
with patch("tradingagents.dataflows.alpha_vantage_scanner._rate_limited_request",
return_value=json.dumps(_TOP_GAINERS_LOSERS)):
result = get_market_movers_alpha_vantage("day_losers")
assert "INTC" in result
def test_most_actives_returns_markdown_table(self):
from tradingagents.dataflows.alpha_vantage_scanner import get_market_movers_alpha_vantage
with patch("tradingagents.dataflows.alpha_vantage_scanner._rate_limited_request",
return_value=json.dumps(_TOP_GAINERS_LOSERS)):
result = get_market_movers_alpha_vantage("most_actives")
assert "TSLA" in result
def test_invalid_category_raises_value_error(self):
from tradingagents.dataflows.alpha_vantage_scanner import get_market_movers_alpha_vantage
with pytest.raises(ValueError, match="Invalid category"):
get_market_movers_alpha_vantage("not_valid")
def test_rate_limit_error_propagates(self):
from tradingagents.dataflows.alpha_vantage_scanner import get_market_movers_alpha_vantage
from tradingagents.dataflows.alpha_vantage_common import RateLimitError
with patch("tradingagents.dataflows.alpha_vantage_scanner._rate_limited_request",
side_effect=RateLimitError("rate limited")):
with pytest.raises(RateLimitError):
get_market_movers_alpha_vantage("day_gainers")
# ---------------------------------------------------------------------------
# Alpha Vantage scanner — get_market_indices_alpha_vantage
# ---------------------------------------------------------------------------
class TestAVScannerMarketIndices:
"""Offline mocked tests for get_market_indices_alpha_vantage."""
def test_returns_markdown_table_with_index_names(self):
from tradingagents.dataflows.alpha_vantage_scanner import get_market_indices_alpha_vantage
def fake_request(function_name, params, **kwargs):
symbol = params.get("symbol", "SPY")
return json.dumps(_global_quote(symbol))
with patch("tradingagents.dataflows.alpha_vantage_scanner._rate_limited_request",
side_effect=fake_request):
result = get_market_indices_alpha_vantage()
assert "Market Indices" in result
assert "|" in result
assert any(name in result for name in ["S&P 500", "Dow Jones", "NASDAQ"])
def test_all_proxies_appear_in_output(self):
from tradingagents.dataflows.alpha_vantage_scanner import get_market_indices_alpha_vantage
def fake_request(function_name, params, **kwargs):
return json.dumps(_global_quote(params.get("symbol", "SPY")))
with patch("tradingagents.dataflows.alpha_vantage_scanner._rate_limited_request",
side_effect=fake_request):
result = get_market_indices_alpha_vantage()
# All 4 ETF proxies should appear
for proxy in ["SPY", "DIA", "QQQ", "IWM"]:
assert proxy in result
# ---------------------------------------------------------------------------
# Alpha Vantage scanner — get_sector_performance_alpha_vantage
# ---------------------------------------------------------------------------
class TestAVScannerSectorPerformance:
"""Offline mocked tests for get_sector_performance_alpha_vantage."""
def _make_fake_request(self):
"""Return a side_effect function handling both GLOBAL_QUOTE and TIME_SERIES_DAILY."""
def fake(function_name, params, **kwargs):
if function_name == "GLOBAL_QUOTE":
symbol = params.get("symbol", "XLK")
return json.dumps(_global_quote(symbol))
elif function_name == "TIME_SERIES_DAILY":
symbol = params.get("symbol", "XLK")
return json.dumps(_time_series_daily(symbol))
return json.dumps({})
return fake
def test_returns_sector_table_with_percentages(self):
from tradingagents.dataflows.alpha_vantage_scanner import get_sector_performance_alpha_vantage
with patch("tradingagents.dataflows.alpha_vantage_scanner._rate_limited_request",
side_effect=self._make_fake_request()):
result = get_sector_performance_alpha_vantage()
assert "Sector Performance Overview" in result
assert "|" in result
def test_all_eleven_sectors_in_output(self):
from tradingagents.dataflows.alpha_vantage_scanner import get_sector_performance_alpha_vantage
with patch("tradingagents.dataflows.alpha_vantage_scanner._rate_limited_request",
side_effect=self._make_fake_request()):
result = get_sector_performance_alpha_vantage()
for sector in ["Technology", "Healthcare", "Financials", "Energy"]:
assert sector in result
def test_all_errors_raises_alpha_vantage_error(self):
"""If ALL sector ETF requests fail, AlphaVantageError is raised for fallback."""
from tradingagents.dataflows.alpha_vantage_scanner import get_sector_performance_alpha_vantage
from tradingagents.dataflows.alpha_vantage_common import AlphaVantageError, RateLimitError
with patch("tradingagents.dataflows.alpha_vantage_scanner._rate_limited_request",
side_effect=RateLimitError("rate limited")):
with pytest.raises(AlphaVantageError):
get_sector_performance_alpha_vantage()
# ---------------------------------------------------------------------------
# Alpha Vantage scanner — get_industry_performance_alpha_vantage
# ---------------------------------------------------------------------------
class TestAVScannerIndustryPerformance:
"""Offline mocked tests for get_industry_performance_alpha_vantage."""
def test_returns_table_for_technology_sector(self):
from tradingagents.dataflows.alpha_vantage_scanner import get_industry_performance_alpha_vantage
def fake_request(function_name, params, **kwargs):
symbol = params.get("symbol", "AAPL")
return json.dumps(_global_quote(symbol, price=185.0, change_pct="+1.20%"))
with patch("tradingagents.dataflows.alpha_vantage_scanner._rate_limited_request",
side_effect=fake_request):
result = get_industry_performance_alpha_vantage("technology")
assert "Industry Performance" in result
assert "|" in result
assert any(t in result for t in ["AAPL", "MSFT", "NVDA"])
def test_invalid_sector_raises_value_error(self):
from tradingagents.dataflows.alpha_vantage_scanner import get_industry_performance_alpha_vantage
with pytest.raises(ValueError, match="Unknown sector"):
get_industry_performance_alpha_vantage("not_a_real_sector")
def test_sorted_by_change_percent_descending(self):
"""Results should be sorted by change % descending."""
from tradingagents.dataflows.alpha_vantage_scanner import get_industry_performance_alpha_vantage
# Alternate high/low changes to verify sort order
prices = {"AAPL": ("180.00", "+5.00%"), "MSFT": ("380.00", "+1.00%"),
"NVDA": ("620.00", "+8.00%"), "GOOGL": ("140.00", "+2.50%"),
"META": ("350.00", "+3.10%"), "AVGO": ("850.00", "+0.50%"),
"ADBE": ("550.00", "+4.20%"), "CRM": ("275.00", "+1.80%"),
"AMD": ("170.00", "+6.30%"), "INTC": ("31.00", "-2.10%")}
def fake_request(function_name, params, **kwargs):
symbol = params.get("symbol", "AAPL")
p, c = prices.get(symbol, ("100.00", "0.00%"))
return json.dumps({
"Global Quote": {"01. symbol": symbol, "05. price": p,
"09. change": "1.00", "10. change percent": c}
})
with patch("tradingagents.dataflows.alpha_vantage_scanner._rate_limited_request",
side_effect=fake_request):
result = get_industry_performance_alpha_vantage("technology")
# NVDA (+8%) should appear before INTC (-2.1%)
nvda_pos = result.find("NVDA")
intc_pos = result.find("INTC")
assert nvda_pos != -1 and intc_pos != -1
assert nvda_pos < intc_pos
# ---------------------------------------------------------------------------
# Alpha Vantage scanner — get_topic_news_alpha_vantage
# ---------------------------------------------------------------------------
class TestAVScannerTopicNews:
"""Offline mocked tests for get_topic_news_alpha_vantage."""
def test_returns_news_articles_for_known_topic(self):
from tradingagents.dataflows.alpha_vantage_scanner import get_topic_news_alpha_vantage
with patch("tradingagents.dataflows.alpha_vantage_scanner._rate_limited_request",
return_value=json.dumps(_NEWS_SENTIMENT)):
result = get_topic_news_alpha_vantage("market", limit=5)
assert "News for Topic" in result
assert "AI Stocks Rally on Positive Earnings" in result
def test_known_topic_is_mapped_to_av_value(self):
"""Topic strings like 'market' are remapped to AV-specific topic keys."""
from tradingagents.dataflows.alpha_vantage_scanner import get_topic_news_alpha_vantage
captured = {}
def capture_request(function_name, params, **kwargs):
captured.update(params)
return json.dumps(_NEWS_SENTIMENT)
with patch("tradingagents.dataflows.alpha_vantage_scanner._rate_limited_request",
side_effect=capture_request):
get_topic_news_alpha_vantage("market", limit=5)
# "market" maps to "financial_markets" in _TOPIC_MAP
assert captured.get("topics") == "financial_markets"
def test_unknown_topic_passed_through(self):
"""Topics not in the map are forwarded to the API as-is."""
from tradingagents.dataflows.alpha_vantage_scanner import get_topic_news_alpha_vantage
captured = {}
def capture_request(function_name, params, **kwargs):
captured.update(params)
return json.dumps(_NEWS_SENTIMENT)
with patch("tradingagents.dataflows.alpha_vantage_scanner._rate_limited_request",
side_effect=capture_request):
get_topic_news_alpha_vantage("custom_topic", limit=3)
assert captured.get("topics") == "custom_topic"
def test_empty_feed_returns_no_articles_message(self):
from tradingagents.dataflows.alpha_vantage_scanner import get_topic_news_alpha_vantage
empty = {"feed": []}
with patch("tradingagents.dataflows.alpha_vantage_scanner._rate_limited_request",
return_value=json.dumps(empty)):
result = get_topic_news_alpha_vantage("earnings", limit=5)
assert "No articles" in result
def test_rate_limit_error_propagates(self):
from tradingagents.dataflows.alpha_vantage_scanner import get_topic_news_alpha_vantage
from tradingagents.dataflows.alpha_vantage_common import RateLimitError
with patch("tradingagents.dataflows.alpha_vantage_scanner._rate_limited_request",
side_effect=RateLimitError("rate limited")):
with pytest.raises(RateLimitError):
get_topic_news_alpha_vantage("technology")
# ---------------------------------------------------------------------------
# Scanner routing — route_to_vendor for scanner methods
# ---------------------------------------------------------------------------
class TestScannerRouting:
"""End-to-end routing tests for scanner_data methods via route_to_vendor."""
def test_get_market_movers_routes_to_yfinance_by_default(self):
"""Default config uses yfinance for scanner_data."""
from tradingagents.dataflows.interface import route_to_vendor
screener_data = {
"quotes": [{"symbol": "NVDA", "shortName": "NVIDIA", "regularMarketPrice": 620.0,
"regularMarketChangePercent": 5.1, "regularMarketVolume": 45_000_000,
"marketCap": 1_500_000_000_000}]
}
with patch("tradingagents.dataflows.yfinance_scanner.yf.screener.screen",
return_value=screener_data):
result = route_to_vendor("get_market_movers", "day_gainers")
assert isinstance(result, str)
assert "NVDA" in result
def test_get_sector_performance_routes_to_yfinance_by_default(self):
from tradingagents.dataflows.interface import route_to_vendor
etfs = ["XLK", "XLV", "XLF", "XLE", "XLY", "XLP", "XLI", "XLB", "XLRE", "XLU", "XLC"]
idx = pd.date_range("2023-07-01", periods=130, freq="B")
close_data = {e: [100.0 + i * 0.01 for i in range(130)] for e in etfs}
df = pd.DataFrame(close_data, index=idx)
df.columns = pd.MultiIndex.from_product([["Close"], etfs])
with patch("tradingagents.dataflows.yfinance_scanner.yf.download", return_value=df):
result = route_to_vendor("get_sector_performance")
assert isinstance(result, str)
assert "Sector Performance Overview" in result
def test_get_market_movers_falls_back_to_yfinance_when_av_fails(self):
"""When AV scanner raises AlphaVantageError, fallback to yfinance is used."""
from tradingagents.dataflows.interface import route_to_vendor
from tradingagents.dataflows.config import get_config
from tradingagents.dataflows.alpha_vantage_common import AlphaVantageError
original_config = get_config()
patched_config = {
**original_config,
"data_vendors": {**original_config.get("data_vendors", {}), "scanner_data": "alpha_vantage"},
}
screener_data = {
"quotes": [{"symbol": "AMD", "shortName": "AMD", "regularMarketPrice": 175.0,
"regularMarketChangePercent": 3.2, "regularMarketVolume": 32_000_000,
"marketCap": 280_000_000_000}]
}
with patch("tradingagents.dataflows.interface.get_config", return_value=patched_config):
# AV market movers raises → fallback to yfinance
with patch("tradingagents.dataflows.alpha_vantage_scanner._rate_limited_request",
side_effect=AlphaVantageError("rate limited")):
with patch("tradingagents.dataflows.yfinance_scanner.yf.screener.screen",
return_value=screener_data):
result = route_to_vendor("get_market_movers", "day_gainers")
assert isinstance(result, str)
assert "AMD" in result
def test_get_topic_news_routes_correctly(self):
from tradingagents.dataflows.interface import route_to_vendor
mock_search = MagicMock()
mock_search.news = [{"title": "Fed Signals Rate Cut", "publisher": "Reuters",
"link": "https://example.com", "summary": "Fed news."}]
with patch("tradingagents.dataflows.yfinance_scanner.yf.Search",
return_value=mock_search):
result = route_to_vendor("get_topic_news", "economy")
assert isinstance(result, str)

162
tests/test_yfinance_integration.py
View File

@ -403,3 +403,165 @@ class TestGetInsiderTransactions:
result = get_insider_transactions("AAPL")
assert "Error" in result
# ---------------------------------------------------------------------------
# get_stock_stats_indicators_window
# ---------------------------------------------------------------------------
class TestGetStockStatsIndicatorsWindow:
"""Tests for get_stock_stats_indicators_window (technical indicators)."""
def _bulk_rsi_data(self):
"""Return a realistic dict of date→rsi_value as _get_stock_stats_bulk would."""
return {
"2024-01-08": "62.34",
"2024-01-07": "N/A", # weekend
"2024-01-06": "N/A", # weekend
"2024-01-05": "59.12",
"2024-01-04": "55.67",
"2024-01-03": "50.00",
}
def test_returns_formatted_indicator_string(self):
"""Success path: returns a multi-line string with dates and RSI values."""
from tradingagents.dataflows.y_finance import get_stock_stats_indicators_window
with patch(
"tradingagents.dataflows.y_finance._get_stock_stats_bulk",
return_value=self._bulk_rsi_data(),
):
result = get_stock_stats_indicators_window("AAPL", "rsi", "2024-01-08", 5)
assert "rsi" in result
assert "2024-01-08" in result
assert "62.34" in result
def test_includes_indicator_description(self):
"""The returned string includes the indicator description / usage notes."""
from tradingagents.dataflows.y_finance import get_stock_stats_indicators_window
with patch(
"tradingagents.dataflows.y_finance._get_stock_stats_bulk",
return_value=self._bulk_rsi_data(),
):
result = get_stock_stats_indicators_window("AAPL", "rsi", "2024-01-08", 5)
# Every supported indicator has a description string
assert "RSI" in result or "momentum" in result.lower()
def test_unsupported_indicator_raises_value_error(self):
"""Requesting an unsupported indicator raises ValueError before any network call."""
from tradingagents.dataflows.y_finance import get_stock_stats_indicators_window
with pytest.raises(ValueError, match="not supported"):
get_stock_stats_indicators_window("AAPL", "unknown_indicator", "2024-01-08", 5)
def test_bulk_exception_triggers_fallback(self):
"""If _get_stock_stats_bulk raises, the function falls back gracefully."""
from tradingagents.dataflows.y_finance import get_stock_stats_indicators_window
with patch(
"tradingagents.dataflows.y_finance._get_stock_stats_bulk",
side_effect=Exception("stockstats unavailable"),
):
with patch(
"tradingagents.dataflows.y_finance.get_stockstats_indicator",
return_value="45.00",
):
result = get_stock_stats_indicators_window("AAPL", "rsi", "2024-01-08", 3)
assert isinstance(result, str)
assert "rsi" in result
# ---------------------------------------------------------------------------
# get_global_news_yfinance
# ---------------------------------------------------------------------------
class TestGetGlobalNewsYfinance:
"""Tests for get_global_news_yfinance."""
def _mock_search_with_article(self):
"""Return a mock yf.Search object with one flat-structured news article."""
mock_search = MagicMock()
mock_search.news = [
{
"title": "Fed Holds Rates Steady",
"publisher": "Reuters",
"link": "https://example.com/fed",
"summary": "The Federal Reserve decided to hold interest rates.",
}
]
return mock_search
def test_returns_string_with_articles(self):
"""When yfinance Search returns articles, a formatted string is returned."""
from tradingagents.dataflows.yfinance_news import get_global_news_yfinance
with patch(
"tradingagents.dataflows.yfinance_news.yf.Search",
return_value=self._mock_search_with_article(),
):
result = get_global_news_yfinance("2024-01-15", look_back_days=7)
assert isinstance(result, str)
assert "Fed Holds Rates Steady" in result
def test_no_news_returns_fallback_message(self):
"""When no articles are found, a 'no news found' message is returned."""
from tradingagents.dataflows.yfinance_news import get_global_news_yfinance
mock_search = MagicMock()
mock_search.news = []
with patch(
"tradingagents.dataflows.yfinance_news.yf.Search",
return_value=mock_search,
):
result = get_global_news_yfinance("2024-01-15")
assert "No global news found" in result
def test_handles_nested_content_structure(self):
"""Articles with nested 'content' key are parsed correctly."""
from tradingagents.dataflows.yfinance_news import get_global_news_yfinance
mock_search = MagicMock()
mock_search.news = [
{
"content": {
"title": "Inflation Report Beats Expectations",
"summary": "CPI data came in below forecasts.",
"provider": {"displayName": "Bloomberg"},
"canonicalUrl": {"url": "https://bloomberg.com/story"},
"pubDate": "2024-01-15T10:00:00Z",
}
}
]
with patch(
"tradingagents.dataflows.yfinance_news.yf.Search",
return_value=mock_search,
):
result = get_global_news_yfinance("2024-01-15", look_back_days=3)
assert "Inflation Report Beats Expectations" in result
def test_deduplicates_articles_across_queries(self):
"""Duplicate titles from multiple search queries appear only once."""
from tradingagents.dataflows.yfinance_news import get_global_news_yfinance
same_article = {"title": "Market Rally Continues", "publisher": "AP", "link": ""}
mock_search = MagicMock()
mock_search.news = [same_article]
with patch(
"tradingagents.dataflows.yfinance_news.yf.Search",
return_value=mock_search,
):
result = get_global_news_yfinance("2024-01-15", look_back_days=7, limit=5)
# Title should appear exactly once despite multiple search queries
assert result.count("Market Rally Continues") == 1