""" 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. """ import json 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") # ── Selling prices (canonical registry, smart-w ADR 0018) ──────────────────── def _sorted_batches(stock_batches: List[Dict]) -> List[Dict]: """Newest-first, active before depleted — mirrors the bot's read order so a stale price on an old empty batch never wins.""" return sorted( stock_batches or [], key=lambda b: (money(b.get("quantity_remaining")) > 0, str(b.get("stocked_at") or "")), reverse=True, ) def build_price_view(item_prices: List[Dict], stock_batches: List[Dict]) -> Dict[str, Dict]: """item name → current selling-price record. The bot keeps ONE authoritative selling price per item in users/{mobile}/item_prices (smart-w ADR 0018): base price, plus sale_price / discount_pct while a promotion runs. The registry wins; the newest ACTIVE stock batch's price_each is only a legacy fallback for items priced before the registry existed. This must stay in lockstep with the bot's _lookup_item_price, or the dashboard shows a different price than the bot charges (the July-2026 USD2.56-vs-USD4.00 mismatch). """ view: Dict[str, Dict] = {} for e in item_prices or []: nm = str(e.get("name") or "").strip().lower() if not nm: continue base = money_opt(e.get("price")) sale = money_opt(e.get("sale_price")) if e.get("on_sale") else None current = sale if (sale is not None and sale > 0) else base if current is None or current <= 0: continue view[nm] = { "price": round(current, 2), "basePrice": round(base, 2) if (base is not None and base > 0) else None, "onSale": bool(e.get("on_sale")), "discountPct": e.get("discount_pct"), "source": "registry", } for b in _sorted_batches(stock_batches): nm = str(b.get("name") or "").strip().lower() if not nm or nm in view: continue p = None if b.get("on_sale"): p = money_opt(b.get("sale_price_each")) if p is None or p <= 0: p = money_opt(b.get("price_each")) if p is None or p <= 0: continue view[nm] = { "price": round(p, 2), "basePrice": round(p, 2), "onSale": bool(b.get("on_sale")), "discountPct": b.get("discount_pct"), "source": "batch", } return view # ── 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", item_prices: Optional[List[Dict]] = None) -> 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) # Current selling prices (canonical registry first — smart-w ADR 0018) and the # stock's value AT those prices. stockValue stays cost-based (accounting); # stockRetailValue is what the shelf would fetch at today's bot prices. price_view = build_price_view(item_prices, stock_batches) on_hand: Dict[str, float] = defaultdict(float) for b in stock_batches or []: nm = str(b.get("name") or "").strip().lower() if nm: on_hand[nm] += money(b.get("quantity_remaining")) stock_retail_value = round(sum( qty * price_view[nm]["price"] for nm, qty in on_hand.items() if qty > 0 and nm in price_view), 2) pricing = sorted( ({"item": nm, "onHand": round(on_hand.get(nm, 0.0), 2), "price": v["price"], "basePrice": v.get("basePrice"), "onSale": v.get("onSale", False), "discountPct": v.get("discountPct")} for nm, v in price_view.items()), key=lambda x: -(x["onHand"] * x["price"]))[:20] # 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, "stockRetailValue": stock_retail_value, "pricing": pricing, "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 # Extraction accuracy split by modality — the scientifically interesting cut for # low-literacy informal traders (voice/photo vs typed). Needs per-example # verdict×modality, computed here in a second light pass. modality_labelled: Dict[str, Dict[str, int]] = defaultdict(lambda: {"confirmed": 0, "rejected": 0}) for e in examples or []: if not _in_range(e.get("created_at"), start, end): continue v = str(e.get("verdict") or "pending") if v in ("confirmed", "rejected"): modality_labelled[str(e.get("modality") or "unknown")][v] += 1 accuracy_by_modality = { m: round(c["confirmed"] / (c["confirmed"] + c["rejected"]) * 100.0, 1) for m, c in modality_labelled.items() if (c["confirmed"] + c["rejected"]) > 0 } return { "totalCaptured": total, "byTask": dict(by_task), "byModality": dict(by_modality), "byVerdict": dict(by_verdict), "confirmRate": confirm_rate, # % of labelled sessions the user confirmed "accuracyByModality": accuracy_by_modality, "labelledCount": labelled, "growthSeries": [{"date": d, "count": growth[d]} for d in sorted(growth)], "optOutCount": int(opt_out_count), "resetSnapshots": reset_snapshots, } # ============================================================================= # Informal-market science — pure metrics for the comprehensive admin layer. # # Qx-SmartLedger is, in effect, an instrument recording the accounting of many # informal-sector micro-enterprises. The functions below turn that ledger into # the aggregate signals a development economist / financial-inclusion researcher # would ask for: trader inequality, the credit/trust economy, price discovery, # product concentration, and (via model_health) the accuracy of AI extraction # across modalities. All pure — no Firebase, no Gemini here (ADR 0015). # ============================================================================= def gini_coefficient(values: List[float]) -> Optional[float]: """Gini of a non-negative distribution (0 = perfect equality, 1 = maximal concentration). The standard measure of trader revenue inequality in a market. Returns None for empty / all-zero inputs.""" xs = sorted(float(v) for v in values if v is not None and float(v) >= 0) n = len(xs) if n == 0: return None total = sum(xs) if total <= 0: return None cum = 0.0 for i, x in enumerate(xs, start=1): cum += i * x # G = (2*Σ i*x_i) / (n*Σ x_i) − (n+1)/n return round((2.0 * cum) / (n * total) - (n + 1.0) / n, 3) def hhi_concentration(shares: Dict[str, float]) -> Optional[float]: """Herfindahl–Hirschman Index of market/product concentration on a value map. Normalised to 0–1 (sum of squared shares). >0.25 ≈ highly concentrated.""" vals = [float(v) for v in shares.values() if v and float(v) > 0] total = sum(vals) if total <= 0: return None return round(sum((v / total) ** 2 for v in vals), 3) def top_share(values: List[float], top_fraction: float = 0.10) -> Optional[float]: """Share of the total held by the top `top_fraction` of holders (e.g. the top-10%-of-traders revenue share) — a plain-language inequality companion to Gini.""" xs = sorted((float(v) for v in values if v is not None and float(v) >= 0), reverse=True) total = sum(xs) if not xs or total <= 0: return None k = max(1, int(round(len(xs) * top_fraction))) return round(sum(xs[:k]) / total * 100.0, 1) def median(values: List[float]) -> float: xs = sorted(float(v) for v in values if v is not None) n = len(xs) if n == 0: return 0.0 mid = n // 2 return round(xs[mid] if n % 2 else (xs[mid - 1] + xs[mid]) / 2.0, 2) def price_dispersion(item_price_samples: Dict[str, List[float]], min_traders: int = 2, top_n: int = 15) -> Dict[str, Any]: """Cross-trader price dispersion — the classic "law of one price" test for a market. For each item priced by ≥ min_traders traders, compute the coefficient of variation (std/mean). High dispersion signals weak price discovery / information asymmetry, a core informal-market research question. """ per_item = [] for name, prices in (item_price_samples or {}).items(): ps = [float(p) for p in prices if p is not None and float(p) > 0] if len(ps) < min_traders: continue n = len(ps) mean = sum(ps) / n if mean <= 0: continue var = sum((p - mean) ** 2 for p in ps) / n cv = (var ** 0.5) / mean per_item.append({ "item": name, "traders": n, "meanPrice": round(mean, 2), "minPrice": round(min(ps), 2), "maxPrice": round(max(ps), 2), "cv": round(cv, 3), "spreadPct": round((max(ps) - min(ps)) / mean * 100.0, 1), }) per_item.sort(key=lambda x: -x["cv"]) cvs = [d["cv"] for d in per_item] return { "itemsCompared": len(per_item), "avgDispersionCV": round(sum(cvs) / len(cvs), 3) if cvs else None, "mostDispersed": per_item[:top_n], } def scan_transactions(txns: List[Dict], start: Optional[datetime] = None, end: Optional[datetime] = None) -> Dict[str, Any]: """One pass over a trader's raw transactions for the platform-science signals that aggregate_ledger doesn't surface: type mix, daily volume, last activity, and the share of sales made on credit (the informal trust economy).""" by_type: Dict[str, int] = defaultdict(int) daily: Dict[str, int] = defaultdict(int) last_activity = None sale_count = 0 credit_sale_count = 0 credit_sales_value = 0.0 for d in txns or []: ttype = (d.get("transaction_type") or d.get("type") or "other").lower() dt = _to_dt(d.get("created_at")) if dt is not None and (last_activity is None or dt > last_activity): last_activity = dt if not _in_range(d.get("created_at"), start, end): continue by_type[ttype] += 1 if dt is not None: daily[dt.strftime("%Y-%m-%d")] += 1 if ttype == "sale": sale_count += 1 det = d.get("details", {}) or {} credit = money(det.get("customer_credit")) if credit > 0: credit_sale_count += 1 credit_sales_value += credit return { "byType": dict(by_type), "daily": dict(daily), "lastActivity": last_activity.isoformat() if last_activity else None, "saleCount": sale_count, "creditSaleCount": credit_sale_count, "creditSalesValue": round(credit_sales_value, 2), "total": sum(by_type.values()), } # ── AI report over the platform stats ───────────────────────────────────────── _REPORT_FOCUS = { "business": ( "Focus on ADOPTION and PLATFORM HEALTH for a WhatsApp bookkeeping tool serving " "informal traders: sign-up→approval→first-transaction→active funnel, active-trader " "retention, transactions per active trader, channel mix (voice/photo/text), and " "concrete growth levers for reaching more informal micro-enterprises."), "market": ( "Focus on the ECONOMICS OF THE INFORMAL MARKET this ledger observes, with the rigour " "a development-economics / financial-inclusion researcher expects: the credit & trust " "economy (receivables vs payables, share of sales on credit, working capital locked in " "credit), trader revenue inequality (Gini, top-decile share), price discovery and the " "law of one price (cross-trader price dispersion), product concentration (HHI), cash " "liquidity, and what these say about market efficiency and trader resilience. Cite the " "exact statistics."), "model": ( "Focus on AI EXTRACTION QUALITY as a scientific instrument-calibration question: the " "Confirm/Cancel verdict rate as an accuracy proxy, accuracy BY MODALITY (voice vs photo " "vs typed — decisive for low-literacy traders), dataset growth, task mix, and consent/" "opt-out. Recommend where to improve the model and what to measure next."), "full": ( "Cover ALL of: platform adoption & retention; the informal-market economy (credit/trust, " "trader inequality, price dispersion, product concentration, cash liquidity) with a " "research-grade lens; and AI extraction quality by modality. Treat the ledger as a " "scientific instrument observing informal micro-enterprises and cite exact statistics."), } def build_admin_report_prompt(stats: Dict[str, Any], report_type: str, now_iso: str) -> str: """Prompt for the admin AI report. Domain-framed for informal-market science.""" focus = _REPORT_FOCUS.get(report_type, _REPORT_FOCUS["full"]) return f"""You are a senior data scientist and development economist analysing Qx-SmartLedger, a WhatsApp bookkeeping assistant used by informal-sector traders in Southern & East Africa. The platform statistics below are aggregated from the traders' own live ledgers (sales, stock, expenses, credit given/taken) plus the AI extraction-quality capture. Treat this as an observational instrument on informal micro-enterprises. {focus} Be quantitative and cite the exact numbers you rely on. Where a figure is null or the sample is tiny, say so honestly rather than inventing a trend. Currency amounts are per-currency; do not sum across currencies. PLATFORM STATISTICS (JSON): {json.dumps(stats, indent=2, default=str)} Return ONLY valid JSON with EXACTLY this structure: {{ "reportType": "{report_type}", "generatedAt": "{now_iso}", "executiveSummary": "<4-6 sentence overview a program director could act on>", "keyFindings": [ {{"finding": "", "significance": "high|medium|low", "evidence": ""}} ], "marketInsights": {{ "creditEconomy": "", "inequality": "", "priceDiscovery": "", "cashAndLiquidity": "", "productConcentration": "" }}, "researchHighlights": [ {{"observation": "", "metric": "", "caveat": ""}} ], "modelQuality": {{ "overallAccuracyProxy": "", "modalityGap": "", "dataMaturity": "" }}, "recommendations": [ {{"action": "", "priority": "immediate|short_term|long_term", "rationale": ""}} ], "riskFlags": [ {{"risk": "", "severity": "critical|moderate|low", "metric": ""}} ] }} Rules: keyFindings 5-8 items; researchHighlights 3-5 items; recommendations 4-6 items; riskFlags 2-4 items. No markdown, no commentary outside the JSON.""" def fallback_admin_report(report_type: str, now_iso: str) -> Dict[str, Any]: """Deterministic skeleton returned when the AI call fails, so the endpoint never 500s on a model hiccup.""" return { "reportType": report_type, "generatedAt": now_iso, "executiveSummary": "AI narrative unavailable right now; the raw statistics are attached under rawStats.", "keyFindings": [], "marketInsights": {}, "researchHighlights": [], "modelQuality": {}, "recommendations": [], "riskFlags": [], "aiError": True, }