"""Products / SKU module — SKU health: YoY movers (risers & decliners), zombie SKUs (catalog rot — sellable, formerly selling, now dead), new winners, coverage collapse (SKUs losing customer breadth — the FFS-recovery early-warning signal), and velocity leaders. Per-SKU margin already lives in the Financial module (low_margin_skus); not duplicated here. Basket / co-purchase (MBA) is intentionally deferred — the existing client app computes it runtime-side, and a local co-occurrence pull over 74k LTM lines is the kind of heavy job the project guardrails keep off the local PC. (See BACKLOG.) All line-level (sale.order.line), RI+FFS scope, excluded accounts removed — reusing sale_line_domain. Coverage (distinct customers per SKU) uses a 2-level read_group and is a touch slow (~15s/ window), so it's its own function the app calls lazily and caches. """ import sys import datetime as dt from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from functools import lru_cache import core.odoo as O import core.periods as P import modules.sales as sales_mod # SKU-health views are about real products — exclude service/delivery pseudo-SKUs # (Delivery Charges otherwise dominate movers/winners). Dot-path FILTER works (groupby # on the dot-path does not). Validation reconciles on this same non-service universe. _NO_SVC = [('product_id.type', '!=', 'service')] @lru_cache(maxsize=1) def _code_map(): """product_id → SKU code. Multiple product records can share a default_code (re-SKUing /duplicates); keying by code merges them so a re-coded item doesn't read as a fake decliner + fake riser. Products without a code fall back to a per-id key. Archived products are INCLUDED — re-SKUing typically archives the old record and creates a new one under the same code, and the old record still carries last-year sales.""" prods = O.search_read('product.product', [('active', 'in', [True, False])], ['id', 'default_code'], limit=50000) return {p['id']: (str(p['default_code']).strip() if p.get('default_code') else f"pid:{p['id']}") for p in prods} def _sku_rev(date_from, date_to, team_id=None): """{sku_code: {'name','rev','qty','orders'}} over a window (services excluded, duplicate product records merged by SKU code).""" g = O.read_group('sale.order.line', O.sale_line_domain(date_from, date_to, team_id, extra=_NO_SVC), ['price_subtotal:sum', 'product_uom_qty:sum'], ['product_id'], lazy=False) codes = _code_map() out = {} for r in g: pid = O.m2o_id(r.get('product_id')) if not pid: continue key = codes.get(pid, f"pid:{pid}") e = out.setdefault(key, {'name': O.m2o_name(r.get('product_id')), 'rev': 0.0, 'qty': 0.0, 'orders': 0}) e['rev'] += r.get('price_subtotal') or 0.0 e['qty'] += r.get('product_uom_qty') or 0.0 e['orders'] += r.get('__count') or 0 return out def yoy_movers(t=None, limit=20, team_id=None): """Top SKU risers and decliners by YTD-vs-same-period-LY revenue change.""" t = t or P.today() yf, yt = P.ytd(t) lf, lt = P.ytd_last_year(t) this = _sku_rev(yf, yt, team_id) last = _sku_rev(lf, lt, team_id) rows = [] for pid in set(this) | set(last): tr = this.get(pid, {'rev': 0.0, 'name': last.get(pid, {}).get('name', '')}) lr = last.get(pid, {'rev': 0.0}) name = this.get(pid, {}).get('name') or last.get(pid, {}).get('name') or '' rows.append({'code': pid, 'product': name, 'rev_ytd': tr['rev'], 'rev_ly': lr['rev'], 'change': tr['rev'] - lr['rev']}) risers = sorted([r for r in rows if r['change'] > 0], key=lambda x: -x['change'])[:limit] decliners = sorted([r for r in rows if r['change'] < 0], key=lambda x: x['change'])[:limit] return {'risers': risers, 'decliners': decliners} def zombie_skus(t=None, limit=30, min_prior=1500.0, team_id=None): """Catalog rot: SKUs that sold materially last year but are ~dead this year.""" t = t or P.today() yf, yt = P.ytd(t) lf, lt = P.ytd_last_year(t) this = _sku_rev(yf, yt, team_id) last = _sku_rev(lf, lt, team_id) rows = [] for pid, lr in last.items(): if lr['rev'] < min_prior: continue tr = this.get(pid, {'rev': 0.0}) if tr['rev'] > 0.05 * lr['rev']: # still selling at >5% of prior → not a zombie continue rows.append({'code': pid, 'product': lr['name'], 'rev_ly': lr['rev'], 'rev_ytd': tr['rev'], 'lost': lr['rev'] - tr['rev']}) rows.sort(key=lambda x: -x['lost']) return rows[:limit] def new_winners(t=None, limit=20, min_this=1500.0, team_id=None): """SKUs that barely sold last year but are selling well this year.""" t = t or P.today() yf, yt = P.ytd(t) lf, lt = P.ytd_last_year(t) this = _sku_rev(yf, yt, team_id) last = _sku_rev(lf, lt, team_id) rows = [] for pid, tr in this.items(): if tr['rev'] < min_this: continue lr = last.get(pid, {'rev': 0.0}) if lr['rev'] > 0.05 * tr['rev']: continue rows.append({'code': pid, 'product': tr['name'], 'rev_ytd': tr['rev'], 'rev_ly': lr['rev'], 'gained': tr['rev'] - lr['rev']}) rows.sort(key=lambda x: -x['gained']) return rows[:limit] def velocity_leaders(t=None, limit=25, team_id=None): """Top SKUs by LTM unit velocity (units/month) and revenue.""" t = t or P.today() lf, lt = P.ltm(t) sku = _sku_rev(lf, lt, team_id) rows = [{'code': k, 'product': v['name'], 'units_ltm': v['qty'], 'units_per_mo': v['qty'] / 12.0, 'rev_ltm': v['rev'], 'orders_ltm': v['orders']} for k, v in sku.items()] rows.sort(key=lambda x: -x['units_ltm']) return rows[:limit] def _code_category(): """code -> category name. Built from _code_map (pid->code) + the sales category map (pid->category); the first record carrying a code sets that code's category.""" codes = _code_map() cats = sales_mod._product_cat() out = {} for pid, code in codes.items(): if code not in out and pid in cats: out[code] = cats[pid] return out def directory(t=None, team_id=None): """Every SKU that sold this YTD or the same period last year, with the fields the SKU drill-down filters/sorts on: category, YTD/LY revenue, YoY %, units and orders. Mirrors the customer directory — the full list (filtering happens in the UI), sorted by YTD revenue.""" t = t or P.today() yf, yt = P.ytd(t) lf, lt = P.ytd_last_year(t) this = _sku_rev(yf, yt, team_id) last = _sku_rev(lf, lt, team_id) code_cat = _code_category() rows = [] for code in set(this) | set(last): tr = this.get(code, {}) lr = last.get(code, {}) rev_ytd = tr.get('rev', 0.0) rev_ly = lr.get('rev', 0.0) rows.append({'code': code, 'product': tr.get('name') or lr.get('name') or code, 'category': code_cat.get(code, '(uncategorized)'), 'rev_ytd': rev_ytd, 'rev_ly': rev_ly, 'change': rev_ytd - rev_ly, 'yoy_pct': P.yoy_pct(rev_ytd, rev_ly), 'qty_ytd': tr.get('qty', 0.0), 'orders_ytd': tr.get('orders', 0)}) rows.sort(key=lambda r: -r['rev_ytd']) return rows def catalogue(): """`{sku_code: {'product', 'category'}}` — EVERY ACTIVE product, sold or not. ⛔ THIS IS THE CATALOGUE UNIVERSE, AND IT IS DELIBERATELY NOT `directory()`. `directory()`'s row set IS the union of two revenue `read_group`s over `sale.order.line` (`:155`), so a SKU that never sold cannot exist in it. That is CORRECT for its own callers — `yoy_movers`, `zombie_skus`, `new_winners` and `categories` all legitimately want a sales-window universe — and it is wrong for the PRODUCT GRID, which is a catalogue and was therefore showing 2,717 of 5,875 SKUs (wave 29, owner item 22 / ruling R12). The fix is this function plus a LEFT JOIN in the consumer, never a window removed from `directory()`: removing the window alone lands at 3,327 (all-time-sold), because ~2,550 active SKUs have never sold in wholesale scope at all. ⚠ NOT BU-SHAPED, and it cannot be: `product.product` carries no team. A catalogue is one catalogue. `directory()` stays the BU-shaped half, which is why the two are joined rather than merged — a scoped caller gets every SKU with ITS OWN revenue, blank where that BU never sold. Keyed exactly like `_code_map()` — the `default_code`, or a `pid:N` fallback for the 33 active records that carry none — so the join against `directory()` is code-for-code with no normaliser. Names come from `display_name` (`[CODE] NAME`), which is what `O.m2o_name` yields off a sale line, so a never-sold row wears the same format as a sold one. ⛔ RAISES on a truncated read rather than returning a short catalogue. A silently short pull would put the grid back at a plausible wrong number with every gate green — the exact failure this function exists to end ([[no-unverifiable-aggregates]], and the same truncation guard `modules/backorders.py:161` already uses). """ dom = [('active', '=', True)] prods = O.search_read('product.product', dom, ['id', 'default_code', 'display_name', 'name'], limit=50000) n = O.get_odoo().search_count('product.product', dom) if len(prods) != n: raise ValueError( f"products.catalogue: the product pull is TRUNCATED — read {len(prods)} rows against " f"a search_count of {n}. A short catalogue renders as a plausible smaller grid with " f"nothing reporting it; raise the limit before shipping this.") code_cat = _code_category() out = {} for p in prods: code = (str(p['default_code']).strip() if p.get('default_code') else f"pid:{p['id']}") # First record wins, matching `_code_category`'s own convention. MEASURED 2026-08-11: zero # active products share a code, so this branch is a guard, not a merge policy. out.setdefault(code, {'product': p.get('display_name') or p.get('name') or code, 'category': code_cat.get(code, '(uncategorized)')}) return out #: ⭐⭐ WAVE 30 / W30-T34 (the carried W29-T52) — THE PRICELIST STRATUM, PER LIST. #: #: MEASURED LIVE 2026-08-12, and every one of these numbers shaped the design rather than #: decorating it: #: #: · **Five pricelists, three of them material.** Date-valid fixed rules: Fisch 5,139, #: Royal 1 2,702, Royal 2 2,605, Public Pricelist 17, Giftware Deals 1. At SKU level that #: is Fisch 3,971 · Royal 1 1,920 · Royal 2 1,857 · Public 6 · Giftware 0. #: · **3,988 of 5,871 active products carry a usable fixed rule; 1,883 carry NONE.** #: · **1,853 SKUs are priced on all three lists AND THE LISTS DISAGREE** — median relative #: spread 16%, max 82%. That is why this is three columns rather than one "the price" or a #: low/high pair: neither can tell a Fisch rep what Fisch sells the SKU for, which is the #: only question the column exists to answer. #: · ⛔ **`list_price` is 1.00 on 5,817 of the 5,871** — so the single `3_global` rule #: computed over it is meaningless, exactly as W29-T52 said. **This reader never falls #: through to it**, and that is the ticket's own negative control: a SKU with no specific #: item renders BLANK, never a fallback dressed as a price. #: #: ⚠ RESOLVED BY NAME, NOT BY ID. A pricelist renamed or deleted in Odoo makes its column go #: blank and makes `report["lists_missing"]` name it — a hard-coded id would keep pointing at #: whatever inherited it and mislabel every cell silently. PRICELIST_COLUMNS = ( ("price_fisch", "Fisch"), ("price_royal_1", "Royal 1"), ("price_royal_2", "Royal 2"), ) def pricelist_by_code(): """`({code: {column_key: price}}, report)` — the date-valid FIXED base-tier price for each declared pricelist, keyed EXACTLY as `catalogue()` keys its rows. ⛔ THE KEYING IS LOAD-BEARING, not a detail. `catalogue()` keys on `default_code` with a `pid:{id}` fallback for the ~33 active records that carry none. Keying this map any other way would leave those SKUs permanently blank while an oracle counting active products counted them — a red gate on a working build, or worse, a silent hole nobody counts. **The base tier, deliberately.** `pricecomp._tier_for` picks the highest `min_quantity` at or below an order's quantity, because it is pricing a LINE that has one. A catalogue column has no quantity in hand, so it takes the LOWEST `min_quantity` — the price at qty 1. Rules above that break are a bulk price, and `report["qty_break_only"]` counts the SKUs whose only rule sits on one (MEASURED: 10 rules of 10,464 carry a break at all). **Variant rules beat template rules**, matching `pricecomp._tier_for`: a `0_product_variant` rule is the more specific statement about this exact SKU. ⚠ DEGRADES TO `({}, report)` on a read failure, matching `_inventory_by_code` rather than `catalogue()`: these are COLUMNS, and a product grid that will not render because pricing is momentarily unreachable is a worse failure than one with blank price columns. The blanks are not silent — `product_data.validate()`'s coverage leg reconciles each column against a fresh Odoo count and goes red at zero. ⭐ **The report exists because R6's second sentence is law** (*"if it can't be done, you need to explicitly tell me why and recommend a fix"*). Everything this reader CANNOT see is counted rather than dropped: rules on pricelists the contract does not declare, non-fixed (`formula`/`percent`) rules, the `3_global` fallback, and prices that only exist above a quantity break. """ # ⚠ THE COUNTERS ARE DISJOINT AND ORDER-DEPENDENT, and saying so is the difference between a # report and a misleading one. A rule is classified ONCE, by the first reason it is skipped: # undeclared list → out of date → not fixed → global → zero price. So `rules_not_fixed: 0` # means "no formula rule on a list we declare", NOT "this Odoo has no formula rules" # (MEASURED 2026-08-12: it has exactly one, and it sits on Public Pricelist, which the # contract does not declare — so it lands in `rules_undeclared_list`). report = {"lists_missing": [], "rules_total": 0, "rules_undeclared_list": 0, "rules_not_fixed": 0, "rules_global": 0, "rules_out_of_date": 0, "rules_zero_price": 0, "qty_break_only": 0} try: pl_rows = O.search_read('product.pricelist', [], ['id', 'name']) by_name = {} for p in pl_rows: by_name.setdefault(str(p.get('name') or '').strip(), p['id']) wanted = {} for col, name in PRICELIST_COLUMNS: pid = by_name.get(name) if pid is None: report["lists_missing"].append(name) else: wanted[pid] = col # ⭐⭐ FILTERED SERVER-SIDE, AND THE REASON IS A MEASUREMENT, NOT A STYLE PREFERENCE. # Reading all 10,464 rules with 9 fields and sorting them in Python costs **30.1s** on # this connection; the same rules under a server-side domain with 6 fields cost # **13.6s** (measured 2026-08-12, back to back). `pool()` runs this on a scope's # first-ever build, so that 16.5s is 16.5s of somebody's page load. # # ⚠ The domain reproduces the Python predicate EXACTLY, and the null legs are the part # that is easy to get wrong: an absent `date_start` is `False`, not a past date, so # `('date_start','<=',today)` ALONE would drop every open-ended rule — which is almost # all of them. today = P.today().isoformat() _declared = sorted(wanted) _kinds = ['0_product_variant', '1_product'] _live = [('pricelist_id', 'in', _declared), ('compute_price', '=', 'fixed'), ('applied_on', 'in', _kinds), '|', ('date_start', '=', False), ('date_start', '<=', today), '|', ('date_end', '=', False), ('date_end', '>=', today), ('fixed_price', '>', 0)] rules = O.search_read( 'product.pricelist.item', _live, ['pricelist_id', 'product_id', 'product_tmpl_id', 'applied_on', 'fixed_price', 'min_quantity']) if _declared else [] # ⭐ R6's SECOND SENTENCE, PAID FOR WITH `search_count` RATHER THAN A WIDER READ. Six # counts cost ~1.2s together; the rules they count would cost 16s to read. What this # reader cannot see is still REPORTED — it is just no longer transferred. def _n(extra): try: return O.get_odoo().search_count('product.pricelist.item', extra) except Exception: return -1 # -1 reads as "not measured", never as zero _dated = ['|', ('date_start', '=', False), ('date_start', '<=', today), '|', ('date_end', '=', False), ('date_end', '>=', today)] report["rules_total"] = _n([]) report["rules_undeclared_list"] = _n([('pricelist_id', 'not in', _declared)]) \ if _declared else report["rules_total"] if _declared: _on = [('pricelist_id', 'in', _declared)] # `percent_price` and formula rules are read NOWHERE in this repo. A rule we cannot # price is one the operator is TOLD about, never one that quietly becomes a blank. report["rules_not_fixed"] = _n(_on + [('compute_price', '!=', 'fixed')]) report["rules_global"] = _n(_on + [('compute_price', '=', 'fixed'), ('applied_on', 'not in', _kinds)]) report["rules_out_of_date"] = ( _n(_on + [('compute_price', '=', 'fixed'), ('applied_on', 'in', _kinds)]) - _n(_on + [('compute_price', '=', 'fixed'), ('applied_on', 'in', _kinds)] + _dated)) # A zero price is not "free" — it is an unset rule. Blank says so; 0 does not. report["rules_zero_price"] = _n( _on + [('compute_price', '=', 'fixed'), ('applied_on', 'in', _kinds)] + _dated + [('fixed_price', '<=', 0)]) by_var, by_tmpl = {}, {} for r in rules: col = wanted[O.m2o_id(r.get('pricelist_id'))] if r.get('applied_on') == '0_product_variant' and r.get('product_id'): by_var.setdefault((col, O.m2o_id(r['product_id'])), []).append(r) elif r.get('product_tmpl_id'): by_tmpl.setdefault((col, O.m2o_id(r['product_tmpl_id'])), []).append(r) dom = [('active', '=', True)] prods = O.search_read('product.product', dom, ['id', 'default_code', 'product_tmpl_id'], limit=50000) n = O.get_odoo().search_count('product.product', dom) if len(prods) != n: # Same guard, same reason as `catalogue()`: a short pull renders as a plausible # smaller set of priced SKUs with nothing reporting it. raise ValueError( f"products.pricelist_by_code: the product pull is TRUNCATED — read {len(prods)} " f"rows against a search_count of {n}.") except Exception as e: report["error"] = f"{type(e).__name__}: {str(e)[:200]}" return {}, report out = {} for p in prods: code = (str(p['default_code']).strip() if p.get('default_code') else f"pid:{p['id']}") tmpl = O.m2o_id(p.get('product_tmpl_id')) cells = {} for col, _name in PRICELIST_COLUMNS: cands = by_var.get((col, p['id'])) or by_tmpl.get((col, tmpl)) if not cands: continue base = min(cands, key=lambda r: r.get('min_quantity') or 0.0) if (base.get('min_quantity') or 0.0) > 1.0: report["qty_break_only"] += 1 cells[col] = base.get('fixed_price') if cells: out.setdefault(code, {}).update(cells) return out, report def catalogue_count(): """The INDEPENDENT population oracle: Odoo's own count of active products. Deliberately a bare `search_count` and not a `len()` over anything this module built — the 2,717 defect shipped silently for a wave because `product_data.validate()` derived BOTH sides of its reconciliation from `_sku_rev`, so the oracle could never see a missing row. """ return O.get_odoo().search_count('product.product', [('active', '=', True)]) def categories(t=None, team_id=None): """Sorted distinct category names present in the SKU directory (for the drill-down filter).""" return sorted({r['category'] for r in directory(t, team_id)}) def _coverage(date_from, date_to, team_id=None): """{sku_code: distinct_customer_count} via 2-level read_group (slow-ish), merged by code. (A customer buying two records sharing a code can count twice; duplicates are rare, and code-level avoids the bigger error of a re-SKUed product reading as full coverage loss.)""" g = O.read_group('sale.order.line', O.sale_line_domain(date_from, date_to, team_id, extra=_NO_SVC), ['__count'], ['product_id', 'order_partner_id'], lazy=False) codes = _code_map() cov = {} names = {} for r in g: pid = O.m2o_id(r.get('product_id')) if not pid: continue key = codes.get(pid, f"pid:{pid}") cov[key] = cov.get(key, 0) + 1 names.setdefault(key, O.m2o_name(r.get('product_id'))) return cov, names def coverage_collapse(t=None, limit=25, min_prior_custs=8, team_id=None): """SKUs that lost the most customer breadth YoY — early warning a SKU is dying even if revenue hasn't fully cratered yet.""" t = t or P.today() yf, yt = P.ytd(t) lf, lt = P.ytd_last_year(t) cov_t, names_t = _coverage(yf, yt, team_id) cov_l, names_l = _coverage(lf, lt, team_id) rows = [] for pid, lc in cov_l.items(): if lc < min_prior_custs: continue tc = cov_t.get(pid, 0) drop = lc - tc if drop <= 0: continue rows.append({'code': pid, 'product': names_l.get(pid) or names_t.get(pid) or '', 'custs_ly': lc, 'custs_ytd': tc, 'lost_custs': drop, 'pct_drop': drop / lc * 100}) rows.sort(key=lambda x: (-x['lost_custs'], -x['pct_drop'])) return rows[:limit] # ====================================================================== SKU DRAWER (mirror of customer drawer) def _sku_product_ids(code): """All product.product ids sharing this SKU code (merged variants / archived records).""" return [pid for pid, c in _code_map().items() if c == code] def _sku_dom(pids, date_from, date_to, team_id=None): return O.sale_line_domain(date_from, date_to, team_id, extra=[('product_id', 'in', pids)] + _NO_SVC) def _sku_name_category(pids): rows = O.search_read('product.product', [('id', 'in', pids)], ['name', 'categ_id']) name = rows[0]['name'] if rows else '(unknown)' catmap = sales_mod._product_cat() main = next((catmap.get(p) for p in pids if catmap.get(p)), '(uncategorized)') return name, main def _sku_buyers(pids, date_from, date_to, team_id=None): """{partner_id: {'name','rev','qty'}} for buyers of this SKU over a window.""" g = O.read_group('sale.order.line', _sku_dom(pids, date_from, date_to, team_id), ['price_subtotal:sum', 'product_uom_qty:sum'], ['order_partner_id'], lazy=False) out = {} for r in g: pid = O.m2o_id(r.get('order_partner_id')) if pid: out[pid] = {'name': O.m2o_name(r.get('order_partner_id')), 'rev': r.get('price_subtotal') or 0.0, 'qty': r.get('product_uom_qty') or 0.0} return out def sku_detail(code, t=None, team_id=None, n_months=13, allsku=None): """KPIs (rev/qty/buyers YoY, GM%), monthly trend, rank & % of BU for one SKU code. Pass `allsku` (a cached _sku_rev YTD map) to skip the ~all-SKU rank read.""" t = t or P.today() pids = _sku_product_ids(code) if not pids: return None yf, yt = P.ytd(t) lf, lt = P.ytd_last_year(t) mf, mt = P.ltm(t) name, category = _sku_name_category(pids) def s(df, dtt, field='price_subtotal'): return O.sum_field('sale.order.line', _sku_dom(pids, df, dtt, team_id), field) rev_ytd, rev_ly = s(yf, yt), s(lf, lt) qty_ytd, qty_ly = s(yf, yt, 'product_uom_qty'), s(lf, lt, 'product_uom_qty') buyers_ytd = len(_sku_buyers(pids, yf, yt, team_id)) buyers_ly = len(_sku_buyers(pids, lf, lt, team_id)) g = O.read_group('sale.order.line', _sku_dom(pids, mf, mt, team_id), ['price_subtotal:sum', 'margin:sum'], [], lazy=False) line_rev = (g[0].get('price_subtotal') if g else 0) or 0.0 margin = (g[0].get('margin') if g else 0) or 0.0 # Monthly trend: bin lines by their order's month in Python (dot-path month groupby is rejected # on sale.order.line), in 2 reads instead of 26 point queries. range_start = dt.date(t.year - 2, t.month, 1).isoformat() lines = O.search_read('sale.order.line', _sku_dom(pids, range_start, t.isoformat(), team_id), ['price_subtotal', 'order_id']) oids = list({O.m2o_id(line['order_id']) for line in lines if line.get('order_id')}) odate = {} for i in range(0, len(oids), 1000): for o in O.search_read('sale.order', [('id', 'in', oids[i:i + 1000])], ['date_order']): if o.get('date_order'): odate[o['id']] = str(o['date_order'])[:7] mrev = {} for line in lines: ym = odate.get(O.m2o_id(line.get('order_id'))) if ym: mrev[ym] = mrev.get(ym, 0.0) + (line.get('price_subtotal') or 0.0) 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)}) allsku = allsku if allsku is not None else _sku_rev(yf, yt, team_id) total = sum(v['rev'] for v in allsku.values()) or 1.0 rank = next((i + 1 for i, (c, _v) in enumerate(sorted(allsku.items(), key=lambda kv: -kv[1]['rev'])) if c == code), None) return { 'code': code, 'name': name, 'category': category, 'rev_ytd': rev_ytd, 'rev_ly': rev_ly, 'rev_yoy_pct': P.yoy_pct(rev_ytd, rev_ly), 'qty_ytd': qty_ytd, 'qty_ly': qty_ly, 'qty_yoy_pct': P.yoy_pct(qty_ytd, qty_ly), 'buyers_ytd': buyers_ytd, 'buyers_ly': buyers_ly, 'buyers_delta': buyers_ytd - buyers_ly, 'gm_pct': (margin / line_rev * 100) if line_rev else 0.0, 'gm_dollars': margin, 'rank': rank, 'n_skus': len(allsku), 'pct_of_bu': rev_ytd / total * 100, 'monthly': monthly, } def sku_buyer_bridge(code, t=None, team_id=None, top=12): """Who drives the SKU's YoY: retained / new / churned buyers + a $-ranked churned call list.""" t = t or P.today() pids = _sku_product_ids(code) yf, yt = P.ytd(t) lf, lt = P.ytd_last_year(t) this, last = _sku_buyers(pids, yf, yt, team_id), _sku_buyers(pids, lf, lt, team_id) tset, lset = set(this), set(last) retained, new, churned = tset & lset, tset - lset, lset - tset churned_list = sorted([{'pid': p, 'customer': last[p]['name'], 'ly_rev': last[p]['rev']} for p in churned], key=lambda x: -x['ly_rev'])[:top] return {'retained': {'n': len(retained), 'rev': sum(this[p]['rev'] for p in retained)}, 'new': {'n': len(new), 'rev': sum(this[p]['rev'] for p in new)}, 'churned': {'n': len(churned), 'rev': sum(last[p]['rev'] for p in churned)}, 'buyer_retention_pct': (len(retained) / len(lset) * 100) if lset else 0.0, 'churned_buyers': churned_list, 'buyers_this': len(tset), 'buyers_last': len(lset), 'this_total': sum(v['rev'] for v in this.values())} def sku_concentration(code, t=None, team_id=None): """Buyer-concentration risk: top-1/top-3 share, Herfindahl index, effective buyer count (LTM).""" t = t or P.today() mf, mt = P.ltm(t) buyers = _sku_buyers(_sku_product_ids(code), mf, mt, team_id) revs = sorted([v['rev'] for v in buyers.values()], reverse=True) total = sum(revs) or 1.0 hhi = sum((r / total) ** 2 for r in revs) return {'n_buyers': len(revs), 'top1_pct': (revs[0] / total * 100) if revs else 0.0, 'top3_pct': (sum(revs[:3]) / total * 100) if revs else 0.0, 'hhi': hhi, 'eff_buyers': (1 / hhi) if hhi else 0.0} def sku_price_dispersion(code, t=None, team_id=None, cap=60): """Realized $/unit per buyer (LTM) vs the volume-weighted average; recoverable $ on below-VWAP accounts. list_price is unreliable, so the dispersion among actual buyers is the margin lever.""" t = t or P.today() mf, mt = P.ltm(t) g = O.read_group('sale.order.line', _sku_dom(_sku_product_ids(code), mf, mt, team_id), ['price_subtotal:sum', 'product_uom_qty:sum'], ['order_partner_id'], lazy=False) rows, tot_rev, tot_qty = [], 0.0, 0.0 for r in g: pid = O.m2o_id(r.get('order_partner_id')) rev = r.get('price_subtotal') or 0.0 qty = r.get('product_uom_qty') or 0.0 if not pid or qty <= 0: continue rows.append({'pid': pid, 'customer': O.m2o_name(r.get('order_partner_id')), 'price': rev / qty, 'qty': qty, 'rev': rev}) tot_rev += rev tot_qty += qty vwap = (tot_rev / tot_qty) if tot_qty else 0.0 for r in rows: r['recoverable'] = max(0.0, vwap - r['price']) * r['qty'] rows.sort(key=lambda x: -x['recoverable']) return {'vwap': vwap, 'n_buyers': len(rows), 'recoverable_total': sum(r['recoverable'] for r in rows), 'rows': rows[:cap]} def sku_top_buyers(code, t=None, team_id=None, top=15): """Ranked buyers of this SKU (YTD) with YoY — clickable to open the customer drawer.""" t = t or P.today() pids = _sku_product_ids(code) yf, yt = P.ytd(t) lf, lt = P.ytd_last_year(t) this, last = _sku_buyers(pids, yf, yt, team_id), _sku_buyers(pids, lf, lt, team_id) rows = [{'pid': p, 'customer': v['name'], 'rev': v['rev'], 'qty': v['qty'], 'yoy_pct': P.yoy_pct(v['rev'], last.get(p, {}).get('rev', 0.0))} for p, v in this.items()] rows.sort(key=lambda x: -x['rev']) return rows[:top] _BUYER_TIERS = [('Whale (≥$25k)', 25000.0), ('Large ($10–25k)', 10000.0), ('Mid ($2–10k)', 2000.0), ('Small (<$2k)', 0.0)] def sku_customer_analysis(code, t=None, team_id=None, top=15): """WHO buys this SKU (LTM), as customers: the value-tier mix of its buyers (by each buyer's TOTAL spend), and the accounts most DEPENDENT on it (this SKU as a share of their spend — who gets hurt most if it stocks out / who to protect).""" t = t or P.today() lf, lt = P.ltm(t) pids = _sku_product_ids(code) empty = {'segments': [], 'dependency': [], 'n_buyers': 0, 'avg_dependency': 0.0} if not pids: return empty buyers = _sku_buyers(pids, lf, lt, team_id) # {pid: {name, rev(this SKU), qty}} if not buyers: return empty bpids = list(buyers) g = O.read_group('sale.order', sales_mod.order_domain(lf, lt, team_id) + [('partner_id', 'in', bpids)], ['amount_untaxed:sum'], ['partner_id'], lazy=False) total = {O.m2o_id(r['partner_id']): (r.get('amount_untaxed') or 0.0) for r in g if r.get('partner_id')} def tier(rev): for nm, lo in _BUYER_TIERS: if rev >= lo: return nm return _BUYER_TIERS[-1][0] seg = {nm: {'tier': nm, 'buyers': 0, 'sku_rev': 0.0} for nm, _ in _BUYER_TIERS} dep = [] for p, v in buyers.items(): ct = total.get(p, v['rev']) or v['rev'] s = seg[tier(ct)] s['buyers'] += 1 s['sku_rev'] += v['rev'] dep.append({'customer': v['name'], 'pid': p, 'sku_rev': v['rev'], 'cust_total': ct, 'dependency_pct': (v['rev'] / ct * 100) if ct else None}) tot = sum(s['sku_rev'] for s in seg.values()) or 1.0 segments = [] for nm, _ in _BUYER_TIERS: s = seg[nm] s['rev_share'] = s['sku_rev'] / tot * 100 s['avg_per_buyer'] = (s['sku_rev'] / s['buyers']) if s['buyers'] else 0.0 segments.append(s) dep.sort(key=lambda x: -x['sku_rev']) deps_known = [d['dependency_pct'] for d in dep if d['dependency_pct'] is not None] return {'segments': segments, 'dependency': dep[:top], 'n_buyers': len(buyers), 'avg_dependency': (sum(deps_known) / len(deps_known)) if deps_known else 0.0} def sku_whitespace(code, t=None, team_id=None, top=15): """Customers who buy this SKU's category but NOT this SKU — ranked prospect list.""" import modules.customers as cust_mod t = t or P.today() pids = _sku_product_ids(code) _name, category = _sku_name_category(pids) cat_buyers = cust_mod.category_buyers(category, team_id=team_id) or set() mf, mt = P.ltm(t) prospects = list(cat_buyers - set(_sku_buyers(pids, mf, mt, team_id))) if not prospects: return {'category': category, 'rows': []} g = O.read_group('sale.order', sales_mod.order_domain(mf, mt, team_id) + [('partner_id', 'in', prospects)], ['amount_untaxed:sum'], ['partner_id'], lazy=False) rows = [{'pid': O.m2o_id(r['partner_id']), 'customer': O.m2o_name(r['partner_id']), 'total_spend': r.get('amount_untaxed') or 0.0} for r in g if r.get('partner_id')] rows.sort(key=lambda x: -x['total_spend']) return {'category': category, 'rows': rows[:top]} def sku_drawer_bundle(code, t=None, team_id=None, allsku=None): """The whole SKU drawer's first paint in one cached unit: sku_detail first (the not-found gate), then the other six pulls CONCURRENTLY (O.parallel). Cuts a ~6-call cold open to ~max(call).""" detail = sku_detail(code, t=t, team_id=team_id, allsku=allsku) if detail is None: return {'detail': None} bridge, conc, price, buyers, white, ca = O.parallel([ lambda: sku_buyer_bridge(code, t, team_id), lambda: sku_concentration(code, t, team_id), lambda: sku_price_dispersion(code, t, team_id), lambda: sku_top_buyers(code, t, team_id), lambda: sku_whitespace(code, t, team_id), lambda: sku_customer_analysis(code, t, team_id), ]) return {'detail': detail, 'bridge': bridge, 'conc': conc, 'price': price, 'buyers': buyers, 'white': white, 'ca': ca} def validate(t=None, team_id=None): """Reconcile SKU metrics to independent Odoo aggregates. When team_id is set (a single BU selected) every check runs SCOPED to that BU, so the validation panel never reconciles against — or exposes — the other BU's numbers. All checks here are BU-scopeable (no cross-BU mirror), so team_id threads straight through.""" t = t or P.today() yf, yt = P.ytd(t) checks = [] sku = _sku_rev(yf, yt, team_id) sku_sum = sum(v['rev'] for v in sku.values()) line_total = O.sum_field('sale.order.line', O.sale_line_domain(yf, yt, team_id, extra=_NO_SVC), 'price_subtotal') checks.append({'check': 'SKU rev: Σ(per-SKU) == total line revenue, ex-services (YTD)', 'a': round(sku_sum, 2), 'b': round(line_total, 2), 'gap': round(sku_sum - line_total, 2), 'ok': abs(sku_sum - line_total) <= 1.0}) m = yoy_movers(t, limit=10**9, team_id=team_id) movers_sum = sum(r['change'] for r in m['risers']) + sum(r['change'] for r in m['decliners']) lf, lt = P.ytd_last_year(t) last_total = O.sum_field('sale.order.line', O.sale_line_domain(lf, lt, team_id, extra=_NO_SVC), 'price_subtotal') checks.append({'check': 'SKU movers: Σ(Δ) == (YTD − LY) total', 'a': round(movers_sum, 2), 'b': round(line_total - last_total, 2), 'gap': round(movers_sum - (line_total - last_total), 2), 'ok': abs(movers_sum - (line_total - last_total)) <= 1.0}) # SKU drawer: buyer-bridge reconciles — retained + new buyer revenue == this-period SKU revenue top_code = max(sku.items(), key=lambda kv: kv[1]['rev'])[0] if sku else None if top_code: bb = sku_buyer_bridge(top_code, t, team_id=team_id) recon = bb['retained']['rev'] + bb['new']['rev'] checks.append({'check': 'SKU drawer: retained+new buyer rev == SKU YTD revenue', 'a': round(recon, 2), 'b': round(bb['this_total'], 2), 'gap': round(recon - bb['this_total'], 2), 'ok': abs(recon - bb['this_total']) <= 1.0}) return checks