Agent/examples/auto_put_ad_mini/roi_control/reporting.py

"""调控结果 Excel 报告生成。"""

from __future__ import annotations

from pathlib import Path
from typing import Dict, Mapping, Sequence

import numpy as np
import pandas as pd
from openpyxl import Workbook
from openpyxl.worksheet.datavalidation import DataValidation
from openpyxl.formatting.rule import ColorScaleRule
from openpyxl.styles import Alignment, Font, PatternFill
from openpyxl.utils import get_column_letter
from .fission_multiplier import DISPLAY_MULTIPLIER_COLUMN
from .metrics import ENTITY_GZH, ENTITY_QIWEI, ENTITY_SELF


HEADER_FILL = PatternFill("solid", fgColor="1F4E78")
HEADER_FONT = Font(color="FFFFFF", bold=True)
STOP_FILL = PatternFill("solid", fgColor="F4CCCC")
UP_FILL = PatternFill("solid", fgColor="D9EAD3")
ADJUST_FILL = PatternFill("solid", fgColor="FFF2CC")
APPROVAL_FILL = PatternFill("solid", fgColor="FFD966")
APPROVAL_HEADER_FILL = PatternFill("solid", fgColor="BF9000")
REPORT_VERSION = "roi_report_v5"
REPORT_RUN_SUFFIX = "r5"


BASE_COLUMNS: Dict[str, Sequence[str]] = {
    "小程序投流": (
        "渠道",
        "代理名称",
        "账号id",
        "广告id",
        "广告名称",
        "包名",
        "广告优化目标",
        "创意id",
        "广告age",
        "日均首层UV",
        "首层效率收入",
        "T0裂变效率收入",
        "LTV预测效率收入",
        "日均成本",
        DISPLAY_MULTIPLIER_COLUMN,
        "预测ROI",
        "动作",
        "关停线",
        "审批选择",
        "执行状态",
        "执行结果",
    ),
    "公众号即转": (
        "渠道",
        "合作方名",
        "公众号名",
        "日均首层UV",
        "首层效率收入",
        "T0裂变效率收入",
        "LTV预测效率收入",
        "日均成本",
        DISPLAY_MULTIPLIER_COLUMN,
        "预测ROI",
        "动作",
        "关停线",
    ),
    "企微群合作": (
        "渠道",
        "合作方名",
        "日均首层UV",
        "首层效率收入",
        "T0裂变效率收入",
        "LTV预测效率收入",
        "日均成本",
        DISPLAY_MULTIPLIER_COLUMN,
        "传播裂变参数状态",
        "调控参与状态",
        "预测ROI",
        "动作",
        "关停线",
    ),
}

ENTITY_TO_SHEET = {
    ENTITY_SELF: "小程序投流",
    ENTITY_GZH: "公众号即转",
    ENTITY_QIWEI: "企微群合作",
}

FREEZE_PANES = {
    "小程序投流": "F2",
    "公众号即转": "D2",
    "企微群合作": "C2",
}


def _sheet_frame(candidates: pd.DataFrame, sheet_name: str) -> pd.DataFrame:
    entity_type = next(
        key for key, value in ENTITY_TO_SHEET.items() if value == sheet_name
    )
    subset = candidates[candidates["entity_type"].eq(entity_type)].copy()
    subset = subset.rename(columns={"channel": "渠道"})
    coverage_source = (
        subset["覆盖天数"]
        if "覆盖天数" in subset
        else pd.Series(3, index=subset.index)
    )
    coverage_days = pd.to_numeric(
        coverage_source, errors="coerce"
    ).replace(0, np.nan)
    subset["日均首层UV"] = pd.to_numeric(
        subset.get("日均首层UV"), errors="coerce"
    ).round().astype("Int64")
    subset["首层效率收入"] = (
        pd.to_numeric(subset.get("效率收入"), errors="coerce")
        / coverage_days
    )
    subset["T0裂变效率收入"] = (
        pd.to_numeric(subset.get("T0实际裂变收入"), errors="coerce")
        / coverage_days
    )
    subset["LTV预测效率收入"] = (
        pd.to_numeric(
            subset.get("预测全链路效率收入"), errors="coerce"
        )
        / coverage_days
    )
    subset["日均成本"] = (
        pd.to_numeric(subset.get("成本"), errors="coerce") / coverage_days
    )
    subset["预测ROI"] = subset["ROI"]
    subset["关停线"] = subset["t_stop"]
    subset["扩量线"] = subset["t_up"]
    visible_columns = list(BASE_COLUMNS[sheet_name])
    for column in visible_columns:
        if column not in subset:
            subset[column] = ""
    hidden_columns = [
        column for column in subset.columns if column not in visible_columns
    ]
    ordered = visible_columns + hidden_columns
    subset = subset[ordered]
    if not subset.empty:
        subset = subset.sort_values("预测ROI", ascending=True)
    return subset


def _write_dataframe(ws, frame: pd.DataFrame) -> None:
    ws.append(list(frame.columns))
    for values in frame.itertuples(index=False, name=None):
        ws.append(
            [
                None if isinstance(value, float) and np.isnan(value) else value
                for value in values
            ]
        )


def _format_sheet(ws, sheet_name: str) -> None:
    max_column = max(ws.max_column, 1)
    max_row = max(ws.max_row, 1)
    ws.freeze_panes = FREEZE_PANES[sheet_name]
    ws.auto_filter.ref = f"A1:{get_column_letter(max_column)}{max_row}"
    ws.row_dimensions[1].height = 28

    headers = {}
    for cell in ws[1]:
        cell.fill = HEADER_FILL
        cell.font = HEADER_FONT
        cell.alignment = Alignment(horizontal="center", vertical="center")
        headers[cell.value] = cell.column

    approval_column = headers.get("审批选择")
    if approval_column:
        approval_header = ws.cell(1, approval_column)
        approval_header.fill = APPROVAL_HEADER_FILL
        approval_header.font = HEADER_FONT
        validation = DataValidation(
            type="list",
            formula1='"批准,拒绝"',
            allow_blank=True,
        )
        validation.error = "请选择批准或拒绝"
        validation.errorTitle = "审批值无效"
        ws.add_data_validation(validation)
        for row in range(2, max_row + 1):
            cell = ws.cell(row, approval_column)
            if cell.value != "不可执行":
                cell.fill = APPROVAL_FILL
                validation.add(cell)

    for column_index in range(1, max_column + 1):
        header = ws.cell(1, column_index).value or ""
        values = [str(ws.cell(row, column_index).value or "") for row in range(1, max_row + 1)]
        width = min(max(max(map(len, values)), len(str(header))) + 2, 42)
        ws.column_dimensions[get_column_letter(column_index)].width = max(width, 11)

        if "ROI" in str(header) or "裂变率" in str(header) or "百分位" in str(header):
            for row in range(2, max_row + 1):
                ws.cell(row, column_index).number_format = "0.000"
        elif "成本" in str(header) or header in {
            "首层效率收入",
            "T0裂变效率收入",
            "LTV预测效率收入",
            "效率收入",
            "T0实际裂变收入",
            "实际全链路效率收入",
            "预测T1-T15裂变收入",
            "预测T0-T15裂变收入",
            "预测全链路效率收入",
        }:
            for row in range(2, max_row + 1):
                ws.cell(row, column_index).number_format = "#,##0.00"
        elif "UV" in str(header) or "裂变数" in str(header):
            for row in range(2, max_row + 1):
                ws.cell(row, column_index).number_format = "#,##0"

        if "ROI" in str(header) and max_row >= 2:
            letter = get_column_letter(column_index)
            ws.conditional_formatting.add(
                f"{letter}2:{letter}{max_row}",
                ColorScaleRule(
                    start_type="min",
                    start_color="F8696B",
                    mid_type="percentile",
                    mid_value=50,
                    mid_color="FFEB84",
                    end_type="max",
                    end_color="63BE7B",
                ),
            )

        if column_index > len(BASE_COLUMNS[sheet_name]):
            ws.column_dimensions[get_column_letter(column_index)].hidden = True

    action_column = headers.get("动作")
    if action_column:
        for row in range(2, max_row + 1):
            action = ws.cell(row, action_column).value
            fill = {
                "关停": STOP_FILL,
                "扩量": UP_FILL,
                "调整封面&落地页视频": ADJUST_FILL,
            }.get(action)
            if fill:
                ws.cell(row, action_column).fill = fill

def _write_summary(
    workbook: Workbook,
    thresholds: pd.DataFrame,
    candidates: pd.DataFrame,
    expected_dates: Sequence[str],
    rule_config: Mapping[str, object] | None,
) -> None:
    ws = workbook.create_sheet("运行摘要")
    self_min_uv = (rule_config or {}).get("self_min_avg_uv", 200)
    partner_min_uv = (rule_config or {}).get("partner_min_avg_uv", 200)
    min_daily_cost = (rule_config or {}).get("min_daily_cost", 100)
    rows = [
        ("统计窗口", f"{expected_dates[0]} 至 {expected_dates[-1]}"),
        ("首层口径", "usersharedepth='0'"),
        ("UV口径", "COUNT(DISTINCT mid)"),
        (
            "ROI口径",
            "T0裂变收入为实际值；仅预测T1-T15增量。预测总收入=首层实际效率收入+T0实际裂变收入+T0实际裂变收入×(传播裂变系数-对T0裂变-1)",
        ),
        (
            "企微口径",
            "企微按渠道/合作方保留展示，临时参考系数为2.5，待补算正式参数；不进入阈值样本池和调控。",
        ),
        ("阈值口径", "小程序和公众号合格实体进入同一全局池：三日聚合ROI的P20=关停线、P80=扩量线；企微不参与"),
        (
            "阈值样本条件",
            f"连续3天数据完整；小程序日均首层UV>{self_min_uv:g}；"
            f"公众号日均首层UV>{partner_min_uv:g}；连续3天每天成本>"
            f"{min_daily_cost:g}；三日成本>0且预测ROI有效；企微仅展示、不进入样本池",
        ),
        ("动作优先级", "关停 > 扩量 > 调整封面&落地页视频"),
        ("审批方式", "在小程序投流表黄色【审批选择】列逐行选择批准或拒绝；批准即为最终确认并自动执行"),
    ]
    for label, value in rows:
        ws.append([label, value])

    if rule_config:
        ws.append([])
        ws.append(["规则配置", "值"])
        for key in (
            "self_stop_min_age",
            "self_up_min_age",
            "self_min_avg_uv",
            "partner_min_avg_uv",
            "min_daily_cost",
            "stop_quantile",
            "up_quantile",
            "scale_ratio",
            "scale_cooldown_days",
            "max_base_ratio",
        ):
            ws.append([key, rule_config.get(key)])
        fission_multiplier = rule_config.get("fission_multiplier")
        if isinstance(fission_multiplier, Mapping):
            ws.append([])
            ws.append(["传播裂变参数", "值"])
            for key in (
                "version",
                "cohort_date",
                "observation_end_date",
                "horizon_days",
                "miniapp_exact_available_rows",
                "gzh_exact_available_rows",
                "miniapp_channel_multiplier",
                "gzh_channel_multiplier",
            ):
                ws.append([key, fission_multiplier.get(key)])

    ws.append([])
    display_thresholds = thresholds.rename(
        columns={"t_stop": "关停线", "t_up": "扩量线"}
    )
    threshold_columns = [
        "统计窗口",
        "关停线",
        "扩量线",
        "阈值样本数",
        "小程序样本数",
        "公众号样本数",
        "企微参考实体数",
    ]
    ws.append(threshold_columns)
    for values in display_thresholds[threshold_columns].itertuples(
        index=False,
        name=None,
    ):
        ws.append(list(values))

    ws.append([])
    ws.append(["渠道", "建议动作数"])
    for entity_type, sheet_name in ENTITY_TO_SHEET.items():
        count = int(
            (
                candidates["entity_type"].eq(entity_type)
                & candidates["动作"].ne("")
            ).sum()
        )
        ws.append([sheet_name, count])

    ws.column_dimensions["A"].width = 22
    ws.column_dimensions["B"].width = 92
    for row in ws.iter_rows():
        if row[0].value in {"统计窗口", "渠道"}:
            for cell in row:
                cell.font = Font(bold=True)
    ws.freeze_panes = "A2"


def write_workbook(
    candidates: pd.DataFrame,
    thresholds: pd.DataFrame,
    expected_dates: Sequence[str],
    output_path: Path,
    rule_config: Mapping[str, object] | None = None,
    fission_match_summary: pd.DataFrame | None = None,
) -> Path:
    output_path.parent.mkdir(parents=True, exist_ok=True)
    workbook = Workbook()
    workbook.remove(workbook.active)

    for sheet_name in BASE_COLUMNS:
        ws = workbook.create_sheet(sheet_name)
        frame = _sheet_frame(candidates, sheet_name)
        _write_dataframe(ws, frame)
        _format_sheet(ws, sheet_name)

    if fission_match_summary is not None:
        ws = workbook.create_sheet("传播裂变系数匹配")
        _write_dataframe(ws, fission_match_summary)
        ws.freeze_panes = "A2"
        ws.auto_filter.ref = ws.dimensions
        for cell in ws[1]:
            cell.fill = HEADER_FILL
            cell.font = HEADER_FONT
        for column_index in range(1, ws.max_column + 1):
            header = str(ws.cell(1, column_index).value or "")
            values = [
                str(ws.cell(row, column_index).value or "")
                for row in range(1, ws.max_row + 1)
            ]
            width = min(max(max(map(len, values)), len(header)) + 2, 42)
            ws.column_dimensions[get_column_letter(column_index)].width = max(
                width, 11
            )
            if header == "匹配率":
                for row in range(2, ws.max_row + 1):
                    ws.cell(row, column_index).number_format = "0.00%"

    _write_summary(
        workbook,
        thresholds,
        candidates,
        expected_dates,
        rule_config,
    )

    workbook.save(output_path)
    return output_path