from __future__ import annotations

import json
from pathlib import Path
import tomllib
from typing import Any

import duckdb
import numpy as np
import pandas as pd

from .config import Settings
from .factor_dataset import build_monthly_factor_panel
from .factor_lab import optimize_weights_walk_forward, run_factor_backtest
from .research_context import load_research_context
from .text_research import analyze_text_sources


M2_MODEL_VERSION = "0.1.0"


def _resolve(root: Path, value: str) -> Path:
    path = Path(value)
    return path if path.is_absolute() else root / path


def _save(frame: pd.DataFrame, stem: str, settings: Settings) -> None:
    frame.to_parquet(settings.processed_dir / f"{stem}.parquet", index=False)
    frame.to_csv(settings.processed_dir / f"{stem}.csv", index=False, encoding="utf-8-sig")


def _aggregate_text(documents: pd.DataFrame, as_of: pd.Timestamp) -> pd.DataFrame:
    columns = [
        "source_code", "text_document_count", "latest_text_date", "text_risk_score",
        "text_catalyst_score", "moat_score", "management_score", "management_risk_score",
        "qualitative_score", "text_model_version",
    ]
    usable = documents.loc[documents["source_code"].notna()].copy()
    usable["announcement_date"] = pd.to_datetime(usable["announcement_date"], errors="coerce")
    usable = usable.loc[usable["announcement_date"].isna() | usable["announcement_date"].le(as_of)]
    if usable.empty:
        return pd.DataFrame(columns=columns)
    usable = usable.sort_values(["source_code", "announcement_date"], na_position="first")
    usable = usable.groupby("source_code", group_keys=False).tail(20)
    result = usable.groupby("source_code", as_index=False).agg(
        text_document_count=("document_id", "size"),
        latest_text_date=("announcement_date", "max"),
        text_risk_score=("risk_score", "max"),
        text_catalyst_score=("catalyst_score", "mean"),
        moat_score=("moat_score", "mean"),
        management_score=("management_score", "mean"),
        management_risk_score=("management_risk_score", "max"),
    )
    positive = result[["moat_score", "management_score", "text_catalyst_score"]].mul([0.4, 0.4, 0.2]).sum(axis=1)
    result["qualitative_score"] = (positive - 0.5 * result["text_risk_score"] - 0.25 * result["management_risk_score"]).clip(0, 100)
    result["text_model_version"] = M2_MODEL_VERSION
    return result.reindex(columns=columns)


def _enrich_candidates(candidates: pd.DataFrame, text: pd.DataFrame, context: pd.DataFrame) -> pd.DataFrame:
    result = candidates.merge(text, on="source_code", how="left").merge(context, on="source_code", how="left")
    parts = [("combined_score", 0.70), ("qualitative_score", 0.15), ("research_context_score", 0.15)]
    numerator = pd.Series(0.0, index=result.index)
    denominator = pd.Series(0.0, index=result.index)
    for column, weight in parts:
        values = pd.to_numeric(result[column], errors="coerce")
        available = values.notna()
        numerator = numerator.add(values.fillna(0) * weight)
        denominator = denominator.add(available.astype(float) * weight)
    result["m2_composite_score"] = numerator.div(denominator.replace(0, np.nan))
    result["m2_research_coverage"] = result[["qualitative_score", "research_context_score"]].notna().mean(axis=1)
    result["m2_model_version"] = M2_MODEL_VERSION
    return result.sort_values(["candidate_eligible", "m2_composite_score"], ascending=[False, False])


def _persist_tables(settings: Settings, tables: dict[str, pd.DataFrame]) -> None:
    database = settings.database_dir / "fundamental_mvp.duckdb"
    with duckdb.connect(str(database)) as connection:
        for name, frame in tables.items():
            connection.register("m2_frame", frame)
            connection.execute(f"CREATE OR REPLACE TABLE {name} AS SELECT * FROM m2_frame")
            connection.unregister("m2_frame")


def _fmt_pct(value: object) -> str:
    return "-" if pd.isna(value) else f"{float(value):.2%}"


def _write_report(
    settings: Settings,
    summary: dict[str, Any],
    factor_summary: pd.DataFrame,
    weights: pd.Series,
    enriched: pd.DataFrame,
) -> None:
    factor_lines = "\n".join(
        f"| {row.factor} | {int(row.periods)} | {row.mean_rank_ic:.4f} | {_fmt_pct(row.mean_long_short_net_return)} | {_fmt_pct(row.cumulative_long_short_net_return)} |"
        for row in factor_summary.sort_values("mean_rank_ic", ascending=False).itertuples(index=False)
    ) or "| - | 0 | - | - | - |"
    weight_lines = "\n".join(f"| {name} | {value:.2%} |" for name, value in weights.sort_values(ascending=False).items()) or "| - | - |"
    top = enriched.loc[enriched.get("candidate_eligible", False)].head(20)
    top_lines = "\n".join(
        f"| {row.security_id} | {row.name} | {row.m2_composite_score:.2f} | {row.m2_research_coverage:.0%} |"
        for row in top.itertuples(index=False)
    ) or "| - | - | - | - |"
    report = f"""# M2 深度研究与因子实验报告

模型版本：{M2_MODEL_VERSION}  
研究截面：{summary['as_of']}  
文本文档：{summary['text_documents']} 份，覆盖 {summary['text_stocks']} 只股票  
行业/概念/催化剂上下文覆盖：{summary['context_stocks']} 只  
因子面板：{summary['factor_rows']} 行，{summary['factor_periods']} 个月度截面  
权重样本外检验：{summary['oos_periods']} 个月度截面

## 1. 单因子结果

| 因子 | 期数 | 平均 RankIC | 月均多空净收益 | 累计多空净收益 |
|---|---:|---:|---:|---:|
{factor_lines}

## 2. 最新走步优化权重

| 因子 | 权重 |
|---|---:|
{weight_lines}

权重仅用调仓日之前 {summary['training_periods']} 个月训练，非负且和为 1；回测已计入双边换手所对应的配置交易成本。

## 3. M2 候选样本

| 代码 | 名称 | M2综合分 | 外部研究覆盖 |
|---|---|---:|---:|
{top_lines}

## 4. 口径与限制

- 财务因子仅在 `available_trade_date` 不晚于调仓截面时使用，标签收益不参与当期因子构建。
- 行业映射目前是 M1 当前截面，历史行业变更仍存在分类偏差。
- 公告文本得分是可解释关键词研究信号，不是对公司质地的事实认定；高风险文档必须复核原文。
- 外部文本、行业高频、产业链和催化剂输入缺失时不填充伪分数，以覆盖率显式标记。

本报告用于研究管线验证，不构成投资建议。
"""
    (settings.report_dir / "M2_深度研究与因子实验.md").write_text(report, encoding="utf-8")


def run_m2(settings: Settings, config_path: Path) -> dict[str, Any]:
    raw = tomllib.loads(config_path.read_text(encoding="utf-8"))
    paths = raw["paths"]
    backtest_config = raw["backtest"]
    settings.processed_dir.mkdir(parents=True, exist_ok=True)
    settings.report_dir.mkdir(parents=True, exist_ok=True)

    panel_result = build_monthly_factor_panel(settings, str(backtest_config["start_date"]))
    panel = pd.read_parquet(panel_result.output_path)
    panel = panel.loc[
        ~panel["industry_l1_code"].isin(["X50", "X51"])
        & panel["avg_amount"].ge(settings.min_avg_amount_20)
        & panel["available_trade_date"].notna()
    ].copy()
    as_of = pd.Timestamp(panel["rebalance_date"].max()).normalize()

    documents = analyze_text_sources(_resolve(settings.project_root, str(paths["disclosure_dir"])))
    text_scores = _aggregate_text(documents, as_of)
    master = pd.read_parquet(settings.processed_dir / "security_master.parquet")
    universe = master[["source_code", "industry_l2_code"]].rename(columns={"industry_l2_code": "industry_code"})
    context_config = dict(raw.get("research_context", {}))
    context_config.update({
        "base_dir": str(settings.project_root),
        "industry_data_path": paths["industry_high_frequency"],
        "theme_mapping_path": paths["theme_mapping"],
        "catalyst_path": paths["catalysts"],
    })
    research_context = load_research_context(context_config, as_of)
    context_scores = research_context.score_stocks(universe)
    context_scores.loc[context_scores["research_context_coverage"].eq(0), "research_context_score"] = np.nan

    factors = list(backtest_config["factor_columns"])
    common = dict(
        available_date_col="available_trade_date",
        quantiles=int(backtest_config["quantiles"]),
        transaction_cost_bps=float(backtest_config["transaction_cost_bps"]),
        min_assets=int(backtest_config["minimum_assets"]),
    )
    factor_test = run_factor_backtest(panel, factors, **common)
    optimized_factors = list(backtest_config.get("optimization_columns", factors))
    optimized = optimize_weights_walk_forward(
        panel, optimized_factors,
        train_periods=int(backtest_config["training_periods"]),
        candidate_count=int(backtest_config["weight_candidates"]),
        max_weight=float(backtest_config.get("max_factor_weight", 1.0)),
        objective=str(backtest_config["optimization_objective"]),
        **common,
    )

    candidates = pd.read_parquet(settings.processed_dir / "stock_screen_candidates.parquet")
    enriched = _enrich_candidates(candidates, text_scores, context_scores)
    tables = {
        "text_research_documents": documents,
        "text_research_scores": text_scores,
        "research_context_scores": context_scores,
        "factor_backtest_summary": factor_test.summary,
        "factor_weight_history": optimized.weight_history,
        "m2_stock_candidates": enriched,
    }
    for name, frame in tables.items():
        _save(frame, name, settings)
    _save(factor_test.group_returns, "factor_group_returns", settings)
    _save(factor_test.period_returns, "factor_period_returns", settings)
    _save(factor_test.ic, "factor_ic", settings)
    _save(optimized.oos_backtest.period_returns, "factor_oos_period_returns", settings)
    _persist_tables(settings, tables)

    summary: dict[str, Any] = {
        "model_version": M2_MODEL_VERSION,
        "as_of": as_of.date().isoformat(),
        "text_documents": len(documents),
        "text_stocks": int(text_scores["source_code"].nunique()) if not text_scores.empty else 0,
        "context_stocks": int(context_scores["research_context_coverage"].gt(0).sum()),
        "factor_rows": panel_result.rows,
        "factor_stocks": panel_result.stocks,
        "factor_periods": panel_result.periods,
        "oos_periods": int(optimized.weight_history["rebalance_date"].nunique()) if not optimized.weight_history.empty else 0,
        "training_periods": int(backtest_config["training_periods"]),
        "latest_weights": {name: round(float(weight), 8) for name, weight in optimized.weights.items()},
        "outputs": {name: str(settings.processed_dir / f"{name}.parquet") for name in tables},
    }
    (settings.report_dir / "m2_run_summary.json").write_text(
        json.dumps(summary, ensure_ascii=False, indent=2), encoding="utf-8"
    )
    _write_report(settings, summary, factor_test.summary, optimized.weights, enriched)
    return summary