#!/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_zero_spend=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" 待优化评估: {data.get('need_review_ads_total', 0)} 个") print(f" 正常运行: {summary.get('normal_ads', 0)} 个") # 从 tier_batches 提取待评估广告（方案 1：tier 分批） tier_batches = data.get('tier_batches', []) need_review_ads = [] for batch in tier_batches: need_review_ads.extend(batch.get('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')})") # 检查阈值字段（基于渠道P50） if 'pause_line_min' in ad: print(f" 关停线（渠道P50×0.70~0.75）: {ad.get('pause_line_min', 'N/A')} ~ {ad.get('pause_line_max', 'N/A')}") print(f" 降价线（渠道P50×0.85~0.90）: {ad.get('bid_down_line_min', 'N/A')} ~ {ad.get('bid_down_line_max', 'N/A')}") print(f" 提价线（渠道P50×1.05~1.10）: {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()))