（7-3-05）基于MCP实现的金融投资Agent(5)市场情绪分析测试+ 个股数据测试-平芜编程栈

7.3.8 市场情绪分析测试

文件test_sentiment.py实现了对市场情绪分析工具的测试，涵盖了CNN恐惧与贪婪指数、加密货币恐惧与贪婪指数以及谷歌趋势数据的获取功能。通过模拟数据请求和返回结果，测试了各类情绪指标在不同参数设置（如获取当前数据、历史数据、指定指标等）、异常情况（如无效指标、无数据返回）下的表现，确保情绪分析工具能准确、稳定地提供数据。

class TestSentimentTools: """测试情绪分析工具。""" @patch('investor_agent.server.fetch_fng_data') async def test_cnn_fear_greed_current(self, mock_fetch, mock_fear_greed_data): """测试获取当前CNN恐惧与贪婪指数。""" mock_fetch.return_value = mock_fear_greed_data result = await get_cnn_fear_greed_index(days=0) # 当days=0时应排除历史数据 assert 'fear_and_greed' in result assert 'put_call_options' in result assert 'market_volatility_vix' in result assert 'fear_and_greed_historical' not in result # 验证模拟函数被调用 mock_fetch.assert_called_once() @patch('investor_agent.server.fetch_fng_data') async def test_cnn_fear_greed_historical(self, mock_fetch): """测试获取历史CNN恐惧与贪婪指数。""" historical_data = { 'fear_and_greed': {'value': 45, 'description': 'Neutral'}, 'fear_and_greed_historical': { 'data': [ {'date': '2024-01-01', 'value': 50}, {'date': '2024-01-02', 'value': 48}, {'date': '2024-01-03', 'value': 45} ] } } mock_fetch.return_value = historical_data result = await get_cnn_fear_greed_index(days=3) assert 'fear_and_greed_historical' in result assert len(result['fear_and_greed_historical']['data']) == 3 @patch('investor_agent.server.fetch_fng_data') async def test_cnn_fear_greed_specific_indicators(self, mock_fetch, mock_fear_greed_data): """测试仅获取特定指标。""" mock_fetch.return_value = mock_fear_greed_data result = await get_cnn_fear_greed_index( days=0, indicators=['fear_and_greed', 'market_volatility_vix'] ) # 应只包含请求的指标 assert 'fear_and_greed' in result assert 'market_volatility_vix' in result assert 'put_call_options' not in result @patch('investor_agent.server.fetch_fng_data') async def test_cnn_fear_greed_invalid_indicators(self, mock_fetch, mock_fear_greed_data): """测试对无效指标的错误处理。""" mock_fetch.return_value = mock_fear_greed_data with pytest.raises(ValueError, match="Invalid indicators"): await get_cnn_fear_greed_index(indicators=['invalid_indicator']) @patch('investor_agent.server.fetch_fng_data') async def test_cnn_fear_greed_no_data(self, mock_fetch): """测试无数据可用时的错误处理。""" mock_fetch.return_value = {} # 空字典在if not data检查中应被视为False with pytest.raises(RuntimeError, match="Unable to fetch CNN Fear & Greed Index data"): await get_cnn_fear_greed_index() @patch('investor_agent.yahoo_finance_utils.create_cached_async_client') async def test_crypto_fear_greed_success(self, mock_client, mock_crypto_fng_data): """测试成功获取加密货币恐惧与贪婪指数。""" # 模拟异步上下文管理器和HTTP响应 mock_response = Mock() mock_response.json.return_value = {"data": mock_crypto_fng_data} # 同步方法 mock_response.raise_for_status = Mock() mock_http_client = AsyncMock() mock_http_client.get = AsyncMock(return_value=mock_response) mock_client.return_value.__aenter__ = AsyncMock(return_value=mock_http_client) result = await get_crypto_fear_greed_index(days=1) assert result == mock_crypto_fng_data mock_http_client.get.assert_called_once_with( "https://api.alternative.me/fng/", params={"limit": 1} ) @patch('investor_agent.yahoo_finance_utils.create_cached_async_client') async def test_crypto_fear_greed_default_days(self, mock_client, mock_crypto_fng_data): """测试使用默认天数参数的加密货币恐惧与贪婪指数。""" mock_response = Mock() mock_response.json.return_value = {"data": mock_crypto_fng_data} # 同步方法 mock_response.raise_for_status = Mock() mock_http_client = AsyncMock() mock_http_client.get = AsyncMock(return_value=mock_response) mock_client.return_value.__aenter__ = AsyncMock(return_value=mock_http_client) result = await get_crypto_fear_greed_index() # 应使用默认的7天 mock_http_client.get.assert_called_once_with( "https://api.alternative.me/fng/", params={"limit": 7} ) @patch('pytrends.request.TrendReq') def test_google_trends_success(self, mock_pytrends): """测试成功获取谷歌趋势数据。""" import pandas as pd # 模拟趋势数据 trends_data = pd.DataFrame({ 'AAPL': [75, 80, 85], 'GOOGL': [60, 65, 70] }) mock_instance = mock_pytrends.return_value mock_instance.interest_over_time.return_value = trends_data result = get_google_trends(['AAPL', 'GOOGL'], period_days=7) # 应返回平均值 expected = {'AAPL': 80.0, 'GOOGL': 65.0} assert result == expected mock_instance.build_payload.assert_called_once_with( ['AAPL', 'GOOGL'], timeframe='now 7-d' ) @patch('pytrends.request.TrendReq') def test_google_trends_no_data(self, mock_pytrends): """测试谷歌趋势无数据返回的情况。""" import pandas as pd mock_instance = mock_pytrends.return_value mock_instance.interest_over_time.return_value = pd.DataFrame() # 空DataFrame with pytest.raises(ValueError, match="No data returned from Google Trends"): get_google_trends(['INVALID'], period_days=7) @patch('pytrends.request.TrendReq') def test_google_trends_single_keyword(self, mock_pytrends): """测试单个关键词的谷歌趋势。""" import pandas as pd trends_data = pd.DataFrame({ 'Bitcoin': [45, 50, 55, 60] }) mock_instance = mock_pytrends.return_value mock_instance.interest_over_time.return_value = trends_data result = get_google_trends(['Bitcoin'], period_days=14) assert result == {'Bitcoin': 52.5} # [45, 50, 55, 60]的平均值 mock_instance.build_payload.assert_called_once_with( ['Bitcoin'], timeframe='now 14-d' )

7.3.9 个股数据测试

文件test_ticker_data.py实现了对get_ticker_data工具的测试，该工具用于获取股票代码的相关信息，包括核心财务指标、即将到来的收益日期和股息日期、最新新闻文章、最新分析师推荐以及近期分析师评级变动。

class TestGetTickerData: """测试get_ticker_data工具，重点关注DataFrame边缘情况。""" @patch('investor_agent.yfinance_utils.get_analyst_data') @patch('investor_agent.yfinance_utils.get_news') @patch('investor_agent.yfinance_utils.get_calendar') @patch('investor_agent.yfinance_utils.get_ticker_info') def test_successful_data_retrieval( self, mock_info, mock_calendar, mock_news, mock_analyst, mock_ticker_info, sample_recommendations_df, sample_news, sample_calendar ): """测试成功获取所有股票数据。""" # 设置模拟 mock_info.return_value = mock_ticker_info mock_calendar.return_value = sample_calendar mock_news.return_value = sample_news mock_analyst.side_effect = [sample_recommendations_df, sample_recommendations_df] # 先推荐，后升级 result = get_ticker_data("AAPL", max_news=2, max_recommendations=2, max_upgrades=2) # 验证结构 assert "info" in result assert "calendar" in result assert "news" in result assert "recommendations" in result assert "upgrades_downgrades" in result # 验证内容 assert result["info"]["symbol"] == "AAPL" assert len(result["news"]) == 2 assert "data" in result["recommendations"] assert "data" in result["upgrades_downgrades"] @patch('investor_agent.yfinance_utils.get_analyst_data') @patch('investor_agent.yfinance_utils.get_news') @patch('investor_agent.yfinance_utils.get_calendar') @patch('investor_agent.yfinance_utils.get_ticker_info') def test_empty_dataframe_handling( self, mock_info, mock_calendar, mock_news, mock_analyst, mock_ticker_info, empty_dataframe ): """测试对空DataFrame的处理——这是我们修复的bug！""" mock_info.return_value = mock_ticker_info mock_calendar.return_value = None mock_news.return_value = None mock_analyst.return_value = empty_dataframe # 两次调用都返回空DataFrame result = get_ticker_data("AAPL") # 应该包含info，但不包含recommendations或upgrades_downgrades assert "info" in result assert result["info"]["symbol"] == "AAPL" assert "recommendations" not in result assert "upgrades_downgrades" not in result assert "calendar" not in result assert "news" not in result # 验证函数在DataFrame布尔值评估时没有崩溃 assert isinstance(result, dict) @patch('investor_agent.yfinance_utils.get_analyst_data') @patch('investor_agent.yfinance_utils.get_news') @patch('investor_agent.yfinance_utils.get_calendar') @patch('investor_agent.yfinance_utils.get_ticker_info') def test_none_dataframe_handling( self, mock_info, mock_calendar, mock_news, mock_analyst, mock_ticker_info ): """测试对分析师数据函数返回None的处理。""" mock_info.return_value = mock_ticker_info mock_calendar.return_value = None mock_news.return_value = None mock_analyst.return_value = None # 两次调用都返回None result = get_ticker_data("AAPL") # 应该包含info，但不包含recommendations或upgrades_downgrades assert "info" in result assert result["info"]["symbol"] == "AAPL" assert "recommendations" not in result assert "upgrades_downgrades" not in result @patch('investor_agent.yfinance_utils.get_analyst_data') @patch('investor_agent.yfinance_utils.get_news') @patch('investor_agent.yfinance_utils.get_calendar') @patch('investor_agent.yfinance_utils.get_ticker_info') def test_mixed_dataframe_states( self, mock_info, mock_calendar, mock_news, mock_analyst, mock_ticker_info, sample_recommendations_df, empty_dataframe ): """测试混合状态：部分数据可用，部分为空，部分为None。""" mock_info.return_value = mock_ticker_info mock_calendar.return_value = None mock_news.return_value = None # 第一次调用（推荐）返回数据，第二次调用（升级）返回空 mock_analyst.side_effect = [sample_recommendations_df, empty_dataframe] result = get_ticker_data("AAPL") # 应该包含info和recommendations，但不包含upgrades_downgrades assert "info" in result assert "recommendations" in result assert "upgrades_downgrades" not in result assert "data" in result["recommendations"] @patch('investor_agent.yfinance_utils.get_ticker_info') def test_no_ticker_info_raises_error(self, mock_info): """测试缺少股票信息时是否抛出ValueError。""" mock_info.return_value = None with pytest.raises(ValueError, match="No information available for INVALID"): get_ticker_data("INVALID") @patch('investor_agent.yfinance_utils.get_ticker_info') def test_empty_ticker_info_raises_error(self, mock_info): """测试股票信息为空时是否抛出ValueError。""" mock_info.return_value = {} with pytest.raises(ValueError, match="No information available for INVALID"): get_ticker_data("INVALID") @patch('investor_agent.yfinance_utils.get_analyst_data') @patch('investor_agent.yfinance_utils.get_news') @patch('investor_agent.yfinance_utils.get_calendar') @patch('investor_agent.yfinance_utils.get_ticker_info') def test_essential_fields_filtering( self, mock_info, mock_calendar, mock_news, mock_analyst, mock_ticker_info ): """测试响应中只包含必要字段。""" # 向模拟数据添加一些非必要字段 extended_info = {** mock_ticker_info, 'nonEssentialField': 'should be filtered out'} mock_info.return_value = extended_info mock_calendar.return_value = None mock_news.return_value = None mock_analyst.return_value = None result = get_ticker_data("AAPL") # 验证必要字段存在 essential_fields = {'symbol', 'longName', 'currentPrice', 'marketCap', 'trailingPE'} for field in essential_fields: if field in mock_ticker_info: # 只检查模拟数据中存在的字段 assert field in result["info"], f"必要字段 {field} 缺失" # 验证非必要字段被过滤掉 assert 'nonEssentialField' not in result["info"] @patch('investor_agent.yfinance_utils.get_analyst_data') @patch('investor_agent.yfinance_utils.get_news') @patch('investor_agent.yfinance_utils.get_calendar') @patch('investor_agent.yfinance_utils.get_ticker_info') def test_limits_respected( self, mock_info, mock_calendar, mock_news, mock_analyst, mock_ticker_info, sample_recommendations_df ): """测试最大限制参数是否传递给底层函数。""" mock_info.return_value = mock_ticker_info mock_calendar.return_value = None mock_news.return_value = [] mock_analyst.return_value = sample_recommendations_df get_ticker_data("AAPL", max_news=10, max_recommendations=15, max_upgrades=20) # 验证限制参数被传递给函数 mock_news.assert_called_with("AAPL", limit=10) assert mock_analyst.call_count == 2 mock_analyst.assert_any_call("AAPL", "recommendations", 15) mock_analyst.assert_any_call("AAPL", "upgrades", 20)