dispute activity is above the 0.75% excessive threshold

"""Measure the Stripe dispute rate against the card network thresholds.

Read only. Three paginated GETs and no writes: give this a RESTRICTED key with
read access to Disputes, Charges and Early Fraud Warnings. There is no API
toggle to repair this, so nothing here could write even if it wanted to; the
remediation is printed instead.
"""
import argparse
import calendar
import datetime as dt
import logging
import os
import sys

import requests

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

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

# Visa VAMP flags a merchant as non-compliant here.
WARN_RATE = 0.005
# Industry practice, and Stripe's own guidance, treats this as excessive.
EXCESSIVE_RATE = 0.0075
# Visa VAMP excessive, and Mastercard ECM once its own count floor is met.
PROGRAM_RATE = 0.015

# Below these counts the programmes do not apply, whatever the ratio says.
VAMP_FLOOR = 5      # disputes plus early fraud warnings
ECM_FLOOR = 100     # disputes alone


def rates(disputes, efws, charges):
    """Return (dispute_rate, vamp_rate). Pure.

    Both are None when there were no successful charges. A month with disputes
    and no charges is a data problem, not an infinite rate, and printing
    infinity would bury the real message.
    """
    if not charges:
        return (None, None)
    return (disputes / charges, (disputes + efws) / charges)


def assess(disputes, efws, charges,
           warn=WARN_RATE, excessive=EXCESSIVE_RATE, program=PROGRAM_RATE):
    """Classify a month of dispute activity. Pure. Returns (state, detail).

    The ratio and the count floors are separate tests on purpose: a high ratio
    on a handful of events is a signal worth reading, but it is not a breach,
    and reporting it as one wastes the credibility of the check.
    """
    dispute_rate, vamp_rate = rates(disputes, efws, charges)
    if dispute_rate is None:
        return ("no_volume",
                "no successful captured charges in the window; there is nothing "
                "to divide by")

    events = disputes + efws
    pct = "disputes %.3f%%, with EFW %.3f%%" % (dispute_rate * 100, vamp_rate * 100)

    if vamp_rate < warn:
        return ("clear", "%s, both under the %.2f%% VAMP line" % (pct, warn * 100))

    if events < VAMP_FLOOR and disputes < ECM_FLOOR:
        return ("below_floor",
                "%s, but only %d countable event(s). VAMP needs %d and ECM needs "
                "%d disputes, so no programme applies yet."
                % (pct, events, VAMP_FLOOR, ECM_FLOOR))

    if dispute_rate >= program or vamp_rate >= program:
        return ("program",
                "%s. At or above %.2f%% this is VAMP excessive territory, and "
                "Mastercard ECM once you pass %d disputes in a month."
                % (pct, program * 100, ECM_FLOOR))
    if dispute_rate >= excessive or vamp_rate >= excessive:
        return ("excessive",
                "%s. Above the %.2f%% the industry treats as excessive; expect "
                "monitoring before it reaches %.2f%%."
                % (pct, excessive * 100, program * 100))
    return ("watch",
            "%s. At or above the %.2f%% VAMP non-compliant line and below the "
            "%.2f%% excessive line: the month to act in."
            % (pct, warn * 100, excessive * 100))


def month_bounds(month):
    """Unix bounds for a YYYY-MM string, or the previous calendar month."""
    if month:
        year, mon = (int(p) for p in month.split("-"))
    else:
        today = dt.date.today()
        year, mon = (today.year, today.month - 1) if today.month > 1 else (today.year - 1, 12)
    start = dt.datetime(year, mon, 1, tzinfo=dt.timezone.utc)
    last = calendar.monthrange(year, mon)[1]
    end = dt.datetime(year, mon, last, 23, 59, 59, tzinfo=dt.timezone.utc)
    return int(start.timestamp()), int(end.timestamp()) + 1, "%04d-%02d" % (year, mon)


def get(session, path, params=None):
    r = session.get(API + path, params=params or {}, 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 count(session, path, start, end, cap, keep=None):
    """Count objects in a created range. Returns (count, truncated).

    `truncated` is the whole point of the return being a tuple: a denominator
    that stopped early makes the ratio read high, and a confidently wrong ratio
    is worse than refusing to print one.
    """
    total = 0
    scanned = 0
    params = {"limit": 100, "created[gte]": start, "created[lt]": end}
    while True:
        page = get(session, path, params)
        data = page.get("data", [])
        for obj in data:
            scanned += 1
            if keep is None or keep(obj):
                total += 1
        if not data or not page.get("has_more"):
            return total, False
        if scanned >= cap:
            return total, True
        params["starting_after"] = data[-1]["id"]


def succeeded_and_captured(charge):
    return charge.get("status") == "succeeded" and charge.get("captured") is True


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--month", help="YYYY-MM; defaults to the previous calendar month")
    ap.add_argument("--max-charges", type=int, default=50000,
                    help="refuse to report a ratio if the denominator needs more than this")
    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})

    start, end, label = month_bounds(args.month)
    log.info("counting %s", label)

    disputes, _ = count(s, "/disputes", start, end, cap=100000)
    efws, _ = count(s, "/radar/early_fraud_warnings", start, end, cap=100000)
    charges, truncated = count(s, "/charges", start, end,
                               cap=args.max_charges, keep=succeeded_and_captured)

    if truncated:
        log.error("stopped after %d charges, so the denominator is short and the "
                  "ratio would read high. Raise --max-charges or narrow the window.",
                  args.max_charges)
        return 2

    state, detail = assess(disputes, efws, charges)
    line = "%-12s %d dispute(s), %d EFW(s), %d successful charge(s): %s" % (
        state, disputes, efws, charges, detail)
    if state in ("clear", "no_volume"):
        log.info(line)
        return 0

    log.warning(line)
    log.warning("  there is no API repair for a ratio: reduce the numerator.")
    log.warning("  block highest risk in Radar, request 3DS on elevated risk,")
    log.warning("  refund actionable early fraud warnings before they escalate,")
    log.warning("  set a recognisable statement descriptor, and make cancelling self-serve.")
    log.warning("  remediation guidance: https://docs.stripe.com/disputes/monitoring-programs")
    return 1 if state != "below_floor" else 0


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

/**
 * Measure the Stripe dispute rate against the card network thresholds.
 *
 * Read only. Three paginated GETs and no writes: give this a RESTRICTED key
 * with read access to Disputes, Charges and Early Fraud Warnings. There is no
 * API toggle to repair this; the remediation is printed instead.
 */
const API = 'https://api.stripe.com/v1';

export const WARN_RATE = 0.005;      // Visa VAMP non-compliant
export const EXCESSIVE_RATE = 0.0075; // industry and Stripe guidance
export const PROGRAM_RATE = 0.015;   // VAMP excessive, Mastercard ECM

export const VAMP_FLOOR = 5;   // disputes plus early fraud warnings
export const ECM_FLOOR = 100;  // disputes alone

/**
 * Return [disputeRate, vampRate], or [null, null] with no charges. Pure.
 */
export function rates(disputes, efws, charges) {
  if (!charges) return [null, null];
  return [disputes / charges, (disputes + efws) / charges];
}

/**
 * Classify a month of dispute activity. Pure. Returns [state, detail].
 */
export function assess(disputes, efws, charges,
                       warn = WARN_RATE, excessive = EXCESSIVE_RATE,
                       program = PROGRAM_RATE) {
  const [disputeRate, vampRate] = rates(disputes, efws, charges);
  if (disputeRate === null) {
    return ['no_volume',
      'no successful captured charges in the window; there is nothing to divide by'];
  }

  const events = disputes + efws;
  const pct = `disputes ${(disputeRate * 100).toFixed(3)}%, ` +
              `with EFW ${(vampRate * 100).toFixed(3)}%`;

  if (vampRate < warn) {
    return ['clear', `${pct}, both under the ${(warn * 100).toFixed(2)}% VAMP line`];
  }
  if (events < VAMP_FLOOR && disputes < ECM_FLOOR) {
    return ['below_floor',
      `${pct}, but only ${events} countable event(s). VAMP needs ${VAMP_FLOOR} ` +
      `and ECM needs ${ECM_FLOOR} disputes, so no programme applies yet.`];
  }
  if (disputeRate >= program || vampRate >= program) {
    return ['program',
      `${pct}. At or above ${(program * 100).toFixed(2)}% this is VAMP excessive ` +
      `territory, and Mastercard ECM once you pass ${ECM_FLOOR} disputes in a month.`];
  }
  if (disputeRate >= excessive || vampRate >= excessive) {
    return ['excessive',
      `${pct}. Above the ${(excessive * 100).toFixed(2)}% the industry treats as ` +
      `excessive; expect monitoring before it reaches ${(program * 100).toFixed(2)}%.`];
  }
  return ['watch',
    `${pct}. At or above the ${(warn * 100).toFixed(2)}% VAMP non-compliant line ` +
    `and below the ${(excessive * 100).toFixed(2)}% excessive line: the month to act in.`];
}

/** Unix bounds for a YYYY-MM string, or the previous calendar month. */
export function monthBounds(month) {
  let year;
  let mon;
  if (month) {
    [year, mon] = month.split('-').map(Number);
  } else {
    const now = new Date();
    year = now.getUTCMonth() > 0 ? now.getUTCFullYear() : now.getUTCFullYear() - 1;
    mon = now.getUTCMonth() > 0 ? now.getUTCMonth() : 12;
  }
  const start = Date.UTC(year, mon - 1, 1) / 1000;
  const end = Date.UTC(mon === 12 ? year + 1 : year, mon === 12 ? 0 : mon, 1) / 1000;
  return [start, end, `${String(year).padStart(4, '0')}-${String(mon).padStart(2, '0')}`];
}

async function get(key, path, params = {}) {
  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}` } });
  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();
}

async function countObjects(key, path, start, end, cap, keep) {
  let total = 0;
  let scanned = 0;
  const params = { limit: 100, 'created[gte]': start, 'created[lt]': end };
  for (;;) {
    const page = await get(key, path, params);
    const data = page.data ?? [];
    for (const obj of data) {
      scanned += 1;
      if (!keep || keep(obj)) total += 1;
    }
    if (data.length === 0 || !page.has_more) return [total, false];
    if (scanned >= cap) return [total, true];
    params.starting_after = data[data.length - 1].id;
  }
}

export function succeededAndCaptured(charge) {
  return charge.status === 'succeeded' && charge.captured === true;
}

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 monthArg = process.argv.includes('--month')
    ? process.argv[process.argv.indexOf('--month') + 1] : undefined;
  const maxCharges = process.argv.includes('--max-charges')
    ? Number(process.argv[process.argv.indexOf('--max-charges') + 1]) : 50000;

  const [start, end, label] = monthBounds(monthArg);
  console.log(`counting ${label}`);

  const [disputes] = await countObjects(key, '/disputes', start, end, 100000);
  const [efws] = await countObjects(key, '/radar/early_fraud_warnings', start, end, 100000);
  const [charges, truncated] = await countObjects(
    key, '/charges', start, end, maxCharges, succeededAndCaptured);

  if (truncated) {
    console.error(`stopped after ${maxCharges} charges, so the denominator is ` +
                  'short and the ratio would read high. Raise --max-charges or ' +
                  'narrow the window.');
    process.exitCode = 2;
    return;
  }

  const [state, detail] = assess(disputes, efws, charges);
  const line = `${state.padEnd(12)} ${disputes} dispute(s), ${efws} EFW(s), ` +
               `${charges} successful charge(s): ${detail}`;
  if (state === 'clear' || state === 'no_volume') {
    console.log(line);
    return;
  }

  console.warn(line);
  console.warn('  there is no API repair for a ratio: reduce the numerator.');
  console.warn('  block highest risk in Radar, request 3DS on elevated risk,');
  console.warn('  refund actionable early fraud warnings before they escalate,');
  console.warn('  set a recognisable statement descriptor, and make cancelling self-serve.');
  console.warn('  remediation guidance: https://docs.stripe.com/disputes/monitoring-programs');
  process.exitCode = state === 'below_floor' ? 0 : 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_dispute_rate import assess, rates


def test_no_successful_charges_is_not_an_infinite_rate():
    assert rates(3, 0, 0) == (None, None)
    state, detail = assess(3, 0, 0)
    assert state == "no_volume"
    assert "divide" in detail


def test_the_half_percent_vamp_line_is_inclusive():
    # 10 / 2000 is exactly 0.5%, which is already non-compliant.
    assert assess(10, 0, 2000)[0] == "watch"
    assert assess(9, 0, 2000)[0] == "clear"


def test_early_fraud_warnings_count_toward_the_visa_ratio():
    # 0.2% on disputes alone, 0.6% once EFWs join the numerator.
    dispute_rate, vamp_rate = rates(4, 8, 2000)
    assert dispute_rate < 0.005 < vamp_rate
    state, detail = assess(4, 8, 2000)
    assert state == "watch"
    assert "EFW" in detail


def test_a_high_ratio_under_the_count_floor_is_not_a_breach():
    # 3 countable events is below VAMP's floor of 5 and ECM's of 100,
    # so 1.5% here is a signal and not a programme risk.
    state, detail = assess(2, 1, 200)
    assert state == "below_floor"
    assert "VAMP needs 5" in detail


def test_the_bands_above_the_line_are_distinct():
    assert assess(16, 0, 2000)[0] == "excessive"   # 0.8%
    assert assess(40, 0, 2000)[0] == "program"     # 2.0%
    assert assess(11, 0, 2000)[0] == "watch"       # 0.55%

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { assess, rates } from './stripe-dispute-rate.mjs';

test('no successful charges is not an infinite rate', () => {
  assert.deepEqual(rates(3, 0, 0), [null, null]);
  const [state, detail] = assess(3, 0, 0);
  assert.equal(state, 'no_volume');
  assert.match(detail, /divide/);
});

test('the half percent vamp line is inclusive', () => {
  assert.equal(assess(10, 0, 2000)[0], 'watch');
  assert.equal(assess(9, 0, 2000)[0], 'clear');
});

test('early fraud warnings count toward the visa ratio', () => {
  const [disputeRate, vampRate] = rates(4, 8, 2000);
  assert.ok(disputeRate < 0.005);
  assert.ok(vampRate > 0.005);
  const [state, detail] = assess(4, 8, 2000);
  assert.equal(state, 'watch');
  assert.match(detail, /EFW/);
});

test('a high ratio under the count floor is not a breach', () => {
  const [state, detail] = assess(2, 1, 200);
  assert.equal(state, 'below_floor');
  assert.match(detail, /VAMP needs 5/);
});

test('the bands above the line are distinct', () => {
  assert.equal(assess(16, 0, 2000)[0], 'excessive');
  assert.equal(assess(40, 0, 2000)[0], 'program');
  assert.equal(assess(11, 0, 2000)[0], 'watch');
});