Spaces:
Runtime error
Runtime error
"""
ๆฐๅญ่ฅ้ไธๅฎถ็บงๅๆๅผๆใ
ๅบๅฑไฝฟ็จๆฅๅบฆๆฐๆฎ๏ผๆ Country ร Platform ร Campaign Type ๅค็ปดๅบฆ่ๅใ
ๆฏๆๅจๅบฆๅๆ๏ผdailyโweekly ่ๅ๏ผๅๆๅบฆๅๆ๏ผๆ่ช็ถๆ่ๅ๏ผใ
"""
from pathlib import Path
from typing import Any
import pandas as pd
from datetime import date, datetime, timedelta
from .loader import load_config, load_daily_data, get_available_weeks, get_latest_week
from .ai_analyzer import generate_ai_analysis
log = __import__("logging").getLogger("ad-agent")
def _safe_ai_analysis(
curr_data, prev_data, wow_total, wow_by_platform, wow_by_campaign, wow_by_country,
**kwargs,
) -> dict[str, Any]:
"""Try AI analysis first; fall back to rule-based generate_expert_analysis on failure."""
try:
result = generate_ai_analysis(
curr_data, prev_data, wow_total, wow_by_platform, wow_by_campaign, wow_by_country,
**kwargs,
)
if result:
return result
except Exception as e:
log.warning(f"AI ๅๆไธๅฏ็จ๏ผๅ้ๅฐ่งๅๅผๆ: {e}")
return generate_expert_analysis(
curr_data, prev_data, wow_by_platform, wow_by_campaign, wow_by_country,
yoy_data=kwargs.get("yoy_data"),
yoy_total=kwargs.get("yoy_total"),
yoy_by_platform=kwargs.get("yoy_by_platform"),
yoy_by_country=kwargs.get("yoy_by_country"),
)
ADDITIVE_METRICS = {
"Cost", "Ads.Sales", "Ads.Sales(1d+7d)", "SPF.Sales",
"GA4.Sessions", "SPF.Sessions", "LPV", "Impressions",
"ATC", "Email", "Video Views",
}
def _safe_div(numerator: float, denominator: float) -> float:
if denominator == 0 or pd.isna(denominator):
return 0.0
return numerator / denominator
def _safe_pct_change(curr: float, prev: float) -> float | None:
if prev == 0 or pd.isna(prev):
return None
return (curr - prev) / prev * 100
def _agg_group(group: pd.DataFrame, metrics: list[str]) -> dict[str, float]:
"""
ๆญฃ็กฎ่ๅ๏ผๅฏๅ ๆๆ ็ดๆฅ sum๏ผๆฏ็ๆๆ ้่ฟ่ฟๅๅๅญๅๆฏ้ๆฐ่ฎก็ฎใ
"""
result: dict[str, float] = {}
g = group.copy()
for m in ADDITIVE_METRICS:
if m in g.columns:
result[m] = g[m].fillna(0).sum()
if "Impressions" in g.columns and "CTR" in g.columns:
g["_clicks"] = g["Impressions"].fillna(0) * g["CTR"].fillna(0)
else:
g["_clicks"] = 0.0
if "CVR" in g.columns and "GA4.Sessions" in g.columns:
g["_conversions"] = g["CVR"].fillna(0) * g["GA4.Sessions"].fillna(0)
else:
g["_conversions"] = 0.0
total_clicks = g["_clicks"].sum()
total_conversions = g["_conversions"].sum()
total_cost = result.get("Cost", 0)
total_sales = result.get("Ads.Sales", 0)
total_imp = result.get("Impressions", 0)
total_sessions = result.get("GA4.Sessions", 0)
total_lpv = result.get("LPV", 0)
total_atc = result.get("ATC", 0)
total_email = result.get("Email", 0)
total_video = result.get("Video Views", 0)
result["ROAS"] = _safe_div(total_sales, total_cost)
result["CTR"] = _safe_div(total_clicks, total_imp)
result["CVR"] = _safe_div(total_conversions, total_sessions)
result["CPC"] = _safe_div(total_cost, total_clicks)
result["CPM"] = _safe_div(total_cost, total_imp) * 1000
result["CPSE"] = _safe_div(total_cost, total_sessions)
result["CPLPV"] = _safe_div(total_cost, total_lpv)
result["CP-ATC"] = _safe_div(total_cost, total_atc)
result["CP-Email"] = _safe_div(total_cost, total_email)
result["CP-Video Views"] = _safe_div(total_cost, total_video)
for m, weight_col in [("Bounce Rate", "GA4.Sessions"), ("Average Duration", "GA4.Sessions")]:
if m in g.columns and weight_col in g.columns:
w = g[weight_col].fillna(0)
v = g[m].fillna(0)
total_w = w.sum()
result[m] = _safe_div((v * w).sum(), total_w)
for m in ["50% Play Rate", "100% Play Rate", "Hook Rate"]:
if m in g.columns and "Video Views" in g.columns:
w = g["Video Views"].fillna(0)
v = g[m].fillna(0)
total_w = w.sum()
result[m] = _safe_div((v * w).sum(), total_w)
return result
def aggregate_week_data(
df: pd.DataFrame,
year: int,
week: int,
metrics: list[str],
) -> dict[str, Any]:
"""
่ๅๆๅฎๅจ็ๆฅๅบฆๆฐๆฎ๏ผ่ฟๅๅคๅฑ็ปๆ๏ผ
{
"total": { metric: value },
"by_country": { "CA": {...}, "US": {...} },
"by_platform": { "Google": {...}, "Meta": {...} },
"by_campaign_type": { "Awareness": {...}, ... },
"by_country_platform": { "CA-Google": {...}, ... },
"by_platform_campaign": { "Google-Awareness": {...}, ... },
}
"""
wk = df[(df["year"] == year) & (df["week"] == week)]
if wk.empty:
return {}
out: dict[str, Any] = {}
out["total"] = _agg_group(wk, metrics)
for dim, col in [
("by_country", "country"),
("by_platform", "platform"),
("by_campaign_type", "campaign_type"),
]:
out[dim] = {}
for val, grp in wk.groupby(col):
out[dim][val] = _agg_group(grp, metrics)
out["by_country_platform"] = {}
for (ctry, plat), grp in wk.groupby(["country", "platform"]):
out["by_country_platform"][f"{ctry}-{plat}"] = _agg_group(grp, metrics)
out["by_platform_campaign"] = {}
for (plat, ctype), grp in wk.groupby(["platform", "campaign_type"]):
out["by_platform_campaign"][f"{plat}-{ctype}"] = _agg_group(grp, metrics)
return out
def compute_wow(
curr: dict[str, float],
prev: dict[str, float],
metrics: list[str],
) -> dict[str, float | None]:
"""่ฎก็ฎไธคๆไน้ด็็ฏๆฏๅๅ็(%)ใ"""
changes: dict[str, float | None] = {}
for m in metrics:
c = curr.get(m, 0)
p = prev.get(m, 0)
changes[m] = _safe_pct_change(c, p)
return changes
def _chg(curr_dict: dict, prev_dict: dict, key: str) -> float | None:
return _safe_pct_change(curr_dict.get(key, 0), prev_dict.get(key, 0))
def _fmt_chg(val: float | None) -> str:
if val is None:
return "โ"
return f"{val:+.1f}%"
def generate_expert_analysis(
curr: dict[str, Any],
prev: dict[str, Any] | None,
wow_by_platform: dict[str, dict] | None,
wow_by_campaign: dict[str, dict] | None,
wow_by_country: dict[str, dict] | None,
*,
yoy_data: dict[str, Any] | None = None,
yoy_total: dict[str, float | None] | None = None,
yoy_by_platform: dict[str, dict] | None = None,
yoy_by_country: dict[str, dict] | None = None,
) -> dict[str, Any]:
"""
ๅง็ป่พๅบๅฎๆดไธๅฎถๅๆ๏ผๅ
ๅซ๏ผ
summary - 1~2 ๅฅๆป็ป่ฏไปท๏ผๅซ YoY๏ผ
efficiency - ๆๆพๆ็ๅๆ
platform - ๅนณๅฐ็ญ็ฅๅๆ
funnel - ๆผๆๅฅๅบทๅบฆ่ฏๆญ
market - ๅธๅบ่กจ็ฐๅๆ
alerts - ๅผๅจ้ข่ญฆ๏ผไป
ๅๅจ่พๅคงๆถๅบ็ฐ๏ผ
recommendations - ไธไธๆญฅๅปบ่ฎฎ
"""
tc = curr.get("total", {})
tp = (prev or {}).get("total", {})
has_prev = bool(tp)
ty = (yoy_data or {}).get("total", {})
has_yoy = bool(ty)
plats = curr.get("by_platform", {})
ct = curr.get("by_campaign_type", {})
countries = curr.get("by_country", {})
result: dict[str, Any] = {}
alerts: list[dict[str, str]] = []
recommendations: list[str] = []
# ไธใๆป็ป่ฏไปท
cost = tc.get("Cost", 0)
sales = tc.get("Ads.Sales", 0)
roas = tc.get("ROAS", 0)
if has_prev:
cost_chg = _chg(tc, tp, "Cost")
sales_chg = _chg(tc, tp, "Ads.Sales")
roas_chg = _chg(tc, tp, "ROAS")
yoy_suffix = ""
if has_yoy:
yoy_roas_chg = _chg(tc, ty, "ROAS")
yoy_sales_chg = _chg(tc, ty, "Ads.Sales")
if yoy_roas_chg is not None:
yoy_suffix = f" ๅๆฏๅปๅนดๅๆ: Sales {_fmt_chg(yoy_sales_chg)}๏ผROAS {_fmt_chg(yoy_roas_chg)}ใ"
if roas_chg is not None and roas_chg > 5:
summary = f"ๆฌๆๆดไฝ่กจ็ฐๅๅฅฝ๏ผ่ฑ่ดน ${cost:,.0f}๏ผ{_fmt_chg(cost_chg)}๏ผ๏ผSales ${sales:,.0f}๏ผ{_fmt_chg(sales_chg)}๏ผ๏ผROAS ๆๅ่ณ {roas:.2f}๏ผ{_fmt_chg(roas_chg)}๏ผ๏ผๆๆพๆ็ๆนๅใ{yoy_suffix}"
elif roas_chg is not None and roas_chg < -5:
summary = f"ๆฌๆๆๆพๆ็ๆฟๅ๏ผ่ฑ่ดน ${cost:,.0f}๏ผ{_fmt_chg(cost_chg)}๏ผ๏ผSales ${sales:,.0f}๏ผ{_fmt_chg(sales_chg)}๏ผ๏ผROAS ้่ณ {roas:.2f}๏ผ{_fmt_chg(roas_chg)}๏ผ๏ผ้ๅ
ณๆณจใ{yoy_suffix}"
else:
summary = f"ๆฌๆๆดไฝ่กจ็ฐๅนณ็จณ๏ผ่ฑ่ดน ${cost:,.0f}๏ผ{_fmt_chg(cost_chg)}๏ผ๏ผSales ${sales:,.0f}๏ผ{_fmt_chg(sales_chg)}๏ผ๏ผROAS {roas:.2f}๏ผ{_fmt_chg(roas_chg)}๏ผใ{yoy_suffix}"
else:
summary = f"่ฑ่ดน ${cost:,.0f}๏ผSales ${sales:,.0f}๏ผROAS {roas:.2f}ใ"
result["summary"] = summary
# ไบใๆๆพๆ็ๅๆ
eff_lines: list[str] = []
if has_prev:
if cost_chg is not None and sales_chg is not None:
if cost_chg < 0 and sales_chg > 0:
eff_lines.append(f"่ฑ่ดนไธ้ {abs(cost_chg):.1f}% ็ๅๆถ Sales ๅข้ฟ {sales_chg:.1f}%๏ผ่พน้
ๆ็ๆๅ๏ผ้ข็ฎไฝฟ็จๆด็ฒพๅใ")
elif cost_chg > 0 and sales_chg > cost_chg:
eff_lines.append(f"่ฑ่ดนๅข้ฟ {cost_chg:.1f}%๏ผไฝ Sales ๅขๅน
ๆดๅคง๏ผ{sales_chg:.1f}%๏ผ๏ผ่งๆจกๆฉๅผ ๆๆใ")
elif cost_chg > 0 and sales_chg < cost_chg:
eff_lines.append(f"่ฑ่ดนๅข้ฟ {cost_chg:.1f}% ไฝ Sales ๅขๅน
ไป
{sales_chg:.1f}%๏ผ่พน้
ๅๆฅ้ๅ๏ผ้่ฏไผฐๆฉ้็ญ็ฅใ")
recommendations.append("ๅฎก่ง่ฟๆๅขๆ็ๅนฟๅ็ป/ๅไผ๏ผๆๅ ROAS ไฝไบๅๅผ็ๆๆพ")
elif cost_chg < 0 and sales_chg < cost_chg:
eff_lines.append(f"่ฑ่ดน็ผฉๅ {abs(cost_chg):.1f}%๏ผSales ้ๅน
ๆดๅคง๏ผ{sales_chg:.1f}%๏ผ๏ผ็ผฉ้ๅฏ่ฝ็ ๅฐไบ้ซๆๆๆพใ")
recommendations.append("ๆๆฅ่ขซ็ผฉๅ็ๅนฟๅๆฏๅฆๅ
ๅซ้ซๆ่ฎกๅ๏ผ้ฟๅ
่ฏฏไผค")
cpc_chg = _chg(tc, tp, "CPC")
cpm_chg = _chg(tc, tp, "CPM")
if cpc_chg is not None:
if cpc_chg > 10:
eff_lines.append(f"CPC ไธๆถจ {cpc_chg:.1f}%๏ผๆต้่ทๅๆๆฌ่ตฐ้ซ๏ผๅ
ณๆณจ็ซไบ็ฏๅขๅๅไผๆฐ้ฒๅบฆใ")
elif cpc_chg < -10:
eff_lines.append(f"CPC ไธ้ {abs(cpc_chg):.1f}%๏ผๆต้ๆๆฌๆนๅใ")
else:
eff_lines.append(f"CPC ๅๅจ {_fmt_chg(cpc_chg)}๏ผๆต้ๆๆฌๅบๆฌ็จณๅฎใ")
if not eff_lines:
eff_lines.append(f"ROAS {roas:.2f}๏ผCPC ${tc.get('CPC', 0):.2f}๏ผCPM ${tc.get('CPM', 0):.2f}ใ")
result["efficiency"] = eff_lines
# ไธใๅนณๅฐ็ญ็ฅๅๆ
plat_lines: list[str] = []
if "Google" in plats and "Meta" in plats:
g = plats["Google"]
m = plats["Meta"]
g_cost, m_cost = g.get("Cost", 0), m.get("Cost", 0)
g_roas, m_roas = g.get("ROAS", 0), m.get("ROAS", 0)
total_c = g_cost + m_cost
g_share = g_cost / total_c * 100 if total_c else 0
better = "Google" if g_roas > m_roas else "Meta"
worse = "Meta" if better == "Google" else "Google"
b_roas = g_roas if better == "Google" else m_roas
w_roas = m_roas if better == "Google" else g_roas
b_share = g_share if better == "Google" else 100 - g_share
plat_lines.append(f"้ข็ฎๅ้
: Google {g_share:.0f}% / Meta {100 - g_share:.0f}%ใ")
plat_lines.append(f"{better} ROAS๏ผ{b_roas:.2f}๏ผไผไบ {worse}๏ผ{w_roas:.2f}๏ผใ")
if has_prev and wow_by_platform:
for pn in ["Google", "Meta"]:
pw = (wow_by_platform or {}).get(pn, {})
p_roas_chg = pw.get("ROAS")
p_sales_chg = pw.get("Ads.Sales")
p_ctr_chg = pw.get("CTR")
if p_roas_chg is not None:
if p_sales_chg is not None and p_sales_chg > 15:
plat_lines.append(f"{pn} Sales ๅข้ฟ {p_sales_chg:.1f}%๏ผROAS {_fmt_chg(p_roas_chg)}๏ผๅข้ฟๅจ่ฝๅผบๅฒใ")
recommendations.append(f"{pn} ่กจ็ฐๅผบๅฒ๏ผๅฏๆต่ฏๅจ่ฏฅๅนณๅฐๅฐๅน
ๆฉ้๏ผ่งๅฏ่พน้
ROAS ๆฏๅฆ็จณๅฎ")
elif p_roas_chg < -10:
plat_lines.append(f"{pn} ROAS ไธ้ {abs(p_roas_chg):.1f}%๏ผๆ็่ตฐไฝ๏ผ้ๅฎก่ง็ด ๆๅๅไผ็ญ็ฅใ")
recommendations.append(f"ๆๆฅ {pn} ่ฟๆ็ด ๆ่กจ็ฐ๏ผ่่ๆดๆฐๅๆๆ่ฐๆดๅไผๅฎๅ")
elif p_roas_chg > 10:
plat_lines.append(f"{pn} ROAS ไธๅ {p_roas_chg:.1f}%๏ผๆๆพๆ็ๆนๅใ")
if p_ctr_chg is not None and p_ctr_chg < -10:
plat_lines.append(f"{pn} CTR ไธ้ {abs(p_ctr_chg):.1f}%๏ผๆณจๆ็ด ๆๆฐ้ฒๅบฆใ")
recommendations.append(f"ๆดๆฐ {pn} ๅนฟๅๅๆ๏ผๆต่ฏๆฐ็็ด ๆๆนๅ")
if b_share < 40:
plat_lines.append(f"ๆณจๆ: {better} ROAS ๆด้ซไฝ้ข็ฎๅ ๆฏไป
{b_share:.0f}%๏ผๅญๅจๆ็้้
ใ")
recommendations.append(f"่่้ๅบฆๅ {better} ๅพๆ้ข็ฎ๏ผๆต่ฏๅขๆๅ็่พน้
ROAS ่กจ็ฐ")
if yoy_by_platform:
for pn in ["Google", "Meta"]:
py = yoy_by_platform.get(pn, {})
yoy_r = py.get("ROAS")
yoy_s = py.get("Ads.Sales")
if yoy_r is not None:
plat_lines.append(f"{pn} ๅๆฏๅปๅนด: ROAS {_fmt_chg(yoy_r)}๏ผSales {_fmt_chg(yoy_s)}ใ")
result["platform"] = plat_lines
# ๅใๆผๆๅฅๅบทๅบฆ
funnel_lines: list[str] = []
aw = ct.get("Awareness", {})
co = ct.get("Consideration", {})
cv = ct.get("Conversion", {})
if cv:
cv_roas = cv.get("ROAS", 0)
cv_cost = cv.get("Cost", 0)
cv_share = cv_cost / cost * 100 if cost else 0
funnel_lines.append(f"Conversion: ่ดก็ฎๆ ธๅฟๆถๅ
ฅ๏ผROAS {cv_roas:.2f}๏ผๅ ๆป้ข็ฎ {cv_share:.0f}%ใ")
if co:
co_roas = co.get("ROAS", 0)
co_atc = co.get("ATC", 0)
funnel_lines.append(f"Consideration: ROAS {co_roas:.2f}๏ผATC {co_atc:,.0f}๏ผ่ด่ดฃๅผๆตๅๅ ่ดญใ")
if aw:
aw_cost = aw.get("Cost", 0)
aw_imp = aw.get("Impressions", 0)
funnel_lines.append(f"Awareness: ่ฑ่ดน ${aw_cost:,.0f}๏ผๆๅ
{aw_imp:,.0f}๏ผๅ็ๆๅ
่่ฝใ")
if has_prev:
ctr_chg = _chg(tc, tp, "CTR")
cvr_chg = _chg(tc, tp, "CVR")
if ctr_chg is not None and cvr_chg is not None:
if ctr_chg > 5 and cvr_chg < -5:
funnel_lines.append(f"ๆผๆ่ฏๆญ: CTR ไธๅ๏ผ{_fmt_chg(ctr_chg)}๏ผไฝ CVR ไธ้๏ผ{_fmt_chg(cvr_chg)}๏ผ๏ผๅนฟๅๅธๅผไบๆดๅค็นๅปไฝ่ฝๅฐ้กตๆฟๆฅไธ่ถณใ")
recommendations.append("ไผๅ
ไผๅ่ฝๅฐ้กตไฝ้ช๏ผๅ ่ฝฝ้ๅบฆใ้ฆๅฑๅ
ๅฎนใCTA ๆ้ฎ๏ผ๏ผๆฃๆฅๅไผๅน้
ๅบฆ")
alerts.append({"level": "critical", "title": "ๆผๆๆญ่ฃ้ฃ้ฉ", "detail": f"CTR {_fmt_chg(ctr_chg)} ไฝ CVR {_fmt_chg(cvr_chg)}๏ผ่ฝๅฐ้กต่ฝฌๅๆช่ฝๆฟๆฅๅนฟๅๅผๆต"})
elif ctr_chg < -5 and cvr_chg > 5:
funnel_lines.append(f"ๆผๆ่ฏๆญ: CTR ไธ้๏ผ{_fmt_chg(ctr_chg)}๏ผไฝ CVR ไธๅ๏ผ{_fmt_chg(cvr_chg)}๏ผ๏ผๆต้ๆด็ฒพๅ๏ผๅฏๅจๆญคๅบ็กไธๆฉ้ใ")
elif ctr_chg < -10:
funnel_lines.append(f"ๆผๆ่ฏๆญ: CTR ไธ้ {abs(ctr_chg):.1f}%๏ผๆณจๆ็ด ๆๆฏๅฆ็ฒๅณใ")
recommendations.append("ๆดๆฐๅนฟๅ็ด ๆ๏ผๆต่ฏๆฐ็ๅๆๆนๅ")
else:
funnel_lines.append(f"ๆผๆ่ฏๆญ: CTR {_fmt_chg(ctr_chg)}๏ผCVR {_fmt_chg(cvr_chg)}๏ผๆผๆๆดไฝๅฅๅบทใ")
atc_chg = _chg(tc, tp, "ATC")
if atc_chg is not None:
if atc_chg < -15:
funnel_lines.append(f"ๅ ่ดญ้๏ผATC๏ผไธ้ {abs(atc_chg):.1f}%๏ผไธญ้ดๆผๆๆฟๅ๏ผ่ดญ็ฉๆๅๅๅผฑใ")
recommendations.append("ๆฃๆฅๅๅ็ซไบๅใๅบๅญ็ถๆไธไปทๆ ผ็ญ็ฅ")
alerts.append({"level": "warning", "title": "ๅ ่ดญ้ไธๆป", "detail": f"ATC {_fmt_chg(atc_chg)}๏ผ่ดญ็ฉๆๅๅๅผฑ"})
elif atc_chg > 15:
funnel_lines.append(f"ๅ ่ดญ้๏ผATC๏ผๅข้ฟ {atc_chg:.1f}%๏ผ่ดญ็ฉๆๅๅขๅผบใ")
else:
funnel_lines.append(f"ๅ ่ดญ้๏ผATC๏ผๅๅจ {_fmt_chg(atc_chg)}๏ผๅบๆฌ็จณๅฎใ")
result["funnel"] = funnel_lines
# ไบใๅธๅบ่กจ็ฐ
market_lines: list[str] = []
if len(countries) > 1:
for cn, cm in countries.items():
c_cost = cm.get("Cost", 0)
c_roas = cm.get("ROAS", 0)
c_share = c_cost / cost * 100 if cost else 0
line = f"{cn}: ROAS {c_roas:.2f}๏ผ้ข็ฎๅ ๆฏ {c_share:.0f}%"
if has_prev and wow_by_country and cn in wow_by_country:
c_roas_chg = wow_by_country[cn].get("ROAS")
if c_roas_chg is not None:
line += f"๏ผ็ฏๆฏ {_fmt_chg(c_roas_chg)}"
if c_roas_chg < -10:
alerts.append({"level": "warning", "title": f"{cn} ๅธๅบๆ็ไธๆป", "detail": f"{cn} ROAS ็ฏๆฏ {_fmt_chg(c_roas_chg)}๏ผ้ๅ
ณๆณจ่ฏฅๅธๅบ็ซไบ็ฏๅขไธๅไผๅๅ"})
elif c_roas_chg > 15:
alerts.append({"level": "positive", "title": f"{cn} ๅธๅบ่กจ็ฐ็ชๅบ", "detail": f"{cn} ROAS ็ฏๆฏ {_fmt_chg(c_roas_chg)}๏ผๅฏ่่ๅ ๅคง่ฏฅๅธๅบๆๅ
ฅ"})
if yoy_by_country and cn in yoy_by_country:
c_yoy_roas = yoy_by_country[cn].get("ROAS")
c_yoy_sales = yoy_by_country[cn].get("Ads.Sales")
if c_yoy_roas is not None:
line += f"๏ผๅๆฏ ROAS {_fmt_chg(c_yoy_roas)}"
if c_yoy_sales is not None:
line += f" / Sales {_fmt_chg(c_yoy_sales)}"
market_lines.append(line + "ใ")
result["market"] = market_lines
# ๅ
ญใROAS ็บงๅซๅผๅจ้ข่ญฆ
if has_prev:
roas_chg = _chg(tc, tp, "ROAS")
if roas_chg is not None:
if roas_chg <= -20:
alerts.append({"level": "critical", "title": "ROAS ๅคงๅน
ไธ้",
"detail": f"ROAS {tp['ROAS']:.2f} -> {tc['ROAS']:.2f}๏ผ{roas_chg:+.1f}%๏ผ๏ผ้็ซๅณๆๆฅๅไผ/็ด ๆ/ๅบไปท็ญ็ฅ"})
elif roas_chg <= -10:
alerts.append({"level": "warning", "title": "ROAS ่ตฐไฝ",
"detail": f"ROAS {tp['ROAS']:.2f} -> {tc['ROAS']:.2f}๏ผ{roas_chg:+.1f}%๏ผ๏ผๅปบ่ฎฎๅฎก่ง่ฟๆ่ฐๆด"})
if cost_chg is not None and cost_chg > 20 and (sales_chg is None or sales_chg < cost_chg * 0.5):
alerts.append({"level": "critical", "title": "่ฑ่ดนๆฟๅขๅๆฅไธ่ถณ",
"detail": f"Cost {_fmt_chg(cost_chg)} ไฝ Sales {_fmt_chg(sales_chg)}๏ผ่พน้
ๆ็ไธ้"})
elif cost_chg is not None and cost_chg < -20:
alerts.append({"level": "warning", "title": "่ฑ่ดนๅคงๅน
็ผฉๅ",
"detail": f"Cost {_fmt_chg(cost_chg)}๏ผ็กฎ่ฎคๆฏไธปๅจ่ฐๆง่ฟๆฏๆๆพๅผๅธธ"})
if wow_by_campaign:
cv_wow = wow_by_campaign.get("Conversion", {})
cv_roas_chg = cv_wow.get("ROAS")
if cv_roas_chg is not None and cv_roas_chg < -10:
alerts.append({"level": "critical", "title": "Conversion ROAS ไธๆป",
"detail": f"ๆ ธๅฟ่ฝฌๅๅนฟๅ ROAS {_fmt_chg(cv_roas_chg)}๏ผ้ๆๆฅ"})
sess_chg = _chg(tc, tp, "GA4.Sessions")
if sess_chg is not None and sess_chg < -10:
alerts.append({"level": "warning", "title": "็ซ็นๆต้ไธ้",
"detail": f"Sessions {_fmt_chg(sess_chg)}๏ผ็กฎ่ฎคๆๆพ่ฆ็ๆฏๅฆไธ่ถณ"})
# YoY ๅๆฏ้ข่ญฆ
if has_yoy:
yoy_roas_chg = _chg(tc, ty, "ROAS")
yoy_sales_chg = _chg(tc, ty, "Ads.Sales")
if yoy_roas_chg is not None:
if yoy_roas_chg < -20:
alerts.append({"level": "critical", "title": "ๅๆฏ ROAS ๅคงๅน
่ฝๅ",
"detail": f"่พๅปๅนดๅๆ ROAS {_fmt_chg(yoy_roas_chg)}๏ผSales {_fmt_chg(yoy_sales_chg)}๏ผ้ๅฎก่งๆดไฝ็ญ็ฅ"})
elif yoy_roas_chg < -10:
alerts.append({"level": "warning", "title": "ๅๆฏ ROAS ไธๆป",
"detail": f"่พๅปๅนดๅๆ ROAS {_fmt_chg(yoy_roas_chg)}๏ผๆณจๆๆ็ๆฏๅฆๆ็ปญ่ตฐไฝ"})
elif yoy_roas_chg > 20:
alerts.append({"level": "positive", "title": "ๅๆฏ ROAS ๅคงๅน
ๆๅ",
"detail": f"่พๅปๅนดๅๆ ROAS {_fmt_chg(yoy_roas_chg)}๏ผSales {_fmt_chg(yoy_sales_chg)}๏ผๅนดๅบฆๅข้ฟๅผบๅฒ"})
seen = set()
unique_alerts = []
for a in alerts:
key = a["title"]
if key not in seen:
seen.add(key)
unique_alerts.append(a)
level_order = {"critical": 0, "warning": 1, "positive": 2}
unique_alerts.sort(key=lambda a: level_order.get(a["level"], 9))
result["alerts"] = unique_alerts
# ไธใๅปบ่ฎฎ
if not recommendations:
if has_prev and roas_chg is not None and roas_chg > 0:
recommendations.append("ๅฝๅๆๆพๆ็ๅๅฅฝ๏ผๅปบ่ฎฎ็ปดๆ็ฐๆ็ญ็ฅ๏ผๅฏๅฐๅน
ๆต่ฏๆฉ้")
else:
recommendations.append("ไฟๆ็ฐๆๆๆพ่ๅฅ๏ผๆ็ปญ็ๆงๆ ธๅฟๆๆ ๅๅ")
result["recommendations"] = recommendations
return result
def _apply_platform_filter(df: pd.DataFrame, platform: str | None) -> pd.DataFrame:
if platform and "platform" in df.columns:
return df[df["platform"] == platform].copy()
return df
def run_analysis(
config_path: Path | None = None,
target_year: int | None = None,
target_week: int | None = None,
trend_table: list[dict] | None = None,
memory_context: str | None = None,
business_context: str | None = None,
platform: str | None = None,
) -> dict[str, Any]:
"""
ๆง่กๅฎๆดๅจๅบฆๅๆ๏ผ่ฟๅ็ปๆๅๆฅๅๆฐๆฎใ
ๅบๅฑไปๆฅๅบฆๆฐๆฎ่ๅๅฐๅจๅบฆใ
่ฅไธๆๅฎ target_year/target_week ๅๅๆๆๆฐไธๅจใ
"""
config = load_config(config_path)
metrics = config.get("metrics", [])
core_metrics = config.get("core_metrics", metrics[:9])
df = _apply_platform_filter(load_daily_data(config_path), platform)
if target_year and target_week:
curr_year, curr_week = target_year, target_week
else:
curr_year, curr_week = get_latest_week(df)
weeks = get_available_weeks(df)
curr_idx = None
for i, (y, w) in enumerate(weeks):
if y == curr_year and w == curr_week:
curr_idx = i
break
prev_year, prev_week = (None, None)
if curr_idx is not None and curr_idx > 0:
prev_year, prev_week = weeks[curr_idx - 1]
curr_data = aggregate_week_data(df, curr_year, curr_week, metrics)
prev_data = aggregate_week_data(df, prev_year, prev_week, metrics) if prev_year else None
wow_total = None
wow_by_platform = None
wow_by_campaign = None
wow_by_country = None
if prev_data:
wow_total = compute_wow(curr_data.get("total", {}), prev_data.get("total", {}), metrics)
wow_by_platform = _compute_dim_wow(curr_data, prev_data, "by_platform", metrics)
wow_by_campaign = _compute_dim_wow(curr_data, prev_data, "by_campaign_type", metrics)
wow_by_country = _compute_dim_wow(curr_data, prev_data, "by_country", metrics)
# YoY โ ๅปๅนดๅๅจ
yoy_year = curr_year - 1
yoy_week = curr_week
yoy_period = None
yoy_data = None
yoy_total = None
yoy_by_platform = None
yoy_by_campaign = None
yoy_by_country = None
if (yoy_year, yoy_week) in set(weeks):
yoy_period = {"year": yoy_year, "week": yoy_week}
yoy_data = aggregate_week_data(df, yoy_year, yoy_week, metrics)
if yoy_data:
yoy_total = compute_wow(curr_data.get("total", {}), yoy_data.get("total", {}), metrics)
yoy_by_platform = _compute_dim_wow(curr_data, yoy_data, "by_platform", metrics)
yoy_by_campaign = _compute_dim_wow(curr_data, yoy_data, "by_campaign_type", metrics)
yoy_by_country = _compute_dim_wow(curr_data, yoy_data, "by_country", metrics)
expert = _safe_ai_analysis(
curr_data, prev_data, wow_total, wow_by_platform, wow_by_campaign, wow_by_country,
yoy_data=yoy_data, yoy_total=yoy_total,
yoy_by_platform=yoy_by_platform, yoy_by_country=yoy_by_country,
period_label=f"{curr_year} W{curr_week}",
trend_table=trend_table, memory_context=memory_context,
business_context=business_context,
)
return {
"current_period": {"year": curr_year, "week": curr_week},
"previous_period": {"year": prev_year, "week": prev_week} if prev_year else None,
"yoy_period": yoy_period,
"current": curr_data,
"previous": prev_data,
"yoy": yoy_data,
"wow_total": wow_total,
"wow_by_platform": wow_by_platform,
"wow_by_campaign": wow_by_campaign,
"wow_by_country": wow_by_country,
"yoy_total": yoy_total,
"yoy_by_platform": yoy_by_platform,
"yoy_by_campaign": yoy_by_campaign,
"yoy_by_country": yoy_by_country,
"expert": expert,
"core_metrics": core_metrics,
"all_metrics": metrics,
}
def _compute_dim_wow(
curr_data: dict[str, Any],
prev_data: dict[str, Any],
dim_key: str,
metrics: list[str],
) -> dict[str, dict]:
result = {}
curr_dim = curr_data.get(dim_key, {})
prev_dim = prev_data.get(dim_key, {})
for key in curr_dim:
if key in prev_dim:
result[key] = compute_wow(curr_dim[key], prev_dim[key], metrics)
return result
# โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
# ๆๅบฆๅๆ โ ๅบไบๆฅๅบฆๆฐๆฎๆ่ช็ถๆ่ๅ
# โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
def aggregate_month_data(
df: pd.DataFrame,
year: int,
month: int,
metrics: list[str],
) -> dict[str, Any]:
"""ๅฏนๆๅฎๅนดๆ็ๆฅๅบฆๆฐๆฎๅ่ๅ๏ผไธ aggregate_week_data ็ปๆไธ่ด๏ผใ"""
cal_year = df["date"].dt.year
subset = df[(cal_year == year) & (df["month"] == month)]
if subset.empty:
return {}
out: dict[str, Any] = {}
out["total"] = _agg_group(subset, metrics)
for dim, col in [
("by_country", "country"),
("by_platform", "platform"),
("by_campaign_type", "campaign_type"),
]:
out[dim] = {}
for val, grp in subset.groupby(col):
out[dim][val] = _agg_group(grp, metrics)
return out
def run_monthly_analysis(
config_path: Path | None = None,
target_year: int = 2025,
target_month: int = 12,
trend_table: list[dict] | None = None,
memory_context: str | None = None,
business_context: str | None = None,
platform: str | None = None,
) -> dict[str, Any]:
"""
ๆๅบฆๆฑๆปๅๆใๅบไบๆฅๅบฆๆฐๆฎๆ่ช็ถๆ่ๅ๏ผไธไธๆๅๅปๅนดๅๆๅฏนๆฏใ
"""
config = load_config(config_path)
metrics = config.get("metrics", [])
core_metrics = config.get("core_metrics", metrics[:9])
df = _apply_platform_filter(load_daily_data(config_path), platform)
curr_data = aggregate_month_data(df, target_year, target_month, metrics)
if not curr_data:
return {
"period_label": f"{target_year}ๅนด{target_month}ๆ",
"current": {},
"previous": None,
"wow_total": None,
"expert": {},
"core_metrics": core_metrics,
"all_metrics": metrics,
"error": f"{target_year}ๅนด{target_month}ๆ ๆ ๅฏ็จๆฐๆฎ",
}
# ไธๆ
prev_month = target_month - 1
prev_year = target_year
if prev_month < 1:
prev_month = 12
prev_year -= 1
prev_data = aggregate_month_data(df, prev_year, prev_month, metrics)
prev_data = prev_data if prev_data else None
# MoM (็ฏๆฏ)
wow_total = None
if prev_data and curr_data.get("total") and prev_data.get("total"):
wow_total = compute_wow(curr_data["total"], prev_data["total"], metrics)
wow_by_platform = _compute_dim_wow(curr_data, prev_data, "by_platform", metrics) if prev_data else None
wow_by_campaign = _compute_dim_wow(curr_data, prev_data, "by_campaign_type", metrics) if prev_data else None
wow_by_country = _compute_dim_wow(curr_data, prev_data, "by_country", metrics) if prev_data else None
# YoY โ ๅปๅนดๅๆ
yoy_year = target_year - 1
yoy_data = aggregate_month_data(df, yoy_year, target_month, metrics)
yoy_data = yoy_data if yoy_data else None
yoy_total = None
yoy_by_platform = None
yoy_by_campaign = None
yoy_by_country = None
yoy_label = None
if yoy_data and curr_data.get("total") and yoy_data.get("total"):
yoy_label = f"{yoy_year}ๅนด{target_month}ๆ"
yoy_total = compute_wow(curr_data["total"], yoy_data["total"], metrics)
yoy_by_platform = _compute_dim_wow(curr_data, yoy_data, "by_platform", metrics)
yoy_by_campaign = _compute_dim_wow(curr_data, yoy_data, "by_campaign_type", metrics)
yoy_by_country = _compute_dim_wow(curr_data, yoy_data, "by_country", metrics)
expert = _safe_ai_analysis(
curr_data, prev_data, wow_total, wow_by_platform, wow_by_campaign, wow_by_country,
yoy_data=yoy_data, yoy_total=yoy_total,
yoy_by_platform=yoy_by_platform, yoy_by_country=yoy_by_country,
period_label=f"{target_year}ๅนด{target_month}ๆ",
trend_table=trend_table, memory_context=memory_context,
business_context=business_context,
) if prev_data else {}
prev_label = f"{prev_year}ๅนด{prev_month}ๆ" if prev_data else None
return {
"period_label": f"{target_year}ๅนด{target_month}ๆ",
"prev_label": prev_label,
"yoy_label": yoy_label,
"current": curr_data,
"previous": prev_data,
"yoy": yoy_data,
"wow_total": wow_total,
"wow_by_platform": wow_by_platform,
"wow_by_campaign": wow_by_campaign,
"wow_by_country": wow_by_country,
"yoy_total": yoy_total,
"yoy_by_platform": yoy_by_platform,
"yoy_by_campaign": yoy_by_campaign,
"yoy_by_country": yoy_by_country,
"expert": expert,
"core_metrics": core_metrics,
"all_metrics": metrics,
}
# โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
# ๆฅๅบฆๅๆ โ ๅๆฅๆฐๆฎ๏ผ็ฏๆฏๅไธๆฅ๏ผๅๆฏๅปๅนดๅๆฅ
# โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
def aggregate_day_data(
df: pd.DataFrame,
target_date: date,
metrics: list[str],
) -> dict[str, Any]:
"""ๅฏนๆๅฎๆฅๆ็ๆฐๆฎๅ่ๅ๏ผ็ปๆๅ aggregate_week_data๏ผใ"""
subset = df[df["date"].dt.date == target_date]
if subset.empty:
return {}
out: dict[str, Any] = {}
out["total"] = _agg_group(subset, metrics)
for dim, col in [
("by_country", "country"),
("by_platform", "platform"),
("by_campaign_type", "campaign_type"),
]:
out[dim] = {}
for val, grp in subset.groupby(col):
out[dim][val] = _agg_group(grp, metrics)
return out
def run_daily_analysis(
config_path: Path | None = None,
target_date: date | None = None,
trend_table: list[dict] | None = None,
memory_context: str | None = None,
business_context: str | None = None,
platform: str | None = None,
) -> dict[str, Any]:
"""
ๆฅๅบฆๅๆๅ
ฅๅฃใ
็ฏๆฏ: ๅไธๆฅ (DoD)
ๅๆฏ: ๅปๅนดๅๆฅ (YoY)
"""
config = load_config(config_path)
metrics = config.get("metrics", [])
core_metrics = config.get("core_metrics", metrics[:9])
df = _apply_platform_filter(load_daily_data(config_path), platform)
if target_date is None:
target_date = df["date"].dt.date.max()
date_str = target_date.strftime("%Y-%m-%d")
curr_data = aggregate_day_data(df, target_date, metrics)
if not curr_data:
return {
"period_label": date_str,
"current": {},
"previous": None,
"wow_total": None,
"expert": {},
"core_metrics": core_metrics,
"all_metrics": metrics,
"error": f"{date_str} ๆ ๅฏ็จๆฐๆฎ",
}
prev_date = target_date - timedelta(days=1)
prev_data = aggregate_day_data(df, prev_date, metrics)
prev_data = prev_data if prev_data else None
dod_total = None
dod_by_platform = None
dod_by_campaign = None
dod_by_country = None
if prev_data and curr_data.get("total") and prev_data.get("total"):
dod_total = compute_wow(curr_data["total"], prev_data["total"], metrics)
dod_by_platform = _compute_dim_wow(curr_data, prev_data, "by_platform", metrics)
dod_by_campaign = _compute_dim_wow(curr_data, prev_data, "by_campaign_type", metrics)
dod_by_country = _compute_dim_wow(curr_data, prev_data, "by_country", metrics)
yoy_date = target_date.replace(year=target_date.year - 1)
yoy_data = aggregate_day_data(df, yoy_date, metrics)
yoy_data = yoy_data if yoy_data else None
yoy_total = None
yoy_by_platform = None
yoy_by_country = None
yoy_label = None
if yoy_data and curr_data.get("total") and yoy_data.get("total"):
yoy_label = yoy_date.strftime("%Y-%m-%d")
yoy_total = compute_wow(curr_data["total"], yoy_data["total"], metrics)
yoy_by_platform = _compute_dim_wow(curr_data, yoy_data, "by_platform", metrics)
yoy_by_country = _compute_dim_wow(curr_data, yoy_data, "by_country", metrics)
expert = _safe_ai_analysis(
curr_data, prev_data, dod_total, dod_by_platform, dod_by_campaign, dod_by_country,
yoy_data=yoy_data, yoy_total=yoy_total,
yoy_by_platform=yoy_by_platform, yoy_by_country=yoy_by_country,
period_label=date_str,
trend_table=trend_table, memory_context=memory_context,
business_context=business_context,
) if prev_data else {}
prev_label = prev_date.strftime("%Y-%m-%d") if prev_data else None
return {
"period_label": date_str,
"prev_label": prev_label,
"yoy_label": yoy_label,
"current": curr_data,
"previous": prev_data,
"yoy": yoy_data,
"wow_total": dod_total,
"wow_by_platform": dod_by_platform,
"wow_by_campaign": dod_by_campaign,
"wow_by_country": dod_by_country,
"yoy_total": yoy_total,
"yoy_by_platform": yoy_by_platform,
"yoy_by_country": yoy_by_country,
"expert": expert,
"core_metrics": core_metrics,
"all_metrics": metrics,
}
# โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
# ่ชๅฎไนๆถ้ดๆฎตๅๆ
# โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
def aggregate_range_data(
df: pd.DataFrame,
start_date: date,
end_date: date,
metrics: list[str],
) -> dict[str, Any]:
"""ๅฏนๆๅฎๆฅๆ่ๅด [start, end] ็ๆฐๆฎๅ่ๅใ"""
mask = (df["date"].dt.date >= start_date) & (df["date"].dt.date <= end_date)
subset = df[mask]
if subset.empty:
return {}
out: dict[str, Any] = {}
out["total"] = _agg_group(subset, metrics)
for dim, col in [
("by_country", "country"),
("by_platform", "platform"),
("by_campaign_type", "campaign_type"),
]:
out[dim] = {}
for val, grp in subset.groupby(col):
out[dim][val] = _agg_group(grp, metrics)
return out
def run_range_analysis(
config_path: Path | None = None,
start_date: date | None = None,
end_date: date | None = None,
trend_table: list[dict] | None = None,
memory_context: str | None = None,
business_context: str | None = None,
platform: str | None = None,
) -> dict[str, Any]:
"""
่ชๅฎไนๆถ้ดๆฎตๅๆใ
็ฏๆฏ: ็ญ้ฟ็ๅไธๆฎต (PoP)
ๅๆฏ: ๅปๅนดๅๆฅๆๆฎต (YoY)
"""
config = load_config(config_path)
metrics = config.get("metrics", [])
core_metrics = config.get("core_metrics", metrics[:9])
df = _apply_platform_filter(load_daily_data(config_path), platform)
if start_date is None or end_date is None:
return {"error": "้่ฆๆไพ่ตทๆญขๆฅๆ"}
if start_date > end_date:
start_date, end_date = end_date, start_date
days_count = (end_date - start_date).days + 1
period_label = f"{start_date.strftime('%Y-%m-%d')} ~ {end_date.strftime('%Y-%m-%d')}"
curr_data = aggregate_range_data(df, start_date, end_date, metrics)
if not curr_data:
return {
"period_label": period_label,
"current": {},
"previous": None,
"wow_total": None,
"expert": {},
"core_metrics": core_metrics,
"all_metrics": metrics,
"error": f"{period_label} ๆ ๅฏ็จๆฐๆฎ",
}
prev_end = start_date - timedelta(days=1)
prev_start = prev_end - timedelta(days=days_count - 1)
prev_data = aggregate_range_data(df, prev_start, prev_end, metrics)
prev_data = prev_data if prev_data else None
pop_total = None
pop_by_platform = None
pop_by_campaign = None
pop_by_country = None
if prev_data and curr_data.get("total") and prev_data.get("total"):
pop_total = compute_wow(curr_data["total"], prev_data["total"], metrics)
pop_by_platform = _compute_dim_wow(curr_data, prev_data, "by_platform", metrics)
pop_by_campaign = _compute_dim_wow(curr_data, prev_data, "by_campaign_type", metrics)
pop_by_country = _compute_dim_wow(curr_data, prev_data, "by_country", metrics)
try:
yoy_start = start_date.replace(year=start_date.year - 1)
yoy_end = end_date.replace(year=end_date.year - 1)
except ValueError:
yoy_start = yoy_end = None
yoy_data = None
yoy_total = None
yoy_by_platform = None
yoy_by_campaign = None
yoy_by_country = None
yoy_label = None
if yoy_start and yoy_end:
yoy_data = aggregate_range_data(df, yoy_start, yoy_end, metrics)
yoy_data = yoy_data if yoy_data else None
if yoy_data and curr_data.get("total") and yoy_data.get("total"):
yoy_label = f"{yoy_start.strftime('%Y-%m-%d')} ~ {yoy_end.strftime('%Y-%m-%d')}"
yoy_total = compute_wow(curr_data["total"], yoy_data["total"], metrics)
yoy_by_platform = _compute_dim_wow(curr_data, yoy_data, "by_platform", metrics)
yoy_by_campaign = _compute_dim_wow(curr_data, yoy_data, "by_campaign_type", metrics)
yoy_by_country = _compute_dim_wow(curr_data, yoy_data, "by_country", metrics)
prev_label = f"{prev_start.strftime('%Y-%m-%d')} ~ {prev_end.strftime('%Y-%m-%d')}" if prev_data else None
expert = _safe_ai_analysis(
curr_data, prev_data, pop_total, pop_by_platform, pop_by_campaign, pop_by_country,
yoy_data=yoy_data, yoy_total=yoy_total,
yoy_by_platform=yoy_by_platform, yoy_by_country=yoy_by_country,
period_label=period_label,
trend_table=trend_table, memory_context=memory_context,
business_context=business_context,
) if prev_data else {}
return {
"period_label": period_label,
"prev_label": prev_label,
"yoy_label": yoy_label,
"days_count": days_count,
"current": curr_data,
"previous": prev_data,
"yoy": yoy_data,
"wow_total": pop_total,
"wow_by_platform": pop_by_platform,
"wow_by_campaign": pop_by_campaign,
"wow_by_country": pop_by_country,
"yoy_total": yoy_total,
"yoy_by_platform": yoy_by_platform,
"yoy_by_campaign": yoy_by_campaign,
"yoy_by_country": yoy_by_country,
"expert": expert,
"core_metrics": core_metrics,
"all_metrics": metrics,
}
# โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
# ่ถๅฟๅผๆ โ ๆๅๆ่ฟ N ไธชๅ็ฒๅบฆๅจๆ็ๆ ธๅฟๆๆ
# โโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโโ
TREND_METRICS = ["Cost", "Ads.Sales", "ROAS", "CPC", "CTR", "CVR", "GA4.Sessions"]
def build_trend_table(
config_path: Path | None = None,
mode: str = "week",
current_year: int | None = None,
current_period: int | None = None,
current_date: date | None = None,
n: int = 8,
) -> list[dict[str, Any]]:
"""
่ฟๅๆ่ฟ n ไธชๅ็ฒๅบฆๅจๆ็ๆ ธๅฟๆๆ ๅ่กจ๏ผๅซๅฝๆ๏ผ๏ผๆๆถ้ดๅๅบใ
ๆฏไธชๅ
็ด : {"label": "W12", "Cost": ..., "ROAS": ..., ...}
"""
config = load_config(config_path)
metrics = config.get("metrics", [])
df = load_daily_data(config_path)
rows: list[dict[str, Any]] = []
if mode == "week":
weeks = get_available_weeks(df)
if current_year and current_period:
idx = None
for i, (y, w) in enumerate(weeks):
if y == current_year and w == current_period:
idx = i
break
if idx is not None:
weeks = weeks[max(0, idx - n + 1): idx + 1]
else:
weeks = weeks[-n:]
else:
weeks = weeks[-n:]
for y, w in weeks:
data = aggregate_week_data(df, y, w, metrics)
total = data.get("total", {})
row = {"label": f"{y} W{w}"}
for m in TREND_METRICS:
row[m] = total.get(m)
rows.append(row)
elif mode == "month":
from .loader import get_available_months
months = get_available_months(df)
if current_year and current_period:
idx = None
for i, (y, mo) in enumerate(months):
if y == current_year and mo == current_period:
idx = i
break
if idx is not None:
months = months[max(0, idx - n + 1): idx + 1]
else:
months = months[-n:]
else:
months = months[-n:]
for y, mo in months:
data = aggregate_month_data(df, y, mo, metrics)
total = data.get("total", {})
row = {"label": f"{y}ๅนด{mo}ๆ"}
for m in TREND_METRICS:
row[m] = total.get(m)
rows.append(row)
elif mode == "day":
all_dates = sorted(df["date"].dt.date.unique())
if current_date:
try:
idx = all_dates.index(current_date)
all_dates = all_dates[max(0, idx - n + 1): idx + 1]
except ValueError:
all_dates = all_dates[-n:]
else:
all_dates = all_dates[-n:]
for d in all_dates:
data = aggregate_day_data(df, d, metrics)
total = data.get("total", {})
row = {"label": str(d)}
for m in TREND_METRICS:
row[m] = total.get(m)
rows.append(row)
return rows
"""
ๆฐๅญ่ฅ้ไธๅฎถ็บงๅๆๅผๆใ
ๅบๅฑไฝฟ็จๆฅๅบฆๆฐๆฎ๏ผๆ Country ร Platform ร Campaign Type ๅค็ปดๅบฆ่ๅใ
ๆฏๆๅจๅบฆๅๆ๏ผdailyโweekly ่ๅ๏ผๅๆๅบฆๅๆ๏ผๆ่ช็ถๆ่ๅ๏ผใ
"""
from pathlib import Path
from typing import Any
import pandas as pd
from datetime import date, datetime, timedelta
from .loader import load_config, load_daily_data, get_available_weeks, get_latest_week
from .ai_analyzer import generate_ai_analysis
log = __import__("logging").getLogger("ad-agent")
def _safe_ai_analysis(
curr_data, prev_data, wow_total, wow_by_platform, wow_by_campaign, wow_by_country,
**kwargs,
) -> dict[str, Any]:
"""Try AI analysis first; fall back to rule-based generate_expert_analysis on failure."""
try:
result = generate_ai_analysis(
curr_data, prev_data, wow_total, wow_by_platform, wow_by_campaign, wow_by_country,
**kwargs,
)
if result:
return result
except Exception as e:
log.warning(f"AI ๅๆไธๅฏ็จ๏ผๅ้ๅฐ่งๅๅผๆ: {e}")
return generate_expert_analysis(
curr_data, prev_data, wow_by_platform, wow_by_campaign, wow_by_country,
yoy_data=kwargs.get("yoy_data"),
yoy_total=kwargs.get("yoy_total"),
yoy_by_platform=kwargs.get("yoy_by_platform"),
yoy_by_country=kwargs.get("yoy_by_country"),
)
ADDITIVE_METRICS = {
"Cost", "Ads.Sales", "Ads.Sales(1d+7d)", "SPF.Sales",
"GA4.Sessions", "SPF.Sessions", "LPV", "Impressions",
"ATC", "Email", "Video Views",
}
def _safe_div(numerator: float, denominator: float) -> float:
if denominator == 0 or pd.isna(denominator):
return 0.0
return numerator / denominator
def _safe_pct_change(curr: float, prev: float) -> float | None:
if prev == 0 or pd.isna(prev):
return None
return (curr - prev) / prev * 100
def _agg_group(group: pd.DataFrame, metrics: list[str]) -> dict[str, float]:
"""
ๆญฃ็กฎ่ๅ๏ผๅฏๅ ๆๆ ็ดๆฅ sum๏ผๆฏ็ๆๆ ้่ฟ่ฟๅๅๅญๅๆฏ้ๆฐ่ฎก็ฎใ
"""
result: dict[str, float] = {}
g = group.copy()
for m in ADDITIVE_METRICS:
if m in g.columns:
result[m] = g[m].fillna(0).sum()
if "Impressions" in g.columns and "CTR" in g.columns:
g["_clicks"] = g["Impressions"].fillna(0) * g["CTR"].fillna(0)
else:
g["_clicks"] = 0.0
if "CVR" in g.columns and "GA4.Sessions" in g.columns:
g["_conversions"] = g["CVR"].fillna(0) * g["GA4.Sessions"].fillna(0)
else:
g["_conversions"] = 0.0
total_clicks = g["_clicks"].sum()
total_conversions = g["_conversions"].sum()
total_cost = result.get("Cost", 0)
total_sales = result.get("Ads.Sales", 0)
total_imp = result.get("Impressions", 0)
total_sessions = result.get("GA4.Sessions", 0)
total_lpv = result.get("LPV", 0)
total_atc = result.get("ATC", 0)
total_email = result.get("Email", 0)
total_video = result.get("Video Views", 0)
result["ROAS"] = _safe_div(total_sales, total_cost)
result["CTR"] = _safe_div(total_clicks, total_imp)
result["CVR"] = _safe_div(total_conversions, total_sessions)
result["CPC"] = _safe_div(total_cost, total_clicks)
result["CPM"] = _safe_div(total_cost, total_imp) * 1000
result["CPSE"] = _safe_div(total_cost, total_sessions)
result["CPLPV"] = _safe_div(total_cost, total_lpv)
result["CP-ATC"] = _safe_div(total_cost, total_atc)
result["CP-Email"] = _safe_div(total_cost, total_email)