Agent/examples/auto_put_ad_mini/tools/posterior_collector.py

"""
后验数据采集 — 为进化奠基（V4 Layer 5）

职责：
  - 给定决策日期，回采 t+1d / t+3d / t+7d 的广告 ROI 和消耗
  - 输出"决策 ↔ 后验"配对 CSV
  - 自动给决策打 outcome_label（good/bad/neutral/executed）
  - 回填 snapshot.json 的 posterior_* 字段

⚠️ 本轮（auto_put_ad_mini_v4）不在主流程调用；下阶段反馈环对接。
   - 接口签名固化：collect_posterior_data(decision_date, posterior_days, update_snapshot)
   - 数据 schema 固化：posterior.csv 字段顺序见下文 POSTERIOR_CSV_COLUMNS
   - 决策 outcome_label 标注规则固化：见 _label_outcome()

──────────────────────────────────────────────────────────────────────
后续接入路径（下阶段反馈环可直接对接）：
  1. 每日定时任务：collect_posterior_data(yesterday, 1) → 滚动回采
  2. T+3 / T+7 定时任务：再回采一次（数据稳定后）
  3. 累计 100~500 条带后验的样本后：
     - 训练 Few-shot 示例池（outcome=good 的高质量决策）
     - 用回归校准 DECAY_WEIGHTS 等规则权重
     - 生成决策质量监控仪表盘（精确率/召回率）
──────────────────────────────────────────────────────────────────────
"""

import json
import logging
import sys
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any, Dict, List, Optional

import pandas as pd

from agent.tools import tool
from agent.tools.models import ToolContext, ToolResult

_MINI_DIR = Path(__file__).resolve().parent.parent
if str(_MINI_DIR) not in sys.path:
    sys.path.insert(0, str(_MINI_DIR))

logger = logging.getLogger(__name__)


# ═══════════════════════════════════════════
# 数据 Schema 定义（固化，下阶段不要改）
# ═══════════════════════════════════════════

POSTERIOR_CSV_COLUMNS: List[str] = [
    # 决策标识
    "ad_id",
    "decision_date",
    "action",
    "dimension",
    "source",                     # rule_pause / llm_main / llm_override
    # 决策时刻的快照
    "decision_roi_7d",            # 决策时的 动态ROI_7日均值
    "decision_cost_7d_avg",
    "decision_bid_amount",
    # 后验回采（多窗口）
    "posterior_t1d_roi",
    "posterior_t1d_cost",
    "posterior_t3d_roi",
    "posterior_t3d_cost",
    "posterior_t7d_roi",
    "posterior_t7d_cost",
    # 计算出的相对变化
    "roi_change_t3d_pct",         # (posterior_t3d - decision) / decision
    "cost_change_t3d_pct",
    # 标签 + 评分溯源（JSON 串，便于按信号统计）
    "outcome_label",              # good / bad / neutral / executed
    "signal_scores_json",
    "reasoning_chain_json",
    "rule_alignment",             # agree / override
]


# ═══════════════════════════════════════════
# 内部工具函数
# ═══════════════════════════════════════════


def _safe_pct_change(new_v: Optional[float], old_v: Optional[float]) -> Optional[float]:
    """安全计算百分比变化，None / 0 兜底。"""
    if new_v is None or old_v is None:
        return None
    try:
        if abs(old_v) < 1e-9:
            return None
        return round((new_v - old_v) / old_v, 4)
    except Exception:
        return None


def _label_outcome(
    decision_action: str,
    decision_roi: Optional[float],
    decision_cost: Optional[float],
    posterior_t3d_roi: Optional[float],
    posterior_t3d_cost: Optional[float],
) -> str:
    """
    后验标注规则（固化，下阶段不要改）：

      - action = pause:
          posterior_t3d_cost ≈ 0  → executed（说明确实暂停了）
          posterior_t3d_cost > 0  → bad（暂停决策没生效或被复活）
      - action = bid_down:
          ROI 提升   → good
          ROI 持平   → neutral
          ROI 下降   → bad
      - action = bid_up:
          cost 提升 + ROI 不掉 → good
          cost 不变           → neutral
          ROI 大跌            → bad
      - action = hold / needs_llm: 默认 neutral
    """
    if posterior_t3d_roi is None and posterior_t3d_cost is None:
        return "unknown"

    act = (decision_action or "").lower()

    if act == "pause":
        if posterior_t3d_cost is None:
            return "unknown"
        return "executed" if posterior_t3d_cost < 1.0 else "bad"

    if act == "bid_down":
        roi_chg = _safe_pct_change(posterior_t3d_roi, decision_roi)
        if roi_chg is None:
            return "unknown"
        if roi_chg > 0.05:
            return "good"
        if roi_chg < -0.05:
            return "bad"
        return "neutral"

    if act == "bid_up":
        cost_chg = _safe_pct_change(posterior_t3d_cost, decision_cost)
        roi_chg = _safe_pct_change(posterior_t3d_roi, decision_roi)
        if cost_chg is None:
            return "unknown"
        if roi_chg is not None and roi_chg < -0.10:
            return "bad"
        if cost_chg > 0.10:
            return "good"
        return "neutral"

    return "neutral"


def _date_offset(date_str: str, days: int) -> str:
    """YYYYMMDD + N 天 → YYYYMMDD"""
    dt = datetime.strptime(date_str, "%Y%m%d") + timedelta(days=days)
    return dt.strftime("%Y%m%d")


def _load_metrics_for_date(date_str: str) -> Optional[pd.DataFrame]:
    """
    尝试加载某一天的 metrics 快照（用于回采当天 ROI / cost）。

    优先策略（下阶段可扩展）：
      1. outputs/metrics_history/{date}.csv
      2. outputs/metrics_temp.csv（如果它的 end_date 等于 date）
      3. None（无数据）
    """
    history_csv = _MINI_DIR / "outputs" / "metrics_history" / f"{date_str}.csv"
    if history_csv.exists():
        try:
            return pd.read_csv(history_csv)
        except Exception as e:
            logger.warning("加载历史 metrics 失败 %s: %s", history_csv, e)

    # fallback: temp（仅当日期吻合）
    temp_csv = _MINI_DIR / "outputs" / "metrics_temp.csv"
    if temp_csv.exists():
        try:
            df = pd.read_csv(temp_csv)
            if "end_date" in df.columns:
                if str(df["end_date"].iloc[0]) == date_str:
                    return df
        except Exception:
            pass

    return None


def _lookup_posterior(
    df: Optional[pd.DataFrame],
    ad_id: int,
) -> Dict[str, Optional[float]]:
    """从 metrics DataFrame 提取某广告的 ROI / cost。"""
    if df is None or df.empty:
        return {"roi": None, "cost": None}
    try:
        row = df[df["ad_id"] == ad_id]
        if row.empty:
            return {"roi": None, "cost": None}
        r = row.iloc[0]
        roi_val = r.get("动态ROI_7日均值")
        cost_val = r.get("cost_7d_avg")
        return {
            "roi": float(roi_val) if pd.notna(roi_val) else None,
            "cost": float(cost_val) if pd.notna(cost_val) else None,
        }
    except Exception as e:
        logger.warning("查找广告 %s 后验数据失败: %s", ad_id, e)
        return {"roi": None, "cost": None}


# ═══════════════════════════════════════════
# 主接口：后验采集（固化签名）
# ═══════════════════════════════════════════


@tool(description="V4 后验数据采集（决策 → 执行后效果配对）— 本轮预留接口，主流程不调用")
async def collect_posterior_data(
    ctx: ToolContext = None,
    decision_date: str = "",
    posterior_days: int = 7,
    update_snapshot: bool = True,
) -> ToolResult:
    """
    后验数据采集 — 决策 ↔ 执行后效果配对。

    Args:
        ctx: 工具上下文
        decision_date: 决策日期 YYYYMMDD（即 snapshot.json 的目录名）
        posterior_days: 采集多少天后验（默认 7）
        update_snapshot: 是否回写 snapshot.json 的 posterior_* 字段（默认 True）

    流程：
      1. 加载 outputs/decisions_history/{decision_date}/snapshot.json
      2. 对每条决策的 ad_id，分别加载 t+1d / t+3d / t+7d 的 metrics
      3. 计算 posterior_t1d/t3d/t7d 的 roi/cost
      4. 给决策打 outcome_label（_label_outcome 规则）
      5. 写 outputs/decisions_history/{decision_date}/posterior.csv
      6. （可选）回写 snapshot.json 的 posterior_* 和 outcome_label

    Returns:
      ToolResult 含统计 summary：good/bad/neutral/executed/unknown 的分布

    ⚠️ 本轮不在 prompts/system.prompt 中引用；仅供下阶段反馈环对接。
    """
    try:
        snap_dir = _MINI_DIR / "outputs" / "decisions_history" / decision_date
        snap_path = snap_dir / "snapshot.json"
        if not snap_path.exists():
            return ToolResult(
                title="collect_posterior_data 失败",
                output=f"决策快照不存在: {snap_path}",
            )

        snapshot = json.loads(snap_path.read_text(encoding="utf-8"))
        decisions = snapshot.get("decisions", [])
        if not decisions:
            return ToolResult(
                title="collect_posterior_data",
                output=f"快照 {decision_date} 无决策记录",
            )

        # 预加载 t+1d / t+3d / t+7d 的 metrics
        t1d_date = _date_offset(decision_date, 1)
        t3d_date = _date_offset(decision_date, 3)
        t7d_date = _date_offset(decision_date, 7)

        df_t1d = _load_metrics_for_date(t1d_date)
        df_t3d = _load_metrics_for_date(t3d_date)
        df_t7d = _load_metrics_for_date(t7d_date) if posterior_days >= 7 else None

        rows: List[Dict[str, Any]] = []
        outcome_dist: Dict[str, int] = {}

        for d in decisions:
            ad_id_raw = d.get("ad_id")
            try:
                ad_id_int = int(ad_id_raw)
            except (ValueError, TypeError):
                continue

            # 决策快照本体
            input_signals = d.get("input_signals") or {}
            decision_roi = input_signals.get("动态ROI_7日均值")
            decision_cost = input_signals.get("cost_7d_avg")
            decision_bid = input_signals.get("bid_amount")

            # 各窗口后验
            p1 = _lookup_posterior(df_t1d, ad_id_int)
            p3 = _lookup_posterior(df_t3d, ad_id_int)
            p7 = _lookup_posterior(df_t7d, ad_id_int) if df_t7d is not None else {"roi": None, "cost": None}

            # 标签
            label = _label_outcome(
                decision_action=d.get("action"),
                decision_roi=decision_roi,
                decision_cost=decision_cost,
                posterior_t3d_roi=p3["roi"],
                posterior_t3d_cost=p3["cost"],
            )
            outcome_dist[label] = outcome_dist.get(label, 0) + 1

            # 写一行
            rows.append({
                "ad_id": ad_id_int,
                "decision_date": decision_date,
                "action": d.get("action"),
                "dimension": d.get("dimension"),
                "source": d.get("source"),
                "decision_roi_7d": decision_roi,
                "decision_cost_7d_avg": decision_cost,
                "decision_bid_amount": decision_bid,
                "posterior_t1d_roi": p1["roi"],
                "posterior_t1d_cost": p1["cost"],
                "posterior_t3d_roi": p3["roi"],
                "posterior_t3d_cost": p3["cost"],
                "posterior_t7d_roi": p7["roi"],
                "posterior_t7d_cost": p7["cost"],
                "roi_change_t3d_pct": _safe_pct_change(p3["roi"], decision_roi),
                "cost_change_t3d_pct": _safe_pct_change(p3["cost"], decision_cost),
                "outcome_label": label,
                "signal_scores_json": json.dumps(d.get("signal_scores") or {}, ensure_ascii=False),
                "reasoning_chain_json": json.dumps(d.get("reasoning_chain") or [], ensure_ascii=False),
                "rule_alignment": d.get("rule_alignment"),
            })

            # 回写 snapshot
            if update_snapshot:
                d["posterior_t1d"] = p1
                d["posterior_t3d"] = p3
                d["posterior_t7d"] = p7
                d["outcome_label"] = label

        # 输出 posterior.csv
        out_csv = snap_dir / "posterior.csv"
        df_out = pd.DataFrame(rows, columns=POSTERIOR_CSV_COLUMNS)
        df_out.to_csv(out_csv, index=False, encoding="utf-8-sig")

        # 回写 snapshot
        if update_snapshot:
            snapshot["metadata"] = snapshot.get("metadata", {})
            snapshot["metadata"]["posterior_collected_at"] = datetime.now().isoformat()
            snapshot["metadata"]["posterior_days"] = posterior_days
            snap_path.write_text(json.dumps(snapshot, ensure_ascii=False, indent=2), encoding="utf-8")

        # 汇总
        lines = [
            f"决策日期: {decision_date}",
            f"采集窗口: t+1d={t1d_date}, t+3d={t3d_date}, t+7d={t7d_date}",
            f"决策总数: {len(decisions)}",
            f"配对 CSV: {out_csv}",
            "",
            "outcome_label 分布:",
        ]
        for k, v in sorted(outcome_dist.items(), key=lambda x: -x[1]):
            lines.append(f"  {k}: {v}")

        return ToolResult(
            title=f"后验采集完成（{decision_date}）",
            output="\n".join(lines),
            metadata={
                "decision_date": decision_date,
                "posterior_csv": str(out_csv),
                "snapshot_updated": update_snapshot,
                "decision_count": len(decisions),
                "outcome_distribution": outcome_dist,
                "posterior_dates": {"t1d": t1d_date, "t3d": t3d_date, "t7d": t7d_date},
            },
        )

    except Exception as e:
        logger.error("collect_posterior_data 失败: %s", e, exc_info=True)
        return ToolResult(title="collect_posterior_data 失败", output=str(e))