352 lines
13 KiB
Python
352 lines
13 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""
|
||
Watcher Agent — 主動偵測與觸發
|
||
|
||
角色：
|
||
- 定期輪詢 sales snapshot，檢查銷量下滑或競品價格突漲
|
||
- 發生異常時構建 event 並 dispatch 到 EventRouter
|
||
- 與 ActionPlanner 配合生成後續計畫
|
||
|
||
設計：
|
||
- 輕量級，無需額外 infra（僅用 PostgreSQL）
|
||
- 異常閾值可透過 env 調整
|
||
- 與 OpenClaw 共享 agent_context 記憶
|
||
"""
|
||
|
||
import json
|
||
import logging
|
||
import os
|
||
import time
|
||
from datetime import datetime, timedelta
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
import requests
|
||
from sqlalchemy import text
|
||
|
||
from database.manager import get_session
|
||
from services.ai_orchestrator import AIOrchestrator
|
||
from services.event_router import dispatch
|
||
|
||
sys_log = logging.getLogger(__name__)
|
||
|
||
# ─── 環境設定 ────────────────────────────────────────────────
|
||
SALES_SNAPSHOT_TABLE = os.getenv("WATCHER_SNAPSHOT_TABLE", "daily_sales_snapshot")
|
||
SALES_DROP_THRESHOLD = float(os.getenv("WATCHER_SALES_DROP_THRESHOLD", "0.20")) # 20%
|
||
PRICE_SURGE_THRESHOLD = float(os.getenv("WATCHER_PRICE_SURGE_THRESHOLD", "0.15")) # 15%
|
||
CACHE_TTL_MIN = int(os.getenv("WATCHER_CACHE_TTL_MIN", "30")) # 輪詻間隔
|
||
|
||
# ─── 共享上下文鍵 ────────────────────────────────────────────
|
||
WATCHER_CTX_NS = "watcher"
|
||
|
||
|
||
class WatcherAgent:
|
||
"""
|
||
主動偵測 Agent
|
||
流程：
|
||
1) 載入最近兩週銷售快照
|
||
2) 計算環比變化
|
||
3) 篩選異常 SKU
|
||
4) 構建 event 並 dispatch
|
||
5) 寫入 agent_context 供後續 Agent 使用
|
||
"""
|
||
|
||
def __init__(self, orchestrator: Optional[AIOrchestrator] = None):
|
||
self.orchestrator = orchestrator or AIOrchestrator()
|
||
|
||
async def scan(self) -> int:
|
||
"""執行一次掃描，回傳觸發的異常數"""
|
||
rows = await self._fetch_sales_snapshot()
|
||
if not rows:
|
||
sys_log.info("[Watcher] 無銷售快照，跳過掃描")
|
||
return 0
|
||
|
||
anomalies = self._detect_anomalies(rows)
|
||
if not anomalies:
|
||
sys_log.info("[Watcher] 未檢測到異常")
|
||
return 0
|
||
|
||
sys_log.info(f"[Watcher] 檢測到 {len(anomalies)} 筆異常，開始 dispatch")
|
||
triggered = 0
|
||
for an in anomalies:
|
||
if await self._dispatch_anomaly(an):
|
||
triggered += 1
|
||
return triggered
|
||
|
||
async def track_outcomes(self, days: int = 7) -> None:
|
||
"""
|
||
排程回撥：執行後 days 天後檢查 action_outcomes，
|
||
並將結果回饋給 OpenClaw 學習。
|
||
這裡僅作佈署示意；實際排程由外部 scheduler 負責。
|
||
"""
|
||
sys_log.info(f"[Watcher] 排程 outcome 回撥（{days} 天後）")
|
||
# 範例：
|
||
# await outcome_tracker.schedule_follow_up(plan_id, sku, metric)
|
||
|
||
# ── 內部方法 ────────────────────────────────────────────────
|
||
|
||
async def _fetch_sales_snapshot(self) -> List[Dict[str, Any]]:
|
||
"""
|
||
讀取銷售快照。
|
||
欄位假設：
|
||
- sku
|
||
- name
|
||
- category
|
||
- sales_curr (最近7天銷售金額)
|
||
- sales_prev (前7天銷售金額)
|
||
- price_momo (MOMO 價格)
|
||
- price_pchome (PChome 價格)
|
||
- stock_status (庫存狀態)
|
||
若實際欄位名不同，請依實際調整。
|
||
"""
|
||
session = get_session()
|
||
try:
|
||
sql = text(f"""
|
||
SELECT sku, name, category,
|
||
COALESCE(sales_curr, 0) AS sales_curr,
|
||
COALESCE(sales_prev, 0) AS sales_prev,
|
||
price_momo, price_pchome, stock_status
|
||
FROM {SALES_SNAPSHOT_TABLE}
|
||
WHERE snapshot_date = CURRENT_DATE - INTERVAL '1 day'
|
||
LIMIT 500
|
||
""")
|
||
result = session.execute(sql).fetchall()
|
||
return [dict(row._mapping) for row in result]
|
||
except Exception as e:
|
||
sys_log.error(f"[Watcher] 無法讀取快照: {e}")
|
||
return []
|
||
finally:
|
||
session.close()
|
||
|
||
def _detect_anomalies(self, rows: List[Dict[str, Any]]) -> List[Dict[str, Any]]:
|
||
anomalies: List[Dict[str, Any]] = []
|
||
for r in rows:
|
||
sku = r["sku"]
|
||
name = r["name"]
|
||
curr = float(r["sales_curr"] or 0)
|
||
prev = float(r["sales_prev"] or 1) # 避免除以 0
|
||
pchome = r["price_pchome"]
|
||
momo = r["price_momo"]
|
||
stock = r.get("stock_status", "unknown")
|
||
|
||
drop_pct = (curr - prev) / prev if prev else 0.0
|
||
price_gap_pct = ((momo - pchome) / pchome * 100) if pchome else 0.0
|
||
|
||
reasons = []
|
||
|
||
# 銷量下滑異常
|
||
if drop_pct <= -SALES_DROP_THRESHOLD:
|
||
reasons.append(
|
||
f"銷量下滑 {drop_pct:+.1%}（閾值 {SALES_DROP_THRESHOLD:+.0%}）"
|
||
)
|
||
|
||
# 競品價格突漲（若我方價格低且差距擴大）
|
||
if price_gap_pct > PRICE_SURGE_THRESHOLD:
|
||
reasons.append(
|
||
f"競品價格突漲 {price_gap_pct:+.1f}% 形成高價差"
|
||
)
|
||
|
||
# 庫存危機（可擴充）
|
||
if stock in ("out_of_stock", "low_stock"):
|
||
reasons.append(f"庫存狀態: {stock}")
|
||
|
||
if not reasons:
|
||
continue
|
||
|
||
anomalies.append({
|
||
"sku": sku,
|
||
"name": name,
|
||
"category": r.get("category", ""),
|
||
"drop_pct": drop_pct,
|
||
"price_gap_pct": price_gap_pct,
|
||
"reasons": reasons,
|
||
"stock": stock,
|
||
"momo_price": momo,
|
||
"pchome_price": pchome,
|
||
})
|
||
return anomalies
|
||
|
||
async def _dispatch_anomaly(self, anom: Dict[str, Any]) -> bool:
|
||
"""
|
||
依異常類型決定路由：
|
||
- 銷量下滑 + 價差微小 → L1（分析原因）
|
||
- 銷量下滑 + 價差大 → L2（規劃 + 審核）
|
||
- 競品價格突漲 → L2（防範被動）
|
||
"""
|
||
drop = anom["drop_pct"]
|
||
gap = anom["price_gap_pct"]
|
||
sku = anom["sku"]
|
||
name = anom["name"]
|
||
session_id = self._ensure_session(sku)
|
||
|
||
# 構建 event payload（與 EventRouter 對齊）
|
||
event = {
|
||
"source": "watcher",
|
||
"event_type": "sales_anomaly",
|
||
"severity": "alert",
|
||
"title": f"銷售異常偵測 — {sku} {name}",
|
||
"summary": "; ".join(anom["reasons"]),
|
||
"payload": {
|
||
"sku": sku,
|
||
"name": name,
|
||
"category": anom["category"],
|
||
"drop_pct": anom["drop_pct"],
|
||
"price_gap_pct": anom["price_gap_pct"],
|
||
"stock": anom["stock"],
|
||
"momo_price": anom["momo_price"],
|
||
"pchome_price": anom["pchome_price"],
|
||
"sales_prev": anom.get("sales_prev"),
|
||
"sales_curr": anom.get("sales_curr"),
|
||
},
|
||
"impact": "銷量下滑可能導致收入損失",
|
||
"status": "open",
|
||
}
|
||
|
||
# 決策路由
|
||
if drop <= -SALES_DROP_THRESHOLD and abs(gap) < PRICE_SURGE_THRESHOLD:
|
||
# 銷量下滑但價差微小 → 檢查是否非價格因素（缺貨/流量）
|
||
event["severity"] = "alert"
|
||
event["payload"]["non_price_factor"] = True
|
||
# 交由 L1 分析原因
|
||
return await self._route_l1(event, session_id)
|
||
else:
|
||
# 銷量下滑 + 價差大 或 競品價格突漲 → L2 規劃
|
||
event["severity"] = "alert"
|
||
return await self._route_l2(event, session_id)
|
||
|
||
async def _route_l1(self, event: Dict[str, Any], session_id: str) -> bool:
|
||
"""L1：Hermes 分析下滑原因"""
|
||
try:
|
||
result = await self.orchestrator.handle_l1(event, session_id)
|
||
sys_log.info(f"[Watcher] L1 dispatch success for {event['payload']['sku']}")
|
||
# 寫入共享上下文
|
||
await self._save_context(session_id, "hermes", {
|
||
"summary": result.get("summary"),
|
||
"probable_cause": result.get("probable_cause"),
|
||
"actions": result.get("actions", []),
|
||
})
|
||
return True
|
||
except Exception as e:
|
||
sys_log.error(f"[Watcher] L1 dispatch failed: {e}")
|
||
# 保底：直接通知
|
||
await self._fallback_notify(event)
|
||
return False
|
||
|
||
async def _route_l2(self, event: Dict[str, Any], session_id: str) -> bool:
|
||
"""L2：NemoTron 規劃 + 審核閘"""
|
||
try:
|
||
result = await self.orchestrator.handle_l2(event, session_id)
|
||
sys_log.info(f"[Watcher] L2 dispatch success for {event['payload']['sku']}")
|
||
# 寫入共享上下文與 action_plans
|
||
await self._save_context(session_id, "nemotron", {
|
||
"plan": result.get("plan"),
|
||
"actions_taken": result.get("actions_taken", []),
|
||
})
|
||
await self._save_action_plan(event, result.get("plan"))
|
||
return True
|
||
except Exception as e:
|
||
sys_log.error(f"[Watcher] L2 dispatch failed: {e}")
|
||
# 保底通知
|
||
await self._fallback_notify(event)
|
||
return False
|
||
|
||
async def _fallback_notify(self, event: Dict[str, Any]) -> None:
|
||
"""當 AI 失敗時，直接通知並記錄原因"""
|
||
sku = event["payload"]["sku"]
|
||
name = event["payload"]["name"]
|
||
text = (
|
||
f"⚠️ [Watcher Fallback] {sku} {name}\n"
|
||
f"原因：{event['summary']}\n"
|
||
f"建議：立即人工檢查銷售與庫存狀態。"
|
||
)
|
||
await self._notify_telegram(text)
|
||
|
||
async def _notify_telegram(self, text: str) -> bool:
|
||
"""透過 Telegram 發送訊息"""
|
||
from services.telegram_templates import alert as render_alert
|
||
bot_token = os.getenv("TELEGRAM_BOT_TOKEN")
|
||
if not bot_token:
|
||
sys_log.warning("[Watcher] TELEGRAM_BOT_TOKEN 未設定")
|
||
return False
|
||
chat_ids_raw = os.getenv("TELEGRAM_CHAT_IDS", "[]")
|
||
try:
|
||
chat_ids = json.loads(chat_ids_raw)
|
||
except json.JSONDecodeError:
|
||
chat_ids = []
|
||
url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
|
||
payload = {
|
||
"chat_id": chat_ids[0] if chat_ids else -1003940688311,
|
||
"text": render_alert(title="銷售異常通知", content=text),
|
||
"parse_mode": "HTML",
|
||
}
|
||
try:
|
||
r = requests.post(url, json=payload, timeout=10)
|
||
return r.ok
|
||
except Exception as e:
|
||
sys_log.error(f"[Watcher] Telegram 通知失敗: {e}")
|
||
return False
|
||
|
||
def _ensure_session(self, sku: str) -> str:
|
||
"""保證 session_id 存在（簡化：skuid 作為 session）"""
|
||
return f"session:{sku}"
|
||
|
||
async def _save_context(self, session_id: str, agent: str, data: Dict[str, Any]) -> None:
|
||
"""寫入 agent_context（共享記憶）"""
|
||
session = get_session()
|
||
try:
|
||
# 刪除舊的 key
|
||
session.execute(
|
||
text("DELETE FROM agent_context WHERE session_id = :sid AND agent_name = :ag"),
|
||
{"sid": session_id, "ag": agent},
|
||
)
|
||
# 寫入新 context
|
||
session.execute(
|
||
text("""
|
||
INSERT INTO agent_context
|
||
(session_id, agent_name, context_key, context_val, created_at, ttl_minutes)
|
||
VALUES
|
||
(:sid, :ag, :ck, :cv, NOW(), :ttl)
|
||
"""),
|
||
{
|
||
"sid": session_id,
|
||
"ag": agent,
|
||
"ck": "latest",
|
||
"cv": json.dumps(data, ensure_ascii=False),
|
||
"ttl": CACHE_TTL_MIN * 2,
|
||
},
|
||
)
|
||
session.commit()
|
||
except Exception as e:
|
||
session.rollback()
|
||
sys_log.warning(f"[Watcher] 寫入 context 失敗: {e}")
|
||
finally:
|
||
session.close()
|
||
|
||
async def _save_action_plan(self, event: Dict[str, Any], plan: Optional[Dict[str, Any]]) -> None:
|
||
"""將 NemoTron 的 plan 寫入 action_plans"""
|
||
if not plan:
|
||
return
|
||
session = get_session()
|
||
try:
|
||
sku = event["payload"]["sku"]
|
||
session.execute(
|
||
text("""
|
||
INSERT INTO action_plans
|
||
(session_id, plan_type, sku, payload, status, created_by)
|
||
VALUES
|
||
(:sid, :pt, :sku, :pl, 'pending', 'nemotron')
|
||
"""),
|
||
{
|
||
"sid": f"session:{sku}",
|
||
"pt": plan.get("type", "price_adjust"),
|
||
"sku": sku,
|
||
"pl": json.dumps(plan, ensure_ascii=False),
|
||
},
|
||
)
|
||
session.commit()
|
||
except Exception as e:
|
||
session.rollback()
|
||
sys_log.warning(f"[Watcher] 寫入 action_plan 失敗: {e}")
|
||
finally:
|
||
session.close()
|