"""Test analyzer module - Factor analysis with alphalens.""" import sys import os _VNPY_SRC = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "vnpy_v4.4.0")) sys.path.insert(0, _VNPY_SRC) from unittest.mock import Mock, patch, MagicMock import tempfile def test_run_factor_analysis_returns_report(): """Test that run_factor_analysis returns FactorReport with correct structure.""" from pathlib import Path from sanguo_factor.analyzer import run_factor_analysis, FactorReport with tempfile.TemporaryDirectory() as tmpdir: output_dir = str(Path(tmpdir) / "factor_analysis") # Patch imports to avoid ImportError when alphalens missing with patch("sanguo_factor.analyzer.AlphaLabSession") as MockSession, \ patch("sanguo_factor.analyzer.get_clean_factor_and_forward_returns"), \ patch("sanguo_factor.analyzer.create_full_tear_sheet"): # Mock the AlphaLabSession mock_session_instance = Mock() MockSession.return_value = mock_session_instance mock_session_instance.load_symbols = Mock() mock_session_instance.compute_factors = Mock(return_value=Mock(to_pandas=Mock(return_value=Mock( set_index=Mock(return_value=Mock(__getitem__=Mock(return_value=Mock())))), pivot=Mock(return_value=Mock()) ))) # Mock get_factor to avoid registry call with patch("sanguo_factor.registry.get_factor", return_value={"expression": "ts_mean(close, 5)"}): # Run the analysis result = run_factor_analysis( symbols=["600000"], factor_names=["ma5"], start="2024-01-01", end="2024-06-30", cfg=Mock(), output_dir=output_dir ) # Verify the result is a FactorReport with expected structure assert isinstance(result, FactorReport) assert result.factor_names == ["ma5"] assert result.output_dir == output_dir assert isinstance(result.ic_summary, dict) def test_run_factor_analysis_calls_load_symbols(): """Test that run_factor_analysis handles missing alphalens gracefully.""" from pathlib import Path from sanguo_factor.analyzer import run_factor_analysis with tempfile.TemporaryDirectory() as tmpdir: output_dir = str(Path(tmpdir) / "factor_analysis") # Patch module-level variables to simulate missing alphalens with patch("sanguo_factor.analyzer.get_clean_factor_and_forward_returns", None), \ patch("sanguo_factor.analyzer.create_full_tear_sheet", None): # alphalens missing - should return skeleton report with error result = run_factor_analysis( symbols=["600000", "000001"], factor_names=["ma5"], start="2024-01-01", end="2024-06-30", cfg=Mock(), output_dir=output_dir ) # Verify skeleton report returned assert "error" in result.ic_summary def test_run_factor_analysis_adds_features(): """Test that run_factor_analysis returns correct structure when alphalens missing.""" from pathlib import Path from sanguo_factor.analyzer import run_factor_analysis with tempfile.TemporaryDirectory() as tmpdir: output_dir = str(Path(tmpdir) / "factor_analysis") # Patch module-level variables to simulate missing alphalens with patch("sanguo_factor.analyzer.get_clean_factor_and_forward_returns", None), \ patch("sanguo_factor.analyzer.create_full_tear_sheet", None): # alphalens missing - verify structure result = run_factor_analysis( symbols=["600000"], factor_names=["ma5"], start="2024-01-01", end="2024-06-30", cfg=Mock(), output_dir=output_dir ) # Verify factor_names preserved even when alphalens missing assert result.factor_names == ["ma5"] assert result.output_dir == output_dir def test_run_factor_analysis_calls_tears(tmp_path): """Test that run_factor_analysis calls alphalens tears pipeline.""" from pathlib import Path from sanguo_factor.analyzer import run_factor_analysis with patch("sanguo_factor.analyzer.AlphaLabSession") as MS, \ patch("sanguo_factor.analyzer.get_clean_factor_and_forward_returns") as MC, \ patch("sanguo_factor.analyzer.create_full_tear_sheet") as MT: import polars as pl MS.return_value.compute_factors.return_value = pl.DataFrame({ "datetime": [], "vt_symbol": [], "ma5": [] }) MC.return_value = MagicMock() report = run_factor_analysis( ["600000"], ["ma5"], "2024-01-01", "2024-06-30", cfg=MagicMock(), output_dir=str(tmp_path) ) assert report.factor_names == ["ma5"] MC.assert_called_once() MT.assert_called_once()