connect_reserved grows as connected accounts go negative

"""Report platform funds held in connect_reserved against negative accounts.

Read only. Paginated GETs and no writes: give this a RESTRICTED key with read
access to Balance, Balance transactions and Connected accounts. The repair is
printed, never performed, because this script holds a credential to a live
payments account.
"""
import argparse
import logging
import os
import sys
import time

import requests

logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s")
log = logging.getLogger("stripe_connect_reserve")

API = "https://api.stripe.com/v1"

DAY = 86400


def classify(entry, reserved, collected):
    """Sort one currency bucket of connect_reserved. Pure, so the states can be
    tested without a network.

    `entry` is one element of balance.connect_reserved. `reserved` is the total
    reserve_transaction activity seen in that currency over the window and
    `collected` the total connect_collection_transfer activity, both passed as
    magnitudes in minor units: the sign of a balance transaction depends on the
    direction of the movement, and only the size matters here.

    Returns (state, detail).
    """
    amount = entry.get("amount")
    if not isinstance(amount, int):
        return ("unknown", "connect_reserved entry has no numeric amount: %r" % (amount,))

    if collected:
        return ("written-off",
                "%d already moved out as connect_collection_transfer: accounts that "
                "stayed negative for 180 days were settled from your reserve, and "
                "that money is not coming back" % collected)

    if amount > 0 and reserved:
        return ("growing",
                "%d held now and %d of reserve_transaction activity in the window: "
                "accounts are still going negative faster than they earn back"
                % (amount, reserved))

    if amount > 0:
        return ("held",
                "%d held with no reserve_transaction activity in the window: the "
                "negative account behind it has stopped trading, so nothing will "
                "release this before the 180 day settlement" % amount)

    if reserved:
        return ("settled",
                "nothing held now, %d of reserve_transaction activity in the window: "
                "reserves were taken and released as accounts earned back" % reserved)

    return ("clear", "nothing reserved")


def get(session, path, headers=None, **params):
    r = session.get(API + path, params=params, headers=headers, timeout=30)
    if r.status_code == 401:
        raise SystemExit("401 from Stripe: the key is wrong, or is for the other mode")
    r.raise_for_status()
    return r.json()


def totals_by_currency(session, btype, since, cap):
    """Sum balance transactions of one type per currency, as magnitudes."""
    totals = {}
    seen = 0
    params = {"type": btype, "limit": 100, "created[gte]": since}
    while True:
        page = get(session, "/balance_transactions", **params)
        data = page.get("data", [])
        for bt in data:
            cur = bt.get("currency", "?")
            totals[cur] = totals.get(cur, 0) + abs(bt.get("amount") or 0)
            seen += 1
        if not data or not page.get("has_more") or seen >= cap:
            return totals
        params["starting_after"] = data[-1]["id"]


def negative_accounts(session, cap):
    """Yield (account_id, currency, amount, liable) for accounts below zero.

    One extra GET per connected account, which is why this is behind a flag.
    """
    params = {"limit": 100}
    seen = 0
    while True:
        page = get(session, "/accounts", **params)
        data = page.get("data", [])
        for acct in data:
            seen += 1
            aid = acct.get("id", "")
            controller = acct.get("controller") or {}
            losses = (controller.get("losses") or {}).get("payments")
            bal = get(session, "/balance", headers={"Stripe-Account": aid})
            for entry in bal.get("available", []):
                if (entry.get("amount") or 0) < 0:
                    yield (aid, entry.get("currency", "?"), entry["amount"],
                           losses == "application")
            if seen >= cap:
                return
        if not data or not page.get("has_more"):
            return
        params["starting_after"] = data[-1]["id"]


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--days", type=int, default=90,
                    help="window for reserve and collection activity")
    ap.add_argument("--accounts", action="store_true",
                    help="also find the negative connected accounts (one GET each)")
    ap.add_argument("--max-accounts", type=int, default=1000,
                    help="stop the per-account pass after this many accounts")
    args = ap.parse_args()

    key = os.environ.get("STRIPE_API_KEY")
    if not key:
        log.error("set STRIPE_API_KEY (use a restricted, read-only key)")
        return 2

    s = requests.Session()
    s.headers.update({"Authorization": "Bearer " + key})

    since = int(time.time()) - args.days * DAY
    balance = get(s, "/balance")
    reserved_now = balance.get("connect_reserved") or []
    if not reserved_now:
        log.info("no connect_reserved on this balance: either not a platform, or "
                 "no account has ever gone negative")

    reserved = totals_by_currency(s, "reserve_transaction", since, 5000)
    collected = totals_by_currency(s, "connect_collection_transfer", since, 5000)

    # A currency can have activity in the window and nothing held now, so union
    # the buckets rather than iterating connect_reserved alone.
    currencies = ({e.get("currency", "?") for e in reserved_now}
                  | set(reserved) | set(collected))

    counts = {}
    for cur in sorted(currencies):
        entry = next((e for e in reserved_now if e.get("currency") == cur),
                     {"currency": cur, "amount": 0})
        state, detail = classify(entry, reserved.get(cur, 0), collected.get(cur, 0))
        counts[state] = counts.get(state, 0) + 1
        line = "%-4s %-11s %s" % (cur, state, detail)
        (log.info if state in ("clear", "settled") else log.warning)(line)

    log.info("%d currency bucket(s): %d growing, %d held, %d written off",
             len(currencies), counts.get("growing", 0), counts.get("held", 0),
             counts.get("written-off", 0))

    if args.accounts:
        found = 0
        for aid, cur, amount, liable in negative_accounts(s, args.max_accounts):
            found += 1
            log.warning("%s  %s %d  liable=%s", aid, cur, amount, liable)
        if not found:
            log.info("no connected account is currently below zero")

    if counts.get("growing") or counts.get("held") or counts.get("written-off"):
        log.warning("  repair, per negative account, in order of preference:")
        log.warning("  1. transfer the shortfall to the account to release the "
                    "reserve now: %s/transfers with destination=acct_x", API)
        log.warning("  2. make future shortfalls come out of the account's own "
                    "bank: %s/balance_settings with Stripe-Account, "
                    "payments[debit_negative_balances]=true", API)
        log.warning("  3. for accounts that will never trade again, reject them "
                    "so nothing more accrues: %s/accounts/{id}/reject", API)
        return 1
    return 0


if __name__ == "__main__":
    sys.exit(main())

/**
 * Report platform funds held in connect_reserved against negative accounts.
 *
 * Read only. Paginated GETs and no writes: give this a RESTRICTED key with read
 * access to Balance, Balance transactions and Connected accounts. The repair is
 * printed, never performed.
 */
const API = 'https://api.stripe.com/v1';
const DAY = 86400;

/**
 * Sort one currency bucket of connect_reserved. Pure, so the states can be
 * tested without a network. `reserved` and `collected` are magnitudes in minor
 * units, because the sign of a balance transaction depends on the direction of
 * the movement and only the size matters here. Returns [state, detail].
 */
export function classify(entry, reserved, collected) {
  const amount = entry.amount;
  if (typeof amount !== 'number' || !Number.isFinite(amount)) {
    return ['unknown',
      `connect_reserved entry has no numeric amount: ${JSON.stringify(amount)}`];
  }

  if (collected) {
    return ['written-off',
      `${collected} already moved out as connect_collection_transfer: accounts ` +
      'that stayed negative for 180 days were settled from your reserve, and ' +
      'that money is not coming back'];
  }

  if (amount > 0 && reserved) {
    return ['growing',
      `${amount} held now and ${reserved} of reserve_transaction activity in the ` +
      'window: accounts are still going negative faster than they earn back'];
  }

  if (amount > 0) {
    return ['held',
      `${amount} held with no reserve_transaction activity in the window: the ` +
      'negative account behind it has stopped trading, so nothing will release ' +
      'this before the 180 day settlement'];
  }

  if (reserved) {
    return ['settled',
      `nothing held now, ${reserved} of reserve_transaction activity in the ` +
      'window: reserves were taken and released as accounts earned back'];
  }

  return ['clear', 'nothing reserved'];
}

async function get(key, path, params = {}, headers = {}) {
  const url = new URL(API + path);
  for (const [k, v] of Object.entries(params)) url.searchParams.set(k, v);
  const res = await fetch(url, {
    headers: { Authorization: `Bearer ${key}`, ...headers },
  });
  if (res.status === 401) {
    throw new Error('401 from Stripe: the key is wrong, or is for the other mode');
  }
  if (!res.ok) throw new Error(`${res.status} from ${url.pathname}`);
  return res.json();
}

export async function totalsByCurrency(key, type, since, cap = 5000) {
  const totals = new Map();
  let seen = 0;
  const params = { type, limit: 100, 'created[gte]': since };
  for (;;) {
    const page = await get(key, '/balance_transactions', params);
    const data = page.data ?? [];
    for (const bt of data) {
      const cur = bt.currency ?? '?';
      totals.set(cur, (totals.get(cur) ?? 0) + Math.abs(bt.amount ?? 0));
      seen += 1;
    }
    if (data.length === 0 || !page.has_more || seen >= cap) return totals;
    params.starting_after = data[data.length - 1].id;
  }
}

async function main() {
  const key = process.env.STRIPE_API_KEY;
  if (!key) {
    console.error('set STRIPE_API_KEY (use a restricted, read-only key)');
    process.exitCode = 2;
    return;
  }

  const days = Number(process.env.WINDOW_DAYS ?? 90);
  const since = Math.floor(Date.now() / 1000) - days * DAY;

  const balance = await get(key, '/balance');
  const reservedNow = balance.connect_reserved ?? [];
  if (reservedNow.length === 0) {
    console.log('no connect_reserved on this balance: either not a platform, or ' +
                'no account has ever gone negative');
  }

  const reserved = await totalsByCurrency(key, 'reserve_transaction', since);
  const collected = await totalsByCurrency(key, 'connect_collection_transfer', since);

  const currencies = new Set([
    ...reservedNow.map((e) => e.currency ?? '?'),
    ...reserved.keys(), ...collected.keys(),
  ]);

  const counts = {};
  for (const cur of [...currencies].sort()) {
    const entry = reservedNow.find((e) => e.currency === cur)
      ?? { currency: cur, amount: 0 };
    const [state, detail] = classify(entry, reserved.get(cur) ?? 0,
                                     collected.get(cur) ?? 0);
    counts[state] = (counts[state] ?? 0) + 1;
    const line = `${cur.padEnd(4)} ${state.padEnd(11)} ${detail}`;
    if (state === 'clear' || state === 'settled') console.log(line);
    else console.warn(line);
  }

  console.log(`${currencies.size} currency bucket(s): ${counts.growing ?? 0} ` +
              `growing, ${counts.held ?? 0} held, ` +
              `${counts['written-off'] ?? 0} written off`);

  if (counts.growing || counts.held || counts['written-off']) {
    console.warn('  repair, per negative account, in order of preference:');
    console.warn(`  1. transfer the shortfall to the account to release the ` +
                 `reserve now: ${API}/transfers with destination=acct_x`);
    console.warn(`  2. make future shortfalls come out of the account's own bank: ` +
                 `${API}/balance_settings with Stripe-Account, ` +
                 `payments[debit_negative_balances]=true`);
    console.warn(`  3. for accounts that will never trade again, reject them so ` +
                 `nothing more accrues: ${API}/accounts/{id}/reject`);
    process.exitCode = 1;
  }
}

// Only run when invoked directly. The test file imports this module, and without
// the guard main() would run there too, fail on the missing key, and set a
// non-zero exit code that fails the whole test file even as every test passes.
if (import.meta.url === `file://${process.argv[1]}`) {
  main().catch((err) => { console.error(err.message); process.exitCode = 2; });
}

from stripe_connect_reserve import classify


def test_a_collection_transfer_outranks_everything_else():
    # The 180 day settlement already happened. Whatever is held now is a
    # secondary concern next to money that has permanently left.
    state, detail = classify({"currency": "usd", "amount": 4000}, 4000, 25000)
    assert state == "written-off"
    assert "connect_collection_transfer" in detail


def test_reserve_with_recent_activity_is_growing():
    state, detail = classify({"currency": "usd", "amount": 12000}, 9000, 0)
    assert state == "growing"
    assert "12000" in detail


def test_reserve_with_no_activity_is_the_dead_account_case():
    # Same positive amount, no movement behind it: nothing will release this
    # on its own, because the account that caused it has stopped trading.
    state, detail = classify({"currency": "usd", "amount": 12000}, 0, 0)
    assert state == "held"
    assert "180 day" in detail


def test_activity_with_nothing_held_is_normal_operation():
    state, _ = classify({"currency": "eur", "amount": 0}, 30000, 0)
    assert state == "settled"


def test_missing_amount_is_not_silently_clear():
    state, _ = classify({"currency": "usd"}, 0, 0)
    assert state == "unknown"

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { classify } from './stripe-connect-reserve.mjs';

test('a collection transfer outranks everything else', () => {
  const [state, detail] = classify({ currency: 'usd', amount: 4000 }, 4000, 25000);
  assert.equal(state, 'written-off');
  assert.match(detail, /connect_collection_transfer/);
});

test('reserve with recent activity is growing', () => {
  const [state, detail] = classify({ currency: 'usd', amount: 12000 }, 9000, 0);
  assert.equal(state, 'growing');
  assert.match(detail, /12000/);
});

test('reserve with no activity is the dead account case', () => {
  const [state, detail] = classify({ currency: 'usd', amount: 12000 }, 0, 0);
  assert.equal(state, 'held');
  assert.match(detail, /180 day/);
});

test('activity with nothing held is normal operation', () => {
  assert.equal(classify({ currency: 'eur', amount: 0 }, 30000, 0)[0], 'settled');
});

test('missing amount is not silently clear', () => {
  assert.equal(classify({ currency: 'usd' }, 0, 0)[0], 'unknown');
});