Skip to content

Ingest Reference

philanthropy.ingest

On-ramps from an upstream donor system to a PhilanthroPy donor-level feature table.

UniSchema: read_constituent_events loads UniSchema's JSON / NDJSON egress files; constituent_events_to_features aggregates them into the one-row-per-donor feature frame the estimators consume.

CiviCRM: read_civicrm_contributions loads a contribution export CSV; civicrm_contributions_to_features aggregates it the same way, dropping test-mode and non-Completed rows first.

Raiser's Edge: read_raisers_edge_gifts loads a Blackbaud gift export CSV; raisers_edge_gifts_to_features aggregates it, dropping the commitment rows (pledges, matching gift pledges, recurring gift templates) first so a pledged dollar is not counted both as the promise and as the payments against it.

Nonprofit Cloud: read_nonprofit_cloud_gifts loads a Salesforce Nonprofit Cloud GiftTransaction export CSV; nonprofit_cloud_gifts_to_features aggregates it, keeping only paid statuses (Paid), so all other documented statuses (Unpaid, Failed, Pending, Written Off, Fully Refunded, Cancelled) are not counted as received gifts.

NPSP: read_npsp_opportunities loads a Salesforce Nonprofit Success Pack Opportunity export CSV; npsp_opportunities_to_features aggregates it, keeping only closed/won stages (Closed Won, Awarded, Posted) so a Pledged instalment, an open pipeline row, or Closed Lost is not counted as a gift.

Bloomerang: read_bloomerang_transactions loads a Bloomerang transaction export CSV; bloomerang_transactions_to_features aggregates it, dropping the commitment rows (Pledge, Recurring Donation) first so a pledged dollar is not counted both as the promise and as the payments against it.

DonorPerfect: read_donorperfect_gifts loads a DonorPerfect gift export CSV; donorperfect_gifts_to_features aggregates it, dropping Pledge records and split-gift Main totals (record_type P / M) first so a pledged or split dollar is not counted both as the promise/total and as the payments/splits against it.

map_columns renames a user-supplied export's headers to the canonical names a bridge above expects, raising one error listing every column still missing after the rename.

activities_to_features aggregates a long, multi-source activity log (event attendance, volunteer shifts, email clicks, ...) into per-donor, per-type engagement features, generalising the pattern above to an open-ended set of activity types discovered from the data itself.

read_gifts(path_or_df, source=...) looks up the CiviCRM, Raiser's Edge, Nonprofit Cloud or NPSP reader-and-aggregator pair by name, for a caller working with more than one CRM export format.

build_leadership_snapshots builds a per-donor, per-fiscal-year training table for an upgrade model: one row per (donor, fiscal year T) for every donor whose FY T giving falls in a mid-level band, features computed only from data through the end of T, and a target reading whether FY T+1 crossed a leadership threshold. build_snapshots is the same idea for every question (upgrade, lapse, response next year, next gift amount) on one shared column set, with an optional minimum number of giving years.

activities_to_features(activities, *, as_of, donors=None, include_momentum=False)

Aggregate a long activity log into per-donor, per-type features.

Parameters:

Name Type Description Default
activities iterable of mapping, or DataFrame

Long-format rows, one per activity: contact_id, activity_date, activity_type are required; amount and hours are used when present (summed, per type, over the trailing 12 months); name is accepted but not aggregated.

 required
as_of str or datetime - like

The cutoff. Rows with activity_date after as_of are dropped before any feature is computed; every window and recency figure is measured back from this date, not from "now" or from the batch's own latest date. A tz-aware timestamp (e.g. pd.Timestamp("2024-12-31", tz="UTC")) is converted to naive UTC, the same treatment activity_date gets, so the two are always comparable.

 required
donors DataFrame, Series, or iterable

The donor population this activity log is being matched against (its index if a DataFrame, its values otherwise). Used only to report the match rate: the share of distinct contact_id values in the (cutoff) activity log that are also in donors. Does not affect which rows appear in the output; a donor absent from the activity log entirely does not get a row here regardless of donors.

 None
include_momentum bool

Also emit, per activity type, <type>_count_slope_{k}y and <type>_count_rel_slope_{k}y for k in (3, 5) (plus the _hours_ / _amount_ equivalents when those columns are present): the closed-form OLS slope, and slope relative to the window mean, of that type's annual activity over the trailing k years before as_of. NaN when fewer than 2 of those years have any row in activities at all. Off by default so existing output shapes are unchanged; see :func:philanthropy.utils._momentum.trailing_slope_features.

 False

Returns:

Name Type Description
features DataFrame

One row per donor with at least one activity at or before as_of, indexed by contact_id. For each activity type present in that (cutoff) data, four columns: <type>_count_12m, <type>_count_36m (counts in the trailing 12 / 36 months before as_of, exclusive of the boundary itself: a row exactly 12 months before as_of falls outside <type>_count_12m, matching the "trailing 12 full months" reading rather than "12 months or less ago"), <type>_days_since_last (days from the type's most recent activity to as_of; NaN for a donor with no activity of that type at all, not 0 -- 0 would read as "did it today"), and <type>_distinct (lifetime count of distinct activity dates of that type, unwindowed). Plus <type>_hours_12m / <type>_amount_12m (trailing-12-month sums) when the input carries an hours / amount column at all. A donor with no rows of a given type gets 0 in that type's count and distinct columns (but NaN in days_since_last, see above); a type with no rows anywhere in the (cutoff) data contributes no columns at all. Rows are sorted by contact_id for determinism.

Raises:

Type Description
KeyError

If contact_id, activity_date, or activity_type is absent.

Warns:

Type Description
UserWarning

If donors is given and fewer than 80% of the distinct contact_id values in the (cutoff) activity log are found in it; the usual cause is an activity export keyed on email while the CRM export it is being joined against is keyed on an internal id.

Examples:

>>> rows = [
...     {"contact_id": "1", "activity_date": "2024-03-01",
...      "activity_type": "volunteer", "hours": 2},
...     {"contact_id": "1", "activity_date": "2023-01-01",
...      "activity_type": "volunteer", "hours": 3},
...     {"contact_id": "2", "activity_date": "2024-06-01",
...      "activity_type": "event"},
... ]
>>> feats = activities_to_features(rows, as_of="2024-12-31")
>>> int(feats.loc["1", "volunteer_count_12m"])
1
>>> int(feats.loc["1", "volunteer_distinct"])
2
>>> float(feats.loc["1", "volunteer_hours_12m"])
2.0
>>> "event_count_12m" in feats.columns
True
>>> int(feats.loc["2", "event_count_12m"])
1
Source code in philanthropy/ingest/_activities.py 
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
def activities_to_features(
    activities: Union[Iterable[Mapping], pd.DataFrame],
    *,
    as_of: Union[str, pd.Timestamp],
    donors: Optional[Union[pd.DataFrame, pd.Series, Iterable]] = None,
    include_momentum: bool = False,
) -> pd.DataFrame:
    """Aggregate a long activity log into per-donor, per-type features.

    Parameters
    ----------
    activities : iterable of mapping, or DataFrame
        Long-format rows, one per activity: ``contact_id``, ``activity_date``,
        ``activity_type`` are required; ``amount`` and ``hours`` are used when
        present (summed, per type, over the trailing 12 months); ``name`` is
        accepted but not aggregated.
    as_of : str or datetime-like
        The cutoff. Rows with ``activity_date`` after ``as_of`` are dropped
        before any feature is computed; every window and recency figure is
        measured back from this date, not from "now" or from the batch's own
        latest date. A tz-aware timestamp (e.g. ``pd.Timestamp("2024-12-31",
        tz="UTC")``) is converted to naive UTC, the same treatment
        ``activity_date`` gets, so the two are always comparable.
    donors : DataFrame, Series, or iterable, optional
        The donor population this activity log is being matched against (its
        index if a DataFrame, its values otherwise). Used only to report the
        match rate: the share of distinct ``contact_id`` values in the
        (cutoff) activity log that are also in ``donors``. Does not affect
        which rows appear in the output; a donor absent from the activity log
        entirely does not get a row here regardless of ``donors``.
    include_momentum : bool, default=False
        Also emit, per activity type, ``<type>_count_slope_{k}y`` and
        ``<type>_count_rel_slope_{k}y`` for k in (3, 5) (plus the ``_hours_``
        / ``_amount_`` equivalents when those columns are present): the
        closed-form OLS slope, and slope relative to the window mean, of
        that type's annual activity over the trailing k years before
        ``as_of``. NaN when fewer than 2 of those years have any row in
        ``activities`` at all. Off by default so existing output shapes are
        unchanged; see :func:`philanthropy.utils._momentum.trailing_slope_features`.

    Returns
    -------
    features : pandas.DataFrame
        One row per donor with at least one activity at or before ``as_of``,
        indexed by ``contact_id``. For each activity type present in that
        (cutoff) data, four columns: ``<type>_count_12m``, ``<type>_count_36m``
        (counts in the trailing 12 / 36 months before ``as_of``, exclusive of
        the boundary itself: a row exactly 12 months before ``as_of`` falls
        outside ``<type>_count_12m``, matching the "trailing 12 *full*
        months" reading rather than "12 months or less ago"),
        ``<type>_days_since_last`` (days from the type's most recent activity
        to ``as_of``; ``NaN`` for a donor with no activity of that type at
        all, not 0 -- 0 would read as "did it today"), and ``<type>_distinct``
        (lifetime count of distinct activity dates of that type, unwindowed).
        Plus ``<type>_hours_12m`` / ``<type>_amount_12m`` (trailing-12-month
        sums) when the input carries an ``hours`` / ``amount`` column at all.
        A donor with no rows of a given type gets 0 in that type's count and
        distinct columns (but ``NaN`` in ``days_since_last``, see above); a
        type with no rows anywhere in the (cutoff) data contributes no
        columns at all. Rows are sorted by ``contact_id`` for determinism.

    Raises
    ------
    KeyError
        If ``contact_id``, ``activity_date``, or ``activity_type`` is absent.

    Warns
    -----
    UserWarning
        If ``donors`` is given and fewer than 80% of the distinct
        ``contact_id`` values in the (cutoff) activity log are found in it;
        the usual cause is an activity export keyed on email while the CRM
        export it is being joined against is keyed on an internal id.

    Examples
    --------
    >>> rows = [
    ...     {"contact_id": "1", "activity_date": "2024-03-01",
    ...      "activity_type": "volunteer", "hours": 2},
    ...     {"contact_id": "1", "activity_date": "2023-01-01",
    ...      "activity_type": "volunteer", "hours": 3},
    ...     {"contact_id": "2", "activity_date": "2024-06-01",
    ...      "activity_type": "event"},
    ... ]
    >>> feats = activities_to_features(rows, as_of="2024-12-31")
    >>> int(feats.loc["1", "volunteer_count_12m"])
    1
    >>> int(feats.loc["1", "volunteer_distinct"])
    2
    >>> float(feats.loc["1", "volunteer_hours_12m"])
    2.0
    >>> "event_count_12m" in feats.columns
    True
    >>> int(feats.loc["2", "event_count_12m"])
    1
    """
    df = _to_frame(activities)

    if df.empty:
        return pd.DataFrame(index=pd.Index([], name="contact_id", dtype="object"))

    missing = [col for col in _REQUIRED if col not in df.columns]
    if missing:
        raise KeyError(
            f"Activity log is missing {missing}. Every activity row needs "
            f"'contact_id', 'activity_date' and 'activity_type'; got "
            f"{sorted(df.columns)}."
        )

    df = df.copy()
    df["_contact_id"] = _normalise_contact_id(df["contact_id"]).str.strip()
    df["_ts"] = _to_datetime(df["activity_date"])
    df["_type"] = df["activity_type"].astype("string").str.strip()
    # as_of is required, so this never falls back to the batch's own latest
    # date; it only normalises as_of (naive or tz-aware) to naive UTC, the
    # same treatment activity_date already got via _to_datetime, so the two
    # sides of every comparison below share a tz-awareness.
    as_of_ts = _resolve_reference_date(as_of, df["_ts"])

    # A row we can't place in time or attribute to a donor and a type
    # contributes to nothing; drop it rather than let a NaT or blank type
    # poison a window or split into its own meaningless column.
    df = df[
        df["_contact_id"].notna()
        & (df["_contact_id"].str.len() > 0)
        & df["_ts"].notna()
        & df["_type"].notna()
        & (df["_type"].str.len() > 0)
    ]
    # The cutoff: nothing after as_of is counted anywhere below, including in
    # which types and donors even appear in the output.
    df = df[df["_ts"] <= as_of_ts]
    if df.empty:
        return pd.DataFrame(index=pd.Index([], name="contact_id", dtype="object"))

    has_amount = "amount" in df.columns
    has_hours = "hours" in df.columns
    if has_amount:
        df["_amount"] = pd.to_numeric(df["amount"], errors="coerce").fillna(0.0)
    if has_hours:
        df["_hours"] = pd.to_numeric(df["hours"], errors="coerce").fillna(0.0)

    if donors is not None:
        _warn_on_low_match_rate(df["_contact_id"], donors)

    window_12m = as_of_ts - pd.DateOffset(months=12)
    window_36m = as_of_ts - pd.DateOffset(months=36)

    donor_ids = pd.Index(sorted(df["_contact_id"].unique()), name="contact_id")
    out = pd.DataFrame(index=donor_ids)
    data_start = df["_ts"].min()

    for activity_type in sorted(df["_type"].unique()):
        type_rows = df[df["_type"] == activity_type]
        grouped = type_rows.groupby("_contact_id")
        recent_12m = type_rows[type_rows["_ts"] > window_12m].groupby("_contact_id")
        recent_36m = type_rows[type_rows["_ts"] > window_36m].groupby("_contact_id")

        prefix = f"{activity_type}_"
        out[prefix + "count_12m"] = recent_12m.size().reindex(donor_ids, fill_value=0)
        out[prefix + "count_36m"] = recent_36m.size().reindex(donor_ids, fill_value=0)
        last_activity = grouped["_ts"].max().reindex(donor_ids)
        # A donor with no rows of this type has no "last time" to measure
        # from; leave this NaN (rather than 0, which reads as "did it
        # today") -- the whole point of a per-type column is telling the two
        # cases apart.
        out[prefix + "days_since_last"] = (as_of_ts - last_activity).dt.days
        out[prefix + "distinct"] = (
            grouped["_ts"].nunique().reindex(donor_ids, fill_value=0)
        )

        if has_hours:
            hours_12m = type_rows[type_rows["_ts"] > window_12m].groupby("_contact_id")["_hours"].sum()
            out[prefix + "hours_12m"] = hours_12m.reindex(donor_ids, fill_value=0.0)
        if has_amount:
            amount_12m = type_rows[type_rows["_ts"] > window_12m].groupby("_contact_id")["_amount"].sum()
            out[prefix + "amount_12m"] = amount_12m.reindex(donor_ids, fill_value=0.0)

        if include_momentum:
            momentum_cols = [("_ts", "count", prefix + "count")]
            if has_hours:
                momentum_cols.append(("_hours", "sum", prefix + "hours"))
            if has_amount:
                momentum_cols.append(("_amount", "sum", prefix + "amount"))
            for value_col, agg, col_prefix in momentum_cols:
                momentum = trailing_slope_features(
                    type_rows, donor_ids, as_of_ts, date_col="_ts", value_col=value_col,
                    agg=agg, prefix=col_prefix, donor_col="_contact_id", data_start=data_start,
                )
                out = out.join(momentum, how="left")

    for col in out.columns:
        if col.endswith(("_count_12m", "_count_36m", "_distinct")):
            out[col] = out[col].astype("int64")
        elif col.endswith("_days_since_last"):
            # float64, not int64: a donor with no rows of this type is NaN
            # here (see above), and int64 has no way to hold that.
            out[col] = out[col].astype("float64")
        else:
            out[col] = out[col].astype("float64")

    return out

bloomerang_transactions_to_features(transactions, *, reference_date=None, exclude_entry_types=DEFAULT_EXCLUDED_ENTRY_TYPES)

Aggregate a Bloomerang transaction export into donor-level features.

Commitment rows (pledges and recurring-donation schedules) are dropped first, then the surviving payment rows are handed to :func:civicrm_contributions_to_features, which produces the donor frame. Bloomerang has no contribution-status column analogous to CiviCRM's, so no status filter is applied.

Parameters:

Name Type Description Default
transactions iterable of mapping, or DataFrame

Bloomerang transaction rows, under a report export label (Account Number, Date, Amount, Transaction Type, Fund) or the REST API field names (AccountId, Date, Amount, EntryType, FundName). contact_id, receive_date and total_amount are required after normalisation; gift_type, financial_type, email, first_name and last_name are used when present.

 required
reference_date str or datetime - like

Anchor for the recency features. If None, the latest transaction date in the batch is used, which keeps the aggregation free of "now" leakage.

 None
exclude_entry_types sequence of str or None

Entry types to drop before aggregating, matched against gift_type ignoring case, spacing and punctuation. None or an empty sequence disables the filter and sums every row, which double-counts pledged dollars and counts a recurring schedule never charged. Rows whose entry type is blank are kept either way: an unlabelled row cannot be shown to be a commitment.

 :data:`DEFAULT_EXCLUDED_ENTRY_TYPES`

Returns:

Name Type Description
features DataFrame

One row per donor, indexed by contact_id, with the same columns :func:civicrm_contributions_to_features emits. distinct_financial_types counts distinct Funds here.

Raises:

Type Description
KeyError

If contact_id, receive_date or total_amount is absent after normalisation.

Warns:

Type Description
UserWarning

If exclude_entry_types was requested but the export carries no entry type column. Silently summing a pledge together with its payments (or an uncharged recurring schedule) is exactly the error this bridge exists to prevent, so it is worth a warning rather than a quiet wrong total.
Notes

Nothing here deduplicates on a transaction id, so concatenating two exports that overlap in time will double-count the overlap; export disjoint date ranges.

Examples:

>>> rows = [
...     {"Account Number": "88", "Date": "2025-01-10",
...      "Amount": "1200.00", "Transaction Type": "Pledge"},
...     {"Account Number": "88", "Date": "2025-02-10",
...      "Amount": "100.00", "Transaction Type": "PledgePayment"},
...     {"Account Number": "88", "Date": "2025-03-10",
...      "Amount": "100.00", "Transaction Type": "PledgePayment"},
... ]
>>> feats = bloomerang_transactions_to_features(rows)
>>> float(feats.loc["88", "total_gift_amount"])  # not 1400.0
200.0
>>> int(feats.loc["88", "gift_count"])
2
Source code in philanthropy/ingest/_bloomerang.py 
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
def bloomerang_transactions_to_features(
    transactions: Union[Iterable[Mapping], pd.DataFrame],
    *,
    reference_date: Optional[Union[str, pd.Timestamp]] = None,
    exclude_entry_types: Optional[Sequence[str]] = DEFAULT_EXCLUDED_ENTRY_TYPES,
) -> pd.DataFrame:
    """Aggregate a Bloomerang transaction export into donor-level features.

    Commitment rows (pledges and recurring-donation schedules) are dropped
    first, then the surviving payment rows are handed to
    :func:`civicrm_contributions_to_features`, which produces the donor
    frame. Bloomerang has no contribution-status column analogous to
    CiviCRM's, so no status filter is applied.

    Parameters
    ----------
    transactions : iterable of mapping, or DataFrame
        Bloomerang transaction rows, under a report export label
        (``Account Number``, ``Date``, ``Amount``, ``Transaction Type``,
        ``Fund``) or the REST API field names (``AccountId``, ``Date``,
        ``Amount``, ``EntryType``, ``FundName``). ``contact_id``,
        ``receive_date`` and ``total_amount`` are required after
        normalisation; ``gift_type``, ``financial_type``, ``email``,
        ``first_name`` and ``last_name`` are used when present.
    reference_date : str or datetime-like, optional
        Anchor for the recency features. If ``None``, the latest transaction
        date in the batch is used, which keeps the aggregation free of "now"
        leakage.
    exclude_entry_types : sequence of str or None, default=:data:`DEFAULT_EXCLUDED_ENTRY_TYPES`
        Entry types to drop before aggregating, matched against ``gift_type``
        ignoring case, spacing and punctuation. ``None`` or an empty sequence
        disables the filter and sums **every** row, which double-counts
        pledged dollars and counts a recurring schedule never charged. Rows
        whose entry type is blank are kept either way: an unlabelled row
        cannot be shown to be a commitment.

    Returns
    -------
    features : pandas.DataFrame
        One row per donor, indexed by ``contact_id``, with the same columns
        :func:`civicrm_contributions_to_features` emits.
        ``distinct_financial_types`` counts distinct Funds here.

    Raises
    ------
    KeyError
        If ``contact_id``, ``receive_date`` or ``total_amount`` is absent
        after normalisation.

    Warns
    -----
    UserWarning
        If ``exclude_entry_types`` was requested but the export carries no
        entry type column. Silently summing a pledge together with its
        payments (or an uncharged recurring schedule) is exactly the error
        this bridge exists to prevent, so it is worth a warning rather than a
        quiet wrong total.

    Notes
    -----
    Nothing here deduplicates on a transaction id, so concatenating two
    exports that overlap in time will double-count the overlap; export
    disjoint date ranges.

    Examples
    --------
    >>> rows = [
    ...     {"Account Number": "88", "Date": "2025-01-10",
    ...      "Amount": "1200.00", "Transaction Type": "Pledge"},
    ...     {"Account Number": "88", "Date": "2025-02-10",
    ...      "Amount": "100.00", "Transaction Type": "PledgePayment"},
    ...     {"Account Number": "88", "Date": "2025-03-10",
    ...      "Amount": "100.00", "Transaction Type": "PledgePayment"},
    ... ]
    >>> feats = bloomerang_transactions_to_features(rows)
    >>> float(feats.loc["88", "total_gift_amount"])  # not 1400.0
    200.0
    >>> int(feats.loc["88", "gift_count"])
    2
    """
    df = _normalise_headers(_to_frame(transactions), _canonical_bloomerang)
    if df.empty:
        return _empty_feature_frame()

    missing = [col for col in _REQUIRED if col not in df.columns]
    if missing:
        raise KeyError(
            f"Bloomerang transaction export is missing {missing}. Export the "
            f"'Account Number', 'Date' and 'Amount' fields (REST API: "
            f"AccountId, Date, Amount); got {sorted(df.columns)}."
        )

    if exclude_entry_types:
        if "gift_type" in df.columns:
            unwanted = {_entry_type_key(t) for t in exclude_entry_types}
            keys = df["gift_type"].map(_entry_type_key)
            df = df[~keys.isin(unwanted)]
        else:
            warnings.warn(
                f"Bloomerang transaction export has no entry type column, so "
                f"exclude_entry_types={tuple(exclude_entry_types)!r} could not "
                f"be applied: pledges and recurring-donation schedules (if "
                f"any) are summed alongside the payments made against them, "
                f"which counts every committed dollar twice. Add the "
                f"'Transaction Type' field to the export.",
                stacklevel=2,
            )

    if df.empty:
        return _empty_feature_frame()
    return civicrm_contributions_to_features(
        df, reference_date=reference_date, statuses=None
    )

read_bloomerang_transactions(path)

Read Bloomerang transaction export CSV(s) into one frame.

Accepts a single .csv or a directory of them, walked recursively and concatenated in sorted relative-path order; symlinks are not followed. Every column is read as text and nothing is filtered: commitment rows are still present, and it is :func:bloomerang_transactions_to_features that drops them.

Reading a gift export is CRM-agnostic once the headers are normalised, so this delegates to :func:read_civicrm_contributions rather than repeating its BOM handling and path hardening. Bloomerang headers are normalised on the way into :func:bloomerang_transactions_to_features, not here.

Parameters:

Name Type Description Default
path str or Path

CSV file, or a directory of them.

 required

Returns:

Name Type Description
transactions DataFrame

The export as written, with text values.

Raises:

Type Description
FileNotFoundError

If path does not exist.
Source code in philanthropy/ingest/_bloomerang.py 
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
def read_bloomerang_transactions(path: Union[str, Path]) -> pd.DataFrame:
    """Read Bloomerang transaction export CSV(s) into one frame.

    Accepts a single ``.csv`` or a directory of them, walked recursively and
    concatenated in sorted relative-path order; symlinks are not followed.
    Every column is read as text and nothing is filtered: **commitment rows
    are still present**, and it is
    :func:`bloomerang_transactions_to_features` that drops them.

    Reading a gift export is CRM-agnostic once the headers are normalised, so
    this delegates to :func:`read_civicrm_contributions` rather than
    repeating its BOM handling and path hardening. Bloomerang headers are
    normalised on the way into
    :func:`bloomerang_transactions_to_features`, not here.

    Parameters
    ----------
    path : str or pathlib.Path
        CSV file, or a directory of them.

    Returns
    -------
    transactions : pandas.DataFrame
        The export as written, with text values.

    Raises
    ------
    FileNotFoundError
        If ``path`` does not exist.
    """
    return read_civicrm_contributions(path)

civicrm_contributions_to_features(contributions, *, reference_date=None, statuses=('Completed',))

Aggregate a CiviCRM contribution log into donor-level features.

Parameters:

Name Type Description Default
contributions iterable of mapping, or DataFrame

CiviCRM contribution rows. Accepts the output of :func:read_civicrm_contributions, an APIv4 Contribution.get result, or a DataFrame of the same fields, under either the export labels or the APIv4 column names. contact_id, receive_date and total_amount are required; id, currency, contribution_status, financial_type, is_test, email, first_name and last_name are used when present.

 required
reference_date str or datetime - like

Anchor for the recency features (years_active, recency_days). If None, the latest receive_date in the batch is used; this keeps the aggregation reproducible and free of "now" leakage. Naive timestamps are interpreted as UTC.

 None
statuses sequence of str or None

Contribution statuses to count, matched case-insensitively against contribution_status. None disables the filter and counts every row, including Failed and Refunded ones. Test-mode rows are dropped either way.

 ``("Completed",)``

Returns:

Name Type Description
features DataFrame

One row per donor, indexed by contact_id, with the columns declared in _FEATURE_DTYPES. recency_days, gift_count and total_gift_amount are the R, F and M of an RFM model. Rows are sorted by contact_id for determinism.

Raises:

Type Description
KeyError

If contact_id, receive_date or total_amount is absent. An export missing one of them cannot be aggregated at all, and failing here names the field instead of surfacing a bare column lookup later.

Warns:

Type Description
UserWarning

If statuses was requested but the batch carries no contribution_status column: the filter silently counting refunded and failed gifts is exactly the error this bridge exists to prevent.
UserWarning

If the batch mixes currencies. total_gift_amount is a plain sum with no FX conversion, so a single-currency export is assumed.

Examples:

>>> rows = [
...     {"Contact ID": "101", "Contribution Date": "2025-01-15",
...      "Total Amount": "250.00", "Contribution Status": "Completed"},
...     {"Contact ID": "101", "Contribution Date": "2025-06-01",
...      "Total Amount": "1,000.00", "Contribution Status": "Completed"},
...     {"Contact ID": "101", "Contribution Date": "2025-06-02",
...      "Total Amount": "99.00", "Contribution Status": "Failed"},
... ]
>>> feats = civicrm_contributions_to_features(rows)
>>> float(feats.loc["101", "total_gift_amount"])
1250.0
>>> int(feats.loc["101", "gift_count"])
2
Source code in philanthropy/ingest/_civicrm.py 
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
def civicrm_contributions_to_features(
    contributions: Union[Iterable[Mapping], pd.DataFrame],
    *,
    reference_date: Optional[Union[str, pd.Timestamp]] = None,
    statuses: Optional[Sequence[str]] = ("Completed",),
) -> pd.DataFrame:
    """Aggregate a CiviCRM contribution log into donor-level features.

    Parameters
    ----------
    contributions : iterable of mapping, or DataFrame
        CiviCRM contribution rows. Accepts the output of
        :func:`read_civicrm_contributions`, an APIv4 ``Contribution.get``
        result, or a DataFrame of the same fields, under either the export
        labels or the APIv4 column names. ``contact_id``, ``receive_date`` and
        ``total_amount`` are required; ``id``, ``currency``,
        ``contribution_status``, ``financial_type``, ``is_test``, ``email``,
        ``first_name`` and ``last_name`` are used when present.
    reference_date : str or datetime-like, optional
        Anchor for the recency features (``years_active``, ``recency_days``).
        If ``None``, the latest ``receive_date`` in the batch is used; this
        keeps the aggregation reproducible and free of "now" leakage. Naive
        timestamps are interpreted as UTC.
    statuses : sequence of str or None, default=``("Completed",)``
        Contribution statuses to count, matched case-insensitively against
        ``contribution_status``. ``None`` disables the filter and counts every
        row, including ``Failed`` and ``Refunded`` ones. Test-mode rows are
        dropped either way.

    Returns
    -------
    features : pandas.DataFrame
        One row per donor, indexed by ``contact_id``, with the columns declared
        in ``_FEATURE_DTYPES``. ``recency_days``, ``gift_count`` and
        ``total_gift_amount`` are the R, F and M of an RFM model. Rows are
        sorted by ``contact_id`` for determinism.

    Raises
    ------
    KeyError
        If ``contact_id``, ``receive_date`` or ``total_amount`` is absent. An
        export missing one of them cannot be aggregated at all, and failing
        here names the field instead of surfacing a bare column lookup later.

    Warns
    -----
    UserWarning
        If ``statuses`` was requested but the batch carries no
        ``contribution_status`` column: the filter silently counting refunded
        and failed gifts is exactly the error this bridge exists to prevent.
    UserWarning
        If the batch mixes currencies. ``total_gift_amount`` is a plain sum with
        no FX conversion, so a single-currency export is assumed.

    Examples
    --------
    >>> rows = [
    ...     {"Contact ID": "101", "Contribution Date": "2025-01-15",
    ...      "Total Amount": "250.00", "Contribution Status": "Completed"},
    ...     {"Contact ID": "101", "Contribution Date": "2025-06-01",
    ...      "Total Amount": "1,000.00", "Contribution Status": "Completed"},
    ...     {"Contact ID": "101", "Contribution Date": "2025-06-02",
    ...      "Total Amount": "99.00", "Contribution Status": "Failed"},
    ... ]
    >>> feats = civicrm_contributions_to_features(rows)
    >>> float(feats.loc["101", "total_gift_amount"])
    1250.0
    >>> int(feats.loc["101", "gift_count"])
    2
    """
    df = _normalise_headers(_to_frame(contributions))
    if df.empty:
        return _empty_feature_frame()

    missing = [col for col in _REQUIRED if col not in df.columns]
    if missing:
        raise KeyError(
            f"CiviCRM contribution log is missing {missing}. Export the "
            f"'Contact ID', 'Contribution Date' and 'Total Amount' fields "
            f"(APIv4: contact_id, receive_date, total_amount); got "
            f"{sorted(df.columns)}."
        )

    df = df.copy()

    # Two rows sharing a CiviCRM contribution id are the same contribution;
    # concatenating overlapping monthly exports is how that happens. Collapse
    # only rows with a real id: pandas' duplicated() treats NaN == NaN, which
    # would drop every id-less gift once any row carries one.
    if "id" in df.columns:
        df = df[~(df["id"].notna() & df.duplicated(subset="id"))]

    if "is_test" in df.columns:
        df = df[~_to_bool(df["is_test"])]

    if statuses is not None:
        if "contribution_status" in df.columns:
            wanted = {str(s).strip().casefold() for s in statuses}
            status = df["contribution_status"].astype("string").str.strip().str.casefold()
            df = df[status.isin(wanted)]
        else:
            warnings.warn(
                f"CiviCRM contribution log has no contribution_status column, so "
                f"statuses={tuple(statuses)!r} could not be applied: Pending, "
                f"Failed, Refunded and Cancelled gifts (if any) are counted in "
                f"total_gift_amount. Add the 'Contribution Status' field to the "
                f"export, or pass statuses=None to accept every row.",
                stacklevel=2,
            )

    # total_gift_amount is a plain sum; CiviCRM carries a per-row `currency` but
    # the export has no FX rates, so a mixed-currency batch would add apples to
    # oranges. Warn rather than convert (rates aren't there) or crash.
    if "currency" in df.columns and df["currency"].dropna().nunique() > 1:
        warnings.warn(
            "CiviCRM contribution log mixes currencies "
            f"({sorted(df['currency'].dropna().unique())}); total_gift_amount is "
            "summed without FX conversion. Normalise to one currency first.",
            stacklevel=2,
        )

    df["_contact_id"] = df["contact_id"].astype("string").str.strip()
    df["_ts"] = _to_datetime(df["receive_date"])
    df["_amount"] = _to_amount(df["total_amount"]).fillna(0.0)
    # A gift we cannot attribute to a donor, or place in time, contributes to no
    # feature; drop it rather than let a NaT poison a donor's recency.
    df = df[
        df["_contact_id"].notna()
        & (df["_contact_id"].str.len() > 0)
        & df["_ts"].notna()
    ]
    if df.empty:
        return _empty_feature_frame()

    ref = _resolve_reference_date(reference_date, df["_ts"])

    grouped = df.groupby("_contact_id", sort=True)
    out = pd.DataFrame(index=grouped.size().index)
    # Optional identity fields: carried through when the export supplies them
    # (this is a donor-level table keyed by CRM id, not a de-identified feature
    # store). ``.first()`` skips nulls, so a donor whose name rode in on only
    # some rows still resolves. Absent column -> None.
    for out_col, src in (
        ("constituent_email", "email"),
        ("first_name", "first_name"),
        ("last_name", "last_name"),
    ):
        out[out_col] = grouped[src].first() if src in df.columns else None
    out["total_gift_amount"] = grouped["_amount"].sum()
    out["gift_count"] = grouped.size()
    out["largest_gift_amount"] = grouped["_amount"].max()
    out["first_gift_date"] = grouped["_ts"].min()
    out["last_gift_date"] = grouped["_ts"].max()
    out["years_active"] = (
        (ref - out["first_gift_date"]).dt.days / DAYS_PER_YEAR
    ).clip(lower=0.0)
    out["recency_days"] = (ref - out["last_gift_date"]).dt.days.clip(lower=0)

    # Financial Type is CiviCRM's gift classification (Donation, Member Dues,
    # Event Fee, ...); breadth across them is the CiviCRM analogue of the
    # UniSchema bridge's distinct_source_systems.
    type_col = next(
        (c for c in ("financial_type", "financial_type_id") if c in df.columns), None
    )
    out["distinct_financial_types"] = grouped[type_col].nunique() if type_col else 0

    out.index.name = "contact_id"
    return _coerce_schema(out)

read_civicrm_contributions(path)

Read CiviCRM contribution export CSV(s) into one normalised frame.

Accepts either a single .csv file or a directory, which is walked recursively and whose *.csv files are concatenated in sorted relative-path order, the shape you get from keeping a folder of monthly exports. Symlinks are not followed.

Every column is read as text and the headers are normalised to the APIv4 spelling, so "Total Amount", "Contact ID" and "Contribution Date" arrive as total_amount, contact_id and receive_date. Nothing else is done to the rows: test-mode and non-Completed contributions are still present, and it is :func:civicrm_contributions_to_features that drops them and types the values. Reading is deliberately lossless so the raw export stays inspectable.

Parameters:

Name Type Description Default
path str or Path

CSV file, or a directory of them.

 required

Returns:

Name Type Description
contributions DataFrame

The export as written, with normalised column names and text values. A directory holding no CSV returns an empty frame.

Raises:

Type Description
FileNotFoundError

If path does not exist. Without this an absent directory falls through to the single-file branch and surfaces as an opaque OSError.
Source code in philanthropy/ingest/_civicrm.py 
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
def read_civicrm_contributions(path: Union[str, Path]) -> pd.DataFrame:
    """Read CiviCRM contribution export CSV(s) into one normalised frame.

    Accepts either a single ``.csv`` file or a directory, which is walked
    **recursively** and whose ``*.csv`` files are concatenated in sorted
    relative-path order, the shape you get from keeping a folder of monthly
    exports. Symlinks are not followed.

    Every column is read as text and the headers are normalised to the APIv4
    spelling, so ``"Total Amount"``, ``"Contact ID"`` and ``"Contribution Date"``
    arrive as ``total_amount``, ``contact_id`` and ``receive_date``. Nothing else
    is done to the rows: **test-mode and non-``Completed`` contributions are
    still present**, and it is
    :func:`civicrm_contributions_to_features` that drops them and types the
    values. Reading is deliberately lossless so the raw export stays inspectable.

    Parameters
    ----------
    path : str or pathlib.Path
        CSV file, or a directory of them.

    Returns
    -------
    contributions : pandas.DataFrame
        The export as written, with normalised column names and text values.
        A directory holding no CSV returns an empty frame.

    Raises
    ------
    FileNotFoundError
        If ``path`` does not exist. Without this an absent directory falls
        through to the single-file branch and surfaces as an opaque OSError.
    """
    p = Path(path)
    if not p.exists():
        raise FileNotFoundError(f"No such file or directory: {p}")
    if p.is_dir():
        files = [
            f
            for f in p.rglob("*.csv")
            # Skip symlinks: one inside the export folder pointing outside it
            # must not be followed and read (path-traversal hardening).
            if f.is_file() and not f.is_symlink()
        ]
        if not files:
            return pd.DataFrame()
        # Sort by relative path so ordering is deterministic across platforms.
        frames = [
            _read_csv(f) for f in sorted(files, key=lambda f: f.relative_to(p).as_posix())
        ]
        return pd.concat(frames, ignore_index=True)
    return _read_csv(p)

constituent_events_to_features(events, *, reference_date=None, deduplicate=True)

Aggregate a UniSchema ConstituentEvent stream into donor features.

Parameters:

Name Type Description Default
events iterable of mapping, or DataFrame

Records following UniSchema's ConstituentEvent schema. Each event carries at least constituentEmail, eventType, sourceSystem, and createdAt (ISO-8601); amount, externalConstituentId, eventId, firstName, and lastName are optional. Accepts the output of :func:read_constituent_events, a list of dicts, or a DataFrame of the same fields.

 required
reference_date str or datetime - like

Anchor for the recency features (years_active, recency_days). If None, the latest createdAt in the batch is used; this keeps the aggregation reproducible and free of "now" leakage. Naive timestamps are interpreted as UTC.

 None
deduplicate bool

Drop repeated eventId values before aggregating. Advancement webhooks are at-least-once, so this prevents a redelivered donation from being counted (and its dollars summed) twice.

 True

Returns:

Name Type Description
features DataFrame

One row per constituent, indexed by constituent_id (the externalConstituentId when present, else constituentEmail), with the columns declared in _FEATURE_DTYPES (first_name / last_name are populated from the feed when available, else null). Rows are sorted by constituent_id for determinism.

Warns:

Type Description
UserWarning

If the batch mixes currencies (more than one distinct currency). total_gift_amount is a plain sum with no FX conversion, so a single-currency feed is assumed; normalise upstream if it isn't.

Examples:

>>> events = [
...     {"constituentEmail": "a@x.edu", "eventType": "DONATION",
...      "sourceSystem": "GIVECAMPUS", "amount": 250.0,
...      "createdAt": "2025-03-01T12:00:00Z"},
...     {"constituentEmail": "a@x.edu", "eventType": "EVENT_REGISTRATION",
...      "sourceSystem": "CVENT", "createdAt": "2025-06-01T09:00:00Z"},
... ]
>>> feats = constituent_events_to_features(events)
>>> float(feats.loc["a@x.edu", "total_gift_amount"])
250.0
>>> int(feats.loc["a@x.edu", "event_attendance_count"])
1
Source code in philanthropy/ingest/_constituent_events.py 
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
def constituent_events_to_features(
    events: Union[Iterable[Mapping], pd.DataFrame],
    *,
    reference_date: Optional[Union[str, pd.Timestamp]] = None,
    deduplicate: bool = True,
) -> pd.DataFrame:
    """Aggregate a UniSchema ``ConstituentEvent`` stream into donor features.

    Parameters
    ----------
    events : iterable of mapping, or DataFrame
        Records following UniSchema's ``ConstituentEvent`` schema.  Each event
        carries at least ``constituentEmail``, ``eventType``, ``sourceSystem``,
        and ``createdAt`` (ISO-8601); ``amount``, ``externalConstituentId``,
        ``eventId``, ``firstName``, and ``lastName`` are optional.  Accepts the
        output of
        :func:`read_constituent_events`, a list of dicts, or a DataFrame of the
        same fields.
    reference_date : str or datetime-like, optional
        Anchor for the recency features (``years_active``, ``recency_days``).
        If ``None``, the latest ``createdAt`` in the batch is used; this keeps
        the aggregation reproducible and free of "now" leakage.  Naive
        timestamps are interpreted as UTC.
    deduplicate : bool, default=True
        Drop repeated ``eventId`` values before aggregating.  Advancement
        webhooks are at-least-once, so this prevents a redelivered donation from
        being counted (and its dollars summed) twice.

    Returns
    -------
    features : pandas.DataFrame
        One row per constituent, indexed by ``constituent_id`` (the
        ``externalConstituentId`` when present, else ``constituentEmail``), with
        the columns declared in ``_FEATURE_DTYPES`` (``first_name`` /
        ``last_name`` are populated from the feed when available, else null).
        Rows are sorted by ``constituent_id`` for determinism.

    Warns
    -----
    UserWarning
        If the batch mixes currencies (more than one distinct ``currency``).
        ``total_gift_amount`` is a plain sum with no FX conversion, so a
        single-currency feed is assumed; normalise upstream if it isn't.

    Examples
    --------
    >>> events = [
    ...     {"constituentEmail": "a@x.edu", "eventType": "DONATION",
    ...      "sourceSystem": "GIVECAMPUS", "amount": 250.0,
    ...      "createdAt": "2025-03-01T12:00:00Z"},
    ...     {"constituentEmail": "a@x.edu", "eventType": "EVENT_REGISTRATION",
    ...      "sourceSystem": "CVENT", "createdAt": "2025-06-01T09:00:00Z"},
    ... ]
    >>> feats = constituent_events_to_features(events)
    >>> float(feats.loc["a@x.edu", "total_gift_amount"])
    250.0
    >>> int(feats.loc["a@x.edu", "event_attendance_count"])
    1
    """
    df = _to_frame(events)
    if df.empty:
        return _empty_feature_frame()

    # total_gift_amount sums raw amounts; UniSchema carries a per-event
    # `currency` but no FX rates, so a mixed-currency feed would sum apples and
    # oranges. Warn rather than convert (rates aren't in the stream) or crash.
    if "currency" in df.columns and df["currency"].dropna().nunique() > 1:
        warnings.warn(
            "ConstituentEvent feed mixes currencies "
            f"({sorted(df['currency'].dropna().unique())}); total_gift_amount "
            "is summed without FX conversion. Normalise to one currency first.",
            stacklevel=2,
        )

    if deduplicate and "eventId" in df.columns:
        # Collapse only rows that share a real eventId.  A missing eventId is not
        # "equal" to another missing one, but pandas' drop_duplicates treats
        # NaN == NaN, which would silently drop every id-less donation (and even
        # whole donors) once any event carries an id.
        is_dup = df["eventId"].notna() & df.duplicated(subset="eventId")
        df = df[~is_dup]

    df = df.copy()
    df["_constituent_id"] = _constituent_id(df)
    df["_ts"] = _to_utc_naive(df["createdAt"])
    # An event we can't place in time can't contribute to time-aware features;
    # drop it rather than let a NaT poison a donor's recency to a crash.
    df = df[df["_ts"].notna()]
    if df.empty:
        return _empty_feature_frame()

    # fillna("") keeps a missing/None eventType from becoming pd.NA in the
    # comparisons below (np.where on a NA-bearing mask raises); such an event
    # simply matches no type and contributes to no typed count.
    event_type = (
        df.get("eventType", pd.Series(index=df.index, dtype="object"))
        .astype("string")
        .fillna("")
    )
    raw_amount = df["amount"] if "amount" in df.columns else pd.Series(np.nan, index=df.index)
    # ponytail: UniSchema guarantees `amount` is a JSON number, so a missing
    # amount coerces to 0 (a real absent gift); an unparseable string would too.
    # Parse currency strings ('$250', '1,250.00') here only if a non-conforming
    # feed is ever fed in directly, bypassing UniSchema's validation.
    amount = pd.to_numeric(raw_amount, errors="coerce")

    is_donation = (event_type == _DONATION).to_numpy()
    df["_gift_amount"] = np.where(is_donation, amount.fillna(0.0).to_numpy(), 0.0)
    df["_is_donation"] = is_donation.astype("int64")
    df["_is_event"] = (event_type == _EVENT_REGISTRATION).astype("int64")
    df["_is_click"] = (event_type == _EMAIL_CLICK).astype("int64")
    df["_donation_ts"] = df["_ts"].where(is_donation)

    ref = _resolve_reference_date(reference_date, df["_ts"])

    grouped = df.groupby("_constituent_id", sort=True)
    out = pd.DataFrame(index=grouped.size().index)
    out["constituent_email"] = grouped["constituentEmail"].first()
    # Optional identity fields: carried through when the feed supplies them
    # (the output is a donor-level table, not a de-identified feature store, so
    # it already holds constituent_email). ``.first()`` skips nulls, so a donor
    # whose name rode in on only some events still resolves. Absent column -> None.
    for out_col, src in (("first_name", "firstName"), ("last_name", "lastName")):
        out[out_col] = grouped[src].first() if src in df.columns else None
    out["total_gift_amount"] = grouped["_gift_amount"].sum()
    out["gift_count"] = grouped["_is_donation"].sum()
    out["event_attendance_count"] = grouped["_is_event"].sum()
    out["email_click_count"] = grouped["_is_click"].sum()
    out["first_gift_date"] = grouped["_donation_ts"].min()
    out["last_gift_date"] = grouped["_donation_ts"].max()

    first_seen = grouped["_ts"].min()
    last_seen = grouped["_ts"].max()
    out["years_active"] = ((ref - first_seen).dt.days / DAYS_PER_YEAR).clip(lower=0.0)
    out["recency_days"] = (ref - last_seen).dt.days.clip(lower=0)

    if "sourceSystem" in df.columns:
        out["distinct_source_systems"] = grouped["sourceSystem"].nunique()
    else:
        out["distinct_source_systems"] = 0

    out.index.name = "constituent_id"
    return _coerce_schema(out)

read_constituent_events(path)

Read UniSchema egress files into a list of ConstituentEvent dicts.

Handles the shapes UniSchema's egress writes:

  • a single .json file holding one event (object) or many (array);
  • a .ndjson / .jsonl batch, one event per line;
  • a directory, which is walked recursively: every *.json, *.ndjson, and *.jsonl file at any depth is read and concatenated, sorted by relative path. This handles UniSchema's date-partitioned egress ({prefix}/{vendor}/{yyyy}/{mm}/{dd}/{eventId}.json); a flat directory still works too. *.manifest.json batch sidecars are skipped.

Parameters:

Name Type Description Default
path str or Path

File or directory to read.

 required

Returns:

Name Type Description
events list of dict

Parsed events, ready to pass to :func:constituent_events_to_features.

Raises:

Type Description
FileNotFoundError

If path does not exist. Without this an absent directory falls through to the single-file branch and surfaces as an opaque OSError.
Source code in philanthropy/ingest/_constituent_events.py 
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
def read_constituent_events(
    path: Union[str, Path],
) -> "list[dict]":
    """Read UniSchema egress files into a list of ``ConstituentEvent`` dicts.

    Handles the shapes UniSchema's egress writes:

    * a single ``.json`` file holding one event (object) or many (array);
    * a ``.ndjson`` / ``.jsonl`` batch, one event per line;
    * a directory, which is walked **recursively**: every ``*.json``,
      ``*.ndjson``, and ``*.jsonl`` file at any depth is read and concatenated,
      sorted by relative path.  This handles UniSchema's date-partitioned egress
      (``{prefix}/{vendor}/{yyyy}/{mm}/{dd}/{eventId}.json``); a flat directory
      still works too.  ``*.manifest.json`` batch sidecars are skipped.

    Parameters
    ----------
    path : str or pathlib.Path
        File or directory to read.

    Returns
    -------
    events : list of dict
        Parsed events, ready to pass to :func:`constituent_events_to_features`.

    Raises
    ------
    FileNotFoundError
        If ``path`` does not exist.  Without this an absent directory falls
        through to the single-file branch and surfaces as an opaque OSError.
    """
    p = Path(path)
    if not p.exists():
        raise FileNotFoundError(f"No such file or directory: {p}")
    if p.is_dir():
        # UniSchema's local egress is date-partitioned: it writes each event to
        # {prefix}/{vendor}/{yyyy}/{mm}/{dd}/{eventId}.json (see UniSchema
        # src/egress/objectKey.ts), so the files sit several levels down and a
        # non-recursive scan of the top dir finds nothing. Walk the whole tree.
        events: "list[dict]" = []
        files = [
            f
            for f in p.rglob("*")
            if f.is_file()
            # Skip symlinks: a symlink inside the egress tree pointing outside
            # it must not be followed and read (path-traversal hardening).
            and not f.is_symlink()
            and f.suffix.lower() in {".json", ".ndjson", ".jsonl"}
            # Skip S3 batch sidecars: batch metadata, not ConstituentEvents.
            and not f.name.lower().endswith(".manifest.json")
        ]
        # Sort by relative path so ordering is deterministic across platforms.
        for child in sorted(files, key=lambda f: f.relative_to(p).as_posix()):
            events.extend(_read_events_file(child))
        return events
    return _read_events_file(p)

donorperfect_gifts_to_features(gifts, *, reference_date=None, exclude_record_types=DEFAULT_EXCLUDED_RECORD_TYPES)

Aggregate a DonorPerfect gift export into donor-level features.

Pledge records (record_type='P') and split-gift Main totals (record_type='M') are dropped first, then the surviving gift rows (regular gifts, pledge payments, and split gift entries, all record_type='G') are handed to :func:civicrm_contributions_to_features, which produces the donor frame. DonorPerfect has no contribution-status column analogous to CiviCRM's, so no status filter is applied.

Parameters:

Name Type Description Default
gifts iterable of mapping, or DataFrame

DonorPerfect gift rows, under a "Gift/Pledge Transactions" export label (Donor ID, Gift Date, Gift Amount, Record Type) or the DB column / XML API field names (DONOR_ID, GIFT_DATE, AMOUNT, RECORD_TYPE). contact_id, receive_date and total_amount are required after normalisation; gift_type, financial_type, email, first_name and last_name are used when present.

 required
reference_date str or datetime - like

Anchor for the recency features. If None, the latest gift date in the batch is used, which keeps the aggregation free of "now" leakage.

 None
exclude_record_types sequence of str or None

Record types to drop before aggregating, matched against gift_type ignoring case, spacing and punctuation. None or an empty sequence disables the filter and sums every row, which double-counts pledged dollars and split-gift totals. Rows whose record type is blank are kept either way: an unlabelled row cannot be shown to be a commitment.

 :data:`DEFAULT_EXCLUDED_RECORD_TYPES`

Returns:

Name Type Description
features DataFrame

One row per donor, indexed by contact_id, with the same columns :func:civicrm_contributions_to_features emits. distinct_financial_types counts distinct GL codes here.

Raises:

Type Description
KeyError

If contact_id, receive_date or total_amount is absent after normalisation.

Warns:

Type Description
UserWarning

If exclude_record_types was requested but the export carries no record type column. Silently summing a pledge together with its payments (or a split gift's Main total together with its splits) is exactly the error this bridge exists to prevent, so it is worth a warning rather than a quiet wrong total.
Notes

Nothing here deduplicates on a gift id, so concatenating two exports that overlap in time will double-count the overlap; export disjoint date ranges. Soft credits (record_type='S') are out of scope for this filter and are summed if present; DonorPerfect creates a soft credit under a different donor than the one who actually gave, so including them attributes the same dollar to two donors by design of that feature, not a bug this bridge is meant to catch.

Examples:

>>> rows = [
...     {"Donor ID": "88", "Gift Date": "2025-01-10",
...      "Gift Amount": "1200.00", "Record Type": "P"},
...     {"Donor ID": "88", "Gift Date": "2025-02-10",
...      "Gift Amount": "100.00", "Record Type": "G"},
...     {"Donor ID": "88", "Gift Date": "2025-03-10",
...      "Gift Amount": "100.00", "Record Type": "G"},
... ]
>>> feats = donorperfect_gifts_to_features(rows)
>>> float(feats.loc["88", "total_gift_amount"])  # not 1400.0
200.0
>>> int(feats.loc["88", "gift_count"])
2
Source code in philanthropy/ingest/_donorperfect.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
def donorperfect_gifts_to_features(
    gifts: Union[Iterable[Mapping], pd.DataFrame],
    *,
    reference_date: Optional[Union[str, pd.Timestamp]] = None,
    exclude_record_types: Optional[Sequence[str]] = DEFAULT_EXCLUDED_RECORD_TYPES,
) -> pd.DataFrame:
    """Aggregate a DonorPerfect gift export into donor-level features.

    Pledge records (``record_type='P'``) and split-gift Main totals
    (``record_type='M'``) are dropped first, then the surviving gift rows
    (regular gifts, pledge payments, and split gift entries, all
    ``record_type='G'``) are handed to :func:`civicrm_contributions_to_features`,
    which produces the donor frame. DonorPerfect has no contribution-status
    column analogous to CiviCRM's, so no status filter is applied.

    Parameters
    ----------
    gifts : iterable of mapping, or DataFrame
        DonorPerfect gift rows, under a "Gift/Pledge Transactions" export
        label (``Donor ID``, ``Gift Date``, ``Gift Amount``, ``Record
        Type``) or the DB column / XML API field names (``DONOR_ID``,
        ``GIFT_DATE``, ``AMOUNT``, ``RECORD_TYPE``). ``contact_id``,
        ``receive_date`` and ``total_amount`` are required after
        normalisation; ``gift_type``, ``financial_type``, ``email``,
        ``first_name`` and ``last_name`` are used when present.
    reference_date : str or datetime-like, optional
        Anchor for the recency features. If ``None``, the latest gift date
        in the batch is used, which keeps the aggregation free of "now"
        leakage.
    exclude_record_types : sequence of str or None, default=:data:`DEFAULT_EXCLUDED_RECORD_TYPES`
        Record types to drop before aggregating, matched against
        ``gift_type`` ignoring case, spacing and punctuation. ``None`` or an
        empty sequence disables the filter and sums **every** row, which
        double-counts pledged dollars and split-gift totals. Rows whose
        record type is blank are kept either way: an unlabelled row cannot
        be shown to be a commitment.

    Returns
    -------
    features : pandas.DataFrame
        One row per donor, indexed by ``contact_id``, with the same columns
        :func:`civicrm_contributions_to_features` emits.
        ``distinct_financial_types`` counts distinct GL codes here.

    Raises
    ------
    KeyError
        If ``contact_id``, ``receive_date`` or ``total_amount`` is absent
        after normalisation.

    Warns
    -----
    UserWarning
        If ``exclude_record_types`` was requested but the export carries no
        record type column. Silently summing a pledge together with its
        payments (or a split gift's Main total together with its splits) is
        exactly the error this bridge exists to prevent, so it is worth a
        warning rather than a quiet wrong total.

    Notes
    -----
    Nothing here deduplicates on a gift id, so concatenating two exports
    that overlap in time will double-count the overlap; export disjoint
    date ranges. Soft credits (``record_type='S'``) are out of scope for
    this filter and are summed if present; DonorPerfect creates a soft
    credit under a *different* donor than the one who actually gave, so
    including them attributes the same dollar to two donors by design of
    that feature, not a bug this bridge is meant to catch.

    Examples
    --------
    >>> rows = [
    ...     {"Donor ID": "88", "Gift Date": "2025-01-10",
    ...      "Gift Amount": "1200.00", "Record Type": "P"},
    ...     {"Donor ID": "88", "Gift Date": "2025-02-10",
    ...      "Gift Amount": "100.00", "Record Type": "G"},
    ...     {"Donor ID": "88", "Gift Date": "2025-03-10",
    ...      "Gift Amount": "100.00", "Record Type": "G"},
    ... ]
    >>> feats = donorperfect_gifts_to_features(rows)
    >>> float(feats.loc["88", "total_gift_amount"])  # not 1400.0
    200.0
    >>> int(feats.loc["88", "gift_count"])
    2
    """
    df = _normalise_headers(_to_frame(gifts), _canonical_donorperfect)
    if df.empty:
        return _empty_feature_frame()

    missing = [col for col in _REQUIRED if col not in df.columns]
    if missing:
        raise KeyError(
            f"DonorPerfect gift export is missing {missing}. Export the "
            f"'Donor ID', 'Gift Date' and 'Gift Amount' fields (DB/API: "
            f"DONOR_ID, GIFT_DATE, AMOUNT); got {sorted(df.columns)}."
        )

    if exclude_record_types:
        if "gift_type" in df.columns:
            unwanted = {_record_type_key(t) for t in exclude_record_types}
            keys = df["gift_type"].map(_record_type_key)
            df = df[~keys.isin(unwanted)]
        else:
            warnings.warn(
                f"DonorPerfect gift export has no record type column, so "
                f"exclude_record_types={tuple(exclude_record_types)!r} could "
                f"not be applied: pledges and split-gift Main totals (if "
                f"any) are summed alongside the payments/splits made "
                f"against them, which counts every committed or split "
                f"dollar twice. Add the 'Record Type' field to the export.",
                stacklevel=2,
            )

    if df.empty:
        return _empty_feature_frame()
    return civicrm_contributions_to_features(
        df, reference_date=reference_date, statuses=None
    )

read_donorperfect_gifts(path)

Read DonorPerfect gift export CSV(s) into one frame.

Accepts a single .csv or a directory of them, walked recursively and concatenated in sorted relative-path order; symlinks are not followed. Every column is read as text and nothing is filtered: pledge and split-Main rows are still present, and it is :func:donorperfect_gifts_to_features that drops them.

Reading a gift export is CRM-agnostic once the headers are normalised, so this delegates to :func:read_civicrm_contributions rather than repeating its BOM handling and path hardening. DonorPerfect headers are normalised on the way into :func:donorperfect_gifts_to_features, not here.

Parameters:

Name Type Description Default
path str or Path

CSV file, or a directory of them.

 required

Returns:

Name Type Description
gifts DataFrame

The export as written, with text values.

Raises:

Type Description
FileNotFoundError

If path does not exist.
Source code in philanthropy/ingest/_donorperfect.py 
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
def read_donorperfect_gifts(path: Union[str, Path]) -> pd.DataFrame:
    """Read DonorPerfect gift export CSV(s) into one frame.

    Accepts a single ``.csv`` or a directory of them, walked recursively and
    concatenated in sorted relative-path order; symlinks are not followed.
    Every column is read as text and nothing is filtered: **pledge and
    split-Main rows are still present**, and it is
    :func:`donorperfect_gifts_to_features` that drops them.

    Reading a gift export is CRM-agnostic once the headers are normalised, so
    this delegates to :func:`read_civicrm_contributions` rather than
    repeating its BOM handling and path hardening. DonorPerfect headers are
    normalised on the way into :func:`donorperfect_gifts_to_features`, not
    here.

    Parameters
    ----------
    path : str or pathlib.Path
        CSV file, or a directory of them.

    Returns
    -------
    gifts : pandas.DataFrame
        The export as written, with text values.

    Raises
    ------
    FileNotFoundError
        If ``path`` does not exist.
    """
    return read_civicrm_contributions(path)

map_columns(df, mapping, *, required=())

Rename a DataFrame's columns per a user-supplied mapping.

Parameters:

Name Type Description Default
df DataFrame

The uploaded export, with whatever headers the source system wrote.

 required
mapping mapping of str to str

User header -> canonical name, e.g. {"Constituent ID": "contact_id", "Gift Date": "activity_date"}. Keys not present in df.columns are ignored, so a mapping built from a superset of known headers is safe to reuse across files.

 required
required sequence of str

Canonical column names that must be present after renaming.

 ()

Returns:

Name Type Description
mapped DataFrame

df with the mapped columns renamed. Unmapped columns are kept as-is.

Raises:

Type Description
ValueError

If, after renaming, any name in required is missing. The message lists every missing column at once, so a user fixing the mapping in a UI does not have to resubmit once per missing field.

Examples:

>>> import pandas as pd
>>> df = pd.DataFrame({"CnID": [1], "Gift Date": ["2025-01-01"]})
>>> map_columns(
...     df,
...     {"CnID": "contact_id", "Gift Date": "activity_date"},
...     required=["contact_id", "activity_date"],
... ).columns.tolist()
['contact_id', 'activity_date']
>>> map_columns(df, {"CnID": "contact_id"}, required=["contact_id", "amount"])
Traceback (most recent call last):
    ...
ValueError: missing required column(s) after mapping: amount
Source code in philanthropy/ingest/_map_columns.py 
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
def map_columns(
    df: pd.DataFrame,
    mapping: Mapping[str, str],
    *,
    required: Sequence[str] = (),
) -> pd.DataFrame:
    """Rename a DataFrame's columns per a user-supplied mapping.

    Parameters
    ----------
    df : pandas.DataFrame
        The uploaded export, with whatever headers the source system wrote.
    mapping : mapping of str to str
        User header -> canonical name, e.g. ``{"Constituent ID":
        "contact_id", "Gift Date": "activity_date"}``. Keys not present in
        ``df.columns`` are ignored, so a mapping built from a superset of
        known headers is safe to reuse across files.
    required : sequence of str, default=()
        Canonical column names that must be present after renaming.

    Returns
    -------
    mapped : pandas.DataFrame
        ``df`` with the mapped columns renamed. Unmapped columns are kept
        as-is.

    Raises
    ------
    ValueError
        If, after renaming, any name in ``required`` is missing. The message
        lists every missing column at once, so a user fixing the mapping in a
        UI does not have to resubmit once per missing field.

    Examples
    --------
    >>> import pandas as pd
    >>> df = pd.DataFrame({"CnID": [1], "Gift Date": ["2025-01-01"]})
    >>> map_columns(
    ...     df,
    ...     {"CnID": "contact_id", "Gift Date": "activity_date"},
    ...     required=["contact_id", "activity_date"],
    ... ).columns.tolist()
    ['contact_id', 'activity_date']
    >>> map_columns(df, {"CnID": "contact_id"}, required=["contact_id", "amount"])
    Traceback (most recent call last):
        ...
    ValueError: missing required column(s) after mapping: amount
    """
    mapping = dict(mapping)
    targets: "dict[str, list[str]]" = {}
    for source, target in mapping.items():
        if source in df.columns:
            targets.setdefault(target, []).append(source)
    collisions = {target: sources for target, sources in targets.items() if len(sources) > 1}
    if collisions:
        detail = "; ".join(
            f"{target!r} <- {sources}" for target, sources in collisions.items()
        )
        raise ValueError(
            "mapping assigns multiple source columns to the same target: " + detail
        )

    renamed = df.rename(columns=mapping)
    missing = [col for col in required if col not in renamed.columns]
    if missing:
        raise ValueError(
            "missing required column(s) after mapping: " + ", ".join(missing)
        )
    return renamed

nonprofit_cloud_gifts_to_features(gifts, *, reference_date=None, include_statuses=DEFAULT_INCLUDED_GIFT_TRANSACTION_STATUSES)

Aggregate a Nonprofit Cloud GiftTransaction export into donor features.

Transactions whose status is not a documented received status (Paid) are dropped first, then the surviving receipts are handed to :func:civicrm_contributions_to_features, which produces the donor frame. Nonprofit Cloud's GiftTransaction carries no separate pledge flag beyond Status, so no other filter is applied.

Parameters:

Name Type Description Default
gifts iterable of mapping, or DataFrame

Nonprofit Cloud GiftTransaction rows, under a report's column labels (Donor ID, Transaction Date, Current Amount, Status) or the raw API field names (DonorId, TransactionDate, CurrentAmount, Status). contact_id, receive_date and total_amount are required after normalisation; gift_type, status, email, first_name and last_name are used when present. If both a Donor column (DonorId) and a Contact column (ContactId) are present, the Donor ID is used as the donor key. The Account / Party reference donor is the aggregation unit, and the individual contact reference is a transaction-level fallback used only when the Donor is absent.

 required
reference_date str or datetime - like

Anchor for the recency features. If None, the latest Transaction Date in the batch is used, which keeps the aggregation free of "now" leakage.

 None
include_statuses sequence of str or None

Statuses to keep before aggregating, matched against status ignoring case, spacing and punctuation. None or an empty sequence disables the filter and sums every row, which double-counts a commitment's full schedule against the receipts that have actually landed: a GiftCommitment for a monthly pledge generates every scheduled GiftTransaction up front, and the two that have been received cannot be told apart from the ten that have not without reading Status. Rows whose status is blank are kept either way: an unlabelled row cannot be shown to be an expected-but-unpaid commitment. A partial refund does not change a transaction's Status (it stays Paid), so a partially refunded gift is kept by default and contributes its net CurrentAmount; only a full refund moves the row to Fully Refunded, which the default drops.

 :data:`DEFAULT_INCLUDED_GIFT_TRANSACTION_STATUSES`

Returns:

Name Type Description
features DataFrame

One row per donor, indexed by contact_id, with the same columns :func:civicrm_contributions_to_features emits. distinct_financial_types is not populated by this bridge (see Notes).

Raises:

Type Description
KeyError

If contact_id, receive_date or total_amount is absent after normalisation.

Warns:

Type Description
UserWarning

If include_statuses was requested but the export carries no Status column. Silently summing a commitment's unpaid and other non-paid schedules together with the receipts that have actually landed is exactly the error this bridge exists to prevent, so it is worth a warning rather than a quiet wrong total.
Notes

Nothing here deduplicates on a GiftTransaction id, so concatenating two exports that overlap in time will double-count the overlap; export disjoint date ranges.

Examples:

>>> rows = [
...     {"Donor ID": "88", "Transaction Date": "2025-01-10",
...      "Current Amount": "100.00", "Status": "Paid"},
...     {"Donor ID": "88", "Transaction Date": "2025-02-10",
...      "Current Amount": "100.00", "Status": "Unpaid"},
...     {"Donor ID": "88", "Transaction Date": "2025-03-10",
...      "Current Amount": "100.00", "Status": "Paid"},
... ]
>>> feats = nonprofit_cloud_gifts_to_features(rows)
>>> float(feats.loc["88", "total_gift_amount"])  # not 300.0
200.0
>>> int(feats.loc["88", "gift_count"])
2
Source code in philanthropy/ingest/_nonprofit_cloud.py 
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
def nonprofit_cloud_gifts_to_features(
    gifts: Union[Iterable[Mapping], pd.DataFrame],
    *,
    reference_date: Optional[Union[str, pd.Timestamp]] = None,
    include_statuses: Optional[Sequence[str]] = DEFAULT_INCLUDED_GIFT_TRANSACTION_STATUSES
) -> pd.DataFrame:
    """Aggregate a Nonprofit Cloud GiftTransaction export into donor features.

    Transactions whose status is not a documented received status (``Paid``) are
    dropped first, then the surviving receipts are handed to
    :func:`civicrm_contributions_to_features`, which produces the donor frame.
    Nonprofit Cloud's GiftTransaction carries no separate pledge flag beyond
    ``Status``, so no other filter is applied.

    Parameters
    ----------
    gifts : iterable of mapping, or DataFrame
        Nonprofit Cloud ``GiftTransaction`` rows, under a report's column
        labels (``Donor ID``, ``Transaction Date``, ``Current Amount``,
        ``Status``) or the raw API field names (``DonorId``,
        ``TransactionDate``, ``CurrentAmount``, ``Status``). ``contact_id``,
        ``receive_date`` and ``total_amount`` are required after
        normalisation; ``gift_type``, ``status``, ``email``, ``first_name``
        and ``last_name`` are used when present. If both a Donor column
        (``DonorId``) and a Contact column (``ContactId``) are present, the
        Donor ID is used as the donor key. The Account / Party reference donor
        is the aggregation unit, and the individual contact reference is a
        transaction-level fallback used only when the Donor is absent.
    reference_date : str or datetime-like, optional
        Anchor for the recency features. If ``None``, the latest
        Transaction Date in the batch is used, which keeps the aggregation
        free of "now" leakage.
    include_statuses : sequence of str or None, default=:data:`DEFAULT_INCLUDED_GIFT_TRANSACTION_STATUSES`
        Statuses to keep before aggregating, matched against ``status``
        ignoring case, spacing and punctuation. ``None`` or an empty sequence
        disables the filter and sums **every** row, which double-counts a
        commitment's full schedule against the receipts that have actually
        landed: a ``GiftCommitment`` for a monthly pledge generates every
        scheduled ``GiftTransaction`` up front, and the two that have been
        received cannot be told apart from the ten that have not without
        reading Status. Rows whose status is blank are kept either way: an
        unlabelled row cannot be shown to be an expected-but-unpaid
        commitment. A partial refund does not change a transaction's ``Status``
        (it stays ``Paid``), so a partially refunded gift is kept by default
        and contributes its net ``CurrentAmount``; only a full refund moves the
        row to ``Fully Refunded``, which the default drops.

    Returns
    -------
    features : pandas.DataFrame
        One row per donor, indexed by ``contact_id``, with the same columns
        :func:`civicrm_contributions_to_features` emits.
        ``distinct_financial_types`` is not populated by this bridge (see Notes).

    Raises
    ------
    KeyError
        If ``contact_id``, ``receive_date`` or ``total_amount`` is absent
        after normalisation.

    Warns
    -----
    UserWarning
        If ``include_statuses`` was requested but the export carries no
        Status column. Silently summing a commitment's unpaid and other non-paid
        schedules together with the receipts that have actually landed is exactly
        the error this bridge exists to prevent, so it is worth a warning rather
        than a quiet wrong total.

    Notes
    -----
    Nothing here deduplicates on a GiftTransaction id, so concatenating two exports
    that overlap in time will double-count the overlap; export disjoint date ranges.

    Examples
    --------
    >>> rows = [
    ...     {"Donor ID": "88", "Transaction Date": "2025-01-10",
    ...      "Current Amount": "100.00", "Status": "Paid"},
    ...     {"Donor ID": "88", "Transaction Date": "2025-02-10",
    ...      "Current Amount": "100.00", "Status": "Unpaid"},
    ...     {"Donor ID": "88", "Transaction Date": "2025-03-10",
    ...      "Current Amount": "100.00", "Status": "Paid"},
    ... ]
    >>> feats = nonprofit_cloud_gifts_to_features(rows)
    >>> float(feats.loc["88", "total_gift_amount"])  # not 300.0
    200.0
    >>> int(feats.loc["88", "gift_count"])
    2
    """
    df = _normalise_headers(_prefer_donor_key(_to_frame(gifts)), _canonical_nonprofit_cloud)
    if df.empty:
        return _empty_feature_frame()

    missing = [col for col in _REQUIRED if col not in df.columns]
    if missing:
        raise KeyError(
            f"Nonprofit Cloud GiftTransaction export is missing {missing}. "
            f"Export the 'Donor ID' (or 'Contact ID'), 'Transaction Date' "
            f"and 'Current Amount' fields (Salesforce API: DonorId, "
            f"TransactionDate, CurrentAmount); got {sorted(df.columns)}."
        )

    if include_statuses:
        if "status" in df.columns:
            wanted = {_status_key(s) for s in include_statuses}
            keys = df["status"].map(_status_key)
            df = df[(keys == "") | keys.isin(wanted)]
        else:
            warnings.warn(
                f"Nonprofit Cloud GiftTransaction export has no Status column, "
                f"so include_statuses={tuple(include_statuses)!r} could not be "
                f"applied: a GiftCommitment's unpaid and non-paid schedules (if "
                f"present) are summed alongside the receipts that have actually "
                f"landed, which counts money that has not yet been given. Add "
                f"the 'Status' field to the export.",
                stacklevel=2,
            )

    if df.empty:
        return _empty_feature_frame()
    return civicrm_contributions_to_features(
        df, reference_date=reference_date, statuses=None
    )

read_nonprofit_cloud_gifts(path)

Read Nonprofit Cloud GiftTransaction export CSV(s) into one frame.

Accepts a single .csv or a directory of them, walked recursively and concatenated in sorted relative-path order; symlinks are not followed. Every column is read as text and nothing is filtered: unpaid and other non-paid transactions are still present, and it is :func:nonprofit_cloud_gifts_to_features that drops them.

Reading a gift export is CRM-agnostic once the headers are normalised, so this delegates to :func:read_civicrm_contributions rather than repeating its BOM handling and path hardening. Nonprofit Cloud headers are normalised on the way into :func:nonprofit_cloud_gifts_to_features, not here.

Parameters:

Name Type Description Default
path str or Path

CSV file, or a directory of them.

 required

Returns:

Name Type Description
gifts DataFrame

The export as written, with text values.

Raises:

Type Description
FileNotFoundError

If path does not exist.
Source code in philanthropy/ingest/_nonprofit_cloud.py 
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
def read_nonprofit_cloud_gifts(path: Union[str, Path]) -> pd.DataFrame:
    """Read Nonprofit Cloud GiftTransaction export CSV(s) into one frame.

    Accepts a single ``.csv`` or a directory of them, walked recursively and
    concatenated in sorted relative-path order; symlinks are not followed.
    Every column is read as text and nothing is filtered: **unpaid and other
    non-paid transactions are still present**, and it is
    :func:`nonprofit_cloud_gifts_to_features` that drops them.

    Reading a gift export is CRM-agnostic once the headers are normalised,
    so this delegates to :func:`read_civicrm_contributions` rather than
    repeating its BOM handling and path hardening. Nonprofit Cloud headers
    are normalised on the way into
    :func:`nonprofit_cloud_gifts_to_features`, not here.

    Parameters
    ----------
    path : str or pathlib.Path
        CSV file, or a directory of them.

    Returns
    -------
    gifts : pandas.DataFrame
        The export as written, with text values.

    Raises
    ------
    FileNotFoundError
        If ``path`` does not exist.
    """
    return read_civicrm_contributions(path)

npsp_opportunities_to_features(opportunities, *, reference_date=None, include_stages=DEFAULT_INCLUDED_STAGES)

Aggregate an NPSP Opportunity export into donor-level features.

Only rows whose stage is a closed/won equivalent are kept, then handed to :func:civicrm_contributions_to_features, which produces the donor frame. NPSP's stage vocabulary carries no test-mode flag, so none is applied.

Parameters:

Name Type Description Default
opportunities iterable of mapping, or DataFrame

NPSP Opportunity export rows, under a report's column labels (Account Name or Primary Contact, Close Date, Amount, Stage), the raw API field names (AccountId, CloseDate, StageName), or NPSP's Data Import template headers (Donation Date, Donation Amount, Donation Stage). contact_id, receive_date and total_amount are required after normalisation; gift_type, financial_type, email, first_name and last_name are used when present. If both an Account column (Account Name, AccountId) and Primary Contact are present, the Account is used as the donor key, per NPSP's default Household Account model.

 required
reference_date str or datetime - like

Anchor for the recency features. If None, the latest Opportunity date in the batch is used, which keeps the aggregation free of "now" leakage.

 None
include_stages sequence of str or None

Allowlist of closed/won-equivalent stages to keep before aggregating, matched against gift_type (the normalised Stage column) ignoring case, spacing and punctuation. Every other named stage, including Pledged, open pipeline stages and Closed Lost, is dropped: an org's own stage vocabulary cannot be enumerated here, so an unrecognised stage is treated as not yet money rather than summed. None or an empty sequence disables the filter and sums every row, which double-counts an instalment recorded in both its Pledged and closed/won stages, and counts open pipeline that was never given. Rows whose stage is blank are kept either way: an unlabelled row cannot be shown to be anything but a gift.

 :data:`DEFAULT_INCLUDED_STAGES`

Returns:

Name Type Description
features DataFrame

One row per donor, indexed by contact_id, with the same columns :func:civicrm_contributions_to_features emits. distinct_financial_types counts distinct Record Types here.

Raises:

Type Description
KeyError

If contact_id, receive_date or total_amount is absent after normalisation.

Warns:

Type Description
UserWarning

If include_stages was requested but the export carries no stage column. Silently summing a Pledged instalment together with the Closed Won row recording its receipt (or an open pipeline row) is exactly the error this bridge exists to prevent, so it is worth a warning rather than a quiet wrong total.
Notes

Nothing here deduplicates on an Opportunity id, so concatenating two exports that overlap in time will double-count the overlap; export disjoint date ranges.

Examples:

>>> rows = [
...     {"Account ID": "88", "Close Date": "2025-01-10",
...      "Amount": "100.00", "Stage": "Pledged"},
...     {"Account ID": "88", "Close Date": "2025-02-10",
...      "Amount": "100.00", "Stage": "Closed Won"},
...     {"Account ID": "88", "Close Date": "2025-03-10",
...      "Amount": "100.00", "Stage": "Closed Won"},
... ]
>>> feats = npsp_opportunities_to_features(rows)
>>> float(feats.loc["88", "total_gift_amount"])  # not 300.0
200.0
>>> int(feats.loc["88", "gift_count"])
2
Source code in philanthropy/ingest/_npsp.py 
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
def npsp_opportunities_to_features(
    opportunities: Union[Iterable[Mapping], pd.DataFrame],
    *,
    reference_date: Optional[Union[str, pd.Timestamp]] = None,
    include_stages: Optional[Sequence[str]] = DEFAULT_INCLUDED_STAGES,
) -> pd.DataFrame:
    """Aggregate an NPSP Opportunity export into donor-level features.

    Only rows whose stage is a closed/won equivalent are kept, then handed to
    :func:`civicrm_contributions_to_features`, which produces the donor
    frame. NPSP's stage vocabulary carries no test-mode flag, so none is
    applied.

    Parameters
    ----------
    opportunities : iterable of mapping, or DataFrame
        NPSP Opportunity export rows, under a report's column labels
        (``Account Name`` or ``Primary Contact``, ``Close Date``, ``Amount``,
        ``Stage``), the raw API field names (``AccountId``, ``CloseDate``,
        ``StageName``), or NPSP's Data Import template headers (``Donation
        Date``, ``Donation Amount``, ``Donation Stage``). ``contact_id``,
        ``receive_date`` and ``total_amount`` are required after
        normalisation; ``gift_type``, ``financial_type``, ``email``,
        ``first_name`` and ``last_name`` are used when present. If both an
        Account column (``Account Name``, ``AccountId``) and ``Primary
        Contact`` are present, the Account is used as the donor key, per
        NPSP's default Household Account model.
    reference_date : str or datetime-like, optional
        Anchor for the recency features. If ``None``, the latest Opportunity
        date in the batch is used, which keeps the aggregation free of "now"
        leakage.
    include_stages : sequence of str or None, default=:data:`DEFAULT_INCLUDED_STAGES`
        Allowlist of closed/won-equivalent stages to keep before aggregating,
        matched against ``gift_type`` (the normalised Stage column) ignoring
        case, spacing and punctuation. Every other named stage, including
        ``Pledged``, open pipeline stages and ``Closed Lost``, is dropped:
        an org's own stage vocabulary cannot be enumerated here, so an
        unrecognised stage is treated as not yet money rather than summed.
        ``None`` or an empty sequence disables the filter and sums **every**
        row, which double-counts an instalment recorded in both its
        ``Pledged`` and closed/won stages, and counts open pipeline that was
        never given. Rows whose stage is blank are kept either way: an
        unlabelled row cannot be shown to be anything but a gift.

    Returns
    -------
    features : pandas.DataFrame
        One row per donor, indexed by ``contact_id``, with the same columns
        :func:`civicrm_contributions_to_features` emits.
        ``distinct_financial_types`` counts distinct Record Types here.

    Raises
    ------
    KeyError
        If ``contact_id``, ``receive_date`` or ``total_amount`` is absent
        after normalisation.

    Warns
    -----
    UserWarning
        If ``include_stages`` was requested but the export carries no stage
        column. Silently summing a ``Pledged`` instalment together with the
        ``Closed Won`` row recording its receipt (or an open pipeline row)
        is exactly the error this bridge exists to prevent, so it is worth a
        warning rather than a quiet wrong total.

    Notes
    -----
    Nothing here deduplicates on an Opportunity id, so concatenating two
    exports that overlap in time will double-count the overlap; export
    disjoint date ranges.

    Examples
    --------
    >>> rows = [
    ...     {"Account ID": "88", "Close Date": "2025-01-10",
    ...      "Amount": "100.00", "Stage": "Pledged"},
    ...     {"Account ID": "88", "Close Date": "2025-02-10",
    ...      "Amount": "100.00", "Stage": "Closed Won"},
    ...     {"Account ID": "88", "Close Date": "2025-03-10",
    ...      "Amount": "100.00", "Stage": "Closed Won"},
    ... ]
    >>> feats = npsp_opportunities_to_features(rows)
    >>> float(feats.loc["88", "total_gift_amount"])  # not 300.0
    200.0
    >>> int(feats.loc["88", "gift_count"])
    2
    """
    df = _normalise_headers(_prefer_account_donor_key(_to_frame(opportunities)), _canonical_npsp)
    if df.empty:
        return _empty_feature_frame()

    missing = [col for col in _REQUIRED if col not in df.columns]
    if missing:
        raise KeyError(
            f"NPSP Opportunity export is missing {missing}. Export the "
            f"'Account ID' (or 'Primary Contact'), 'Close Date' and 'Amount' "
            f"fields (Salesforce API: AccountId, CloseDate, Amount); got "
            f"{sorted(df.columns)}."
        )

    if include_stages:
        if "gift_type" in df.columns:
            wanted = {_stage_key(s) for s in include_stages}
            keys = df["gift_type"].map(_stage_key)
            df = df[(keys == "") | keys.isin(wanted)]
        else:
            warnings.warn(
                f"NPSP Opportunity export has no Stage column, so "
                f"include_stages={tuple(include_stages)!r} could not be "
                f"applied: a Pledged instalment or an open-pipeline row (if "
                f"any) is summed alongside the Closed Won row recording a "
                f"gift's receipt, which counts money that was never given. "
                f"Add the 'Stage' field to the export.",
                stacklevel=2,
            )

    if df.empty:
        return _empty_feature_frame()
    return civicrm_contributions_to_features(
        df, reference_date=reference_date, statuses=None
    )

read_npsp_opportunities(path)

Read NPSP Opportunity export CSV(s) into one frame.

Accepts a single .csv or a directory of them, walked recursively and concatenated in sorted relative-path order; symlinks are not followed. Every column is read as text and nothing is filtered: Pledged rows are still present, and it is :func:npsp_opportunities_to_features that drops them.

Reading a gift export is CRM-agnostic once the headers are normalised, so this delegates to :func:read_civicrm_contributions rather than repeating its BOM handling and path hardening. NPSP headers are normalised on the way into :func:npsp_opportunities_to_features, not here.

Parameters:

Name Type Description Default
path str or Path

CSV file, or a directory of them.

 required

Returns:

Name Type Description
opportunities DataFrame

The export as written, with text values.

Raises:

Type Description
FileNotFoundError

If path does not exist.
Source code in philanthropy/ingest/_npsp.py 
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
def read_npsp_opportunities(path: Union[str, Path]) -> pd.DataFrame:
    """Read NPSP Opportunity export CSV(s) into one frame.

    Accepts a single ``.csv`` or a directory of them, walked recursively and
    concatenated in sorted relative-path order; symlinks are not followed.
    Every column is read as text and nothing is filtered: **Pledged rows are
    still present**, and it is :func:`npsp_opportunities_to_features` that
    drops them.

    Reading a gift export is CRM-agnostic once the headers are normalised, so
    this delegates to :func:`read_civicrm_contributions` rather than repeating
    its BOM handling and path hardening. NPSP headers are normalised on the
    way into :func:`npsp_opportunities_to_features`, not here.

    Parameters
    ----------
    path : str or pathlib.Path
        CSV file, or a directory of them.

    Returns
    -------
    opportunities : pandas.DataFrame
        The export as written, with text values.

    Raises
    ------
    FileNotFoundError
        If ``path`` does not exist.
    """
    return read_civicrm_contributions(path)

raisers_edge_gifts_to_features(gifts, *, reference_date=None, exclude_gift_types=DEFAULT_EXCLUDED_GIFT_TYPES)

Aggregate a Raiser's Edge gift export into donor-level features.

Commitment rows (pledges, matching gift pledges, recurring gift templates) and ledger corrections are dropped first, then the surviving payment rows are handed to :func:civicrm_contributions_to_features, which produces the donor frame. Raiser's Edge has no contribution-status column, so no status filter is applied.

Parameters:

Name Type Description Default
gifts iterable of mapping, or DataFrame

Raiser's Edge gift rows, under the desktop export labels (Constituent ID, Gift Date, Gift Amount, Gift Type), the RE7 database columns (CONSTIT_ID, DTE, TYPE) or the RE NXT SKY API field names (constituent_id, date, amount, type). contact_id, receive_date and total_amount are required after normalisation; gift_type, fund, email, first_name and last_name are used when present.

 required
reference_date str or datetime - like

Anchor for the recency features. If None, the latest gift date in the batch is used, which keeps the aggregation free of "now" leakage.

 None
exclude_gift_types sequence of str or None

Gift types to drop before aggregating, matched against gift_type ignoring case, spacing and punctuation. None or an empty sequence disables the filter and sums every row, which double-counts pledged dollars. Rows whose gift type is blank are kept either way: an unlabelled row cannot be shown to be a commitment.

 :data:`DEFAULT_EXCLUDED_GIFT_TYPES`

Returns:

Name Type Description
features DataFrame

One row per donor, indexed by contact_id, with the same columns :func:civicrm_contributions_to_features emits. distinct_financial_types counts distinct Funds here.

Raises:

Type Description
KeyError

If contact_id, receive_date or total_amount is absent after normalisation.

Warns:

Type Description
UserWarning

If exclude_gift_types was requested but the export carries no gift type column. Silently summing pledges together with their payments is exactly the error this bridge exists to prevent, so it is worth a warning rather than a quiet wrong total.
Notes

A Raiser's Edge export of split gifts writes one row per split, all sharing the gift's system record id, and their amounts are meant to be summed. Nothing here deduplicates on a gift id for that reason, so concatenating two exports that overlap in time will double-count the overlap; export disjoint date ranges.

Examples:

>>> rows = [
...     {"Constituent ID": "88", "Gift Date": "2025-01-10",
...      "Gift Amount": "1200.00", "Gift Type": "Pledge"},
...     {"Constituent ID": "88", "Gift Date": "2025-02-10",
...      "Gift Amount": "100.00", "Gift Type": "Pay-Cash"},
...     {"Constituent ID": "88", "Gift Date": "2025-03-10",
...      "Gift Amount": "100.00", "Gift Type": "Pay-Cash"},
... ]
>>> feats = raisers_edge_gifts_to_features(rows)
>>> float(feats.loc["88", "total_gift_amount"])  # not 1400.0
200.0
>>> int(feats.loc["88", "gift_count"])
2
Source code in philanthropy/ingest/_raisers_edge.py 
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
def raisers_edge_gifts_to_features(
    gifts: Union[Iterable[Mapping], pd.DataFrame],
    *,
    reference_date: Optional[Union[str, pd.Timestamp]] = None,
    exclude_gift_types: Optional[Sequence[str]] = DEFAULT_EXCLUDED_GIFT_TYPES,
) -> pd.DataFrame:
    """Aggregate a Raiser's Edge gift export into donor-level features.

    Commitment rows (pledges, matching gift pledges, recurring gift templates)
    and ledger corrections are dropped first, then the surviving payment rows
    are handed to :func:`civicrm_contributions_to_features`, which produces the
    donor frame. Raiser's Edge has no contribution-status column, so no status
    filter is applied.

    Parameters
    ----------
    gifts : iterable of mapping, or DataFrame
        Raiser's Edge gift rows, under the desktop export labels
        (``Constituent ID``, ``Gift Date``, ``Gift Amount``, ``Gift Type``), the
        RE7 database columns (``CONSTIT_ID``, ``DTE``, ``TYPE``) or the RE NXT
        SKY API field names (``constituent_id``, ``date``, ``amount``,
        ``type``). ``contact_id``, ``receive_date`` and ``total_amount`` are
        required after normalisation; ``gift_type``, ``fund``, ``email``,
        ``first_name`` and ``last_name`` are used when present.
    reference_date : str or datetime-like, optional
        Anchor for the recency features. If ``None``, the latest gift date in
        the batch is used, which keeps the aggregation free of "now" leakage.
    exclude_gift_types : sequence of str or None, default=:data:`DEFAULT_EXCLUDED_GIFT_TYPES`
        Gift types to drop before aggregating, matched against ``gift_type``
        ignoring case, spacing and punctuation. ``None`` or an empty sequence
        disables the filter and sums **every** row, which double-counts pledged
        dollars. Rows whose gift type is blank are kept either way: an unlabelled
        row cannot be shown to be a commitment.

    Returns
    -------
    features : pandas.DataFrame
        One row per donor, indexed by ``contact_id``, with the same columns
        :func:`civicrm_contributions_to_features` emits.
        ``distinct_financial_types`` counts distinct Funds here.

    Raises
    ------
    KeyError
        If ``contact_id``, ``receive_date`` or ``total_amount`` is absent after
        normalisation.

    Warns
    -----
    UserWarning
        If ``exclude_gift_types`` was requested but the export carries no gift
        type column. Silently summing pledges together with their payments is
        exactly the error this bridge exists to prevent, so it is worth a
        warning rather than a quiet wrong total.

    Notes
    -----
    A Raiser's Edge export of *split* gifts writes one row per split, all
    sharing the gift's system record id, and their amounts are meant to be
    summed. Nothing here deduplicates on a gift id for that reason, so
    concatenating two exports that overlap in time will double-count the
    overlap; export disjoint date ranges.

    Examples
    --------
    >>> rows = [
    ...     {"Constituent ID": "88", "Gift Date": "2025-01-10",
    ...      "Gift Amount": "1200.00", "Gift Type": "Pledge"},
    ...     {"Constituent ID": "88", "Gift Date": "2025-02-10",
    ...      "Gift Amount": "100.00", "Gift Type": "Pay-Cash"},
    ...     {"Constituent ID": "88", "Gift Date": "2025-03-10",
    ...      "Gift Amount": "100.00", "Gift Type": "Pay-Cash"},
    ... ]
    >>> feats = raisers_edge_gifts_to_features(rows)
    >>> float(feats.loc["88", "total_gift_amount"])  # not 1400.0
    200.0
    >>> int(feats.loc["88", "gift_count"])
    2
    """
    df = _normalise_headers(_to_frame(gifts), _canonical_raisers_edge)
    if df.empty:
        return _empty_feature_frame()

    missing = [col for col in _REQUIRED if col not in df.columns]
    if missing:
        raise KeyError(
            f"Raiser's Edge gift export is missing {missing}. Export the "
            f"'Constituent ID', 'Gift Date' and 'Gift Amount' fields (SKY API: "
            f"constituent_id, date, amount); got {sorted(df.columns)}."
        )

    if exclude_gift_types:
        if "gift_type" in df.columns:
            unwanted = {_gift_type_key(t) for t in exclude_gift_types}
            keys = df["gift_type"].map(_gift_type_key)
            df = df[~keys.isin(unwanted)]
        else:
            warnings.warn(
                f"Raiser's Edge gift export has no gift type column, so "
                f"exclude_gift_types={tuple(exclude_gift_types)!r} could not be "
                f"applied: pledges and recurring gift templates (if any) are "
                f"summed alongside the payments made against them, which counts "
                f"every committed dollar twice. Add the 'Gift Type' field to the "
                f"export.",
                stacklevel=2,
            )

    if df.empty:
        return _empty_feature_frame()
    return civicrm_contributions_to_features(
        df, reference_date=reference_date, statuses=None
    )

read_raisers_edge_gifts(path)

Read Raiser's Edge gift export CSV(s) into one frame.

Accepts a single .csv or a directory of them, walked recursively and concatenated in sorted relative-path order; symlinks are not followed. Every column is read as text and nothing is filtered: commitment rows are still present, and it is :func:raisers_edge_gifts_to_features that drops them.

Reading a gift export is CRM-agnostic once the headers are normalised, so this delegates to :func:read_civicrm_contributions rather than repeating its BOM handling and path hardening. Raiser's Edge headers are normalised on the way into :func:raisers_edge_gifts_to_features, not here.

Parameters:

Name Type Description Default
path str or Path

CSV file, or a directory of them.

 required

Returns:

Name Type Description
gifts DataFrame

The export as written, with text values.

Raises:

Type Description
FileNotFoundError

If path does not exist.
Source code in philanthropy/ingest/_raisers_edge.py 
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
def read_raisers_edge_gifts(path: Union[str, Path]) -> pd.DataFrame:
    """Read Raiser's Edge gift export CSV(s) into one frame.

    Accepts a single ``.csv`` or a directory of them, walked recursively and
    concatenated in sorted relative-path order; symlinks are not followed. Every
    column is read as text and nothing is filtered: **commitment rows are still
    present**, and it is :func:`raisers_edge_gifts_to_features` that drops them.

    Reading a gift export is CRM-agnostic once the headers are normalised, so
    this delegates to :func:`read_civicrm_contributions` rather than repeating
    its BOM handling and path hardening. Raiser's Edge headers are normalised on
    the way into :func:`raisers_edge_gifts_to_features`, not here.

    Parameters
    ----------
    path : str or pathlib.Path
        CSV file, or a directory of them.

    Returns
    -------
    gifts : pandas.DataFrame
        The export as written, with text values.

    Raises
    ------
    FileNotFoundError
        If ``path`` does not exist.
    """
    return read_civicrm_contributions(path)

read_gifts(path_or_df, *, source, **kwargs)

Read (if given a path) and aggregate a CRM gift export in one call.

Looks source up in a small preset registry mapping each of "civicrm", "raisers_edge", "npsp", and "nonprofit_cloud" to that CRM's read_<source>_... loader and <source>_..._to_features aggregator, then runs the pair. read_gifts("gifts.csv", source="npsp") is equivalent to npsp_opportunities_to_features(read_npsp_opportunities("gifts.csv")).

Parameters:

Name Type Description Default
path_or_df str, pathlib.Path, DataFrame, or iterable of mapping

A gift export CSV file, a directory of them, or gift rows already in memory (a DataFrame or an iterable of mappings, e.g. an API result). A path is only meaningful for a single source's own export format; already-in-memory rows are handed straight to the aggregator.

 required
source str

Which CRM the export came from. One of :data:GIFT_SOURCES.

 required
**kwargs Any

Passed through to the source's aggregator, e.g. statuses= for "civicrm", exclude_gift_types= for "raisers_edge", include_stages= for "npsp", include_statuses= for "nonprofit_cloud", exclude_entry_types= for "bloomerang", exclude_record_types= for "donorperfect", or the reference_date= every preset accepts.

 {}

Returns:

Name Type Description
features DataFrame

One row per donor, indexed by contact_id, identical to calling the source's own reader and aggregator directly.

Raises:

Type Description
ValueError

If source is not one of :data:GIFT_SOURCES.

Examples:

>>> rows = [
...     {"Constituent ID": "88", "Gift Date": "2025-01-10",
...      "Gift Amount": "1200.00", "Gift Type": "Pledge"},
...     {"Constituent ID": "88", "Gift Date": "2025-02-10",
...      "Gift Amount": "100.00", "Gift Type": "Pay-Cash"},
... ]
>>> feats = read_gifts(rows, source="raisers_edge")
>>> float(feats.loc["88", "total_gift_amount"])  # the pledge is excluded
100.0
Source code in philanthropy/ingest/_read_gifts.py 
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
def read_gifts(
    path_or_df: Union[str, Path, pd.DataFrame, Iterable[Mapping]],
    *,
    source: str,
    **kwargs: Any,
) -> pd.DataFrame:
    """Read (if given a path) and aggregate a CRM gift export in one call.

    Looks ``source`` up in a small preset registry mapping each of
    ``"civicrm"``, ``"raisers_edge"``, ``"npsp"``, and ``"nonprofit_cloud"``
    to that CRM's ``read_<source>_...`` loader and ``<source>_..._to_features``
    aggregator, then runs the pair. ``read_gifts("gifts.csv", source="npsp")``
    is equivalent to
    ``npsp_opportunities_to_features(read_npsp_opportunities("gifts.csv"))``.

    Parameters
    ----------
    path_or_df : str, pathlib.Path, DataFrame, or iterable of mapping
        A gift export CSV file, a directory of them, or gift rows already in
        memory (a DataFrame or an iterable of mappings, e.g. an API result).
        A path is only meaningful for a single source's own export format;
        already-in-memory rows are handed straight to the aggregator.
    source : str
        Which CRM the export came from. One of :data:`GIFT_SOURCES`.
    **kwargs
        Passed through to the source's aggregator, e.g. ``statuses=`` for
        ``"civicrm"``, ``exclude_gift_types=`` for ``"raisers_edge"``,
        ``include_stages=`` for ``"npsp"``, ``include_statuses=`` for
        ``"nonprofit_cloud"``, ``exclude_entry_types=`` for ``"bloomerang"``,
        ``exclude_record_types=`` for ``"donorperfect"``, or the
        ``reference_date=`` every preset accepts.

    Returns
    -------
    features : pandas.DataFrame
        One row per donor, indexed by ``contact_id``, identical to calling
        the source's own reader and aggregator directly.

    Raises
    ------
    ValueError
        If ``source`` is not one of :data:`GIFT_SOURCES`.

    Examples
    --------
    >>> rows = [
    ...     {"Constituent ID": "88", "Gift Date": "2025-01-10",
    ...      "Gift Amount": "1200.00", "Gift Type": "Pledge"},
    ...     {"Constituent ID": "88", "Gift Date": "2025-02-10",
    ...      "Gift Amount": "100.00", "Gift Type": "Pay-Cash"},
    ... ]
    >>> feats = read_gifts(rows, source="raisers_edge")
    >>> float(feats.loc["88", "total_gift_amount"])  # the pledge is excluded
    100.0
    """
    try:
        read, to_features = _REGISTRY[source]
    except KeyError:
        raise ValueError(
            f"Unknown gift source {source!r}; expected one of {GIFT_SOURCES}."
        ) from None

    if isinstance(path_or_df, (str, Path)):
        path_or_df = read(path_or_df)
    return to_features(path_or_df, **kwargs)

build_leadership_snapshots(gifts, *, fiscal_years, threshold=1000, band=(100, 999), fiscal_year_start=7, activities=None, donors=None, include_momentum=False)

Build upgrade-candidate snapshots, one row per (donor, fiscal year).

For each fiscal year T in fiscal_years, the population is every donor whose FY T total giving falls inside band (inclusive) and below threshold: donors already at or above threshold in T are not "upgrade candidates" and get no row for that T. Every feature is computed from gift (and, if given, activity) rows dated at or before the end of FY T; target is 1 if the donor's FY T+1 total reaches threshold, else 0, including when the donor gave nothing at all in T+1.

Parameters:

Name Type Description Default
gifts iterable of mapping, or DataFrame

Gift-level rows, already normalised (not a raw CRM export): each row needs donor_id, gift_date and gift_amount, the same columns :class:~philanthropy.preprocessing.RFMTransformer and :func:~philanthropy.datasets.make_donor_panel use. Extra columns are ignored.

 required
fiscal_years iterable of int

The snapshot years T to build. One donor can appear once per year it qualified for the band, so passing several years stacks rows.

 required
threshold float or str

The leadership-giving level an upgrade crosses into. Also the upper exclusive bound of the candidate population: a donor at or above this in FY T already gave at that level and is not a candidate. A string "pNN" (e.g. "p93") is a percentile instead of a dollar amount: the NNth percentile of every positive donor FY total in the snapshot years fiscal_years (never their T+1 label years), so a file in small dollars and one in large dollars ask the same question. The resolved dollar value is in snapshots.attrs["threshold"].

 1000
band (float, float)

Inclusive (low, high) bounds on FY T total giving that define the upgrade-candidate population. high should sit below threshold for the two limits to describe one band; a donor is still excluded once the FY T total reaches threshold even if high does not.

 (100, 999)
fiscal_year_start int

Month (1-12) the organisation's fiscal year begins, the same parameter :class:~philanthropy.preprocessing.FiscalYearTransformer takes; fiscal-year boundaries are computed with that transformer so the two stay in agreement.

 7
activities iterable of mapping, or DataFrame

A long activity log in the shape :func:~philanthropy.ingest.activities_to_features expects (contact_id, activity_date, activity_type, ...). When given, that function is called once per snapshot year with as_of set to the end of FY T, and its columns are joined in.

 None
donors DataFrame

Static donor attributes (constituency, wealth rating, ...), indexed by donor id. Joined in as-is; also passed through to activities_to_features for its match-rate check when activities is given. The caller is responsible for not including a column that already encodes the answer (e.g. a precomputed giving tier); this function does not attempt to detect that.

 None
include_momentum bool

Also emit, for each of fy_total, gift_count and largest_gift, a trailing 3- and 5-year OLS slope and relative slope (and, when activities is given, the same for each activity type's count/hours/amount), plus fy_total_growth_ratio. Off by default so existing callers, including the shipped :func:~philanthropy.models.score_leadership_prospects model, see an unchanged feature set; opt in explicitly once you want the model trained on momentum too. See :func:philanthropy.utils._momentum.trailing_slope_features.

 False

Returns:

Name Type Description
snapshots DataFrame

One row per qualifying (donor, T), indexed by donor_id, sorted by fiscal_year then donor_id. Columns: fiscal_year (the snapshot year T, not T+1); fy_total, fy_total_prior1, fy_total_prior2 (summed gift amount in FY T, T-1, T-2); fy_trend (fy_total - fy_total_prior1); fy_total_growth_ratio ((fy_total + 1) / (fy_total_prior1 + 1)); largest_gift and gift_count (within FY T); consecutive_years_given (count of unbroken prior fiscal years, ending at and including T, with positive giving); months_since_last_gift (from the donor's most recent gift on or before the end of FY T); when include_momentum=True, for each of fy_total, gift_count and largest_gift, a trailing 3- and 5-year OLS slope and relative slope (<base>_slope_3y, <base>_rel_slope_3y, <base>_slope_5y, <base>_rel_slope_5y; NaN with fewer than 2 observed years in the window, see :func:philanthropy.utils._momentum.trailing_slope_features); any activities_to_features columns, including its own momentum columns when include_momentum=True; any donors columns; and target. A fiscal year with no qualifying donors contributes no rows. Returns an empty, columnless frame (index name donor_id) if no year has any.

Raises:

Type Description
KeyError

If gifts is missing donor_id, gift_date or gift_amount.
ValueError

If fiscal_year_start is not between 1 and 12, band[0] > band[1], or threshold is a string that is not "pNN" with 0 < NN < 100.

Examples:

>>> import pandas as pd
>>> from philanthropy.ingest import build_leadership_snapshots
>>> gifts = pd.DataFrame({
...     "donor_id": ["1", "1", "1", "1", "2"],
...     "gift_date": ["2017-08-01", "2018-08-01", "2019-08-01",
...                   "2020-08-01", "2019-08-01"],
...     "gift_amount": [200, 300, 500, 1500, 5000],
... })
>>> snaps = build_leadership_snapshots(gifts, fiscal_years=[2020])
>>> list(snaps.index)
['1']
>>> int(snaps.loc["1", "fiscal_year"])
2020
>>> float(snaps.loc["1", "fy_total"])
500.0
>>> int(snaps.loc["1", "consecutive_years_given"])
3
>>> int(snaps.loc["1", "target"])
1

Donor "2" gave 5000 in FY2020, at or above threshold, so is excluded from the band even though it never appears in snaps:

>>> "2" in snaps.index
False
Source code in philanthropy/ingest/_upgrade_snapshots.py 
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
def build_leadership_snapshots(
    gifts: Union[Iterable[Mapping], pd.DataFrame],
    *,
    fiscal_years: Iterable[int],
    threshold: Union[float, str] = 1000,
    band: Tuple[float, float] = (100, 999),
    fiscal_year_start: int = 7,
    activities: Optional[Union[Iterable[Mapping], pd.DataFrame]] = None,
    donors: Optional[pd.DataFrame] = None,
    include_momentum: bool = False,
) -> pd.DataFrame:
    """Build upgrade-candidate snapshots, one row per (donor, fiscal year).

    For each fiscal year ``T`` in ``fiscal_years``, the population is every
    donor whose FY T total giving falls inside ``band`` (inclusive) and below
    ``threshold``: donors already at or above ``threshold`` in T are not
    "upgrade candidates" and get no row for that T. Every feature is computed
    from gift (and, if given, activity) rows dated at or before the end of FY
    T; ``target`` is 1 if the donor's FY T+1 total reaches ``threshold``, else
    0, including when the donor gave nothing at all in T+1.

    Parameters
    ----------
    gifts : iterable of mapping, or DataFrame
        Gift-level rows, already normalised (not a raw CRM export): each row
        needs ``donor_id``, ``gift_date`` and ``gift_amount``, the same
        columns :class:`~philanthropy.preprocessing.RFMTransformer` and
        :func:`~philanthropy.datasets.make_donor_panel` use. Extra columns
        are ignored.
    fiscal_years : iterable of int
        The snapshot years ``T`` to build. One donor can appear once per year
        it qualified for the band, so passing several years stacks rows.
    threshold : float or str, default=1000
        The leadership-giving level an upgrade crosses into. Also the upper
        exclusive bound of the candidate population: a donor at or above this
        in FY T already gave at that level and is not a candidate. A string
        ``"pNN"`` (e.g. ``"p93"``) is a percentile instead of a dollar
        amount: the NNth percentile of every positive donor FY total in the
        snapshot years ``fiscal_years`` (never their T+1 label years), so a
        file in small dollars and one in large dollars ask the same
        question. The resolved dollar value is in ``snapshots.attrs["threshold"]``.
    band : (float, float), default=(100, 999)
        Inclusive ``(low, high)`` bounds on FY T total giving that define the
        upgrade-candidate population. ``high`` should sit below ``threshold``
        for the two limits to describe one band; a donor is still excluded
        once the FY T total reaches ``threshold`` even if ``high`` does not.
    fiscal_year_start : int, default=7
        Month (1-12) the organisation's fiscal year begins, the same
        parameter :class:`~philanthropy.preprocessing.FiscalYearTransformer`
        takes; fiscal-year boundaries are computed with that transformer so
        the two stay in agreement.
    activities : iterable of mapping, or DataFrame, optional
        A long activity log in the shape
        :func:`~philanthropy.ingest.activities_to_features` expects
        (``contact_id``, ``activity_date``, ``activity_type``, ...). When
        given, that function is called once per snapshot year with
        ``as_of`` set to the end of FY T, and its columns are joined in.
    donors : DataFrame, optional
        Static donor attributes (constituency, wealth rating, ...), indexed
        by donor id. Joined in as-is; also passed through to
        ``activities_to_features`` for its match-rate check when
        ``activities`` is given. The caller is responsible for not including
        a column that already encodes the answer (e.g. a precomputed giving
        tier); this function does not attempt to detect that.
    include_momentum : bool, default=False
        Also emit, for each of ``fy_total``, ``gift_count`` and
        ``largest_gift``, a trailing 3- and 5-year OLS slope and relative
        slope (and, when ``activities`` is given, the same for each activity
        type's count/hours/amount), plus ``fy_total_growth_ratio``. Off by
        default so existing callers, including the shipped
        :func:`~philanthropy.models.score_leadership_prospects` model, see an
        unchanged feature set; opt in explicitly once you want the model
        trained on momentum too. See
        :func:`philanthropy.utils._momentum.trailing_slope_features`.

    Returns
    -------
    snapshots : pandas.DataFrame
        One row per qualifying ``(donor, T)``, indexed by ``donor_id``, sorted
        by ``fiscal_year`` then ``donor_id``. Columns: ``fiscal_year`` (the
        snapshot year T, not T+1); ``fy_total``, ``fy_total_prior1``,
        ``fy_total_prior2`` (summed gift amount in FY T, T-1, T-2); ``fy_trend``
        (``fy_total - fy_total_prior1``); ``fy_total_growth_ratio``
        (``(fy_total + 1) / (fy_total_prior1 + 1)``); ``largest_gift`` and
        ``gift_count`` (within FY T); ``consecutive_years_given`` (count of
        unbroken prior fiscal years, ending at and including T, with positive
        giving); ``months_since_last_gift`` (from the donor's most recent
        gift on or before the end of FY T); when ``include_momentum=True``,
        for each of ``fy_total``, ``gift_count`` and ``largest_gift``, a
        trailing 3- and 5-year OLS slope and relative slope
        (``<base>_slope_3y``, ``<base>_rel_slope_3y``, ``<base>_slope_5y``,
        ``<base>_rel_slope_5y``; NaN with fewer than 2 observed years in the
        window, see :func:`philanthropy.utils._momentum.trailing_slope_features`);
        any ``activities_to_features`` columns, including its own momentum
        columns when ``include_momentum=True``; any ``donors`` columns; and
        ``target``. A fiscal year with no
        qualifying donors contributes no rows. Returns an empty, columnless
        frame (index name ``donor_id``) if no year has any.

    Raises
    ------
    KeyError
        If ``gifts`` is missing ``donor_id``, ``gift_date`` or
        ``gift_amount``.
    ValueError
        If ``fiscal_year_start`` is not between 1 and 12, ``band[0] >
        band[1]``, or ``threshold`` is a string that is not ``"pNN"`` with
        ``0 < NN < 100``.

    Examples
    --------
    >>> import pandas as pd
    >>> from philanthropy.ingest import build_leadership_snapshots
    >>> gifts = pd.DataFrame({
    ...     "donor_id": ["1", "1", "1", "1", "2"],
    ...     "gift_date": ["2017-08-01", "2018-08-01", "2019-08-01",
    ...                   "2020-08-01", "2019-08-01"],
    ...     "gift_amount": [200, 300, 500, 1500, 5000],
    ... })
    >>> snaps = build_leadership_snapshots(gifts, fiscal_years=[2020])
    >>> list(snaps.index)
    ['1']
    >>> int(snaps.loc["1", "fiscal_year"])
    2020
    >>> float(snaps.loc["1", "fy_total"])
    500.0
    >>> int(snaps.loc["1", "consecutive_years_given"])
    3
    >>> int(snaps.loc["1", "target"])
    1

    Donor "2" gave 5000 in FY2020, at or above ``threshold``, so is excluded
    from the band even though it never appears in ``snaps``:

    >>> "2" in snaps.index
    False
    """
    validate_fiscal_year_start(fiscal_year_start)
    low, high = band
    if low > high:
        raise ValueError(f"band[0] ({low}) must be <= band[1] ({high}).")

    df, pivot_sum, pivot_max, pivot_count = _prepare_gifts(gifts, fiscal_year_start)
    fiscal_years = [int(fy) for fy in fiscal_years]
    threshold = _resolve_threshold(
        threshold, pivot_sum.reindex(columns=fiscal_years).to_numpy().ravel()
    )

    donors_norm = None
    if donors is not None:
        donors_norm = donors.copy()
        donors_norm.index = donors_norm.index.astype("string").str.strip()

    snapshots = []
    for fy_t in fiscal_years:
        fy_t = int(fy_t)
        totals_t = _column(pivot_sum, fy_t)
        candidates = totals_t[(totals_t >= low) & (totals_t <= high) & (totals_t < threshold)]
        if candidates.empty:
            continue
        donor_ids = candidates.index

        snap = _snapshot_features_for_year(
            df, pivot_sum, pivot_max, pivot_count, donor_ids, fy_t,
            fiscal_year_start, activities, donors, donors_norm,
            include_momentum=include_momentum,
        )

        next_totals = _column(pivot_sum, fy_t + 1).reindex(donor_ids, fill_value=0.0)
        snap["target"] = (next_totals >= threshold).astype("int64")

        snapshots.append(snap)

    if not snapshots:
        empty = pd.DataFrame(index=pd.Index([], name="donor_id", dtype="object"))
        empty.attrs["threshold"] = threshold
        return empty

    out = pd.concat(snapshots)
    out = out.reset_index().sort_values(["fiscal_year", "donor_id"], kind="stable")
    out = out.set_index("donor_id")
    out.attrs["threshold"] = threshold
    return out

build_snapshots(gifts, *, kind, fiscal_years=None, threshold=1000.0, band=(100.0, 999.0), min_years_given=1, fiscal_year_start=7, scale_free=False)

Build labelled donor x fiscal-year snapshots for one question.

Every feature is computed from gifts dated at or before the end of fiscal year T; the label alone is read from T + 1. A fiscal year with no following year in the file contributes no rows, so "gave nothing next year" and "next year not in the file" are never confused.

Parameters:

Name Type Description Default
gifts iterable of mapping, or DataFrame

Gift-level rows with donor_id, gift_date and gift_amount.

 required
kind ('upgrade', 'lapse', 'response_next_year', 'next_amount')

Which population and label (see :func:period_snapshots).

 "upgrade"
fiscal_years iterable of int

Snapshot years T to keep. Default: every year with a following year in the file.

 None
threshold (float, (float, float))

The upgrade question's leadership level and candidate band, as in :func:~philanthropy.ingest.build_leadership_snapshots. Ignored by the other kinds.

 1000.0
band (float, (float, float))

The upgrade question's leadership level and candidate band, as in :func:~philanthropy.ingest.build_leadership_snapshots. Ignored by the other kinds.

 1000.0
min_years_given int

Keep only donors who gave in at least this many fiscal years up to and including T. 2 restricts lapse to multi-year donors, the donors a retention program can act on.

 1
fiscal_year_start int

Month (1-12) the fiscal year begins.

 7
scale_free bool

Also emit :data:SCALE_FREE_COLUMNS.

 False

Returns:

Name Type Description
snapshots DataFrame

Indexed by donor_id, sorted by period then donor_id: period (the fiscal year T), :data:CORE_COLUMNS, gift_count (gifts in T), months_since_last_gift (to the end of T), the scale-free columns when asked, and target.

Examples:

>>> import pandas as pd
>>> from philanthropy.ingest import build_snapshots
>>> gifts = pd.DataFrame({
...     "donor_id": ["1", "1", "2", "2", "2"],
...     "gift_date": ["2018-08-01", "2019-08-01", "2018-08-01",
...                   "2019-08-01", "2020-08-01"],
...     "gift_amount": [50, 60, 20, 30, 40],
... })
>>> snaps = build_snapshots(gifts, kind="lapse", min_years_given=2)
>>> snaps["period"].tolist(), snaps["target"].tolist()
([2020, 2020], [1, 0])

Donor "1" gave in FY2019 and FY2020 and nothing in FY2021, so it is a lapse; donor "2" gave again. FY2019 has no rows: neither donor had two giving years by then.

Source code in philanthropy/ingest/_snapshots.py 
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
def build_snapshots(
    gifts: Union[Iterable[Mapping], pd.DataFrame],
    *,
    kind: str,
    fiscal_years: Optional[Iterable[int]] = None,
    threshold: float = 1000.0,
    band: Tuple[float, float] = (100.0, 999.0),
    min_years_given: int = 1,
    fiscal_year_start: int = 7,
    scale_free: bool = False,
) -> pd.DataFrame:
    """Build labelled donor x fiscal-year snapshots for one question.

    Every feature is computed from gifts dated at or before the end of
    fiscal year ``T``; the label alone is read from ``T + 1``. A fiscal year
    with no following year in the file contributes no rows, so "gave
    nothing next year" and "next year not in the file" are never confused.

    Parameters
    ----------
    gifts : iterable of mapping, or DataFrame
        Gift-level rows with ``donor_id``, ``gift_date`` and ``gift_amount``.
    kind : {"upgrade", "lapse", "response_next_year", "next_amount"}
        Which population and label (see :func:`period_snapshots`).
    fiscal_years : iterable of int, optional
        Snapshot years ``T`` to keep. Default: every year with a following
        year in the file.
    threshold, band : float, (float, float)
        The upgrade question's leadership level and candidate band, as in
        :func:`~philanthropy.ingest.build_leadership_snapshots`. Ignored by
        the other kinds.
    min_years_given : int, default=1
        Keep only donors who gave in at least this many fiscal years up to
        and including ``T``. ``2`` restricts lapse to multi-year donors, the
        donors a retention program can act on.
    fiscal_year_start : int, default=7
        Month (1-12) the fiscal year begins.
    scale_free : bool, default=False
        Also emit :data:`SCALE_FREE_COLUMNS`.

    Returns
    -------
    snapshots : pandas.DataFrame
        Indexed by ``donor_id``, sorted by ``period`` then ``donor_id``:
        ``period`` (the fiscal year T), :data:`CORE_COLUMNS`,
        ``gift_count`` (gifts in T), ``months_since_last_gift`` (to the end
        of T), the scale-free columns when asked, and ``target``.

    Examples
    --------
    >>> import pandas as pd
    >>> from philanthropy.ingest import build_snapshots
    >>> gifts = pd.DataFrame({
    ...     "donor_id": ["1", "1", "2", "2", "2"],
    ...     "gift_date": ["2018-08-01", "2019-08-01", "2018-08-01",
    ...                   "2019-08-01", "2020-08-01"],
    ...     "gift_amount": [50, 60, 20, 30, 40],
    ... })
    >>> snaps = build_snapshots(gifts, kind="lapse", min_years_given=2)
    >>> snaps["period"].tolist(), snaps["target"].tolist()
    ([2020, 2020], [1, 0])

    Donor "1" gave in FY2019 and FY2020 and nothing in FY2021, so it is a
    lapse; donor "2" gave again. FY2019 has no rows: neither donor had two
    giving years by then.
    """
    validate_fiscal_year_start(fiscal_year_start)
    df, pivot_sum, pivot_max, pivot_count = _prepare_gifts(gifts, fiscal_year_start)
    if df.empty:
        return pd.DataFrame(index=pd.Index([], name="donor_id"))
    # Fiscal years are contiguous: a year with no gift at all on file is
    # still a period in which every donor gave $0.
    years = list(range(int(df["_fy"].min()), int(df["_fy"].max()) + 1))
    totals = pivot_sum.reindex(columns=years, fill_value=0.0).fillna(0.0)
    largest = pivot_max.reindex(columns=years, fill_value=0.0).fillna(0.0)
    counts = pivot_count.reindex(columns=years, fill_value=0).fillna(0)

    snap = period_snapshots(
        totals, kind=kind, largest=largest, counts=counts, periods=fiscal_years,
        threshold=threshold, band=band, min_years_given=min_years_given, scale_free=scale_free,
    )
    if snap.empty:
        return snap

    last_date = {}
    for fy in snap["period"].unique():
        fy = int(fy)
        last_date[fy] = (
            (_fy_end(fy, fiscal_year_start) - df[df["_fy"] <= fy].groupby("donor_id")["_date"].max()).dt.days / 30.0
        ).round(2)
    snap["months_since_last_gift"] = [
        last_date[int(p)].get(d, np.nan) for d, p in zip(snap.index, snap["period"])
    ]
    target = snap.pop("target")
    snap["target"] = target
    return snap