Liquidity operations cockpit

-- historical.shared_cash: returns the aggregated rollup that the
-- backend stores under historical_analytics.shared_cash in the snapshot.
WITH raw AS (
    SELECT id, sim_time, delta_bdt
      FROM shared.shared_cash_movement
     WHERE agent_id = :agent_id
       AND sim_time >= :cutoff
       AND sim_time <= :as_of
     ORDER BY sim_time DESC, id DESC
     LIMIT :row_cap
), balance_points AS (
    SELECT id, sim_time,
           (SELECT balance_bdt FROM shared.shared_cash_ledger
             WHERE agent_id = :agent_id)
           - COALESCE(
               sum(delta_bdt) OVER (
                   ORDER BY sim_time DESC, id DESC
                   ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING
               ), 0
           ) AS balance_after
      FROM raw
), daily AS (
    SELECT balance_after,
           row_number() OVER (
               PARTITION BY (sim_time AT TIME ZONE 'UTC')::date
               ORDER BY sim_time DESC, id DESC
           ) AS daily_rank
      FROM balance_points
)
SELECT count(*)                                    AS transaction_count,
       avg(-delta_bdt) FILTER (WHERE delta_bdt < 0) AS average_outflow,
       avg(delta_bdt)  FILTER (WHERE delta_bdt > 0) AS average_inflow,
       sum(delta_bdt)                                AS net_delta,
       extract(epoch FROM (max(sim_time) - min(sim_time))) / 60.0
                                                     AS span_minutes,
       stddev_pop(abs(delta_bdt))                    AS amount_stddev,
       avg(abs(delta_bdt))                           AS average_absolute_delta,
       (SELECT avg(balance_after)
          FROM daily WHERE daily_rank = 1)            AS average_daily_balance,
       (SELECT balance_bdt
          FROM shared.shared_cash_ledger
         WHERE agent_id = :agent_id)                 AS current_balance
  FROM raw;