from __future__ import annotations

from dataclasses import dataclass
from pathlib import Path

import duckdb
import numpy as np
import pandas as pd

from .config import Settings


QUALITY_SIGNAL_VERSION = "0.1.0"


@dataclass(frozen=True)
class QualitySignalResult:
    rows: int
    covered: int
    output_path: Path


def _industry_rank(frame: pd.DataFrame, column: str, higher: bool = True) -> pd.Series:
    values = pd.to_numeric(frame[column], errors="coerce").replace([np.inf, -np.inf], np.nan)
    return values.groupby(frame["industry_l2_code"]).rank(pct=True, ascending=higher)


def build_quality_signals(settings: Settings) -> QualitySignalResult:
    scores = pd.read_parquet(settings.processed_dir / "fundamental_scores.parquet")
    finance = pd.read_parquet(settings.financial_pit_path)
    as_of = pd.Timestamp(scores["available_trade_date"].max())
    annual = finance.loc[
        finance["period_type"].eq("FY") & finance["available_trade_date"].le(as_of)
    ].sort_values(["source_code", "report_period"]).groupby("source_code", group_keys=False).tail(5)
    annual["is_profitable"] = annual["parent_net_profit_ytd"].gt(0)
    annual["roe_above_10"] = annual["weighted_roe_reported"].ge(10)
    history = annual.groupby("source_code").agg(
        annual_observations_5y=("report_period", "size"),
        profitable_years_5y=("is_profitable", "sum"),
        roe_above_10_years_5y=("roe_above_10", "sum"),
        gross_margin_stability_5y=("gross_margin_reported", "std"),
    ).reset_index()
    frame = scores.merge(history, on="source_code", how="left")
    frame["profit_persistence"] = frame["profitable_years_5y"] / frame["annual_observations_5y"].replace(0, np.nan)
    frame["roe_persistence"] = frame["roe_above_10_years_5y"] / frame["annual_observations_5y"].replace(0, np.nan)
    frame["margin_stability"] = -frame["gross_margin_stability_5y"]

    moat_columns = [
        ("weighted_roe_reported", True), ("gross_margin_reported", True),
        ("cfo_to_parent_profit", True), ("asset_turnover_reported", True),
        ("profit_persistence", True), ("roe_persistence", True), ("margin_stability", True),
    ]
    innovation_columns = [
        ("research_development_ratio", True), ("operating_revenue_yoy_reported", True),
        ("net_profit_yoy_reported", True), ("revenue_cagr_3y", True),
        ("parent_profit_cagr_3y", True),
    ]
    management_columns = [
        ("cfo_to_parent_profit", True), ("parent_equity_yoy", True),
        ("warning_count", False), ("goodwill_equity_ratio", False),
    ]

    def composite(specs: list[tuple[str, bool]]) -> tuple[pd.Series, pd.Series]:
        matrix = pd.concat([_industry_rank(frame, column, higher) for column, higher in specs], axis=1)
        return matrix.mean(axis=1, skipna=True) * 100, matrix.notna().mean(axis=1)

    frame["moat_quant_proxy_score"], moat_coverage = composite(moat_columns)
    frame["innovation_quant_proxy_score"], innovation_coverage = composite(innovation_columns)
    frame["management_quant_proxy_score"], management_coverage = composite(management_columns)
    frame["quality_signal_coverage"] = pd.concat([moat_coverage, innovation_coverage, management_coverage], axis=1).mean(axis=1)
    insufficient_moat_history = frame["annual_observations_5y"].lt(3)
    frame.loc[insufficient_moat_history, "moat_quant_proxy_score"] = np.nan
    frame.loc[frame["is_financial_sector"], [
        "moat_quant_proxy_score", "innovation_quant_proxy_score", "management_quant_proxy_score",
        "quality_signal_coverage",
    ]] = np.nan
    frame["moat_proxy_level"] = pd.cut(
        frame["moat_quant_proxy_score"], [-np.inf, 40, 60, 75, np.inf],
        labels=["弱", "一般", "较强", "强"], right=False,
    ).astype("string")
    frame["innovation_proxy_level"] = pd.cut(
        frame["innovation_quant_proxy_score"], [-np.inf, 40, 60, 75, np.inf],
        labels=["弱", "一般", "较强", "强"], right=False,
    ).astype("string")
    frame["evidence_status"] = np.where(
        frame["quality_signal_coverage"].ge(0.7), "QUANTITATIVE_PROXY_ONLY", "INSUFFICIENT_QUANTITATIVE_EVIDENCE"
    )
    frame["moat_evidence_status"] = np.select([
        frame["is_financial_sector"], insufficient_moat_history,
    ], ["FINANCIAL_MODEL_NOT_APPLICABLE", "INSUFFICIENT_HISTORY"], default="QUANTITATIVE_PROXY_ONLY")
    output_columns = [
        "security_id", "source_code", "name", "industry_l2_code", "industry_l2_name",
        "moat_quant_proxy_score", "moat_proxy_level", "innovation_quant_proxy_score",
        "innovation_proxy_level", "management_quant_proxy_score", "quality_signal_coverage",
        "annual_observations_5y", "profitable_years_5y", "roe_above_10_years_5y",
        "gross_margin_stability_5y", "evidence_status",
        "moat_evidence_status",
    ]
    output = frame[output_columns].copy()
    output["model_version"] = QUALITY_SIGNAL_VERSION
    path = settings.processed_dir / "quality_signal_proxies.parquet"
    output.to_parquet(path, index=False)
    output.to_csv(settings.processed_dir / "quality_signal_proxies.csv", index=False, encoding="utf-8-sig")
    with duckdb.connect(str(settings.database_dir / "fundamental_mvp.duckdb")) as connection:
        connection.register("quality_df", output)
        connection.execute("CREATE OR REPLACE TABLE quality_signal_proxies AS SELECT * FROM quality_df")
    return QualitySignalResult(len(output), int(output["quality_signal_coverage"].ge(0.7).sum()), path)