Agent/examples/auto_put_ad_mini/tools/ad_decision.py

"""
广告决策引擎 — auto_put_ad_mini V3

V3 三维度决策引擎：
  维度 1（最高优先级）: ROI 过低 → 关停
  维度 2: 长期无消耗 → 关停
  维度 3: 广告衰退 → 关停

设计：
  - DecisionDimension 基类 + 优先级注册
  - 第一个命中的维度决定动作，后续不再评估
  - 不满 7 天的广告不参与决策
  - 所有阈值通过参数传入（来自 SKILL）
"""

import logging
import sys
from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime, timedelta
from pathlib import Path
from typing import Dict, List, Optional, Tuple

import numpy as np
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))

from config import (
    AUDIENCE_TIER_PATTERNS,
    BID_ADJUSTMENT_ENABLED,
    BID_DOWN_ROI_FACTOR,
    BID_UP_ROI_FACTOR,
    BID_UP_LOW_SPEND_FACTOR,
    BID_CHANGE_MIN_PCT,
    BID_CHANGE_MAX_PCT,
    BID_FLOOR_YUAN,
    BID_CEILING_YUAN,
    COLD_START_DAYS,       # ≤3天：冷启动期（极度保护）
    EARLY_GROWTH_DAYS,     # 4-7天：早期成长期（可提价）
    AD_AGE_MATURE,         # >7天：成熟期（全面调控）
    HIGH_BURN_AGE_THRESHOLD,
    HIGH_BURN_COST_THRESHOLD,
    ROI_LOW_FACTOR,
)

logger = logging.getLogger(__name__)

# ═══════════════════════════════════════════
# 策略参数动态加载（阈值不写死在代码中）
# ═══════════════════════════════════════════

STRATEGY_PARAMS_FILE = _MINI_DIR / "strategy_params.json"

def _load_strategy_params():
    """从json文件加载策略参数，如不存在则使用config.py默认值"""
    import json

    if STRATEGY_PARAMS_FILE.exists():
        try:
            with open(STRATEGY_PARAMS_FILE) as f:
                data = json.load(f)
                return data.get("params", {})
        except Exception as e:
            logger.warning(f"加载strategy_params.json失败，使用config.py默认值: {e}")

    # 使用config.py默认值
    return {
        "ROI_LOW_FACTOR": ROI_LOW_FACTOR,
        "BID_DOWN_ROI_FACTOR": BID_DOWN_ROI_FACTOR,
        "BID_UP_ROI_FACTOR": BID_UP_ROI_FACTOR,
        "BID_UP_LOW_SPEND_FACTOR": BID_UP_LOW_SPEND_FACTOR,
    }


# ═══════════════════════════════════════════
# 决策动作类型（扩展支持）
# ═══════════════════════════════════════════

VALID_ACTIONS = [
    "pause",            # 关停
    "bid_down",         # 降价
    "bid_up",           # 提价
    "hold",             # 保持
    "creative_adjust",  # 调整素材方向（需人工执行）
    "observe",          # 观察等待（数据不稳定或接近阈值）
    "scale_up",         # 扩量：建议新增广告/创意（需人工执行）
]

# ═══════════════════════════════════════════
# 辅助函数
# ═══════════════════════════════════════════


def _extract_audience_tier(ad_name: str) -> str:
    """从广告名称提取人群包 R 层级（保留自 V2）。"""
    if not ad_name:
        return "default"
    for tier, patterns in AUDIENCE_TIER_PATTERNS:
        for pat in patterns:
            if pat.lower() in str(ad_name).lower():
                return tier
    return "default"


def _calculate_ad_age_days(create_time) -> Optional[int]:
    """计算广告从创建到现在的天数。"""
    if pd.isna(create_time):
        return None
    try:
        if isinstance(create_time, str):
            ct = datetime.strptime(create_time[:19], "%Y-%m-%d %H:%M:%S")
        else:
            ct = pd.Timestamp(create_time).to_pydatetime()
        return (datetime.now() - ct).days
    except Exception:
        return None


# ═══════════════════════════════════════════
# 决策结果数据类
# ═══════════════════════════════════════════


@dataclass
class Decision:
    """单个广告的决策结果。"""
    ad_id: int
    action: str  # "pause" / "bid_down" / "bid_up" / "hold" / "creative_adjust" / "observe"
    dimension: str  # "ROI过低" / "长期无消耗" / "广告衰退" / "ROI偏低-降价" / "高ROI低量-提价" / "保持"
    reason: str  # 详细原因
    recommended_change_pct: Optional[float] = None  # +0.05 = 提价5%, -0.08 = 降价8%
    current_bid: Optional[float] = None             # 当前出价(元)
    recommended_bid: Optional[float] = None          # 建议出价(元)


# ═══════════════════════════════════════════
# 决策维度基类（可扩展）
# ═══════════════════════════════════════════


class DecisionDimension(ABC):
    """决策维度基类。"""

    def __init__(self, priority: int):
        self.priority = priority

    @abstractmethod
    def evaluate(self, row: pd.Series, context: Dict) -> Optional[Decision]:
        """
        评估单个广告是否命中该维度。

        Args:
            row: 广告数据行（包含所有指标）
            context: 全局上下文（如全体均值、阈值参数）

        Returns:
            Decision 对象（命中）或 None（不命中）
        """
        pass


# ═══════════════════════════════════════════
# 维度 1: ROI 过低
# ═══════════════════════════════════════════


class ROITooLowDimension(DecisionDimension):
    """维度 1: 动态ROI_7日均值 < 全体均值 × 0.5 → 关停。"""

    def __init__(self):
        super().__init__(priority=1)

    def evaluate(self, row: pd.Series, context: Dict) -> Optional[Decision]:
        ad_age_days = row.get("ad_age_days")
        cost_7d_avg = row.get("cost_7d_avg", 0)
        f_roi_7d = row.get("动态ROI_7日均值")   # 决策参考值
        f_roi_mean_all = context.get("动态ROI_mean_all")

        min_ad_age = context.get("min_ad_age_days", 7)
        min_daily_cost = context.get("min_daily_cost", 100)
        roi_low_factor = context.get("roi_low_factor", 0.5)

        # 前置条件
        if ad_age_days is None or ad_age_days < min_ad_age:
            return None
        if cost_7d_avg < min_daily_cost:
            return None
        if pd.isna(f_roi_7d) or pd.isna(f_roi_mean_all):
            return None

        # 判断
        threshold = f_roi_mean_all * roi_low_factor
        if f_roi_7d < threshold:
            return Decision(
                ad_id=int(row["ad_id"]),
                action="pause",
                dimension="ROI过低",
                reason=f"动态ROI_7日均值={f_roi_7d:.4f} < 全体均值×{roi_low_factor}={threshold:.4f}"
            )

        return None


# ═══════════════════════════════════════════
# 维度 2: 长期无消耗
# ═══════════════════════════════════════════


class NoSpendDimension(DecisionDimension):
    """维度 2: 7日消耗均值 < 10元 → 关停。"""

    def __init__(self):
        super().__init__(priority=2)

    def evaluate(self, row: pd.Series, context: Dict) -> Optional[Decision]:
        ad_age_days = row.get("ad_age_days")
        cost_7d_avg = row.get("cost_7d_avg", 0)

        min_ad_age = context.get("min_ad_age_days", 7)
        no_spend_threshold = context.get("no_spend_threshold", 10)

        # 前置条件
        if ad_age_days is None or ad_age_days < min_ad_age:
            return None

        # 判断
        if cost_7d_avg < no_spend_threshold:
            return Decision(
                ad_id=int(row["ad_id"]),
                action="pause",
                dimension="长期无消耗",
                reason=f"7日消耗均值={cost_7d_avg:.2f}元 < {no_spend_threshold}元"
            )

        return None


# ═══════════════════════════════════════════
# 维度 3: 广告衰退
# ═══════════════════════════════════════════


class AdDecayDimension(DecisionDimension):
    """
    维度 3: 广告衰退 → 关停。

    条件：
      - 30 天内曾连续稳定消耗（>100元/天）
      - 近 7 天已提价或换创意
      - 但消耗仍低（<100元）
    """

    def __init__(self):
        super().__init__(priority=3)

    def evaluate(self, row: pd.Series, context: Dict) -> Optional[Decision]:
        stable_spend_days_30d = row.get("stable_spend_days_30d", 0)
        cost_7d_avg = row.get("cost_7d_avg", 0)
        bid_increased_7d = row.get("bid_increased_7d", False)
        creative_changed_7d = row.get("creative_changed_7d", False)

        stable_threshold = context.get("stable_spend_threshold", 100)

        # 判断
        if stable_spend_days_30d >= 7:  # 曾稳定消耗
            if cost_7d_avg < stable_threshold:  # 现在消耗低
                if bid_increased_7d or creative_changed_7d:  # 已干预
                    reason_parts = []
                    if bid_increased_7d:
                        reason_parts.append("已提价")
                    if creative_changed_7d:
                        reason_parts.append("已换创意")
                    reason = f"30天内曾稳定消耗{stable_spend_days_30d}天，近7天{'+'.join(reason_parts)}，但消耗仍低({cost_7d_avg:.2f}元)"

                    return Decision(
                        ad_id=int(row["ad_id"]),
                        action="pause",
                        dimension="广告衰退",
                        reason=reason
                    )

        return None


# ═══════════════════════════════════════════
# 维度 4: 出价偏高 — 降价
# ═══════════════════════════════════════════


class BidDownDimension(DecisionDimension):
    """
    维度 4: ROI 偏低但未达关停线 → 降价。

    触发条件：
      - ROI 在 均值×0.5 ~ 均值×0.8 之间（偏低但非极低）
      - 日消耗 ≥ 100 元（数据有统计意义）
      - 非冷启动期（> {COLD_START_DAYS} 天，即≥4天）
      - 有出价数据（bid_amount > 0）

    降幅计算：
      ROI 距离关停线越近 → 降幅越大（最大 -10%）
      ROI 刚好低于正常线 → 小幅降价（-3%~-5%）
      公式：change_pct = -3% - 7% × (1 - (ROI - hard_stop) / (normal_line - hard_stop))
    """

    def __init__(self):
        super().__init__(priority=4)

    def evaluate(self, row: pd.Series, context: Dict) -> Optional[Decision]:
        if not context.get("bid_adjustment_enabled", BID_ADJUSTMENT_ENABLED):
            return None

        ad_age_days = row.get("ad_age_days")
        cost_7d_avg = row.get("cost_7d_avg", 0)
        f_roi_7d = row.get("动态ROI_7日均值")
        f_roi_mean_all = context.get("动态ROI_mean_all")
        bid_amount = row.get("bid_amount", 0)  # 元

        cold_start = context.get("cold_start_days", COLD_START_DAYS)
        cautious = context.get("cautious_days", CAUTIOUS_DAYS)
        min_daily_cost = context.get("min_daily_cost", 100)
        roi_low_factor = context.get("roi_low_factor", 0.5)
        bid_down_factor = context.get("bid_down_roi_factor", BID_DOWN_ROI_FACTOR)

        # 前置条件
        if ad_age_days is None or ad_age_days < cold_start:
            return None
        if cost_7d_avg < min_daily_cost:
            return None
        if pd.isna(f_roi_7d) or pd.isna(f_roi_mean_all) or f_roi_mean_all <= 0:
            return None
        if not bid_amount or bid_amount <= 0:
            return None

        hard_stop_line = f_roi_mean_all * roi_low_factor
        normal_line = f_roi_mean_all * bid_down_factor

        # ROI 必须在 hard_stop ~ normal_line 之间
        if f_roi_7d >= normal_line or f_roi_7d < hard_stop_line:
            return None

        # 计算降幅
        range_width = normal_line - hard_stop_line
        if range_width <= 0:
            return None
        ratio = 1 - (f_roi_7d - hard_stop_line) / range_width
        change_pct = -(BID_CHANGE_MIN_PCT + (BID_CHANGE_MAX_PCT - BID_CHANGE_MIN_PCT) * ratio)

        # 谨慎期（4-7天）限制最大降幅 5%
        if ad_age_days < cautious:
            change_pct = max(change_pct, -0.05)

        # 计算建议出价
        new_bid = bid_amount * (1 + change_pct)
        new_bid = max(new_bid, BID_FLOOR_YUAN)
        new_bid = min(new_bid, BID_CEILING_YUAN)
        new_bid = round(new_bid, 2)

        # 实际调幅重算（边界钳位后）
        actual_pct = (new_bid - bid_amount) / bid_amount if bid_amount > 0 else 0

        return Decision(
            ad_id=int(row["ad_id"]),
            action="bid_down",
            dimension="ROI偏低-降价",
            reason=(
                f"动态ROI_7日均值={f_roi_7d:.4f}，"
                f"在关停线{hard_stop_line:.4f}~正常线{normal_line:.4f}之间，"
                f"当前出价{bid_amount:.2f}元，建议降{abs(actual_pct)*100:.1f}%至{new_bid:.2f}元"
            ),
            recommended_change_pct=round(actual_pct, 4),
            current_bid=round(bid_amount, 2),
            recommended_bid=new_bid,
        )


# ═══════════════════════════════════════════
# 维度 5: 高ROI低量 — 提价
# ═══════════════════════════════════════════


class BidUpDimension(DecisionDimension):
    """
    维度 5: ROI 远超均值但消耗不足 → 提价放量。

    触发条件：
      - ROI > 均值×1.2
      - 7日均消耗 < 全体中位数×0.5（消耗不足）
      - 非冷启动期（> COLD_START_DAYS 天）
      - 有出价数据（bid_amount > 0）

    提幅计算：
      提幅 = min(10%, (ROI/均值 - 1.2) × 20%)
      最小提幅 3%
    """

    def __init__(self):
        super().__init__(priority=5)

    def evaluate(self, row: pd.Series, context: Dict) -> Optional[Decision]:
        if not context.get("bid_adjustment_enabled", BID_ADJUSTMENT_ENABLED):
            return None

        ad_age_days = row.get("ad_age_days")
        cost_7d_avg = row.get("cost_7d_avg", 0)
        f_roi_7d = row.get("动态ROI_7日均值")
        f_roi_mean_all = context.get("动态ROI_mean_all")
        cost_median = context.get("cost_7d_avg_median", 0)
        bid_amount = row.get("bid_amount", 0)  # 元

        cold_start = context.get("cold_start_days", COLD_START_DAYS)
        bid_up_factor = context.get("bid_up_roi_factor", BID_UP_ROI_FACTOR)
        low_spend_factor = context.get("bid_up_low_spend_factor", BID_UP_LOW_SPEND_FACTOR)

        # 前置条件
        if ad_age_days is None or ad_age_days < cold_start:
            return None
        if pd.isna(f_roi_7d) or pd.isna(f_roi_mean_all) or f_roi_mean_all <= 0:
            return None
        if not bid_amount or bid_amount <= 0:
            return None

        # ROI 必须远超均值
        roi_threshold = f_roi_mean_all * bid_up_factor
        if f_roi_7d <= roi_threshold:
            return None

        # 消耗必须不足
        spend_threshold = cost_median * low_spend_factor
        if cost_7d_avg >= spend_threshold and spend_threshold > 0:
            return None

        # 计算提幅
        roi_excess = f_roi_7d / f_roi_mean_all - bid_up_factor
        change_pct = min(BID_CHANGE_MAX_PCT, roi_excess * 0.20)
        change_pct = max(change_pct, BID_CHANGE_MIN_PCT)

        # 计算建议出价
        new_bid = bid_amount * (1 + change_pct)
        new_bid = max(new_bid, BID_FLOOR_YUAN)
        new_bid = min(new_bid, BID_CEILING_YUAN)
        new_bid = round(new_bid, 2)

        # 实际调幅重算
        actual_pct = (new_bid - bid_amount) / bid_amount if bid_amount > 0 else 0

        return Decision(
            ad_id=int(row["ad_id"]),
            action="bid_up",
            dimension="高ROI低量-提价",
            reason=(
                f"动态ROI_7日均值={f_roi_7d:.4f} > 均值{f_roi_mean_all:.4f}×{bid_up_factor}={roi_threshold:.4f}，"
                f"但7日均消耗仅{cost_7d_avg:.2f}元 < 中位数{cost_median:.2f}×{low_spend_factor}={spend_threshold:.2f}，"
                f"当前出价{bid_amount:.2f}元，建议提{actual_pct*100:.1f}%至{new_bid:.2f}元"
            ),
            recommended_change_pct=round(actual_pct, 4),
            current_bid=round(bid_amount, 2),
            recommended_bid=new_bid,
        )


# ═══════════════════════════════════════════
# 决策引擎
# ═══════════════════════════════════════════


def _run_decision_engine(
    df: pd.DataFrame,
    context: Dict
) -> pd.DataFrame:
    """
    运行三维度决策引擎。

    流程：
      1. 注册所有维度（按优先级排序）
      2. 对每个广告，按优先级评估维度
      3. 第一个命中的维度决定动作
      4. 不满 7 天的广告标记为"投放不足7日"

    输入：
      df: 广告级指标表（包含 动态ROI, cost_7d_avg, ad_age_days 等）
      context: 全局上下文（阈值参数、全体均值）

    输出：
      添加 action, dimension, reason 列的 DataFrame
    """
    # 注册维度（含出价调整维度）
    dimensions = [
        ROITooLowDimension(),
        NoSpendDimension(),
        AdDecayDimension(),
        BidDownDimension(),
        BidUpDimension(),
    ]
    dimensions.sort(key=lambda d: d.priority)

    decisions = []

    for _, row in df.iterrows():
        ad_age_days = row.get("ad_age_days")
        min_ad_age = context.get("min_ad_age_days", 7)

        # 不满 7 天的广告
        if ad_age_days is None or ad_age_days < min_ad_age:
            decisions.append(Decision(
                ad_id=int(row["ad_id"]),
                action="hold",
                dimension="保持",
                reason=f"投放不足{min_ad_age}日（当前{ad_age_days}日）"
            ))
            continue

        # 按优先级评估维度
        decision = None
        for dim in dimensions:
            decision = dim.evaluate(row, context)
            if decision is not None:
                break

        # 无维度命中 → 保持
        if decision is None:
            decision = Decision(
                ad_id=int(row["ad_id"]),
                action="hold",
                dimension="保持",
                reason="各项指标正常"
            )

        decisions.append(decision)

    # 转换为 DataFrame
    decision_df = pd.DataFrame([
        {
            "ad_id": d.ad_id,
            "action": d.action,
            "dimension": d.dimension,
            "reason": d.reason,
            "recommended_change_pct": d.recommended_change_pct,
            "current_bid": d.current_bid,
            "recommended_bid": d.recommended_bid,
        }
        for d in decisions
    ])

    # 合并回原 DataFrame
    df = df.merge(decision_df, on="ad_id", how="left")

    return df


# ═══════════════════════════════════════════
# 衰退检测辅助函数
# ═══════════════════════════════════════════


def _detect_decay_signals(
    ad_ids: List[int],
    raw_dir: Path,
    ad_status_dir: Path,
    end_date: str
) -> pd.DataFrame:
    """
    检测广告衰退信号（提价、换创意）。

    输入：
      ad_ids: 需要检测的广告 ID 列表
      raw_dir: 创意级原始 CSV 目录
      ad_status_dir: 广告状态 CSV 目录
      end_date: 结束日期（YYYYMMDD）

    输出：
      DataFrame，列：ad_id, bid_increased_7d, creative_changed_7d, stable_spend_days_30d
    """
    end_dt = datetime.strptime(end_date, "%Y%m%d")

    # 加载近 14 天创意数据（用于检测创意变化）
    creative_dfs = []
    for i in range(14):
        date = (end_dt - timedelta(days=i)).strftime("%Y%m%d")
        csv_path = raw_dir / f"creative_{date}.csv"
        if csv_path.exists():
            df = pd.read_csv(csv_path)
            df["date"] = date
            creative_dfs.append(df)

    if not creative_dfs:
        logger.warning("无创意数据，无法检测衰退信号")
        return pd.DataFrame(columns=["ad_id", "bid_increased_7d", "creative_changed_7d", "stable_spend_days_30d"])

    creative_df = pd.concat(creative_dfs, ignore_index=True)
    creative_df = creative_df[creative_df["ad_id"].isin(ad_ids)]

    # 加载近 14 天广告状态（用于检测提价）
    status_dfs = []
    for i in range(14):
        date = (end_dt - timedelta(days=i)).strftime("%Y%m%d")
        csv_path = ad_status_dir / f"ad_status_{date}.csv"
        if csv_path.exists():
            df = pd.read_csv(csv_path)
            df["date"] = date
            status_dfs.append(df)

    if not status_dfs:
        logger.warning("无广告状态数据，无法检测提价")
        status_df = pd.DataFrame()
    else:
        status_df = pd.concat(status_dfs, ignore_index=True)
        status_df = status_df[status_df["ad_id"].isin(ad_ids)]

    # 检测创意变化（近 7 天 vs 前 7-14 天）
    recent_7d_start = (end_dt - timedelta(days=6)).strftime("%Y%m%d")
    prior_7d_start = (end_dt - timedelta(days=13)).strftime("%Y%m%d")
    prior_7d_end = (end_dt - timedelta(days=7)).strftime("%Y%m%d")

    recent_creatives = (
        creative_df[creative_df["date"] >= recent_7d_start]
        .groupby("ad_id")["creative_id"]
        .apply(set)
    )

    prior_creatives = (
        creative_df[
            (creative_df["date"] >= prior_7d_start) & (creative_df["date"] <= prior_7d_end)
        ]
        .groupby("ad_id")["creative_id"]
        .apply(set)
    )

    creative_changed = {}
    for ad_id in ad_ids:
        recent_set = recent_creatives.get(ad_id, set())
        prior_set = prior_creatives.get(ad_id, set())
        creative_changed[ad_id] = (recent_set != prior_set) and len(recent_set) > 0 and len(prior_set) > 0

    # 检测提价（近 7 天最大出价 > 前 7-14 天最大出价）
    bid_increased = {}
    if not status_df.empty:
        recent_bids = (
            status_df[status_df["date"] >= recent_7d_start]
            .groupby("ad_id")["bid_amount"]
            .max()
        )

        prior_bids = (
            status_df[
                (status_df["date"] >= prior_7d_start) & (status_df["date"] <= prior_7d_end)
            ]
            .groupby("ad_id")["bid_amount"]
            .max()
        )

        for ad_id in ad_ids:
            recent_bid = recent_bids.get(ad_id, 0)
            prior_bid = prior_bids.get(ad_id, 0)
            bid_increased[ad_id] = recent_bid > prior_bid
    else:
        bid_increased = {ad_id: False for ad_id in ad_ids}

    # 计算 30 天稳定消耗天数（加载 30 天创意数据）
    creative_30d_dfs = []
    for i in range(30):
        date = (end_dt - timedelta(days=i)).strftime("%Y%m%d")
        csv_path = raw_dir / f"creative_{date}.csv"
        if csv_path.exists():
            df = pd.read_csv(csv_path)
            df["date"] = date
            creative_30d_dfs.append(df)

    if creative_30d_dfs:
        creative_30d_df = pd.concat(creative_30d_dfs, ignore_index=True)
        creative_30d_df = creative_30d_df[creative_30d_df["ad_id"].isin(ad_ids)]

        # 按 ad_id + date 聚合消耗
        daily_cost = (
            creative_30d_df.groupby(["ad_id", "date"])["cost"]
            .sum()
            .reset_index()
        )

        stable_days = {}
        for ad_id in ad_ids:
            ad_cost = daily_cost[daily_cost["ad_id"] == ad_id]
            stable_days[ad_id] = (ad_cost["cost"] >= 100).sum()
    else:
        stable_days = {ad_id: 0 for ad_id in ad_ids}

    # 组装结果（不含 stable_spend_days_30d，该值已在 metrics CSV 中）
    result = pd.DataFrame({
        "ad_id": ad_ids,
        "bid_increased_7d": [bid_increased.get(ad_id, False) for ad_id in ad_ids],
        "creative_changed_7d": [creative_changed.get(ad_id, False) for ad_id in ad_ids],
    })

    return result


# ═══════════════════════════════════════════
# V3 工具：三维度决策
# ═══════════════════════════════════════════


@tool(description="V3 三维度决策引擎：ROI过低 / 长期无消耗 / 广告衰退")
async def analyze_ads(
    ctx: ToolContext,
    metrics_csv: str,
    end_date: str = "yesterday",
    min_ad_age_days: int = 7,
    min_daily_cost: float = 100.0,
    roi_low_factor: float = 0.5,
    no_spend_threshold: float = 10.0,
    stable_spend_threshold: float = 100.0,
) -> ToolResult:
    """
    V3 三维度决策引擎。

    Args:
        ctx: 工具上下文
        metrics_csv: ROI 指标 CSV 路径（calculate_roi_metrics 输出）
        end_date: 结束日期（YYYYMMDD 或 "yesterday"）
        min_ad_age_days: 最小广告年龄（天）
        min_daily_cost: 最小日消耗（元）
        roi_low_factor: ROI 过低因子（< 全体均值 × factor）
        no_spend_threshold: 长期无消耗阈值（元）
        stable_spend_threshold: 稳定消耗阈值（元/天）

    Returns:
        ToolResult，包含决策结果 DataFrame
    """
    try:
        # 加载指标数据
        df = pd.read_csv(metrics_csv)

        if df.empty:
            return ToolResult(
                title="决策引擎",
                output="指标数据为空，无法决策",
            )

        # 解析日期
        if end_date == "yesterday":
            end_date = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d")

        # 计算广告年龄
        df["ad_age_days"] = df["create_time"].apply(_calculate_ad_age_days)

        # 提取人群包层级
        df["audience_tier"] = df["ad_name"].apply(_extract_audience_tier)

        # 检测衰退信号
        raw_dir = _MINI_DIR / "outputs" / "raw"
        ad_status_dir = _MINI_DIR / "outputs" / "ad_status"

        decay_signals = _detect_decay_signals(
            ad_ids=df["ad_id"].tolist(),
            raw_dir=raw_dir,
            ad_status_dir=ad_status_dir,
            end_date=end_date
        )

        df = df.merge(decay_signals, on="ad_id", how="left")

        # 填充缺失值
        df["bid_increased_7d"] = df["bid_increased_7d"].fillna(False)
        df["creative_changed_7d"] = df["creative_changed_7d"].fillna(False)
        df["stable_spend_days_30d"] = df["stable_spend_days_30d"].fillna(0)

        # 计算全体 动态ROI_7日均值 的均值（决策基准线）
        f_roi_mean_all = df["动态ROI_7日均值"].mean()

        # 构建上下文
        context = {
            "动态ROI_mean_all": f_roi_mean_all,
            "min_ad_age_days": min_ad_age_days,
            "min_daily_cost": min_daily_cost,
            "roi_low_factor": roi_low_factor,
            "no_spend_threshold": no_spend_threshold,
            "stable_spend_threshold": stable_spend_threshold,
        }

        # 运行决策引擎
        df = _run_decision_engine(df, context)

        # 统计
        total_ads = len(df)
        pause_ads = (df["action"] == "pause").sum()
        hold_ads = (df["action"] == "hold").sum()

        dimension_counts = df["dimension"].value_counts().to_dict()

        output_lines = [
            f"决策完成（共 {total_ads} 个广告）",
            f"  - 关停: {pause_ads} 个",
            f"  - 保持: {hold_ads} 个",
            "",
            "维度分布:",
        ]

        for dim, count in dimension_counts.items():
            output_lines.append(f"  - {dim}: {count} 个")

        output_lines.extend([
            "",
            f"全体 动态ROI_7日均值 均值: {f_roi_mean_all:.4f}",
            f"ROI 过低阈值: {f_roi_mean_all * roi_low_factor:.4f}",
        ])

        # 保存决策结果（临时 CSV，供 generate_report 使用）
        decision_csv = _MINI_DIR / "outputs" / "decision_temp.csv"
        decision_csv.parent.mkdir(parents=True, exist_ok=True)
        df.to_csv(decision_csv, index=False)

        return ToolResult(
            title=f"决策引擎（{total_ads}个广告）",
            output="\n".join(output_lines),
            metadata={
                "total_ads": total_ads,
                "pause_ads": pause_ads,
                "hold_ads": hold_ads,
                "dimension_counts": dimension_counts,
                "动态ROI_mean_all": f_roi_mean_all,
                "decision_csv": str(decision_csv),
            },
        )

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


# ═══════════════════════════════════════════
# 智能引擎工具 1：整理待评估广告数据
# ═══════════════════════════════════════════


@tool(description="智能引擎：整理需要关注的广告数据，供LLM推理决策")
async def get_ads_for_review(
    ctx: ToolContext,
    metrics_csv: str = "",
    end_date: str = "yesterday",
    roi_review_factor: float = 0.8,
    min_spend_for_class_a: float = 10.0,
) -> ToolResult:
    """
    不做决策，将广告分为三类，返回结构化摘要供 LLM 推理。

    类别 A【已确认异常，建议直接关停】：7日均消耗 < 10元（几乎零活动）
    类别 B【待LLM评估】：消耗有意义但指标异常（ROI偏低或衰退信号）
    类别 C【正常运行】：仅返回摘要统计

    Args:
        metrics_csv: ROI 指标 CSV 路径（calculate_roi_metrics 输出）
        end_date: 结束日期
        roi_review_factor: 动态ROI < 全体均值 × 此值 → 进入 B 类（默认 0.8）
        min_spend_for_class_a: 7日均消耗低于此值（元）→ A 类（默认 10.0）
    """
    try:
        # 加载策略参数（动态阈值，不写死在代码中）
        params = _load_strategy_params()

        if not metrics_csv:
            metrics_csv = str(_MINI_DIR / "outputs" / "metrics_temp.csv")

        df = pd.read_csv(metrics_csv)
        if df.empty:
            return ToolResult(title="get_ads_for_review", output="指标数据为空")

        if end_date == "yesterday":
            end_date = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d")

        # ===== 新增：读取人群包级别统计数据（同类对比基准）=====
        logger.info("读取人群包级别统计数据...")
        by_tier_stats = {}
        try:
            # 读取 portfolio_summary JSON 文件
            portfolio_dir = _MINI_DIR / "outputs" / "portfolio_summary"
            portfolio_file = portfolio_dir / f"portfolio_summary_{end_date}.json"

            if portfolio_file.exists():
                import json
                with open(portfolio_file, "r", encoding="utf-8") as f:
                    portfolio_data = json.load(f)
                by_tier_stats = portfolio_data.get("by_audience_tier", {})
                logger.info(f"✅ 从 {portfolio_file.name} 加载了 {len(by_tier_stats)} 个人群包的统计数据")
            else:
                logger.warning(f"未找到 portfolio_summary 文件: {portfolio_file}，将使用全局均值兜底")
                # 可以选择在这里调用 calculate_portfolio_summary 生成文件
                # 但为了简化，我们先用空字典兜底
        except Exception as e:
            logger.warning(f"读取人群包统计数据失败，使用空字典兜底: {e}")
            by_tier_stats = {}

        # 计算广告年龄
        df["ad_age_days"] = df["create_time"].apply(_calculate_ad_age_days)

        # 检测衰退信号
        raw_dir = _MINI_DIR / "outputs" / "raw"
        ad_status_dir = _MINI_DIR / "outputs" / "ad_status"
        decay_signals = _detect_decay_signals(
            ad_ids=df["ad_id"].tolist(),
            raw_dir=raw_dir,
            ad_status_dir=ad_status_dir,
            end_date=end_date,
        )
        df = df.merge(decay_signals, on="ad_id", how="left")
        df["bid_increased_7d"] = df["bid_increased_7d"].fillna(False)
        df["creative_changed_7d"] = df["creative_changed_7d"].fillna(False)
        df["stable_spend_days_30d"] = df["stable_spend_days_30d"].fillna(0)

        # 全体 ROI 分布
        roi_series = df["动态ROI_7日均值"].dropna()
        roi_mean = float(roi_series.mean()) if len(roi_series) > 0 else 0.0
        roi_p25 = float(roi_series.quantile(0.25)) if len(roi_series) > 0 else 0.0
        roi_p50 = float(roi_series.quantile(0.50)) if len(roi_series) > 0 else 0.0
        roi_p75 = float(roi_series.quantile(0.75)) if len(roi_series) > 0 else 0.0
        roi_p90 = float(roi_series.quantile(0.90)) if len(roi_series) > 0 else 0.0

        # 消耗中位数（供出价提升判断）
        cost_series = df["cost_7d_avg"].dropna()
        cost_median = float(cost_series.median()) if len(cost_series) > 0 else 0.0

        # 分类（业务语言）
        zero_spend_ads = []      # 零消耗待关停
        need_review_ads = []     # 待优化评估
        normal_ads_count = 0     # 正常运行

        for _, row in df.iterrows():
            cost_7d_avg = float(row.get("cost_7d_avg", 0) or 0)
            f_roi = row.get("动态ROI_7日均值")
            ad_age = row.get("ad_age_days")
            bid_inc = bool(row.get("bid_increased_7d", False))
            creative_chg = bool(row.get("creative_changed_7d", False))
            stable_days = float(row.get("stable_spend_days_30d", 0) or 0)
            bid_amount = float(row.get("bid_amount", 0) or 0)

            # 零消耗待关停：7日均消耗 < 10元，几乎无活动（强规则，仍保留）
            if cost_7d_avg < min_spend_for_class_a:
                zero_spend_ads.append({
                    "ad_id": int(row["ad_id"]),
                    "ad_name": str(row.get("ad_name", "")),
                    "cost_7d_avg": round(cost_7d_avg, 2),
                })
                continue

            # 待优化评估：ROI 偏低 或 衰退信号 或 出价调整候选（需要智能判断）
            roi_low = (not pd.isna(f_roi)) and (f_roi < roi_mean * roi_review_factor)
            decay_signal = (
                stable_days >= 7
                and cost_7d_avg < 100
                and (bid_inc or creative_chg)
            )
            # 出价调整候选：高ROI低量（提价）或 ROI偏低（降价）
            bid_up_candidate = (
                (not pd.isna(f_roi))
                and f_roi > roi_mean * params["BID_UP_ROI_FACTOR"]
                and cost_7d_avg < cost_median * params["BID_UP_LOW_SPEND_FACTOR"]
                and bid_amount > 0
            ) if BID_ADJUSTMENT_ENABLED else False
            bid_down_candidate = (
                (not pd.isna(f_roi))
                and f_roi < roi_mean * params["BID_DOWN_ROI_FACTOR"]
                and f_roi >= roi_mean * params["ROI_LOW_FACTOR"]
                and cost_7d_avg >= 100
                and bid_amount > 0
            ) if BID_ADJUSTMENT_ENABLED else False

            # 扩量候选：成熟期 + 消耗稳定 + 高消耗 + ROI正常（基于决策树）
            scale_up_candidate = (
                ad_age is not None
                and ad_age > 7  # 成熟期（>7天）
                and stable_days >= 7  # 消耗稳定（≥7天）
                and cost_7d_avg > 1000  # 高消耗（>1000元/天）
                and (not pd.isna(f_roi))
                and f_roi >= roi_mean * 0.9  # ROI正常（≥均值的90%）
            )

            # ===== 年龄保护（第一优先级）=====
            # 无论是否满足候选条件，年龄保护都是第一层判断
            age_protected_skip = False  # 标记是否被年龄保护排除

            if ad_age is not None:
                # 冷启动期（≤3天）：极度保护，直接排除所有评估
                if ad_age <= COLD_START_DAYS:
                    normal_ads_count += 1
                    logger.debug(
                        f"广告 {row['ad_id']} 处于冷启动期（{ad_age}天≤{COLD_START_DAYS}天），"
                        f"年龄保护规则自动排除（无论是否满足候选条件）"
                    )
                    age_protected_skip = True

                # 早期成长期（4-7天）：仅允许提价和扩量评估
                # ⚠️ 关键修复：完全阻断非提价/扩量候选，无论何种候选标志
                elif ad_age <= EARLY_GROWTH_DAYS:
                    # 只有提价候选或扩量候选才允许进入LLM评估
                    # 其他所有候选标志（roi_low, decay_signal, bid_down_candidate）都被排除
                    if not (bid_up_candidate or scale_up_candidate):
                        # 检查是否有任何候选标志（即使不是提价/扩量）
                        has_any_candidate = roi_low or decay_signal or bid_down_candidate
                        if has_any_candidate:
                            # 有候选标志但不是提价/扩量 → 直接排除
                            normal_ads_count += 1
                            logger.debug(
                                f"广告 {row['ad_id']} 处于早期成长期（{ad_age}天，4-{EARLY_GROWTH_DAYS}天），"
                                f"年龄保护规则：仅允许提价/扩量评估，其他候选已排除"
                                f"（roi_low={roi_low}, decay={decay_signal}, bid_down={bid_down_candidate}）"
                            )
                            age_protected_skip = True
                        # else: 无任何候选标志，正常计入normal_ads_count
                    # else: 是提价或扩量候选，允许进入评估

            # 年龄保护排除的广告，直接跳过
            if age_protected_skip:
                continue

            # ===== 业务逻辑判断（第二层）=====
            # 只有通过年龄保护的广告才会到这里
            # 早期成长期的广告只会带着 bid_up_candidate 或 scale_up_candidate 到这里
            if roi_low or decay_signal or bid_up_candidate or bid_down_candidate or scale_up_candidate:
                # ===== 构建广告字典（基础字段）=====
                ad_dict = {
                    "ad_id": int(row["ad_id"]),
                    "ad_name": str(row.get("ad_name", "")),
                    "动态ROI_7日均值": round(float(f_roi), 4) if not pd.isna(f_roi) else None,
                    "cost_7d_avg": round(cost_7d_avg, 2),
                    "cost_7d_total": round(float(row.get("cost_7d_total", 0) or 0), 2),
                    "ad_age_days": int(ad_age) if ad_age is not None else None,
                    "bid_increased_7d": bid_inc,
                    "creative_changed_7d": creative_chg,
                    "stable_spend_days_30d": int(stable_days),
                    "bid_amount": round(bid_amount, 2),
                    "bid_candidate": "bid_up" if bid_up_candidate else ("bid_down" if bid_down_candidate else None),
                    "scale_up_candidate": scale_up_candidate,  # 新增：扩量候选标记
                }

                # ===== 新增：添加 audience_tier 和 roi_valid_days =====
                ad_dict["audience_tier"] = str(row.get("audience_tier", "default"))
                ad_dict["roi_valid_days"] = int(row.get("roi_valid_days", 0) or 0)

                # ===== 新增：添加同类对比数据 =====
                tier = ad_dict.get("audience_tier", "default")
                tier_stats = by_tier_stats.get(tier, {})

                ad_dict["tier_roi_p25"] = tier_stats.get("roi_p25")
                ad_dict["tier_roi_p50"] = tier_stats.get("roi_p50")
                ad_dict["tier_roi_p75"] = tier_stats.get("roi_p75")
                ad_dict["tier_roi_mean"] = tier_stats.get("roi_mean")

                # ===== 新增：裂变率同类对比数据（如果有）=====
                ad_dict["tier_fission_mean"] = tier_stats.get("fission_mean")
                ad_dict["tier_fission_p50"] = tier_stats.get("fission_p50")

                # 计算动态阈值（供LLM参考）
                tier_roi_p50 = tier_stats.get("roi_p50", roi_mean)  # 兜底用全局均值

                # 关停线：中位数的 70-75%（低于25-30%）
                ad_dict["pause_line_min"] = round(tier_roi_p50 * 0.70, 4) if tier_roi_p50 else None
                ad_dict["pause_line_max"] = round(tier_roi_p50 * 0.75, 4) if tier_roi_p50 else None

                # 降价线：中位数的 85-90%（低于10-15%）
                ad_dict["bid_down_line_min"] = round(tier_roi_p50 * 0.85, 4) if tier_roi_p50 else None
                ad_dict["bid_down_line_max"] = round(tier_roi_p50 * 0.90, 4) if tier_roi_p50 else None

                # 提价线：中位数的 105-110%（高于5-10%）— 决策树标准
                ad_dict["bid_up_line_min"] = round(tier_roi_p50 * 1.05, 4) if tier_roi_p50 else None
                ad_dict["bid_up_line_max"] = round(tier_roi_p50 * 1.10, 4) if tier_roi_p50 else None

                # ===== 新增：年龄分段标签（基于决策树图片）=====
                if ad_age is not None:
                    if ad_age <= COLD_START_DAYS:  # ≤3天：冷启动期
                        ad_dict["age_segment"] = "cold_start"
                        ad_dict["age_protection_level"] = "极度保护（冷启动期）"
                        ad_dict["allow_bid_down"] = False  # 不允许降价
                        ad_dict["allow_bid_up"] = False    # 不允许提价
                    elif ad_age <= EARLY_GROWTH_DAYS:  # 4-7天：早期成长期
                        ad_dict["age_segment"] = "early_growth"
                        ad_dict["age_protection_level"] = "仅允许提价（早期成长期）"
                        ad_dict["allow_bid_down"] = False  # 不允许降价
                        ad_dict["allow_bid_up"] = True     # 允许提价（满足ROI+消耗条件时）
                        ad_dict["max_bid_down_pct"] = 0     # 不允许降价
                    else:  # >7天：成熟期
                        ad_dict["age_segment"] = "mature"
                        ad_dict["age_protection_level"] = "正常调控（成熟期）"
                        ad_dict["allow_bid_down"] = True
                        ad_dict["allow_bid_up"] = True
                        ad_dict["max_bid_down_pct"] = 0.05  # 最大降价5%（决策树上限）

                    # ⚠️ 高燃烧预警：广告年龄>3天 且 昨日消耗>300元
                    yesterday_cost = float(row.get("前1日消耗", 0) or 0)
                    if ad_age > HIGH_BURN_AGE_THRESHOLD and yesterday_cost > HIGH_BURN_COST_THRESHOLD:
                        ad_dict["high_burn_alert"] = True
                        ad_dict["yesterday_cost"] = round(yesterday_cost, 2)
                    else:
                        ad_dict["high_burn_alert"] = False

                need_review_ads.append(ad_dict)
                continue

            # 正常运行：ROI 正常且无异常信号
            normal_ads_count += 1

        import json
        result = {
            "summary": {
                "total": len(df),
                "zero_spend_ads": len(zero_spend_ads),
                "need_review_ads": len(need_review_ads),
                "normal_ads": normal_ads_count,
            },
            "distribution": {
                "roi_mean": round(roi_mean, 4),
                "p25": round(roi_p25, 4),
                "p50": round(roi_p50, 4),
                "p75": round(roi_p75, 4),
                "p90": round(roi_p90, 4),
                "cost_7d_avg_median": round(cost_median, 2),
            },
            "bid_adjustment": {
                "enabled": BID_ADJUSTMENT_ENABLED,
                "bid_down_line": round(roi_mean * params["BID_DOWN_ROI_FACTOR"], 4),
                "bid_up_line": round(roi_mean * params["BID_UP_ROI_FACTOR"], 4),
                "low_spend_line": round(cost_median * params["BID_UP_LOW_SPEND_FACTOR"], 2),
            },
            "thresholds_used": {
                "ROI_LOW_FACTOR": params["ROI_LOW_FACTOR"],
                "BID_DOWN_ROI_FACTOR": params["BID_DOWN_ROI_FACTOR"],
                "BID_UP_ROI_FACTOR": params["BID_UP_ROI_FACTOR"],
                "BID_UP_LOW_SPEND_FACTOR": params["BID_UP_LOW_SPEND_FACTOR"],
                "roi_mean": round(roi_mean, 4),
                "pause_line": round(roi_mean * params["ROI_LOW_FACTOR"], 4),
                "bid_down_line": round(roi_mean * params["BID_DOWN_ROI_FACTOR"], 4),
                "bid_up_line": round(roi_mean * params["BID_UP_ROI_FACTOR"], 4),
            },
            "zero_spend_ads": zero_spend_ads,
            "need_review_ads": need_review_ads,
        }

        output_json = json.dumps(result, ensure_ascii=False, indent=2)

        return ToolResult(
            title=f"广告分类（零消耗:{len(zero_spend_ads)} 待评估:{len(need_review_ads)} 正常:{normal_ads_count}）",
            output=output_json,
            metadata={
                "total": len(df),
                "zero_spend_ads": len(zero_spend_ads),
                "need_review_ads": len(need_review_ads),
                "normal_ads": normal_ads_count,
                "roi_mean": roi_mean,
                "end_date": end_date,
            },
        )

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


# ═══════════════════════════════════════════
# 智能引擎工具 2：保存 LLM 决策结果
# ═══════════════════════════════════════════


@tool(description="智能引擎：接收LLM的决策列表，合并A/C类自动决策，保存为结构化结果")
async def apply_decisions(
    ctx: ToolContext,
    decisions: str,
    end_date: str = "yesterday",
    metrics_csv: str = "",
) -> ToolResult:
    """
    接收 LLM 的决策，合并 A 类广告（自动关停）和 C 类广告（自动保持），保存到 llm_decisions_{date}.csv。

    决策分类：
      - 零消耗待关停：7日均消耗 < 10元，几乎无活动 → 规则判断自动关停
      - 待优化评估：ROI 偏低、衰退信号、出价调整候选 → 智能判断
      - 正常运行：ROI 正常且无异常信号 → 规则判断自动保持

    Args:
        decisions: JSON 字符串，LLM 输出的"待优化评估"类广告决策列表
                   格式：[{"ad_id": 123, "action": "pause"/"hold"/"bid_up"/"bid_down",
                           "dimension": "...", "reason": "...", "confidence": "high"/"medium"/"low"}]
        end_date: 结束日期
        metrics_csv: ROI 指标 CSV 路径（用于获取 A/C 类广告）
    """
    import json

    try:
        if end_date == "yesterday":
            end_date = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d")

        # 解析 LLM 决策
        try:
            llm_list = json.loads(decisions)
        except json.JSONDecodeError as e:
            return ToolResult(title="apply_decisions 失败", output=f"decisions 不是合法 JSON: {e}")

        if not isinstance(llm_list, list):
            return ToolResult(title="apply_decisions 失败", output="decisions 必须是 JSON 数组")

        # 加载零消耗待关停广告（规则判断）
        zero_spend_rows = []
        if not metrics_csv:
            metrics_csv = str(_MINI_DIR / "outputs" / "metrics_temp.csv")

        try:
            df_metrics = pd.read_csv(metrics_csv)
            for _, row in df_metrics.iterrows():
                cost_7d_avg = float(row.get("cost_7d_avg", 0) or 0)
                if cost_7d_avg < 10.0:
                    # 优化reason表达：避免"0.00元"显示，改用"几乎无消耗"
                    if cost_7d_avg == 0:
                        reason_text = "7日几乎无消耗，长期无活动"
                    else:
                        reason_text = f"7日均消耗={cost_7d_avg:.2f}元，长期低消耗"
                    zero_spend_rows.append({
                        "ad_id": int(row["ad_id"]),
                        "action": "pause",
                        "dimension": "长期零消耗",
                        "reason": reason_text,
                        "confidence": "high",
                        "source": "规则判断",
                        "cost_7d_avg": cost_7d_avg,  # 用于排序
                    })
        except Exception as e:
            logger.warning("加载零消耗待关停广告失败（跳过）: %s", e)

        # 合并 LLM 决策（标注来源 + 添加cost_7d_avg用于排序 + 冷启动期决策过滤）
        for item in llm_list:
            item["source"] = "智能判断"
            ad_id = item.get("ad_id")
            action = item.get("action", "hold")

            # 从metrics中获取广告信息
            try:
                cost_row = df_metrics[df_metrics["ad_id"] == ad_id]
                if not cost_row.empty:
                    row_data = cost_row.iloc[0]
                    item["cost_7d_avg"] = float(row_data.get("cost_7d_avg", 0) or 0)

                    # ===== 年龄保护兜底检查（阶段3）=====
                    # 阶段1已做前置过滤，这里仅作兜底检查（理论上不应触发）
                    ad_age_days = row_data.get("ad_age_days")
                    if ad_age_days is not None:
                        if ad_age_days <= COLD_START_DAYS:  # ≤3天：冷启动期（极度保护）
                            # 所有操作都改为observe
                            if action in ["bid_down", "pause", "bid_up"]:
                                original_action = action
                                original_reason = item.get("reason", "")
                                item["action"] = "observe"
                                item["reason"] = f"{original_reason}（LLM建议{original_action}，但广告处于冷启动期{ad_age_days}天，年龄保护规则自动改为观察）"
                                item["confidence"] = "low"
                                item["recommended_change_pct"] = None
                                logger.error(
                                    f"⚠️ 兜底检查触发！广告 {ad_id} 处于冷启动期（{ad_age_days}天≤{COLD_START_DAYS}天），"
                                    f"LLM建议 {original_action}，已自动转换为 observe。"
                                    f"这不应该发生（阶段1应已过滤），请检查逻辑！"
                                )
                        elif ad_age_days <= EARLY_GROWTH_DAYS:  # 4-7天：早期成长期（仅允许提价）
                            # 不允许降价/关停
                            if action in ["bid_down", "pause"]:
                                original_action = action
                                original_reason = item.get("reason", "")
                                item["action"] = "observe"
                                item["reason"] = f"{original_reason}（LLM建议{original_action}，但广告处于早期成长期{ad_age_days}天，年龄保护规则仅允许提价，改为观察）"
                                item["confidence"] = "low"
                                item["recommended_change_pct"] = None
                                logger.error(
                                    f"⚠️ 兜底检查触发！广告 {ad_id} 处于早期成长期（{ad_age_days}天，4-{EARLY_GROWTH_DAYS}天），"
                                    f"LLM建议 {original_action}，已自动转换为 observe。"
                                    f"这不应该发生（阶段1应已过滤），请检查逻辑！"
                                )
                else:
                    item["cost_7d_avg"] = 0.0
            except Exception as e:
                item["cost_7d_avg"] = 0.0
                logger.warning(f"处理广告 {ad_id} 信息时出错: {e}")

        # 加载正常运行广告（规则判断）
        normal_running_rows = []
        try:
            # 收集零消耗和待评估的 ad_id
            zero_spend_ad_ids = {row["ad_id"] for row in zero_spend_rows}
            need_review_ad_ids = {item["ad_id"] for item in llm_list}

            # 正常运行 = 所有广告 - 零消耗 - 待评估
            for _, row in df_metrics.iterrows():
                ad_id = int(row["ad_id"])
                if ad_id not in zero_spend_ad_ids and ad_id not in need_review_ad_ids:
                    cost_7d_avg = float(row.get("cost_7d_avg", 0) or 0)
                    f_roi = row.get("动态ROI_7日均值")
                    ad_age_days = row.get("ad_age_days")

                    # 冷启动保护：广告年龄 ≤ 3天（基于决策树）
                    if ad_age_days is not None and ad_age_days <= COLD_START_DAYS:
                        roi_str = f"{f_roi:.2f}" if not pd.isna(f_roi) else "数据不足"
                        normal_running_rows.append({
                            "ad_id": ad_id,
                            "action": "hold",
                            "dimension": "冷启动保护",
                            "reason": f"广告年龄{ad_age_days}天 ≤ {COLD_START_DAYS}天（冷启动期），ROI={roi_str}，消耗{cost_7d_avg:.2f}元/天，极度保护",
                            "confidence": "high",
                            "source": "规则判断",
                            "cost_7d_avg": cost_7d_avg,  # 用于排序
                        })
                    else:
                        # 正常运行
                        roi_str = f"{f_roi:.2f}" if not pd.isna(f_roi) else "数据不足"
                        normal_running_rows.append({
                            "ad_id": ad_id,
                            "action": "hold",
                            "dimension": "正常运行",
                            "reason": f"ROI={roi_str}，消耗正常({cost_7d_avg:.2f}元/天)，保持当前出价",
                            "confidence": "high",
                            "source": "规则判断",
                            "cost_7d_avg": cost_7d_avg,  # 用于排序
                        })
        except Exception as e:
            logger.warning("加载正常运行广告失败（跳过）: %s", e)

        all_decisions = zero_spend_rows + llm_list + normal_running_rows

        if not all_decisions:
            return ToolResult(title="apply_decisions", output="无决策数据")

        df_out = pd.DataFrame(all_decisions)

        # ===== 关键修复：合并 metrics CSV 中的字段 =====
        # 从 metrics CSV 补充 ad_name, ad_age_days, cost_7d_avg, 动态ROI 等字段
        try:
            df_metrics_full = pd.read_csv(metrics_csv)
            # 选择需要合并的列（OUTPUT_COLUMNS中定义的所有列）
            merge_cols = [
                "ad_id", "account_id", "ad_name", "audience_tier", "create_time", "ad_age_days",
                "bid_amount", "yesterday_cost", "yesterday_revenue", "yesterday_roi",
                "cost_7d_total", "cost_7d_avg", "revenue_7d_total",
                "动态ROI", "动态ROI_7日均值", "cost_30d_total", "cost_30d_avg",
                "stable_spend_days_30d", "creative_count", "roi_valid_days"
            ]
            # 只保留存在的列
            merge_cols = [c for c in merge_cols if c in df_metrics_full.columns]
            df_metrics_merge = df_metrics_full[merge_cols]

            # 左连接：保留df_out的所有行，补充字段
            df_out = df_out.merge(df_metrics_merge, on="ad_id", how="left", suffixes=("", "_metrics"))
            logger.info(f"已从 metrics CSV 合并 {len(merge_cols)} 个字段")
        except Exception as e:
            logger.warning(f"合并 metrics 字段失败（决策CSV将缺少扩展字段）: {e}")

        # 过滤：已经是 AD_STATUS_SUSPEND 的广告不应出现在决策表中（已暂停无需再决策）
        ad_status_path = _MINI_DIR / "outputs" / "ad_status" / f"ad_status_{end_date}.csv"
        if ad_status_path.exists():
            try:
                df_status = pd.read_csv(ad_status_path)
                suspended_ads = set(
                    df_status[df_status["ad_status"] == "AD_STATUS_SUSPEND"]["ad_id"].tolist()
                )

                # 过滤掉所有已暂停的广告（不论决策是什么，已暂停的广告不应出现在决策表中）
                before_count = len(df_out)
                df_out = df_out[~df_out["ad_id"].isin(suspended_ads)]
                filtered_count = before_count - len(df_out)
                if filtered_count > 0:
                    logger.info(f"过滤掉 {filtered_count} 个已暂停广告（AD_STATUS_SUSPEND）")
            except Exception as e:
                logger.warning(f"加载广告状态数据失败，跳过过滤: {e}")

        # 确保必要列存在
        for col in ["ad_id", "action", "dimension", "reason", "confidence", "source"]:
            if col not in df_out.columns:
                df_out[col] = ""
        # 数值列用 None 而非空字符串，避免 float("") 异常
        for col in ["recommended_change_pct", "current_bid", "recommended_bid", "cost_7d_avg"]:
            if col not in df_out.columns:
                df_out[col] = None

        # 按7日均消耗降序排列（消耗高的广告排在前面，更需要关注）
        if "cost_7d_avg" in df_out.columns:
            df_out["cost_7d_avg"] = pd.to_numeric(df_out["cost_7d_avg"], errors="coerce").fillna(0)
            df_out = df_out.sort_values("cost_7d_avg", ascending=False).reset_index(drop=True)

        # ⚠️ 不再删除 cost_7d_avg，保留所有字段到最终报告

        # 保存
        reports_dir = _MINI_DIR / "outputs" / "reports"
        reports_dir.mkdir(parents=True, exist_ok=True)
        out_path = reports_dir / f"llm_decisions_{end_date}.csv"
        df_out.to_csv(out_path, index=False, encoding="utf-8-sig")

        pause_count = (df_out["action"] == "pause").sum()
        hold_count = (df_out["action"] == "hold").sum()
        bid_up_count = (df_out["action"] == "bid_up").sum()
        bid_down_count = (df_out["action"] == "bid_down").sum()

        output_parts = [
            f"智能引擎决策已保存: {out_path}",
            f"  关停: {pause_count} 个（含零消耗待关停: {len(zero_spend_rows)} 个）",
            f"  保持: {hold_count} 个（含正常运行: {len(normal_running_rows)} 个）",
        ]
        if bid_up_count > 0:
            output_parts.append(f"  提价: {bid_up_count} 个")
        if bid_down_count > 0:
            output_parts.append(f"  降价: {bid_down_count} 个")

        return ToolResult(
            title=f"智能引擎决策已保存（{len(df_out)}条）",
            output="\n".join(output_parts),
            metadata={
                "csv_path": str(out_path),
                "total": len(df_out),
                "pause": int(pause_count),
                "hold": int(hold_count),
                "bid_up": int(bid_up_count),
                "bid_down": int(bid_down_count),
                "zero_spend_ads": len(zero_spend_rows),
                "normal_running_ads": len(normal_running_rows),
                "end_date": end_date,
            },
        )

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


# ═══════════════════════════════════════════
# 智能引擎工具 3：查询单个广告详情（Mode 2 支撑）
# ═══════════════════════════════════════════


@tool(description="查询单个广告的当前指标和历史数据")
async def query_ad_detail(
    ctx: ToolContext,
    ad_id: str,
    metrics_csv: str = "",
) -> ToolResult:
    """
    查询单个广告的当前指标 + 全局分布上下文（Mode 2 定向操作用）。

    Args:
        ctx: 工具上下文
        ad_id: 广告 ID（字符串或数字均可）
        metrics_csv: ROI 指标 CSV 路径（默认 outputs/metrics_temp.csv）

    Returns:
        ToolResult，包含该广告的详细指标和全局上下文
    """
    import json
    import os

    try:
        if not metrics_csv:
            metrics_csv = str(_MINI_DIR / "outputs" / "metrics_temp.csv")

        metrics_path = Path(metrics_csv)
        if not metrics_path.exists():
            return ToolResult(
                title="query_ad_detail 失败",
                output=f"指标文件不存在: {metrics_csv}，请先执行 calculate_roi_metrics",
            )

        # 检查数据新鲜度
        file_mtime = os.path.getmtime(metrics_path)
        age_hours = (datetime.now().timestamp() - file_mtime) / 3600
        freshness_warning = ""
        if age_hours > 24:
            freshness_warning = f"⚠️ 数据已过期（{age_hours:.1f}小时前更新），建议先执行 fetch_creative_data + calculate_roi_metrics 刷新数据。\n\n"

        df = pd.read_csv(metrics_csv)

        # 查找目标广告
        ad_id_int = int(ad_id)
        ad_row = df[df["ad_id"] == ad_id_int]

        if ad_row.empty:
            return ToolResult(
                title="query_ad_detail",
                output=f"{freshness_warning}未找到广告 {ad_id}，共有 {len(df)} 个广告",
            )

        row = ad_row.iloc[0]

        # 计算广告年龄
        ad_age_days = _calculate_ad_age_days(row.get("create_time"))

        # 全局 ROI 分布
        roi_series = df["动态ROI_7日均值"].dropna()
        roi_mean = float(roi_series.mean()) if len(roi_series) > 0 else 0.0
        cost_series = df["cost_7d_avg"].dropna()
        cost_median = float(cost_series.median()) if len(cost_series) > 0 else 0.0

        roi_low_line = roi_mean * ROI_LOW_FACTOR if "ROI_LOW_FACTOR" in dir() else roi_mean * 0.5
        bid_down_line = roi_mean * BID_DOWN_ROI_FACTOR
        bid_up_line = roi_mean * BID_UP_ROI_FACTOR

        # 构建广告详情
        f_roi = row.get("动态ROI_7日均值")
        ad_detail = {
            "ad_id": ad_id_int,
            "ad_name": str(row.get("ad_name", "")),
            "bid_amount": round(float(row.get("bid_amount", 0) or 0), 2),
            "动态ROI_7日均值": round(float(f_roi), 4) if not pd.isna(f_roi) else None,
            "cost_7d_avg": round(float(row.get("cost_7d_avg", 0) or 0), 2),
            "cost_7d_total": round(float(row.get("cost_7d_total", 0) or 0), 2),
            "ad_age_days": ad_age_days,
            "configured_status": str(row.get("configured_status", "")),
        }

        # 检测干预信号
        try:
            end_date = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d")
            raw_dir = _MINI_DIR / "outputs" / "raw"
            ad_status_dir = _MINI_DIR / "outputs" / "ad_status"
            decay_signals = _detect_decay_signals(
                ad_ids=[ad_id_int],
                raw_dir=raw_dir,
                ad_status_dir=ad_status_dir,
                end_date=end_date,
            )
            if not decay_signals.empty:
                ds_row = decay_signals.iloc[0]
                ad_detail["bid_increased_7d"] = bool(ds_row.get("bid_increased_7d", False))
                ad_detail["creative_changed_7d"] = bool(ds_row.get("creative_changed_7d", False))
        except Exception as e:
            logger.warning("检测干预信号失败: %s", e)

        # 全局上下文
        global_context = {
            "全体动态ROI均值": round(roi_mean, 4),
            "ROI关停线": round(roi_mean * 0.5, 4),
            "ROI降价线": round(bid_down_line, 4),
            "ROI提价线": round(bid_up_line, 4),
            "全体消耗中位数": round(cost_median, 2),
        }

        result = {
            "ad_detail": ad_detail,
            "global_context": global_context,
        }

        output = freshness_warning + json.dumps(result, ensure_ascii=False, indent=2)

        return ToolResult(
            title=f"广告 {ad_id} 详情",
            output=output,
            metadata=result,
        )

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


# ═══════════════════════════════════════════
# 智能引擎工具 4：修改已有决策（Mode 3 支撑）
# ═══════════════════════════════════════════


@tool(description="修改已有决策：修改指定广告的操作或调幅，也可新增决策")
async def modify_decisions(
    ctx: ToolContext,
    modifications: str,
    decisions_csv: str = "",
    end_date: str = "yesterday",
) -> ToolResult:
    """
    修改已有 llm_decisions_{date}.csv 中的决策（Mode 3 反馈修改用）。

    支持两种修改方式：
    1. 按 ad_id 精确修改/新增（upsert）：
       [{"ad_id": "90289631207", "new_action": "bid_down", "new_change_pct": -0.05}]
    2. 按过滤器批量修改：
       [{"filter": "all_bid_down", "new_change_pct": -0.03}]
       支持: all_pause / all_bid_down / all_bid_up / all_llm

    Args:
        ctx: 工具上下文
        modifications: JSON 字符串，修改列表
        decisions_csv: 决策 CSV 路径（默认自动查找最新）
        end_date: 结束日期（用于查找默认 CSV）

    Returns:
        ToolResult，包含修改日志和新的 action 分布
    """
    import json
    import glob as glob_mod

    try:
        if end_date == "yesterday":
            end_date = (datetime.now() - timedelta(days=1)).strftime("%Y%m%d")

        # 解析修改列表
        try:
            mod_list = json.loads(modifications)
        except json.JSONDecodeError as e:
            return ToolResult(title="modify_decisions 失败", output=f"modifications 不是合法 JSON: {e}")

        if not isinstance(mod_list, list):
            return ToolResult(title="modify_decisions 失败", output="modifications 必须是 JSON 数组")

        # 定位决策 CSV
        if not decisions_csv:
            reports_dir = _MINI_DIR / "outputs" / "reports"
            # 先找当天的，再找最新的
            target_path = reports_dir / f"llm_decisions_{end_date}.csv"
            if target_path.exists():
                decisions_csv = str(target_path)
            else:
                # 查找最新的 llm_decisions_*.csv
                pattern = str(reports_dir / "llm_decisions_*.csv")
                files = sorted(glob_mod.glob(pattern), reverse=True)
                if files:
                    decisions_csv = files[0]
                else:
                    return ToolResult(
                        title="modify_decisions 失败",
                        output="未找到任何已有决策文件（llm_decisions_*.csv），请先执行全量分析",
                    )

        decisions_path = Path(decisions_csv)
        if not decisions_path.exists():
            return ToolResult(title="modify_decisions 失败", output=f"决策文件不存在: {decisions_csv}")

        df = pd.read_csv(decisions_csv)
        if df.empty:
            return ToolResult(title="modify_decisions 失败", output="决策文件为空")

        # 加载 metrics 获取 bid_amount
        metrics_csv_path = str(_MINI_DIR / "outputs" / "metrics_temp.csv")
        bid_map = {}
        try:
            df_metrics = pd.read_csv(metrics_csv_path)
            bid_map = dict(zip(df_metrics["ad_id"].astype(int), df_metrics["bid_amount"].fillna(0)))
        except Exception as e:
            logger.warning("加载 metrics 获取 bid_amount 失败: %s", e)

        change_log = []
        new_rows = []

        for mod in mod_list:
            if "filter" in mod:
                # 批量修改
                filter_type = mod["filter"]
                filter_map = {
                    "all_pause": "pause",
                    "all_bid_down": "bid_down",
                    "all_bid_up": "bid_up",
                    "all_llm": None,  # 所有 LLM 决策
                }
                if filter_type not in filter_map:
                    change_log.append(f"⚠️ 未知 filter: {filter_type}，跳过")
                    continue

                target_action = filter_map[filter_type]
                if target_action:
                    mask = df["action"] == target_action
                else:
                    mask = df["source"] == "llm"

                matched = mask.sum()
                if matched == 0:
                    change_log.append(f"filter={filter_type}: 无匹配行")
                    continue

                # 应用修改
                if "new_action" in mod:
                    df.loc[mask, "action"] = mod["new_action"]
                if "new_change_pct" in mod:
                    df.loc[mask, "recommended_change_pct"] = mod["new_change_pct"]
                    # 重算 recommended_bid
                    for idx in df[mask].index:
                        ad_id_val = int(df.at[idx, "ad_id"])
                        bid = bid_map.get(ad_id_val, 0)
                        if bid > 0:
                            new_bid = round(bid * (1 + mod["new_change_pct"]), 2)
                            new_bid = max(new_bid, BID_FLOOR_YUAN)
                            new_bid = min(new_bid, BID_CEILING_YUAN)
                            df.at[idx, "recommended_bid"] = new_bid
                            df.at[idx, "current_bid"] = round(bid, 2)
                if "new_dimension" in mod:
                    df.loc[mask, "dimension"] = mod["new_dimension"]
                if "new_reason" in mod:
                    df.loc[mask, "reason"] = mod["new_reason"]

                df.loc[mask, "source"] = "llm_modified"
                change_log.append(f"filter={filter_type}: 修改 {matched} 行")

            elif "ad_id" in mod:
                # 精确修改/新增（upsert）
                target_id = int(mod["ad_id"])
                mask = df["ad_id"] == target_id

                if mask.any():
                    # 修改已有行
                    if "new_action" in mod:
                        old_action = df.loc[mask, "action"].iloc[0]
                        df.loc[mask, "action"] = mod["new_action"]
                        change_log.append(f"ad_id={target_id}: action {old_action} → {mod['new_action']}")
                    if "new_change_pct" in mod:
                        df.loc[mask, "recommended_change_pct"] = mod["new_change_pct"]
                        bid = bid_map.get(target_id, 0)
                        if bid > 0:
                            new_bid = round(bid * (1 + mod["new_change_pct"]), 2)
                            new_bid = max(new_bid, BID_FLOOR_YUAN)
                            new_bid = min(new_bid, BID_CEILING_YUAN)
                            df.loc[mask, "recommended_bid"] = new_bid
                            df.loc[mask, "current_bid"] = round(bid, 2)
                        change_log.append(f"ad_id={target_id}: change_pct → {mod['new_change_pct']}")
                    if "new_dimension" in mod:
                        df.loc[mask, "dimension"] = mod["new_dimension"]
                    if "new_reason" in mod:
                        df.loc[mask, "reason"] = mod["new_reason"]
                    df.loc[mask, "source"] = "llm_modified"
                else:
                    # 新增行
                    new_action = mod.get("new_action", "hold")
                    change_pct = mod.get("new_change_pct")
                    bid = bid_map.get(target_id, 0)
                    new_bid = None
                    if change_pct is not None and bid > 0:
                        new_bid = round(bid * (1 + change_pct), 2)
                        new_bid = max(new_bid, BID_FLOOR_YUAN)
                        new_bid = min(new_bid, BID_CEILING_YUAN)

                    new_row = {
                        "ad_id": target_id,
                        "action": new_action,
                        "dimension": mod.get("new_dimension", "用户指定"),
                        "reason": mod.get("new_reason", "用户定向操作"),
                        "confidence": mod.get("confidence", "high"),
                        "source": "llm_modified",
                        "recommended_change_pct": change_pct,
                        "current_bid": round(bid, 2) if bid > 0 else None,
                        "recommended_bid": new_bid,
                    }
                    new_rows.append(new_row)
                    change_log.append(f"ad_id={target_id}: 新增 action={new_action}")
            else:
                change_log.append(f"⚠️ 修改项缺少 ad_id 或 filter，跳过: {mod}")

        # 合并新增行
        if new_rows:
            df = pd.concat([df, pd.DataFrame(new_rows)], ignore_index=True)

        # 保存（覆盖原文件）
        df.to_csv(decisions_csv, index=False, encoding="utf-8-sig")

        # 统计新的 action 分布
        action_dist = df["action"].value_counts().to_dict()

        output_parts = [
            f"决策已修改并保存: {decisions_csv}",
            "",
            "修改日志:",
        ]
        for log in change_log:
            output_parts.append(f"  {log}")

        output_parts.extend([
            "",
            "当前 action 分布:",
        ])
        for action, count in action_dist.items():
            output_parts.append(f"  {action}: {count} 个")
        output_parts.append(f"  总计: {len(df)} 个")

        return ToolResult(
            title=f"决策修改完成（{len(change_log)}项变更）",
            output="\n".join(output_parts),
            metadata={
                "csv_path": str(decisions_csv),
                "changes": len(change_log),
                "action_distribution": action_dist,
                "total": len(df),
            },
        )

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