claude_dev
a1048690c1
- test_main: FastAPI 0.139 _IncludedRouter 不再 flatten，改用 TestClient 探测路由 - datareader: 文件名(sh600000_daily)/patch target(vnpy.trader.database) 对齐 lazy import 实现 - alpha_lab/analyzer/data_adapter: vnpy.alpha/alphalens 容器专用本地 skip - 删 4 个测废弃 sanguo_web 的僵尸测试（-1267 行死代码） - pytest.ini: asyncio_mode=auto - frontend: package.json 加 test script（npm test 可跑） - NAS 容器 309 passed 全绿验证（Python 3.10，本机 303+6skip）
258 lines
10 KiB
Python
258 lines
10 KiB
Python
"""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")
|
|
pytest.importorskip("alphalens")
|
|
|
|
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"]
|