terraflow.drought¶

Impact-labeled drought-loss prediction benchmark. Predicts insured drought loss (USDA RMA Cause of Loss) from within-season climate/vegetation anomalies. See the Drought benchmark guide for the task, splits, and reference results.

Configuration¶

`config` ¶

Configuration model for the drought-impact benchmark.

The benchmark predicts insured drought loss (USDA RMA Cause of Loss) from within-season climate/vegetation predictors. This module defines the single Pydantic config that drives the whole build so runs are reproducible and fingerprintable.

`DroughtConfig` ¶

Bases: BaseModel

End-to-end configuration for building and evaluating the drought-impact benchmark.

`state_alphas` `property` ¶

USPS postal codes for the configured FIPS states (for the NASS query).

`from_yaml(path)` `classmethod` ¶

Load a :class:DroughtConfig from a YAML file.

Source code in terraflow/drought/config.py

@classmethod
def from_yaml(cls, path: Path) -> "DroughtConfig":
    """Load a :class:`DroughtConfig` from a YAML file."""
    data = yaml.safe_load(Path(path).read_text(encoding="utf-8")) or {}
    return cls(**data)

RMA Cause of Loss ingest¶

`rma` ¶

Ingest USDA RMA Cause of Loss Summary-of-Business files.

The Cause of Loss Historical Data Files are public, pipe-delimited flat files (one ZIP per year, 1989–present) published at https://www.rma.usda.gov/tools-reports/summary-of-business/cause-loss

Each record is broken down by year × state × county × commodity × insurance-plan × coverage × stage × cause-of-loss × month-of-loss. The 30-field record layout is stable across the benchmark's year range (verified 2000 & 2012 → exactly 30 fields). Fields are space-padded within the pipe delimiters and numerics are zero-/dot-padded (e.g. 00.0000000000), so string fields are stripped and numeric fields coerced.

`col_url(year)` ¶

Public download URL for a given commodity year's Cause of Loss ZIP.

Source code in terraflow/drought/rma.py

def col_url(year: int) -> str:
    """Public download URL for a given commodity year's Cause of Loss ZIP."""
    return f"{BASE_URL}/colsom_{year}.zip"

`download_col_years(years, dest_dir, *, overwrite=False)` ¶

Download (and cache) the Cause of Loss ZIP for each year into dest_dir.

Returns the list of local ZIP paths. Existing files are reused unless overwrite=True.

Source code in terraflow/drought/rma.py

def download_col_years(years: list[int], dest_dir: Path, *, overwrite: bool = False) -> list[Path]:
    """Download (and cache) the Cause of Loss ZIP for each year into *dest_dir*.

    Returns the list of local ZIP paths. Existing files are reused unless ``overwrite=True``.
    """
    dest_dir = Path(dest_dir)
    dest_dir.mkdir(parents=True, exist_ok=True)
    paths: list[Path] = []
    for year in years:
        target = dest_dir / f"colsom_{year}.zip"
        if overwrite or not target.exists():
            urllib.request.urlretrieve(col_url(year), target)  # noqa: S310 (trusted USDA host)
        paths.append(target)
    return paths

`load_col(rma_dir, years, *, states=None, commodity=None)` ¶

Load and concatenate Cause of Loss records for years from rma_dir.

Accepts either colsom_YYYY.zip or the extracted colsomYY.txt in rma_dir. states / commodity filters are pushed down into each file for speed.

Source code in terraflow/drought/rma.py

def load_col(
    rma_dir: Path,
    years: list[int],
    *,
    states: list[str] | None = None,
    commodity: str | None = None,
) -> pd.DataFrame:
    """Load and concatenate Cause of Loss records for *years* from *rma_dir*.

    Accepts either ``colsom_YYYY.zip`` or the extracted ``colsomYY.txt`` in *rma_dir*. ``states`` /
    ``commodity`` filters are pushed down into each file for speed.
    """
    rma_dir = Path(rma_dir)
    frames: list[pd.DataFrame] = []
    for year in years:
        candidates = [
            rma_dir / f"colsom_{year}.zip",
            rma_dir / f"colsom{year % 100:02d}.txt",
            rma_dir / f"colsom_{year}.txt",
        ]
        path = next((c for c in candidates if c.exists()), None)
        if path is None:
            raise FileNotFoundError(
                f"No Cause of Loss file for {year} in {rma_dir} "
                f"(expected one of: {', '.join(c.name for c in candidates)})"
            )
        frames.append(parse_col_file(path, states=states, commodity=commodity))
    return pd.concat(frames, ignore_index=True)

`parse_col_file(path, *, states=None, commodity=None)` ¶

Parse a single Cause of Loss file (.txt or .zip) into a tidy DataFrame.

String fields are stripped; numeric fields are coerced. Adds a 5-digit GEOID (state FIPS + county FIPS) matching the flashdry predictor tables. Optional states / commodity filters are applied before the (costly) strip/convert of every column, which keeps whole-corpus loads fast when only one crop/region is needed.

Source code in terraflow/drought/rma.py

def parse_col_file(
    path: Path,
    *,
    states: list[str] | None = None,
    commodity: str | None = None,
) -> pd.DataFrame:
    """Parse a single Cause of Loss file (``.txt`` or ``.zip``) into a tidy DataFrame.

    String fields are stripped; numeric fields are coerced. Adds a 5-digit ``GEOID`` (state FIPS +
    county FIPS) matching the flashdry predictor tables. Optional ``states`` / ``commodity`` filters
    are applied *before* the (costly) strip/convert of every column, which keeps whole-corpus loads
    fast when only one crop/region is needed.
    """
    handle = _open_col_text(path)
    try:
        df = pd.read_csv(
            handle,
            sep="|",
            header=None,
            names=list(COL_COLUMNS),
            dtype=str,
            encoding="latin-1",
            na_filter=False,
        )
    finally:
        handle.close()

    # Cheap early filter on just the two key columns to shrink the frame before full processing.
    if states is not None:
        df = df[df["state_code"].str.strip().str.zfill(2).isin(states)]
    if commodity is not None:
        df = df[df["commodity_name"].str.strip() == commodity]
    df = df.copy()

    for col in COL_COLUMNS:
        df[col] = df[col].str.strip()
    for col in _NUMERIC_COLUMNS:
        df[col] = pd.to_numeric(df[col], errors="coerce")

    df["state_code"] = df["state_code"].str.zfill(2)
    df["county_code"] = df["county_code"].str.zfill(3)
    df["GEOID"] = df["state_code"] + df["county_code"]
    return df

Labels¶

`labels` ¶

Build county-crop-year drought-loss labels from parsed RMA Cause of Loss records.

The impact label is what makes this benchmark distinct from severity (USDM D2+) and yield (CY-Bench/SustainBench) benchmarks: it is the insured drought loss actually paid out.

:func:build_labels aggregates Cause of Loss into per-(GEOID, year) numerators (drought/total indemnity) plus a fallback loss-experience liability. :func:finalize_targets then computes the primary targets once the true total insured liability (from the Summary-of-Business coverage file) and planted acres are joined:

drought_indemnity = Σ indemnity where cause_of_loss_desc ∈ drought_causes
total_indemnity = Σ indemnity over all causes
drought_share = drought_indemnity / total_indemnity (clean, in [0, 1])
drought_loss_ratio = drought_indemnity / total insured liability (primary regression target)
significant_drought_loss = drought_loss_ratio ≥ threshold (binary target)
insured_acre_fraction = insured acres / NASS planted acres (coverage-bias column)

Using the Summary-of-Business total liability (all policies) as the denominator removes the >1 artifact of the Cause-of-Loss loss-experience liability (only loss-experiencing policies).

`build_labels(col, cfg)` ¶

Aggregate parsed Cause of Loss records into per-(GEOID, year) numerators.

Produces drought/total indemnity, the loss-experience (Cause-of-Loss) liability as a fallback denominator col_liability, and the bounded drought_share. The primary drought_loss_ratio / significant_drought_loss targets are computed by :func:finalize_targets once the true Summary-of-Business total liability is joined.

Source code in terraflow/drought/labels.py

def build_labels(col: pd.DataFrame, cfg: DroughtConfig) -> pd.DataFrame:
    """Aggregate parsed Cause of Loss records into per-(GEOID, year) numerators.

    Produces drought/total indemnity, the loss-experience (Cause-of-Loss) liability as a fallback
    denominator ``col_liability``, and the bounded ``drought_share``. The primary
    ``drought_loss_ratio`` / ``significant_drought_loss`` targets are computed by
    :func:`finalize_targets` once the true Summary-of-Business total liability is joined.
    """
    scoped = col[
        col["state_code"].isin(cfg.states)
        & (col["commodity_name"] == cfg.crop)
        & col["commodity_year"].between(cfg.year_min, cfg.year_max)
    ].copy()

    scoped["_is_drought"] = scoped["cause_of_loss_desc"].isin(cfg.drought_causes)
    scoped["_drought_indemnity"] = np.where(scoped["_is_drought"], scoped["indemnity_amount"], 0.0)

    grouped = scoped.groupby(["GEOID", "commodity_year"], as_index=False).agg(
        drought_indemnity=("_drought_indemnity", "sum"),
        total_indemnity=("indemnity_amount", "sum"),
        col_liability=("liability", "sum"),
    )
    grouped = grouped.rename(columns={"commodity_year": "year"})
    grouped["drought_share"] = _safe_ratio(grouped["drought_indemnity"], grouped["total_indemnity"])
    grouped["year"] = grouped["year"].astype(int)
    return grouped[list(LABEL_COLUMNS)].sort_values(["GEOID", "year"]).reset_index(drop=True)

`finalize_targets(df, cfg)` ¶

Compute the final targets on the fully-joined benchmark frame.

Uses the true Summary-of-Business total_liability as the drought-loss-ratio denominator when present (removing the loss-experience >1 artifact), else falls back to col_liability. Adds the insured_acre_fraction coverage-bias column when both insured and planted acres are available.

Source code in terraflow/drought/labels.py

def finalize_targets(df: pd.DataFrame, cfg: DroughtConfig) -> pd.DataFrame:
    """Compute the final targets on the fully-joined benchmark frame.

    Uses the true Summary-of-Business ``total_liability`` as the drought-loss-ratio denominator when
    present (removing the loss-experience >1 artifact), else falls back to ``col_liability``. Adds the
    ``insured_acre_fraction`` coverage-bias column when both insured and planted acres are available.
    """
    out = df.copy()
    liability = out["total_liability"] if "total_liability" in out.columns else out["col_liability"]
    out["drought_loss_ratio"] = _safe_ratio(out["drought_indemnity"], liability)
    out["significant_drought_loss"] = out["drought_loss_ratio"] >= cfg.loss_ratio_threshold
    if "insured_acres" in out.columns and "planted_acres" in out.columns:
        planted = out["planted_acres"].to_numpy(dtype=float)
        insured = out["insured_acres"].to_numpy(dtype=float)
        # Missing/withheld planted acres -> unknown coverage (NaN), not a spurious 0.0.
        out["insured_acre_fraction"] = np.divide(insured, planted, out=np.full(len(out), np.nan), where=planted > 0)
    return out

Predictors¶

`predictors` ¶

Aggregate flashdry within-season predictors to one vector per (GEOID, crop-year).

The flashdry feature_table.parquet is sub-annual (growing-season, ~biweekly). For an early-warning task we aggregate each base feature over the season up to cutoff_doy — so a model only sees signal available by, e.g., end of July — into {mean, min, max, last} summaries.

Two feature families are tagged so baselines can select them: - climate/vegetation — the 30 deseasonalized *_anom features + NDVI_anom_z - severity — dm_gte_d2 / dm_class (USDM drought severity). Aggregating these lets us build a severity-only baseline and quantify how much within-season climate signal adds on top of the (strong) USDM severity indicator for predicting realized drought loss.

`aggregate_predictors(feature_table, cfg)` ¶

Aggregate within-season predictors up to cfg.cutoff_doy per (GEOID, year).

Returns one row per (GEOID, year) with {mean, min, max, last} of every climate and severity feature, plus n_obs (observations before the cutoff) and n_stress_weeks (vegetation ≤ −1 SD). last is the value at the largest DOY at or before the cutoff.

Source code in terraflow/drought/predictors.py

def aggregate_predictors(feature_table: pd.DataFrame, cfg: DroughtConfig) -> pd.DataFrame:
    """Aggregate within-season predictors up to ``cfg.cutoff_doy`` per (GEOID, year).

    Returns one row per (GEOID, year) with {mean, min, max, last} of every climate and severity
    feature, plus ``n_obs`` (observations before the cutoff) and ``n_stress_weeks`` (vegetation
    ≤ −1 SD). ``last`` is the value at the largest DOY at or before the cutoff.
    """
    df = feature_table[
        (feature_table["doy"] <= cfg.cutoff_doy)
        & feature_table["year"].between(cfg.year_min, cfg.year_max)
        & feature_table["STATEFP"].isin(cfg.states)
    ].copy()
    df = df.sort_values(["GEOID", "year", "doy"])

    bases = list(CLIMATE_FEATURES + SEVERITY_FEATURES)
    agg_spec = {b: list(_STATS) for b in bases}
    grouped = df.groupby(["GEOID", "year"]).agg(agg_spec)
    grouped.columns = [f"{base}_{stat}" for base, stat in grouped.columns]

    extras = df.groupby(["GEOID", "year"]).agg(
        n_obs=("doy", "size"),
        n_stress_weeks=("NDVI_anom_z", lambda s: int((s <= -1.0).sum())),
    )
    out = grouped.join(extras).reset_index()
    out["year"] = out["year"].astype(int)
    return out

Splits¶

`splits` ¶

Reproducible evaluation splits for the drought-impact benchmark.

Three complementary regimes, all deterministic from the config (no RNG): - temporal — held-out years (incl. the 2012 extreme + recent 2022/2023) test extrapolation to unseen seasons. - spatial — leave-one-state-out blocks test spatial transfer (guards against spatially autocorrelated leakage; finer lon/lat grids are a follow-up). - loyo — leave-one-year-out folds for a stability estimate across all years.

Split definitions (not enumerated rows) are serialized so the artifact is small and the masks are re-derivable from any benchmark table.