"""Versioned three-day policy with a latest-day creative stop supplement."""

from __future__ import annotations

from dataclasses import dataclass
from typing import Iterable

import numpy as np
import pandas as pd

from .fission_multiplier import FissionMultiplierParameters
from .metrics import (
    ENTITY_GZH,
    ENTITY_SELF,
    ENTITY_SELF_AD,
    compute_roi_summary,
)


POLICY_VERSION = "roi_policy_v12"
POLICY_RUN_SUFFIX = "p12"


@dataclass(frozen=True)
class RuleConfig:
    self_stop_min_age: int = 4
    self_up_min_age: int = 3
    self_min_daily_uv: float = 200
    partner_min_daily_uv: float = 200
    observe_min_latest_uv: float = 200
    one_day_min_uv: float = 200
    one_day_p30_min_uv: float = 500
    one_day_hard_stop_roi: float = 0.20
    one_day_stop_quantile: float = 0.30
    stop_quantile: float = 0.20
    up_quantile: float = 0.80


def _daily_sample_mask(
    summary: pd.DataFrame,
    entity_type: str,
    dt: str,
    config: RuleConfig,
) -> pd.Series:
    min_uv = (
        config.partner_min_daily_uv
        if entity_type == ENTITY_GZH
        else config.self_min_daily_uv
    )
    return (
        summary["entity_type"].eq(entity_type)
        & pd.to_numeric(summary[f"首层UV_{dt}"], errors="coerce").gt(min_uv)
        & pd.to_numeric(summary[f"成本_{dt}"], errors="coerce").gt(0)
        & np.isfinite(pd.to_numeric(summary[f"ROI_{dt}"], errors="coerce"))
    )


def entity_eligibility_mask(
    summary: pd.DataFrame,
    entity_type: str,
    config: RuleConfig,
    expected_dates: list[str],
) -> pd.Series:
    eligible = summary["entity_type"].eq(entity_type)
    for dt in expected_dates:
        eligible &= _daily_sample_mask(summary, entity_type, dt, config)
    return eligible


def threshold_eligibility_mask(
    summary: pd.DataFrame,
    config: RuleConfig,
    expected_dates: list[str],
) -> pd.Series:
    """Only creative-level miniapp and official accounts enter global P20."""

    return entity_eligibility_mask(
        summary, ENTITY_SELF, config, expected_dates
    ) | entity_eligibility_mask(summary, ENTITY_GZH, config, expected_dates)


def report_formal_mask(
    summary: pd.DataFrame,
    config: RuleConfig,
    expected_dates: list[str],
) -> pd.Series:
    return threshold_eligibility_mask(
        summary, config, expected_dates
    ) | entity_eligibility_mask(summary, ENTITY_SELF_AD, config, expected_dates)


def observation_mask(
    summary: pd.DataFrame,
    config: RuleConfig,
    expected_dates: list[str],
) -> pd.Series:
    latest = expected_dates[-1]
    formal = report_formal_mask(summary, config, expected_dates)
    return (
        summary["entity_type"].isin([ENTITY_SELF, ENTITY_SELF_AD, ENTITY_GZH])
        & ~formal
        & pd.to_numeric(summary[f"首层UV_{latest}"], errors="coerce").gt(
            config.observe_min_latest_uv
        )
    )


def one_day_p30_pool_mask(
    summary: pd.DataFrame,
    config: RuleConfig,
    expected_dates: list[str],
) -> pd.Series:
    latest = expected_dates[-1]
    latest_roi = pd.to_numeric(summary[f"ROI_{latest}"], errors="coerce")
    return (
        summary["entity_type"].eq(ENTITY_SELF)
        & pd.to_numeric(summary[f"首层UV_{latest}"], errors="coerce").gt(
            config.one_day_p30_min_uv
        )
        & pd.to_numeric(summary[f"成本_{latest}"], errors="coerce").gt(0)
        & np.isfinite(latest_roi)
    )


def one_day_supplement_mask(
    summary: pd.DataFrame,
    config: RuleConfig,
    expected_dates: list[str],
) -> pd.Series:
    latest = expected_dates[-1]
    latest_roi = pd.to_numeric(summary[f"ROI_{latest}"], errors="coerce")
    formal_creative = entity_eligibility_mask(
        summary, ENTITY_SELF, config, expected_dates
    )
    return (
        summary["entity_type"].eq(ENTITY_SELF)
        & ~formal_creative
        & pd.to_numeric(summary[f"首层UV_{latest}"], errors="coerce").gt(
            config.one_day_min_uv
        )
        & pd.to_numeric(summary[f"成本_{latest}"], errors="coerce").gt(0)
        & np.isfinite(latest_roi)
    )


def compute_global_threshold(
    summary: pd.DataFrame,
    expected_dates: list[str],
    config: RuleConfig,
) -> pd.DataFrame:
    eligible = threshold_eligibility_mask(summary, config, expected_dates)
    sample = pd.to_numeric(summary.loc[eligible, "ROI"], errors="coerce")
    sample = sample[np.isfinite(sample)]
    if sample.empty:
        raise ValueError("最近三日没有满足每日UV和成本门槛的统一阈值样本")
    creative_eligible = entity_eligibility_mask(
        summary, ENTITY_SELF, config, expected_dates
    )
    creative_sample = pd.to_numeric(
        summary.loc[creative_eligible, "ROI"], errors="coerce"
    )
    creative_sample = creative_sample[np.isfinite(creative_sample)]
    t_up = (
        float(creative_sample.quantile(config.up_quantile))
        if not creative_sample.empty
        else np.nan
    )
    latest = expected_dates[-1]
    one_day_pool = pd.to_numeric(
        summary.loc[
            one_day_p30_pool_mask(summary, config, expected_dates),
            f"ROI_{latest}",
        ],
        errors="coerce",
    )
    one_day_pool = one_day_pool[np.isfinite(one_day_pool)]
    t_one_day_stop = (
        float(one_day_pool.quantile(config.one_day_stop_quantile))
        if not one_day_pool.empty
        else np.nan
    )
    return pd.DataFrame(
        [
            {
                "统计窗口": f"{expected_dates[0]} 至 {expected_dates[-1]}",
                "entity_type": "global",
                "渠道": "小程序创意级+公众号",
                "t_stop": float(sample.quantile(config.stop_quantile)),
                "t_up": t_up,
                "t_one_day_stop": t_one_day_stop,
                "阈值样本数": int(len(sample)),
                "扩量样本数": int(len(creative_sample)),
                "单日P30样本数": int(len(one_day_pool)),
                "单日硬关停ROI": config.one_day_hard_stop_roi,
                "单日最低UV": config.one_day_min_uv,
                "单日P30最低UV": config.one_day_p30_min_uv,
                "关停线口径": f"合格实体等权P{int(config.stop_quantile * 100)}",
                "扩量线口径": f"合格小程序创意实体等权P{int(config.up_quantile * 100)}",
                "广告级是否入池": "否_仅复用统一关停线",
            }
        ]
    )


def _sample_percentile(values: pd.Series, target: float) -> float:
    clean = pd.to_numeric(values, errors="coerce")
    clean = clean[np.isfinite(clean)]
    if clean.empty or not np.isfinite(target):
        return np.nan
    return float((clean <= target).mean())


def apply_actions(
    summary: pd.DataFrame,
    thresholds: pd.DataFrame,
    config: RuleConfig,
    expected_dates: list[str],
) -> pd.DataFrame:
    result = summary.copy()
    result["动作"] = ""
    result["动作原因"] = ""
    result["t_stop"] = float(thresholds.iloc[0]["t_stop"])
    result["t_up"] = float(thresholds.iloc[0]["t_up"])
    result["t_one_day_stop"] = float(
        thresholds.iloc[0]["t_one_day_stop"]
    )

    threshold_eligible = threshold_eligibility_mask(result, config, expected_dates)
    ad_eligible = entity_eligibility_mask(
        result, ENTITY_SELF_AD, config, expected_dates
    )
    observe_only = observation_mask(result, config, expected_dates)
    one_day_supplement = one_day_supplement_mask(
        result, config, expected_dates
    )
    sample_roi = result.loc[threshold_eligible, "ROI"]
    creative_eligible = entity_eligibility_mask(
        result, ENTITY_SELF, config, expected_dates
    )
    creative_sample_roi = result.loc[creative_eligible, "ROI"]
    result["整体三日ROI排名百分位"] = result["ROI"].apply(
        lambda value: _sample_percentile(sample_roi, float(value))
        if pd.notna(value)
        else np.nan
    )
    result["是否位于三日ROI后20%"] = False
    result["创意三日ROI排名百分位"] = result["ROI"].apply(
        lambda value: _sample_percentile(creative_sample_roi, float(value))
        if pd.notna(value)
        else np.nan
    )
    result["是否位于创意三日ROI前20%"] = False
    comparable = threshold_eligible | ad_eligible
    result.loc[comparable, "是否位于三日ROI后20%"] = (
        pd.to_numeric(result.loc[comparable, "ROI"], errors="coerce")
        <= result.loc[comparable, "t_stop"]
    )
    if np.isfinite(float(result["t_up"].iloc[0])):
        result.loc[creative_eligible, "是否位于创意三日ROI前20%"] = (
            pd.to_numeric(result.loc[creative_eligible, "ROI"], errors="coerce")
            >= result.loc[creative_eligible, "t_up"]
        )

    for index, row in result.iterrows():
        entity_type = str(row["entity_type"])
        if bool(one_day_supplement.loc[index]):
            latest = expected_dates[-1]
            latest_uv = float(row[f"首层UV_{latest}"])
            latest_roi = float(row[f"ROI_{latest}"])
            three_day_roi = float(row["ROI"])
            raw_age = row.get("广告age")
            age = int(raw_age) if pd.notna(raw_age) else 0
            decision_context = (
                "未满足连续三天每天首层UV>200、成本>0且ROI有效，"
                "启用单日补充规则；"
            )
            three_day_context = (
                f"；三日预测总效率ROI={three_day_roi:.2f}仅作参考，"
                "不参与本次单日判断"
            )
            stop_reason = ""
            if latest_roi <= config.one_day_hard_stop_roi:
                stop_reason = (
                    f"最新日首层UV={latest_uv:.0f}>{config.one_day_min_uv:g}，"
                    f"最新日预测总效率ROI={latest_roi:.2f}≤单日硬关停线"
                    f"{config.one_day_hard_stop_roi:.2f}"
                )
            elif (
                latest_uv > config.one_day_p30_min_uv
                and np.isfinite(float(row["t_one_day_stop"]))
                and latest_roi <= float(row["t_one_day_stop"])
            ):
                stop_reason = (
                    f"最新日首层UV={latest_uv:.0f}>{config.one_day_p30_min_uv:g}，"
                    f"最新日预测总效率ROI={latest_roi:.2f}≤单日实体等权P30关停线"
                    f"{float(row['t_one_day_stop']):.2f}"
                )

            if stop_reason and age >= config.self_stop_min_age:
                result.at[index, "动作"] = "关停"
                result.at[index, "动作原因"] = (
                    f"{decision_context}{stop_reason}{three_day_context}；"
                    f"广告age={age}>{config.self_stop_min_age - 1}天，"
                    "建议审批后暂停动态创意"
                )
            elif stop_reason:
                result.at[index, "动作"] = "观察"
                result.at[index, "动作原因"] = (
                    f"{decision_context}{stop_reason}{three_day_context}；"
                    f"广告age={age}≤{config.self_stop_min_age - 1}天，暂不关停"
                )
            else:
                result.at[index, "动作"] = "观察"
                result.at[index, "动作原因"] = (
                    f"{decision_context}"
                    f"最新日首层UV={latest_uv:.0f}>{config.one_day_min_uv:g}，"
                    f"最新日预测总效率ROI={latest_roi:.2f}未命中单日关停规则"
                    f"{three_day_context}；建议继续观察"
                )
            continue
        if bool(observe_only.loc[index]):
            result.at[index, "动作"] = "观察"
            result.at[index, "动作原因"] = (
                f"最新日首层UV>{config.observe_min_latest_uv:g}，但未满足连续三天"
                "每天首层UV>200、成本>0且ROI有效；仅置底展示，不进入P20和自动执行"
            )
            continue
        if bool(ad_eligible.loc[index]):
            if float(row["ROI"]) <= float(row["t_stop"]):
                raw_age = row.get("广告age")
                age = int(raw_age) if pd.notna(raw_age) else 0
                if age >= config.self_stop_min_age:
                    result.at[index, "动作"] = "关停"
                    result.at[index, "动作原因"] = (
                        f"广告级三日加权平均效率ROI≤统一实体等权P20，"
                        f"广告age>{config.self_stop_min_age - 1}天；审批后暂停整个广告"
                    )
                else:
                    result.at[index, "动作"] = "观察"
                    result.at[index, "动作原因"] = (
                        f"广告级三日加权平均效率ROI≤统一实体等权P20，但广告age≤"
                        f"{config.self_stop_min_age - 1}天"
                    )
            continue
        if not bool(threshold_eligible.loc[index]):
            continue
        if entity_type == ENTITY_SELF:
            raw_age = row.get("广告age")
            age = int(raw_age) if pd.notna(raw_age) else 0
            if float(row["ROI"]) <= float(row["t_stop"]):
                if age >= config.self_stop_min_age:
                    result.at[index, "动作"] = "关停"
                    result.at[index, "动作原因"] = (
                        f"三日加权平均效率ROI≤统一实体等权P20，"
                        f"广告age>{config.self_stop_min_age - 1}天；审批后仅暂停动态创意"
                    )
                else:
                    result.at[index, "动作"] = "观察"
                    result.at[index, "动作原因"] = (
                        f"三日加权平均效率ROI≤统一实体等权P20，但广告age≤"
                        f"{config.self_stop_min_age - 1}天"
                    )
            elif bool(row["是否位于创意三日ROI前20%"]):
                if age >= config.self_up_min_age:
                    result.at[index, "动作"] = "扩量"
                    result.at[index, "动作原因"] = (
                        f"三日加权平均效率ROI≥合格创意实体等权P80，"
                        f"广告age≥{config.self_up_min_age}天；审批后提高广告永久基础出价"
                    )
                else:
                    result.at[index, "动作"] = "观察"
                    result.at[index, "动作原因"] = (
                        f"三日加权平均效率ROI≥合格创意实体等权P80，但广告age<"
                        f"{config.self_up_min_age}天"
                    )
        elif entity_type == ENTITY_GZH and float(row["ROI"]) <= float(row["t_stop"]):
            result.at[index, "动作"] = "关停"
            result.at[index, "动作原因"] = (
                "三日加权平均效率ROI≤统一实体等权P20；公众号当前仅通知参考"
            )
    return result


def evaluate_rules(
    raw_daily: pd.DataFrame,
    expected_dates: Iterable[str],
    ad_age: pd.DataFrame | None = None,
    config: RuleConfig = RuleConfig(),
    *,
    fission_parameters: FissionMultiplierParameters,
) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame]:
    summary, dates = compute_roi_summary(
        raw_daily,
        expected_dates,
        ad_age,
        fission_parameters=fission_parameters,
    )
    thresholds = compute_global_threshold(summary, dates, config)
    evaluated = apply_actions(summary, thresholds, config, dates)
    formal_threshold = threshold_eligibility_mask(evaluated, config, dates)
    formal_ad = entity_eligibility_mask(evaluated, ENTITY_SELF_AD, config, dates)
    one_day_supplement = one_day_supplement_mask(evaluated, config, dates)
    observe_only = observation_mask(evaluated, config, dates) & ~one_day_supplement
    evaluated["阈值样本状态"] = "未达到三日正式样本门槛"
    evaluated.loc[formal_threshold, "阈值样本状态"] = "进入三日统一阈值样本池"
    evaluated.loc[formal_ad, "阈值样本状态"] = "广告级三日合格_不进入阈值样本池"
    evaluated.loc[one_day_supplement, "阈值样本状态"] = (
        "单日补充决策_昨日UV>200"
    )
    evaluated.loc[observe_only, "阈值样本状态"] = "补充观察_昨日UV>200"
    candidates = evaluated[evaluated["动作"].ne("")].copy()
    return candidates, thresholds, evaluated