"""admin 大盘聚合查询(全局 count/sum/DAU/成功率)。全部只读、不改任何数据。

⚠️ 性能:这些是全表 count/sum,P0 数据量小够用;用户量上来后热点字段(user.created_at /
user.last_login_at / comparison_record.status / withdraw_order.status)要加索引,或改增量统计表。
"""
from __future__ import annotations

from collections import Counter
from datetime import UTC, date, datetime, time, timedelta, timezone
from decimal import ROUND_HALF_UP, Decimal, InvalidOperation

from sqlalchemy import case, func, or_, select
from sqlalchemy.orm import Session

from app.admin.repositories.coupon_data import _percentile
from app.models.ad_feed_reward import AdFeedRewardRecord
from app.models.ad_reward import AdRewardRecord
from app.models.analytics_event import AnalyticsEvent
from app.models.comparison import ComparisonRecord
from app.models.coupon_state import (
    CouponClaimRecord,
    CouponPromptEngagement,
    CouponSession,
)
from app.models.cps_order import CpsOrder
from app.models.feedback import Feedback
from app.models.savings import SavingsRecord
from app.models.signin import SigninRecord
from app.models.user import User
from app.models.wallet import CoinTransaction, WithdrawOrder

_BEIJING = timezone(timedelta(hours=8))
REWARD_VIDEO_BIZ_TYPES = ("reward_video", "ad_reward")
# 领券/比价奖励金币的真实来源是信息流广告发奖(ad_feed_reward_record,按 feed_scene 分场景);
# coin_transaction 里只有扁平的 feed_ad_reward、biz_type 不分 coupon/comparison,故这俩桶历史从未
# 被写入,仅留作未来兜底,实际金额在下方按 feed_scene 汇总 ad_feed_reward_record 得出。reward_video/
# ad_reward 是激励视频,单独成桶、不再混进领券奖励(历史误并会把激励视频金币双计进领券)。
COUPON_REWARD_BIZ_TYPES = ("coupon", "coupon_reward")
COMPARISON_REWARD_BIZ_TYPES = ("comparison", "compare_reward", "comparison_reward")
# 常规任务必须按明确来源相加；不能从全部正向流水反减排除项，否则新增广告/运营
# biz_type 时会在排除清单更新前自动混入该桶。task_ 前缀在查询处单独覆盖现有及未来任务。
REGULAR_TASK_EXACT_BIZ_TYPES = (
    "signin",
    "signin_boost",
    "price_report_reward",
    "feedback_reward",
)
MEITUAN_CPS_INVALID_STATUSES = ("4", "5")
MEITUAN_CPS_SETTLED_STATUS = "6"
COMPARE_START_EVENT = "real_compare_start"
COUPON_START_EVENT = "real_coupon_start"
JD_CPS_INVALID_CODES = {
    "2", "3", "4", "5", "6", "7", "8", "9", "11", "13", "14", "19", "20", "21",
    "22", "23", "25", "26", "27", "28", "29", "30", "31", "34", "35", "36",
}
JD_CPS_UNPAID_CODES = {"15"}


def _beijing_today_start_utc() -> datetime:
    """北京时间今天 0 点对应的 UTC 时刻(DAU / 今日新增按北京时区切天)。"""
    now_bj = datetime.now(_BEIJING)
    start_bj = now_bj.replace(hour=0, minute=0, second=0, microsecond=0)
    return start_bj.astimezone(UTC)


def today_dau(db: Session) -> int:
    """今日活跃用户数(DAU):登录 + 开始比价 + 开始领券,按用户去重。

    = period_active_dau(今天, 今天);历史 / 多天窗口用 period_active_dau 传区间(广告收益报表复用)。
    """
    today_bj = datetime.now(_BEIJING).date()
    return period_active_dau(db, today_bj, today_bj)


def _default_period_end() -> date:
    """新版大盘不含今日,默认窗口结束日=北京时间昨天。"""
    return datetime.now(_BEIJING).date() - timedelta(days=1)


def _normalize_period(date_from: date | None, date_to: date | None) -> tuple[date, date]:
    end = date_to or _default_period_end()
    start = date_from or end
    if start > end:
        start, end = end, start
    return start, end


def _period_bounds(date_from: date, date_to: date) -> tuple[datetime, datetime, datetime, datetime]:
    """返回同一北京自然日窗口的 UTC aware 边界和北京 naive 边界。

    user.created_at / last_login_at 是 UTC aware 口径;比较/金币等历史上有北京 naive
    写入,所以两套边界同时保留。
    """
    start_bj = datetime.combine(date_from, time.min, tzinfo=_BEIJING)
    end_bj = datetime.combine(date_to + timedelta(days=1), time.min, tzinfo=_BEIJING)
    start_utc = start_bj.astimezone(UTC)
    end_utc = end_bj.astimezone(UTC)
    return (
        start_utc,
        end_utc,
        start_bj.replace(tzinfo=None),
        end_bj.replace(tzinfo=None),
    )


def _date_range(date_from: date, date_to: date) -> list[date]:
    days = (date_to - date_from).days
    return [date_from + timedelta(days=i) for i in range(days + 1)]


def _duration_percentile(sorted_values: list[int], q: float) -> int | None:
    """Linear-interpolated percentile with the same half-up rounding as Math.round."""
    if not sorted_values:
        return None
    if len(sorted_values) == 1:
        return sorted_values[0]
    index = (len(sorted_values) - 1) * q
    lower = int(index)
    upper = min(lower + 1, len(sorted_values) - 1)
    fraction = Decimal(str(index - lower))
    value = (
        Decimal(sorted_values[lower]) * (Decimal(1) - fraction)
        + Decimal(sorted_values[upper]) * fraction
    )
    return int(value.quantize(Decimal("1"), rounding=ROUND_HALF_UP))


def _id_set(db: Session, stmt) -> set[int]:
    return {int(v) for v in db.execute(stmt).scalars().all() if v is not None}


def _event_user_ids(
    db: Session, event_names: tuple[str, ...], start_utc: datetime, end_utc: datetime
) -> set[int]:
    return _id_set(
        db,
        select(AnalyticsEvent.user_id).where(
            AnalyticsEvent.user_id.is_not(None),
            AnalyticsEvent.event.in_(event_names),
            AnalyticsEvent.created_at >= start_utc,
            AnalyticsEvent.created_at < end_utc,
        ),
    )


def _period_active_user_ids(
    db: Session,
    *,
    start_utc: datetime,
    end_utc: datetime,
    period_from: date,
    period_to: date,
) -> set[int]:
    """区间去重活跃用户 id 集合:登录(last_login_at)+ 开始比价(real_compare_start)+
    开始领券(real_coupon_start / claim_started)。

    user / analytics_event 用 UTC aware 边界 [start_utc, end_utc);coupon_prompt_engagement
    的 engage_date 是北京自然日 date 列,用 [period_from, period_to] 闭区间。今日 / 历史 / 多天通用。
    """
    login_user_ids = _id_set(
        db,
        select(User.id).where(User.last_login_at >= start_utc, User.last_login_at < end_utc),
    )
    compare_start_user_ids = _event_user_ids(db, (COMPARE_START_EVENT,), start_utc, end_utc)
    coupon_event_user_ids = _event_user_ids(db, (COUPON_START_EVENT,), start_utc, end_utc)
    coupon_claim_user_ids = _id_set(
        db,
        select(CouponPromptEngagement.user_id).where(
            CouponPromptEngagement.engage_date >= period_from,
            CouponPromptEngagement.engage_date <= period_to,
            CouponPromptEngagement.engage_type == "claim_started",
        ),
    )
    return login_user_ids | compare_start_user_ids | coupon_event_user_ids | coupon_claim_user_ids


def period_active_dau(db: Session, date_from: date, date_to: date) -> int:
    """任意北京自然日区间 [date_from, date_to] 的去重活跃用户数。

    与数据大盘 period.users.active 同口径(登录 + 开始比价 + 开始领券);广告收益报表按所选
    日期区间(含今日)复用,ARPU = 区间预估收益 ÷ 本数。全局口径,不随用户 / 类型 / 场景筛选变化。
    """
    period_from, period_to = _normalize_period(date_from, date_to)
    start_utc, end_utc, _start_local, _end_local = _period_bounds(period_from, period_to)
    return len(
        _period_active_user_ids(
            db,
            start_utc=start_utc,
            end_utc=end_utc,
            period_from=period_from,
            period_to=period_to,
        )
    )


def _commission_rate_percent(raw: str | None) -> Decimal | None:
    """美团 commissionRate 原值: "300"=3%, "10"=0.1%;也兼容 "3%"。"""
    if raw is None:
        return None
    s = str(raw).strip()
    if not s:
        return None
    try:
        if s.endswith("%"):
            return Decimal(s[:-1])
        val = Decimal(s)
    except (InvalidOperation, ValueError):
        return None
    return val / Decimal("100")


def _jd_valid_order(order: CpsOrder) -> bool:
    code = str(order.jd_valid_code).strip() if order.jd_valid_code is not None else ""
    return bool(code and code not in JD_CPS_INVALID_CODES and code not in JD_CPS_UNPAID_CODES)


def dashboard_overview(
    db: Session, *, date_from: date | None = None, date_to: date | None = None
) -> dict:
    today_start = _beijing_today_start_utc()
    period_from, period_to = _normalize_period(date_from, date_to)
    start_utc, end_utc, start_local, end_local = _period_bounds(period_from, period_to)

    def _count(model, *conds) -> int:
        stmt = select(func.count(model.id))
        if conds:
            stmt = stmt.where(*conds)
        return int(db.execute(stmt).scalar_one())

    def _sum(col, *conds) -> int:
        stmt = select(func.coalesce(func.sum(col), 0))
        if conds:
            stmt = stmt.where(*conds)
        return int(db.execute(stmt).scalar_one())

    def _user_id_set(stmt) -> set[int]:
        return {int(v) for v in db.execute(stmt).scalars().all() if v is not None}

    # ===== 用户 =====
    by_status = dict(
        db.execute(select(User.status, func.count(User.id)).group_by(User.status)).all()
    )

    # ===== 提现状态分布 =====
    wd_by_status = dict(
        db.execute(
            select(WithdrawOrder.status, func.count(WithdrawOrder.id)).group_by(
                WithdrawOrder.status
            )
        ).all()
    )

    # ===== 比价 =====
    comparison_total = _count(ComparisonRecord)
    comparison_success = _count(ComparisonRecord, ComparisonRecord.status == "success")
    success_rate = round(comparison_success / comparison_total, 4) if comparison_total else 0.0
    period_comparison_conds = (
        ComparisonRecord.created_at >= start_local,
        ComparisonRecord.created_at < end_local,
    )
    period_comparison_stats = db.execute(
        select(
            func.count(ComparisonRecord.id),
            func.coalesce(
                func.sum(
                    case(
                        (ComparisonRecord.status.in_(("success", "failed")), 1),
                        else_=0,
                    )
                ),
                0,
            ),
            func.coalesce(
                func.sum(
                    case((ComparisonRecord.status == "cancelled", 1), else_=0)
                ),
                0,
            ),
            func.coalesce(
                func.sum(case((ComparisonRecord.status == "success", 1), else_=0)),
                0,
            ),
            func.coalesce(func.sum(ComparisonRecord.llm_cost_yuan), 0.0),
        ).where(*period_comparison_conds)
    ).one()
    period_comparison_total = int(period_comparison_stats[0])
    period_comparison_completed = int(period_comparison_stats[1])
    period_comparison_cancelled = int(period_comparison_stats[2])
    period_comparison_success = int(period_comparison_stats[3])
    period_comparison_token_cost_yuan = float(period_comparison_stats[4])
    period_comparison_success_denominator = (
        period_comparison_total - period_comparison_cancelled
    )
    period_comparison_success_rate = (
        round(
            period_comparison_success / period_comparison_success_denominator,
            4,
        )
        if period_comparison_success_denominator > 0
        else None
    )
    period_saved_positive_count = _count(
        ComparisonRecord,
        *period_comparison_conds,
        ComparisonRecord.status == "success",
        ComparisonRecord.saved_amount_cents > 0,
    )
    period_saved_positive_sum = _sum(
        ComparisonRecord.saved_amount_cents,
        *period_comparison_conds,
        ComparisonRecord.status == "success",
        ComparisonRecord.saved_amount_cents > 0,
    )
    period_avg_saved_cents = (
        round(period_saved_positive_sum / period_saved_positive_count)
        if period_saved_positive_count
        else None
    )
    period_avg_duration_ms = db.execute(
        select(func.avg(ComparisonRecord.total_ms)).where(
            *period_comparison_conds,
            ComparisonRecord.total_ms.is_not(None),
            ComparisonRecord.total_ms > 0,
        )
    ).scalar_one()
    period_avg_duration_ms = (
        round(float(period_avg_duration_ms))
        if period_avg_duration_ms is not None
        else None
    )
    completed_duration_conds = (
        *period_comparison_conds,
        ComparisonRecord.status.in_(("success", "failed")),
        ComparisonRecord.total_ms.is_not(None),
    )
    if db.bind is not None and db.bind.dialect.name == "postgresql":
        period_median_duration_ms, period_p95_duration_ms = db.execute(
            select(
                func.percentile_cont(0.5).within_group(ComparisonRecord.total_ms),
                func.percentile_cont(0.95).within_group(ComparisonRecord.total_ms),
            ).where(*completed_duration_conds)
        ).one()
        period_median_duration_ms = (
            int(
                Decimal(str(period_median_duration_ms)).quantize(
                    Decimal("1"), rounding=ROUND_HALF_UP
                )
            )
            if period_median_duration_ms is not None
            else None
        )
        period_p95_duration_ms = (
            int(
                Decimal(str(period_p95_duration_ms)).quantize(
                    Decimal("1"), rounding=ROUND_HALF_UP
                )
            )
            if period_p95_duration_ms is not None
            else None
        )
    else:
        # SQLite 测试环境没有 percentile_cont；仅回退读取耗时单列，不加载完整记录。
        completed_durations = list(
            db.execute(
                select(ComparisonRecord.total_ms)
                .where(*completed_duration_conds)
                .order_by(ComparisonRecord.total_ms)
            ).scalars()
        )
        period_median_duration_ms = _duration_percentile(completed_durations, 0.5)
        period_p95_duration_ms = _duration_percentile(completed_durations, 0.95)

    ordered_exists = (
        select(SavingsRecord.id)
        .where(
            SavingsRecord.user_id == ComparisonRecord.user_id,
            SavingsRecord.source == "compare",
            SavingsRecord.shop_name.is_not(None),
            SavingsRecord.shop_name == ComparisonRecord.store_name,
        )
        .exists()
    )
    period_ordered_count = _count(
        ComparisonRecord,
        *period_comparison_conds,
        ComparisonRecord.store_name.is_not(None),
        ordered_exists,
    )

    # ===== 日期窗口用户 =====
    period_new_user_ids = _user_id_set(
        select(User.id).where(User.created_at >= start_utc, User.created_at < end_utc)
    )
    period_active_user_ids = _period_active_user_ids(
        db,
        start_utc=start_utc,
        end_utc=end_utc,
        period_from=period_from,
        period_to=period_to,
    )
    # 留存口径(2026-07-05 产品改):次日留存——窗口内每天 D,取 **D-1 日(前日)新增**用户,
    # 统计其 D 日活跃(登录/开始比价/开始领券)比例,逐日累加。默认窗口=昨日单天,即
    # 「前日新增用户的昨日留存」。原口径(窗口内新增∩窗口内活跃)在单日窗口下≈100% 无意义
    # (注册即登录,当天新增必然当天活跃)。逐日 cohort 在下方 trend 循环内顺带累计。
    retention_cohort_total = 0
    retention_retained_total = 0
    trend_points: list[dict] = []
    for cur_date in _date_range(period_from, period_to):
        day_start_utc, day_end_utc, day_start_local, day_end_local = _period_bounds(
            cur_date, cur_date
        )
        daily_comparison_conds = (
            ComparisonRecord.created_at >= day_start_local,
            ComparisonRecord.created_at < day_end_local,
        )
        daily_active_user_ids = _period_active_user_ids(
            db,
            start_utc=day_start_utc,
            end_utc=day_end_utc,
            period_from=cur_date,
            period_to=cur_date,
        )
        # 次日留存:cohort = 前一日(D-1)新增用户,留存 = 其中当日(D)活跃者(口径见上)。
        cohort_ids = _user_id_set(
            select(User.id).where(
                User.created_at >= day_start_utc - timedelta(days=1),
                User.created_at < day_end_utc - timedelta(days=1),
            )
        )
        retention_cohort_total += len(cohort_ids)
        retention_retained_total += len(cohort_ids & daily_active_user_ids)
        trend_points.append(
            {
                "date": cur_date,
                "active_users": len(daily_active_user_ids),
                "new_users": _count(
                    User,
                    User.created_at >= day_start_utc,
                    User.created_at < day_end_utc,
                ),
                "comparisons": _count(ComparisonRecord, *daily_comparison_conds),
            }
        )
    period_retention_rate = (
        round(retention_retained_total / retention_cohort_total, 4)
        if retention_cohort_total
        else None
    )

    period_coin_conds = (
        CoinTransaction.created_at >= start_local,
        CoinTransaction.created_at < end_local,
        CoinTransaction.amount > 0,
    )
    period_reward_video_coin_total = _sum(
        CoinTransaction.amount,
        *period_coin_conds,
        CoinTransaction.biz_type.in_(REWARD_VIDEO_BIZ_TYPES),
    )
    period_feed_ad_coin_total = _sum(
        CoinTransaction.amount,
        *period_coin_conds,
        CoinTransaction.biz_type == "feed_ad_reward",
    )
    period_signin_coin_total = _sum(
        CoinTransaction.amount,
        *period_coin_conds,
        CoinTransaction.biz_type == "signin",
    )
    period_signin_boost_coin_total = _sum(
        CoinTransaction.amount,
        *period_coin_conds,
        CoinTransaction.biz_type == "signin_boost",
    )
    period_task_coin_total = _sum(
        CoinTransaction.amount,
        *period_coin_conds,
        CoinTransaction.biz_type.like("task_%"),
    )
    # 领券/比价奖励金币 = biz_type 桶(历史空,兜底)+ 该场景信息流广告实发金币
    # (ad_feed_reward_record.feed_scene,granted;reward_date 是北京日期串,与 period 同自然日窗口)。
    period_coupon_reward_coin_total = _sum(
        CoinTransaction.amount,
        *period_coin_conds,
        CoinTransaction.biz_type.in_(COUPON_REWARD_BIZ_TYPES),
    ) + _sum(
        AdFeedRewardRecord.coin,
        AdFeedRewardRecord.status == "granted",
        AdFeedRewardRecord.feed_scene == "coupon",
        AdFeedRewardRecord.reward_date >= period_from.isoformat(),
        AdFeedRewardRecord.reward_date <= period_to.isoformat(),
    )
    period_comparison_reward_coin_total = _sum(
        CoinTransaction.amount,
        *period_coin_conds,
        CoinTransaction.biz_type.in_(COMPARISON_REWARD_BIZ_TYPES),
    ) + _sum(
        AdFeedRewardRecord.coin,
        AdFeedRewardRecord.status == "granted",
        AdFeedRewardRecord.feed_scene == "comparison",
        AdFeedRewardRecord.reward_date >= period_from.isoformat(),
        AdFeedRewardRecord.reward_date <= period_to.isoformat(),
    )
    period_regular_task_coin_total = _sum(
        CoinTransaction.amount,
        *period_coin_conds,
        or_(
            CoinTransaction.biz_type.in_(REGULAR_TASK_EXACT_BIZ_TYPES),
            CoinTransaction.biz_type.like(r"task\_%", escape="\\"),
        ),
    )
    period_cps_orders = list(
        db.execute(
            select(CpsOrder).where(
                CpsOrder.pay_time >= start_utc,
                CpsOrder.pay_time < end_utc,
            )
        ).scalars()
    )
    period_meituan_orders = [
        o for o in period_cps_orders if (o.platform or "meituan") == "meituan"
    ]
    period_meituan_valid_orders = [
        o for o in period_meituan_orders if o.mt_status not in MEITUAN_CPS_INVALID_STATUSES
    ]
    period_jd_orders = [o for o in period_cps_orders if o.platform == "jd"]
    period_jd_valid_orders = [o for o in period_jd_orders if _jd_valid_order(o)]
    period_jd_invalid_orders = [
        o for o in period_jd_orders if o.jd_valid_code and not _jd_valid_order(o)
    ]
    period_meituan_hit_count = 0
    period_meituan_miss_count = 0
    period_meituan_unknown_rate_count = 0
    for order in period_meituan_valid_orders:
        rate = _commission_rate_percent(order.commission_rate)
        if rate is None:
            period_meituan_unknown_rate_count += 1
        elif rate < Decimal("1"):
            period_meituan_miss_count += 1
        else:
            period_meituan_hit_count += 1
    period_meituan_hit_denominator = period_meituan_hit_count + period_meituan_miss_count
    period_meituan_hit_rate = (
        round(period_meituan_hit_count / period_meituan_hit_denominator, 4)
        if period_meituan_hit_denominator
        else None
    )

    # ===== 领券核心数据(2026-07-05 产品新增)=====
    # 数据源:coupon_session(一次领券一行,started_date 北京自然日)+ coupon_claim_record
    # (一券/点位一天一条终态,claim_date 北京自然日)。点位与 session 不按 trace_id 关联——
    # record_claims 更新路径不覆盖 trace_id(同设备同券同日重跑归第一次的 trace),按
    # (device_id, 自然日) 桶关联才可靠;同桶多次发起共享同一份点位终态。
    period_coupon_sessions = db.execute(
        select(
            CouponSession.device_id,
            CouponSession.started_date,
            CouponSession.status,
            CouponSession.elapsed_ms,
        ).where(
            CouponSession.started_date >= period_from,
            CouponSession.started_date <= period_to,
            # 只统计正式环境,同「领券数据」页默认口径(防 debug 包调试数据串台;
            # 命中 ix_coupon_session_date_env)。点位表无 app_env 列,但点位指标只经
            # 下方 prod session 触达的 (device, 日) 桶进入统计,随之收敛到 prod。
            CouponSession.app_env == "prod",
        )
    ).all()
    coupon_started = len(period_coupon_sessions)
    coupon_completed_elapsed = sorted(
        s.elapsed_ms
        for s in period_coupon_sessions
        if s.status == "completed" and s.elapsed_ms is not None
    )
    # 点位桶:(device, 日) → (点位总数, 成功点位数)。成功口径与「我的」页累计领券一致
    # (sum_claimed_count,2026-06-15 产品定):success + already_claimed(已领过=持有券)都算成功。
    point_buckets: dict[tuple[str, date], tuple[int, int]] = {
        (dev, d): (int(total), int(succ or 0))
        for dev, d, total, succ in db.execute(
            select(
                CouponClaimRecord.device_id,
                CouponClaimRecord.claim_date,
                func.count(),
                func.sum(
                    case(
                        (CouponClaimRecord.status.in_(("success", "already_claimed")), 1),
                        else_=0,
                    )
                ),
            )
            .where(
                CouponClaimRecord.claim_date >= period_from,
                # 上界放宽一天:跨零点场次(23:5x 发起)的点位 claim_date 落在发起日+1,
                # 桶只经下方 session 触达的键参与计数,放宽不会引入无关数据。
                CouponClaimRecord.claim_date <= period_to + timedelta(days=1),
            )
            .group_by(CouponClaimRecord.device_id, CouponClaimRecord.claim_date)
        ).all()
    }
    # 全部领成功的次数:completed 且其 (device, 日) 桶内点位全部成功(桶为空不算)。
    coupon_all_success = 0
    completed_bucket_totals: list[int] = []
    session_bucket_keys: set[tuple[str, date]] = set()
    for s in period_coupon_sessions:
        key = (s.device_id, s.started_date)
        if key not in point_buckets:
            # 跨零点回退:发起日桶不存在(点位终态全部落在次日)时取 (device, 发起日+1)。
            # 仅在发起日桶完全缺失时回退,避免抢占该设备次日 session 自己的桶。
            next_key = (s.device_id, s.started_date + timedelta(days=1))
            if next_key in point_buckets:
                key = next_key
        bucket = point_buckets.get(key)
        if bucket is not None:
            session_bucket_keys.add(key)
        if s.status != "completed" or bucket is None:
            continue
        total, succ = bucket
        completed_bucket_totals.append(total)
        if total > 0 and succ == total:
            coupon_all_success += 1
    # 每次发起的应领点位数:取「完成过的领券」实际点位数的众数(done 帧会给所有点位终态,
    # 完成场的点位数=当前配置的全量点位数;数据自校准,配置改点位数无需改代码)。本期无完成场
    # 时给不出,点位成功率置空。
    coupon_points_per_session = (
        Counter(completed_bucket_totals).most_common(1)[0][0]
        if completed_bucket_totals
        else None
    )
    # 成功点位数:本期 session 触达过的 (device, 日) 桶内成功点位之和(桶级去重,同桶重试不重复计)。
    coupon_point_success = sum(point_buckets[k][1] for k in session_bucket_keys)
    # 点位成功率 = 成功点位数 / (发起数 × 应领点位数):中途退出未跑到的点位不产生记录,
    # 但发起数×点位数把它们计入分母 → 视为失败,符合产品口径;重试会拉低该率(分母按次数计)。
    coupon_point_success_rate = (
        round(
            min(1.0, coupon_point_success / (coupon_started * coupon_points_per_session)), 4
        )
        if coupon_started and coupon_points_per_session
        else None
    )

    return {
        "users": {
            "total": _count(User),
            "active": by_status.get("active", 0),
            "disabled": by_status.get("disabled", 0),
            "deleted": by_status.get("deleted", 0),
            "new_today": _count(User, User.created_at >= today_start),
            "dau": today_dau(db),
        },
        "coins": {
            # 累计发放金币(coin_transaction 里所有 amount>0 之和;负数是兑换/扣减不计)
            "granted_total": _sum(CoinTransaction.amount, CoinTransaction.amount > 0),
            "reward_video_coin_total": _sum(
                CoinTransaction.amount,
                CoinTransaction.amount > 0,
                CoinTransaction.biz_type.in_(REWARD_VIDEO_BIZ_TYPES),
            ),
            "reward_video_watch_count": _count(
                AdRewardRecord,
                AdRewardRecord.reward_scene == "reward_video",
                AdRewardRecord.status == "granted",
            ),
            "feed_ad_coin_total": _sum(
                CoinTransaction.amount,
                CoinTransaction.amount > 0,
                CoinTransaction.biz_type == "feed_ad_reward",
            ),
            "feed_ad_watch_count": _count(
                AdFeedRewardRecord,
                AdFeedRewardRecord.status == "granted",
            ),
            "signin_coin_total": _sum(
                CoinTransaction.amount,
                CoinTransaction.amount > 0,
                CoinTransaction.biz_type == "signin",
            ),
            "signin_count": _count(SigninRecord),
            "signin_boost_coin_total": _sum(
                CoinTransaction.amount,
                CoinTransaction.amount > 0,
                CoinTransaction.biz_type == "signin_boost",
            ),
            # 签到膨胀 2026-07 已下线,signin_boost_record 表随之 drop。这两项保留为**历史口径**
            # (钱是真发过的,账要能查回)。次数改数金币流水:一次膨胀 = 一笔 signin_boost 流水,
            # 与原来数 signin_boost_record 行数等价。
            "signin_boost_watch_count": _count(
                CoinTransaction,
                CoinTransaction.biz_type == "signin_boost",
                CoinTransaction.amount > 0,
            ),
        },
        "cash": {
            "withdraw_success_cents": _sum(
                WithdrawOrder.amount_cents, WithdrawOrder.status == "success"
            ),
            "withdraw_pending_count": wd_by_status.get("pending", 0),
            "withdraw_success_count": wd_by_status.get("success", 0),
            "withdraw_failed_count": wd_by_status.get("failed", 0),
        },
        "comparison": {
            "total": comparison_total,
            "success": comparison_success,
            "success_rate": success_rate,
        },
        "period": {
            "date_from": period_from,
            "date_to": period_to,
            "users": {
                "new": len(period_new_user_ids),
                "active": len(period_active_user_ids),
                "retained_new_users": retention_retained_total,
                "retention_cohort": retention_cohort_total,
                "retention_rate": period_retention_rate,
                "retention_note": (
                    "次日留存:窗口内每天取前一日新增用户,统计其当日活跃"
                    "(登录/开始比价/开始领券,按用户去重)比例,逐日累加;"
                    "默认窗口=昨日,即前日新增用户的昨日留存"
                ),
            },
            "comparison": {
                "total": period_comparison_total,
                "completed": period_comparison_completed,
                "cancelled": period_comparison_cancelled,
                "success": period_comparison_success,
                "success_rate": period_comparison_success_rate,
                "ordered": period_ordered_count,
                "average_duration_ms": period_avg_duration_ms,
                "median_duration_ms": period_median_duration_ms,
                "p95_duration_ms": period_p95_duration_ms,
                "average_saved_cents": period_avg_saved_cents,
                "token_cost_total_yuan": period_comparison_token_cost_yuan,
            },
            "coupon": {
                "started": coupon_started,
                "all_success": coupon_all_success,
                "success_rate": (
                    round(coupon_all_success / coupon_started, 4) if coupon_started else None
                ),
                "point_success": coupon_point_success,
                "points_per_session": coupon_points_per_session,
                "point_success_rate": coupon_point_success_rate,
                "median_elapsed_ms": _percentile(coupon_completed_elapsed, 50),
            },
            "coins": {
                "granted_total": _sum(CoinTransaction.amount, *period_coin_conds),
                "reward_video_coin_total": period_reward_video_coin_total,
                "feed_ad_coin_total": period_feed_ad_coin_total,
                "signin_coin_total": period_signin_coin_total,
                "signin_boost_coin_total": period_signin_boost_coin_total,
                "task_coin_total": period_task_coin_total,
                "coupon_reward_coin_total": period_coupon_reward_coin_total,
                "comparison_reward_coin_total": period_comparison_reward_coin_total,
                "regular_task_coin_total": period_regular_task_coin_total,
            },
            "cash": {
                "withdraw_success_cents": _sum(
                    WithdrawOrder.amount_cents,
                    WithdrawOrder.status == "success",
                    WithdrawOrder.created_at >= start_local,
                    WithdrawOrder.created_at < end_local,
                ),
            },
            "trend": trend_points,
        },
        "feedback": {
            "new": _count(Feedback, Feedback.status.in_(("pending", "new"))),
        },
        "cps": {
            "available": True,
            "note": "美团/JD CPS 读 cps_order 对账订单;淘宝佣金暂空",
            "meituan_order_count": len(period_meituan_valid_orders),
            "meituan_commission_cents": sum(
                o.commission_cents or 0 for o in period_meituan_valid_orders
            ),
            "meituan_hit_count": period_meituan_hit_count,
            "meituan_miss_count": period_meituan_miss_count,
            "meituan_unknown_rate_count": period_meituan_unknown_rate_count,
            "meituan_hit_rate": period_meituan_hit_rate,
            "jd_order_count": len(period_jd_valid_orders),
            # 数据大盘京东 CPS 只看实际佣金,不再用预估佣金兜底。
            "jd_commission_cents": sum(
                o.actual_commission_cents or 0 for o in period_jd_valid_orders
            ),
            "jd_actual_commission_cents": sum(
                o.actual_commission_cents or 0 for o in period_jd_valid_orders
            ),
            "jd_estimated_commission_cents": sum(
                o.estimated_commission_cents or 0 for o in period_jd_valid_orders
            ),
            "jd_invalid_count": len(period_jd_invalid_orders),
        },
    }