analytics.py

"""
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": "<str>", "significance": "high|medium|low", "evidence": "<exact stat cited>"}}
  ],
  "marketInsights": {{
    "creditEconomy": "<what receivables/payables & credit-sale share reveal about the trust economy>",
    "inequality": "<interpret the Gini / top-decile share across traders>",
    "priceDiscovery": "<interpret cross-trader price dispersion; where is the law of one price violated>",
    "cashAndLiquidity": "<cash on hand vs credit locked up; resilience read>",
    "productConcentration": "<what the HHI and top items say about basket diversity>"
  }},
  "researchHighlights": [
    {{"observation": "<a finding worth a working-paper footnote>", "metric": "<supporting number>", "caveat": "<sampling/coverage limitation>"}}
  ],
  "modelQuality": {{
    "overallAccuracyProxy": "<confirmRate read>",
    "modalityGap": "<voice vs photo vs typed accuracy — implication for low-literacy traders>",
    "dataMaturity": "<dataset size & growth read>"
  }},
  "recommendations": [
    {{"action": "<str>", "priority": "immediate|short_term|long_term", "rationale": "<str>"}}
  ],
  "riskFlags": [
    {{"risk": "<str>", "severity": "critical|moderate|low", "metric": "<supporting number>"}}
  ]
}}
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,
    }