radar is blocking a large share of your charge attempts

"""Report a Stripe account where Radar blocks too large a share of attempts.

Read only. One paginated GET and no writes: give this a RESTRICTED key with read
access to Charges. 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_block_rate")

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

HIGH_RATE = 0.05    # one attempt in twenty stopped before the issuer saw it
WATCH_RATE = 0.02   # worth a look before it becomes the conversion investigation
DOMINANT = 0.5      # one predicate causing at least half of the blocks
MOSTLY_NORMAL = 0.8  # and those charges scoring normal risk, not elevated


def verdict(total, blocked, adaptive=0, top_rule=None):
    """Classify one window of charge attempts. Pure, so the thresholds are testable.

    `total` is every charge attempt in the window, `blocked` the ones Radar stopped,
    `adaptive` the subset blocked as low_probability_of_authorization, and
    `top_rule` is (predicate, blocked_count, normal_risk_count) for the single
    predicate responsible for the most blocks, or None. Returns (state, detail).
    """
    if not total:
        return ("no-data", "no charge attempts in the window")
    if not blocked:
        return ("normal", "no blocked charges in %d attempt(s)" % total)

    own = max(blocked - adaptive, 0)
    if not own:
        return ("adaptive-only",
                "%d of %d attempt(s) blocked (%.1f%%), every one of them "
                "low_probability_of_authorization: that is Adaptive Acceptance "
                "skipping a decline, not a rule of yours"
                % (blocked, total, 100.0 * blocked / total))

    pct = 100.0 * own / total
    if own / float(total) >= HIGH_RATE:
        if top_rule:
            predicate, count, normal = top_rule
            if (count >= DOMINANT * own
                    and count and normal >= MOSTLY_NORMAL * count):
                return ("overblocking-rule",
                        "%d of %d attempt(s) blocked (%.1f%%), and %d of those came "
                        "from one predicate (%s) on charges Radar scored normal risk"
                        % (own, total, pct, count, predicate))
        return ("elevated",
                "%d of %d attempt(s) blocked by rules or risk (%.1f%%), spread "
                "across predicates: check the risk threshold as well as the rules"
                % (own, total, pct))
    if own / float(total) >= WATCH_RATE:
        return ("watch",
                "%d of %d attempt(s) blocked by rules or risk (%.1f%%). Track it as "
                "a series; a step change dates the rule edit." % (own, total, pct))
    return ("normal",
            "%d of %d attempt(s) blocked by rules or risk (%.1f%%)"
            % (own, total, pct))


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 scan(session, since, until, cap):
    """Tally blocks across one window.

    outcome.rule is expanded in the request, because unexpanded it is an id and the
    predicate is the only part of it worth reading.
    """
    total = blocked = adaptive = 0
    per_rule = {}
    params = {"created[gte]": since, "created[lt]": until, "limit": 100,
              "expand[]": "data.outcome.rule"}
    while True:
        page = get(session, "/charges", params)
        data = page.get("data", [])
        for charge in data:
            total += 1
            outcome = charge.get("outcome") or {}
            if outcome.get("type") != "blocked":
                continue
            blocked += 1
            reason = outcome.get("reason")
            if reason == "low_probability_of_authorization":
                adaptive += 1
                continue
            rule = outcome.get("rule")
            if isinstance(rule, dict):
                predicate = rule.get("predicate") or rule.get("id") or "<no predicate>"
            elif rule:
                predicate = str(rule)
            else:
                predicate = reason or "<no rule>"
            count, normal = per_rule.get(predicate, (0, 0))
            per_rule[predicate] = (
                count + 1,
                normal + (1 if outcome.get("risk_level") == "normal" else 0))
        if not data or not page.get("has_more") or total >= cap:
            break
        params["starting_after"] = data[-1]["id"]
    return total, blocked, adaptive, per_rule


def worst(per_rule):
    """The predicate with the most blocks, as (predicate, count, normal_risk)."""
    if not per_rule:
        return None
    predicate, (count, normal) = max(per_rule.items(), key=lambda kv: kv[1][0])
    return (predicate, count, normal)


def main():
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--days", type=int, default=30,
                    help="window to measure, in days (keep it fixed between runs)")
    ap.add_argument("--max-charges", type=int, default=5000,
                    help="stop paginating after this many charges per window")
    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})

    now = int(time.time())
    span = args.days * 86400
    total, blocked, adaptive, per_rule = scan(s, now - span, now, args.max_charges)
    state, detail = verdict(total, blocked, adaptive, worst(per_rule))

    log.info("%-17s %s", state, detail)
    if adaptive:
        log.info("  %d adaptive block(s) excluded (low_probability_of_authorization)",
                 adaptive)
    for predicate, (count, normal) in sorted(per_rule.items(),
                                             key=lambda kv: -kv[1][0])[:5]:
        log.info("  %4d blocked  %4d at normal risk  %s", count, normal, predicate)

    prev_total, prev_blocked, prev_adaptive, _ = scan(
        s, now - 2 * span, now - span, args.max_charges)
    if prev_total:
        log.info("  previous window: %.1f%% blocked by rules or risk",
                 100.0 * max(prev_blocked - prev_adaptive, 0) / prev_total)

    if state in ("normal", "no-data", "adaptive-only"):
        return 0
    log.warning("  repair: narrow the predicate in Dashboard > Radar > Rules rather "
                "than deleting the rule, e.g. add: and :risk_level: = 'elevated'")
    log.warning("  or convert it to a review rule while you gather data, and check "
                "its estimated false positive rate before re-enabling")
    return 1


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

/**
 * Report a Stripe account where Radar blocks too large a share of attempts.
 *
 * Read only. One paginated GET and no writes: give this a RESTRICTED key with
 * read access to Charges. The repair is printed, never performed.
 */
const API = 'https://api.stripe.com/v1';

const HIGH_RATE = 0.05;      // one attempt in twenty stopped before the issuer saw it
const WATCH_RATE = 0.02;     // worth a look before it becomes a conversion problem
const DOMINANT = 0.5;        // one predicate causing at least half of the blocks
const MOSTLY_NORMAL = 0.8;   // and those charges scoring normal risk, not elevated

/**
 * Classify one window of charge attempts. Pure, so the thresholds are testable.
 * `topRule` is [predicate, blockedCount, normalRiskCount] or null.
 */
export function verdict(total, blocked, adaptive = 0, topRule = null) {
  if (!total) return ['no-data', 'no charge attempts in the window'];
  if (!blocked) return ['normal', `no blocked charges in ${total} attempt(s)`];

  const own = Math.max(blocked - adaptive, 0);
  if (!own) {
    return ['adaptive-only',
      `${blocked} of ${total} attempt(s) blocked ` +
      `(${(100 * blocked / total).toFixed(1)}%), every one of them ` +
      'low_probability_of_authorization: that is Adaptive Acceptance skipping a ' +
      'decline, not a rule of yours'];
  }

  const pct = (100 * own / total).toFixed(1);
  if (own / total >= HIGH_RATE) {
    if (topRule) {
      const [predicate, count, normal] = topRule;
      if (count >= DOMINANT * own && count && normal >= MOSTLY_NORMAL * count) {
        return ['overblocking-rule',
          `${own} of ${total} attempt(s) blocked (${pct}%), and ${count} of those ` +
          `came from one predicate (${predicate}) on charges Radar scored normal risk`];
      }
    }
    return ['elevated',
      `${own} of ${total} attempt(s) blocked by rules or risk (${pct}%), spread ` +
      'across predicates: check the risk threshold as well as the rules'];
  }
  if (own / total >= WATCH_RATE) {
    return ['watch',
      `${own} of ${total} attempt(s) blocked by rules or risk (${pct}%). Track it ` +
      'as a series; a step change dates the rule edit.'];
  }
  return ['normal',
    `${own} of ${total} attempt(s) blocked by rules or risk (${pct}%)`];
}

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();
}

export async function scan(key, since, until, cap = 5000) {
  let total = 0; let blocked = 0; let adaptive = 0;
  const perRule = new Map();
  const params = {
    'created[gte]': since, 'created[lt]': until, limit: 100,
    'expand[]': 'data.outcome.rule',
  };
  for (;;) {
    const page = await get(key, '/charges', params);
    const data = page.data ?? [];
    for (const charge of data) {
      total += 1;
      const outcome = charge.outcome ?? {};
      if (outcome.type !== 'blocked') continue;
      blocked += 1;
      if (outcome.reason === 'low_probability_of_authorization') { adaptive += 1; continue; }
      const rule = outcome.rule;
      const predicate = (rule && typeof rule === 'object')
        ? (rule.predicate ?? rule.id ?? '<no predicate>')
        : (rule ?? outcome.reason ?? '<no rule>');
      const [count, normal] = perRule.get(predicate) ?? [0, 0];
      perRule.set(predicate,
        [count + 1, normal + (outcome.risk_level === 'normal' ? 1 : 0)]);
    }
    if (data.length === 0 || !page.has_more || total >= cap) break;
    params.starting_after = data[data.length - 1].id;
  }
  return { total, blocked, adaptive, perRule };
}

export function worst(perRule) {
  let best = null;
  for (const [predicate, [count, normal]] of perRule) {
    if (!best || count > best[1]) best = [predicate, count, normal];
  }
  return best;
}

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.argv[2] ?? 30);
  const now = Math.floor(Date.now() / 1000);
  const span = days * 86400;
  const { total, blocked, adaptive, perRule } = await scan(key, now - span, now);
  const [state, detail] = verdict(total, blocked, adaptive, worst(perRule));

  console.log(`${state.padEnd(17)} ${detail}`);
  if (adaptive) {
    console.log(`  ${adaptive} adaptive block(s) excluded ` +
                '(low_probability_of_authorization)');
  }
  for (const [predicate, [count, normal]] of
       [...perRule].sort((a, b) => b[1][0] - a[1][0]).slice(0, 5)) {
    console.log(`  ${count} blocked  ${normal} at normal risk  ${predicate}`);
  }

  const prev = await scan(key, now - 2 * span, now - span);
  if (prev.total) {
    const prevOwn = Math.max(prev.blocked - prev.adaptive, 0);
    console.log(`  previous window: ${(100 * prevOwn / prev.total).toFixed(1)}% ` +
                'blocked by rules or risk');
  }

  if (state === 'normal' || state === 'no-data' || state === 'adaptive-only') return;
  console.warn('  repair: narrow the predicate in Dashboard > Radar > Rules rather ' +
               "than deleting the rule, e.g. add: and :risk_level: = 'elevated'");
  console.warn('  or convert it to a review rule while you gather data, and check ' +
               'its estimated false positive rate before re-enabling');
  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_block_rate import verdict


def test_an_empty_window_is_not_a_finding():
    assert verdict(0, 0)[0] == "no-data"


def test_a_low_block_rate_is_normal():
    assert verdict(1000, 4)[0] == "normal"


def test_blocks_that_are_all_adaptive_acceptance_are_not_yours():
    # 6% raw, and not one of them came from a rule. Editing rules here changes
    # nothing at all.
    state, detail = verdict(1000, 60, 60)
    assert state == "adaptive-only"
    assert "Adaptive Acceptance" in detail


def test_a_dominant_predicate_on_normal_risk_charges_names_the_rule():
    state, detail = verdict(1000, 80, 10, (":card_country: != 'US'", 60, 58))
    assert state == "overblocking-rule"
    assert ":card_country:" in detail


def test_a_high_rate_spread_across_predicates_is_still_elevated():
    state, _ = verdict(1000, 80, 0, ("amount > 20000", 20, 18))
    assert state == "elevated"


def test_a_dominant_predicate_on_risky_charges_is_the_rule_working():
    # Same share of blocks, but Radar scored those charges risky too, so the rule
    # is agreeing with the fraud signal rather than replacing it.
    state, _ = verdict(1000, 80, 0, ("card_country != 'US'", 60, 4))
    assert state == "elevated"


def test_a_middling_rate_is_worth_watching_not_paging():
    state, detail = verdict(1000, 30)
    assert state == "watch"
    assert "series" in detail

import { test } from 'node:test';
import assert from 'node:assert/strict';
import { verdict } from './stripe-block-rate.mjs';

test('an empty window is not a finding', () => {
  assert.equal(verdict(0, 0)[0], 'no-data');
});

test('a low block rate is normal', () => {
  assert.equal(verdict(1000, 4)[0], 'normal');
});

test('blocks that are all adaptive acceptance are not yours', () => {
  const [state, detail] = verdict(1000, 60, 60);
  assert.equal(state, 'adaptive-only');
  assert.match(detail, /Adaptive Acceptance/);
});

test('a dominant predicate on normal risk charges names the rule', () => {
  const [state, detail] = verdict(1000, 80, 10, [":card_country: != 'US'", 60, 58]);
  assert.equal(state, 'overblocking-rule');
  assert.match(detail, /card_country/);
});

test('a high rate spread across predicates is still elevated', () => {
  assert.equal(verdict(1000, 80, 0, ['amount > 20000', 20, 18])[0], 'elevated');
});

test('a dominant predicate on risky charges is the rule working', () => {
  assert.equal(verdict(1000, 80, 0, ["card_country != 'US'", 60, 4])[0], 'elevated');
});

test('a middling rate is worth watching not paging', () => {
  const [state, detail] = verdict(1000, 30);
  assert.equal(state, 'watch');
  assert.match(detail, /series/);
});