File size: 8,328 Bytes
c14ceee | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 | """Price compliance — selling below the customer's pricelist tier (pocket-price floor).
Discount FIELDS are unused in this Odoo (verified dead), but reps can still key a low
price_unit directly — leakage that is invisible today. Per (customer × SKU) over the LTM:
realized unit price (Σ subtotal ÷ Σ qty — discount-true) vs the customer's applicable
pricelist tier (fixed-price rules, qty break matched on the average order quantity).
Exceptions below THRESHOLD of tier form the worklist, each valued at its annualized leak
(= (tier − realized) × LTM qty). Pricing canon: ~1% realization ≈ 8-9% operating profit.
v1 simplifications (documented, deliberate):
- qty break matched on AVG order qty per (customer, SKU) — not per order line;
- fixed-price rules only (this instance prices via fixed tiers, like the legacy SKU app);
- LTM window, wholesale scope.
"""
import datetime as dt
import core.odoo as O
import core.periods as P
import modules.customers as cust
THRESHOLD = 0.97 # below 97% of tier = exception
def _rules():
"""All date-valid fixed-price pricelist rules, indexed by (pricelist, product-variant)
and (pricelist, product-template). Highest min_quantity ≤ qty wins at match time."""
today = P.today().isoformat()
rows = O.search_read(
'product.pricelist.item',
[('compute_price', '=', 'fixed'),
('applied_on', 'in', ['0_product_variant', '1_product'])],
['pricelist_id', 'product_id', 'product_tmpl_id', 'min_quantity',
'fixed_price', 'date_start', 'date_end', 'applied_on'])
ok = []
for r in rows:
ds = str(r.get('date_start') or '')[:10]
de = str(r.get('date_end') or '')[:10]
if (ds and ds > today) or (de and de < today):
continue
ok.append(r)
by_var, by_tmpl = {}, {}
for r in ok:
pl = O.m2o_id(r['pricelist_id'])
if r['applied_on'] == '0_product_variant' and r.get('product_id'):
by_var.setdefault((pl, O.m2o_id(r['product_id'])), []).append(r)
elif r.get('product_tmpl_id'):
by_tmpl.setdefault((pl, O.m2o_id(r['product_tmpl_id'])), []).append(r)
return ok, by_var, by_tmpl
def _tier_for(pl, pid, tmpl, qty, by_var, by_tmpl):
"""The applicable fixed tier price: variant rules beat template rules; within a scope,
the highest qty break ≤ qty wins."""
for cands in (by_var.get((pl, pid)), by_tmpl.get((pl, tmpl))):
if not cands:
continue
eligible = [r for r in cands if (r.get('min_quantity') or 0) <= max(qty, 1)]
if eligible:
best = max(eligible, key=lambda r: r.get('min_quantity') or 0)
return best.get('fixed_price') or None
return None
def matched_pairs(team_id=None, t=None):
"""The shared substrate: every (customer × SKU) pair bought in the LTM with its applicable
fixed tier attached (tier=None when the customer has no pricelist / no rule matches).
Used by the compliance sweep here AND by the price corridor + pocket waterfall
(modules/reprice.py) — one definition of 'realized vs tier' everywhere."""
t = t or P.today()
lf, lt = P.ltm(t)
# realized price per (customer, SKU) — one server-side aggregate
g = O.read_group('sale.order.line', O.sale_line_domain(lf, lt, team_id),
['price_subtotal:sum', 'product_uom_qty:sum'],
['order_partner_id', 'product_id'], lazy=False)
pairs = []
for r in g:
pid = O.m2o_id(r.get('order_partner_id'))
prod = O.m2o_id(r.get('product_id'))
qty = r.get('product_uom_qty') or 0.0
rev = r.get('price_subtotal') or 0.0
n = r.get('__count') or 1
if pid and prod and qty > 0 and rev > 0:
pairs.append({'pid': pid, 'customer': O.m2o_name(r.get('order_partner_id')),
'prod': prod, 'product': O.m2o_name(r.get('product_id')),
'qty': qty, 'rev': rev, 'lines': n,
'unit': rev / qty, 'avg_order_qty': qty / n})
# customer pricelists + product template/code lookups
pids = list({p['pid'] for p in pairs})
plists = {}
for i in range(0, len(pids), 2000):
for r in O.search_read('res.partner', [('id', 'in', pids[i:i + 2000])],
['property_product_pricelist']):
plists[r['id']] = O.m2o_id(r.get('property_product_pricelist'))
prods = list({p['prod'] for p in pairs})
pinfo = {}
for i in range(0, len(prods), 2000):
for r in O.search_read('product.product',
[('id', 'in', prods[i:i + 2000]),
('active', 'in', [True, False])],
['product_tmpl_id', 'default_code']):
pinfo[r['id']] = {'tmpl': O.m2o_id(r.get('product_tmpl_id')),
'code': (r.get('default_code') or '').strip()}
all_rules, by_var, by_tmpl = _rules()
for p in pairs:
info = pinfo.get(p['prod'], {})
p['sku'] = info.get('code', '')
pl = plists.get(p['pid'])
tier = (_tier_for(pl, p['prod'], info.get('tmpl'), p['avg_order_qty'],
by_var, by_tmpl) if pl else None)
p['tier'] = tier if (tier and tier > 0) else None
return {'pairs': pairs, 'checked': len(pairs), 'n_rules': len(all_rules),
'window': (lf, lt)}
def build(team_id=None, t=None, pre=None):
"""The compliance sweep. Returns dict(rows[exceptions], checked, matched, kpis…)."""
mp = pre or matched_pairs(team_id, t)
matched_list = [p for p in mp['pairs'] if p.get('tier')]
exceptions = [p for p in matched_list if p['unit'] / p['tier'] < THRESHOLD]
attrs = cust._partner_attrs(list({p['pid'] for p in exceptions}))
rows = []
for p in exceptions:
tier = p['tier']
rows.append({
'pid': p['pid'], 'customer': p['customer'],
'sku': p['sku'], 'product': p['product'],
'agent': (attrs.get(p['pid']) or {}).get('agent', '(none)'),
'tier_price': tier, 'realized': p['unit'],
'pct_of_tier': p['unit'] / tier * 100.0,
'qty_ltm': p['qty'], 'rev_ltm': p['rev'],
'leak': (tier - p['unit']) * p['qty'], 'lines': p['lines'],
})
rows.sort(key=lambda x: -x['leak'])
return {
'rows': rows, 'checked': mp['checked'], 'matched': len(matched_list),
'n_rules': mp['n_rules'],
'leak_total': sum(r['leak'] for r in rows),
'exception_rate': (len(rows) / len(matched_list) * 100.0) if matched_list else 0.0,
'window': mp['window'],
}
def validate(t=None, team_id=None):
"""(1) rule pull complete vs search_count; (2) the TOP exception re-derived row-level:
its realized unit price recomputed from its raw order lines (independent path) must match
the read_group aggregate the sweep used."""
b = build(team_id, t)
n = O.get_odoo().search_count(
'product.pricelist.item',
[('compute_price', '=', 'fixed'),
('applied_on', 'in', ['0_product_variant', '1_product'])])
checks = [{'check': 'fixed pricelist rules — pull complete (pre date-filter)',
'a': b['n_rules'], 'b': f'<= {n}', 'gap': '',
'ok': b['n_rules'] <= n}]
if b['rows']:
top = b['rows'][0]
lf, lt = b['window']
lines = O.search_read('sale.order.line',
O.sale_line_domain(lf, lt, team_id,
partner_ids=[top['pid']],
extra=[('product_id.default_code', '=',
top['sku'])]),
['price_subtotal', 'product_uom_qty'])
qty = sum(l.get('product_uom_qty') or 0 for l in lines)
rev = sum(l.get('price_subtotal') or 0 for l in lines)
unit = rev / qty if qty else 0.0
checks.append({'check': f"top exception realized price — {top['sku']}",
'a': round(unit, 4), 'b': round(top['realized'], 4),
'gap': round(unit - top['realized'], 4),
'ok': abs(unit - top['realized']) < 0.01})
return checks
|