From 48000b0dcad95f7f78e81fdfb686bae12f2523c5 Mon Sep 17 00:00:00 2001
From: Graham McIntire <graham@mcintire.me>
Date: Mon, 4 May 2026 16:18:17 -0500
Subject: [PATCH] feat(pskr): weekly recalibration analysis on the calibration
 corpus
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

`Pskr.Recalibrator.run/0` reads `pskr_calibration_samples`, bins
each sample per (band × feature), and writes spot-density stats to
`pskr_feature_bins` so an operator can read whether a feature
actually discriminates propagation at the threshold granularity
the scorer uses.

Bins, not regression: `BandConfig` already encodes scoring as
discrete thresholds, so the bin output matches that shape and an
operator can copy adjusted thresholds directly without translating
from regression coefficients.

Self-healing: corpus too thin ⇒ run row written with status
`skipped_insufficient_data` and the analysis is a no-op until next
fire. `min_total_samples = 1000` (≈ 4-5 days of CONUS PSKR
activity); per-band threshold is 100. Both surface in the run row's
`notes`.

Auto-applies nothing. Weight changes still go through human review
of `BandConfig.@band_configs` and a code commit. The recalibrator
is a read-only analyst that stays out of the production scoring
path.

Features binned (matching the scorer's discriminating fields):
  * pwat_mm — humidity U-shape candidate
  * hpbl_m — boundary layer (mechanism vs scoring re-eval)
  * min_refractivity_gradient — refractivity threshold validation
  * surface_pressure_mb — pressure-front proxy
  * kp_index — aurora boost magnitude tuning

Schema: two tables.
  * `pskr_recalibration_runs` — one row per fire with corpus
    stats, status, notes
  * `pskr_feature_bins` — one row per (run, band, feature, bin)
    with sample_count, spot_count_total/avg/p50/p90

Cron: `0 4 * * 0` (Sundays 04:00 UTC, off-peak, post-climatology).
Manual reruns enqueue with no args.

Tests cover the empty-corpus skip path, sub-threshold totals,
per-band threshold gating, the actual bin emission, nil-feature
handling, spot-count averaging, and the always-records-a-run
audit invariant. 8 new tests, 3282 total passing.

Backfill pipeline untouched.
---
 config/runtime.exs                            |  10 +-
 lib/microwaveprop/pskr/feature_bin.ex         |  58 ++++
 lib/microwaveprop/pskr/recalibration_run.ex   |  54 ++++
 lib/microwaveprop/pskr/recalibrator.ex        | 265 ++++++++++++++++++
 .../workers/pskr_recalibration_worker.ex      |  35 +++
 ...11415_create_pskr_recalibration_tables.exs |  61 ++++
 test/microwaveprop/pskr/recalibrator_test.exs | 163 +++++++++++
 .../pskr_recalibration_worker_test.exs        |  16 ++
 8 files changed, 661 insertions(+), 1 deletion(-)
 create mode 100644 lib/microwaveprop/pskr/feature_bin.ex
 create mode 100644 lib/microwaveprop/pskr/recalibration_run.ex
 create mode 100644 lib/microwaveprop/pskr/recalibrator.ex
 create mode 100644 lib/microwaveprop/workers/pskr_recalibration_worker.ex
 create mode 100644 priv/repo/migrations/20260504211415_create_pskr_recalibration_tables.exs
 create mode 100644 test/microwaveprop/pskr/recalibrator_test.exs
 create mode 100644 test/microwaveprop/workers/pskr_recalibration_worker_test.exs

diff --git a/config/runtime.exs b/config/runtime.exs
index e51abed8..8eb460a3 100644
--- a/config/runtime.exs
+++ b/config/runtime.exs
@@ -316,7 +316,15 @@ if config_env() == :prod do
        # aggregator flush, so the previous-hour sample has the
        # atmospheric and signal data it needs. Idempotent UPSERT, so
        # missed fires are safe to backfill manually.
-       {"25 * * * *", Microwaveprop.Workers.PskrCalibrationWorker}
+       {"25 * * * *", Microwaveprop.Workers.PskrCalibrationWorker},
+       # Weekly recalibration analysis pass. The recalibrator
+       # checks corpus density and bins each scoring feature ×
+       # band when the corpus is large enough; otherwise records
+       # a `skipped_insufficient_data` run. Sundays 04:00 UTC sits
+       # off-peak and after the nightly climatology rebuild.
+       # Auto-applies nothing — weight changes still go through
+       # human review of `BandConfig`.
+       {"0 4 * * 0", Microwaveprop.Workers.PskrRecalibrationWorker}
      ]}
 
   base_plugins = [
diff --git a/lib/microwaveprop/pskr/feature_bin.ex b/lib/microwaveprop/pskr/feature_bin.ex
new file mode 100644
index 00000000..b1ebc2d5
--- /dev/null
+++ b/lib/microwaveprop/pskr/feature_bin.ex
@@ -0,0 +1,58 @@
+defmodule Microwaveprop.Pskr.FeatureBin do
+  @moduledoc """
+  One feature-bucket row from a `Pskr.RecalibrationRun`. Stores the
+  spot-density statistics for a (band, feature, bin) cell of the
+  corpus so an operator can read whether a feature actually
+  discriminates propagation at the granularity the scorer cares
+  about.
+
+  Example: at run R, on the 10 GHz band, the `pwat_mm` feature has
+  five rows (one per bin label: `<15`, `15-25`, `25-40`, `40-55`,
+  `>55`). Each row carries:
+    * `sample_count` — how many corpus samples landed in the bin
+    * `spot_count_total` — sum of spot_count across those samples
+    * `spot_count_avg` / `spot_count_p50` / `spot_count_p90` —
+      central tendency + tail for the spot-density distribution
+
+  The unique key (run_id, band, feature, bin_label) means a rerun
+  of the same recalibration run is idempotent.
+  """
+  use Ecto.Schema
+
+  import Ecto.Changeset
+
+  @primary_key {:id, :binary_id, autogenerate: true}
+  @foreign_key_type :binary_id
+
+  schema "pskr_feature_bins" do
+    belongs_to :run, Microwaveprop.Pskr.RecalibrationRun, foreign_key: :run_id
+
+    field :band, :string
+    field :feature, :string
+    field :bin_label, :string
+    field :bin_min, :float
+    field :bin_max, :float
+    field :sample_count, :integer, default: 0
+    field :spot_count_total, :integer, default: 0
+    field :spot_count_avg, :float
+    field :spot_count_p50, :float
+    field :spot_count_p90, :float
+
+    timestamps(type: :utc_datetime)
+  end
+
+  @type t :: %__MODULE__{}
+
+  @cast_fields ~w(run_id band feature bin_label bin_min bin_max sample_count
+                  spot_count_total spot_count_avg spot_count_p50 spot_count_p90)a
+
+  @required_fields ~w(run_id band feature bin_label sample_count)a
+
+  @spec changeset(t(), map()) :: Ecto.Changeset.t()
+  def changeset(record, attrs) do
+    record
+    |> cast(attrs, @cast_fields)
+    |> validate_required(@required_fields)
+    |> unique_constraint([:run_id, :band, :feature, :bin_label])
+  end
+end
diff --git a/lib/microwaveprop/pskr/recalibration_run.ex b/lib/microwaveprop/pskr/recalibration_run.ex
new file mode 100644
index 00000000..0b56fc8f
--- /dev/null
+++ b/lib/microwaveprop/pskr/recalibration_run.ex
@@ -0,0 +1,54 @@
+defmodule Microwaveprop.Pskr.RecalibrationRun do
+  @moduledoc """
+  One execution of the PSKR-driven recalibration analysis.
+
+  A row is written every time `Pskr.Recalibrator.run/0` fires
+  (weekly via `Workers.PskrRecalibrationWorker`). The status field
+  separates an actually-completed analysis from runs that bailed
+  early because the corpus was too thin to produce useful bins.
+
+  Status values:
+    * `"completed"` — analysis ran and `pskr_feature_bins` rows
+      were written for at least one (band, feature)
+    * `"skipped_insufficient_data"` — corpus had < threshold
+      samples, no bins emitted
+    * `"failed"` — exception during analysis (notes carries
+      reason)
+  """
+  use Ecto.Schema
+
+  import Ecto.Changeset
+
+  @primary_key {:id, :binary_id, autogenerate: true}
+  @foreign_key_type :binary_id
+
+  schema "pskr_recalibration_runs" do
+    field :run_at, :utc_datetime
+    field :status, :string
+    field :sample_count, :integer, default: 0
+    field :band_count, :integer, default: 0
+    field :earliest_sample, :utc_datetime
+    field :latest_sample, :utc_datetime
+    field :avg_spots_per_sample, :float
+    field :notes, :string
+
+    has_many :feature_bins, Microwaveprop.Pskr.FeatureBin, foreign_key: :run_id
+
+    timestamps(type: :utc_datetime)
+  end
+
+  @type t :: %__MODULE__{}
+
+  @cast_fields ~w(run_at status sample_count band_count earliest_sample latest_sample
+                  avg_spots_per_sample notes)a
+
+  @required_fields ~w(run_at status)a
+
+  @spec changeset(t(), map()) :: Ecto.Changeset.t()
+  def changeset(record, attrs) do
+    record
+    |> cast(attrs, @cast_fields)
+    |> validate_required(@required_fields)
+    |> validate_inclusion(:status, ~w(completed skipped_insufficient_data failed))
+  end
+end
diff --git a/lib/microwaveprop/pskr/recalibrator.ex b/lib/microwaveprop/pskr/recalibrator.ex
new file mode 100644
index 00000000..22c229f1
--- /dev/null
+++ b/lib/microwaveprop/pskr/recalibrator.ex
@@ -0,0 +1,265 @@
+defmodule Microwaveprop.Pskr.Recalibrator do
+  @moduledoc """
+  PSKR-driven analysis pipeline. Reads `pskr_calibration_samples`,
+  bins each sample by every scoring feature × band, and writes per-
+  (run, band, feature, bin) spot-density stats to
+  `pskr_feature_bins`. The operator queries the bin tables to see
+  whether each feature actually discriminates spot density at the
+  threshold granularity the scorer uses.
+
+  Auto-applies *nothing* to the live algorithm. Weight changes still
+  go through human review of `BandConfig.@band_configs` and a code
+  commit. The recalibrator's job is to surface evidence; it stays
+  out of the production scoring path.
+
+  ## Why bins, not regression
+
+  `BandConfig` already encodes scoring as discrete thresholds (e.g.
+  refractivity gradient at -500 / -300 / -200 / -100). The bin
+  output matches that shape so an operator can read "PWAT 25-40 mm
+  on 10 GHz had n=4,200 cells with avg 18 spots, vs PWAT >55 mm
+  with n=620 and avg 4 spots" and decide directly whether a
+  threshold needs to move. A continuous regression would obscure
+  the threshold structure that the scorer needs to consume.
+
+  ## Self-healing
+
+  `run/0` is idempotent at the run level — every fire creates a
+  fresh `pskr_recalibration_runs` row. Fresh runs replace stale
+  ones at query time (operators look at the latest). Mid-run
+  failure logs a `failed` status and surfaces the exception in
+  `notes` so the next fire can retry without manual cleanup.
+  """
+
+  import Ecto.Query
+
+  alias Microwaveprop.Pskr.CalibrationSample
+  alias Microwaveprop.Pskr.FeatureBin
+  alias Microwaveprop.Pskr.RecalibrationRun
+  alias Microwaveprop.Repo
+
+  require Logger
+
+  # Minimum corpus size to attempt analysis. Below this the bin
+  # statistics are so noisy they're misleading; better to log a
+  # `skipped_insufficient_data` run and wait. ~1000 samples ≈
+  # 10 cells × 100 hours of CONUS PSKR activity, achievable in 4-5
+  # days even with sparse traffic.
+  @min_total_samples 1_000
+
+  # Per-band threshold for emitting bins for that band. With < this
+  # many samples on a band, the per-bin counts collapse to single
+  # digits and the analysis tells you nothing.
+  @min_band_samples 100
+
+  # Feature → bin definitions. Each list is `{label, min, max}` with
+  # `min`/`max` as inclusive lower / exclusive upper. nil sides
+  # mean unbounded. Bins are ordered for human-readable output but
+  # SQL returns them set-style; the worker re-sorts on read if
+  # needed.
+  @feature_bins %{
+    pwat_mm: [
+      {"<15", nil, 15.0},
+      {"15-25", 15.0, 25.0},
+      {"25-40", 25.0, 40.0},
+      {"40-55", 40.0, 55.0},
+      {">55", 55.0, nil}
+    ],
+    hpbl_m: [
+      {"<250", nil, 250.0},
+      {"250-500", 250.0, 500.0},
+      {"500-1000", 500.0, 1000.0},
+      {"1000-1500", 1000.0, 1500.0},
+      {">1500", 1500.0, nil}
+    ],
+    min_refractivity_gradient: [
+      {"<-200", nil, -200.0},
+      {"-200..-100", -200.0, -100.0},
+      {"-100..-50", -100.0, -50.0},
+      {">-50", -50.0, nil}
+    ],
+    surface_pressure_mb: [
+      {"<1010", nil, 1010.0},
+      {"1010-1015", 1010.0, 1015.0},
+      {"1015-1020", 1015.0, 1020.0},
+      {">1020", 1020.0, nil}
+    ],
+    kp_index: [
+      {"0-2 quiet", nil, 3.0},
+      {"3-4 unsettled", 3.0, 5.0},
+      {"5+ storm", 5.0, nil}
+    ]
+  }
+
+  @doc """
+  Run the analysis end-to-end. Returns the `RecalibrationRun` row
+  written. Safe to call from a worker, an iex shell, or a test —
+  always inserts a run record so the audit trail is complete.
+  """
+  @spec run() :: RecalibrationRun.t()
+  def run do
+    now = DateTime.truncate(DateTime.utc_now(), :second)
+    stats = corpus_stats()
+
+    if stats.sample_count < @min_total_samples do
+      insert_skipped_run(now, stats, "corpus has #{stats.sample_count} samples, need ≥ #{@min_total_samples}")
+    else
+      run_analysis(now, stats)
+    end
+  end
+
+  # ── corpus stats ────────────────────────────────────────────────
+
+  defp corpus_stats do
+    %{rows: [[total, bands, earliest, latest, avg_spots]]} =
+      Repo.query!(
+        """
+        SELECT
+          count(*)::int,
+          count(DISTINCT band)::int,
+          min(hour_utc),
+          max(hour_utc),
+          avg(spot_count)::float
+        FROM pskr_calibration_samples
+        """,
+        []
+      )
+
+    %{
+      sample_count: total || 0,
+      band_count: bands || 0,
+      earliest_sample: cast_to_utc(earliest),
+      latest_sample: cast_to_utc(latest),
+      avg_spots_per_sample: avg_spots
+    }
+  end
+
+  defp cast_to_utc(%NaiveDateTime{} = ndt), do: DateTime.from_naive!(ndt, "Etc/UTC")
+  defp cast_to_utc(nil), do: nil
+  defp cast_to_utc(%DateTime{} = dt), do: dt
+
+  # ── skipped run ────────────────────────────────────────────────
+
+  defp insert_skipped_run(now, stats, note) do
+    Logger.info("Pskr.Recalibrator: skipping — #{note}")
+
+    {:ok, run} =
+      %RecalibrationRun{}
+      |> RecalibrationRun.changeset(
+        Map.merge(stats, %{
+          run_at: now,
+          status: "skipped_insufficient_data",
+          notes: note
+        })
+      )
+      |> Repo.insert()
+
+    run
+  end
+
+  # ── analysis run ───────────────────────────────────────────────
+
+  defp run_analysis(now, stats) do
+    Logger.info("Pskr.Recalibrator: starting analysis on #{stats.sample_count} samples across #{stats.band_count} bands")
+
+    {:ok, run} =
+      %RecalibrationRun{}
+      |> RecalibrationRun.changeset(
+        Map.merge(stats, %{
+          run_at: now,
+          status: "completed"
+        })
+      )
+      |> Repo.insert()
+
+    bin_rows = build_bin_rows(run.id)
+
+    if bin_rows != [] do
+      now_seconds = DateTime.truncate(DateTime.utc_now(), :second)
+
+      timestamped =
+        Enum.map(bin_rows, &Map.merge(&1, %{inserted_at: now_seconds, updated_at: now_seconds}))
+
+      {count, _} = Repo.insert_all(FeatureBin, timestamped)
+      Logger.info("Pskr.Recalibrator: wrote #{count} feature_bins for run #{run.id}")
+    end
+
+    Repo.preload(run, :feature_bins)
+  end
+
+  # ── bin generation ─────────────────────────────────────────────
+
+  defp build_bin_rows(run_id) do
+    eligible_bands = eligible_bands_for_run()
+
+    for band <- eligible_bands,
+        {feature, defs} <- @feature_bins,
+        {label, lo, hi} <- defs,
+        row = bin_for(run_id, band, feature, label, lo, hi),
+        not is_nil(row) do
+      row
+    end
+  end
+
+  defp eligible_bands_for_run do
+    Repo.all(
+      from(s in CalibrationSample,
+        group_by: s.band,
+        having: count(s.id) >= ^@min_band_samples,
+        select: s.band
+      )
+    )
+  end
+
+  defp bin_for(run_id, band, feature, label, lo, hi) do
+    %{rows: rows} =
+      Repo.query!(bin_sql(feature), bin_params(band, lo, hi))
+
+    [[count, total, avg, p50, p90]] = rows
+
+    if count == 0 do
+      nil
+    else
+      %{
+        id: Ecto.UUID.generate(),
+        run_id: run_id,
+        band: band,
+        feature: Atom.to_string(feature),
+        bin_label: label,
+        bin_min: lo,
+        bin_max: hi,
+        sample_count: count,
+        spot_count_total: total || 0,
+        spot_count_avg: avg,
+        spot_count_p50: p50,
+        spot_count_p90: p90
+      }
+    end
+  end
+
+  # Per-feature SQL: count samples that have a non-null value of the
+  # feature within the bin range, plus spot-count statistics. Built
+  # at compile time per feature so we don't take an Ecto fragment
+  # interpolation hit on the hot path.
+  for feature <- Map.keys(@feature_bins) do
+    col = Atom.to_string(feature)
+
+    defp bin_sql(unquote(feature)) do
+      """
+      SELECT
+        count(*)::int,
+        sum(spot_count)::bigint,
+        avg(spot_count)::float,
+        percentile_cont(0.5) WITHIN GROUP (ORDER BY spot_count)::float,
+        percentile_cont(0.9) WITHIN GROUP (ORDER BY spot_count)::float
+      FROM pskr_calibration_samples
+      WHERE band = $1
+        AND #{unquote(col)} IS NOT NULL
+        AND ($2::float IS NULL OR #{unquote(col)} >= $2)
+        AND ($3::float IS NULL OR #{unquote(col)} < $3)
+      """
+    end
+  end
+
+  defp bin_params(band, lo, hi), do: [band, lo, hi]
+end
diff --git a/lib/microwaveprop/workers/pskr_recalibration_worker.ex b/lib/microwaveprop/workers/pskr_recalibration_worker.ex
new file mode 100644
index 00000000..7c1830d6
--- /dev/null
+++ b/lib/microwaveprop/workers/pskr_recalibration_worker.ex
@@ -0,0 +1,35 @@
+defmodule Microwaveprop.Workers.PskrRecalibrationWorker do
+  @moduledoc """
+  Weekly cron that runs `Pskr.Recalibrator` against the
+  PSKR calibration corpus.
+
+  Scheduled for Sunday 04:00 UTC (off-peak, after the climatology
+  rebuild and before the Monday-morning operator review window). At
+  fire time the recalibrator decides whether the corpus is dense
+  enough to bin — when it isn't, the run still records a row with
+  status `"skipped_insufficient_data"` so the audit trail captures
+  every fire.
+
+  Manual reruns: enqueue this worker with no args. It always
+  produces a fresh run record; previous runs stay queryable for
+  trend comparison.
+  """
+  use Oban.Worker,
+    queue: :backfill_enqueue,
+    max_attempts: 3,
+    unique: [
+      period: 86_400,
+      states: [:available, :scheduled, :executing, :retryable]
+    ]
+
+  alias Microwaveprop.Pskr.Recalibrator
+
+  require Logger
+
+  @impl Oban.Worker
+  def perform(%Oban.Job{}) do
+    run = Recalibrator.run()
+    Logger.info("PskrRecalibrationWorker: status=#{run.status} sample_count=#{run.sample_count}")
+    :ok
+  end
+end
diff --git a/priv/repo/migrations/20260504211415_create_pskr_recalibration_tables.exs b/priv/repo/migrations/20260504211415_create_pskr_recalibration_tables.exs
new file mode 100644
index 00000000..aea5d99d
--- /dev/null
+++ b/priv/repo/migrations/20260504211415_create_pskr_recalibration_tables.exs
@@ -0,0 +1,61 @@
+defmodule Microwaveprop.Repo.Migrations.CreatePskrRecalibrationTables do
+  use Ecto.Migration
+
+  # Two-table layout for the PSKR-driven recalibration analysis:
+  #
+  #   * `pskr_recalibration_runs` — one row per scheduled run with
+  #     corpus stats and run status. Cheap to scan, fronts the
+  #     analysis tables.
+  #   * `pskr_feature_bins` — one row per (run, band, feature, bin)
+  #     with spot-density stats. The recalibrator generates these by
+  #     bucketing the corpus on each scoring feature so an operator
+  #     can read directly: "at PWAT 40-55 mm, 10 GHz cells averaged
+  #     N spots vs M at 15-25 mm — is that real or sampling noise?"
+  #
+  # Bins beat continuous regression for the operator workflow:
+  # the algorithm's per-band weights are also discrete thresholds
+  # (`@band_configs`), so matching the shape lets the operator copy
+  # adjusted thresholds back into `BandConfig` without translation.
+
+  def change do
+    create table(:pskr_recalibration_runs, primary_key: false) do
+      add :id, :binary_id, primary_key: true
+      add :run_at, :utc_datetime, null: false
+      add :status, :string, null: false
+      add :sample_count, :integer, null: false, default: 0
+      add :band_count, :integer, null: false, default: 0
+      add :earliest_sample, :utc_datetime
+      add :latest_sample, :utc_datetime
+      add :avg_spots_per_sample, :float
+      add :notes, :text
+
+      timestamps(type: :utc_datetime)
+    end
+
+    create index(:pskr_recalibration_runs, [:run_at])
+
+    create table(:pskr_feature_bins, primary_key: false) do
+      add :id, :binary_id, primary_key: true
+
+      add :run_id,
+          references(:pskr_recalibration_runs, type: :binary_id, on_delete: :delete_all),
+          null: false
+
+      add :band, :string, null: false
+      add :feature, :string, null: false
+      add :bin_label, :string, null: false
+      add :bin_min, :float
+      add :bin_max, :float
+      add :sample_count, :integer, null: false, default: 0
+      add :spot_count_total, :integer, null: false, default: 0
+      add :spot_count_avg, :float
+      add :spot_count_p50, :float
+      add :spot_count_p90, :float
+
+      timestamps(type: :utc_datetime)
+    end
+
+    create unique_index(:pskr_feature_bins, [:run_id, :band, :feature, :bin_label])
+    create index(:pskr_feature_bins, [:run_id])
+  end
+end
diff --git a/test/microwaveprop/pskr/recalibrator_test.exs b/test/microwaveprop/pskr/recalibrator_test.exs
new file mode 100644
index 00000000..96449fd2
--- /dev/null
+++ b/test/microwaveprop/pskr/recalibrator_test.exs
@@ -0,0 +1,163 @@
+defmodule Microwaveprop.Pskr.RecalibratorTest do
+  use Microwaveprop.DataCase, async: true
+
+  alias Microwaveprop.Pskr.CalibrationSample
+  alias Microwaveprop.Pskr.FeatureBin
+  alias Microwaveprop.Pskr.RecalibrationRun
+  alias Microwaveprop.Pskr.Recalibrator
+  alias Microwaveprop.Repo
+
+  defp insert_sample!(attrs) do
+    base = %{
+      hour_utc: ~U[2026-05-04 18:00:00Z],
+      band: "10000",
+      midpoint_lat: 33.0,
+      midpoint_lon: -97.0,
+      spot_count: 5,
+      pwat_mm: 30.0,
+      hpbl_m: 500.0,
+      min_refractivity_gradient: -100.0,
+      surface_pressure_mb: 1013.0,
+      kp_index: 2
+    }
+
+    {:ok, _} =
+      %CalibrationSample{}
+      |> CalibrationSample.changeset(Map.merge(base, attrs))
+      |> Repo.insert()
+  end
+
+  # Insert N samples for a band, optionally varying one feature.
+  # Uses a per-call unique offset so successive `insert_n!` invocations
+  # don't collide on the (hour, band, midpoint_lat, midpoint_lon) key.
+  defp insert_n!(n, band, attrs_fn) do
+    offset = :erlang.unique_integer([:positive])
+
+    Enum.each(1..n, fn i ->
+      attrs = attrs_fn.(i)
+      mp_lat = 33.0 + (offset * 1000 + i) * 0.0001
+      mp_lon = -97.0 - (offset * 1000 + i) * 0.0001
+      insert_sample!(Map.merge(attrs, %{band: band, midpoint_lat: mp_lat, midpoint_lon: mp_lon}))
+    end)
+  end
+
+  describe "run/0" do
+    test "skips with status `skipped_insufficient_data` when corpus is empty" do
+      run = Recalibrator.run()
+
+      assert run.status == "skipped_insufficient_data"
+      assert run.sample_count == 0
+      assert run.notes =~ "≥ 1000"
+      assert Repo.aggregate(FeatureBin, :count) == 0
+    end
+
+    test "skips when corpus has < 1000 total samples" do
+      insert_n!(500, "10000", fn _ -> %{} end)
+      run = Recalibrator.run()
+
+      assert run.status == "skipped_insufficient_data"
+      assert run.sample_count == 500
+      assert Repo.aggregate(FeatureBin, :count) == 0
+    end
+
+    test "completes and writes feature bins when corpus is dense enough" do
+      # Spread samples across PWAT bins so several have non-zero counts.
+      insert_n!(1200, "10000", fn i ->
+        pwat = 10.0 + rem(i, 60)
+        %{pwat_mm: pwat, spot_count: rem(i, 10) + 1}
+      end)
+
+      run = Recalibrator.run()
+      assert run.status == "completed"
+      assert run.sample_count == 1200
+      assert run.band_count == 1
+      assert run.avg_spots_per_sample > 0
+
+      bins = Repo.all(from(fb in FeatureBin, where: fb.run_id == ^run.id))
+      assert length(bins) > 0
+
+      # Every bin row should reference the run, target this band, and
+      # carry non-negative stats.
+      Enum.each(bins, fn bin ->
+        assert bin.run_id == run.id
+        assert bin.band == "10000"
+        assert bin.sample_count > 0
+        assert bin.spot_count_total >= 0
+      end)
+
+      # PWAT should populate every defined bin label since we
+      # spread values 10-69 across the range.
+      pwat_bins =
+        bins
+        |> Enum.filter(&(&1.feature == "pwat_mm"))
+        |> Enum.map(& &1.bin_label)
+        |> Enum.sort()
+
+      assert "<15" in pwat_bins
+      assert "15-25" in pwat_bins
+      assert "25-40" in pwat_bins
+      assert "40-55" in pwat_bins
+      assert ">55" in pwat_bins
+    end
+
+    test "skips bands below the per-band threshold even when total corpus passes" do
+      # Band A gets enough samples; band B (50 samples) is too thin
+      # to bin even though the total corpus passes the global
+      # threshold.
+      insert_n!(1100, "10000", fn _ -> %{spot_count: 3} end)
+      insert_n!(50, "144", fn _ -> %{spot_count: 1} end)
+
+      run = Recalibrator.run()
+      assert run.status == "completed"
+      assert run.sample_count == 1150
+      # band_count counts distinct bands in the corpus, not just the
+      # bands we ran analysis for, so 2 here is correct.
+      assert run.band_count == 2
+
+      bins = Repo.all(from(fb in FeatureBin, where: fb.run_id == ^run.id))
+      bands_with_bins = bins |> Enum.map(& &1.band) |> Enum.uniq()
+
+      assert "10000" in bands_with_bins
+      refute "144" in bands_with_bins
+    end
+
+    test "spot_count_avg reflects the actual cell mean" do
+      # Force every cell to have spot_count = 4 so we know the avg.
+      insert_n!(1000, "10000", fn _ -> %{spot_count: 4} end)
+
+      run = Recalibrator.run()
+      pwat_bins = Repo.all(from(fb in FeatureBin, where: fb.run_id == ^run.id and fb.feature == "pwat_mm"))
+
+      Enum.each(pwat_bins, fn bin ->
+        if bin.sample_count > 0 do
+          assert bin.spot_count_avg == 4.0
+        end
+      end)
+    end
+
+    test "ignores samples with nil feature value when binning that feature" do
+      # Half have pwat_mm; half don't. The pwat bins should only
+      # count the half that does.
+      insert_n!(500, "10000", fn _ -> %{pwat_mm: 30.0, spot_count: 1} end)
+      insert_n!(500, "10000", fn _ -> %{pwat_mm: nil, spot_count: 1} end)
+
+      run = Recalibrator.run()
+      assert run.status == "completed"
+
+      pwat_total =
+        FeatureBin
+        |> where([fb], fb.run_id == ^run.id and fb.feature == "pwat_mm")
+        |> Repo.all()
+        |> Enum.map(& &1.sample_count)
+        |> Enum.sum()
+
+      assert pwat_total == 500
+    end
+
+    test "every run records a row in pskr_recalibration_runs regardless of status" do
+      Recalibrator.run()
+      Recalibrator.run()
+      assert Repo.aggregate(RecalibrationRun, :count) == 2
+    end
+  end
+end
diff --git a/test/microwaveprop/workers/pskr_recalibration_worker_test.exs b/test/microwaveprop/workers/pskr_recalibration_worker_test.exs
new file mode 100644
index 00000000..5bfe44c7
--- /dev/null
+++ b/test/microwaveprop/workers/pskr_recalibration_worker_test.exs
@@ -0,0 +1,16 @@
+defmodule Microwaveprop.Workers.PskrRecalibrationWorkerTest do
+  use Microwaveprop.DataCase, async: true
+  use Oban.Testing, repo: Microwaveprop.Repo
+
+  alias Microwaveprop.Pskr.RecalibrationRun
+  alias Microwaveprop.Repo
+  alias Microwaveprop.Workers.PskrRecalibrationWorker
+
+  test "perform/1 records a run row even when the corpus is empty" do
+    assert :ok = perform_job(PskrRecalibrationWorker, %{})
+
+    [run] = Repo.all(RecalibrationRun)
+    assert run.status == "skipped_insufficient_data"
+    assert run.sample_count == 0
+  end
+end