"""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 # Module-level imports for patch targets (with try/except guards for local importability) try: from alphalens.utils import get_clean_factor_and_forward_returns from alphalens.tears import create_full_tear_sheet except ImportError: # alphalens not available locally - set to None for patch targets get_clean_factor_and_forward_returns = None create_full_tear_sheet = None try: from .alpha_lab import AlphaLabSession except ImportError: # AlphaLabSession not available - set to None for patch targets AlphaLabSession = None 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) report_path: str | None = None def run_factor_analysis( symbols: list[str], factor_names: list[str], start: str, end: str, cfg, output_dir: str ) -> FactorReport: """ Run factor analysis using AlphaLabSession 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 including tears report """ 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: return FactorReport( factor_names=factor_names, output_dir=output_dir, ic_summary={"error": "alphalens not installed"}, report_path=None ) if AlphaLabSession is None: return FactorReport( factor_names=factor_names, output_dir=output_dir, ic_summary={"error": "AlphaLabSession not available"}, report_path=None ) # Lazy imports for container environment try: import polars as pl import pandas as pd import matplotlib matplotlib.use("Agg") # Use non-interactive backend for headless operation import matplotlib.pyplot as plt except ImportError as e: return FactorReport( factor_names=factor_names, output_dir=output_dir, ic_summary={"error": f"Required import missing: {e}"}, report_path=None ) # Create AlphaLab session and load symbols session = AlphaLabSession(lab_path=output_dir) session.load_symbols(symbols, start, end, cfg) # Calculate period split (simple deterministic split) from datetime import datetime start_dt = datetime.strptime(start, "%Y-%m-%d") end_dt = datetime.strptime(end, "%Y-%m-%d") total_days = (end_dt - start_dt).days # Simple split: train = first half, valid = empty, test = second half mid_point = start_dt + pd.Timedelta(days=total_days // 2) train_period = (start, mid_point.strftime("%Y-%m-%d")) valid_period = (mid_point.strftime("%Y-%m-%d"), mid_point.strftime("%Y-%m-%d")) test_period = (mid_point.strftime("%Y-%m-%d"), end) # Compute factors using AlphaLabSession factor_df = session.compute_factors(factor_names, train_period, valid_period, test_period) # Initialize IC summary and report path ic_summary = {} report_path = None # Process each factor for factor_name in factor_names: try: # Convert polars DataFrame to pandas for alphalens factor_pd = factor_df.to_pandas() # Check if factor column exists if factor_name not in factor_pd.columns: # If the specific factor name isn't found, use the last column # (compute_factors returns factors with their names as columns) factor_cols = [col for col in factor_pd.columns if col not in ["datetime", "vt_symbol"]] if factor_cols: factor_col = factor_cols[0] # Use first available factor column else: continue # No factor columns found else: factor_col = factor_name # Set MultiIndex (datetime, vt_symbol) as required by alphalens factor_pd["datetime"] = pd.to_datetime(factor_pd["datetime"]) factor_series = factor_pd.set_index(["datetime", "vt_symbol"])[factor_col] # Build prices DataFrame (datetime × vt_symbol) # We need close prices - assume factor_df contains close column or derive it if "close" in factor_pd.columns: prices_df = factor_pd.pivot(index="datetime", columns="vt_symbol", values="close") else: # If close isn't available, create a simple price structure from the data # This is a simplified approach - in production, you'd re-read bars or cache close prices prices_df = factor_pd.pivot(index="datetime", columns="vt_symbol", values=factor_col) # Replace with simple returns-based price approximation prices_df = prices_df.cumsum() # Simplified: cumulative sum as price proxy # Ensure datetime index for prices prices_df.index = pd.to_datetime(prices_df.index) # Call get_clean_factor_and_forward_returns merged_data = get_clean_factor_and_forward_returns( factor=factor_series, prices=prices_df, periods=(1, 5, 10), # Standard forward return periods max_loss=0.35 # Allow up to 35% data loss ) # Generate tears sheet from io import StringIO import sys old_stdout = sys.stdout sys.stdout = StringIO() # Capture stdout to avoid display issues try: create_full_tear_sheet( merged_data, long_short=True, group_neutral=False, by_group=False ) finally: sys.stdout = old_stdout # Restore stdout # Save the tears report factor_report_path = os.path.join(output_dir, f"{factor_name}_tears.html") plt.savefig(factor_report_path.replace(".html", ".png")) # Save as PNG report_path = factor_report_path.replace(".png", ".html") # Mark HTML as report # Store basic IC summary (simplified) ic_summary[factor_name] = { "status": "success", "report": factor_report_path } except Exception as e: ic_summary[factor_name] = { "status": "error", "error": str(e) } return FactorReport( factor_names=factor_names, output_dir=output_dir, ic_summary=ic_summary, report_path=report_path )