diff --git a/CHANGELOG.md b/CHANGELOG.md
index 060303c..4916455 100644
--- a/CHANGELOG.md
+++ b/CHANGELOG.md
@@ -25,6 +25,17 @@ Format: [Keep a Changelog](https://keepachangelog.com/en/1.1.0/)
   `<type>_hours_12m` / `<type>_amount_12m` when those columns are present),
   cut at `as_of` so nothing dated after the cutoff is counted. A new activity
   type never needs new model code; it just yields its own columns.
+- `philanthropy.ingest.build_upgrade_snapshots(gifts, *, fiscal_years,
+  threshold=1000, band=(100, 999), fiscal_year_start=7, activities=None,
+  donors=None)`: builds a per-donor, per-fiscal-year training table for an
+  upgrade model, one row per donor whose fiscal-year-T giving lands in the
+  upgrade band, with gift-derived features (prior-year totals, trend,
+  largest gift, gift count, consecutive years given, months since last
+  gift), optional joined activity and donor-attribute columns, and a
+  `target` reading whether the donor crossed `threshold` in fiscal year T+1.
+  Everything but `target` is computed from data through the end of T; the
+  output's `fiscal_year` and donor-id columns feed directly into
+  `FiscalYearGroupedSplitter`.
 
 ## [0.8.0] - 2026-09-24
 
diff --git a/docs/reference/index.md b/docs/reference/index.md
index f97571a..367857c 100755
--- a/docs/reference/index.md
+++ b/docs/reference/index.md
@@ -58,6 +58,7 @@ Everything reachable from `philanthropy.__all__` is listed below. A symbol not l
 | `npsp_opportunities_to_features`, `read_npsp_opportunities`, `DEFAULT_EXCLUDED_STAGES` | `ingest` | Tracks NPSP Opportunity export labels and a per-org-configurable stage vocabulary; the excluded-stage default will grow as real exports arrive. |
 | `map_columns` | `ingest` | The one-error-per-missing-column message shape may still change. |
 | `activities_to_features` | `ingest` | The activity-type feature set (`_count_12m`, `_distinct`, ...) may grow as more source types are onboarded. |
+| `build_upgrade_snapshots` | `ingest` | The gift-derived feature set (trend, consecutive years given, ...) is a minimal starting recipe and is likely to be refined. |
 | `plot_affinity_distribution`, `plot_retention_waterfall` | `visualisation` | Chart composition is presentation, not contract. |
 | `fetch_kdd98_donors` | `datasets` | Returns the raw upstream columns untyped; may gain as-of date parsing as the real-data leakage replication in #124 lands. |
 | `make_donor_panel` | `datasets` | The returned dict may gain keys (pledges, appeals, soft credits) as more of the library needs panel-shaped fixtures; existing keys and their columns will not change silently. |
diff --git a/philanthropy/ingest/__init__.py b/philanthropy/ingest/__init__.py
index 4862102..41c6444 100644
--- a/philanthropy/ingest/__init__.py
+++ b/philanthropy/ingest/__init__.py
@@ -30,6 +30,12 @@
 (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.
+
+``build_upgrade_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.
 """
 
 from ._activities import activities_to_features
@@ -52,11 +58,13 @@
     raisers_edge_gifts_to_features,
     read_raisers_edge_gifts,
 )
+from ._upgrade_snapshots import build_upgrade_snapshots
 
 __all__ = [
     "DEFAULT_EXCLUDED_GIFT_TYPES",
     "DEFAULT_EXCLUDED_STAGES",
     "activities_to_features",
+    "build_upgrade_snapshots",
     "civicrm_contributions_to_features",
     "constituent_events_to_features",
     "map_columns",
diff --git a/philanthropy/ingest/_upgrade_snapshots.py b/philanthropy/ingest/_upgrade_snapshots.py
new file mode 100644
index 0000000..1b25935
--- /dev/null
+++ b/philanthropy/ingest/_upgrade_snapshots.py
@@ -0,0 +1,297 @@
+"""
+philanthropy.ingest._upgrade_snapshots
+=======================================
+Build a per-donor, per-fiscal-year training table for an upgrade model: will
+a donor currently giving in a mid-level band move up to a leadership level
+next year?
+
+``build_upgrade_snapshots`` turns a gift log (optionally joined with an
+activity log and static donor attributes) into one row per
+``(donor, fiscal year T)`` for every donor whose FY T giving falls inside the
+"upgrade band", plus a ``target`` column: did that donor cross ``threshold``
+in FY T+1? Everything a row's features touch is dated at or before the end
+of FY T; the target alone is read from FY T+1, and nothing later.
+
+This is a plain function, not an estimator: it changes the row count (one
+gift log row becomes zero or one donor-year row) and manufactures a label
+column, neither of which an ``sklearn`` ``transform()`` can do. It lives in
+``philanthropy.ingest`` because it is the same shape as every other bridge in
+this subpackage, raw/normalised source tables in, one donor-level feature
+table out, and it reuses :func:`~philanthropy.ingest.activities_to_features`
+directly when an activity log is supplied.
+"""
+
+from __future__ import annotations
+
+from typing import Iterable, Mapping, Optional, Tuple, Union
+
+import pandas as pd
+
+from philanthropy.preprocessing import FiscalYearTransformer
+from philanthropy.utils._validation import validate_fiscal_year_start
+
+from ._activities import activities_to_features
+
+__all__ = ["build_upgrade_snapshots"]
+
+_REQUIRED = ("donor_id", "gift_date", "gift_amount")
+
+
+def build_upgrade_snapshots(
+    gifts: Union[Iterable[Mapping], pd.DataFrame],
+    *,
+    fiscal_years: Iterable[int],
+    threshold: float = 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,
+) -> 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, 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.
+    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.
+
+    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``); ``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); any ``activities_to_features`` columns;
+        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, or ``band[0] >
+        band[1]``.
+
+    Examples
+    --------
+    >>> import pandas as pd
+    >>> from philanthropy.ingest import build_upgrade_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_upgrade_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 = _to_frame(gifts)
+    missing = [col for col in _REQUIRED if col not in df.columns]
+    if missing:
+        raise KeyError(
+            f"gifts is missing {missing}. Every gift row needs 'donor_id', "
+            f"'gift_date' and 'gift_amount'; got {sorted(df.columns)}."
+        )
+
+    df = df.copy()
+    df["donor_id"] = df["donor_id"].astype("string").str.strip()
+    df["_date"] = pd.to_datetime(df["gift_date"], errors="coerce")
+    df["_amount"] = pd.to_numeric(df["gift_amount"], errors="coerce")
+    df = df[df["donor_id"].notna() & df["_date"].notna() & df["_amount"].notna()]
+
+    if not df.empty:
+        fy = (
+            FiscalYearTransformer(fiscal_year_start=fiscal_year_start)
+            .set_output(transform="pandas")
+            .fit_transform(df[["_date"]].rename(columns={"_date": "gift_date"}))
+        )
+        df["_fy"] = fy["fiscal_year"].to_numpy()
+        df = df[df["_fy"].notna()]
+        df["_fy"] = df["_fy"].astype("int64")
+
+    pivot_sum = _pivot(df, "sum")
+    pivot_max = _pivot(df, "max")
+    pivot_count = _pivot(df, "count")
+
+    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 = pd.DataFrame(index=donor_ids)
+        snap.index.name = "donor_id"
+        snap["fiscal_year"] = fy_t
+        snap["fy_total"] = candidates
+        snap["fy_total_prior1"] = _column(pivot_sum, fy_t - 1).reindex(donor_ids, fill_value=0.0)
+        snap["fy_total_prior2"] = _column(pivot_sum, fy_t - 2).reindex(donor_ids, fill_value=0.0)
+        snap["fy_trend"] = snap["fy_total"] - snap["fy_total_prior1"]
+        snap["largest_gift"] = _column(pivot_max, fy_t).reindex(donor_ids, fill_value=0.0)
+        snap["gift_count"] = _column(pivot_count, fy_t).reindex(donor_ids, fill_value=0).astype("int64")
+        snap["consecutive_years_given"] = _consecutive_years_given(pivot_sum, donor_ids, fy_t)
+
+        fy_end = _fy_end(fy_t, fiscal_year_start)
+        last_gift = df[df["_fy"] <= fy_t].groupby("donor_id")["_date"].max().reindex(donor_ids)
+        snap["months_since_last_gift"] = ((fy_end - last_gift).dt.days / 30.0).round(2)
+
+        if activities is not None:
+            act_feats = activities_to_features(activities, as_of=fy_end, donors=donors)
+            snap = snap.join(act_feats, how="left")
+
+        if donors_norm is not None:
+            snap = snap.join(donors_norm, how="left")
+
+        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:
+        return pd.DataFrame(index=pd.Index([], name="donor_id", dtype="object"))
+
+    out = pd.concat(snapshots)
+    out = out.reset_index().sort_values(["fiscal_year", "donor_id"], kind="stable")
+    return out.set_index("donor_id")
+
+
+# --------------------------------------------------------------------------- #
+# Internals
+# --------------------------------------------------------------------------- #
+def _to_frame(gifts: Union[Iterable[Mapping], pd.DataFrame]) -> pd.DataFrame:
+    if isinstance(gifts, pd.DataFrame):
+        return gifts
+    return pd.DataFrame(list(gifts))
+
+
+def _pivot(df: pd.DataFrame, aggfunc: str) -> pd.DataFrame:
+    """Donor x fiscal-year table of ``aggfunc`` over gift amounts.
+
+    Empty when there is no valid gift row; a fiscal year with no data at all
+    is simply absent as a column, never a column of zeros.
+    """
+    if df.empty:
+        return pd.DataFrame()
+    fill = 0
+    return (
+        df.groupby(["donor_id", "_fy"])["_amount"]
+        .agg(aggfunc)
+        .unstack(fill_value=fill)
+    )
+
+
+def _column(pivot: pd.DataFrame, fy: int) -> pd.Series:
+    """The pivot's column for ``fy``, or an empty float series if absent."""
+    if fy not in pivot.columns:
+        return pd.Series(dtype="float64")
+    return pivot[fy]
+
+
+def _consecutive_years_given(
+    pivot_sum: pd.DataFrame, donor_ids: pd.Index, fy_t: int
+) -> pd.Series:
+    """Unbroken run of fiscal years with positive giving, ending at ``fy_t``.
+
+    A plain per-donor walk backwards from ``fy_t``: donor counts are small
+    (one row per candidate) and streak lengths are bounded by the years on
+    file, so this stays a simple loop rather than a vectorised scan.
+    """
+    counts = []
+    for donor_id in donor_ids:
+        streak = 0
+        year = fy_t
+        while True:
+            total = float(_column(pivot_sum, year).get(donor_id, 0.0))
+            if total <= 0:
+                break
+            streak += 1
+            year -= 1
+        counts.append(streak)
+    return pd.Series(counts, index=donor_ids, dtype="int64")
+
+
+def _fy_end(fy: int, fiscal_year_start: int) -> pd.Timestamp:
+    """The last calendar day of fiscal year ``fy``.
+
+    Matches :class:`~philanthropy.preprocessing.FiscalYearTransformer`'s
+    convention (``fiscal_year = year + 1`` once the month reaches
+    ``fiscal_year_start``): FY ``fy`` ends the day before ``fiscal_year_start``
+    rolls over in calendar year ``fy``.
+    """
+    return pd.Timestamp(year=fy, month=fiscal_year_start, day=1) - pd.Timedelta(days=1)
diff --git a/tests/test_upgrade_snapshots.py b/tests/test_upgrade_snapshots.py
new file mode 100644
index 0000000..b5e57af
--- /dev/null
+++ b/tests/test_upgrade_snapshots.py
@@ -0,0 +1,368 @@
+"""
+tests/test_upgrade_snapshots.py
+Tests for philanthropy.ingest.build_upgrade_snapshots.
+
+The population for a snapshot year T is the "upgrade band": donors giving
+between band[0] and band[1] in FY T, and strictly below threshold even if
+band[1] reaches or exceeds it. Every feature is computed from data through
+the end of FY T; target alone reads FY T+1. Most tests below are built
+around proving that boundary never moves.
+"""
+
+import numpy as np
+import pandas as pd
+import pytest
+
+from philanthropy.ingest import build_upgrade_snapshots
+from philanthropy.model_selection import FiscalYearGroupedSplitter
+
+# fiscal_year_start=7 (the default): FY N spans N-1's July 1 to N's June 30.
+FY2018 = "2017-08-01"  # -> fiscal_year 2018
+FY2019 = "2018-08-01"  # -> fiscal_year 2019
+FY2020 = "2019-08-01"  # -> fiscal_year 2020
+FY2021 = "2020-08-01"  # -> fiscal_year 2021
+FY2022 = "2021-08-01"  # -> fiscal_year 2022
+
+
+def _gifts(rows):
+    """rows: iterable of (donor_id, date, amount)."""
+    return pd.DataFrame(rows, columns=["donor_id", "gift_date", "gift_amount"])
+
+
+# --------------------------------------------------------------------------- #
+# Band population
+# --------------------------------------------------------------------------- #
+def test_donor_inside_band_qualifies():
+    gifts = _gifts([("1", FY2020, 500)])
+    snaps = build_upgrade_snapshots(gifts, fiscal_years=[2020])
+    assert list(snaps.index) == ["1"]
+
+
+def test_donor_below_band_excluded():
+    gifts = _gifts([("1", FY2020, 50)])
+    snaps = build_upgrade_snapshots(gifts, fiscal_years=[2020])
+    assert snaps.empty
+
+
+def test_band_bounds_are_inclusive():
+    gifts = _gifts([("1", FY2020, 100), ("2", FY2020, 999)])
+    snaps = build_upgrade_snapshots(gifts, fiscal_years=[2020])
+    assert set(snaps.index) == {"1", "2"}
+
+
+def test_donor_at_threshold_excluded():
+    gifts = _gifts([("1", FY2020, 1000)])
+    snaps = build_upgrade_snapshots(gifts, fiscal_years=[2020])
+    assert snaps.empty
+
+
+def test_donor_above_threshold_excluded():
+    gifts = _gifts([("1", FY2020, 5000)])
+    snaps = build_upgrade_snapshots(gifts, fiscal_years=[2020])
+    assert snaps.empty
+
+
+def test_threshold_excludes_even_when_band_is_wider():
+    # band[1] alone would admit 1200; threshold must still exclude it.
+    gifts = _gifts([("1", FY2020, 999), ("2", FY2020, 1200), ("3", FY2020, 1500)])
+    snaps = build_upgrade_snapshots(
+        gifts, fiscal_years=[2020], threshold=1000, band=(100, 1500)
+    )
+    assert set(snaps.index) == {"1"}
+
+
+def test_donor_with_no_gifts_at_all_never_appears():
+    gifts = _gifts([("1", FY2020, 500)])
+    snaps = build_upgrade_snapshots(gifts, fiscal_years=[2020])
+    assert "ghost" not in snaps.index
+
+
+# --------------------------------------------------------------------------- #
+# fiscal_year column and multi-year stacking
+# --------------------------------------------------------------------------- #
+def test_fiscal_year_column_is_the_snapshot_year_not_t_plus_1():
+    gifts = _gifts([("1", FY2020, 500)])
+    snaps = build_upgrade_snapshots(gifts, fiscal_years=[2020])
+    assert int(snaps.loc["1", "fiscal_year"]) == 2020
+
+
+def test_multiple_fiscal_years_stack_rows_for_the_same_donor():
+    gifts = _gifts([("1", FY2020, 500), ("1", FY2021, 300)])
+    snaps = build_upgrade_snapshots(gifts, fiscal_years=[2020, 2021])
+    assert list(snaps.index) == ["1", "1"]
+    assert sorted(snaps["fiscal_year"].tolist()) == [2020, 2021]
+
+
+def test_year_with_no_candidates_contributes_no_rows():
+    gifts = _gifts([("1", FY2020, 500)])
+    snaps = build_upgrade_snapshots(gifts, fiscal_years=[2019, 2020])
+    assert list(snaps["fiscal_year"]) == [2020]
+
+
+def test_every_gift_row_unparseable_returns_empty_frame():
+    gifts = _gifts([("1", "not-a-date", 500)])
+    snaps = build_upgrade_snapshots(gifts, fiscal_years=[2020])
+    assert snaps.empty
+    assert snaps.index.name == "donor_id"
+
+
+def test_no_year_has_any_candidate_returns_empty_frame():
+    gifts = _gifts([("1", FY2020, 5000)])
+    snaps = build_upgrade_snapshots(gifts, fiscal_years=[2019, 2020])
+    assert snaps.empty
+    assert snaps.index.name == "donor_id"
+
+
+def test_rows_sorted_by_fiscal_year_then_donor_id():
+    gifts = _gifts([("9", FY2021, 500), ("1", FY2021, 500), ("1", FY2020, 500)])
+    snaps = build_upgrade_snapshots(gifts, fiscal_years=[2020, 2021])
+    assert list(zip(snaps["fiscal_year"], snaps.index)) == [
+        (2020, "1"),
+        (2021, "1"),
+        (2021, "9"),
+    ]
+
+
+# --------------------------------------------------------------------------- #
+# Target: reads FY T+1 only
+# --------------------------------------------------------------------------- #
+def test_target_1_when_next_year_reaches_threshold():
+    gifts = _gifts([("1", FY2020, 500), ("1", FY2021, 1500)])
+    snaps = build_upgrade_snapshots(gifts, fiscal_years=[2020])
+    assert int(snaps.loc["1", "target"]) == 1
+
+
+def test_target_0_when_next_year_stays_below_threshold():
+    gifts = _gifts([("1", FY2020, 500), ("1", FY2021, 600)])
+    snaps = build_upgrade_snapshots(gifts, fiscal_years=[2020])
+    assert int(snaps.loc["1", "target"]) == 0
+
+
+def test_target_0_when_no_gift_at_all_in_next_year():
+    gifts = _gifts([("1", FY2020, 500)])
+    snaps = build_upgrade_snapshots(gifts, fiscal_years=[2020])
+    assert int(snaps.loc["1", "target"]) == 0
+    assert not pd.isna(snaps.loc["1", "target"])
+
+
+def test_target_reads_t_plus_1_and_nothing_further():
+    # A huge gift two years out must not affect target for T.
+    gifts = _gifts([("1", FY2020, 500), ("1", FY2022, 100_000)])
+    snaps = build_upgrade_snapshots(gifts, fiscal_years=[2020])
+    assert int(snaps.loc["1", "target"]) == 0
+
+
+def test_appending_a_t_plus_1_gift_can_flip_target():
+    base = _gifts([("1", FY2020, 500)])
+    before = build_upgrade_snapshots(base, fiscal_years=[2020])
+    after_gifts = pd.concat([base, _gifts([("1", FY2021, 1500)])], ignore_index=True)
+    after = build_upgrade_snapshots(after_gifts, fiscal_years=[2020])
+
+    assert int(before.loc["1", "target"]) == 0
+    assert int(after.loc["1", "target"]) == 1
+    # Every feature column (everything but target) is untouched by the new
+    # FY T+1 row: only the FY T+1-derived target may change.
+    feature_cols = [c for c in before.columns if c != "target"]
+    pd.testing.assert_frame_equal(before[feature_cols], after[feature_cols])
+
+
+# --------------------------------------------------------------------------- #
+# Leakage: features never see FY T+1 or later
+# --------------------------------------------------------------------------- #
+def test_features_unaffected_by_size_of_t_plus_1_gift():
+    small = _gifts([("1", FY2020, 500), ("1", FY2021, 1)])
+    large = _gifts([("1", FY2020, 500), ("1", FY2021, 999_999)])
+    snaps_small = build_upgrade_snapshots(small, fiscal_years=[2020])
+    snaps_large = build_upgrade_snapshots(large, fiscal_years=[2020])
+
+    feature_cols = [c for c in snaps_small.columns if c != "target"]
+    pd.testing.assert_frame_equal(
+        snaps_small[feature_cols], snaps_large[feature_cols]
+    )
+
+
+def test_features_unaffected_by_rows_appended_at_t_plus_2_or_later():
+    before = _gifts([("1", FY2020, 500)])
+    after = pd.concat(
+        [before, _gifts([("1", FY2022, 100_000)])], ignore_index=True
+    )
+    snaps_before = build_upgrade_snapshots(before, fiscal_years=[2020])
+    snaps_after = build_upgrade_snapshots(after, fiscal_years=[2020])
+
+    pd.testing.assert_frame_equal(snaps_before, snaps_after)
+
+
+def test_idempotent_on_repeat():
+    gifts = _gifts(
+        [("1", FY2018, 200), ("1", FY2019, 300), ("1", FY2020, 500), ("1", FY2021, 1500)]
+    )
+    first = build_upgrade_snapshots(gifts, fiscal_years=[2020])
+    second = build_upgrade_snapshots(gifts, fiscal_years=[2020])
+    pd.testing.assert_frame_equal(first, second)
+
+
+# --------------------------------------------------------------------------- #
+# Gift-derived features
+# --------------------------------------------------------------------------- #
+def test_fy_totals_and_trend():
+    gifts = _gifts(
+        [("1", FY2018, 200), ("1", FY2019, 300), ("1", FY2020, 500)]
+    )
+    snaps = build_upgrade_snapshots(gifts, fiscal_years=[2020])
+    row = snaps.loc["1"]
+
+    assert float(row["fy_total"]) == 500.0
+    assert float(row["fy_total_prior1"]) == 300.0
+    assert float(row["fy_total_prior2"]) == 200.0
+    assert float(row["fy_trend"]) == 200.0
+
+
+def test_prior_year_totals_default_to_zero_without_history():
+    gifts = _gifts([("1", FY2020, 500)])
+    snaps = build_upgrade_snapshots(gifts, fiscal_years=[2020])
+    row = snaps.loc["1"]
+
+    assert float(row["fy_total_prior1"]) == 0.0
+    assert float(row["fy_total_prior2"]) == 0.0
+
+
+def test_largest_gift_and_gift_count():
+    gifts = _gifts([("1", FY2020, 300), ("1", "2020-01-15", 200)])
+    snaps = build_upgrade_snapshots(gifts, fiscal_years=[2020])
+    row = snaps.loc["1"]
+
+    assert float(row["fy_total"]) == 500.0
+    assert float(row["largest_gift"]) == 300.0
+    assert int(row["gift_count"]) == 2
+
+
+def test_consecutive_years_given():
+    gifts = _gifts(
+        [("1", FY2018, 200), ("1", FY2019, 300), ("1", FY2020, 500)]
+    )
+    snaps = build_upgrade_snapshots(gifts, fiscal_years=[2020])
+    assert int(snaps.loc["1", "consecutive_years_given"]) == 3
+
+
+def test_consecutive_years_given_breaks_on_a_gap():
+    # No gift at all in FY2019 breaks the streak before FY2018.
+    gifts = _gifts([("1", FY2018, 200), ("1", FY2020, 500)])
+    snaps = build_upgrade_snapshots(gifts, fiscal_years=[2020])
+    assert int(snaps.loc["1", "consecutive_years_given"]) == 1
+
+
+def test_months_since_last_gift():
+    gifts = _gifts([("1", FY2019, 300), ("1", FY2020, 500)])
+    snaps = build_upgrade_snapshots(gifts, fiscal_years=[2020])
+    # Last gift on or before end of FY2020 (2020-06-30) is the FY2020 gift
+    # itself, dated 2019-08-01.
+    expected_days = (pd.Timestamp("2020-06-30") - pd.Timestamp(FY2020)).days
+    assert float(snaps.loc["1", "months_since_last_gift"]) == round(
+        expected_days / 30.0, 2
+    )
+
+
+# --------------------------------------------------------------------------- #
+# activities integration
+# --------------------------------------------------------------------------- #
+def test_activities_columns_are_joined_in():
+    gifts = _gifts([("1", FY2020, 500)])
+    activities = [
+        {"contact_id": "1", "activity_date": FY2020, "activity_type": "event"},
+    ]
+    snaps = build_upgrade_snapshots(gifts, fiscal_years=[2020], activities=activities)
+    assert "event_count_12m" in snaps.columns
+
+
+def test_activities_cutoff_matches_end_of_fiscal_year_t():
+    gifts = _gifts([("1", FY2020, 500)])
+    # This event is dated in FY2021, after the end of FY2020: must not be
+    # counted in the FY2020 snapshot's activity features.
+    activities = [
+        {"contact_id": "1", "activity_date": FY2021, "activity_type": "event"},
+    ]
+    snaps = build_upgrade_snapshots(gifts, fiscal_years=[2020], activities=activities)
+    assert "event_count_12m" not in snaps.columns
+
+
+def test_donor_missing_from_activities_gets_nan_activity_columns():
+    gifts = _gifts([("1", FY2020, 500), ("2", FY2020, 200)])
+    activities = [
+        {"contact_id": "1", "activity_date": FY2020, "activity_type": "event"},
+    ]
+    snaps = build_upgrade_snapshots(gifts, fiscal_years=[2020], activities=activities)
+    assert pd.isna(snaps.loc["2", "event_count_12m"])
+
+
+# --------------------------------------------------------------------------- #
+# donors integration
+# --------------------------------------------------------------------------- #
+def test_donor_attribute_columns_are_joined_in():
+    gifts = _gifts([("1", FY2020, 500)])
+    donors = pd.DataFrame({"wealth_rating": ["A"]}, index=pd.Index(["1"], name="donor_id"))
+    snaps = build_upgrade_snapshots(gifts, fiscal_years=[2020], donors=donors)
+    assert snaps.loc["1", "wealth_rating"] == "A"
+
+
+def test_donors_does_not_add_rows_for_non_candidates():
+    gifts = _gifts([("1", FY2020, 500)])
+    donors = pd.DataFrame(
+        {"wealth_rating": ["A", "B"]}, index=pd.Index(["1", "2"], name="donor_id")
+    )
+    snaps = build_upgrade_snapshots(gifts, fiscal_years=[2020], donors=donors)
+    assert list(snaps.index) == ["1"]
+
+
+# --------------------------------------------------------------------------- #
+# FiscalYearGroupedSplitter integration
+# --------------------------------------------------------------------------- #
+def test_output_feeds_fiscal_year_grouped_splitter():
+    # Three donors, each qualifying in exactly one distinct fiscal year, so
+    # every test fold's donor is unseen in training and drop_repeat_donors's
+    # default guard against an emptied fold never fires.
+    gifts = _gifts(
+        [("A", FY2020, 500), ("B", FY2021, 500), ("C", FY2022, 500)]
+    )
+    snaps = build_upgrade_snapshots(gifts, fiscal_years=[2020, 2021, 2022])
+    assert len(snaps) == 3
+
+    groups = snaps.reset_index()[["fiscal_year", "donor_id"]].to_numpy()
+    splitter = FiscalYearGroupedSplitter(n_splits=2)
+    X = np.zeros((len(snaps), 1))
+
+    splits = list(splitter.split(X, groups=groups))
+    assert len(splits) == 2
+    for train_idx, test_idx in splits:
+        assert len(test_idx) > 0
+        assert len(train_idx) > 0
+        train_fy = snaps["fiscal_year"].to_numpy()[train_idx]
+        test_fy = snaps["fiscal_year"].to_numpy()[test_idx]
+        assert train_fy.max() < test_fy.min()
+
+
+# --------------------------------------------------------------------------- #
+# Validation
+# --------------------------------------------------------------------------- #
+def test_missing_required_column_raises():
+    gifts = pd.DataFrame({"donor_id": ["1"], "gift_date": [FY2020]})
+    with pytest.raises(KeyError, match="gift_amount"):
+        build_upgrade_snapshots(gifts, fiscal_years=[2020])
+
+
+def test_invalid_fiscal_year_start_raises():
+    gifts = _gifts([("1", FY2020, 500)])
+    with pytest.raises(ValueError):
+        build_upgrade_snapshots(gifts, fiscal_years=[2020], fiscal_year_start=13)
+
+
+def test_band_low_above_high_raises():
+    gifts = _gifts([("1", FY2020, 500)])
+    with pytest.raises(ValueError):
+        build_upgrade_snapshots(gifts, fiscal_years=[2020], band=(999, 100))
+
+
+def test_dataframe_and_iterable_of_mappings_both_accepted():
+    rows = [{"donor_id": "1", "gift_date": FY2020, "gift_amount": 500}]
+    from_df = build_upgrade_snapshots(pd.DataFrame(rows), fiscal_years=[2020])
+    from_iter = build_upgrade_snapshots(rows, fiscal_years=[2020])
+    pd.testing.assert_frame_equal(from_df, from_iter)