"""Financial module — gross margin (the bridge between Sales revenue and Inventory cost), margin by brand / category / SKU, margin trend & YoY, margin-dilutive SKUs, and the cash-conversion-cycle capstone (DSO + DIO − DPO) that ties the AR + Inventory + Financial working-capital picture together. Margin source: Odoo's Margin module is installed, so sale.order.line carries `margin` (= price_subtotal − purchase_price×qty) and `purchase_price` (snapshot cost at sale). Both are read_group-aggregatable (verified). COGS is derived as revenue − margin. Data-quality caveats surfaced and reported, not hidden: - ~4% of sales-line revenue runs through zero-cost lines → their margin reads as 100% and slightly overstates GM. Quantified in `uncosted_share()` and the validation panel. - product.list_price is uniformly 1.0 in this Odoo (pricing lives in pricelists), so "discount vs list" is not computable here — price realization is expressed as realized GM% instead. (A pricelist-join version is on the backlog.) """ import sys import datetime as dt 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.inventory as inv_mod import modules.ar as ar_mod # ---------- core margin window ---------- def _window(date_from, date_to, team_id=None): """Revenue, margin, derived COGS and GM% for a sale-line window (RI+FFS scope).""" dom = O.sale_line_domain(date_from=str(date_from), date_to=str(date_to), team_id=team_id) rev = O.sum_field('sale.order.line', dom, 'price_subtotal') margin = O.sum_field('sale.order.line', dom, 'margin') cogs = rev - margin return { 'revenue': rev, 'margin': margin, 'cogs': cogs, 'gm_pct': (margin / rev * 100) if rev else 0.0, } def headline(t=None, team_id=None): """YTD gross margin vs same-period last year (consistent with the Sales headline).""" t = t or P.today() yf, yt = P.ytd(t) lf, lt = P.ytd_last_year(t) this = _window(yf, yt, team_id) last = _window(lf, lt, team_id) lf2, lt2 = P.ltm(t) ltm = _window(lf2, lt2, team_id) return { 'window': f'{yf} → {yt}', 'revenue': this['revenue'], 'revenue_ly': last['revenue'], 'revenue_yoy_pct': P.yoy_pct(this['revenue'], last['revenue']), 'cogs': this['cogs'], 'cogs_ly': last['cogs'], 'cogs_yoy_pct': P.yoy_pct(this['cogs'], last['cogs']), 'gm_dollars': this['margin'], 'gm_pct': this['gm_pct'], 'gm_pct_ly': last['gm_pct'], 'gm_pp_delta': this['gm_pct'] - last['gm_pct'], 'gm_dollars_ly': last['margin'], 'gm_dollars_yoy_pct': P.yoy_pct(this['margin'], last['margin']), 'ltm_revenue': ltm['revenue'], 'ltm_gm_dollars': ltm['margin'], 'ltm_gm_pct': ltm['gm_pct'], } def by_brand(t=None): """Fisch vs Royal: revenue, GM$, GM% (YTD) + same-period-LY for variance columns.""" t = t or P.today() yf, yt = P.ytd(t) lf, lt = P.ytd_last_year(t) rows = [] for tid in O.TEAM_IDS: w = _window(yf, yt, team_id=tid) ly = _window(lf, lt, team_id=tid) rows.append({'brand': O.TEAM_NAMES[tid], 'revenue': w['revenue'], 'gm_dollars': w['margin'], 'gm_pct': w['gm_pct'], 'gm_dollars_ly': ly['margin'], 'gm_change': w['margin'] - ly['margin'], 'gm_pct_ly': ly['gm_pct'], 'gm_pp_delta': w['gm_pct'] - ly['gm_pct']}) return sorted(rows, key=lambda r: -r['gm_dollars']) def _product_cat_map(): """product_id → main category name (reusing inventory's category resolver).""" o = O.get_odoo() catmap = inv_mod._cat_main_map() prods = o.search_read('product.product', [('default_code', '!=', False)], ['id', 'categ_id']) return {p['id']: (catmap.get(O.m2o_id(p.get('categ_id'))) or '(uncategorized)') for p in prods} def _per_product(date_from, date_to, team_id=None): """read_group sale lines by product → rev, margin, qty (YTD window).""" o = O.get_odoo() dom = O.sale_line_domain(date_from=str(date_from), date_to=str(date_to), team_id=team_id) g = o.read_group('sale.order.line', domain=dom, fields=['price_subtotal:sum', 'margin:sum', 'product_uom_qty:sum'], groupby=['product_id'], lazy=False) out = [] for r in g: pid = O.m2o_id(r.get('product_id')) if not pid: continue out.append({'product_id': pid, 'name': O.m2o_name(r.get('product_id')), 'revenue': r.get('price_subtotal') or 0.0, 'margin': r.get('margin') or 0.0, 'qty': r.get('product_uom_qty') or 0.0}) return out def by_category(t=None, limit=20, team_id=None): """Where the gross-profit dollars come from, by main category (YTD) — with same-period-LY so the page can show each category's GM$ change and pp drift (the mix/variance read).""" t = t or P.today() yf, yt = P.ytd(t) lf, lt = P.ytd_last_year(t) catmap = _product_cat_map() agg = {} for r in _per_product(yf, yt, team_id): cat = catmap.get(r['product_id'], '(uncategorized)') a = agg.setdefault(cat, {'category': cat, 'revenue': 0.0, 'gm_dollars': 0.0, 'revenue_ly': 0.0, 'gm_dollars_ly': 0.0}) a['revenue'] += r['revenue'] a['gm_dollars'] += r['margin'] for r in _per_product(lf, lt, team_id): cat = catmap.get(r['product_id'], '(uncategorized)') a = agg.setdefault(cat, {'category': cat, 'revenue': 0.0, 'gm_dollars': 0.0, 'revenue_ly': 0.0, 'gm_dollars_ly': 0.0}) a['revenue_ly'] += r['revenue'] a['gm_dollars_ly'] += r['margin'] for a in agg.values(): a['gm_pct'] = (a['gm_dollars'] / a['revenue'] * 100) if a['revenue'] else 0.0 a['gm_pct_ly'] = (a['gm_dollars_ly'] / a['revenue_ly'] * 100) if a['revenue_ly'] else None a['gm_change'] = a['gm_dollars'] - a['gm_dollars_ly'] a['gm_pp_delta'] = (a['gm_pct'] - a['gm_pct_ly']) if a['gm_pct_ly'] is not None else None return sorted(agg.values(), key=lambda x: -x['gm_dollars'])[:limit] def _shift_years(d, years): """Same calendar date shifted by whole years (Feb-29 → Feb-28).""" d = dt.date.fromisoformat(str(d)) try: return d.replace(year=d.year + years).isoformat() except ValueError: return d.replace(year=d.year + years, day=28).isoformat() # Bridge cutoff presets: label + window fn. The compare window is ALWAYS the same dates shifted # back one year (for LTM that IS the prior 12 months), so seasonality never distorts the bridge. BRIDGE_BASES = { 'ytd': ('YTD', P.ytd), 'qtd': ('QTD', P.qtd), 'mtd': ('MTD', P.mtd), 'ltm': ('LTM', P.ltm), } def margin_bridge(t=None, team_id=None, basis='ytd'): """The gross-margin bridge — decomposes ΔGM$ (the `basis` window vs the same window one year earlier) into per-SKU effects that sum EXACTLY (validated to the cent): volume = (Q1−Q0)·(P0−C0) price = (P1−P0)·Q1 cost = −(C1−C0)·Q1 (SKUs in both periods) new = GM1 of SKUs with no prior-window sales lost = −GM0 of SKUs gone this window basis ∈ BRIDGE_BASES: 'ytd' (default), 'qtd', 'mtd', 'ltm' — the owner-selectable cutoff. P/C are realized $-per-unit from the window aggregates (Odoo margin basis). Zero-qty lines (services, adjustments) can't carry a unit price — their ΔGM lands in `other` so the bridge still ties. The classic FP&A "why did margin move" chart (IBCS/PVM canon).""" t = t or P.today() _lbl, _win = BRIDGE_BASES.get(basis, BRIDGE_BASES['ytd']) yf, yt = _win(t) lf, lt = _shift_years(yf, -1), _shift_years(yt, -1) this = {r['product_id']: r for r in _per_product(yf, yt, team_id)} last = {r['product_id']: r for r in _per_product(lf, lt, team_id)} vol = prc = cst = new = lost = other = 0.0 fx = [] # per-SKU effect rows — the drill-down behind each bridge bar for pid, r1 in this.items(): r0 = last.get(pid) if r0 is None: new += r1['margin'] fx.append({'name': r1['name'], 'bucket': 'New SKUs', 'volume': 0.0, 'price': 0.0, 'cost': 0.0, 'change': r1['margin']}) continue q0, q1 = r0['qty'], r1['qty'] if q0 > 0 and q1 > 0: p0, c0 = r0['revenue'] / q0, (r0['revenue'] - r0['margin']) / q0 p1, c1 = r1['revenue'] / q1, (r1['revenue'] - r1['margin']) / q1 v, pr, ct = (q1 - q0) * (p0 - c0), (p1 - p0) * q1, -(c1 - c0) * q1 vol += v prc += pr cst += ct fx.append({'name': r1['name'], 'bucket': 'Continuing', 'volume': v, 'price': pr, 'cost': ct, 'change': r1['margin'] - r0['margin']}) else: # unit price undefined on either side → exact residual bucket other += r1['margin'] - r0['margin'] fx.append({'name': r1['name'], 'bucket': 'Other', 'volume': 0.0, 'price': 0.0, 'cost': 0.0, 'change': r1['margin'] - r0['margin']}) for pid, r0 in last.items(): if pid not in this: lost += -r0['margin'] fx.append({'name': r0['name'], 'bucket': 'Lost SKUs', 'volume': 0.0, 'price': 0.0, 'cost': 0.0, 'change': -r0['margin']}) gm0 = sum(r['margin'] for r in last.values()) gm1 = sum(r['margin'] for r in this.values()) fx.sort(key=lambda r: -abs(r['change'])) return {'gm_ly': gm0, 'gm_ytd': gm1, 'volume': vol, 'price': prc, 'cost': cst, 'new': new, 'lost': lost, 'other': other, 'delta': gm1 - gm0, 'sku_effects': fx[:600], 'basis': basis, 'basis_label': _lbl, 'cmp_label': 'prior LTM' if basis == 'ltm' else 'LY', 'date_from': str(yf), 'date_to': str(yt), 'cmp_from': str(lf), 'cmp_to': str(lt), 'window': f'{yf} → {yt}', 'cmp_window': f'{lf} → {lt}', 'ties': abs((vol + prc + cst + new + lost + other) - (gm1 - gm0)) <= 1.0} def margin_trend(n=13, t=None, team_id=None): """Monthly revenue, GM$ and GM% over the last n months (trend + seasonality).""" t = t or P.today() out = [] for label, mf, mt in P.month_starts(n, t): w = _window(mf, mt, team_id) out.append({'month': label, 'revenue': w['revenue'], 'gm_dollars': w['margin'], 'gm_pct': w['gm_pct']}) return out def low_margin_skus(t=None, limit=25, min_rev=2000.0, team_id=None): """Margin-dilutive SKUs: meaningful YTD revenue but low realized GM%. Zero-cost (uncosted) lines are flagged separately — their 100% GM is a data gap, not real.""" t = t or P.today() yf, yt = P.ytd(t) rows = [] for r in _per_product(yf, yt, team_id): if r['revenue'] < min_rev: continue gm_pct = (r['margin'] / r['revenue'] * 100) if r['revenue'] else 0.0 rows.append({'name': r['name'], 'revenue': r['revenue'], 'gm_dollars': r['margin'], 'gm_pct': gm_pct, 'uncosted': abs(r['margin'] - r['revenue']) < 0.01}) real = [r for r in rows if not r['uncosted']] return sorted(real, key=lambda x: x['gm_pct'])[:limit] def uncosted_share(t=None, team_id=None): """Share of YTD revenue running through zero-cost lines (margin overstated).""" t = t or P.today() yf, yt = P.ytd(t) dom = O.sale_line_domain(date_from=str(yf), date_to=str(yt), team_id=team_id) rev_all = O.sum_field('sale.order.line', dom, 'price_subtotal') rev_zero = O.sum_field('sale.order.line', dom + [('purchase_price', '=', 0)], 'price_subtotal') return {'rev_all': rev_all, 'rev_uncosted': rev_zero, 'pct': (rev_zero / rev_all * 100) if rev_all else 0.0} def cash_conversion_cycle(t=None): """CCC = DIO + DSO − DPO (days). Directional, company-wide. Throughput basis = LTM purchases (vendor bills net of refunds), the most consistent company-wide proxy for COGS available without a P&L-account reconciliation. DSO comes from the AR module (revenue-based, solid). The dominant driver — inventory days — is robust to the exact basis, which is the point. """ t = t or P.today() o = O.get_odoo() lf, lt = P.ltm(t) base = [('state', '=', 'posted'), ('invoice_date', '>=', str(lf)), ('invoice_date', '<=', str(lt))] purch = (O.sum_field('account.move', base + [('move_type', '=', 'in_invoice')], 'amount_total') - O.sum_field('account.move', base + [('move_type', '=', 'in_refund')], 'amount_total')) daily = (purch / 365.0) if purch else None ap_rows = o.search_read('account.move', [('move_type', 'in', ['in_invoice', 'in_refund']), ('state', '=', 'posted'), ('payment_state', 'in', ['not_paid', 'partial'])], ['amount_residual_signed']) ap_open = abs(sum(r.get('amount_residual_signed') or 0.0 for r in ap_rows)) inv_value = inv_mod.summary(t)['total_inv_value'] dso = ar_mod.summary(t)['dso_days'] dio = (inv_value / daily) if daily else None dpo = (ap_open / daily) if daily else None ccc = (dio + dso - dpo) if (dio is not None and dso is not None and dpo is not None) else None return { 'ltm_purchases': purch, 'inventory_value': inv_value, 'ap_open': ap_open, 'dio_days': dio, 'dso_days': dso, 'dpo_days': dpo, 'ccc_days': ccc, 'basis': 'LTM purchases (vendor bills net of refunds) as COGS/throughput proxy', } def validate(t=None, team_id=None): """Reconcile margin 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. The cross-BU brand-mirror check (#1) only makes sense consolidated, so it runs only when team_id is None.""" t = t or P.today() yf, yt = P.ytd(t) o = O.get_odoo() checks = [] total = _window(yf, yt, team_id) # 1. Σ(brand margin) == total margin (consolidated only — cross-BU) if team_id is None: brand_sum = sum(b['gm_dollars'] for b in by_brand(t)) checks.append({'check': 'Margin: Σ(brand) == total (YTD)', 'a': round(brand_sum, 2), 'b': round(total['margin'], 2), 'gap': round(brand_sum - total['margin'], 2), 'ok': abs(brand_sum - total['margin']) <= 1.0}) # 2. Σ(per-product margin) == total margin (grouping integrity) prod_sum = sum(r['margin'] for r in _per_product(yf, yt, team_id)) checks.append({'check': 'Margin: Σ(per-product) == total (YTD)', 'a': round(prod_sum, 2), 'b': round(total['margin'], 2), 'gap': round(prod_sum - total['margin'], 2), 'ok': abs(prod_sum - total['margin']) <= 1.0}) # 2b. The margin bridge decomposition sums exactly to ΔGM (volume+price+cost+new+lost+other) br = margin_bridge(t, team_id=team_id) _parts = br['volume'] + br['price'] + br['cost'] + br['new'] + br['lost'] + br['other'] checks.append({'check': 'Margin bridge: Σ(effects) == ΔGM YTD vs LY', 'a': round(_parts, 2), 'b': round(br['delta'], 2), 'gap': round(_parts - br['delta'], 2), 'ok': abs(_parts - br['delta']) <= 1.0}) # 3. Odoo line-level margin identity: margin == price_subtotal − purchase_price×qty dom = O.sale_line_domain(date_from=str(yf), date_to=str(yt), team_id=team_id) smp = o.search_read('sale.order.line', dom, ['price_subtotal', 'purchase_price', 'product_uom_qty', 'margin'], limit=1) if smp: r = smp[0] recomputed = (r.get('price_subtotal') or 0) - (r.get('purchase_price') or 0) * (r.get('product_uom_qty') or 0) checks.append({'check': 'Margin identity: price_subtotal − cost×qty (sample line)', 'a': round(r.get('margin') or 0, 2), 'b': round(recomputed, 2), 'gap': round((r.get('margin') or 0) - recomputed, 2), 'ok': abs((r.get('margin') or 0) - recomputed) <= 0.05}) return checks