feat(ppt): period_review unified generator for quarterly/half/annual/ttm
All checks were successful
CD Pipeline / deploy (push) Successful in 3m37s

Wave 1.2:用一份 generator 解 4 種時間維度報表(季/半年/年/TTM)。

generate_period_review_ppt(services/ppt_generator.py)— 12 頁
- 期間類型徽章自動切換(季報/半年報/年報/TTM 滾動)
- P1 封面:含業績狀態徽章 + elevator pitch + 期間動能
- P2 執行摘要:KPI v2 含 △% (vs 上季/上半/去年) + AI 解讀 + YoY 對比帶
- P3 月度業績走勢(matplotlib 折線:本期 + 上期 + 月均線 + 高低點)+ 4 卡指標
- P4 品類分析:橫條 + 帕雷托雙視圖
- P5-P7 TOP 50 商品(自動分頁)
- P8 TOP 30 廠商
- P9 MCP 市場情報
- P10 AI 結構化洞察
- P11 附錄

query_period_summary(routes/openclaw_bot_routes.py)
- 一次拉齊:kpis / monthly_breakdown / top_products(50) / top_categories(8) /
  top_vendors(30)
- 自動算月度聚合(PostgreSQL TO_CHAR)

路由層加 4 種 sub_type 分支:
- /ppt quarterly [YYYY/Q1-4]    當季或指定季
- /ppt half_yearly [YYYY/H1-2]  當半年或指定半年
- /ppt annual [YYYY]            當年或指定年
- /ppt ttm                      最近 12 個月(自動算到本月底)

每種自動抓三段資料:本期 + 上期同等 + 去年同期
- quarterly: 上季 + 去年同季
- half_yearly: 上半年 + 去年同半年
- annual: 去年 + 前年
- ttm: 上一個 TTM (24-12 月前) + 同上

_ppt_ai_analysis 加 is_period 分支
- 角色:策略顧問 + BU 主管 + CFO 三合一視角
- 結構:整體解讀 / 市場趨勢對位 / 月度走勢分析 / 品類結構 / 戰略級 SMART /
  風險預警
- SMART 行動分三層:下期立即啟動(30 天)/ 下期戰略(60-90 天)/
  下下期預備(6-12 月)
- max_tokens 2600,字數 1000-1300

Telegram 按鈕:報表選單加 4 顆季/半/年/TTM 按鈕

bump TEMPLATE_VERSIONS:quarterly / half_yearly / annual / ttm 全 v3.1.0

煙霧測試:4 種全綠,每份 12 頁 220KB。

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
OoO
2026-05-03 02:10:14 +08:00
parent b6fdb4f473
commit 1af96f5be4
3 changed files with 659 additions and 2 deletions

View File

@@ -1884,6 +1884,7 @@ def _ppt_ai_analysis(prompt_data: str, report_type: str = '') -> str:
is_competitor = '競品' in report_type
is_promo = '促銷' in report_type
is_vendor = '廠商' in report_type
is_period = any(k in report_type for k in ('quarterly', 'half_yearly', 'annual', 'ttm', '季報', '半年報', '年報'))
# ── 格式鐵律(所有 prompt 共用後綴)────────────────────────
FORMAT_RULES = (
@@ -2041,6 +2042,47 @@ def _ppt_ai_analysis(prompt_data: str, report_type: str = '') -> str:
+ FORMAT_RULES
)
max_tokens = 1400
elif is_period:
sys_instruction = (
"你身兼三職:(1) 資深電商策略顧問10 年 BCG / 麥肯錫零售諮詢經驗)"
"(2) momo BU 主管(決策季度/半年/年度資源分配)"
"(3) CFO 觀點(看 P&L 結構、毛利歸因、預算 vs 實際)。\n"
"你的客戶是 momo 高層管理層,會用本報告做下季度/半年度/年度的"
"戰略校正、資源重分配、OKR 設定。所有判斷必須有量化依據。\n\n"
f"請針對以下{report_type}資料,輸出戰略級期間回顧報告,結構嚴格如下:\n\n"
"【期間整體解讀】4-5句\n"
"引用期間業績、訂單、毛利率、客單價,評估等級(卓越/穩健/普通/警訊);"
"與上期 / 去年同期分別作 △% 比較;指出最關鍵亮點與最大警訊;"
"與台灣電商市場同期表現比較定位(健康成長 5~15%、強勁 >20%、警訊 <0%)。\n\n"
"【市場趨勢與本期對位】3-4句\n"
"結合本期所跨檔期(依期間落點:母親節/520/618/雙11/雙12回顧檔期效益"
"對比歷史拉動幅度如雙11 +50~80%、618 +30~50%"
"點出本期是否充分捕捉市場紅利或錯失機會。\n\n"
"【月度走勢分析】3-4句\n"
"解讀月度業績曲線:高點月成因(檔期/活動/季節)、低點月成因;"
"識別連續 2 個月以上的趨勢(持續上升/下降/震盪);"
"QoQ / HoH / YoY 的成長動能差異。\n\n"
"【品類與商品結構洞察】3-4句\n"
"TOP3 主力品類佔比與健康度(前一品類 >60% 為集中度過高);"
"新進榜商品 vs 跌出榜商品的比例與業績規模;"
"毛利結構(高毛利品類 vs 低毛利品類)的 mix 健康度。\n\n"
"【行動建議 — 戰略級 SMART】\n"
"■ 下期立即啟動3 條,✅ 開頭,含期限):\n"
" 針對庫存補貨、廣告投放、定價調整、品類 mix 調整等 30 天內可見效的決策。\n"
"■ 下期戰略重點3 條,✅ 開頭,含 60-90 天目標):\n"
" 針對品類 mix、商品組合、廠商議價、會員活動等 1 季可改善的結構性議題。\n"
"■ 下下期預備佈局2 條,✅ 開頭,含 6-12 個月目標):\n"
" 針對年度大檔雙11/雙12/618、新品類進入、自有品牌、平台戰略等長期議題。\n"
"每條必須含「具體商品/品類 + 量化目標(業績 +X% / 毛利 +Y pp / 客單 +NT$Z+ 期限」。\n\n"
"【最大三大風險與防禦】2-3句\n"
"點出 3 項最大潛在風險(集中度 / 毛利下滑 / 競品價格戰 / 庫存積壓 / 廠商斷供 等),"
"對應「立即啟動」防禦動作(具體至:建立 N 天安全庫存 / 與 TOP3 簽 N 年協議)。\n\n"
"要求:每段引用至少 2 個具體數字,全文 1000~1300 字,"
"語氣為資深顧問遞交給 BU 主管/CEO 的戰略決策報告,禁用模糊用詞,要明確期限與量化。"
+ MARKET_TREND_2026
+ FORMAT_RULES
)
max_tokens = 2600
elif is_vendor:
sys_instruction = (
"你身兼 (1) 資深採購主管10 年零售/電商採購實戰經驗,精通議價、選品、"
@@ -2581,7 +2623,7 @@ def _generate_ppt_cmd(sub_type: str, sub_arg: str, _chat_id: int, target: str,
generate_daily_ppt, generate_weekly_ppt,
generate_monthly_ppt, generate_strategy_ppt,
generate_competitor_ppt, generate_promo_ppt,
generate_vendor_ppt,
generate_vendor_ppt, generate_period_review_ppt,
check_pptx_available
)
except ImportError:
@@ -3106,10 +3148,158 @@ def _generate_ppt_cmd(sub_type: str, sub_arg: str, _chat_id: int, target: str,
})
return ppt_path
elif sub_type in ('quarterly', '季報', 'half_yearly', '半年報', 'annual', '年報', 'ttm'):
# 期間回顧報告 — period_review 共用 generator
# /ppt quarterly [YYYY/Q1-4] 季報
# /ppt half_yearly [YYYY/H1-2] 半年報
# /ppt annual [YYYY] 年報
# /ppt ttm 最近 12 個月(滾動)
from datetime import datetime as _dt, timedelta as _td
import calendar as _cal
# ── 解析期間 ────────────────────────────────────────
if sub_type in ('quarterly', '季報'):
period_type = 'quarterly'
yr = int(sub_arg.split('/')[0]) if sub_arg else now.year
q = int(sub_arg.split('/Q')[1]) if sub_arg and 'Q' in sub_arg else ((now.month - 1) // 3 + 1)
start_mo, end_mo = (q - 1) * 3 + 1, q * 3
start_str = f"{yr}/{start_mo:02d}/01"
end_last = _cal.monthrange(yr, end_mo)[1]
end_str = f"{yr}/{end_mo:02d}/{end_last:02d}"
period_label = f"{yr} Q{q}"
# 上期 = 上一季
prev_q = q - 1 if q > 1 else 4
prev_yr = yr if q > 1 else yr - 1
prev_start_mo, prev_end_mo = (prev_q - 1) * 3 + 1, prev_q * 3
prev_start = f"{prev_yr}/{prev_start_mo:02d}/01"
prev_end_last = _cal.monthrange(prev_yr, prev_end_mo)[1]
prev_end = f"{prev_yr}/{prev_end_mo:02d}/{prev_end_last:02d}"
yoy_start = f"{yr-1}/{start_mo:02d}/01"
yoy_end_last = _cal.monthrange(yr-1, end_mo)[1]
yoy_end = f"{yr-1}/{end_mo:02d}/{yoy_end_last:02d}"
elif sub_type in ('half_yearly', '半年報'):
period_type = 'half_yearly'
yr = int(sub_arg.split('/')[0]) if sub_arg else now.year
h = int(sub_arg.split('/H')[1]) if sub_arg and 'H' in sub_arg else (1 if now.month <= 6 else 2)
start_mo, end_mo = (1, 6) if h == 1 else (7, 12)
start_str = f"{yr}/{start_mo:02d}/01"
end_last = _cal.monthrange(yr, end_mo)[1]
end_str = f"{yr}/{end_mo:02d}/{end_last:02d}"
period_label = f"{yr} H{h}"
prev_h = h - 1 if h > 1 else 2
prev_yr = yr if h > 1 else yr - 1
prev_start_mo, prev_end_mo = (1, 6) if prev_h == 1 else (7, 12)
prev_start = f"{prev_yr}/{prev_start_mo:02d}/01"
prev_end_last = _cal.monthrange(prev_yr, prev_end_mo)[1]
prev_end = f"{prev_yr}/{prev_end_mo:02d}/{prev_end_last:02d}"
yoy_start = f"{yr-1}/{start_mo:02d}/01"
yoy_end_last = _cal.monthrange(yr-1, end_mo)[1]
yoy_end = f"{yr-1}/{end_mo:02d}/{yoy_end_last:02d}"
elif sub_type in ('annual', '年報'):
period_type = 'annual'
yr = int(sub_arg) if sub_arg and sub_arg.isdigit() else now.year
start_str = f"{yr}/01/01"
end_str = f"{yr}/12/31"
period_label = f"{yr}"
prev_start = f"{yr-1}/01/01"
prev_end = f"{yr-1}/12/31"
yoy_start = f"{yr-2}/01/01"
yoy_end = f"{yr-2}/12/31"
else: # ttm 滾動 12 月
period_type = 'ttm'
today = now.date() if hasattr(now, 'date') else now
ttm_end = today
ttm_start = today.replace(day=1) - _td(days=365)
ttm_start = ttm_start.replace(day=1)
start_str = ttm_start.strftime('%Y/%m/%d')
end_str = ttm_end.strftime('%Y/%m/%d')
period_label = f"TTM {start_str[:7]}~{end_str[:7]}"
# 上期 TTM = 再往前 12 個月
prev_end_d = ttm_start - _td(days=1)
prev_start_d = prev_end_d.replace(day=1) - _td(days=365)
prev_start_d = prev_start_d.replace(day=1)
prev_start = prev_start_d.strftime('%Y/%m/%d')
prev_end = prev_end_d.strftime('%Y/%m/%d')
yoy_start = prev_start
yoy_end = prev_end
params = {'report_type': period_type, 'period': period_label}
cached, cached_ai = _load_cached_ppt_path_and_analysis(period_type, params)
if cached:
return cached
mcp_text = ''
if not cached_ai:
mcp_text = _fetch_mcp_context()
# 抓三段資料:本期、上期、去年同期
curr = query_period_summary(start_str, end_str)
if not curr.get('found'):
raise RuntimeError(f'{period_type} 期間 {period_label} 無資料,請確認 DB')
try:
prev = query_period_summary(prev_start, prev_end)
except Exception:
prev = {'found': False}
try:
yoy = query_period_summary(yoy_start, yoy_end)
if yoy.get('found'):
yoy['period_label'] = f"{yoy_start} ~ {yoy_end}"
except Exception:
yoy = {'found': False}
# 組 db_data
kpis = curr.get('kpis', {})
prod_breakdown = '\n'.join(
f" {i+1}. {p.get('name','')[:30]} — NT${p.get('revenue', 0):,.0f}"
for i, p in enumerate(curr.get('top_products', [])[:5])
)
cat_breakdown = '\n'.join(
f" - {c.get('cat','')}: NT${c.get('revenue', 0):,.0f}"
for c in curr.get('top_categories', [])[:5]
)
data_summary = (
f"【期間】{period_label}{period_type}\n"
f"【業績】NT${kpis.get('revenue', 0):,.0f}{kpis.get('revenue', 0)/10000:.1f}萬)\n"
f"【訂單】{kpis.get('orders', 0):,}\n"
f"【毛利率】{kpis.get('gross_margin', 0):.1f}%\n"
f"【平均客單】NT${kpis.get('avg_order', 0):,.0f}\n"
f"【商品數】{kpis.get('product_count', 0)}\n"
f"【廠商數】{kpis.get('vendor_count', 0)}\n\n"
f"【品類 TOP 5】\n{cat_breakdown}\n\n"
f"【熱銷商品 TOP 5】\n{prod_breakdown}\n\n"
f"【上期業績】NT${prev.get('kpis', {}).get('revenue', 0):,.0f}\n"
f"【去年同期業績】NT${yoy.get('kpis', {}).get('revenue', 0):,.0f}\n\n"
f"【MCP 外部市場情報】\n{mcp_text[:600] if mcp_text else '(無)'}"
)
ai_text = cached_ai or _ppt_ai_analysis(
data_summary,
f'{period_type}{period_label}'
)
if not cached_ai and _ppt_needs_fallback(ai_text):
ai_text = _ppt_fallback_insight(period_type, data_summary, mcp_text)
db_data_pr = dict(curr)
db_data_pr['prev_period'] = prev if prev.get('found') else None
db_data_pr['yoy_period'] = yoy if yoy.get('found') else None
db_data_pr['mcp'] = mcp_text
ppt_path = generate_period_review_ppt(period_type, period_label, db_data_pr, ai_text)
_store_ppt_cache(period_type, params, ppt_path, {
'report_type': period_type,
'parameters': params,
'data_summary': data_summary,
'analysis': ai_text,
'mcp': mcp_text,
})
return ppt_path
else:
raise RuntimeError(
f'不支援的簡報類型:{sub_type}'
f'支援daily / weekly / monthly / strategy / competitor / promo / vendor'
f'支援daily / weekly / monthly / quarterly / half_yearly / annual / ttm / strategy / competitor / promo / vendor'
)
@@ -4129,6 +4319,123 @@ def query_date_range(start_str: str, end_str: str) -> dict:
return {'found': False, 'range': f'{start_str}~{end_str}'}
def query_period_summary(start_date: str, end_date: str) -> dict:
"""期間業績完整摘要quarterly / half_yearly / annual / ttm 共用)
回傳:{
kpis: {revenue, orders, gross_margin, avg_order, product_count, vendor_count, days},
monthly_breakdown: [{month, revenue, orders, gross_margin}],
top_products: [...],
top_categories: [...],
top_vendors: [...],
found: bool
}
"""
try:
s = start_date.replace('/', '-')
e = end_date.replace('/', '-')
with _db().connect() as c:
row = c.execute(text("""
SELECT COUNT(DISTINCT "訂單編號"),
COALESCE(SUM(CAST("總業績" AS FLOAT)), 0),
COALESCE(SUM(CAST("總成本" AS FLOAT)), 0),
COUNT(DISTINCT "商品ID"),
COUNT(DISTINCT "廠商名稱"),
COUNT(DISTINCT "日期")
FROM realtime_sales_monthly
WHERE CAST("日期" AS DATE) BETWEEN CAST(:s AS DATE) AND CAST(:e AS DATE)
"""), {'s': s, 'e': e}).fetchone()
# 月度聚合YYYY-MM
monthly_rows = c.execute(text("""
SELECT TO_CHAR(CAST("日期" AS DATE), 'YYYY-MM') AS ym,
SUM(CAST("總業績" AS FLOAT)) AS rev,
SUM(CAST("總成本" AS FLOAT)) AS cost,
COUNT(DISTINCT "訂單編號") AS orders
FROM realtime_sales_monthly
WHERE CAST("日期" AS DATE) BETWEEN CAST(:s AS DATE) AND CAST(:e AS DATE)
GROUP BY ym ORDER BY ym ASC
"""), {'s': s, 'e': e}).fetchall()
# TOP 50 商品
prod_rows = c.execute(text("""
SELECT "商品ID", "商品名稱",
SUM(CAST("總業績" AS FLOAT)) AS rev,
SUM(CAST("數量" AS INTEGER)) AS qty,
COUNT(DISTINCT "訂單編號") AS orders
FROM realtime_sales_monthly
WHERE CAST("日期" AS DATE) BETWEEN CAST(:s AS DATE) AND CAST(:e AS DATE)
GROUP BY "商品ID", "商品名稱"
ORDER BY 3 DESC LIMIT 50
"""), {'s': s, 'e': e}).fetchall()
# TOP 8 品類
cat_rows = c.execute(text("""
SELECT "商品分類L1", SUM(CAST("總業績" AS FLOAT)) AS rev
FROM realtime_sales_monthly
WHERE CAST("日期" AS DATE) BETWEEN CAST(:s AS DATE) AND CAST(:e AS DATE)
AND "商品分類L1" IS NOT NULL AND "商品分類L1" != ''
GROUP BY "商品分類L1" ORDER BY 2 DESC LIMIT 8
"""), {'s': s, 'e': e}).fetchall()
# TOP 30 廠商
vendor_rows = c.execute(text("""
SELECT "廠商名稱",
SUM(CAST("總業績" AS FLOAT)) AS rev,
SUM(CAST("總成本" AS FLOAT)) AS cost
FROM realtime_sales_monthly
WHERE CAST("日期" AS DATE) BETWEEN CAST(:s AS DATE) AND CAST(:e AS DATE)
AND "廠商名稱" IS NOT NULL AND "廠商名稱" != ''
GROUP BY "廠商名稱" ORDER BY 2 DESC LIMIT 30
"""), {'s': s, 'e': e}).fetchall()
if not row or row[0] == 0:
return {'found': False}
orders, revenue, cost = int(row[0]), float(row[1]), float(row[2])
gm = (revenue - cost) / revenue * 100 if revenue > 0 else 0
return {
'found': True,
'kpis': {
'revenue': revenue,
'orders': orders,
'gross_margin': gm,
'avg_order': revenue / orders if orders else 0,
'product_count': int(row[3] or 0),
'vendor_count': int(row[4] or 0),
'days': int(row[5] or 0),
},
'monthly_breakdown': [
{'month': r[0],
'revenue': float(r[1] or 0),
'gross_margin': ((float(r[1] or 0) - float(r[2] or 0)) / float(r[1] or 1) * 100)
if float(r[1] or 0) else 0,
'orders': int(r[3] or 0)}
for r in monthly_rows
],
'top_products': [
{'id': r[0], 'name': r[1], 'revenue': float(r[2]),
'qty': int(r[3] or 0), 'orders': int(r[4] or 0)}
for r in prod_rows
],
'top_categories': [
{'cat': r[0], 'revenue': float(r[1])} for r in cat_rows
],
'top_vendors': [
{'name': r[0],
'sales': float(r[1] or 0),
'profit': float(r[1] or 0) - float(r[2] or 0),
'margin': ((float(r[1] or 0) - float(r[2] or 0)) / float(r[1] or 1) * 100)
if float(r[1] or 0) else 0}
for r in vendor_rows
],
}
except Exception as e:
sys_log.error(f"[query_period_summary] {e}")
return {'found': False}
def query_available_months() -> list:
"""取得 DB 中有資料的月份清單(支援 YYYY/MM/DD 和 YYYY-MM-DD 兩種日期格式)"""
try: