platform/modules/expenses.py

"""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