feat(pskr): weekly recalibration analysis on the calibration corpus

`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.
This commit is contained in:
Graham McIntire 2026-05-04 16:18:17 -05:00
parent 7a2b1f292c
commit 48000b0dca
No known key found for this signature in database
GPG key ID: F4ABF488E6029E59
8 changed files with 661 additions and 1 deletions

View file

@ -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 = [

View file

@ -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

View file

@ -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

View file

@ -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

View file

@ -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

View file

@ -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

View file

@ -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

View file

@ -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