"""modules/expenses.py — Expenses: the operating-expense analytics dashboard (the CFO cost-control surface). The company's operating cost BASE, read straight from the general ledger: every posted line to an expense-type account (account_type in {'expense','expense_depreciation'}) over time. COGS ('expense_direct_cost') is the product side and is EXCLUDED — this is opex, not cost of sales. Because it reads the GL directly (not just vendor bills) it also captures payroll journal entries, depreciation and bank/merchant fees that never pass through account.move in_invoice. This is the MONITOR / DIAGNOSE complement to the Spend & Payables WORKFLOW (which recovers vendor-bill waste). It answers one executive question: *what is our operating cost base, how is it trending against revenue, what grew, and where did a category spike?* It NAMES the categories to look at; it never performs an action. 1. TREND + LEVERAGE — monthly opex and opex-as-%-of-revenue (the operating-leverage picture: are costs growing faster than the top line?). Revenue denominator = income-type GL accounts (all channels), so the ratio is internally consistent and independently reconcilable. 2. YoY BRIDGE — prior-LTM opex → per-category movers → current-LTM opex (what grew / fell). 3. COST SPIKES — a Wheeler XmR (individuals) control chart per category flags the months a category ran above its own upper control limit: a diagnostic watch list, worst first. 4. FIXED VS VARIABLE — categories split by month-to-month coefficient of variation + coverage: fixed/recurring (rent, wages, subscriptions) vs variable (where in-year cuts actually land). 5. DIRECTORY — every expense account with LTM $, share, YoY, % of revenue and type, each drilling to the raw ledger lines behind it (no unverifiable aggregates). Company-level: opex is shared overhead, not cleanly BU-splittable → an HQ module that ignores the brand selector. READ-ONLY. """ import os import core.odoo as O import core.periods as P EXP_TYPES = ('expense', 'expense_depreciation') # opex — COGS 'expense_direct_cost' excluded INC_TYPES = ('income',) # operating revenue (the leverage denominator) XMR_CONST = 2.66 # individuals-chart control-limit constant (Wheeler): UCL = mean + 2.66·mR̄ MIN_MONTHS = 6 # minimum completed months of history before an XmR flag is trustworthy MATERIAL_LTM = 5000.0 # ignore sub-$5k/yr accounts for spike/leverage noise (still in the directory) FIXED_CV = 0.35 # monthly coefficient-of-variation below this + broad coverage = fixed/recurring FIXED_COVERAGE = 9 # months (of the completed window) with activity to count as recurring BRIDGE_TOP = 5 # named category movers in the YoY bridge (rest roll into "Other"; # kept small so the waterfall's x-axis labels stay legible) def _odoo_base(): return os.environ.get('ODOO_URL', '').rstrip('/') def _short_lbl(s, n=16): s = str(s or '') return s if len(s) <= n else s[:n - 1] + '…' # ---- data primitives: STORE-backed (OM-2 retrofit 2026-07-12) with LIVE fallback ------------- # build() reads the tenant DuckDB store (kept minutes-fresh by the app's auto-sync) — the page's # ~30 monthly read_groups become millisecond SQL. validate() STAYS on live Odoo, so the existing # to-the-cent contract doubles as the retrofit's standing parity proof. Any store problem # (missing file, cold table) falls back to the live reads — the page never breaks on a cold store. USE_STORE = True def _store_con(): import harness.datastore as DS return DS.ro_con() # per-thread cached read-only (shared retrofit primitive) def _accounts_live(): rows = O.search_read('account.account', [('account_type', 'in', list(EXP_TYPES) + list(INC_TYPES))], ['code', 'name', 'account_type']) exp = {r['id']: {'code': r.get('code') or '', 'name': r.get('name') or ''} for r in rows if r.get('account_type') in EXP_TYPES} inc_ids = [r['id'] for r in rows if r.get('account_type') in INC_TYPES] return exp, inc_ids def _accounts_store(): types = list(EXP_TYPES) + list(INC_TYPES) rows = _store_con().execute( 'SELECT id, code, name, account_type FROM account_account ' 'WHERE account_type IN (' + ','.join('?' * len(types)) + ')', types).fetchall() exp = {r[0]: {'code': r[1] or '', 'name': r[2] or ''} for r in rows if r[3] in EXP_TYPES} inc_ids = [r[0] for r in rows if r[3] in INC_TYPES] return exp, inc_ids def _accounts(): """{id: {code, name}} for expense-type accounts, and the list of income-account ids.""" if USE_STORE: try: return _accounts_store() except Exception: pass return _accounts_live() def _acct_balances_live(acc_ids, date_from, date_to): g = O.read_group('account.move.line', [('account_id', 'in', list(acc_ids)), ('parent_state', '=', 'posted'), ('date', '>=', date_from), ('date', '<=', date_to)], ['balance:sum'], ['account_id'], lazy=False) return {O.m2o_id(r['account_id']): (r.get('balance') or 0.0) for r in g if r.get('account_id')} def _acct_balances_store(acc_ids, date_from, date_to): ids = list(acc_ids) rows = _store_con().execute( 'SELECT l.account_id, sum(l.balance) FROM account_move_line l ' 'JOIN account_move m ON m.id = l.move_id ' "WHERE m.state = 'posted' AND l.account_id IN (" + ','.join('?' * len(ids)) + ') ' 'AND CAST(l.date AS TIMESTAMP) >= ? AND CAST(l.date AS TIMESTAMP) <= ? ' 'GROUP BY 1', ids + [str(date_from), str(date_to)]).fetchall() return {r[0]: (r[1] or 0.0) for r in rows} def _acct_balances(acc_ids, date_from, date_to): """{account_id: Σ balance} over posted lines in the window (debit-positive: expense accounts read positive, income accounts read negative).""" if not acc_ids: return {} if USE_STORE: try: return _acct_balances_store(acc_ids, date_from, date_to) except Exception: pass return _acct_balances_live(acc_ids, date_from, date_to) def _income_total(inc_ids, date_from, date_to): """Operating revenue over the window (income accounts carry credit balances → negate).""" if not inc_ids: return 0.0 bal = _acct_balances(inc_ids, date_from, date_to) return -sum(bal.values()) def _xmr(series): """Wheeler individuals-chart limits over a numeric series. Returns (mean, ucl, lcl, mrbar) or None when there isn't enough history.""" xs = list(series) if len(xs) < MIN_MONTHS: return None mean = sum(xs) / len(xs) mrs = [abs(xs[i] - xs[i - 1]) for i in range(1, len(xs))] mrbar = (sum(mrs) / len(mrs)) if mrs else 0.0 return mean, mean + XMR_CONST * mrbar, mean - XMR_CONST * mrbar, mrbar def _cv_coverage(series): """(coefficient of variation, months-with-activity) over a completed monthly series.""" xs = list(series) n = len(xs) cov = sum(1 for x in xs if abs(x) > 0.005) if n < 2: return None, cov mean = sum(xs) / n if mean <= 0: return None, cov sd = (sum((x - mean) ** 2 for x in xs) / n) ** 0.5 return sd / mean, cov def build(t=None): """The full Expenses bundle: scorecard inputs, monthly trend + leverage, YoY bridge steps, the category directory, the XmR spike watch list and the fixed/variable split.""" t = t or P.today() lf, lt = P.ltm(t) pf, pt = P.prior_ltm(t) exp, inc_ids = _accounts() exp_ids = list(exp.keys()) # per-account LTM & prior-LTM opex, + LTM/prior revenue — four independent aggregates, parallel ltm_by, prior_by, rev_ltm, rev_prior, inc_ltm_by = O.parallel([ lambda: _acct_balances(exp_ids, lf, lt), lambda: _acct_balances(exp_ids, pf, pt), lambda: _income_total(inc_ids, lf, lt), lambda: _income_total(inc_ids, pf, pt), lambda: _acct_balances(inc_ids, lf, lt), ]) opex_ltm = sum(ltm_by.values()) opex_prior = sum(prior_by.values()) # monthly matrix (24 calendar months, ascending; last row = current partial month) months = P.month_starts(24, t) mres = O.parallel([ (lambda mf=mf, mt=mt: (_acct_balances(exp_ids, mf, mt), _income_total(inc_ids, mf, mt))) for (_lbl, mf, mt) in months]) cur_ym = f'{t.year}-{t.month:02d}' monthly, month_acct = [], {aid: [] for aid in exp_ids} for (lbl, _mf, _mt), (opx, rev) in zip(months, mres): total = sum(opx.values()) monthly.append({'month': lbl, 'opex': total, 'revenue': rev, 'ratio': (total / rev * 100) if rev else None, 'partial': lbl == cur_ym, 'closing': False}) for aid in exp_ids: month_acct[aid].append(opx.get(aid, 0.0)) # 3-month moving average on the opex column series (for a quiet trend line) op_series = [m['opex'] for m in monthly] for i, m in enumerate(monthly): w = op_series[max(0, i - 2):i + 1] m['ma3'] = sum(w) / len(w) complete_idx = [i for i, (lbl, _mf, _mt) in enumerate(months) if lbl != cur_ym] # Month-end close lags: the newest completed calendar month is often still being booked # (vendor bills / payroll JEs / marketplace fee settlements not yet posted), which shows up as # a near-empty month. Detect the trailing contiguous under-booked months (< CLOSE_FRAC of the # completed-month median) and anchor month-level analysis on the last FULLY-BOOKED month, so a # −90% "drop" that is really an open period never reads as a cost win. LTM/YoY stay rolling # (they tie to the GL) and are captioned as-booked. CLOSE_FRAC = 0.5 comp_all = [monthly[i]['opex'] for i in complete_idx] _srt = sorted(comp_all) med = _srt[len(_srt) // 2] if _srt else 0.0 closing_idx = set() for i in reversed(complete_idx): if med > 0 and monthly[i]['opex'] < CLOSE_FRAC * med: closing_idx.add(i) else: break for i in closing_idx: monthly[i]['closing'] = True closed_idx = [i for i in complete_idx if i not in closing_idx] last_c = closed_idx[-1] if closed_idx else (complete_idx[-1] if complete_idx else None) last_month = months[last_c][0] if last_c is not None else cur_ym # ---- category directory + fixed/variable classification ---- rows, flags = [], [] for aid in exp_ids: ltm_v = ltm_by.get(aid, 0.0) pr_v = prior_by.get(aid, 0.0) if abs(ltm_v) < 0.005 and abs(pr_v) < 0.005: continue # never active in either window comp = [month_acct[aid][i] for i in closed_idx] cv, cov = _cv_coverage(comp) ftype = ('Fixed/recurring' if (cov >= FIXED_COVERAGE and cv is not None and cv < FIXED_CV) else 'Variable') rows.append({'aid': aid, 'code': exp[aid]['code'], 'account': exp[aid]['name'], 'ltm': ltm_v, 'prior': pr_v, 'yoy_d': ltm_v - pr_v, 'yoy_pct': P.yoy_pct(ltm_v, pr_v), 'pct_opex': (ltm_v / opex_ltm * 100) if opex_ltm else 0.0, 'pct_rev': (ltm_v / rev_ltm * 100) if rev_ltm else 0.0, 'type': ftype}) # ---- XmR spike flag (latest completed month above the account's own UCL) ---- stat = _xmr(comp) if stat and last_c is not None: mean, ucl, _lcl, mrbar = stat latest = month_acct[aid][last_c] if latest > ucl and mrbar > 0 and ltm_v >= MATERIAL_LTM: flags.append({'aid': aid, 'code': exp[aid]['code'], 'account': exp[aid]['name'], 'month': last_month, 'value': latest, 'mean': mean, 'ucl': ucl, 'exceed': latest - ucl}) rows.sort(key=lambda r: -r['ltm']) flags.sort(key=lambda x: -x['exceed']) fixed_ltm = sum(r['ltm'] for r in rows if r['type'] == 'Fixed/recurring') var_ltm = sum(r['ltm'] for r in rows if r['type'] == 'Variable') # ---- income (revenue) breakdown — the denominator's verify list ---- inc_names = {} for r in O.search_read('account.account', [('id', 'in', inc_ids)], ['name']): inc_names[r['id']] = r.get('name') or '' inc_rows = [{'account': inc_names.get(aid, ''), 'ltm': -bal} for aid, bal in inc_ltm_by.items()] inc_rows.sort(key=lambda r: -r['ltm']) # ---- YoY bridge: prior LTM → top movers → other → current LTM ---- movers = sorted([r for r in rows if r.get('yoy_d')], key=lambda r: -abs(r['yoy_d'])) top = movers[:BRIDGE_TOP] steps = [{'label': 'Prior LTM', 'amount': round(opex_prior, 0), 'kind': 'total'}] used, seen = 0.0, set() for r in top: d = r['yoy_d'] used += d lbl = _short_lbl(r['account']) # the waterfall x-axis is categorical — if lbl in seen: # two similarly-named accounts must not collide lbl = f"{_short_lbl(r['account'], 11)} {r['code']}" seen.add(lbl) steps.append({'label': lbl, 'amount': round(d, 0), 'kind': 'down' if d > 0 else 'up'}) # cost up = red(bad); down = green(good) other = (opex_ltm - opex_prior) - used if abs(other) > 1: steps.append({'label': 'Other', 'amount': round(other, 0), 'kind': 'down' if other > 0 else 'up'}) steps.append({'label': 'Current LTM', 'amount': round(opex_ltm, 0), 'kind': 'total'}) # ---- scorecard scalars ---- opex_g = P.yoy_pct(opex_ltm, opex_prior) rev_g = P.yoy_pct(rev_ltm, rev_prior) opex_ratio = (opex_ltm / rev_ltm * 100) if rev_ltm else None prior_ratio = (opex_prior / rev_prior * 100) if rev_prior else None ratio_delta = (opex_ratio - prior_ratio) if (opex_ratio is not None and prior_ratio is not None) else None comp_totals = [monthly[i]['opex'] for i in closed_idx] latest_full = comp_totals[-1] if comp_totals else 0.0 trail = comp_totals[-13:-1] # the 12 closed months before the latest trail_avg = (sum(trail) / len(trail)) if trail else 0.0 latest_vs_avg = P.yoy_pct(latest_full, trail_avg) top5_share = (sum(r['ltm'] for r in rows[:5]) / opex_ltm * 100) if opex_ltm else 0.0 top_cat = rows[0] if rows else None grower = max(rows, key=lambda r: (r.get('yoy_d') or -1e18)) if rows else None n_up_fast = sum(1 for r in rows if r.get('yoy_pct') is not None and rev_g is not None and r['yoy_pct'] > rev_g and r['yoy_d'] > 0 and r['ltm'] > MATERIAL_LTM) return { 'window': (lf, lt), 'last_month': last_month, 'n_closing': len(closing_idx), 'closing_months': [months[i][0] for i in sorted(closing_idx)], 'opex_ltm': opex_ltm, 'opex_prior': opex_prior, 'opex_g': opex_g, 'rev_ltm': rev_ltm, 'rev_prior': rev_prior, 'rev_g': rev_g, 'opex_ratio': opex_ratio, 'prior_ratio': prior_ratio, 'ratio_delta': ratio_delta, 'latest_full': latest_full, 'trail_avg': trail_avg, 'latest_vs_avg': latest_vs_avg, 'top5_share': top5_share, 'top_cat': top_cat, 'grower': grower, 'n_up_fast': n_up_fast, 'n_accounts': len(rows), 'monthly': monthly, 'steps': steps, 'rows': rows, 'flags': flags, 'inc_rows': inc_rows, 'fixed_ltm': fixed_ltm, 'var_ltm': var_ltm, } def account_detail(aid, t=None): """Raw-GL drill for one expense account: 15-month balance trend with its own control band, LTM vs prior-LTM totals, the recent posted ledger lines behind it, and an Odoo deep-link. Ties to the directory's numbers (same GL basis) — not the management-basis 'account' drawer.""" t = t or P.today() acc = O.search_read('account.account', [('id', '=', aid)], ['code', 'name', 'account_type']) if not acc: return None acc = acc[0] months = P.month_starts(15, t) cur_ym = f'{t.year}-{t.month:02d}' def _m(mf, mt): return O.sum_field('account.move.line', [('account_id', '=', aid), ('parent_state', '=', 'posted'), ('date', '>=', mf), ('date', '<=', mt)], 'balance') lf, lt = P.ltm(t) pf, pt = P.prior_ltm(t) vals = O.parallel( [(lambda mf=mf, mt=mt: _m(mf, mt)) for (_lbl, mf, mt) in months] + [lambda: _m(lf, lt), lambda: _m(pf, pt)]) mvals, ltm_v, prior_v = vals[:-2], vals[-2], vals[-1] stat = _xmr([v for (lbl, _mf, _mt), v in zip(months, mvals) if lbl != cur_ym]) mean = ucl = None if stat: mean, ucl = stat[0], stat[1] trend = [{'month': lbl, 'amount': v, 'partial': lbl == cur_ym, 'over': bool(ucl is not None and v > ucl and lbl != cur_ym)} for (lbl, _mf, _mt), v in zip(months, mvals)] base = _odoo_base() jl = O.search_read('account.move.line', [('account_id', '=', aid), ('parent_state', '=', 'posted')], ['date', 'move_id', 'partner_id', 'name', 'balance'], order='date desc', limit=40) lines = [{'date': l.get('date') or '', 'entry': O.m2o_name(l.get('move_id')), 'partner': O.m2o_name(l.get('partner_id')) or '', 'label': l.get('name') or '', 'amount': round(l.get('balance') or 0.0, 2), 'link': (f"{base}/web#id={O.m2o_id(l['move_id'])}&model=account.move&view_type=form" if base and l.get('move_id') else None)} for l in jl] return {'aid': aid, 'code': acc.get('code') or '', 'name': acc.get('name') or '', 'ltm': ltm_v, 'prior': prior_v, 'yoy_pct': P.yoy_pct(ltm_v, prior_v), 'avg_month': (ltm_v / 12.0), 'mean': mean, 'ucl': ucl, 'trend': trend, 'lines': lines, 'odoo_link': (f"{base}/web#id={aid}&model=account.account&view_type=form" if base else None)} def validate(t=None, team_id=None, pre=None): """(1) client opex LTM ties an independent server Σ balance over the same expense accounts; (2) client revenue LTM ties the independent server income Σ; (3) the directory rows re-add to the opex headline (nothing dropped). Company-level — team_id ignored (opex is consolidated).""" t = t or P.today() b = pre or build(t) lf, lt = b['window'] exp, inc_ids = _accounts() exp_ids = list(exp.keys()) srv_opex = O.sum_field('account.move.line', [('account_id', 'in', exp_ids), ('parent_state', '=', 'posted'), ('date', '>=', lf), ('date', '<=', lt)], 'balance') srv_rev = _income_total(inc_ids, lf, lt) dir_sum = sum(r['ltm'] for r in b['rows']) checks = [ {'check': 'operating expense LTM — client Σ == server Σ balance', 'a': round(b['opex_ltm'], 2), 'b': round(srv_opex, 2), 'gap': round(b['opex_ltm'] - srv_opex, 2), 'ok': abs(b['opex_ltm'] - srv_opex) <= max(1.0, abs(srv_opex) * 0.001)}, {'check': 'operating revenue LTM — client Σ == server income Σ', 'a': round(b['rev_ltm'], 2), 'b': round(srv_rev, 2), 'gap': round(b['rev_ltm'] - srv_rev, 2), 'ok': abs(b['rev_ltm'] - srv_rev) <= max(1.0, abs(srv_rev) * 0.001)}, {'check': 'category directory Σ == opex headline (nothing dropped)', 'a': round(dir_sum, 2), 'b': round(b['opex_ltm'], 2), 'gap': round(dir_sum - b['opex_ltm'], 2), 'ok': abs(dir_sum - b['opex_ltm']) < 1.0}, ] return checks