"""Factor analysis with alphalens - lazy import to avoid ImportError."""
import sys
import os
import warnings
import traceback
_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
    from alphalens.performance import factor_information_coefficient
    # alphalens-reloaded × pandas2 兼容：demean_forward_returns 用 groupby.transform(lambda)
    # 在 pandas2 走 _transform_general → concat 空(No objects to concatenate)，tears 崩。
    # 改 transform("mean") 走 _transform_fast（广播组均值，不 concat）再相减，等价且兼容。
    # 仅 monkey-patch，不改 alphalens 源码（同 empyrical np.NINF 适配思路）。
    import alphalens.utils as _al_utils
    import pandas as _al_pd
    def _demean_forward_returns_pandas2(factor_data, grouper=None):
        factor_data = factor_data.copy()
        if not grouper:
            grouper = factor_data.index.get_level_values("date")
        cols = _al_utils.get_forward_returns_columns(factor_data.columns)
        means = factor_data.groupby(grouper)[cols].transform("mean")
        factor_data[cols] = factor_data[cols] - means
        return factor_data
    _al_utils.demean_forward_returns = _demean_forward_returns_pandas2
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
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_paths: dict = field(default_factory=dict)
    symbols: list[str] = field(default_factory=list)
    start: str = ""
    end: str = ""


def run_factor_analysis(
    symbols: list[str],
    factor_names: list[str],
    start: str,
    end: str,
    cfg,
    output_dir: str,
    periods: tuple = (1, 5, 10)
) -> 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
        periods: Forward return periods for IC analysis (default: 1, 5, 10 days)

    Returns:
        FactorReport with analysis results including tears report and IC values
    """
    from .registry import get_factor

    # symbols 兼容:前端/vnpy 可能传 vt_symbol("600000.SSE"),read_db_daily 查 DB 的 key
    # 是裸代码(同 cta_engine),带后缀查不到 → bars 空 → factor/prices 全空 → alphalens 崩。
    # 统一归一成裸代码(DB key)。
    symbols = [str(s).split(".")[0] for s in symbols]

    # API path passes cfg=None → load default data_platform.yaml (so read_db_daily
    # and AlphaLabSession can find the A-share DB).
    if cfg is None:
        from sanguo_data.config import load_config, find_config_path
        cfg = load_config(find_config_path())

    # Check if alphalens is available
    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,
            ic_summary={"error": "alphalens not installed"},
            report_paths={}
        )

    if AlphaLabSession is None:
        return FactorReport(
            factor_names=factor_names,
            output_dir=output_dir,
            ic_summary={"error": "AlphaLabSession not available"},
            report_paths={}
        )

    # 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_paths={}
        )

    # 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)

    # Load close prices separately for tears computation
    # (factor_df only contains factor columns, not OHLCV)
    from sanguo_data.datareader import read_db_daily
    from datetime import datetime
    from zoneinfo import ZoneInfo

    # Convert string dates to datetime for database query
    _SH = ZoneInfo("Asia/Shanghai")
    start_dt = datetime.strptime(start, "%Y-%m-%d").replace(tzinfo=_SH)
    end_dt = datetime.strptime(end, "%Y-%m-%d").replace(tzinfo=_SH)

    # Load bars for close prices
    all_bars = []
    for symbol in symbols:
        try:
            bars = read_db_daily(symbol, start_dt.strftime("%Y-%m-%d"), end_dt.strftime("%Y-%m-%d"), cfg)
            all_bars.extend(bars)
        except Exception as e:
            warnings.warn(f"Failed to load bars for {symbol}: {e}")
            continue

    # Create close price DataFrame
    if all_bars:
        close_df = pl.DataFrame({
            "datetime": [b.datetime for b in all_bars],
            "vt_symbol": [b.vt_symbol for b in all_bars],
            "close": [b.close_price for b in all_bars]
        })
    else:
        warnings.warn("No bars loaded for close prices - tears computation will fail")
        close_df = pl.DataFrame(schema={"datetime": pl.Datetime, "vt_symbol": pl.Utf8, "close": pl.Float64})

    # Initialize IC summary and report paths
    ic_summary = {}
    report_paths = {}

    # Process each factor
    for factor_name in factor_names:
        try:
            # Convert polars DataFrame to pandas for alphalens
            factor_pd = factor_df.to_pandas()
            close_pd = close_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 from separately loaded close prices
            # Localize close datetimes to Asia/Shanghai-aware to match factor_df's
            # aware datetimes (compute_factors localizes), else the date-alignment
            # filter (prices.index.isin(factor_dates)) empties prices → concat error.
            _close_dt = pd.to_datetime(close_pd["datetime"])
            if _close_dt.dt.tz is None:
                _close_dt = _close_dt.dt.tz_localize("Asia/Shanghai")
            else:
                _close_dt = _close_dt.dt.tz_convert("Asia/Shanghai")
            close_pd["datetime"] = _close_dt
            prices_df = close_pd.pivot(index="datetime", columns="vt_symbol", values="close")

            # CRITICAL FIX: Align price data with factor data date range
            # Factor data only contains test period, but price data contains full range
            # Filter prices to only include dates that exist in factor data
            factor_dates = factor_series.index.get_level_values('datetime').unique()
            prices_df = prices_df[prices_df.index.isin(factor_dates)]

            # Ensure datetime index for prices
            prices_df.index = pd.to_datetime(prices_df.index)

            # busdaycal 调试:定位 factor/prices 数据是否空或日期不对齐
            if prices_df.empty or len(factor_dates) == 0:
                fmin = factor_pd['datetime'].min() if len(factor_pd) else None
                fmax = factor_pd['datetime'].max() if len(factor_pd) else None
                cmin = close_pd['datetime'].min() if len(close_pd) else None
                cmax = close_pd['datetime'].max() if len(close_pd) else None
                ic_summary[factor_name] = {
                    "status": "error",
                    "error": f"DBG empty: factor_df.height={factor_df.height}, factor_dates={len(factor_dates)}, factor_series={len(factor_series)}, close_rows={len(close_pd)}, prices_df={prices_df.shape}, factor_dt={fmin}~{fmax}, close_dt={cmin}~{cmax}",
                }
                continue

            # Call get_clean_factor_and_forward_returns
            merged_data = get_clean_factor_and_forward_returns(
                factor=factor_series,
                prices=prices_df,
                periods=periods,  # Use configurable periods
                max_loss=1.0  # TEMPORARY: Allow 100% loss to see IC data
            )

            # 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}"}

            # IC 已算完(ic_data)——先存成功。因子分析的核心指标(IC/ICIR/t_stat)可用,
            # 即使下面的 tears 报告因 alphalens-reloaded × pandas2 的 groupby.transform
            # 兼容问题崩,也不影响 IC 数值。
            ic_summary[factor_name] = {
                "status": "success",
                "ic": ic_data,
            }

            # Generate tears sheet（独立 try：tears 失败只标注，不覆盖上面的 IC 成功）
            #
            # alphalens-reloaded 的 create_full_tear_sheet 内部，每个 tear sheet
            # 末尾调 GridFigure.close() → plt.close(fig)。Agg 后端下 figure 被销毁，
            # 之后 plt.savefig 只能拿到最后一个空占位 figure（实证 2.3KB 空 PNG），
            # 且原代码 savefig 存 .png 却把 report_paths 记成 .html → report endpoint
            # 404。monkey-patch GridFigure.close 收集 figure，base64 嵌入真 HTML 文件。
            # 同 empyrical np.NINF / demean_forward_returns 适配思路，不改 alphalens 源码。
            from io import StringIO, BytesIO
            import base64
            import alphalens.tears as _al_tears
            old_stdout = sys.stdout
            sys.stdout = StringIO()  # Capture stdout (alphalens 打印统计表)
            try:
                _collected_figs: list = []
                _orig_gridfig_close = _al_tears.GridFigure.close
                def _collect_gridfig_close(self, _sink=_collected_figs):
                    fig = getattr(self, "fig", None)
                    if fig is not None:
                        _sink.append(fig)
                    self.gs = None  # 不调 plt.close，保留 figure
                _al_tears.GridFigure.close = _collect_gridfig_close
                try:
                    plt.close("all")
                    create_full_tear_sheet(
                        merged_data,
                        long_short=True,
                        group_neutral=False,
                        by_group=False,
                    )
                finally:
                    _al_tears.GridFigure.close = _orig_gridfig_close

                # 合并 collected + 仍存活的 figure（plot_quantile_statistics_table
                # 的表格图不经过 GridFigure，需从 fignums 补）
                _all_figs = list(_collected_figs)
                for _num in plt.get_fignums():
                    _f = plt.figure(_num)
                    if _f not in _all_figs:
                        _all_figs.append(_f)

                _img_tags = []
                for _fig in _all_figs:
                    _buf = BytesIO()
                    _fig.savefig(_buf, format="png", dpi=72, bbox_inches="tight")
                    _buf.seek(0)
                    _b64 = base64.b64encode(_buf.read()).decode("ascii")
                    _img_tags.append(f'<img src="data:image/png;base64,{_b64}"/>')
                    plt.close(_fig)

                _html = (
                    f'<html><head><meta charset="utf-8"><title>{factor_name} tears</title>'
                    '<style>body{font-family:-apple-system,Segoe UI,sans-serif;margin:0;'
                    'padding:12px;background:#fff;}h2{margin:0 0 12px;font-size:18px;}'
                    'img{display:block;max-width:100%;margin:0 0 14px;border:1px solid #eee;}'
                    '</style></head><body>'
                    f'<h2>{factor_name} — Alphalens Tearsheet</h2>'
                    + "".join(_img_tags)
                    + "</body></html>"
                )
                factor_report_path = os.path.join(output_dir, f"{factor_name}_tears.html")
                with open(factor_report_path, "w", encoding="utf-8") as _fp:
                    _fp.write(_html)
                report_paths[factor_name] = factor_report_path
                ic_summary[factor_name]["report"] = factor_report_path
            except Exception as tears_e:
                ic_summary[factor_name]["tears_error"] = f"{type(tears_e).__name__}: {tears_e}"
            finally:
                sys.stdout = old_stdout  # Restore stdout

        except Exception as e:
            err_type = type(e).__name__
            ic_summary[factor_name] = {
                "status": "error",
                "error": f"{err_type}: {e}",
                "traceback": traceback.format_exc()
            }

    return FactorReport(
        factor_names=factor_names,
        output_dir=output_dir,
        ic_summary=ic_summary,
        report_paths=report_paths,
        symbols=symbols,
        start=start,
        end=end,
    )