From fa7237b996451687b031d70a7a5f70d705c8aa35 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Mon, 6 Jul 2026 21:32:27 +0800 Subject: [PATCH] =?UTF-8?q?feat(polish):=20=E5=9B=A0=E5=AD=90=E6=8A=A5?= =?UTF-8?q?=E5=91=8A=20IC=20=E6=8F=90=E5=8F=96=20+=20periods=20=E6=8F=90?= =?UTF-8?q?=E5=8F=82=20+=20=E7=BC=93=E5=AD=98=E4=B8=8A=E9=99=90=20+=20?= =?UTF-8?q?=E4=BB=A3=E7=A0=81=E6=95=B4=E6=B4=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - analyzer.py: 提取 IC 值到 ic_summary (mean/std/icir/t_stat)，periods 提参 (默认 1,5,10) - alpha_lab.py: _loaded_bars 缓存 LRU 上限 (_MAX_CACHED_SYMBOLS=50) - runner.py: 统一阶段文案 (参数优化中/因子分析中)，worker 类型标注，_wait_future 文档 - pool.py: submit_work 添加 task_id debug 日志 Co-Authored-By: Claude --- sanguo_factor/alpha_lab.py | 22 +++++- sanguo_factor/analyzer.py | 52 +++++++++++-- sanguo_orchestrator/pool.py | 4 + sanguo_orchestrator/runner.py | 19 +++-- tests/factor/test_alpha_lab.py | 92 +++++++++++++++++++++- tests/factor/test_analyzer.py | 135 ++++++++++++++++++++++++++++++++- 6 files changed, 307 insertions(+), 17 deletions(-) diff --git a/sanguo_factor/alpha_lab.py b/sanguo_factor/alpha_lab.py index 4e7abdb..b7ca3a6 100644 --- a/sanguo_factor/alpha_lab.py +++ b/sanguo_factor/alpha_lab.py @@ -5,6 +5,12 @@ _VNPY_SRC = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "vnpy_ if _VNPY_SRC not in sys.path: sys.path.insert(0, _VNPY_SRC) +from collections import OrderedDict + +# Maximum number of symbols to cache in _loaded_bars to prevent unbounded growth +# Single-user internal tool - 50 symbols is a generous ceiling +_MAX_CACHED_SYMBOLS = 50 + class AlphaLabSession: """Session manager for vnpy.alpha AlphaLab operations.""" @@ -21,7 +27,8 @@ class AlphaLabSession: self.lab_path = lab_path self.lab = AlphaLab(lab_path) self._loaded_symbols: list[str] = [] - self._loaded_bars: dict[str, list] = {} + # Use OrderedDict to maintain insertion order for LRU eviction + self._loaded_bars: OrderedDict[str, list] = OrderedDict() def load_symbols(self, symbols: list[str], start: str, end: str, cfg) -> None: """ @@ -40,11 +47,22 @@ class AlphaLabSession: bars = read_db_daily(symbol, start, end, cfg) if bars: save_alpha_lab_data(bars, self.lab_path) - # Cache bars for compute_factors + # Cache bars for compute_factors with LRU eviction if symbol not in self._loaded_symbols: self._loaded_symbols.append(symbol) + + # Add or update symbol in cache (moves to end if exists) + if symbol in self._loaded_bars: + del self._loaded_bars[symbol] # Remove to re-insert for LRU self._loaded_bars[symbol] = bars + # Enforce cache cap - evict oldest symbol if exceeded + while len(self._loaded_bars) > _MAX_CACHED_SYMBOLS: + oldest_symbol = next(iter(self._loaded_bars)) + del self._loaded_bars[oldest_symbol] + if oldest_symbol in self._loaded_symbols: + self._loaded_symbols.remove(oldest_symbol) + def compute_factors(self, factor_names: list[str], train_period: tuple, valid_period: tuple, test_period: tuple): """ Compute factors using cached bars and vnpy.alpha AlphaDataset. diff --git a/sanguo_factor/analyzer.py b/sanguo_factor/analyzer.py index 4601cb8..acc44ec 100644 --- a/sanguo_factor/analyzer.py +++ b/sanguo_factor/analyzer.py @@ -13,10 +13,12 @@ from typing import TYPE_CHECKING try: from alphalens.utils import get_clean_factor_and_forward_returns from alphalens.tears import create_full_tear_sheet + from alphalens.performance import factor_information_coefficient except ImportError: # alphalens not available locally - set to None for patch targets get_clean_factor_and_forward_returns = None create_full_tear_sheet = None + factor_information_coefficient = None try: from .alpha_lab import AlphaLabSession @@ -44,7 +46,8 @@ def run_factor_analysis( start: str, end: str, cfg, - output_dir: str + output_dir: str, + periods: tuple = (1, 5, 10) ) -> FactorReport: """ Run factor analysis using AlphaLabSession and alphalens. @@ -56,14 +59,15 @@ def run_factor_analysis( end: End date (YYYY-MM-DD) cfg: Database configuration object output_dir: Output directory for analysis results + periods: Forward return periods for IC analysis (default: 1, 5, 10 days) Returns: - FactorReport with analysis results including tears report + FactorReport with analysis results including tears report and IC values """ from .registry import get_factor # Check if alphalens is available - if get_clean_factor_and_forward_returns is None or create_full_tear_sheet is None: + if get_clean_factor_and_forward_returns is None or create_full_tear_sheet is None or factor_information_coefficient is None: return FactorReport( factor_names=factor_names, output_dir=output_dir, @@ -160,10 +164,47 @@ def run_factor_analysis( merged_data = get_clean_factor_and_forward_returns( factor=factor_series, prices=prices_df, - periods=(1, 5, 10), # Standard forward return periods + periods=periods, # Use configurable periods max_loss=0.35 # Allow up to 35% data loss ) + # Extract IC values using factor_information_coefficient + ic_data = {} + try: + ic_df = factor_information_coefficient(merged_data) + + # Compute IC statistics for each period + for period_col in ic_df.columns: + period_name = f"{period_col}D" if period_col.isdigit() else period_col + + # Extract IC values for this period (drop NaN values) + period_ic_values = ic_df[period_col].dropna() + + if len(period_ic_values) > 0: + ic_mean = float(period_ic_values.mean()) + ic_std = float(period_ic_values.std()) + icir = ic_mean / ic_std if ic_std > 0 else 0.0 + + # Compute t-statistic if we have enough samples + n = len(period_ic_values) + t_stat = ic_mean / (ic_std / (n ** 0.5)) if ic_std > 0 and n > 1 else 0.0 + + ic_data[period_name] = { + "mean": ic_mean, + "std": ic_std, + "icir": icir, + "t_stat": t_stat, + "count": n + } + else: + ic_data[period_name] = { + "error": "No valid IC values for period" + } + + except Exception as ic_error: + # Capture IC extraction error but continue with tears report + ic_data = {"error": f"IC extraction failed: {type(ic_error).__name__}: {ic_error}"} + # Generate tears sheet from io import StringIO import sys @@ -189,7 +230,8 @@ def run_factor_analysis( status = "warning_unreliable_prices" if use_cumsum_fallback else "success" ic_summary[factor_name] = { "status": status, - "report": factor_report_path + "report": factor_report_path, + "ic": ic_data # Add IC statistics } except Exception as e: diff --git a/sanguo_orchestrator/pool.py b/sanguo_orchestrator/pool.py index 09adef7..e23d283 100644 --- a/sanguo_orchestrator/pool.py +++ b/sanguo_orchestrator/pool.py @@ -2,10 +2,13 @@ Task pool for managing multiple tasks in memory Provides task storage and status tracking (not actual multiprocessing) """ +import logging from concurrent.futures import ProcessPoolExecutor, Future from multiprocessing import get_context from .task import Task, TaskState +logger = logging.getLogger(__name__) + class TaskPool: """Manages task storage and status tracking""" @@ -26,6 +29,7 @@ class TaskPool: def submit_work(self, task_id: str, func, *args) -> Future: """Submit work to the process pool executor""" + logger.debug("submit_work task_id=%s", task_id) return self.executor.submit(func, *args) def update_stage(self, task_id: str, stage: str): diff --git a/sanguo_orchestrator/runner.py b/sanguo_orchestrator/runner.py index e0ad7ad..003724b 100644 --- a/sanguo_orchestrator/runner.py +++ b/sanguo_orchestrator/runner.py @@ -23,7 +23,7 @@ class Orchestrator: """Set callback for stage updates (async callable)""" self._on_stage = cb - async def _notify_stage(self, task_id: str, stage: str): + async def _notify_stage(self, task_id: str, stage: str) -> None: """Update task stage and fire callback if set""" self.pool.update_stage(task_id, stage) if self._on_stage: @@ -79,7 +79,7 @@ class Orchestrator: task = self.pool.get_task(task_id) task.start() - await self._notify_stage(task_id, "优化中") + await self._notify_stage(task_id, "参数优化中") asyncio.ensure_future(self._wait_future(task_id, fut)) return task_id @@ -110,8 +110,11 @@ class Orchestrator: asyncio.ensure_future(self._wait_future(task_id, fut)) return task_id - async def _wait_future(self, task_id: str, fut: Future): - """Wait for Future to complete and handle result/exception""" + async def _wait_future(self, task_id: str, fut: Future) -> None: + """Wait for Future to complete and handle result/exception + + Bridges concurrent.futures.Future (from ProcessPoolExecutor) to asyncio coroutine. + """ try: result = await asyncio.wrap_future(fut) await self._on_done(task_id, result) @@ -121,7 +124,7 @@ class Orchestrator: task.fail(f"{type(e).__name__}: {e}") await self._notify_stage(task_id, "失败") - async def _on_done(self, task_id: str, result): + async def _on_done(self, task_id: str, result) -> None: """Handle task completion (with None-guard for unknown tasks)""" task = self.pool.get_task(task_id) if task is None: @@ -147,19 +150,19 @@ class Orchestrator: # Module-level worker functions (must be top-level for ProcessPoolExecutor pickle) -def _cta_worker(strategy_class, symbol: str, params: dict, start: str, end: str, cfg, db_path: str): +def _cta_worker(strategy_class, symbol: str, params: dict, start: str, end: str, cfg, db_path: str) -> any: """Worker for CTA backtest (lazy import, spawn-friendly)""" from sanguo_backtest.cta_engine import run_cta_backtest return run_cta_backtest(strategy_class, symbol, params, start, end, cfg, db_path) -def _opt_worker(strategy_class, symbol: str, grid: dict, start: str, end: str, cfg, db_path: str): +def _opt_worker(strategy_class, symbol: str, grid: dict, start: str, end: str, cfg, db_path: str) -> any: """Worker for CTA optimization (lazy import, spawn-friendly)""" from sanguo_backtest.cta_optimizer import run_cta_optimization return run_cta_optimization(strategy_class, symbol, grid, start, end, cfg, db_path) -def _factor_worker(symbols: list, factor_names: list, start: str, end: str, cfg, output_dir: str): +def _factor_worker(symbols: list, factor_names: list, start: str, end: str, cfg, output_dir: str) -> any: """Worker for factor analysis (lazy import, spawn-friendly)""" from sanguo_factor.analyzer import run_factor_analysis return run_factor_analysis(symbols, factor_names, start, end, cfg, output_dir) diff --git a/tests/factor/test_alpha_lab.py b/tests/factor/test_alpha_lab.py index 257a15a..cd6eb26 100644 --- a/tests/factor/test_alpha_lab.py +++ b/tests/factor/test_alpha_lab.py @@ -38,7 +38,13 @@ def test_load_symbols_calls_read_db_daily(): def test_compute_factors_calls_prepare_and_fetch(tmp_path): - """Test compute_factors calls AlphaDataset methods correctly.""" + """Test compute_factors calls AlphaDataset methods correctly. + + Requires polars - runs in container, skips locally. + """ + import pytest + pytest.importorskip("polars") + from pathlib import Path from datetime import datetime from unittest.mock import MagicMock @@ -124,3 +130,87 @@ def test_compute_factors_calls_prepare_and_fetch(tmp_path): # Verify return value is a DataFrame assert isinstance(df, pl.DataFrame) + + +def test_loaded_bars_cache_eviction(tmp_path): + """Test that _loaded_bars cache evicts oldest entries when exceeding cap.""" + from datetime import datetime + from unittest.mock import MagicMock + from sanguo_factor.alpha_lab import AlphaLabSession, _MAX_CACHED_SYMBOLS + from collections import OrderedDict + + # Create mock bar data + def create_mock_bar(symbol: str): + mock_bar = MagicMock() + mock_bar.vt_symbol = symbol + mock_bar.datetime = datetime(2024, 1, 1) + mock_bar.open_price = 1.0 + mock_bar.high_price = 1.0 + mock_bar.low_price = 1.0 + mock_bar.close_price = 1.0 + mock_bar.volume = 1 + mock_bar.turnover = 0 + mock_bar.open_interest = 0 + return [mock_bar] + + # Create a mock session and manually test cache eviction logic + lab_path = str(tmp_path / "alpha_lab") + + # Manually initialize the session to avoid vnpy.alpha import + import sanguo_factor.alpha_lab as alpha_lab_module + original_init = AlphaLabSession.__init__ + + def mock_init(self, lab_path): + self.lab_path = lab_path + self._loaded_symbols = [] + self._loaded_bars = OrderedDict() + + # Temporarily replace __init__ and test the cache logic + AlphaLabSession.__init__ = mock_init + + try: + session = AlphaLabSession(lab_path=lab_path) + + # Directly simulate the cache eviction logic from load_symbols + symbols_to_load = [f"60000{i}.SSE" for i in range(_MAX_CACHED_SYMBOLS + 10)] + + for i, symbol in enumerate(symbols_to_load): + # Simulate adding symbol to cache (from load_symbols logic) + bars = create_mock_bar(symbol) + + # Add symbol to _loaded_symbols + if symbol not in session._loaded_symbols: + session._loaded_symbols.append(symbol) + + # Add or update symbol in cache (LRU logic) + if symbol in session._loaded_bars: + del session._loaded_bars[symbol] + session._loaded_bars[symbol] = bars + + # Enforce cache cap - evict oldest symbol if exceeded + while len(session._loaded_bars) > _MAX_CACHED_SYMBOLS: + oldest_symbol = next(iter(session._loaded_bars)) + del session._loaded_bars[oldest_symbol] + if oldest_symbol in session._loaded_symbols: + session._loaded_symbols.remove(oldest_symbol) + + # Check that cache size never exceeds cap + assert len(session._loaded_bars) <= _MAX_CACHED_SYMBOLS, \ + f"Cache exceeded cap at iteration {i}: {len(session._loaded_bars)} > {_MAX_CACHED_SYMBOLS}" + + # Final check: cache should be exactly at cap + assert len(session._loaded_bars) == _MAX_CACHED_SYMBOLS + + # Verify that the oldest symbols were evicted (first loaded symbols should be gone) + oldest_symbols = symbols_to_load[:10] # First 10 symbols should be evicted + for symbol in oldest_symbols: + assert symbol not in session._loaded_bars, f"Oldest symbol {symbol} should have been evicted" + + # Verify that the newest symbols are still in cache + newest_symbols = symbols_to_load[-10:] # Last 10 symbols should be present + for symbol in newest_symbols: + assert symbol in session._loaded_bars, f"Newest symbol {symbol} should be in cache" + + finally: + # Restore original __init__ + AlphaLabSession.__init__ = original_init diff --git a/tests/factor/test_analyzer.py b/tests/factor/test_analyzer.py index 5912a15..f19ebbd 100644 --- a/tests/factor/test_analyzer.py +++ b/tests/factor/test_analyzer.py @@ -100,7 +100,13 @@ def test_run_factor_analysis_adds_features(): def test_run_factor_analysis_calls_tears(tmp_path): - """Test that run_factor_analysis calls alphalens tears pipeline.""" + """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 @@ -121,3 +127,130 @@ def test_run_factor_analysis_calls_tears(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"]