金融数据工具测试+市场动态工具测试)
7.3.6 金融数据工具测试
测试功能对金融Agent至关重要，通过测试可以验证财务、市场动态、期权、个股等各类金融数据获取的准确性，保障市场情绪分析可靠，增强系统稳定性与健壮性，提升用户信任和满意度，还能助力符合金融合规与监管要求，为金融Agent精准、稳定地提供金融分析和投资建议筑牢基础。
在本项目中，文件test_financial_data.py包含了针对金融数据工具的测试用例，这些工具包括价格历史、财务报表、机构持股、收益历史和内部交易等。每个测试函数都验证了特定功能的预期行为，包括成功获取数据、处理无数据或空数据框的情况，以及正确调用被测试函数。
"""针对金融数据工具（价格历史、财务报表、机构持股等）的测试。""" class TestFinancialDataTools: """测试金融数据检索工具。""" @patch('investor_agent.yfinance_utils.get_price_history') def test_price_history_success(self, mock_price_history, sample_price_history): """测试成功获取价格历史。""" mock_price_history.return_value = sample_price_history result = get_price_history("AAPL", period="1mo") assert "data" in result assert "columns" in result assert "index" in result assert len(result["data"]) == 5 # 5天的数据 mock_price_history.assert_called_once_with("AAPL", "1mo", "1d") @patch('investor_agent.yfinance_utils.get_price_history') def test_price_history_long_period_uses_monthly_interval(self, mock_price_history, sample_price_history): """测试长时间周期使用月度间隔。""" mock_price_history.return_value = sample_price_history result = get_price_history("AAPL", period="5y") # 对于5年周期应使用月度间隔 mock_price_history.assert_called_once_with("AAPL", "5y", "1mo") @patch('investor_agent.yfinance_utils.get_price_history') def test_price_history_short_period_uses_daily_interval(self, mock_price_history, sample_price_history): """测试短时间周期使用日度间隔。""" mock_price_history.return_value = sample_price_history result = get_price_history("AAPL", period="1d") # 对于1天周期应使用日度间隔 mock_price_history.assert_called_once_with("AAPL", "1d", "1d") @patch('investor_agent.yfinance_utils.get_price_history') def test_price_history_no_data(self, mock_price_history): """测试当没有价格历史可用时的错误处理。""" mock_price_history.return_value = None with pytest.raises(ValueError, match="No historical data found for INVALID"): get_price_history("INVALID") @patch('investor_agent.yfinance_utils.get_price_history') def test_price_history_empty_dataframe(self, mock_price_history): """测试当返回空DataFrame时的错误处理。""" mock_price_history.return_value = pd.DataFrame() with pytest.raises(ValueError, match="No historical data found for INVALID"): get_price_history("INVALID") @patch('investor_agent.yfinance_utils.get_financial_statements') def test_financial_statements_success(self, mock_statements): """测试成功获取财务报表。""" statements_df = pd.DataFrame({ '2024-Q1': [1000000, 500000, 300000], '2023-Q4': [900000, 450000, 250000], '2023-Q3': [850000, 400000, 200000] }, index=['Revenue', 'Gross Profit', 'Net Income']) mock_statements.return_value = statements_df result = get_financial_statements("AAPL", statement_type="income", frequency="quarterly") assert "data" in result assert "columns" in result assert "index" in result assert len(result["columns"]) == 3 # 3个季度 mock_statements.assert_called_once_with("AAPL", "income", "quarterly") @patch('investor_agent.yfinance_utils.get_financial_statements') def test_financial_statements_max_periods_limit(self, mock_statements): """测试max_periods限制返回的列数。""" # 创建具有许多列的DataFrame many_periods = {f'2024-Q{i}': [1000000, 500000] for i in range(1, 15)} # 14个季度 statements_df = pd.DataFrame(many_periods, index=['Revenue', 'Gross Profit']) mock_statements.return_value = statements_df result = get_financial_statements("AAPL", max_periods=5) # 应限制为5个时期 assert len(result["columns"]) == 5 @patch('investor_agent.yfinance_utils.get_financial_statements') def test_financial_statements_no_data(self, mock_statements): """测试当没有财务数据可用时的错误处理。""" mock_statements.return_value = None with pytest.raises(ValueError, match="No income statement data found for INVALID"): get_financial_statements("INVALID", statement_type="income") @patch('investor_agent.yfinance_utils.get_financial_statements') def test_financial_statements_empty_dataframe(self, mock_statements): """测试当返回空DataFrame时的错误处理。""" mock_statements.return_value = pd.DataFrame() with pytest.raises(ValueError, match="No balance statement data found for INVALID"): get_financial_statements("INVALID", statement_type="balance") @patch('investor_agent.yfinance_utils.get_institutional_holders') def test_institutional_holders_success(self, mock_holders): """测试成功获取机构持股。""" inst_df = pd.DataFrame({ 'Holder': ['Vanguard', 'BlackRock', 'State Street'], 'Shares': [100000000, 90000000, 80000000], 'Date Reported': ['2024-01-01', '2024-01-01', '2024-01-01'] }) fund_df = pd.DataFrame({ 'Holder': ['Fund A', 'Fund B'], 'Shares': [5000000, 4000000], 'Date Reported': ['2024-01-01', '2024-01-01'] }) mock_holders.return_value = (inst_df, fund_df) result = get_institutional_holders("AAPL", top_n=20) assert "institutional_holders" in result assert "mutual_fund_holders" in result assert "data" in result["institutional_holders"] assert "data" in result["mutual_fund_holders"] mock_holders.assert_called_once_with("AAPL", 20) @patch('investor_agent.yfinance_utils.get_institutional_holders') def test_institutional_holders_partial_data(self, mock_holders): """测试仅机构持股数据可用时的情况。""" inst_df = pd.DataFrame({ 'Holder': ['Vanguard'], 'Shares': [100000000] }) mock_holders.return_value = (inst_df, None) # 没有共同基金数据 result = get_institutional_holders("AAPL") assert "institutional_holders" in result assert "mutual_fund_holders" not in result @patch('investor_agent.yfinance_utils.get_institutional_holders') def test_institutional_holders_no_data(self, mock_holders): """测试当没有机构持股数据可用时的错误处理。""" mock_holders.return_value = (None, None) with pytest.raises(ValueError, match="No institutional holder data found for INVALID"): get_institutional_holders("INVALID") @patch('investor_agent.yfinance_utils.get_institutional_holders') def test_institutional_holders_empty_dataframes(self, mock_holders): """测试当返回空DataFrame时的错误处理。""" mock_holders.return_value = (pd.DataFrame(), pd.DataFrame()) with pytest.raises(ValueError, match="No institutional holder data found for INVALID"): get_institutional_holders("INVALID") @patch('investor_agent.yfinance_utils.get_earnings_history') def test_earnings_history_success(self, mock_earnings): """测试成功获取收益历史。""" earnings_df7.3.7 市场动态工具测试
文件test_market_movers.py实现了针对get_market_movers工具的测试，通过模拟yahoo_finance_utils.get_market_movers_data的返回结果，测试了获取市场涨幅股、跌幅股、最活跃股，以及在盘前、盘后交易时段获取最活跃股等多种场景下该工具的行为，还包括默认参数使用、空响应和错误处理等情况的测试。
class TestGetMarketMovers: """测试get_market_movers工具。""" @pytest.fixture def sample_movers_data(self): """市场动态数据样本。""" return { 'stocks': [ { 'symbol': 'AAPL', 'name': 'Apple Inc.', 'price': 150.0, 'change': 5.0, 'percentChange': 3.45, 'volume': 50000000 }, { 'symbol': 'GOOGL', 'name': 'Alphabet Inc.', 'price': 2800.0, 'change': -25.0, 'percentChange': -0.88, 'volume': 25000000 } ] } @patch('investor_agent.yahoo_finance_utils.get_market_movers_data') async def test_get_gainers(self, mock_movers, sample_movers_data): """测试获取市场涨幅股。""" mock_movers.return_value = sample_movers_data result = await get_market_movers(category="gainers", count=25) assert result == sample_movers_data mock_movers.assert_called_once_with("gainers", 25, "regular") @patch('investor_agent.yahoo_finance_utils.get_market_movers_data') async def test_get_losers(self, mock_movers, sample_movers_data): """测试获取市场跌幅股。""" mock_movers.return_value = sample_movers_data result = await get_market_movers(category="losers", count=50) assert result == sample_movers_data mock_movers.assert_called_once_with("losers", 50, "regular") @patch('investor_agent.yahoo_finance_utils.get_market_movers_data') async def test_get_most_active(self, mock_movers, sample_movers_data): """测试获取最活跃股票。""" mock_movers.return_value = sample_movers_data result = await get_market_movers(category="most-active", count=30) assert result == sample_movers_data mock_movers.assert_called_once_with("most-active", 30, "regular") @patch('investor_agent.yahoo_finance_utils.get_market_movers_data') async def test_premarket_session(self, mock_movers, sample_movers_data): """测试获取盘前交易时段最活跃股票。""" mock_movers.return_value = sample_movers_data result = await get_market_movers( category="most-active", count=15, market_session="pre-market" ) assert result == sample_movers_data mock_movers.assert_called_once_with("most-active", 15, "pre-market") @patch('investor_agent.yahoo_finance_utils.get_market_movers_data') async def test_after_hours_session(self, mock_movers, sample_movers_data): """测试获取盘后交易时段最活跃股票。""" mock_movers.return_value = sample_movers_data result = await get_market_movers( category="most-active", count=20, market_session="after-hours" ) assert result == sample_movers_data mock_movers.assert_called_once_with("most-active", 20, "after-hours") @patch('investor_agent.yahoo_finance_utils.get_market_movers_data') async def test_default_parameters(self, mock_movers, sample_movers_data): """测试默认参数。""" mock_movers.return_value = sample_movers_data result = await get_market_movers() assert result == sample_movers_data mock_movers.assert_called_once_with("most-active", 25, "regular") @patch('investor_agent.yahoo_finance_utils.get_market_movers_data') async def test_empty_response(self, mock_movers): """测试处理空响应。""" mock_movers.return_value = {'stocks': []} result = await get_market_movers(category="gainers") assert result == {'stocks': []} assert len(result['stocks']) == 0 @patch('investor_agent.yahoo_finance_utils.get_market_movers_data') async def test_error_handling(self, mock_movers): """测试底层函数失败时的错误处理。""" mock_movers.side_effect = Exception("API Error") with pytest.raises(Exception, match="API Error"): await get_market_movers(category="gainers")
