Agent/examples/auto_put_ad_mini/run_decision_test.py

#!/usr/bin/env python3
"""
运行决策引擎并验证新策略是否生效
"""
import asyncio
import sys
import json
from pathlib import Path

# 添加项目根目录到路径
PROJECT_ROOT = Path(__file__).parent.parent.parent
sys.path.insert(0, str(PROJECT_ROOT))

from examples.auto_put_ad_mini.tools.ad_decision import get_ads_for_review, apply_decisions

# 模拟ToolContext
class SimpleContext:
    """简单的上下文模拟"""
    def __init__(self):
        self.config = {}

async def run_decision_test(end_date='20260415'):
    """运行决策测试"""
    ctx = SimpleContext()

    print("=" * 70)
    print("  决策引擎测试 - 验证新策略")
    print("=" * 70)

    # Step 1: 获取需要评估的广告
    print(f"\n📊 步骤 1/2: 获取需要评估的广告 (end_date={end_date})")
    try:
        result = await get_ads_for_review(
            ctx,
            metrics_csv="",
            end_date=end_date,
            roi_review_factor=0.8,
            min_spend_for_class_a=10.0
        )

        print(f"✅ {result.title}")

        # 解析结果
        data = json.loads(result.output)

        summary = data.get('summary', {})
        print(f"\n📊 广告分类统计:")
        print(f"   总广告数: {summary.get('total', 0)}")
        print(f"   零消耗待关停: {summary.get('zero_spend_ads', 0)} 个")
        print(f"   待优化评估: {summary.get('need_review_ads', 0)} 个")
        print(f"   正常运行: {summary.get('normal_ads', 0)} 个")

        # 检查待评估广告是否包含新字段
        need_review_ads = data.get('need_review_ads', [])
        if need_review_ads:
            print(f"\n✅ 验证新字段（前3个样本）:")
            for i, ad in enumerate(need_review_ads[:3]):
                print(f"\n   样本 {i+1}: 广告 {ad['ad_id']}")
                print(f"      人群包: {ad.get('audience_tier', 'N/A')}")
                print(f"      ROI有效天数: {ad.get('roi_valid_days', 'N/A')} 天")
                print(f"      年龄分段: {ad.get('age_segment', 'N/A')} ({ad.get('age_protection_level', 'N/A')})")

                # 检查同类对比字段
                if 'tier_roi_p50' in ad:
                    print(f"      同类中位数: {ad['tier_roi_p50']}")
                    print(f"      关停线: {ad.get('pause_line_min', 'N/A')} ~ {ad.get('pause_line_max', 'N/A')}")
                    print(f"      降价线: {ad.get('bid_down_line_min', 'N/A')} ~ {ad.get('bid_down_line_max', 'N/A')}")
                    print(f"      提价线: {ad.get('bid_up_line_min', 'N/A')} ~ {ad.get('bid_up_line_max', 'N/A')}")

                # 检查操作限制
                if ad.get('age_segment') == 'cold_start':
                    print(f"      ⚠️ 冷启动期限制:")
                    print(f"         - 允许提价: {ad.get('allow_bid_up', 'N/A')}")
                    print(f"         - 允许降价: {ad.get('allow_bid_down', 'N/A')}")

                if ad.get('high_burn_alert'):
                    print(f"      🔥 高燃烧预警: 昨日消耗 {ad.get('yesterday_cost', 'N/A')} 元")

        # 统计年龄分段分布
        age_segments = {}
        for ad in need_review_ads:
            seg = ad.get('age_segment', 'unknown')
            age_segments[seg] = age_segments.get(seg, 0) + 1

        if age_segments:
            print(f"\n📊 待评估广告的年龄分段分布:")
            for seg, count in sorted(age_segments.items()):
                print(f"   {seg}: {count} 个")

        # 检查是否有冷启动期广告
        cold_start_ads = [ad for ad in need_review_ads if ad.get('age_segment') == 'cold_start']
        if cold_start_ads:
            print(f"\n⚠️ 发现 {len(cold_start_ads)} 个冷启动期广告（4-7天），这些广告:")
            print(f"   - ✅ 可以提价")
            print(f"   - ❌ 不允许降价/关停")
            print(f"   示例广告ID: {cold_start_ads[0]['ad_id']}")

        # 保存结果用于后续分析
        output_dir = Path(__file__).parent / "outputs" / "test_results"
        output_dir.mkdir(parents=True, exist_ok=True)

        output_file = output_dir / f"decision_test_{end_date}.json"
        with open(output_file, 'w', encoding='utf-8') as f:
            json.dump(data, f, ensure_ascii=False, indent=2)

        print(f"\n💾 详细结果已保存到: {output_file}")

        return True

    except Exception as e:
        print(f"❌ 决策引擎失败: {e}")
        import traceback
        traceback.print_exc()
        return False

async def check_im_config():
    """检查IM审批配置"""
    print("\n" + "=" * 70)
    print("  检查飞书审批配置")
    print("=" * 70)

    from examples.auto_put_ad_mini.config import (
        IM_ENABLED,
        FEISHU_APP_ID,
        FEISHU_APP_SECRET,
        FEISHU_OPERATOR_OPEN_ID,
        FEISHU_OPERATOR_CHAT_ID,
        EXECUTION_ENABLED
    )

    print(f"\n📋 当前配置:")
    print(f"   IM_ENABLED (飞书审批开关): {IM_ENABLED}")
    print(f"   EXECUTION_ENABLED (执行开关): {EXECUTION_ENABLED}")
    print(f"   FEISHU_APP_ID: {FEISHU_APP_ID}")
    print(f"   FEISHU_APP_SECRET: {'*' * 20}{FEISHU_APP_SECRET[-8:]}")
    print(f"   FEISHU_OPERATOR_OPEN_ID: {FEISHU_OPERATOR_OPEN_ID}")
    print(f"   FEISHU_OPERATOR_CHAT_ID: {FEISHU_OPERATOR_CHAT_ID}")

    # 检查配置问题
    issues = []

    if not IM_ENABLED:
        issues.append("⚠️ IM_ENABLED=False，飞书审批功能已关闭")

    if not EXECUTION_ENABLED:
        issues.append("ℹ️ EXECUTION_ENABLED=False，系统只验证不执行（正常的保护机制）")

    if not FEISHU_APP_ID or FEISHU_APP_ID == "your_app_id_here":
        issues.append("❌ FEISHU_APP_ID 未配置")

    if not FEISHU_APP_SECRET or FEISHU_APP_SECRET == "your_app_secret_here":
        issues.append("❌ FEISHU_APP_SECRET 未配置")

    if not FEISHU_OPERATOR_OPEN_ID:
        issues.append("❌ FEISHU_OPERATOR_OPEN_ID 未配置（接收审批的人员ID）")

    if issues:
        print(f"\n🔍 配置检查结果:")
        for issue in issues:
            print(f"   {issue}")
    else:
        print(f"\n✅ 飞书审批配置正常")

    # 说明何时发送审批
    print(f"\n📖 飞书审批触发时机:")
    print(f"   1. 运行 execute_decisions() 工具时")
    print(f"   2. IM_ENABLED=True 且 EXECUTION_ENABLED=True")
    print(f"   3. 有待执行的决策（pause/bid_down/bid_up）")

    print(f"\n💡 当前状态分析:")
    if not IM_ENABLED:
        print(f"   → 飞书审批已关闭，决策不会发送到飞书")
        print(f"   → 建议：config.py 中设置 IM_ENABLED=True")
    elif not EXECUTION_ENABLED:
        print(f"   → 执行开关关闭，系统处于【只验证不执行】模式")
        print(f"   → 在此模式下，决策会被验证但不会真正调用API")
        print(f"   → 飞书审批也不会发送（因为没有实际执行）")
        print(f"   → 建议：如需测试审批流程，临时设置 EXECUTION_ENABLED=True")
    else:
        print(f"   → 配置正常，执行决策时会发送飞书审批")

    return len(issues) == 0

async def main():
    """主测试流程"""
    print("\n" + "🧪" * 35)
    print(" " * 15 + "决策引擎 + IM配置验证")
    print("🧪" * 35)

    # 测试1: 运行决策引擎
    decision_ok = await run_decision_test('20260415')

    # 测试2: 检查IM配置
    im_ok = await check_im_config()

    # 总结
    print("\n" + "=" * 70)
    print("  测试总结")
    print("=" * 70)

    print(f"   决策引擎测试: {'✅ 通过' if decision_ok else '❌ 失败'}")
    print(f"   IM配置检查: {'✅ 正常' if im_ok else '⚠️ 有问题'}")

    if decision_ok and im_ok:
        print(f"\n🎉 所有测试通过！")
    elif decision_ok:
        print(f"\n⚠️ 决策引擎工作正常，但IM配置需要调整")

    return 0 if decision_ok else 1

if __name__ == "__main__":
    sys.exit(asyncio.run(main()))