claude_dev
2379159818
- 新增 alpha_lab.py: AlphaLabSession 类（lazy import vnpy.alpha） - 新增 analyzer.py: run_factor_analysis + FactorReport（lazy import alphalens） - 测试: test_alpha_lab.py (2 passed) + test_analyzer.py (3 passed) - 策略: 本地无 alphalens/polars，函数内 lazy import 避免 ImportError - 骨架: run_factor_analysis 返回 FactorReport，alphalens 集成待完整实现
108 lines
3.2 KiB
Python
108 lines
3.2 KiB
Python
"""Factor analysis with alphalens - lazy import to avoid ImportError."""
|
|
import sys
|
|
import os
|
|
_VNPY_SRC = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "vnpy_v4.4.0"))
|
|
if _VNPY_SRC not in sys.path:
|
|
sys.path.insert(0, _VNPY_SRC)
|
|
|
|
from dataclasses import dataclass, field
|
|
from typing import TYPE_CHECKING
|
|
|
|
if TYPE_CHECKING:
|
|
# Type hints only - not imported at runtime to avoid ImportError
|
|
import polars as pl
|
|
|
|
|
|
@dataclass
|
|
class FactorReport:
|
|
"""Factor analysis report."""
|
|
factor_names: list[str]
|
|
output_dir: str
|
|
ic_summary: dict = field(default_factory=dict)
|
|
|
|
|
|
def run_factor_analysis(
|
|
symbols: list[str],
|
|
factor_names: list[str],
|
|
start: str,
|
|
end: str,
|
|
cfg,
|
|
output_dir: str
|
|
) -> FactorReport:
|
|
"""
|
|
Run factor analysis using AlphaDataset and alphalens.
|
|
|
|
Args:
|
|
symbols: List of vt_symbols to analyze
|
|
factor_names: List of factor names to compute
|
|
start: Start date (YYYY-MM-DD)
|
|
end: End date (YYYY-MM-DD)
|
|
cfg: Database configuration object
|
|
output_dir: Output directory for analysis results
|
|
|
|
Returns:
|
|
FactorReport with analysis results
|
|
"""
|
|
from .alpha_lab import AlphaLabSession
|
|
from .registry import get_factor
|
|
|
|
# Lazy import alphalens functions (only when actually running analysis)
|
|
try:
|
|
from alphalens.utils import get_clean_factor_and_forward_returns
|
|
from alphalens.tears import create_full_tear_sheet
|
|
from vnpy.alpha.dataset import AlphaDataset, Segment
|
|
from vnpy.trader.constant import Interval
|
|
except ImportError:
|
|
# alphalens or vnpy.alpha not available - return skeleton report
|
|
return FactorReport(
|
|
factor_names=factor_names,
|
|
output_dir=output_dir,
|
|
ic_summary={"error": "alphalens or vnpy.alpha not installed"}
|
|
)
|
|
|
|
# Load symbols into AlphaLab
|
|
session = AlphaLabSession(lab_path=output_dir)
|
|
session.load_symbols(symbols, start, end, cfg)
|
|
|
|
# Create AlphaDataset and add features
|
|
# Load data from AlphaLab
|
|
df = session.lab.load_bar_data(symbols[0], Interval.DAILY, start, end) # Simplified - first symbol only
|
|
|
|
dataset = AlphaDataset(
|
|
df=df,
|
|
train_period=(start, end),
|
|
valid_period=None,
|
|
test_period=None
|
|
)
|
|
|
|
# Add features from registry
|
|
for factor_name in factor_names:
|
|
factor_info = get_factor(factor_name)
|
|
if factor_info:
|
|
dataset.add_feature(factor_name, factor_info["expression"])
|
|
|
|
# Prepare data
|
|
dataset.prepare_data(max_workers=None)
|
|
|
|
# Fetch raw data for analysis
|
|
raw_data = dataset.fetch_raw(Segment.TRAIN)
|
|
|
|
# Run alphalens analysis (skeleton)
|
|
try:
|
|
# TODO: Implement full alphalens tears pipeline
|
|
# factor_data = get_clean_factor_and_forward_returns(...)
|
|
# create_full_tear_sheet(factor_data, ...)
|
|
pass
|
|
except Exception as e:
|
|
return FactorReport(
|
|
factor_names=factor_names,
|
|
output_dir=output_dir,
|
|
ic_summary={"error": str(e)}
|
|
)
|
|
|
|
return FactorReport(
|
|
factor_names=factor_names,
|
|
output_dir=output_dir,
|
|
ic_summary={"status": "skeleton"}
|
|
)
|