Realign dashboard API to the v2 bot schema (transactions)

The endpoints read the legacy sales/expenses subcollections that the upgraded
bot no longer writes, so dashboards returned zeros. Read the v2 `transactions`
schema instead and mirror the bot's accounting.

- analytics.py: pure aggregation (per-currency sales/COGS/gross/expenses/net,
cash-on-hand, receivables/payables, stock value, top items/customers/expenses,
daily series) + model_health over distillation_examples.
- /api/user/dashboard rewritten (v2 payload + legacy aliases); add
/api/user/transactions.
- /api/admin/dashboard/stats rewritten (per-user v2 aggregation, per-currency
leaderboards, global standing); add /api/admin/model-health.
- Auth, user CRUD, org and phone-migration endpoints unchanged.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

analytics.py

@@ -0,0 +1,306 @@
+"""
+analytics.py — v2 ledger aggregation for the Smart Qx dashboards.
+
+The WhatsApp bot (smart-w) records ONE `transactions` collection per user
+(`users/{mobile}/transactions`), each doc carrying a `transaction_type`
+(sale | stock_in | expense | asset | loan | repayment | other) and a `details`
+map with an `items[]` array. These helpers reproduce the bot's own accounting
+(see smart-w/utility.py: _compute_cogs, _compute_cash_position, build_period_report)
+so the dashboards and the WhatsApp reports agree to the cent.
+
+Pure functions — they take plain dicts (Firestore `doc.to_dict()`), no Firebase here.
+"""
+
+from collections import defaultdict
+from datetime import datetime, timezone
+from typing import Any, Dict, List, Optional
+
+
+# ── Parsing helpers ──────────────────────────────────────────────────────────
+_CURRENCY_MAP = {
+    "$": "USD", "dollar": "USD", "dollars": "USD", "usd": "USD",
+    "r": "ZAR", "rand": "ZAR", "rands": "ZAR", "zar": "ZAR",
+    "zwg": "ZWG", "zwl": "ZWG", "kes": "KES", "ngn": "NGN", "ghs": "GHS",
+    "eur": "EUR", "€": "EUR", "gbp": "GBP", "£": "GBP",
+}
+
+
+def normalize_currency_code(raw_code: Any, default_code: str = "USD") -> str:
+    """Messy currency string ('$', 'rand', 'R') → ISO code ('USD', 'ZAR')."""
+    if not raw_code or not isinstance(raw_code, str):
+        return default_code
+    return _CURRENCY_MAP.get(raw_code.lower().strip(), (raw_code.upper().strip()[:3] or default_code))
+
+
+def money(value: Any) -> float:
+    """Best-effort money → float. Accepts numbers and strings like 'USD4', '$3.50', '50c'."""
+    if value is None:
+        return 0.0
+    if isinstance(value, bool):
+        return 0.0
+    if isinstance(value, (int, float)):
+        return float(value)
+    s = str(value).strip().replace(",", "")
+    if not s or s.lower() in ("none", "null", "nan", "?"):
+        return 0.0
+    import re
+    m = re.search(r"-?\d+(?:\.\d+)?", s)
+    if not m:
+        return 0.0
+    try:
+        return float(m.group(0))
+    except ValueError:
+        return 0.0
+
+
+def money_opt(value: Any) -> Optional[float]:
+    """Like money() but returns None when there is no numeric value (distinguishes 0 from absent)."""
+    if value is None:
+        return None
+    if isinstance(value, (int, float)) and not isinstance(value, bool):
+        return float(value)
+    import re
+    m = re.search(r"-?\d+(?:\.\d+)?", str(value))
+    return float(m.group(0)) if m else None
+
+
+def _to_dt(ts: Any) -> Optional[datetime]:
+    """Firestore Timestamp / datetime / ISO string → tz-aware datetime (UTC)."""
+    if ts is None:
+        return None
+    if isinstance(ts, datetime):
+        return ts if ts.tzinfo else ts.replace(tzinfo=timezone.utc)
+    if hasattr(ts, "timestamp"):  # Firestore Timestamp / DatetimeWithNanoseconds
+        try:
+            return datetime.fromtimestamp(ts.timestamp(), tz=timezone.utc)
+        except Exception:
+            return None
+    if isinstance(ts, str):
+        try:
+            return datetime.fromisoformat(ts.replace("Z", "+00:00"))
+        except Exception:
+            return None
+    return None
+
+
+def _in_range(ts: Any, start: Optional[datetime], end: Optional[datetime]) -> bool:
+    if start is None and end is None:
+        return True
+    dt = _to_dt(ts)
+    if dt is None:
+        return False
+    if start and dt < start:
+        return False
+    if end and dt > end:
+        return False
+    return True
+
+
+def _txn_type(d: Dict) -> str:
+    return (d.get("transaction_type") or d.get("type") or "other").lower()
+
+
+def _items(details: Dict) -> List[Dict]:
+    it = details.get("items")
+    return it if isinstance(it, list) else []
+
+
+def _truthy(v: Any) -> bool:
+    return str(v).strip().lower() in ("true", "1", "yes", "y", "service")
+
+
+# ── Cost basis (COGS) ────────────────────────────────────────────────────────
+def build_cost_map(stock_batches: List[Dict]) -> Dict[str, float]:
+    """item name → mean known unit cost (cost_each) from stock batches."""
+    sums: Dict[str, float] = defaultdict(float)
+    counts: Dict[str, int] = defaultdict(int)
+    for b in stock_batches or []:
+        name = str(b.get("name") or "").strip().lower()
+        c = money_opt(b.get("cost_each"))
+        if name and c is not None and c > 0:
+            sums[name] += c
+            counts[name] += 1
+    return {k: sums[k] / counts[k] for k in sums if counts[k]}
+
+
+# ── Main ledger aggregation ──────────────────────────────────────────────────
+def aggregate_ledger(txns: List[Dict], stock_batches: List[Dict], customers: List[Dict],
+                     start: Optional[datetime] = None, end: Optional[datetime] = None,
+                     default_currency: str = "USD") -> Dict[str, Any]:
+    """Return per-currency financials + insight lists. Mirrors the bot's report math.
+
+    Standing figures (cash on hand, receivables, payables, stock value) are CURRENT — the
+    date range only filters period flows (sales/expenses/cogs/net and the series/lists).
+    """
+    cur_metrics: Dict[str, Dict[str, float]] = defaultdict(
+        lambda: {"sales": 0.0, "cogs": 0.0, "expenses": 0.0, "stock_purchases": 0.0, "sales_count": 0})
+    item_rev: Dict[str, Dict[str, float]] = defaultdict(lambda: defaultdict(float))   # item → cur → revenue
+    item_qty: Dict[str, float] = defaultdict(float)
+    expense_cat: Dict[str, Dict[str, float]] = defaultdict(lambda: defaultdict(float))  # cat → cur → amount
+    daily: Dict[str, Dict[str, float]] = defaultdict(lambda: {"revenue": 0.0, "cogs": 0.0, "expenses": 0.0})
+
+    cost_map = build_cost_map(stock_batches)
+    last_cur = normalize_currency_code(default_currency, "USD")
+    cash = 0.0  # cash on hand is CURRENT (whole trail), not date-filtered
+
+    for d in txns or []:
+        ttype = _txn_type(d)
+        det = d.get("details", {}) or {}
+        cur = normalize_currency_code(det.get("currency"), last_cur)
+        last_cur = cur
+        amount = money(det.get("amount") or det.get("total"))
+        paid = money_opt(det.get("amount_paid"))
+        in_period = _in_range(d.get("created_at"), start, end)
+        day_key = (_to_dt(d.get("created_at")) or datetime.now(timezone.utc)).strftime("%Y-%m-%d")
+
+        # --- cash on hand (current, all-time) ---
+        if ttype == "sale":
+            credit = money(det.get("customer_credit"))
+            cash += max(amount - credit, 0.0)
+        elif ttype in ("stock_in", "expense", "asset"):
+            cash -= (paid if paid is not None else amount)
+        elif ttype == "loan":
+            direction = (det.get("loan_direction") or "").lower()
+            if direction == "lent":
+                cash -= amount
+            elif direction == "borrowed":
+                cash += amount
+        elif ttype == "repayment":
+            cash += money(det.get("cash_in"))
+            cash -= money(det.get("cash_out"))
+
+        if not in_period:
+            continue
+
+        # --- period flows ---
+        if ttype == "sale":
+            cur_metrics[cur]["sales"] += amount
+            cur_metrics[cur]["sales_count"] += 1
+            daily[day_key]["revenue"] += amount
+            line_cogs = 0.0
+            for it in _items(det):
+                if _truthy(it.get("is_service")):
+                    continue
+                nm = str(it.get("item") or it.get("name") or "").strip().lower()
+                qty = money(it.get("quantity"))
+                if nm and qty > 0:
+                    unit_cost = cost_map.get(nm, 0.0)
+                    line_cogs += qty * unit_cost
+                    item_qty[nm] += qty
+                # per-item revenue share for leaderboards
+                ippu = money_opt(it.get("price_per_unit") or it.get("price_each"))
+                iamt = money_opt(it.get("amount"))
+                rev = iamt if iamt is not None else (ippu * qty if (ippu is not None and qty) else None)
+                if nm and rev:
+                    item_rev[nm][cur] += rev
+            cur_metrics[cur]["cogs"] += line_cogs
+            daily[day_key]["cogs"] += line_cogs
+        elif ttype == "expense":
+            cur_metrics[cur]["expenses"] += amount
+            cat = (det.get("category") or det.get("description") or "other") or "other"
+            expense_cat[str(cat)][cur] += amount
+            daily[day_key]["expenses"] += amount
+        elif ttype == "stock_in":
+            cur_metrics[cur]["stock_purchases"] += amount
+
+    # finalize per-currency
+    by_cur: Dict[str, Dict[str, float]] = {}
+    for cur, m in cur_metrics.items():
+        gross = round(m["sales"] - m["cogs"], 2)
+        net = round(gross - m["expenses"], 2)
+        by_cur[cur] = {
+            "sales": round(m["sales"], 2),
+            "cogs": round(m["cogs"], 2),
+            "grossProfit": gross,
+            "grossMarginPct": round((gross / m["sales"] * 100.0), 1) if m["sales"] else 0.0,
+            "expenses": round(m["expenses"], 2),
+            "netProfit": net,
+            "stockPurchases": round(m["stock_purchases"], 2),
+            "salesCount": int(m["sales_count"]),
+        }
+
+    # standing (current)
+    receivables = round(sum(money(c.get("receivable", c.get("outstanding_credit"))) for c in (customers or [])), 2)
+    payables = round(sum(money(c.get("payable", c.get("outstanding_change"))) for c in (customers or [])), 2)
+    stock_value = round(sum(money(b.get("quantity_remaining")) * money(b.get("cost_each")) for b in (stock_batches or [])), 2)
+
+    # leaderboards / lists
+    top_items = sorted(
+        ({"item": nm, "quantity": round(item_qty.get(nm, 0.0), 2),
+          "revenue": round(sum(cm.values()), 2), "byCurrency": {k: round(v, 2) for k, v in cm.items()}}
+         for nm, cm in item_rev.items()),
+        key=lambda x: -x["revenue"])[:10]
+    top_customers = sorted(
+        ({"name": c.get("name", "Customer"),
+          "receivable": round(money(c.get("receivable", c.get("outstanding_credit"))), 2),
+          "totalPurchases": round(money(c.get("total_purchases")), 2),
+          "visits": int(money(c.get("visit_count")))}
+         for c in (customers or [])),
+        key=lambda x: -x["totalPurchases"])[:10]
+    top_expenses = sorted(
+        ({"category": cat, "amount": round(sum(cm.values()), 2),
+          "byCurrency": {k: round(v, 2) for k, v in cm.items()}}
+         for cat, cm in expense_cat.items()),
+        key=lambda x: -x["amount"])[:10]
+    daily_series = [
+        {"date": day, "revenue": round(v["revenue"], 2), "cogs": round(v["cogs"], 2),
+         "expenses": round(v["expenses"], 2),
+         "profit": round(v["revenue"] - v["cogs"] - v["expenses"], 2)}
+        for day, v in sorted(daily.items())]
+
+    return {
+        "byCurrency": by_cur,
+        "cashOnHand": round(cash, 2),
+        "receivables": receivables,
+        "payables": payables,
+        "stockValue": stock_value,
+        "topItems": top_items,
+        "topCustomers": top_customers,
+        "topExpenses": top_expenses,
+        "dailySeries": daily_series,
+    }
+
+
+# ── Model-Health (distillation session capture) ──────────────────────────────
+def model_health(examples: List[Dict], start: Optional[datetime] = None,
+                 end: Optional[datetime] = None, opt_out_count: int = 0) -> Dict[str, Any]:
+    """Aggregate distillation_examples: how we record & close sessions.
+
+    Confirm/Cancel verdicts are a live extraction-accuracy proxy. `archived` are the
+    anonymised reset snapshots (reason == account_reset)."""
+    by_task: Dict[str, int] = defaultdict(int)
+    by_modality: Dict[str, int] = defaultdict(int)
+    by_verdict: Dict[str, int] = defaultdict(int)
+    growth: Dict[str, int] = defaultdict(int)
+    reset_snapshots = 0
+    total = 0
+
+    for e in examples or []:
+        if not _in_range(e.get("created_at"), start, end):
+            continue
+        total += 1
+        by_task[str(e.get("task") or "unknown")] += 1
+        by_modality[str(e.get("modality") or "unknown")] += 1
+        verdict = str(e.get("verdict") or "pending")
+        by_verdict[verdict] += 1
+        if str(e.get("reason") or "") == "account_reset" or e.get("task") == "ledger_snapshot":
+            reset_snapshots += 1
+        day = (_to_dt(e.get("created_at")) or datetime.now(timezone.utc)).strftime("%Y-%m-%d")
+        growth[day] += 1
+
+    confirmed = by_verdict.get("confirmed", 0)
+    rejected = by_verdict.get("rejected", 0)
+    labelled = confirmed + rejected
+    confirm_rate = round(confirmed / labelled * 100.0, 1) if labelled else None
+
+    return {
+        "totalCaptured": total,
+        "byTask": dict(by_task),
+        "byModality": dict(by_modality),
+        "byVerdict": dict(by_verdict),
+        "confirmRate": confirm_rate,          # % of labelled sessions the user confirmed
+        "labelledCount": labelled,
+        "growthSeries": [{"date": d, "count": growth[d]} for d in sorted(growth)],
+        "optOutCount": int(opt_out_count),
+        "resetSnapshots": reset_snapshots,
+    }

main.py

@@ -7,6 +7,8 @@ from flask_cors import CORS
 import firebase_admin
 from firebase_admin import credentials, auth, firestore
 
+import analytics  # v2 ledger aggregation (reads the transactions schema)
+
 # --- Basic Configuration ---
 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
 
@@ -92,6 +94,28 @@ def normalize_currency_code(raw_code, default_code='USD'):
     clean_code = raw_code.lower().strip()
     return currency_map.get(clean_code, default_code)
 
+
+# ── v2 ledger helpers ────────────────────────────────────────────────────────
+def parse_date_range(req):
+    """Parse ?start_date&end_date (ISO) into tz-aware datetimes (or None)."""
+    def _p(s):
+        if not s:
+            return None
+        try:
+            return datetime.fromisoformat(s.replace('Z', '+00:00'))
+        except Exception:
+            return None
+    return _p(req.args.get('start_date')), _p(req.args.get('end_date'))
+
+
+def fetch_user_ledger(bot_data_id):
+    """Read a bot user's v2 ledger: (transactions, stock_batches, customers) as dicts."""
+    ref = db.collection('users').document(bot_data_id)
+    txns = [d.to_dict() for d in ref.collection('transactions').stream()]
+    batches = [d.to_dict() for d in ref.collection('stock_batches').stream()]
+    customers = [d.to_dict() for d in ref.collection('customers').stream()]
+    return txns, batches, customers
+
 # -----------------------------------------------------------------------------
 # 3. AUTHENTICATION & USER MANAGEMENT ENDPOINTS
 # -----------------------------------------------------------------------------
@@ -244,82 +268,80 @@ def get_user_dashboard():
     if not phone_number: return jsonify({'error': 'No phone number is associated with your account.'}), 404
     
     try:
-        start_date_str = request.args.get('start_date')
-        end_date_str = request.args.get('end_date')
-
-        start_date, end_date = None, None
-        if start_date_str:
-            start_date = datetime.fromisoformat(start_date_str.replace('Z', '+00:00'))
-        if end_date_str:
-            end_date = datetime.fromisoformat(end_date_str.replace('Z', '+00:00'))
-
+        start_date, end_date = parse_date_range(request)
+        default_currency = user_data.get('defaultCurrency', 'USD')
         bot_data_id = phone_number.lstrip('+')
-        bot_user_ref = db.collection('users').document(bot_data_id)
-        
-        sales_revenue_by_currency, cogs_by_currency, expenses_by_currency = {}, {}, {}
-        sales_count = 0
-
-        default_currency_code = normalize_currency_code(user_data.get('defaultCurrency'), 'USD')
-        last_seen_currency_code = default_currency_code
-
-        # Process Sales
-        sales_query = bot_user_ref.collection('sales').stream()
-        for doc in sales_query:
-            doc_data = doc.to_dict()
-            created_at = doc_data.get('createdAt')
 
-            if start_date and (not created_at or created_at < start_date):
-                continue
-            if end_date and (not created_at or created_at > end_date):
-                continue
+        txns, batches, customers = fetch_user_ledger(bot_data_id)
+        agg = analytics.aggregate_ledger(txns, batches, customers, start_date, end_date,
+                                         default_currency=default_currency)
+
+        # Legacy-compatible per-currency maps (so existing UI keeps working during the switch)
+        by_cur = agg['byCurrency']
+        legacy = {
+            'totalSalesRevenueByCurrency': {c: v['sales'] for c, v in by_cur.items()},
+            'totalCostOfGoodsSoldByCurrency': {c: v['cogs'] for c, v in by_cur.items()},
+            'grossProfitByCurrency': {c: v['grossProfit'] for c, v in by_cur.items()},
+            'totalExpensesByCurrency': {c: v['expenses'] for c, v in by_cur.items()},
+            'netProfitByCurrency': {c: v['netProfit'] for c, v in by_cur.items()},
+            'salesCount': sum(v['salesCount'] for v in by_cur.values()),
+        }
+        # New v2 insight payload
+        payload = {
+            **legacy,
+            'byCurrency': by_cur,
+            'cashOnHand': agg['cashOnHand'],
+            'receivables': agg['receivables'],
+            'payables': agg['payables'],
+            'stockValue': agg['stockValue'],
+            'topItems': agg['topItems'],
+            'topCustomers': agg['topCustomers'],
+            'dailySeries': agg['dailySeries'],
+        }
+        return jsonify(payload), 200
 
-            details = doc_data.get('details', {})
-            currency_code = normalize_currency_code(details.get('currency'), last_seen_currency_code)
-            last_seen_currency_code = currency_code
-            quantity, price, cost = int(details.get('quantity', 1)), float(details.get('price', 0)), float(details.get('cost', 0))
-            sales_revenue_by_currency[currency_code] = sales_revenue_by_currency.get(currency_code, 0) + (price * quantity)
-            cogs_by_currency[currency_code] = cogs_by_currency.get(currency_code, 0) + (cost * quantity)
-            sales_count += 1
-        
-        # Process Expenses
-        expenses_query = bot_user_ref.collection('expenses').stream()
-        for doc in expenses_query:
-            doc_data = doc.to_dict()
-            created_at = doc_data.get('createdAt')
+    except Exception as e:
+        logging.error(f"Error fetching dashboard data for user {uid} (phone: {phone_number}): {e}", exc_info=True)
+        return jsonify({'error': 'An error occurred while fetching your dashboard data.'}), 500
 
-            if start_date and (not created_at or created_at < start_date):
-                continue
-            if end_date and (not created_at or created_at > end_date):
-                continue
 
-            details = doc.to_dict().get('details', {})
-            currency_code = normalize_currency_code(details.get('currency'), last_seen_currency_code)
-            last_seen_currency_code = currency_code
-            amount = float(details.get('amount', 0))
-            expenses_by_currency[currency_code] = expenses_by_currency.get(currency_code, 0) + amount
-
-        # Calculate final totals using the clean, normalized currency codes
-        all_currencies = set(sales_revenue_by_currency.keys()) | set(cogs_by_currency.keys()) | set(expenses_by_currency.keys())
-        gross_profit_by_currency, net_profit_by_currency = {}, {}
-
-        for curr in all_currencies:
-            revenue, cogs, expenses = sales_revenue_by_currency.get(curr, 0), cogs_by_currency.get(curr, 0), expenses_by_currency.get(curr, 0)
-            gross_profit_by_currency[curr] = round(revenue - cogs, 2)
-            net_profit_by_currency[curr] = round(revenue - cogs - expenses, 2)
-
-        dashboard_data = {
-            'totalSalesRevenueByCurrency': sales_revenue_by_currency,
-            'totalCostOfGoodsSoldByCurrency': cogs_by_currency,
-            'grossProfitByCurrency': gross_profit_by_currency,
-            'totalExpensesByCurrency': expenses_by_currency,
-            'netProfitByCurrency': net_profit_by_currency,
-            'salesCount': sales_count,
-        }
-        return jsonify(dashboard_data), 200
-        
+@app.route('/api/user/transactions', methods=['GET'])
+def get_user_transactions():
+    """Recent ledger rows for the user's dashboard activity feed."""
+    uid = verify_token(request.headers.get('Authorization'))
+    if not uid:
+        return jsonify({'error': 'Invalid or expired token'}), 401
+    user_doc = db.collection('users').document(uid).get()
+    if not user_doc.exists:
+        return jsonify({'error': 'User not found'}), 404
+    user_data = user_doc.to_dict()
+    if user_data.get('phoneStatus') != 'approved' or not user_data.get('phone'):
+        return jsonify({'error': 'Your phone number is not yet approved.'}), 403
+    try:
+        limit = min(int(request.args.get('limit', 25)), 100)
+        bot_data_id = user_data['phone'].lstrip('+')
+        q = (db.collection('users').document(bot_data_id).collection('transactions')
+             .order_by('created_at', direction=firestore.Query.DESCENDING).limit(limit))
+        rows = []
+        for d in q.stream():
+            v = d.to_dict()
+            det = v.get('details', {}) or {}
+            ts = v.get('created_at')
+            rows.append({
+                'id': d.id,
+                'type': (v.get('transaction_type') or v.get('type') or 'other'),
+                'description': det.get('description') or ', '.join(
+                    f"{it.get('quantity','')} {it.get('item') or it.get('name','')}".strip()
+                    for it in (det.get('items') or []) if isinstance(it, dict)),
+                'amount': analytics.money(det.get('amount') or det.get('total')),
+                'currency': normalize_currency_code(det.get('currency'), 'USD'),
+                'party': det.get('customer') or det.get('party') or '',
+                'createdAt': ts.isoformat() if hasattr(ts, 'isoformat') else (str(ts) if ts else None),
+            })
+        return jsonify({'transactions': rows}), 200
     except Exception as e:
-        logging.error(f"Error fetching dashboard data for user {uid} (phone: {phone_number}): {e}")
-        return jsonify({'error': 'An error occurred while fetching your dashboard data.'}), 500
+        logging.error(f"Error fetching transactions for {uid}: {e}", exc_info=True)
+        return jsonify({'error': 'An error occurred while fetching your transactions.'}), 500
 
 # Replace your existing PUT /api/user/profile with this one.
 @app.route('/api/user/profile', methods=['PUT'])
@@ -812,151 +834,96 @@ def get_admin_dashboard_stats():
     """
     try:
         verify_admin_and_get_uid(request.headers.get('Authorization'))
+        start_date, end_date = parse_date_range(request)
 
-        start_date_str = request.args.get('start_date')
-        end_date_str = request.args.get('end_date')
-
-        start_date, end_date = None, None
-        if start_date_str:
-            start_date = datetime.fromisoformat(start_date_str.replace('Z', '+00:00'))
-        if end_date_str:
-            end_date = datetime.fromisoformat(end_date_str.replace('Z', '+00:00'))
-
-        # --- Initialization ---
         all_users_docs = list(db.collection('users').stream())
         all_orgs_docs = list(db.collection('organizations').stream())
-        
-        user_sales_data, global_item_revenue, global_expense_totals, phone_to_user_map = {}, {}, {}, {}
-        global_sales_rev_by_curr, global_cogs_by_curr, global_expenses_by_curr = {}, {}, {}
-
-        # --- First Pass: Process all documents to gather stats and approved phones ---
-        pending_approvals, approved_users, admin_count = 0, 0, 0
-        approved_phone_numbers = []
 
+        # User/profile stats + approved phone map (profile docs have an 'email')
+        pending_approvals = approved_users = admin_count = 0
+        phone_to_user = {}
         for doc in all_users_docs:
-            user_data = doc.to_dict()
-            # This check ensures we only process documents that are actual user profiles
-            if user_data.get('email'):
-                phone = user_data.get('phone')
-                if user_data.get('phoneStatus') == 'pending':
-                    pending_approvals += 1
-                elif user_data.get('phoneStatus') == 'approved' and phone:
-                    approved_users += 1
-                    approved_phone_numbers.append(phone)
-                    phone_to_user_map[phone] = {
-                        'displayName': user_data.get('displayName', 'N/A'), 'uid': user_data.get('uid'),
-                        'defaultCurrency': user_data.get('defaultCurrency', 'USD')
-                    }
-                if user_data.get('isAdmin', False):
-                    admin_count += 1
-        
-        user_profile_docs = [doc for doc in all_users_docs if doc.to_dict().get('email')]
-        
-        user_stats = {
-            'total': len(user_profile_docs),
-            'admins': admin_count,
-            'approvedForBot': approved_users,
-            'pendingApproval': pending_approvals
-        }
-        
-        org_stats = {'total': len(all_orgs_docs)}
-
-        # --- Second Pass: Aggregate financial data from bot documents ---
-        sales_count = 0
-        for phone in approved_phone_numbers:
+            u = doc.to_dict()
+            if not u.get('email'):
+                continue
+            if u.get('isAdmin'):
+                admin_count += 1
+            status, phone = u.get('phoneStatus'), u.get('phone')
+            if status == 'pending':
+                pending_approvals += 1
+            elif status == 'approved' and phone:
+                approved_users += 1
+                phone_to_user[phone] = {'displayName': u.get('displayName', 'N/A'), 'uid': u.get('uid'),
+                                        'defaultCurrency': u.get('defaultCurrency', 'USD')}
+        user_profile_count = sum(1 for d in all_users_docs if d.to_dict().get('email'))
+
+        # Aggregate every approved user's v2 ledger and merge globally
+        from collections import defaultdict as _dd
+        gcur = _dd(lambda: {'sales': 0.0, 'cogs': 0.0, 'expenses': 0.0, 'salesCount': 0})
+        item_rev = _dd(lambda: _dd(float))       # item -> cur -> revenue
+        exp_cat = _dd(lambda: _dd(float))         # category -> cur -> amount
+        user_rev = _dd(lambda: _dd(float))        # phone -> cur -> revenue
+        g_cash = g_recv = g_pay = g_stock = 0.0
+
+        for phone in phone_to_user:
             try:
-                bot_data_id = phone.lstrip('+')
-                bot_user_ref = db.collection('users').document(bot_data_id)
-                
-                user_sales_data[phone] = {'total_revenue_by_currency': {}, 'item_sales': {}}
-                last_seen_currency_code = normalize_currency_code(phone_to_user_map.get(phone, {}).get('defaultCurrency'), 'USD')
-
-                sales_query = bot_user_ref.collection('sales').stream()
-                for sale_doc in sales_query:
-                    sale_data = sale_doc.to_dict()
-                    created_at = sale_data.get('createdAt')
-
-                    if start_date and (not created_at or created_at < start_date):
-                        continue
-                    if end_date and (not created_at or created_at > end_date):
-                        continue
-
-                    details = sale_data.get('details', {})
-                    currency_code = normalize_currency_code(details.get('currency'), last_seen_currency_code)
-                    last_seen_currency_code = currency_code
-                    quantity, price, cost = int(details.get('quantity', 1)), float(details.get('price', 0)), float(details.get('cost', 0))
-                    sale_revenue = price * quantity
-                    item_name = details.get('item', 'Unknown Item')
-
-                    global_sales_rev_by_curr[currency_code] = global_sales_rev_by_curr.get(currency_code, 0) + sale_revenue
-                    global_cogs_by_curr[currency_code] = global_cogs_by_curr.get(currency_code, 0) + (cost * quantity)
-                    sales_count += 1
-                    
-                    user_sales_data[phone]['total_revenue_by_currency'][currency_code] = user_sales_data[phone]['total_revenue_by_currency'].get(currency_code, 0) + sale_revenue
-                    
-                    if item_name not in global_item_revenue: global_item_revenue[item_name] = {}
-                    global_item_revenue[item_name][currency_code] = global_item_revenue[item_name].get(currency_code, 0) + sale_revenue
-                
-                expenses_query = bot_user_ref.collection('expenses').stream()
-                for expense_doc in expenses_query:
-                    expense_data = expense_doc.to_dict()
-                    created_at = expense_data.get('createdAt')
-
-                    if start_date and (not created_at or created_at < start_date):
-                        continue
-                    if end_date and (not created_at or created_at > end_date):
-                        continue
-
-                    details = expense_data.get('details', {})
-                    currency_code = normalize_currency_code(details.get('currency'), last_seen_currency_code)
-                    last_seen_currency_code = currency_code
-                    amount = float(details.get('amount', 0))
-                    category = details.get('description', 'Uncategorized')
-
-                    global_expenses_by_curr[currency_code] = global_expenses_by_curr.get(currency_code, 0) + amount
-                    if category not in global_expense_totals: global_expense_totals[category] = {}
-                    global_expense_totals[category][currency_code] = global_expense_totals[category].get(currency_code, 0) + amount
+                txns, batches, customers = fetch_user_ledger(phone.lstrip('+'))
             except Exception as e:
-                logging.error(f"Admin stats: Could not process data for phone {phone}. Error: {e}")
+                logging.error(f"Admin stats: ledger read failed for {phone}: {e}")
                 continue
-        
-        # --- Post-Processing: Generate Leaderboards For Each Currency ---
-        all_currencies = set(global_sales_rev_by_curr.keys()) | set(global_expenses_by_curr.keys())
-        
-        leaderboards = {'topUsersByRevenue': {}, 'topSellingItems': {}, 'topExpenses': {}}
-
-        for currency in all_currencies:
-            users_in_curr = [(phone, data['total_revenue_by_currency'].get(currency, 0)) for phone, data in user_sales_data.items() if data['total_revenue_by_currency'].get(currency, 0) > 0]
-            sorted_users = sorted(users_in_curr, key=lambda item: item[1], reverse=True)
-            leaderboards['topUsersByRevenue'][currency] = [{'displayName': phone_to_user_map.get(phone, {}).get('displayName'), 'uid': phone_to_user_map.get(phone, {}).get('uid'), 'totalRevenue': round(revenue, 2)} for phone, revenue in sorted_users[:5]]
+            agg = analytics.aggregate_ledger(txns, batches, customers, start_date, end_date,
+                                             default_currency=phone_to_user[phone]['defaultCurrency'])
+            for cur, m in agg['byCurrency'].items():
+                gcur[cur]['sales'] += m['sales']; gcur[cur]['cogs'] += m['cogs']
+                gcur[cur]['expenses'] += m['expenses']; gcur[cur]['salesCount'] += m['salesCount']
+                user_rev[phone][cur] += m['sales']
+            for it in agg['topItems']:
+                for cur, v in it['byCurrency'].items():
+                    item_rev[it['item']][cur] += v
+            for ex in agg['topExpenses']:
+                for cur, v in ex['byCurrency'].items():
+                    exp_cat[ex['category']][cur] += v
+            g_cash += agg['cashOnHand']; g_recv += agg['receivables']
+            g_pay += agg['payables']; g_stock += agg['stockValue']
+
+        all_currencies = set(gcur)
+        system_by_cur, total_count = {}, 0
+        for cur, m in gcur.items():
+            gross = round(m['sales'] - m['cogs'], 2)
+            system_by_cur[cur] = {
+                'sales': round(m['sales'], 2), 'cogs': round(m['cogs'], 2), 'grossProfit': gross,
+                'expenses': round(m['expenses'], 2), 'netProfit': round(gross - m['expenses'], 2),
+                'salesCount': int(m['salesCount'])}
+            total_count += int(m['salesCount'])
 
-            items_in_curr = [(name, totals.get(currency, 0)) for name, totals in global_item_revenue.items() if totals.get(currency, 0) > 0]
-            sorted_items = sorted(items_in_curr, key=lambda item: item[1], reverse=True)
-            leaderboards['topSellingItems'][currency] = [{'item': name, 'totalRevenue': round(revenue, 2)} for name, revenue in sorted_items[:5]]
-            
-            expenses_in_curr = [(name, totals.get(currency, 0)) for name, totals in global_expense_totals.items() if totals.get(currency, 0) > 0]
-            sorted_expenses = sorted(expenses_in_curr, key=lambda item: item[1], reverse=True)
-            leaderboards['topExpenses'][currency] = [{'category': name, 'totalAmount': round(amount, 2)} for name, amount in sorted_expenses[:5]]
-
-        # --- Final Assembly ---
-        global_net_profit_by_curr = {}
-        for curr in all_currencies:
-            revenue = global_sales_rev_by_curr.get(curr, 0)
-            cogs = global_cogs_by_curr.get(curr, 0)
-            expenses = global_expenses_by_curr.get(curr, 0)
-            global_net_profit_by_curr[curr] = round(revenue - cogs - expenses, 2)
-
-        system_stats = {
-            'totalSalesRevenueByCurrency': {k: round(v, 2) for k, v in global_sales_rev_by_curr.items()},
-            'totalCostOfGoodsSoldByCurrency': {k: round(v, 2) for k, v in global_cogs_by_curr.items()},
-            'totalExpensesByCurrency': {k: round(v, 2) for k, v in global_expenses_by_curr.items()},
-            'totalNetProfitByCurrency': global_net_profit_by_curr, 
-            'totalSalesCount': sales_count,
-        }
+        leaderboards = {'topUsersByRevenue': {}, 'topSellingItems': {}, 'topExpenses': {}}
+        for cur in all_currencies:
+            users_c = sorted(((p, r.get(cur, 0.0)) for p, r in user_rev.items() if r.get(cur, 0) > 0),
+                             key=lambda x: -x[1])[:5]
+            leaderboards['topUsersByRevenue'][cur] = [
+                {'displayName': phone_to_user.get(p, {}).get('displayName'),
+                 'uid': phone_to_user.get(p, {}).get('uid'), 'totalRevenue': round(rev, 2)} for p, rev in users_c]
+            items_c = sorted(((n, t.get(cur, 0.0)) for n, t in item_rev.items() if t.get(cur, 0) > 0),
+                             key=lambda x: -x[1])[:5]
+            leaderboards['topSellingItems'][cur] = [{'item': n, 'totalRevenue': round(v, 2)} for n, v in items_c]
+            exp_c = sorted(((n, t.get(cur, 0.0)) for n, t in exp_cat.items() if t.get(cur, 0) > 0),
+                           key=lambda x: -x[1])[:5]
+            leaderboards['topExpenses'][cur] = [{'category': n, 'totalAmount': round(v, 2)} for n, v in exp_c]
 
         return jsonify({
-            'userStats': user_stats, 'organizationStats': org_stats, 
-            'systemStats': system_stats, 'leaderboards': leaderboards
+            'userStats': {'total': user_profile_count, 'admins': admin_count,
+                          'approvedForBot': approved_users, 'pendingApproval': pending_approvals},
+            'organizationStats': {'total': len(all_orgs_docs)},
+            'systemStats': {
+                'byCurrency': system_by_cur,
+                'totalSalesRevenueByCurrency': {c: v['sales'] for c, v in system_by_cur.items()},
+                'totalCostOfGoodsSoldByCurrency': {c: v['cogs'] for c, v in system_by_cur.items()},
+                'totalExpensesByCurrency': {c: v['expenses'] for c, v in system_by_cur.items()},
+                'totalNetProfitByCurrency': {c: v['netProfit'] for c, v in system_by_cur.items()},
+                'totalSalesCount': total_count,
+                'cashOnHand': round(g_cash, 2), 'receivables': round(g_recv, 2),
+                'payables': round(g_pay, 2), 'stockValue': round(g_stock, 2)},
+            'leaderboards': leaderboards,
         }), 200
 
     except PermissionError as e:
@@ -964,6 +931,25 @@ def get_admin_dashboard_stats():
     except Exception as e:
         logging.error(f"Admin failed to fetch dashboard stats: {e}", exc_info=True)
         return jsonify({'error': 'An internal error occurred while fetching stats'}), 500
+
+
+@app.route('/api/admin/model-health', methods=['GET'])
+def get_model_health():
+    """Scientific view of how we record & close sessions: the distillation dataset."""
+    try:
+        verify_admin_and_get_uid(request.headers.get('Authorization'))
+        start_date, end_date = parse_date_range(request)
+        examples = [d.to_dict() for d in db.collection('distillation_examples').limit(20000).stream()]
+        opt_out = sum(1 for d in db.collection('users').stream()
+                      if d.to_dict().get('distill_consent') is False)
+        return jsonify(analytics.model_health(examples, start_date, end_date, opt_out)), 200
+    except PermissionError as e:
+        return jsonify({'error': str(e)}), 403
+    except Exception as e:
+        logging.error(f"Admin failed to fetch model health: {e}", exc_info=True)
+        return jsonify({'error': 'An internal error occurred while fetching model health'}), 500
+
+
 # -----------------------------------------------------------------------------
 # 8. SERVER EXECUTION
 # -----------------------------------------------------------------------------