sanguo_vnpy_v2/tests/factor/test_analyzer.py

"""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.

    Requires polars - runs in container, skips locally.
    """
    import pytest
    pytest.importorskip("polars")

    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()


def test_run_factor_analysis_extracts_ic_values(tmp_path):
    """Test that run_factor_analysis extracts IC values from alphalens.

    Requires polars/alphalens - runs in container, skips locally.
    """
    import pytest
    pytest.importorskip("polars")
    pytest.importorskip("alphalens")

    import pandas as pd
    from datetime import datetime
    from sanguo_factor.analyzer import run_factor_analysis

    # Create mock factor_data with MultiIndex (datetime, asset) and IC columns
    dates = pd.date_range("2024-01-01", periods=10, freq="D")
    assets = ["AAPL", "GOOGL"]
    index = pd.MultiIndex.from_product([dates, assets], names=["datetime", "asset"])

    # Mock factor_data with forward returns
    mock_factor_data = pd.DataFrame(index=index)
    mock_factor_data["factor"] = [0.5] * 20  # Factor values
    mock_factor_data["1D"] = [0.01] * 20      # 1-day forward returns
    mock_factor_data["5D"] = [0.05] * 20      # 5-day forward returns
    mock_factor_data["10D"] = [0.10] * 20    # 10-day forward returns

    # Mock IC DataFrame returned by factor_information_coefficient
    mock_ic_df = pd.DataFrame({
        "1D": [0.05, 0.03, 0.07, 0.04, 0.06, 0.05, 0.04, 0.06, 0.05, 0.04],
        "5D": [0.08, 0.06, 0.09, 0.07, 0.08, 0.07, 0.08, 0.06, 0.07, 0.08],
        "10D": [0.12, 0.10, 0.13, 0.11, 0.12, 0.11, 0.12, 0.10, 0.11, 0.12]
    }, index=dates)

    # Mock polars DataFrame
    mock_pl_df = MagicMock()
    mock_pl_df.to_pandas.return_value = pd.DataFrame({
        "datetime": [d.isoformat() for d in dates for _ in assets],
        "vt_symbol": assets * len(dates),
        "ma5": [0.5] * 20,
        "close": [100.0] * 20
    })

    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, \
         patch("sanguo_factor.analyzer.factor_information_coefficient") as MIC:

        # Mock compute_factors to return polars DataFrame
        MS.return_value.compute_factors.return_value = mock_pl_df

        # Mock get_clean_factor_and_forward_returns to return our factor_data
        MC.return_value = mock_factor_data

        # Mock IC function to return our IC DataFrame
        MIC.return_value = mock_ic_df

        report = run_factor_analysis(
            ["AAPL"], ["ma5"], "2024-01-01", "2024-01-10",
            cfg=MagicMock(), output_dir=str(tmp_path)
        )

        # Verify IC values were extracted
        assert "ma5" in report.ic_summary
        assert "ic" in report.ic_summary["ma5"]

        # Check IC structure contains expected periods
        ic_data = report.ic_summary["ma5"]["ic"]
        assert "1D" in ic_data
        assert "5D" in ic_data
        assert "10D" in ic_data

        # Verify IC statistics are computed
        assert "mean" in ic_data["1D"]
        assert "icir" in ic_data["1D"]
        assert "std" in ic_data["1D"]

        # Verify approximate values (mean should be around 0.05 for 1D)
        assert abs(ic_data["1D"]["mean"] - 0.05) < 0.01  # Allow small rounding errors


def test_run_factor_analysis_ic_extraction_fails_gracefully(tmp_path):
    """Test that IC extraction failures don't crash the pipeline.

    Requires polars/alphalens - runs in container, skips locally.
    """
    import pytest
    pytest.importorskip("polars")
    pytest.importorskip("alphalens")

    from sanguo_factor.analyzer import run_factor_analysis
    import pandas as pd

    # Mock polars DataFrame
    mock_pl_df = MagicMock()
    mock_pl_df.to_pandas.return_value = pd.DataFrame({
        "datetime": ["2024-01-01"],
        "vt_symbol": ["AAPL"],
        "ma5": [0.5],
        "close": [100.0]
    })

    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, \
         patch("sanguo_factor.analyzer.factor_information_coefficient") as MIC:

        MS.return_value.compute_factors.return_value = mock_pl_df

        # Mock get_clean_factor_and_forward_returns to return valid data
        mock_factor_data = MagicMock()
        MC.return_value = mock_factor_data

        # Mock IC function to raise an exception
        MIC.side_effect = Exception("IC calculation failed")

        report = run_factor_analysis(
            ["AAPL"], ["ma5"], "2024-01-01", "2024-01-10",
            cfg=MagicMock(), output_dir=str(tmp_path)
        )

        # Verify IC error is captured but status/report still exist
        assert "ma5" in report.ic_summary
        assert "ic" in report.ic_summary["ma5"]
        assert "error" in report.ic_summary["ma5"]["ic"]
        # Status and report should still be present
        assert "status" in report.ic_summary["ma5"]