#!/usr/bin/env python3 """ detect_drops.py — анализирует funnel_raw.json (формат после переноса на реальные поля WB), считает дельты vs бейзлайн, фильтрует, локализует первый сломанный этап воронки. Воронка WB: openCount → cartCount → orderCount → buyoutCount. CTR как отдельная метрика НЕ существует. addToCartConversion = % показов → корзина. Все производные считаются от сумм за период (а не как среднее процентов). """ from __future__ import annotations import argparse import json import sys from datetime import date, datetime, timedelta from typing import Any ABS_KEYS = [ "openCount", "cartCount", "orderCount", "orderSum", "buyoutCount", "buyoutSum", "addToWishlistCount", ] # buyoutCount / buyoutPercent НЕ участвуют в drop/growth-критериях: # выкуп на WB фиксируется с задержкой 5-14 дней (заказ → доставка → подтверждение), # поэтому за вчера всегда близок к 0 — это не сигнал. Показываем как справочное поле. PCT_METRICS = [ "openCount", "cartCount", "orderCount", "orderSum", ] PP_METRICS = [ "addToCartConversion", "cartToOrderConversion", ] # Поля справочные (отображаем в отчёте, но не триггерим по ним) INFO_PCT_METRICS = ["buyoutCount"] INFO_PP_METRICS = ["buyoutPercent"] FUNNEL_STAGE_ORDER = [ "openCount", "addToCartConversion", "cartCount", "cartToOrderConversion", "orderCount", "orderSum", ] def safe_div(a: float, b: float) -> float: return (a / b) if b else 0.0 def compute_derived(summed: dict[str, float]) -> dict[str, float]: """Производные от сумм — корректный «средний CR за период».""" opens = summed.get("openCount", 0) or 0 carts = summed.get("cartCount", 0) or 0 orders = summed.get("orderCount", 0) or 0 buyouts = summed.get("buyoutCount", 0) or 0 return { "addToCartConversion": safe_div(carts, opens) * 100, "cartToOrderConversion": safe_div(orders, carts) * 100, "buyoutPercent": safe_div(buyouts, orders) * 100, } def aggregate_days(dates_dict: dict[str, dict[str, float]], date_list: list[str]) -> dict[str, float]: """Per-day средние для абсолютных + производные от сумм.""" if not date_list: return {"_days": 0} summed: dict[str, float] = {m: 0.0 for m in ABS_KEYS} count = 0 for d in date_list: if d not in dates_dict: continue count += 1 for m in ABS_KEYS: summed[m] += dates_dict[d].get(m, 0) or 0 if count == 0: return {"_days": 0} avg = {m: summed[m] / count for m in ABS_KEYS} avg["_days"] = count avg.update(compute_derived(summed)) return avg def compute_today_metrics(day: dict[str, float]) -> dict[str, float]: """Метрики target-дня. Производные пересчитываем из абсолютных, чтобы был единый источник правды (WB иногда округляет проценты).""" out = {k: day.get(k, 0) or 0 for k in ABS_KEYS} out.update(compute_derived(out)) return out def delta_pct(today_val: float, baseline_val: float) -> float | None: if baseline_val == 0: return None if today_val == 0 else float("inf") return (today_val - baseline_val) / baseline_val * 100 def first_broken_stage(deltas: dict[str, dict[str, Any]], drop_pct: float) -> str | None: for stage in FUNNEL_STAGE_ORDER: d = deltas.get(stage) if not d: continue if stage in PCT_METRICS: pct = d.get("pct") if pct is not None and pct != float("inf") and pct <= -drop_pct: return stage elif stage in PP_METRICS: pp = d.get("pp") if pp is not None and pp <= -drop_pct / 4: return stage return None def analyze_sku(nm_id: int, sku_data: dict[str, Any], target_date: str, baseline_dates: list[str], settings: dict[str, Any]) -> dict[str, Any]: card = sku_data.get("card", {}) dates_dict = sku_data.get("dates", {}) created_at = (card.get("createdAt") or "")[:10] if created_at: try: created_dt = datetime.fromisoformat(created_at).date() target_dt = date.fromisoformat(target_date) if (target_dt - created_dt).days < 7: return {"nmID": nm_id, "card": card, "category": "skipped_new", "reason": f"создан {created_at} (< 7 дней)"} except ValueError: pass today_raw = dates_dict.get(target_date) if not today_raw: return {"nmID": nm_id, "card": card, "category": "skipped_no_data", "reason": f"нет данных за {target_date}"} today = compute_today_metrics(today_raw) promo_dates = set(settings.get("promo_dates", [])) effective_baseline_dates = [d for d in baseline_dates if d not in promo_dates and d in dates_dict] baseline_unreliable = len(effective_baseline_dates) < 3 # минимум 3 дня (раньше 4, теперь окно меньше) baseline = aggregate_days(dates_dict, effective_baseline_dates) min_orders = settings.get("min_baseline_orders_per_day", 3) min_revenue = settings.get("min_baseline_revenue_per_day_rub", 5000) base_orders = baseline.get("orderCount", 0) base_revenue = baseline.get("orderSum", 0) if not (base_orders >= min_orders or base_revenue >= min_revenue): return {"nmID": nm_id, "card": card, "category": "skipped_low_volume", "reason": f"бейзлайн {base_orders:.1f} зак/день, {base_revenue:.0f} ₽/день — ниже фильтра"} deltas: dict[str, dict[str, Any]] = {} for m in PCT_METRICS + INFO_PCT_METRICS: deltas[m] = {"today": today.get(m, 0), "baseline": baseline.get(m, 0), "abs": (today.get(m, 0) or 0) - (baseline.get(m, 0) or 0), "pct": delta_pct(today.get(m, 0), baseline.get(m, 0))} for m in PP_METRICS + INFO_PP_METRICS: deltas[m] = {"today": today.get(m, 0), "baseline": baseline.get(m, 0), "pp": (today.get(m, 0) or 0) - (baseline.get(m, 0) or 0)} drop_pct = settings.get("drop_pct", 20) growth_pct = settings.get("growth_pct", 20) zero_rule = settings.get("zero_orders_rule", "oos_section") if today.get("orderCount", 0) == 0 and base_orders > 0: if zero_rule == "hide": return {"nmID": nm_id, "card": card, "category": "skipped_zero_hidden"} if zero_rule == "oos_section": return {"nmID": nm_id, "card": card, "category": "oos_candidate", "today": today, "baseline": baseline, "deltas": deltas, "first_break": "orderCount", "baseline_unreliable": baseline_unreliable, "effective_baseline_days": len(effective_baseline_dates)} # drop_100 → проваливается в обычную обработку ниже drop_flag = growth_flag = False for m in PCT_METRICS: pct = deltas[m].get("pct") if pct is None or pct == float("inf"): continue if pct <= -drop_pct: drop_flag = True if pct >= growth_pct: growth_flag = True first_break = first_broken_stage(deltas, drop_pct) if drop_flag else None if drop_flag and growth_flag: cat = "mixed" elif drop_flag: cat = "drop" elif growth_flag: cat = "growth" else: cat = "ok" return {"nmID": nm_id, "card": card, "category": cat, "today": today, "baseline": baseline, "deltas": deltas, "first_break": first_break, "drop_flag": drop_flag, "growth_flag": growth_flag, "baseline_unreliable": baseline_unreliable, "effective_baseline_days": len(effective_baseline_dates)} def account_overview(all_skus: list[dict[str, Any]]) -> dict[str, Any]: sum_today: dict[str, float] = {k: 0.0 for k in ABS_KEYS} sum_base: dict[str, float] = {k: 0.0 for k in ABS_KEYS} counted = 0 for s in all_skus: if s["category"].startswith("skipped"): continue counted += 1 t = s.get("today") or {} b = s.get("baseline") or {} for k in ABS_KEYS: sum_today[k] += t.get(k, 0) or 0 sum_base[k] += b.get(k, 0) or 0 today_d = dict(sum_today); today_d.update(compute_derived(sum_today)) base_d = dict(sum_base); base_d.update(compute_derived(sum_base)) deltas: dict[str, dict[str, Any]] = {} for m in PCT_METRICS + INFO_PCT_METRICS: deltas[m] = {"today": sum_today.get(m, 0), "baseline": sum_base.get(m, 0), "pct": delta_pct(sum_today.get(m, 0), sum_base.get(m, 0))} for m in PP_METRICS + INFO_PP_METRICS: deltas[m] = {"today": today_d.get(m, 0), "baseline": base_d.get(m, 0), "pp": today_d.get(m, 0) - base_d.get(m, 0)} return {"counted_skus": counted, "today": today_d, "baseline": base_d, "deltas": deltas} def main() -> int: p = argparse.ArgumentParser() p.add_argument("--funnel", required=True) p.add_argument("--config", required=True) p.add_argument("--output", required=True) args = p.parse_args() with open(args.funnel, encoding="utf-8") as f: funnel = json.load(f) with open(args.config, encoding="utf-8") as f: cfg = json.load(f) if funnel.get("status") == "no_data_yet": print("ERROR: funnel_raw has status=no_data_yet", file=sys.stderr) return 2 th = cfg.get("thresholds", {}) settings = { "drop_pct": th.get("drop_pct", 20), "growth_pct": th.get("growth_pct", 20), "min_baseline_orders_per_day": th.get("min_baseline_orders_per_day", 3), "min_baseline_revenue_per_day_rub": th.get("min_baseline_revenue_per_day_rub", 5000), "zero_orders_rule": cfg.get("zero_orders_rule", "oos_section"), "promo_dates": cfg.get("promo_dates", []), } target_date = funnel["target_date"] baseline_dates = funnel["baseline_dates"] all_analyzed: list[dict[str, Any]] = [] for nm_id_str, sku_data in funnel["by_nm"].items(): result = analyze_sku(int(nm_id_str), sku_data, target_date, baseline_dates, settings) all_analyzed.append(result) drops = [s for s in all_analyzed if s["category"] in ("drop", "mixed")] growths = [s for s in all_analyzed if s["category"] in ("growth", "mixed")] oos = [s for s in all_analyzed if s["category"] == "oos_candidate"] skipped = [s for s in all_analyzed if s["category"].startswith("skipped")] ok = [s for s in all_analyzed if s["category"] == "ok"] drops.sort(key=lambda s: -((s.get("deltas") or {}).get("orderSum", {}).get("abs") or 0)) growths.sort(key=lambda s: ((s.get("deltas") or {}).get("orderSum", {}).get("abs") or 0), reverse=True) oos.sort(key=lambda s: (s.get("baseline") or {}).get("orderSum", 0), reverse=True) out = { "target_date": target_date, "baseline_dates": baseline_dates, "settings_used": settings, "account_overview": account_overview(all_analyzed), "drops": drops, "growths": growths, "oos_candidates": oos, "ok_skus": ok, "skipped": skipped, "totals": { "drops": len(drops), "growths": len(growths), "oos_candidates": len(oos), "ok": len(ok), "skipped": len(skipped), "skipped_new": sum(1 for s in skipped if s["category"] == "skipped_new"), "skipped_low_volume": sum(1 for s in skipped if s["category"] == "skipped_low_volume"), }, } # Компактный JSON: indent=2 раздувал файл в 2-3 раза без всякой пользы для downstream with open(args.output, "w", encoding="utf-8") as f: json.dump(out, f, ensure_ascii=False, separators=(",", ":")) print(f"[detect] drops={len(drops)} growths={len(growths)} oos={len(oos)} ok={len(ok)} skipped={len(skipped)}", file=sys.stderr) return 0 if __name__ == "__main__": sys.exit(main())