"""Customers module — the "why" behind the brand divergence (Fisch −5.5% vs Royal +40%). Centerpiece is the **customer revenue bridge**: it decomposes the YoY revenue change into New (+), Expansion (+), Contraction (−) and Lost (−). That decomposition both answers the business question and self-validates — the four components must reconcile last-period total to this-period total exactly (built into validate()). Plus a MECE value-tier segmentation (LTM), an at-risk / win-back list ranked by dollars at stake, and new/lost customer lists. All order-level (sale.order), RI+FFS scope, excluded accounts removed — reusing the Sales module's order_domain so scope is identical everywhere. """ import sys import datetime as dt import statistics from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1])) import core.odoo as O import core.periods as P import modules.sales as sales_mod # ---- per-customer grouped reads: STORE-backed (OM-2 retrofit 2026-07-12) with LIVE fallback -- # _cust_rev/_last_order_dates are the choke points behind the Customers bundle, the Map's slow # wave-1 and the Customer List queues. SQL mirrors sales.order_domain incl. the agent filter # (partner_ids; an EMPTY set matches no orders, same as the live domain). validate() stays live. USE_STORE = True def _cust_group_store(date_from, date_to, team_id, agent_pids, select): import harness.datastore as DS params, w = [], ["state IN ('sale','done')"] teams = sales_mod.TEAMS(team_id) w.append("team_id IN (" + ",".join("?" * len(teams)) + ")") params += list(teams) if date_from: w.append("CAST(date_order AS TIMESTAMP) >= CAST(? AS TIMESTAMP)") params.append(f"{date_from} 00:00:00") if date_to: w.append("CAST(date_order AS TIMESTAMP) <= CAST(? AS TIMESTAMP)") params.append(f"{date_to} 23:59:59") ex = O.excluded_partner_ids() if ex: w.append("partner_id NOT IN (" + ",".join("?" * len(ex)) + ")") params += list(ex) if agent_pids is not None: if not agent_pids: return [] # empty set matches no orders w.append("partner_id IN (" + ",".join("?" * len(agent_pids)) + ")") params += list(agent_pids) if O.doc_mode() == 'invoice': w.append("invoice_status = 'invoiced'") return DS.ro_con().execute( f"SELECT o.partner_id, coalesce(p.name, '#' || o.partner_id), {select} " "FROM sale_order o LEFT JOIN res_partner p ON p.id = o.partner_id " "WHERE " + " AND ".join(w) + " GROUP BY 1, 2", params).fetchall() def _cust_rev(date_from, date_to, team_id=None, agent_pids=None): """{partner_id: {'name', 'rev', 'orders'}} over a window (RI+FFS, excluded accounts removed). agent_pids (frozenset|None) restricts to one Agent's customers — the module-wide Agent filter.""" if USE_STORE: try: rows = _cust_group_store(date_from, date_to, team_id, agent_pids, 'sum(amount_untaxed), count(*)') return {r[0]: {'name': r[1], 'rev': r[2] or 0.0, 'orders': r[3]} for r in rows if r[0]} except Exception: pass g = O.read_group('sale.order', sales_mod.order_domain(date_from, date_to, team_id, partner_ids=agent_pids), ['amount_untaxed:sum'], ['partner_id'], lazy=False) out = {} for r in g: pid = O.m2o_id(r.get('partner_id')) if not pid: continue out[pid] = {'name': O.m2o_name(r.get('partner_id')), 'rev': r.get('amount_untaxed') or 0.0, 'orders': r.get('__count') or 0} return out # ----------------------------------------------------------- agent filter (module-wide) # Agent = res.partner.agent_ids (the assigned sales agent on a customer; the same attribute the # rollups slice by). The Customer module can be filtered to one agent's whole book: we resolve the # agent name → the set of partner ids assigned to them, then pass it as agent_pids everywhere. def agent_options(t=None, team_id=None): """Agent names that have customers with YTD activity in scope — for the module filter dropdown. Reuses the agent rollup (cheap, cached at the app layer). '(none)' = unassigned customers.""" return [r['group'] for r in by_dimension('agent', t, team_id=team_id)] def agent_partner_ids(agent_name, customers_only=False): """frozenset of partner ids assigned to `agent_name` via res.partner.agent_ids. None when no agent is selected ('All agents'); an EMPTY frozenset (matches nobody) when the agent has no customers. '(none)' resolves to customers with no agent assigned. ⭐ WAVE 20 (R3, DEBT D-30) — `customers_only` IS THE EXPLICIT POLICY THIS FUNCTION WAS MISSING, and the two callers genuinely want different answers: * **Permission scope (default, `False`)** — an agent-LOGIN user's whole app is bounded by this set, so it must be GENEROUS: archived accounts and Odoo address records included. Narrowing it would hide an agent's own data from them, and the fails-closed trap in the note below is what that costs. * **Display/reporting (`True`)** — "how many accounts does Martin have" must answer what Odoo answers. MEASURED 2026-08-05: the generous set is 503 for Martin and the honest one is **494**, the owner's number; the 9-row gap is entirely `type in (delivery, other)` ADDRESS records that ride the pool because they appear on orders, two of them with no name at all. Both are the SAME m2m-contains resolver — which is D-30's actual requirement. The bug was never the generosity; it was that the Agent COLUMN used a THIRD rule (`agent_ids[0]`, first agent only) that agreed with neither, so an admin filtering `Agent = X` and X's own login saw different books. The column now lists every agent on the partner and this states its policy out loud, so the two can be reconciled by reading them instead of by measuring them. INCLUDES INACTIVE/archived partners (active in [True,False]) — the agent's book is the whole book (dormant + archived accounts too), which is what the Agents page promises AND what an AGENT-LOGIN user must see as their complete, isolated book. Active-only used to drop archived customers entirely (e.g. an agent whose whole small book is archived resolved to an EMPTY book — the fails-closed trap). Resolves the agent name to EVERY matching partner id (not just the first) so a same-name collision can't silently mis-scope the book (owner 2026-07-21).""" if not agent_name or agent_name in ('All agents', 'All'): return None _all = [('active', 'in', [True, False])] # The display policy: a real customer record, still active. See the docstring for the # measured 503-vs-494 this closes. _qual = [('customer_rank', '>', 0), ('active', '=', True)] if customers_only else _all if agent_name == '(none)': rows = O.search_read('res.partner', [('customer_rank', '>', 0), ('agent_ids', '=', False)] + _all, ['id'], limit=100000) return frozenset(r['id'] for r in rows) ag = O.search_read('res.partner', [('name', '=', agent_name)] + _all, ['id'], limit=10) if not ag: return frozenset() rows = O.search_read('res.partner', [('agent_ids', 'in', [a['id'] for a in ag])] + _qual, ['id'], limit=100000) return frozenset(r['id'] for r in rows) # ----------------------------------------------------------- revenue bridge def revenue_bridge(t=None, team_id=None, agent_pids=None): """Decompose YTD-vs-same-period-LY revenue change into New/Expansion/Contraction/Lost.""" t = t or P.today() yf, yt = P.ytd(t) lf, lt = P.ytd_last_year(t) this = _cust_rev(yf, yt, team_id, agent_pids) last = _cust_rev(lf, lt, team_id, agent_pids) tset, lset = set(this), set(last) new_ids = tset - lset lost_ids = lset - tset both = tset & lset new_rev = sum(this[p]['rev'] for p in new_ids) lost_rev = sum(last[p]['rev'] for p in lost_ids) expansion = sum(this[p]['rev'] - last[p]['rev'] for p in both if this[p]['rev'] > last[p]['rev']) contraction = sum(this[p]['rev'] - last[p]['rev'] for p in both if this[p]['rev'] < last[p]['rev']) this_total = sum(v['rev'] for v in this.values()) last_total = sum(v['rev'] for v in last.values()) return { 'last_total': last_total, 'this_total': this_total, 'change': this_total - last_total, 'new': {'rev': new_rev, 'n': len(new_ids)}, 'expansion': {'rev': expansion, 'n': sum(1 for p in both if this[p]['rev'] > last[p]['rev'])}, 'contraction': {'rev': contraction, 'n': sum(1 for p in both if this[p]['rev'] < last[p]['rev'])}, 'lost': {'rev': -lost_rev, 'n': len(lost_ids)}, 'retained_n': len(both), 'active_this': len(tset), 'active_last': len(lset), } def bridge_by_brand(t=None, agent_pids=None): rows = [] for tid in O.TEAM_IDS: b = revenue_bridge(t, team_id=tid, agent_pids=agent_pids) rows.append({'brand': O.TEAM_NAMES[tid], **{ 'last': b['last_total'], 'this': b['this_total'], 'change': b['change'], 'new': b['new']['rev'], 'expansion': b['expansion']['rev'], 'contraction': b['contraction']['rev'], 'lost': b['lost']['rev'], 'new_n': b['new']['n'], 'lost_n': b['lost']['n']}}) return rows def bridge_component_customers(component, t=None, team_id=None, agent_pids=None): """The customers contributing to one revenue-bridge component (New / Expansion / Contraction / Lost), each with LY, YTD, the bridge delta and share of that component — for the click-through drawer off the Revenue-bridge chart.""" t = t or P.today() yf, yt = P.ytd(t) lf, lt = P.ytd_last_year(t) this = _cust_rev(yf, yt, team_id, agent_pids) last = _cust_rev(lf, lt, team_id, agent_pids) tset, lset = set(this), set(last) comp = (component or '').strip().lower() rows = [] if comp == 'new': rows = [{'pid': p, 'customer': this[p]['name'], 'rev_ly': 0.0, 'rev_ytd': this[p]['rev'], 'change': this[p]['rev']} for p in (tset - lset)] elif comp == 'lost': rows = [{'pid': p, 'customer': last[p]['name'], 'rev_ly': last[p]['rev'], 'rev_ytd': 0.0, 'change': -last[p]['rev']} for p in (lset - tset)] elif comp in ('expansion', 'contraction'): up = comp == 'expansion' for p in (tset & lset): d = this[p]['rev'] - last[p]['rev'] if (d > 0) == up and d != 0: rows.append({'pid': p, 'customer': this[p]['name'], 'rev_ly': last[p]['rev'], 'rev_ytd': this[p]['rev'], 'change': d}) denom = sum(abs(r['change']) for r in rows) or 1.0 for r in rows: r['at_risk'] = max(0.0, r['rev_ly'] - r['rev_ytd']) r['pct_of_component'] = abs(r['change']) / denom * 100.0 rows.sort(key=lambda r: -abs(r['change'])) _attach_attrs(rows) return {'component': component, 'rows': rows, 'n': len(rows), 'total': sum(r['change'] for r in rows)} def period_customers(ym, t=None, team_id=None, agent_pids=None): """Customers who purchased in calendar month `ym` (YYYY-MM) this year — that month's revenue, the same month last year, order count, and each customer's YTD-vs-LY at-risk for context. Powers the click-through drawer off the Monthly (this-year vs last-year) bar chart.""" t = t or P.today() y, m = int(ym[:4]), int(ym[5:7]) def _bounds(yr): s = dt.date(yr, m, 1) e = dt.date(yr + (m // 12), (m % 12) + 1, 1) - dt.timedelta(days=1) return s.isoformat(), e.isoformat() ts, te = _bounds(y) ls, le = _bounds(y - 1) this = _cust_rev(ts, te, team_id, agent_pids) last = _cust_rev(ls, le, team_id, agent_pids) yf, yt = P.ytd(t) lf, lt = P.ytd_last_year(t) ytd = _cust_rev(yf, yt, team_id, agent_pids) lytd = _cust_rev(lf, lt, team_id, agent_pids) rows = [{'pid': p, 'customer': v['name'], 'rev_this': v['rev'], 'orders': v['orders'], 'rev_ly': last.get(p, {}).get('rev', 0.0), 'at_risk': max(0.0, lytd.get(p, {}).get('rev', 0.0) - ytd.get(p, {}).get('rev', 0.0))} for p, v in this.items()] rows.sort(key=lambda r: -r['rev_this']) _attach_attrs(rows) return {'month': ym, 'rows': rows, 'n': len(rows), 'rev_this': sum(r['rev_this'] for r in rows), 'rev_ly': sum(r['rev_ly'] for r in rows)} def monthly_kpis(t=None, team_id=None, n_months=24): """Monthly time-series for the acquisition/engagement trends at the top of the page: order count, AOV, distinct purchasing customers, and newly-acquired customers (first-ever order that month).""" t = t or P.today() months = P.month_starts(n_months, t) win_start = months[0][1] dom = sales_mod.order_domain(win_start, t.isoformat(), team_id) def _ym(r): rng = (r.get('__range') or {}).get('date_order:month') or {} return (rng.get('from') or '')[:7] mrev, morders = {}, {} for r in O.read_group('sale.order', dom, ['amount_untaxed:sum'], ['date_order:month'], lazy=False): ym = _ym(r) if ym: mrev[ym] = r.get('amount_untaxed') or 0.0 morders[ym] = r.get('__count') or 0 mbuyers = {} for r in O.read_group('sale.order', dom, ['__count'], ['date_order:month', 'partner_id'], lazy=False): ym = _ym(r) if ym: mbuyers[ym] = mbuyers.get(ym, 0) + 1 # new customers: first-ever order month per partner (all history), bucketed into our window mnew = {} for r in O.read_group('sale.order', sales_mod.order_domain('2000-01-01', t.isoformat(), team_id), ['date_order:min'], ['partner_id'], lazy=False): ym = str(r.get('date_order') or '')[:7] if ym: mnew[ym] = mnew.get(ym, 0) + 1 out = [] for ym, _s, _e in months: orders = morders.get(ym, 0) out.append({'month': ym, 'orders': orders, 'revenue': mrev.get(ym, 0.0), 'aov': (mrev.get(ym, 0.0) / orders) if orders else 0.0, 'buyers': mbuyers.get(ym, 0), 'new_customers': mnew.get(ym, 0)}) return out def customer_trends(t=None, team_id=None, agent_pids=None): """12 months aligned THIS-year vs same-month-LAST-year, with both the monthly value and the running CUMULATIVE (this vs last) for orders, AOV, purchasing customers (distinct) and new customers. Powers the Trends chart's Monthly-bars / Cumulative-line toggle.""" t = t or P.today() months = P.month_starts(24, t) # 24 ascending: prior 12 then recent 12 win_start = months[0][1] dom = sales_mod.order_domain(win_start, t.isoformat(), team_id, partner_ids=agent_pids) def _ym(r): rng = (r.get('__range') or {}).get('date_order:month') or {} return (rng.get('from') or '')[:7] mrev, morders, mbuyers = {}, {}, {} for r in O.read_group('sale.order', dom, ['amount_untaxed:sum'], ['date_order:month'], lazy=False): ym = _ym(r) if ym: mrev[ym] = r.get('amount_untaxed') or 0.0 morders[ym] = r.get('__count') or 0 for r in O.read_group('sale.order', dom, ['__count'], ['date_order:month', 'partner_id'], lazy=False): ym = _ym(r) pid = O.m2o_id(r.get('partner_id')) if ym and pid: mbuyers.setdefault(ym, set()).add(pid) mnew = {} for r in O.read_group('sale.order', sales_mod.order_domain('2000-01-01', t.isoformat(), team_id, partner_ids=agent_pids), ['date_order:min'], ['partner_id'], lazy=False): ym = str(r.get('date_order') or '')[:7] if ym: mnew[ym] = mnew.get(ym, 0) + 1 recent = [m[0] for m in months[12:24]] # the 12 most recent months (ascending) prior = [m[0] for m in months[0:12]] # the same 12 months a year earlier out = [] cr_t = co_t = cn_t = 0.0 cr_l = co_l = cn_l = 0.0 cb_t, cb_l = set(), set() for i in range(12): ty, ly = recent[i], prior[i] ot, rt, nt = morders.get(ty, 0), mrev.get(ty, 0.0), mnew.get(ty, 0) ol, rl, nl = morders.get(ly, 0), mrev.get(ly, 0.0), mnew.get(ly, 0) bt, bl = mbuyers.get(ty, set()), mbuyers.get(ly, set()) co_t += ot; cr_t += rt; cn_t += nt; cb_t |= bt co_l += ol; cr_l += rl; cn_l += nl; cb_l |= bl out.append({ 'month': ty, 'month_ly': ly, 'orders': ot, 'orders_ly': ol, 'aov': (rt / ot if ot else 0.0), 'aov_ly': (rl / ol if ol else 0.0), 'buyers': len(bt), 'buyers_ly': len(bl), 'new_customers': nt, 'new_customers_ly': nl, 'orders_cum': co_t, 'orders_cum_ly': co_l, 'aov_cum': (cr_t / co_t if co_t else 0.0), 'aov_cum_ly': (cr_l / co_l if co_l else 0.0), 'buyers_cum': len(cb_t), 'buyers_cum_ly': len(cb_l), 'new_customers_cum': cn_t, 'new_customers_cum_ly': cn_l, }) return out # ----------------------------------------------------------- segmentation (MECE) TIERS = [('Whale (≥$25k)', 25000), ('Large ($10–25k)', 10000), ('Mid ($2–10k)', 2000), ('Small (<$2k)', 0)] def _tier(rev): for name, lo in TIERS: if rev >= lo: return name return TIERS[-1][0] def segments(t=None, team_id=None, agent_pids=None): """MECE value-tier segmentation over LTM revenue (each active customer in one tier).""" t = t or P.today() lf, lt = P.ltm(t) cust = _cust_rev(lf, lt, team_id, agent_pids) agg = {name: {'segment': name, 'customers': 0, 'revenue': 0.0} for name, _ in TIERS} for v in cust.values(): a = agg[_tier(v['rev'])] a['customers'] += 1 a['revenue'] += v['rev'] total_rev = sum(a['revenue'] for a in agg.values()) or 1.0 total_n = sum(a['customers'] for a in agg.values()) or 1 rows = [] for name, _ in TIERS: a = agg[name] a['rev_share'] = a['revenue'] / total_rev * 100 a['cust_share'] = a['customers'] / total_n * 100 a['avg_rev'] = a['revenue'] / a['customers'] if a['customers'] else 0.0 rows.append(a) return rows def tier_customers(tier, t=None, team_id=None, agent_pids=None): """The customers in ONE LTM value tier — each with LTM revenue, order count, units bought and average order value — for the Value-tiers drill. Tier membership matches segments() exactly (same LTM window + _tier).""" t = t or P.today() lf, lt = P.ltm(t) cust = _cust_rev(lf, lt, team_id, agent_pids) pids = [pid for pid, v in cust.items() if _tier(v['rev']) == tier] if not pids: return [] units = {} for r in O.read_group('sale.order.line', O.sale_line_domain(lf, lt, team_id, partner_ids=pids), ['product_uom_qty:sum'], ['order_partner_id'], lazy=False): pid = O.m2o_id(r.get('order_partner_id')) if pid: units[pid] = r.get('product_uom_qty') or 0.0 pdata = {p['id']: p for p in O.search_read('res.partner', [('id', 'in', pids)], ['city', 'state_id'])} rows = [] for pid in pids: v = cust[pid] o = v.get('orders', 0) p = pdata.get(pid, {}) rows.append({'pid': pid, 'customer': v['name'], 'city': p.get('city') or '(none)', 'state': O.m2o_name(p.get('state_id')) or '(none)', 'agent': '(none)', 'rev_ltm': v['rev'], 'orders': o, 'units': units.get(pid, 0.0), 'aov': (v['rev'] / o) if o else 0.0}) rows.sort(key=lambda x: -x['rev_ltm']) return rows # ----------------------------------------------------------- at-risk / win-back def at_risk(t=None, team_id=None, limit=30, min_prior=2000.0, agent_pids=None): """Customers who bought materially last year but are down or gone this year, ranked by dollars at risk (last − this). The win-back target list.""" t = t or P.today() yf, yt = P.ytd(t) lf, lt = P.ytd_last_year(t) this = _cust_rev(yf, yt, team_id, agent_pids) last = _cust_rev(lf, lt, team_id, agent_pids) rows = [] for pid, lv in last.items(): if lv['rev'] < min_prior: continue tv = this.get(pid, {'rev': 0.0}) down = lv['rev'] - tv['rev'] if down <= 0: continue rows.append({'pid': pid, 'customer': lv['name'], 'rev_ly': lv['rev'], 'rev_ytd': tv['rev'], 'orders': tv.get('orders', 0), 'at_risk': down, 'status': 'Lost' if pid not in this else 'Declining'}) rows.sort(key=lambda x: -x['at_risk']) return rows[:limit] def _first_order_dates(team_id=None, t=None, agent_pids=None): """{pid: first-ever confirmed order date (ISO)} — one all-history read_group(min).""" t = t or P.today() g = O.read_group('sale.order', sales_mod.order_domain('2000-01-01', t.isoformat(), team_id, partner_ids=agent_pids), ['date_order:min'], ['partner_id'], lazy=False) out = {} for r in g: pid = O.m2o_id(r.get('partner_id')) if pid and r.get('date_order'): out[pid] = str(r['date_order'])[:10] return out def new_customers(t=None, team_id=None, limit=30, agent_pids=None): """Customers acquired or reactivated this year (active this YTD, not in same-period LY), each tagged New (first-ever order this year) vs Reactivated (ordered in a prior year, had lapsed).""" t = t or P.today() yf, yt = P.ytd(t) lf, lt = P.ytd_last_year(t) this = _cust_rev(yf, yt, team_id, agent_pids) last = _cust_rev(lf, lt, team_id, agent_pids) cand = [pid for pid in this if pid not in last] firsts = _first_order_dates(team_id, t, agent_pids) if cand else {} rows = [] for pid in cand: v = this[pid] first = firsts.get(pid, '') status = 'New' if first[:4] == str(t.year) else 'Reactivated' rows.append({'pid': pid, 'customer': v['name'], 'rev_ytd': v['rev'], 'orders': v['orders'], 'first_order': first, 'status': status}) rows.sort(key=lambda x: -x['rev_ytd']) return rows[:limit] def lists_bundle(t=None, team_id=None, agent_pids=None): """At-risk, New/reactivated and Follow-up lists, all enriched with a CONSISTENT insight column set — YTD $, YoY %, orders, last order, days overdue vs the customer's own cadence, agent — plus each list's own metric (at-risk $ / status / est. missed $).""" t = t or P.today() cad = _cadence_bulk(t, team_id, agent_pids=agent_pids) ar = at_risk(t, team_id, agent_pids=agent_pids) nw = new_customers(t, team_id, agent_pids=agent_pids) fu = contact_recommendations(t, team_id, agent_pids=agent_pids) pids = {r['pid'] for r in ar} | {r['pid'] for r in nw} | {r['pid'] for r in fu} attrs = _partner_attrs(list(pids)) def enrich(rows): for r in rows: c = cad.get(r['pid'], {}) r['last_order'] = r.get('last_order') or c.get('last_order', '') r['overdue_days'] = c.get('overdue_days') r['agent'] = (attrs.get(r['pid']) or {}).get('agent', '(none)') if 'rev_ytd' not in r and 'ltm_rev' in r: r['rev_ytd'] = r['ltm_rev'] if 'yoy_pct' not in r: r['yoy_pct'] = P.yoy_pct(r.get('rev_ytd', 0.0), r.get('rev_ly', 0.0)) return rows return {'at_risk': enrich(ar), 'new': enrich(nw), 'followups': enrich(fu)} def _cadence_bulk(t=None, team_id=None, months=24, agent_pids=None): """{pid: {n_orders, last_order, typical_gap_days, days_since, overdue_days, aov}} from ONE read_group over the window (count + min/max date + revenue per partner). Bulk approximation of per-customer cadence — the basis for the follow-up list and the lists' cadence columns.""" t = t or P.today() months = max(1, months) start = dt.date(t.year - (months // 12) - (1 if t.month <= (months % 12) else 0), ((t.month - 1 - (months % 12)) % 12) + 1, 1) g = O.read_group('sale.order', sales_mod.order_domain(start.isoformat(), t.isoformat(), team_id, partner_ids=agent_pids), ['amount_untaxed:sum', 'mind:min(date_order)', 'maxd:max(date_order)'], ['partner_id'], lazy=False) out = {} for r in g: pid = O.m2o_id(r.get('partner_id')) if not pid: continue n = r.get('__count') or 0 rev = r.get('amount_untaxed') or 0.0 try: dmin = dt.date.fromisoformat(str(r.get('mind'))[:10]) dmax = dt.date.fromisoformat(str(r.get('maxd'))[:10]) except (TypeError, ValueError): continue span = (dmax - dmin).days gap = (span / (n - 1)) if n >= 2 and span > 0 else None days_since = (t - dmax).days overdue = (days_since - gap) if gap else None out[pid] = {'n_orders': n, 'last_order': dmax.isoformat(), 'typical_gap_days': gap, 'days_since': days_since, 'overdue_days': overdue, 'aov': (rev / n) if n else 0.0} return out def contact_recommendations(t=None, team_id=None, limit=40, months=24, agent_pids=None): """Prioritised follow-up list: customers overdue against their OWN purchase cadence, ranked by estimated missed revenue (how far past due × their average order value). Reps' call list.""" t = t or P.today() cad = _cadence_bulk(t, team_id, months, agent_pids=agent_pids) attrs = _partner_attrs(list(cad)) names = {} # names from the LTM revenue map (cheap, already scoped) lf, lt = P.ltm(t) rev_map = _cust_rev(lf, lt, team_id, agent_pids) rows = [] for pid, c in cad.items(): gap = c['typical_gap_days'] if c['n_orders'] < 3 or not gap or gap <= 0: continue # need a real, repeated cadence overdue = c['overdue_days'] if overdue is None or overdue <= 0: continue # only those past due if c['days_since'] > 365: continue # long-dead → not a cadence follow-up ltm_rev = rev_map.get(pid, {}).get('rev', 0.0) cycles_missed = overdue / gap # how many reorder cycles past due # cap at 3 cycles so a long-churned account (belongs in win-back) doesn't dominate the # "call them now" list; this rewards valuable regulars who are RECENTLY due. est_missed = min(cycles_missed, 3.0) * c['aov'] rows.append({ 'pid': pid, 'customer': rev_map.get(pid, {}).get('name', '?'), 'last_order': c['last_order'], 'typical_gap_days': gap, 'overdue_days': overdue, 'ltm_rev': ltm_rev, 'orders': c['n_orders'], 'est_missed': est_missed, 'agent': (attrs.get(pid) or {}).get('agent', '(none)'), }) rows.sort(key=lambda x: -x['est_missed']) return rows[:limit] # ====================================================================== EXPLORER # Granular (per-customer drill-down) + sum-level (filter/group by city, state, etc). # All revenue is order-level (sale.order, amount_untaxed) in the same RI+FFS scope as the # rest of Sales, so every rollup sums back to the headline number (proven in validate()). # Group-by dimensions for the sum-level rollups (key → display label). DIMENSIONS = { 'city': 'City', 'state': 'State / Region', 'country': 'Country', 'agent': 'Agent', 'segment': 'Value tier', } def _partner_attrs(pids): """{pid: {street, street2, city, state, country, agent, zip, payment_terms, customer_since, tags, pricelist}} — the customer attributes the sum-level rollups slice by and the Customer table displays. Agent is res.partner.agent_ids (the assigned sales agent; ~one per customer), NOT Odoo's user_id 'salesperson' — MEASURED 2026-07-27 at 26 of 1,548 (2%), which is why agent_ids has always been the field here. Odoo's `credit_limit` is not read for the same reason: 20 of 1,548 (1%). Credit EXPOSURE comes from AR (modules/collections.py), not from that field. Every value collapses to '(none)' when blank (MECE), so a group-by has no null bucket and a filter has something to match. ⚠ `zip` is TEXT, never numeric: postal codes carry leading zeros, and 01730 read as a number is 1730 — a different town. """ pids = list(pids) if not pids: return {} rows = O.search_read('res.partner', [('id', 'in', pids)], ['street', 'street2', 'city', 'state_id', 'country_id', 'agent_ids', 'zip', 'property_payment_term_id', 'create_date', 'category_id', 'property_product_pricelist']) aids = {a for r in rows for a in (r.get('agent_ids') or [])} anames = ({p['id']: p['name'] for p in O.search_read('res.partner', [('id', 'in', list(aids))], ['name'])} if aids else {}) # Tags are m2m: ids on the partner, names in res.partner.category. One extra read for the # whole set rather than one per customer. tids = {t for r in rows for t in (r.get('category_id') or [])} tnames = ({c['id']: c['name'] for c in O.search_read('res.partner.category', [('id', 'in', list(tids))], ['name'])} if tids else {}) out = {} for r in rows: ag = r.get('agent_ids') or [] city = (r.get('city') or '').strip() tags = [tnames.get(t) for t in (r.get('category_id') or []) if tnames.get(t)] # create_date is a DATETIME ('2024-01-15 10:23:45'); the column is a DATE, and the grid # compares dates ISO-LEXICALLY, so a trailing time would sort and filter as text noise. since = (r.get('create_date') or '') out[r['id']] = { # ⭐ W30-T33 (carried W29-T51) — the STREET half of the address. `city`/`state`/`zip`/ # `country` were already here; the two street lines were read nowhere on the customer # path, only by `modules/map.py` for geocoding, and a geocoder's input never reached # the grid. ⚠ NOT ROUTED THROUGH `map.py`: these come off `res.partner` directly, so a # customer the geocoder could not place still shows its address. # ⚠ `.strip()` WITHOUT `.title()`, unlike `city`. A street line carries unit numbers, # directionals and abbreviations ("123 NW 4TH ST APT 2B"), and title-casing turns # those into "123 Nw 4Th St Apt 2B" — a mangling `city` does not risk because a city # is one word-set of ordinary nouns. 'street': (r.get('street') or '').strip() or '(none)', 'street2': (r.get('street2') or '').strip() or '(none)', 'city': city.title() if city else '(none)', 'state': O.m2o_name(r.get('state_id')) or '(none)', 'country': O.m2o_name(r.get('country_id')) or '(none)', # ⭐ WAVE 20 (R3, closes DEBT D-30) — EVERY agent on the partner, not just the first. # # This was `anames.get(ag[0])`, and it made TWO definitions of "an agent's book" that # disagreed: `agent_partner_ids()` (which scopes an agent-LOGIN user's whole app) is # m2m-CONTAINS, while this column named only `agent_ids[0]`. So an admin filtering # `Agent = X` and X's own login saw different sets — MEASURED book-wide 2026-08-05: # Tara Devon Gallager 9 rows, Sang Ching 6, Moishe Rubenstein 1, Martin Pasternak 1. # # Joined with ', ' rather than kept as a list because the column is a TEXT field the # grid groups and filters on; `contains` then matches any agent on a shared account, # which is the m2m question asked in the vocabulary the column already speaks. 'agent': ', '.join(n for n in (anames.get(a) for a in ag) if n) or '(none)', 'zip': (r.get('zip') or '').strip() or '(none)', 'payment_terms': O.m2o_name(r.get('property_payment_term_id')) or '(none)', 'customer_since': str(since)[:10] if since else '', # The FULL datetime, for the grid's `created_time` field type (wave-5 item 11) — # customer_since above stays the DATE the date-typed column sorts/filters on. 'created_at': str(since) if since else '', 'tags': ', '.join(tags) if tags else '(none)', 'pricelist': O.m2o_name(r.get('property_product_pricelist')) or '(none)', } return out def _attach_attrs(rows, key='pid'): """Enrich a list of customer rows (each carrying a partner id under `key`) in place with the standardized city / state / agent fields — so every customer list (page or drawer) can show the same filters/columns. Returns the same list.""" pids = [r[key] for r in rows if r.get(key) is not None] if not pids: return rows attrs = _partner_attrs(pids) for r in rows: a = attrs.get(r.get(key), {}) r.setdefault('city', a.get('city', '(none)')) r.setdefault('state', a.get('state', '(none)')) r.setdefault('agent', a.get('agent', '(none)')) return rows def _last_order_dates(date_from, date_to, team_id=None, agent_pids=None): """{pid: last order date (ISO)} within the window — for the directory's recency column.""" if USE_STORE: try: rows = _cust_group_store(date_from, date_to, team_id, agent_pids, 'max(date_order)') return {r[0]: str(r[2])[:10] for r in rows if r[0] and r[2]} except Exception: pass g = O.read_group('sale.order', sales_mod.order_domain(date_from, date_to, team_id, partner_ids=agent_pids), ['date_order:max'], ['partner_id'], lazy=False) out = {} for r in g: pid = O.m2o_id(r.get('partner_id')) if pid and r.get('date_order'): out[pid] = str(r['date_order'])[:10] return out def directory(t=None, team_id=None, limit=None, agent_pids=None): """Per-customer table (YTD): revenue, YoY, orders, AOV, last order, city/state/salesperson. Basis for the drill-down picker and (indirectly) the rollups. Sorted by revenue desc.""" t = t or P.today() yf, yt = P.ytd(t) lf, lt = P.ytd_last_year(t) this = _cust_rev(yf, yt, team_id, agent_pids) last = _cust_rev(lf, lt, team_id, agent_pids) attrs = _partner_attrs(set(this) | set(last)) lastord = _last_order_dates(yf, yt, team_id, agent_pids) rows = [] for pid, v in this.items(): a = attrs.get(pid, {}) rev, orders = v['rev'], v['orders'] ly = last.get(pid, {}).get('rev', 0.0) rows.append({ 'pid': pid, 'customer': v['name'], 'revenue': rev, 'revenue_ly': ly, 'yoy_pct': P.yoy_pct(rev, ly), 'orders': orders, 'aov': (rev / orders) if orders else 0.0, 'last_order': lastord.get(pid, ''), 'city': a.get('city', '(none)'), 'state': a.get('state', '(none)'), 'agent': a.get('agent', '(none)'), }) rows.sort(key=lambda x: -x['revenue']) return rows[:limit] if limit else rows def by_dimension(dim, t=None, team_id=None, agent_pids=None): """Sum-level rollup: YTD revenue (vs LY) grouped by a customer attribute. dim is a key of DIMENSIONS. MECE — each active customer lands in exactly one group, so Σ(groups) == total YTD revenue (verified in validate()). When consolidated (team_id=None) each group also carries its Fisch + Royal revenue, mirroring the HQ rollup down to each brand. Sorted desc.""" t = t or P.today() yf, yt = P.ytd(t) lf, lt = P.ytd_last_year(t) this = _cust_rev(yf, yt, team_id, agent_pids) last = _cust_rev(lf, lt, team_id, agent_pids) attrs = _partner_attrs(set(this) | set(last)) split = team_id is None f_rev = _cust_rev(yf, yt, 5, agent_pids) if split else {} r_rev = _cust_rev(yf, yt, 6, agent_pids) if split else {} def keyfor(pid, rev): if dim == 'segment': return _tier(rev) return (attrs.get(pid) or {}).get(dim) or '(none)' agg = {} def bucket(k): return agg.setdefault(k, {'group': k, 'revenue': 0.0, 'revenue_ly': 0.0, 'fisch': 0.0, 'royal': 0.0, 'customers': 0, 'orders': 0}) for pid, v in this.items(): d = bucket(keyfor(pid, v['rev'])) d['revenue'] += v['rev']; d['customers'] += 1; d['orders'] += v['orders'] if split: d['fisch'] += f_rev.get(pid, {}).get('rev', 0.0) d['royal'] += r_rev.get(pid, {}).get('rev', 0.0) for pid, v in last.items(): bucket(keyfor(pid, v['rev']))['revenue_ly'] += v['rev'] rows = list(agg.values()) for d in rows: d['yoy_pct'] = P.yoy_pct(d['revenue'], d['revenue_ly']) d['avg_per_customer'] = d['revenue'] / d['customers'] if d['customers'] else 0.0 rows.sort(key=lambda x: -x['revenue']) return rows def _order_dom(pid, date_from, date_to, team_id=None): return sales_mod.order_domain(date_from, date_to, team_id) + [('partner_id', '=', pid)] def _line_dom(pid, date_from, date_to, team_id=None): return O.sale_line_domain(date_from, date_to, team_id, extra=[('order_partner_id', '=', pid)]) def _monthly_rev(pid, date_from, date_to, team_id=None): """{'YYYY-MM': revenue} over a window via ONE month-grouped read_group (was 26 point queries).""" g = O.read_group('sale.order', _order_dom(pid, date_from, date_to, team_id), ['amount_untaxed:sum'], ['date_order:month'], lazy=False) out = {} for r in g: rng = (r.get('__range') or {}).get('date_order:month') or {} ym = (rng.get('from') or '')[:7] if ym: out[ym] = r.get('amount_untaxed') or 0.0 return out def customer_detail(pid, t=None, team_id=None, n_months=13, top=12): """Granular drill-down for one customer: profile, RFM-style KPIs, monthly YoY trend, top SKUs + categories bought (LTM, with margin), and recent orders. The ~14 independent Odoo reads run CONCURRENTLY (O.parallel_map) so a cold drawer loads in ~1-2s instead of ~6s.""" t = t or P.today() yf, yt = P.ytd(t) lf, lt = P.ytd_last_year(t) mf, mt = P.ltm(t) range_start = dt.date(t.year - 2, t.month, 1).isoformat() # Fisch/Royal mix only when consolidated (team_id=None). When a single BU is selected we never # compute or expose the other BU — strict isolation for per-BU permissioning. split = team_id is None r = O.parallel_map({ 'prof': lambda: O.search_read('res.partner', [('id', '=', pid)], ['name', 'email', 'phone']), 'attrs': lambda: _partner_attrs([pid]), 'ytd_rev': lambda: O.sum_field('sale.order', _order_dom(pid, yf, yt, team_id), 'amount_untaxed'), 'ytd_rev_ly': lambda: O.sum_field('sale.order', _order_dom(pid, lf, lt, team_id), 'amount_untaxed'), 'ytd_fisch': (lambda: O.sum_field('sale.order', _order_dom(pid, yf, yt, 5), 'amount_untaxed')) if split else (lambda: None), 'ytd_royal': (lambda: O.sum_field('sale.order', _order_dom(pid, yf, yt, 6), 'amount_untaxed')) if split else (lambda: None), 'ltm_rev': lambda: O.sum_field('sale.order', _order_dom(pid, mf, mt, team_id), 'amount_untaxed'), 'orders_ltm': lambda: O.get_odoo().search_count('sale.order', _order_dom(pid, mf, mt, team_id)), 'orders_ytd': lambda: O.get_odoo().search_count('sale.order', _order_dom(pid, yf, yt, team_id)), 'lastrow': lambda: O.search_read('sale.order', _order_dom(pid, None, None, team_id), ['date_order'], order='date_order desc', limit=1), 'firstrow': lambda: O.search_read('sale.order', _order_dom(pid, None, None, team_id), ['date_order'], order='date_order asc', limit=1), 'mrev': lambda: _monthly_rev(pid, range_start, t.isoformat(), team_id), 'lg': lambda: O.read_group('sale.order.line', _line_dom(pid, mf, mt, team_id), ['price_subtotal:sum', 'product_uom_qty:sum', 'margin:sum'], ['product_id'], lazy=False), 'recent': lambda: O.search_read('sale.order', _order_dom(pid, None, None, team_id), ['name', 'date_order', 'amount_untaxed', 'state'], order='date_order desc', limit=10), }) prof = (r['prof'] or [{}])[0] a = r['attrs'].get(pid, {}) ytd_rev, ytd_rev_ly, ltm_rev = r['ytd_rev'], r['ytd_rev_ly'], r['ltm_rev'] ytd_fisch, ytd_royal = r['ytd_fisch'], r['ytd_royal'] orders_ltm, orders_ytd = r['orders_ltm'], r['orders_ytd'] last_order = str(r['lastrow'][0]['date_order'])[:10] if r['lastrow'] else None first_order = str(r['firstrow'][0]['date_order'])[:10] if r['firstrow'] else None recency = (t - dt.date.fromisoformat(last_order)).days if last_order else None mrev = r['mrev'] monthly = [] for ym, start, end in P.month_starts(n_months, t): y, m = int(ym[:4]) - 1, int(ym[5:7]) monthly.append({'month': ym, 'revenue': mrev.get(ym, 0.0), 'revenue_ly': mrev.get(f'{y:04d}-{m:02d}', 0.0)}) lg = r['lg'] skus = [{'product': O.m2o_name(x.get('product_id')), 'revenue': x.get('price_subtotal') or 0.0, 'qty': x.get('product_uom_qty') or 0.0, 'margin': x.get('margin') or 0.0} for x in lg if x.get('product_id')] skus.sort(key=lambda x: -x['revenue']) line_rev = sum(s['revenue'] for s in skus) margin = sum(s['margin'] for s in skus) cat = sales_mod._product_cat() catagg = {} for x in lg: prodid = O.m2o_id(x.get('product_id')) if not prodid: continue c = cat.get(prodid, '(uncategorized)') catagg[c] = catagg.get(c, 0.0) + (x.get('price_subtotal') or 0.0) cats = sorted([{'category': k, 'revenue': v} for k, v in catagg.items()], key=lambda x: -x['revenue'])[:10] recent_rows = [{'order': x.get('name'), 'date': str(x.get('date_order'))[:10], 'amount': x.get('amount_untaxed') or 0.0, 'status': 'Confirmed' if x.get('state') in ('sale', 'done') else x.get('state')} for x in r['recent']] return { 'pid': pid, 'name': prof.get('name') or '(unknown)', 'city': a.get('city', '(none)'), 'state': a.get('state', '(none)'), 'country': a.get('country', '(none)'), 'agent': a.get('agent', '(none)'), 'email': prof.get('email') or '', 'phone': prof.get('phone') or '', 'ytd_rev': ytd_rev, 'ytd_rev_ly': ytd_rev_ly, 'yoy_pct': P.yoy_pct(ytd_rev, ytd_rev_ly), 'ytd_fisch': ytd_fisch, 'ytd_royal': ytd_royal, 'ltm_rev': ltm_rev, 'orders_ytd': orders_ytd, 'orders_ltm': orders_ltm, 'aov_ltm': (ltm_rev / orders_ltm) if orders_ltm else 0.0, 'first_order': first_order, 'last_order': last_order, 'recency_days': recency, 'ltm_margin': margin, 'ltm_gm_pct': (margin / line_rev * 100) if line_rev else 0.0, 'monthly': monthly, 'top_skus': skus[:top], 'top_categories': cats, 'recent_orders': recent_rows, } def customer_drawer_bundle(pid, t=None, team_id=None): """Everything the customer drawer's first paint needs, in as few round-trips as possible: customer_detail (itself parallel) then the other three pulls CONCURRENTLY. One cached unit, so switching drawer sections never re-hits Odoo. customer_affinity (the look-alike co-buyer scan — 3 dependent heavy reads, 4-8s) and customer_stockout stay LAZY, loaded only by their own sections, so a cold drawer opens in ~2-3s regardless of how big the customer is.""" detail = customer_detail(pid, t=t, team_id=team_id) decomp, winback, cadence = O.parallel([ lambda: customer_yoy_decomp(pid, t, team_id), lambda: customer_winback(pid, t, team_id), lambda: customer_cadence(pid, t, team_id), ]) return {'detail': detail, 'decomp': decomp, 'winback': winback, 'cadence': cadence} def customer_stockout(pid, t=None, team_id=None, top=25): """Stockout exposure for one customer: of the SKUs they buy, which are currently OUT / LOW on hand, and how much of their YoY $ decline sits on SKUs that are now out of stock (a likely stockout-driven loss) vs other causes. On-hand is a CURRENT snapshot (no historical stock), so 'decline on a now-out SKU' is a strong proxy, not proof, of a stockout cause.""" t = t or P.today() yf, yt = P.ytd(t) lf, lt = P.ytd_last_year(t) def skumap(df, dtt): # storable goods only — services (delivery charges etc.) can't "stock out" dom = O.sale_line_domain(df, dtt, team_id, extra=[('order_partner_id', '=', pid), ('product_id.type', '!=', 'service')]) g = O.read_group('sale.order.line', dom, ['price_subtotal:sum', 'product_uom_qty:sum'], ['product_id'], lazy=False) return {O.m2o_id(r['product_id']): {'name': O.m2o_name(r.get('product_id')), 'rev': r.get('price_subtotal') or 0.0, 'qty': r.get('product_uom_qty') or 0.0} for r in g if r.get('product_id')} this, last = skumap(yf, yt), skumap(lf, lt) pids = list(set(this) | set(last)) empty = {'rows': [], 'n_out': 0, 'n_low': 0, 'rev_at_risk': 0.0, 'risk_pct': 0.0, 'drop_stockout': 0.0, 'drop_other': 0.0, 'total_drop': 0.0} if not pids: return empty q = O.read_group('stock.quant', [('location_id.usage', '=', 'internal'), ('product_id', 'in', pids)], ['quantity:sum'], ['product_id'], lazy=False) onhand = {O.m2o_id(r['product_id']): (r.get('quantity') or 0.0) for r in q if r.get('product_id')} codemap = {} for pr in O.search_read('product.product', [('id', 'in', pids)], ['default_code']): codemap[pr['id']] = str(pr['default_code']).strip() if pr.get('default_code') else None rows, drop_stockout, drop_other = [], 0.0, 0.0 for p in pids: tr = this.get(p, {}).get('rev', 0.0) lr = last.get(p, {}).get('rev', 0.0) oh = onhand.get(p, 0.0) ly_qty = last.get(p, {}).get('qty', 0.0) out = oh <= 0 low = (not out) and oh < max(1.0, ly_qty * 0.25) # under ~a quarter of their annual usage change = tr - lr rows.append({'sku': (this.get(p) or last.get(p) or {}).get('name', '?'), 'code': codemap.get(p), 'on_hand': oh, 'ly_rev': lr, 'ytd_rev': tr, 'change': change, 'status': 'OUT' if out else ('LOW' if low else 'OK')}) if change < 0: if out: drop_stockout += -change else: drop_other += -change rows.sort(key=lambda r: ({'OUT': 0, 'LOW': 1, 'OK': 2}[r['status']], -r['ly_rev'])) this_total = sum(v['rev'] for v in this.values()) or 1.0 rev_at_risk = sum(r['ytd_rev'] for r in rows if r['status'] in ('OUT', 'LOW')) return {'rows': rows[:top], 'n_out': sum(1 for r in rows if r['status'] == 'OUT'), 'n_low': sum(1 for r in rows if r['status'] == 'LOW'), 'rev_at_risk': rev_at_risk, 'risk_pct': rev_at_risk / this_total * 100, 'drop_stockout': drop_stockout, 'drop_other': drop_other, 'total_drop': drop_stockout + drop_other} # ====================================================================== ADVANCED FILTERS # "Find customers who bought X" — resolve a SKU query / category to the set of partner ids that # purchased it, so the directory can be filtered by purchase behaviour (not just attributes). def _buyer_window(t): """24-month look-back for 'who bought this' — catches recent AND lapsed buyers (win-back).""" t = t or P.today() return (t - dt.timedelta(days=730)).isoformat(), t.isoformat() def sku_buyers(query, t=None, team_id=None): """Set of partner ids who bought any SKU whose code/name matches `query` (last 24 months). Returns None when the query is blank (= no filter).""" q = (query or '').strip() # A 1-char ilike matches half the catalogue — treat it as "still typing" (no filter yet). if len(q) < 2: return None df, dtt = _buyer_window(t) prods = O.search_read('product.product', ['|', ('default_code', 'ilike', q), ('name', 'ilike', q)], ['id'], limit=3000) pids = [p['id'] for p in prods] if not pids: return set() g = O.read_group('sale.order.line', O.sale_line_domain(df, dtt, team_id, extra=[('product_id', 'in', pids)]), ['order_partner_id'], ['order_partner_id'], lazy=False) return {O.m2o_id(r.get('order_partner_id')) for r in g if r.get('order_partner_id')} def category_buyers(category, t=None, team_id=None): """Set of partner ids who bought from a main category (last 24 months). None = no filter.""" if not category or category == '(any)': return None df, dtt = _buyer_window(t) catmap = sales_mod._product_cat() prod_ids = [pid for pid, c in catmap.items() if c == category] if not prod_ids: return set() g = O.read_group('sale.order.line', O.sale_line_domain(df, dtt, team_id, extra=[('product_id', 'in', prod_ids)]), ['order_partner_id'], ['order_partner_id'], lazy=False) return {O.m2o_id(r.get('order_partner_id')) for r in g if r.get('order_partner_id')} def main_categories(): """Sorted list of main category names (for the 'bought in category' selector).""" return sorted(set(sales_mod._product_cat().values())) # ====================================================================== WIN-BACK / SKU MOVES def customer_winback(pid, t=None, team_id=None, top=10): """SKU-level YoY moves for one customer (this YTD vs same period last year): losers — SKUs down YoY, each with its share of the customer's TOTAL gross loss (pct_of_loss) gainers — SKUs up YoY (bought more) lapsed — losers gone to zero this year (the re-pitch hooks) lapsed_categories — category-level gaps total_loss / this_total / last_total / net_change 'change' is signed (this − last); negative = decline so the UI tints it red.""" t = t or P.today() yf, yt = P.ytd(t) lf, lt = P.ytd_last_year(t) def sku_map(df, dtt): g = O.read_group('sale.order.line', _line_dom(pid, df, dtt, team_id), ['price_subtotal:sum', 'product_uom_qty:sum'], ['product_id'], lazy=False) return {O.m2o_id(r['product_id']): {'name': O.m2o_name(r.get('product_id')), 'rev': r.get('price_subtotal') or 0.0, 'qty': r.get('product_uom_qty') or 0.0} for r in g if r.get('product_id')} this, last = sku_map(yf, yt), sku_map(lf, lt) # product_id -> SKU code, so each row can deep-link to its SKU drawer (code is how the SKU # views key products; merged-by-code duplicates resolve to the same drawer) pids = list(set(this) | set(last)) codemap = {} if pids: for pr in O.search_read('product.product', [('id', 'in', pids)], ['default_code']): codemap[pr['id']] = str(pr['default_code']).strip() if pr.get('default_code') else None rows = [] for p in (set(this) | set(last)): tv, lv = (this.get(p) or {}), (last.get(p) or {}) tr, lr = tv.get('rev', 0.0), lv.get('rev', 0.0) tq, lq = tv.get('qty', 0.0), lv.get('qty', 0.0) name = (this.get(p) or last.get(p) or {}).get('name') or '(?)' p_ly = (lr / lq) if lq else None # realised $/unit last year p_ytd = (tr / tq) if tq else None # realised $/unit this year rows.append({'sku': name, 'code': codemap.get(p), 'ly_rev': lr, 'ytd_rev': tr, 'change': tr - lr, 'qty_ly': lq, 'qty_ytd': tq, 'qty_change': tq - lq, 'price_ly': p_ly, 'price_ytd': p_ytd, 'price_chg_pct': (((p_ytd - p_ly) / p_ly * 100) if (p_ly and p_ytd) else None), # split the $ change into volume vs price effects (sum to change) 'vol_effect': ((tq - lq) * p_ly) if p_ly is not None else (tr - lr), 'price_effect': ((p_ytd - p_ly) * tq) if (p_ly is not None and p_ytd is not None) else 0.0}) total_loss = sum(-r['change'] for r in rows if r['change'] < 0) for r in rows: r['pct_of_loss'] = (-r['change'] / total_loss * 100) if (r['change'] < 0 and total_loss) else 0.0 losers = sorted([r for r in rows if r['change'] < 0], key=lambda x: x['change']) gainers = sorted([r for r in rows if r['change'] > 0], key=lambda x: -x['change']) lapsed = [r for r in losers if r['ytd_rev'] == 0] cat = sales_mod._product_cat() def cat_rev(m): agg = {} for p, v in m.items(): agg[cat.get(p, '(uncategorized)')] = agg.get(cat.get(p, '(uncategorized)'), 0.0) + v['rev'] return agg tcat, lcat = cat_rev(this), cat_rev(last) lapsed_categories = sorted([{'category': c, 'ly_rev': lcat[c], 'ytd_rev': tcat.get(c, 0.0), 'change': tcat.get(c, 0.0) - lcat[c]} for c in lcat if lcat[c] > tcat.get(c, 0.0)], key=lambda x: x['change']) this_total = sum(v['rev'] for v in this.values()) last_total = sum(v['rev'] for v in last.values()) # 'all_skus' feeds the in-drawer SKU list; cap it so a mega-account doesn't ship thousands # of rows to the browser (the list is searchable/filtered, top-by-revenue is what matters). all_skus = sorted(rows, key=lambda x: -max(x['ytd_rev'], x['ly_rev']))[:500] return { 'losers': losers[:top], 'gainers': gainers[:top], 'lapsed': lapsed[:top], 'lapsed_categories': lapsed_categories[:6], 'all_skus': all_skus, 'n_skus': len(rows), 'n_lapsed': len(lapsed), 'n_losers': len(losers), 'n_gainers': len(gainers), 'total_loss': total_loss, 'lost_dollars': total_loss, 'this_total': this_total, 'last_total': last_total, 'net_change': this_total - last_total, } def customer_yoy_decomp(pid, t=None, team_id=None): """Decompose the customer's YoY sales change into a volume (order count) effect and a price (basket size / AOV) effect. Identity: ΔSales = ΔOrders·AOV_last + Orders_this·ΔAOV.""" t = t or P.today() yf, yt = P.ytd(t) lf, lt = P.ytd_last_year(t) this_sales = O.sum_field('sale.order', _order_dom(pid, yf, yt, team_id), 'amount_untaxed') last_sales = O.sum_field('sale.order', _order_dom(pid, lf, lt, team_id), 'amount_untaxed') this_orders = O.get_odoo().search_count('sale.order', _order_dom(pid, yf, yt, team_id)) last_orders = O.get_odoo().search_count('sale.order', _order_dom(pid, lf, lt, team_id)) this_aov = (this_sales / this_orders) if this_orders else 0.0 last_aov = (last_sales / last_orders) if last_orders else 0.0 volume_effect = (this_orders - last_orders) * last_aov price_effect = this_orders * (this_aov - last_aov) return { 'this_sales': this_sales, 'last_sales': last_sales, 'd_sales': this_sales - last_sales, 'sales_yoy_pct': P.yoy_pct(this_sales, last_sales), 'this_orders': this_orders, 'last_orders': last_orders, 'd_orders': this_orders - last_orders, 'this_aov': this_aov, 'last_aov': last_aov, 'd_aov': this_aov - last_aov, 'aov_yoy_pct': P.yoy_pct(this_aov, last_aov), 'volume_effect': volume_effect, 'price_effect': price_effect, } def customer_affinity(pid, t=None, team_id=None, top=12, max_cobuyers=60): """'Customers who buy similar products also buy …'. Finds the customers who bought this customer's SKUs (co-buyers), then ranks the OTHER SKUs those co-buyers buy (that this customer doesn't) by spend among them — a cross-sell list. LTM window + capped co-buyers to stay fast.""" df, dtt = P.ltm(t) g = O.read_group('sale.order.line', _line_dom(pid, df, dtt, team_id), ['price_subtotal:sum'], ['product_id'], lazy=False) mine = {O.m2o_id(r['product_id']) for r in g if r.get('product_id')} if not mine: return {'recs': [], 'n_cobuyers': 0, 'mine': 0} cg = O.read_group('sale.order.line', O.sale_line_domain(df, dtt, team_id, extra=[('product_id', 'in', list(mine))]), ['price_subtotal:sum'], ['order_partner_id'], lazy=False) cobuyers = sorted([(O.m2o_id(r['order_partner_id']), r.get('price_subtotal') or 0.0) for r in cg if r.get('order_partner_id') and O.m2o_id(r['order_partner_id']) != pid], key=lambda x: -x[1])[:max_cobuyers] cob_ids = [c[0] for c in cobuyers] if not cob_ids: return {'recs': [], 'n_cobuyers': 0, 'mine': len(mine)} # Single groupby over the co-buyers' lines (one aggregated row per product) — fast. pg = O.read_group('sale.order.line', O.sale_line_domain(df, dtt, team_id, extra=[('order_partner_id', 'in', cob_ids)]), ['price_subtotal:sum'], ['product_id'], lazy=False) recs = sorted([{'sku': O.m2o_name(r['product_id']), 'pid': O.m2o_id(r['product_id']), 'rev': r.get('price_subtotal') or 0.0, 'orders': r.get('__count') or 0} for r in pg if r.get('product_id') and O.m2o_id(r['product_id']) not in mine], key=lambda x: -x['rev'])[:top] if recs: codemap = {} for pr in O.search_read('product.product', [('id', 'in', [r['pid'] for r in recs])], ['default_code']): codemap[pr['id']] = str(pr['default_code']).strip() if pr.get('default_code') else None for r in recs: r['code'] = codemap.get(r['pid']) return {'recs': recs, 'n_cobuyers': len(cob_ids), 'mine': len(mine)} # ====================================================================== CADENCE / CHURN / BENCHMARK def _order_dates(pid, team_id=None): """Recent confirmed order dates (date objects), ascending. Capped at the 3000 most recent so a very high-volume buyer can't hang the drawer — cadence/frequency only need recent orders.""" rows = O.search_read('sale.order', _order_dom(pid, None, None, team_id), ['date_order'], order='date_order desc', limit=3000) return sorted(dt.date.fromisoformat(str(r['date_order'])[:10]) for r in rows if r.get('date_order')) def customer_cadence(pid, t=None, team_id=None): """Reorder rhythm from the gaps between orders: typical gap (median of recent), predicted next order, days overdue vs that rhythm, and whether the rhythm is slowing (last-4 vs prior-4 gap).""" t = t or P.today() dates = _order_dates(pid, team_id) n = len(dates) out = {'n_orders': n, 'median_gap_days': None, 'last_order': dates[-1].isoformat() if dates else None, 'predicted_next': None, 'overdue_days': None, 'drift_pct': None, 'recent_gap': None, 'prior_gap': None, 'dates': [d.isoformat() for d in dates]} if n < 2: return out gaps = [(dates[i] - dates[i - 1]).days for i in range(1, n)] median_gap = statistics.median(gaps[-8:]) last = dates[-1] predicted = last + dt.timedelta(days=round(median_gap)) recent_gap = statistics.mean(gaps[-4:]) if len(gaps) >= 4 else statistics.mean(gaps) prior_gap = statistics.mean(gaps[-8:-4]) if len(gaps) >= 8 else None out.update({'median_gap_days': median_gap, 'predicted_next': predicted.isoformat(), 'overdue_days': (t - predicted).days, 'recent_gap': recent_gap, 'prior_gap': prior_gap, 'drift_pct': ((recent_gap - prior_gap) / prior_gap * 100) if prior_gap else None}) return out def customer_churn_score(pid, t=None, team_id=None, cad=None, sales_yoy=None): """0–100 churn-risk score (higher = more at risk): overdue-vs-cadence (50%) + frequency decay last-90 vs prior-90 (30%) + YoY sales trend (20%). Bucketed Healthy/Watch/At-risk/Critical.""" t = t or P.today() cad = cad or customer_cadence(pid, t, team_id) mg = cad.get('median_gap_days') if mg and cad.get('overdue_days') is not None: s_overdue = min(1.0, max(0.0, cad['overdue_days'] / mg) / 2.0) # 2 cycles late = max else: s_overdue = 0.5 dates = [dt.date.fromisoformat(d) for d in cad.get('dates', [])] last90 = sum(1 for d in dates if (t - d).days <= 90) prior90 = sum(1 for d in dates if 90 < (t - d).days <= 180) if prior90 == 0: # recent activity but no 90–180d baseline (reactivated / new) → cautious mid risk, not zero s_freq = 0.6 if last90 == 0 else 0.3 else: s_freq = min(1.0, max(0.0, (prior90 - last90) / prior90)) if sales_yoy is None: sales_yoy = customer_yoy_decomp(pid, t, team_id)['sales_yoy_pct'] s_yoy = 0.5 if sales_yoy is None else min(1.0, max(0.0, -sales_yoy / 50.0)) # -50% YoY = max score = 100 * (0.5 * s_overdue + 0.3 * s_freq + 0.2 * s_yoy) bucket = ('Critical' if score >= 70 else 'At-risk' if score >= 45 else 'Watch' if score >= 25 else 'Healthy') driver = max([('overdue rhythm', s_overdue), ('fewer recent orders', s_freq), ('falling spend', s_yoy)], key=lambda x: x[1])[0] return {'score': round(score), 'bucket': bucket, 'driver': driver} def _pctile(values, x): """Percentile rank (0–100) of x within values — mean/midpoint method (ties count as half), so the median of a peer set lands at the 50th percentile.""" if not values: return None below = sum(1 for v in values if v < x) equal = sum(1 for v in values if v == x) return (below + 0.5 * equal) / len(values) * 100 def rank_contribution(directory_rows, pid): """Rank by YTD revenue + % of BU YTD — pure, from the already-cached directory list.""" total = sum(r['revenue'] for r in directory_rows) or 1.0 ranked = sorted(directory_rows, key=lambda r: -r['revenue']) rank = next((i + 1 for i, r in enumerate(ranked) if r['pid'] == pid), None) me = next((r for r in directory_rows if r['pid'] == pid), None) return {'rank': rank, 'n': len(directory_rows), 'pct_of_bu': (me['revenue'] / total * 100) if me else None} def peer_benchmark(directory_rows, pid): """Percentile rank vs same value-tier peers on revenue, AOV, frequency, YoY — pure.""" me = next((r for r in directory_rows if r['pid'] == pid), None) if not me: return None tier = _tier(me['revenue']) peers = [r for r in directory_rows if _tier(r['revenue']) == tier] yoy_vals = [r['yoy_pct'] for r in peers if r['yoy_pct'] is not None] return {'cohort_n': len(peers), 'tier': tier, 'rev_pctile': _pctile([r['revenue'] for r in peers], me['revenue']), 'aov_pctile': _pctile([r['aov'] for r in peers], me['aov']), 'freq_pctile': _pctile([r['orders'] for r in peers], me['orders']), 'yoy_pctile': (_pctile(yoy_vals, me['yoy_pct']) if me['yoy_pct'] is not None else None)} def customer_whitespace(mine_categories, company_categories, top=6): """Pure: breadth (# categories bought / company total) + the biggest company categories this customer buys $0 of, ranked by company revenue. `company_categories` = sales.by_category rows.""" skip = {'(uncategorized)', 'All'} mine = {c for c in mine_categories if c not in skip} cats = [c for c in company_categories if c['category'] not in skip and c['revenue'] > 0] total = len(cats) ws = sorted([c for c in cats if c['category'] not in mine], key=lambda x: -x['revenue'])[:top] return {'breadth_x': len(mine), 'breadth_y': total, 'whitespace': ws} # ====================================================================== PAGE-LEVEL: NRR / MIGRATION def nrr(t=None, team_id=None, agent_pids=None): """Net Revenue Retention from the revenue bridge (existing book only; new logos excluded).""" b = revenue_bridge(t, team_id, agent_pids=agent_pids) last = b['last_total'] or 1.0 ending = b['last_total'] + b['expansion']['rev'] + b['contraction']['rev'] + b['lost']['rev'] return {'nrr_pct': ending / last * 100, 'starting': b['last_total'], 'expansion': b['expansion']['rev'], 'contraction': b['contraction']['rev'], 'churned': b['lost']['rev'], 'ending_existing': ending, 'new': b['new']['rev']} def tier_migration(t=None, team_id=None, agent_pids=None): """Value-tier flows LTM vs prior-LTM: upgraded / held / downgraded / new / lapsed + net $.""" lf, lt = P.ltm(t) pf, pt = P.prior_ltm(t) this = _cust_rev(lf, lt, team_id, agent_pids) last = _cust_rev(pf, pt, team_id, agent_pids) order = {name: i for i, (name, _) in enumerate(TIERS)} # 0 = Whale (top) … 3 = Small flows = {k: {'flow': k, 'n': 0, 'net': 0.0} for k in ['Upgraded', 'Held', 'Downgraded', 'New / reactivated', 'Lapsed']} for pid in set(this) | set(last): tr = this.get(pid, {}).get('rev', 0.0) lr = last.get(pid, {}).get('rev', 0.0) if pid in this and pid in last: k = ('Upgraded' if order[_tier(tr)] < order[_tier(lr)] else 'Downgraded' if order[_tier(tr)] > order[_tier(lr)] else 'Held') elif pid in this: k = 'New / reactivated' else: k = 'Lapsed' flows[k]['n'] += 1 flows[k]['net'] += (tr - lr) return {'flows': list(flows.values())} # ====================================================================== GROUP DRAWER (rollup → drawer) def group_detail(dim, value, t=None, team_id=None, n_months=13, top=15, agent_pids=None): """For one rollup group (e.g. dim='city', value='Brooklyn'): KPIs, monthly trend, the customers in the group (clickable), and the top SKUs sold there. dim is a DIMENSIONS key.""" t = t or P.today() yf, yt = P.ytd(t) lf, lt = P.ytd_last_year(t) this = _cust_rev(yf, yt, team_id, agent_pids) last = _cust_rev(lf, lt, team_id, agent_pids) attrs = _partner_attrs(set(this) | set(last)) def keyfor(pid): if dim == 'segment': return _tier(this.get(pid, {}).get('rev', last.get(pid, {}).get('rev', 0.0))) return (attrs.get(pid) or {}).get(dim) or '(none)' pids = [pid for pid in (set(this) | set(last)) if keyfor(pid) == value] if not pids: return None # KPI totals and the customer list cover EVERY member of the group (cheap, in-memory). rev_ytd = sum(this.get(p, {}).get('rev', 0.0) for p in pids) rev_ly = sum(last.get(p, {}).get('rev', 0.0) for p in pids) custs = sorted([{'pid': p, 'customer': (this.get(p) or last.get(p) or {}).get('name', '?'), 'revenue': this.get(p, {}).get('rev', 0.0), 'orders': this.get(p, {}).get('orders', 0), 'yoy_pct': P.yoy_pct(this.get(p, {}).get('rev', 0.0), last.get(p, {}).get('rev', 0.0)), 'city': (attrs.get(p) or {}).get('city', '(none)'), 'state': (attrs.get(p) or {}).get('state', '(none)'), 'agent': (attrs.get(p) or {}).get('agent', '(none)')} for p in pids], key=lambda x: -x['revenue']) # The trend + top-SKU read_groups put every pid in an IN(...) clause, so a huge group # (a big city, the '(none)' bucket) would build a slow, oversized query. Cap those two # queries to the group's top buyers by revenue; KPIs/customer list above stay complete. QCAP = 300 capped = len(pids) > QCAP qpids = sorted( pids, key=lambda p: -max(this.get(p, {}).get('rev', 0.0), last.get(p, {}).get('rev', 0.0)) )[:QCAP] if capped else pids range_start = dt.date(t.year - 2, t.month, 1).isoformat() g = O.read_group('sale.order', sales_mod.order_domain(range_start, t.isoformat(), team_id) + [('partner_id', 'in', qpids)], ['amount_untaxed:sum'], ['date_order:month'], lazy=False) mrev = {} for r in g: rng = (r.get('__range') or {}).get('date_order:month') or {} ym = (rng.get('from') or '')[:7] if ym: mrev[ym] = r.get('amount_untaxed') or 0.0 monthly = [] for ym, _s, _e in P.month_starts(n_months, t): y, m = int(ym[:4]) - 1, int(ym[5:7]) monthly.append({'month': ym, 'revenue': mrev.get(ym, 0.0), 'revenue_ly': mrev.get(f'{y:04d}-{m:02d}', 0.0)}) mf, mt = P.ltm(t) lg = O.read_group('sale.order.line', O.sale_line_domain(mf, mt, team_id, extra=[('order_partner_id', 'in', qpids)]), ['price_subtotal:sum', 'product_uom_qty:sum', 'margin:sum'], ['product_id'], lazy=False) skus = sorted([sales_mod._sku_profit( {'product': O.m2o_name(r.get('product_id')), 'pid': O.m2o_id(r.get('product_id')), 'rev': r.get('price_subtotal') or 0.0, 'revenue': r.get('price_subtotal') or 0.0, 'qty': r.get('product_uom_qty') or 0.0, 'margin': r.get('margin') or 0.0, 'lines': r.get('__count') or 0}) for r in lg if r.get('product_id')], key=lambda x: -x['rev'])[:top] if skus: # attach SKU code for deep-linking each to its SKU drawer scode = {} for pr in O.search_read('product.product', [('id', 'in', [s['pid'] for s in skus])], ['default_code']): scode[pr['id']] = str(pr['default_code']).strip() if pr.get('default_code') else None for s in skus: s['code'] = scode.get(s['pid']) return {'dim': dim, 'value': value, 'rev_ytd': rev_ytd, 'rev_ly': rev_ly, 'yoy_pct': P.yoy_pct(rev_ytd, rev_ly), 'n_customers': len([p for p in pids if p in this]), 'n_total': len(pids), 'monthly': monthly, 'top_skus': skus, 'customers': custs, 'capped': capped, 'qcap': QCAP} # ----------------------------------------------------------- KPI-card customer sets _KPI_SETS = { 'active': 'Active customers (YTD)', 'new': 'New / reactivated (YTD)', 'lost': 'Lost customers (YTD)', 'retained': 'Retained customers (YTD)', 'existing': 'Existing book (NRR base)', } def customer_set(kind, t=None, team_id=None, agent_pids=None): """Standardized customer list behind a headline KPI card (Active / New / Lost / Retained / Existing-book), each row carrying the same shape (pid, customer, rev_ytd, rev_ly, change, yoy_pct, orders, city, state, agent) so the drawer renders them with one standardized view.""" t = t or P.today() yf, yt = P.ytd(t) lf, lt = P.ytd_last_year(t) this = _cust_rev(yf, yt, team_id, agent_pids) last = _cust_rev(lf, lt, team_id, agent_pids) tset, lset = set(this), set(last) ids = {'active': tset, 'new': tset - lset, 'lost': lset - tset, 'retained': tset & lset, 'existing': lset}.get(kind, tset) attrs = _partner_attrs(list(ids)) rows = [] for p in ids: tr = this.get(p, {}).get('rev', 0.0) lr = last.get(p, {}).get('rev', 0.0) a = attrs.get(p, {}) rows.append({'pid': p, 'customer': (this.get(p) or last.get(p) or {}).get('name', '?'), 'rev_ytd': tr, 'rev_ly': lr, 'change': tr - lr, 'yoy_pct': P.yoy_pct(tr, lr), 'orders': this.get(p, {}).get('orders', 0), 'city': a.get('city', '(none)'), 'state': a.get('state', '(none)'), 'agent': a.get('agent', '(none)')}) rows.sort(key=lambda r: -(r['rev_ly'] if kind == 'lost' else r['rev_ytd'])) return {'kind': kind, 'label': _KPI_SETS.get(kind, kind), 'rows': rows, 'n': len(rows), 'rev_ytd': sum(r['rev_ytd'] for r in rows), 'rev_ly': sum(r['rev_ly'] for r in rows)} # ----------------------------------------------------------- VALIDATION def validate(t=None, team_id=None): """Reconcile every headline number to an independent Odoo aggregate. When team_id is set (a single BU is selected) the checks run SCOPED to that BU so the validation panel never exposes other-BU figures — BU isolation holds even here. The two cross-BU brand-mirror checks (#2, #4b) only make sense consolidated, so they run only when team_id is None. """ t = t or P.today() checks = [] # 1. Bridge identity: last + new + expansion + contraction + lost == this b = revenue_bridge(t, team_id=team_id) recon = (b['last_total'] + b['new']['rev'] + b['expansion']['rev'] + b['contraction']['rev'] + b['lost']['rev']) checks.append({'check': 'Revenue bridge reconciles last→this (New+Exp+Contr+Lost)', 'a': round(recon, 2), 'b': round(b['this_total'], 2), 'gap': round(recon - b['this_total'], 2), 'ok': abs(recon - b['this_total']) <= 1.0}) # 2. Σ(brand this_total) == company this_total (consolidated only — cross-BU) if team_id is None: brand_this = sum(r['this'] for r in bridge_by_brand(t)) checks.append({'check': 'Σ(brand YTD) == company YTD', 'a': round(brand_this, 2), 'b': round(b['this_total'], 2), 'gap': round(brand_this - b['this_total'], 2), 'ok': abs(brand_this - b['this_total']) <= 1.0}) # 3. Σ(segment LTM rev) == total LTM rev lf, lt = P.ltm(t) seg_sum = sum(s['revenue'] for s in segments(t, team_id=team_id)) ltm_total = sum(v['rev'] for v in _cust_rev(lf, lt, team_id=team_id).values()) checks.append({'check': 'Σ(segment LTM rev) == total LTM rev', 'a': round(seg_sum, 2), 'b': round(ltm_total, 2), 'gap': round(seg_sum - ltm_total, 2), 'ok': abs(seg_sum - ltm_total) <= 1.0}) # 4. Sum-level rollup is MECE: Σ(by_dimension('agent')) == total YTD revenue yf, yt = P.ytd(t) ytd_total = sum(v['rev'] for v in _cust_rev(yf, yt, team_id=team_id).values()) roll = by_dimension('agent', t, team_id=team_id) dim_sum = sum(r['revenue'] for r in roll) checks.append({'check': 'Σ(agent rollup) == total YTD revenue', 'a': round(dim_sum, 2), 'b': round(ytd_total, 2), 'gap': round(dim_sum - ytd_total, 2), 'ok': abs(dim_sum - ytd_total) <= 1.0}) # 4b. Brand mirror reconciles: Σ(Fisch)+Σ(Royal) == total YTD (consolidated only) if team_id is None: brand_sum = sum(r['fisch'] + r['royal'] for r in roll) checks.append({'check': 'Rollup brand mirror: Σ(Fisch)+Σ(Royal) == total YTD', 'a': round(brand_sum, 2), 'b': round(ytd_total, 2), 'gap': round(brand_sum - ytd_total, 2), 'ok': abs(brand_sum - ytd_total) <= 1.0}) # 5. Drill-down scope holds: top customer's LTM order-level rev == line-level rev top = directory(t, team_id=team_id, limit=1) if top: pid = top[0]['pid'] mf, mt = P.ltm(t) ord_rev = O.sum_field('sale.order', _order_dom(pid, mf, mt), 'amount_untaxed') line_rev = O.sum_field('sale.order.line', _line_dom(pid, mf, mt), 'price_subtotal') checks.append({'check': 'Drill-down: top customer LTM order==line revenue', 'a': round(ord_rev, 2), 'b': round(line_rev, 2), 'gap': round(ord_rev - line_rev, 2), 'ok': abs(ord_rev - line_rev) <= max(1.0, 0.005 * ord_rev)}) # 6. YoY decomposition identity: volume effect + price effect == Δsales (drawer math) if top: dc = customer_yoy_decomp(top[0]['pid'], t, team_id=team_id) recon = dc['volume_effect'] + dc['price_effect'] checks.append({'check': 'Drawer: volume + price effect == Δsales (top customer)', 'a': round(recon, 2), 'b': round(dc['d_sales'], 2), 'gap': round(recon - dc['d_sales'], 2), 'ok': abs(recon - dc['d_sales']) <= max(1.0, 0.005 * abs(dc['d_sales']) + 1)}) # 7. NRR ties to the bridge: ending-existing == last + expansion + contraction + churned nr = nrr(t, team_id=team_id) nrr_recon = nr['starting'] + nr['expansion'] + nr['contraction'] + nr['churned'] checks.append({'check': 'NRR: ending-existing == start + exp + contr + churn', 'a': round(nrr_recon, 2), 'b': round(nr['ending_existing'], 2), 'gap': round(nrr_recon - nr['ending_existing'], 2), 'ok': abs(nrr_recon - nr['ending_existing']) <= 1.0}) # 8. Churn score bounded [0,100] with a valid bucket (drawer pill) if top: cs = customer_churn_score(top[0]['pid'], t, team_id=team_id) ok = 0 <= cs['score'] <= 100 and cs['bucket'] in ('Healthy', 'Watch', 'At-risk', 'Critical') checks.append({'check': f"Churn score in [0,100] with valid bucket ({cs['bucket']})", 'a': cs['score'], 'b': cs['score'], 'gap': 0, 'ok': ok}) # 9. Tier migration is MECE: Σ(flow customers) == distinct customers active in either window tm = tier_migration(t) lf, lt = P.ltm(t) pf, pt = P.prior_ltm(t) active = len(set(_cust_rev(lf, lt)) | set(_cust_rev(pf, pt))) flow_n = sum(f['n'] for f in tm['flows']) checks.append({'check': 'Tier migration: Σ(flow customers) == active (LTM ∪ prior-LTM)', 'a': flow_n, 'b': active, 'gap': flow_n - active, 'ok': flow_n == active}) return checks