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 集成待完整实现
41 lines
1.3 KiB
Python
41 lines
1.3 KiB
Python
"""AlphaLab session management - lazy import vnpy.alpha to avoid ImportError when alphalens missing."""
|
|
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)
|
|
|
|
|
|
class AlphaLabSession:
|
|
"""Session manager for vnpy.alpha AlphaLab operations."""
|
|
|
|
def __init__(self, lab_path: str):
|
|
"""
|
|
Initialize AlphaLab session.
|
|
|
|
Args:
|
|
lab_path: Path to AlphaLab directory
|
|
"""
|
|
from vnpy.alpha.lab import AlphaLab
|
|
|
|
self.lab_path = lab_path
|
|
self.lab = AlphaLab(lab_path)
|
|
|
|
def load_symbols(self, symbols: list[str], start: str, end: str, cfg) -> None:
|
|
"""
|
|
Load symbol data from database and save to AlphaLab.
|
|
|
|
Args:
|
|
symbols: List of vt_symbols to load
|
|
start: Start date (YYYY-MM-DD)
|
|
end: End date (YYYY-MM-DD)
|
|
cfg: Database configuration object
|
|
"""
|
|
from sanguo_data.datareader import read_db_daily
|
|
from .data_adapter import save_alpha_lab_data
|
|
|
|
for symbol in symbols:
|
|
bars = read_db_daily(symbol, start, end, cfg)
|
|
if bars:
|
|
save_alpha_lab_data(bars, self.lab_path)
|