diff --git a/sanguo_factor/analyzer.py b/sanguo_factor/analyzer.py index 48d4144..4601cb8 100644 --- a/sanguo_factor/analyzer.py +++ b/sanguo_factor/analyzer.py @@ -1,6 +1,7 @@ """Factor analysis with alphalens - lazy import to avoid ImportError.""" import sys import os +import warnings _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) @@ -34,7 +35,7 @@ class FactorReport: factor_names: list[str] output_dir: str ic_summary: dict = field(default_factory=dict) - report_path: str | None = None + report_paths: dict = field(default_factory=dict) def run_factor_analysis( @@ -67,7 +68,7 @@ def run_factor_analysis( factor_names=factor_names, output_dir=output_dir, ic_summary={"error": "alphalens not installed"}, - report_path=None + report_paths={} ) if AlphaLabSession is None: @@ -75,7 +76,7 @@ def run_factor_analysis( factor_names=factor_names, output_dir=output_dir, ic_summary={"error": "AlphaLabSession not available"}, - report_path=None + report_paths={} ) # Lazy imports for container environment @@ -90,7 +91,7 @@ def run_factor_analysis( factor_names=factor_names, output_dir=output_dir, ic_summary={"error": f"Required import missing: {e}"}, - report_path=None + report_paths={} ) # Create AlphaLab session and load symbols @@ -112,9 +113,9 @@ def run_factor_analysis( # Compute factors using AlphaLabSession factor_df = session.compute_factors(factor_names, train_period, valid_period, test_period) - # Initialize IC summary and report path + # Initialize IC summary and report paths ic_summary = {} - report_path = None + report_paths = {} # Process each factor for factor_name in factor_names: @@ -140,11 +141,14 @@ def run_factor_analysis( # Build prices DataFrame (datetime × vt_symbol) # We need close prices - assume factor_df contains close column or derive it + use_cumsum_fallback = False 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 + warnings.warn(f"close 列缺失,因子 {factor_name} 使用 cumsum 兜底价格,tears 结果不可靠", UserWarning) + use_cumsum_fallback = True 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 @@ -179,23 +183,25 @@ def run_factor_analysis( # 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 + report_paths[factor_name] = factor_report_path.replace(".png", ".html") # Mark HTML as report # Store basic IC summary (simplified) + status = "warning_unreliable_prices" if use_cumsum_fallback else "success" ic_summary[factor_name] = { - "status": "success", + "status": status, "report": factor_report_path } except Exception as e: + err_type = type(e).__name__ ic_summary[factor_name] = { "status": "error", - "error": str(e) + "error": f"{err_type}: {e}" } return FactorReport( factor_names=factor_names, output_dir=output_dir, ic_summary=ic_summary, - report_path=report_path + report_paths=report_paths )