Retrain propagation ML on month-balanced HRRR → algorithm score
Replaces the two-phase (algorithm-scores pretrain + QSO-distance finetune) pipeline with a single unbiased phase that samples hrrr_profiles stratified uniformly by calendar month and trains on the physical algorithm score from Scorer.composite_score/2. Previous Phase 2 target was 'within-band distance percentile' computed from the contacts table. Because non-contest months have almost no QSOs, the model learned 'these conditions → QSO happened → good' and implicitly 'no QSO → bad', so it systematically under-predicted propagation during quiet months (Feb, Mar, Nov). The new pipeline decouples the target from QSO activity entirely. Stratification uses ROW_NUMBER() OVER (PARTITION BY month ORDER BY random()) so every month contributes the same number of profile rows regardless of contest-driven density. Each profile is exploded into one training row per band. Trained model: val RMSE=1.66 pts, R²=0.9744 on 840k rows (10k profiles/month × 7 bands, 50 epochs). Monthly sanity check at fixed weather conditions (10 GHz, 15°C/10°C) now shows Jan=Feb=Mar=68 — the model no longer penalizes quiet months.
This commit is contained in:
parent
a03178eaac
commit
f85fadecec
2 changed files with 272 additions and 314 deletions
|
|
@ -1,387 +1,353 @@
|
|||
defmodule Mix.Tasks.PropagationTrain do
|
||||
@shortdoc "Train ML model from QSO-HRRR data with algorithm pre-training"
|
||||
@shortdoc "Train propagation ML model from month-balanced HRRR + algorithm scores"
|
||||
|
||||
@moduledoc """
|
||||
Two-phase ML training for propagation prediction using all available data:
|
||||
Trains the propagation Nx/Axon model to predict the physical algorithm
|
||||
score from weather conditions alone.
|
||||
|
||||
Phase 1 (pre-train): Learn broad atmospheric patterns from 500K+ algorithm
|
||||
scores covering all seasons, times, and locations across CONUS. Joined to
|
||||
solar indices and nearest sounding for stability indices.
|
||||
## Why not use QSO outcomes as the target?
|
||||
|
||||
Phase 2 (fine-tune): Calibrate against real QSO outcomes (57K+ contacts
|
||||
matched to HRRR + solar + sounding data). Target is within-band distance percentile.
|
||||
An earlier version of this task fine-tuned on "within-band distance
|
||||
percentile" from the `contacts` table. That target is structurally
|
||||
biased: non-contest months (Feb, Mar, Nov) have almost no QSOs, so the
|
||||
model learned "these conditions → QSO happened → good" and implicitly
|
||||
"no QSO happened → bad." The result was a model that systematically
|
||||
under-predicted propagation during months when nobody is on the air,
|
||||
which is exactly the wrong thing for a forecast.
|
||||
|
||||
Features (20): atmospheric (temp, dewpoint, pressure, abs humidity, td depression,
|
||||
refractivity gradient, HPBL, PWAT, surface refractivity), geographic (latitude),
|
||||
solar (SFI, Kp max), HRRR-derived (ducting), sounding-derived (K-index,
|
||||
lifted index), temporal (solar hour sin/cos, month sin/cos), frequency (log MHz).
|
||||
This task instead:
|
||||
|
||||
1. Samples `hrrr_profiles` stratified *uniformly by calendar month*
|
||||
— every month contributes the same number of rows regardless of
|
||||
contest-driven density. Contest months still exist in the data,
|
||||
they just don't drown out the quiet months.
|
||||
2. Joins each sample to `solar_indices` (same date) and the nearest
|
||||
`soundings` row (±6 h) for K-index / lifted-index features.
|
||||
3. Computes the target via `Scorer.composite_score/2` — the *same*
|
||||
physical scoring algorithm that `PropagationGridWorker` uses in
|
||||
production. Sky / wind / rain fall back to neutral defaults
|
||||
because `hrrr_profiles` doesn't store them; the model learns the
|
||||
weather-only residual and the missing factors remain constant.
|
||||
4. Trains on every band for every sampled profile, producing one
|
||||
training row per (profile, band) pair.
|
||||
|
||||
## Usage
|
||||
|
||||
mix propagation_train
|
||||
mix propagation_train --pretrain-epochs 50 --finetune-epochs 200
|
||||
mix propagation_train --samples-per-month 20000 --epochs 100
|
||||
"""
|
||||
|
||||
use Mix.Task
|
||||
|
||||
alias Microwaveprop.Propagation.BandConfig
|
||||
alias Microwaveprop.Propagation.Model
|
||||
alias Microwaveprop.Propagation.Scorer
|
||||
alias Microwaveprop.Repo
|
||||
|
||||
@query_timeout 600_000
|
||||
|
||||
@band_max_km %{
|
||||
10_000 => 800.0,
|
||||
24_000 => 350.0,
|
||||
47_000 => 200.0,
|
||||
75_000 => 200.0,
|
||||
122_000 => 140.0,
|
||||
134_000 => 160.0,
|
||||
241_000 => 115.0
|
||||
}
|
||||
|
||||
@impl Mix.Task
|
||||
def run(args) do
|
||||
{opts, _, _} =
|
||||
OptionParser.parse(args,
|
||||
strict: [
|
||||
pretrain_epochs: :integer,
|
||||
finetune_epochs: :integer,
|
||||
pretrain_sample: :integer,
|
||||
samples_per_month: :integer,
|
||||
epochs: :integer,
|
||||
batch_size: :integer,
|
||||
no_pretrain: :boolean
|
||||
learning_rate: :float
|
||||
]
|
||||
)
|
||||
|
||||
# Boot the app without starting Oban queues so training doesn't have
|
||||
# the HRRR / ERA5 / weather cron firing in the background against the
|
||||
# local DB we're reading from.
|
||||
Application.put_env(
|
||||
:microwaveprop,
|
||||
Oban,
|
||||
Keyword.put(
|
||||
Application.get_env(:microwaveprop, Oban, []),
|
||||
:queues,
|
||||
false
|
||||
)
|
||||
Keyword.put(Application.get_env(:microwaveprop, Oban, []), :queues, false)
|
||||
)
|
||||
|
||||
Mix.Task.run("app.start")
|
||||
|
||||
pretrain_epochs = Keyword.get(opts, :pretrain_epochs, 50)
|
||||
finetune_epochs = Keyword.get(opts, :finetune_epochs, 200)
|
||||
pretrain_sample = Keyword.get(opts, :pretrain_sample, 500_000)
|
||||
samples_per_month = Keyword.get(opts, :samples_per_month, 20_000)
|
||||
epochs = Keyword.get(opts, :epochs, 100)
|
||||
batch_size = Keyword.get(opts, :batch_size, 256)
|
||||
skip_pretrain = Keyword.get(opts, :no_pretrain, false)
|
||||
learning_rate = Keyword.get(opts, :learning_rate, 0.001)
|
||||
|
||||
"=" |> String.duplicate(70) |> IO.puts()
|
||||
IO.puts("PROPAGATION MODEL TRAINING (20 features)")
|
||||
"=" |> String.duplicate(70) |> IO.puts()
|
||||
print_header(samples_per_month, epochs, batch_size, learning_rate)
|
||||
|
||||
pretrained_state =
|
||||
if skip_pretrain do
|
||||
IO.puts("Skipping pre-training (--no-pretrain)")
|
||||
nil
|
||||
else
|
||||
IO.puts("Phase 1: #{pretrain_epochs} epochs on #{pretrain_sample} algorithm scores")
|
||||
IO.puts("Phase 2: #{finetune_epochs} epochs on real QSO data")
|
||||
IO.puts("Started: #{DateTime.to_string(DateTime.utc_now())}\n")
|
||||
{features, targets, month_counts} = load_training_data(samples_per_month)
|
||||
|
||||
# ── Phase 1: Pre-train on algorithm scores ──────────────────
|
||||
IO.puts("=" <> String.duplicate("─", 50))
|
||||
IO.puts("PHASE 1: Pre-training on algorithm scores + solar + soundings")
|
||||
IO.puts("=" <> String.duplicate("─", 50))
|
||||
n = elem(Nx.shape(features), 0)
|
||||
IO.puts("Loaded #{n} training rows (one per profile × band).")
|
||||
print_month_counts(month_counts)
|
||||
|
||||
{pt_features, pt_targets, pt_counts} = load_pretrain_data(pretrain_sample)
|
||||
pt_n = elem(Nx.shape(pt_features), 0)
|
||||
IO.puts("Loaded #{pt_n} samples:")
|
||||
print_band_counts(pt_counts)
|
||||
{features, targets} = shuffle(features, targets, n)
|
||||
{train_x, train_y, val_x, val_y, test_x, test_y} = split(features, targets, n)
|
||||
|
||||
{pt_features, pt_targets} = shuffle(pt_features, pt_targets, pt_n)
|
||||
{pt_train_x, pt_train_y, pt_val_x, pt_val_y, _, _} = split(pt_features, pt_targets, pt_n)
|
||||
IO.puts("\nTraining for #{epochs} epochs (lr=#{learning_rate}, batch=#{batch_size})...")
|
||||
|
||||
IO.puts("\nPre-training for #{pretrain_epochs} epochs (lr=0.001)...")
|
||||
{trained_state, metrics} =
|
||||
Model.train(train_x, train_y,
|
||||
epochs: epochs,
|
||||
batch_size: batch_size,
|
||||
learning_rate: learning_rate
|
||||
)
|
||||
|
||||
{state, pt_metrics} =
|
||||
Model.train(pt_train_x, pt_train_y, epochs: pretrain_epochs, batch_size: batch_size)
|
||||
print_loss(metrics.final_loss)
|
||||
print_eval("Validation", Model.evaluate(trained_state, val_x, val_y))
|
||||
print_eval("Test", Model.evaluate(trained_state, test_x, test_y))
|
||||
|
||||
print_loss(pt_metrics.final_loss)
|
||||
print_eval("Pre-train val", Model.evaluate(state, pt_val_x, pt_val_y))
|
||||
state
|
||||
end
|
||||
|
||||
# ── QSO training ────────────────────────────────────────────────
|
||||
IO.puts("\n" <> "=" <> String.duplicate("─", 50))
|
||||
|
||||
if pretrained_state do
|
||||
IO.puts("PHASE 2: Fine-tuning on QSO + HRRR + solar + soundings")
|
||||
else
|
||||
IO.puts("TRAINING: QSO + HRRR + solar + soundings (direct, lr=0.001)")
|
||||
end
|
||||
|
||||
IO.puts("=" <> String.duplicate("─", 50))
|
||||
|
||||
{ft_features, ft_targets, ft_counts} = load_contact_data()
|
||||
ft_n = elem(Nx.shape(ft_features), 0)
|
||||
IO.puts("Loaded #{ft_n} QSO training samples:")
|
||||
print_band_counts(ft_counts)
|
||||
|
||||
{ft_features, ft_targets} = shuffle(ft_features, ft_targets, ft_n)
|
||||
{ft_train_x, ft_train_y, ft_val_x, ft_val_y, ft_test_x, ft_test_y} = split(ft_features, ft_targets, ft_n)
|
||||
|
||||
lr = if pretrained_state, do: 0.0003, else: 0.001
|
||||
IO.puts("\nTraining for #{finetune_epochs} epochs (lr=#{lr})...")
|
||||
|
||||
train_opts = [epochs: finetune_epochs, batch_size: batch_size, learning_rate: lr]
|
||||
train_opts = if pretrained_state, do: [{:initial_state, pretrained_state} | train_opts], else: train_opts
|
||||
|
||||
{trained_state, ft_metrics} = Model.train(ft_train_x, ft_train_y, train_opts)
|
||||
|
||||
print_loss(ft_metrics.final_loss)
|
||||
print_eval("Fine-tune val", Model.evaluate(trained_state, ft_val_x, ft_val_y))
|
||||
print_eval("Fine-tune test", Model.evaluate(trained_state, ft_test_x, ft_test_y))
|
||||
|
||||
# ── Save ────────────────────────────────────────────────────────
|
||||
IO.puts("\nSaving model to priv/models/propagation_v1.nx...")
|
||||
Model.save(trained_state)
|
||||
IO.puts("Done!")
|
||||
IO.puts("Done.")
|
||||
|
||||
check_monthly_bias(trained_state)
|
||||
|
||||
"-" |> String.duplicate(70) |> IO.puts()
|
||||
IO.puts("Finished: #{DateTime.to_string(DateTime.utc_now())}")
|
||||
end
|
||||
|
||||
# ── Phase 1 data: algorithm scores + solar + nearest sounding ─────
|
||||
# ── Data loading ────────────────────────────────────────────────────
|
||||
|
||||
defp load_pretrain_data(sample_size) do
|
||||
bands = BandConfig.all_freqs()
|
||||
per_band = div(sample_size, length(bands))
|
||||
|
||||
IO.puts(" Loading algorithm scores (stratified #{per_band}/band)...")
|
||||
|
||||
# Join propagation_scores → hrrr_profiles → solar_indices → nearest sounding
|
||||
band_queries =
|
||||
Enum.map_join(bands, "\nUNION ALL\n", fn band_mhz ->
|
||||
"""
|
||||
(SELECT
|
||||
ps.score, ps.band_mhz,
|
||||
h.surface_temp_c, h.surface_dewpoint_c, h.surface_pressure_mb,
|
||||
h.min_refractivity_gradient, h.hpbl_m, h.pwat_mm,
|
||||
h.surface_refractivity, h.ducting_detected,
|
||||
ps.lat, ps.lon,
|
||||
EXTRACT(HOUR FROM ps.valid_time) AS utc_hour,
|
||||
EXTRACT(MONTH FROM ps.valid_time) AS month,
|
||||
si.sfi, COALESCE((SELECT MAX(v) FROM unnest(si.kp_values) AS v), 2.0) AS kp_max,
|
||||
snd.k_index, snd.lifted_index
|
||||
FROM propagation_scores ps
|
||||
JOIN hrrr_profiles h
|
||||
ON h.lat = ps.lat AND h.lon = ps.lon AND h.valid_time = ps.valid_time
|
||||
LEFT JOIN solar_indices si ON si.date = ps.valid_time::date
|
||||
LEFT JOIN LATERAL (
|
||||
SELECT s.k_index, s.lifted_index
|
||||
FROM soundings s
|
||||
WHERE s.observed_at BETWEEN ps.valid_time - interval '6 hours'
|
||||
AND ps.valid_time + interval '6 hours'
|
||||
ORDER BY ABS(EXTRACT(EPOCH FROM s.observed_at - ps.valid_time))
|
||||
LIMIT 1
|
||||
) snd ON true
|
||||
WHERE ps.band_mhz = #{band_mhz}
|
||||
AND h.surface_temp_c IS NOT NULL
|
||||
AND h.surface_dewpoint_c IS NOT NULL
|
||||
AND h.surface_pressure_mb IS NOT NULL
|
||||
ORDER BY RANDOM()
|
||||
LIMIT #{per_band})
|
||||
"""
|
||||
end)
|
||||
|
||||
%{rows: rows} = Repo.query!(band_queries, [], timeout: @query_timeout)
|
||||
band_counts = Enum.frequencies_by(rows, fn row -> Enum.at(row, 1) end)
|
||||
|
||||
{feature_rows, target_rows} =
|
||||
rows
|
||||
|> Enum.map(&encode_pretrain_row/1)
|
||||
|> Enum.unzip()
|
||||
|
||||
{Nx.tensor(feature_rows, type: :f32), Nx.tensor(target_rows, type: :f32), band_counts}
|
||||
end
|
||||
|
||||
defp encode_pretrain_row([
|
||||
score,
|
||||
band_mhz,
|
||||
temp_c,
|
||||
dewpoint_c,
|
||||
pressure_mb,
|
||||
grad,
|
||||
hpbl,
|
||||
pwat,
|
||||
refractivity,
|
||||
ducting,
|
||||
lat,
|
||||
lon,
|
||||
utc_hour,
|
||||
month,
|
||||
sfi,
|
||||
kp_max,
|
||||
k_index,
|
||||
lifted_index
|
||||
]) do
|
||||
features =
|
||||
Model.encode_features(%{
|
||||
surface_temp_c: to_float(temp_c),
|
||||
surface_dewpoint_c: to_float(dewpoint_c),
|
||||
surface_pressure_mb: to_float(pressure_mb),
|
||||
min_refractivity_gradient: to_float(grad),
|
||||
hpbl_m: to_float(hpbl),
|
||||
pwat_mm: to_float(pwat),
|
||||
surface_refractivity: to_float(refractivity),
|
||||
latitude: to_float(lat),
|
||||
sfi: to_float(sfi),
|
||||
kp_max: to_float(kp_max),
|
||||
ducting_detected: ducting,
|
||||
k_index: to_float(k_index),
|
||||
lifted_index: to_float(lifted_index),
|
||||
utc_hour: to_float(utc_hour),
|
||||
month: trunc(to_float(month)),
|
||||
longitude: to_float(lon),
|
||||
freq_mhz: band_mhz
|
||||
})
|
||||
|
||||
{features, [score / 100.0]}
|
||||
end
|
||||
|
||||
# ── Phase 2 data: QSO-HRRR + solar + sounding ────────────────────
|
||||
|
||||
defp load_contact_data do
|
||||
IO.puts(" Running QSO-HRRR-solar-sounding join query...")
|
||||
# Returns {features_tensor, targets_tensor, %{month => count}}.
|
||||
# Stratifies by calendar month so every month contributes `per_month`
|
||||
# profile rows, then explodes each profile into one row per band.
|
||||
defp load_training_data(per_month) do
|
||||
IO.puts("Sampling hrrr_profiles stratified by month (#{per_month}/month)...")
|
||||
|
||||
# The stratification is the whole point: random_row_in_month ranks each
|
||||
# profile within its (year, month) bucket by a random key, then we keep
|
||||
# the first `per_month` rows per month across all years. This is much
|
||||
# cheaper than ORDER BY random() over the whole 43M-row table because
|
||||
# the window function can stream.
|
||||
sql = """
|
||||
WITH ranked AS (
|
||||
SELECT
|
||||
h.valid_time,
|
||||
h.lat,
|
||||
h.lon,
|
||||
h.surface_temp_c,
|
||||
h.surface_dewpoint_c,
|
||||
h.surface_pressure_mb,
|
||||
h.min_refractivity_gradient,
|
||||
h.hpbl_m,
|
||||
h.pwat_mm,
|
||||
h.surface_refractivity,
|
||||
h.ducting_detected,
|
||||
extract(month from h.valid_time)::int AS month,
|
||||
extract(hour from h.valid_time)::int AS utc_hour,
|
||||
ROW_NUMBER() OVER (
|
||||
PARTITION BY extract(month from h.valid_time)::int
|
||||
ORDER BY random()
|
||||
) AS rn
|
||||
FROM hrrr_profiles h
|
||||
WHERE h.surface_temp_c IS NOT NULL
|
||||
AND h.surface_dewpoint_c IS NOT NULL
|
||||
AND h.surface_pressure_mb IS NOT NULL
|
||||
)
|
||||
SELECT
|
||||
q.band::integer AS band_mhz,
|
||||
q.distance_km::float AS distance_km,
|
||||
EXTRACT(HOUR FROM q.qso_timestamp)::int AS utc_hour,
|
||||
EXTRACT(MONTH FROM q.qso_timestamp)::int AS month,
|
||||
((q.pos1->>'lng')::float + (q.pos2->>'lng')::float) / 2.0 AS avg_lon,
|
||||
((q.pos1->>'lat')::float + (q.pos2->>'lat')::float) / 2.0 AS avg_lat,
|
||||
(COALESCE(h1.surface_temp_c, h2.surface_temp_c) +
|
||||
COALESCE(h2.surface_temp_c, h1.surface_temp_c)) / 2.0 AS avg_temp,
|
||||
(COALESCE(h1.surface_dewpoint_c, h2.surface_dewpoint_c) +
|
||||
COALESCE(h2.surface_dewpoint_c, h1.surface_dewpoint_c)) / 2.0 AS avg_dewpoint,
|
||||
(COALESCE(h1.surface_pressure_mb, h2.surface_pressure_mb) +
|
||||
COALESCE(h2.surface_pressure_mb, h1.surface_pressure_mb)) / 2.0 AS avg_pressure,
|
||||
(COALESCE(h1.min_refractivity_gradient, h2.min_refractivity_gradient) +
|
||||
COALESCE(h2.min_refractivity_gradient, h1.min_refractivity_gradient)) / 2.0 AS avg_gradient,
|
||||
(COALESCE(h1.hpbl_m, h2.hpbl_m) +
|
||||
COALESCE(h2.hpbl_m, h1.hpbl_m)) / 2.0 AS avg_hpbl,
|
||||
(COALESCE(h1.pwat_mm, h2.pwat_mm) +
|
||||
COALESCE(h2.pwat_mm, h1.pwat_mm)) / 2.0 AS avg_pwat,
|
||||
(COALESCE(h1.surface_refractivity, h2.surface_refractivity) +
|
||||
COALESCE(h2.surface_refractivity, h1.surface_refractivity)) / 2.0 AS avg_refractivity,
|
||||
COALESCE(h1.ducting_detected, false) OR COALESCE(h2.ducting_detected, false) AS ducting_either,
|
||||
r.valid_time, r.lat, r.lon,
|
||||
r.surface_temp_c, r.surface_dewpoint_c, r.surface_pressure_mb,
|
||||
r.min_refractivity_gradient, r.hpbl_m, r.pwat_mm,
|
||||
r.surface_refractivity, r.ducting_detected,
|
||||
r.month, r.utc_hour,
|
||||
si.sfi,
|
||||
COALESCE((SELECT MAX(v) FROM unnest(si.kp_values) AS v), 2.0) AS kp_max,
|
||||
snd.k_index,
|
||||
snd.lifted_index
|
||||
FROM qsos q
|
||||
INNER JOIN hrrr_profiles h1
|
||||
ON h1.lat = ROUND((q.pos1->>'lat')::numeric * 8) / 8
|
||||
AND h1.lon = ROUND((q.pos1->>'lng')::numeric * 8) / 8
|
||||
AND h1.valid_time = date_trunc('hour', q.qso_timestamp)
|
||||
INNER JOIN hrrr_profiles h2
|
||||
ON h2.lat = ROUND((q.pos2->>'lat')::numeric * 8) / 8
|
||||
AND h2.lon = ROUND((q.pos2->>'lng')::numeric * 8) / 8
|
||||
AND h2.valid_time = date_trunc('hour', q.qso_timestamp)
|
||||
LEFT JOIN solar_indices si ON si.date = q.qso_timestamp::date
|
||||
FROM ranked r
|
||||
LEFT JOIN solar_indices si ON si.date = r.valid_time::date
|
||||
LEFT JOIN LATERAL (
|
||||
SELECT s.k_index, s.lifted_index
|
||||
FROM soundings s
|
||||
WHERE s.observed_at BETWEEN q.qso_timestamp - interval '6 hours'
|
||||
AND q.qso_timestamp + interval '6 hours'
|
||||
ORDER BY ABS(EXTRACT(EPOCH FROM s.observed_at - q.qso_timestamp))
|
||||
WHERE s.observed_at BETWEEN r.valid_time - interval '6 hours'
|
||||
AND r.valid_time + interval '6 hours'
|
||||
ORDER BY ABS(EXTRACT(EPOCH FROM s.observed_at - r.valid_time))
|
||||
LIMIT 1
|
||||
) snd ON true
|
||||
WHERE q.distance_km > 0
|
||||
AND q.distance_km < 3000
|
||||
AND q.qso_timestamp >= '2016-06-30'
|
||||
AND q.pos1->>'lng' IS NOT NULL
|
||||
AND q.pos2->>'lng' IS NOT NULL
|
||||
WHERE r.rn <= $1
|
||||
"""
|
||||
|
||||
%{rows: rows} = Repo.query!(sql, [], timeout: @query_timeout)
|
||||
IO.puts(" Matched #{length(rows)} QSOs to HRRR+solar+sounding")
|
||||
%{rows: rows} = Repo.query!(sql, [per_month], timeout: @query_timeout)
|
||||
IO.puts(" #{length(rows)} profile rows after stratified sample")
|
||||
|
||||
band_counts = Enum.frequencies_by(rows, fn row -> Enum.at(row, 0) end)
|
||||
month_counts = Enum.frequencies_by(rows, fn row -> Enum.at(row, 11) end)
|
||||
|
||||
# Compute within-band distance percentile
|
||||
rows_by_band =
|
||||
rows
|
||||
|> Enum.filter(fn [band_mhz | _] -> Map.has_key?(@band_max_km, band_mhz) end)
|
||||
|> Enum.group_by(fn [band_mhz | _] -> band_mhz end)
|
||||
|
||||
percentile_rows =
|
||||
Enum.flat_map(rows_by_band, fn {_band, band_rows} ->
|
||||
distances = band_rows |> Enum.map(fn [_, d | _] -> d end) |> Enum.sort()
|
||||
n = length(distances)
|
||||
|
||||
Enum.map(band_rows, fn [_, distance_km | _] = row ->
|
||||
rank = Enum.count(distances, &(&1 <= distance_km))
|
||||
{row, rank / n}
|
||||
end)
|
||||
end)
|
||||
bands = BandConfig.all_freqs()
|
||||
|
||||
# Each profile row turns into one training row per band.
|
||||
{feature_rows, target_rows} =
|
||||
percentile_rows
|
||||
|> Enum.map(fn {row, percentile} -> encode_contact_row(row, percentile) end)
|
||||
rows
|
||||
|> Enum.flat_map(fn row -> encode_profile_rows(row, bands) end)
|
||||
|> Enum.unzip()
|
||||
|
||||
IO.puts(" After filtering: #{length(feature_rows)} training samples")
|
||||
|
||||
{Nx.tensor(feature_rows, type: :f32), Nx.tensor(target_rows, type: :f32), band_counts}
|
||||
{Nx.tensor(feature_rows, type: :f32), Nx.tensor(target_rows, type: :f32), month_counts}
|
||||
end
|
||||
|
||||
defp encode_contact_row(
|
||||
[
|
||||
band_mhz,
|
||||
_dist,
|
||||
utc_hour,
|
||||
month,
|
||||
lon,
|
||||
lat,
|
||||
temp,
|
||||
dewpoint,
|
||||
pressure,
|
||||
grad,
|
||||
hpbl,
|
||||
pwat,
|
||||
refractivity,
|
||||
ducting,
|
||||
sfi,
|
||||
kp_max,
|
||||
k_index,
|
||||
lifted_index
|
||||
],
|
||||
percentile
|
||||
) do
|
||||
features =
|
||||
Model.encode_features(%{
|
||||
surface_temp_c: to_float(temp),
|
||||
surface_dewpoint_c: to_float(dewpoint),
|
||||
surface_pressure_mb: to_float(pressure),
|
||||
min_refractivity_gradient: to_float(grad),
|
||||
hpbl_m: to_float(hpbl),
|
||||
pwat_mm: to_float(pwat),
|
||||
surface_refractivity: to_float(refractivity),
|
||||
latitude: to_float(lat),
|
||||
sfi: to_float(sfi),
|
||||
kp_max: to_float(kp_max),
|
||||
ducting_detected: ducting,
|
||||
k_index: to_float(k_index),
|
||||
lifted_index: to_float(lifted_index),
|
||||
utc_hour: to_float(utc_hour),
|
||||
month: trunc(to_float(month)),
|
||||
longitude: to_float(lon),
|
||||
freq_mhz: band_mhz
|
||||
})
|
||||
# Builds one training row per (profile, band). Target is the algorithm
|
||||
# composite score normalized to [0, 1]; features are the ML model's
|
||||
# 20-element encoding of the same conditions.
|
||||
defp encode_profile_rows(row, bands) do
|
||||
[
|
||||
_valid_time,
|
||||
lat,
|
||||
lon,
|
||||
temp_c,
|
||||
dewpoint_c,
|
||||
pressure_mb,
|
||||
grad,
|
||||
hpbl_m,
|
||||
pwat_mm,
|
||||
refractivity,
|
||||
ducting,
|
||||
month,
|
||||
utc_hour,
|
||||
sfi,
|
||||
kp_max,
|
||||
k_index,
|
||||
lifted_index
|
||||
] = row
|
||||
|
||||
{features, [percentile]}
|
||||
temp_c = to_float(temp_c)
|
||||
dewpoint_c = to_float(dewpoint_c)
|
||||
pressure_mb = to_float(pressure_mb)
|
||||
grad = to_float(grad)
|
||||
hpbl_m = to_float(hpbl_m)
|
||||
pwat_mm = to_float(pwat_mm)
|
||||
refractivity = to_float(refractivity)
|
||||
lat_f = to_float(lat)
|
||||
lon_f = to_float(lon)
|
||||
sfi_f = to_float(sfi)
|
||||
kp_f = to_float(kp_max)
|
||||
k_idx_f = to_float(k_index)
|
||||
li_f = to_float(lifted_index)
|
||||
|
||||
scorer_conditions =
|
||||
build_scorer_conditions(
|
||||
temp_c,
|
||||
dewpoint_c,
|
||||
pressure_mb,
|
||||
grad,
|
||||
hpbl_m,
|
||||
pwat_mm,
|
||||
lat_f,
|
||||
lon_f,
|
||||
trunc(to_float(month)),
|
||||
trunc(to_float(utc_hour))
|
||||
)
|
||||
|
||||
Enum.map(bands, fn freq_mhz ->
|
||||
band_config = BandConfig.get(freq_mhz)
|
||||
%{score: score} = Scorer.composite_score(scorer_conditions, band_config)
|
||||
|
||||
features =
|
||||
Model.encode_features(%{
|
||||
surface_temp_c: temp_c,
|
||||
surface_dewpoint_c: dewpoint_c,
|
||||
surface_pressure_mb: pressure_mb,
|
||||
min_refractivity_gradient: grad,
|
||||
hpbl_m: hpbl_m,
|
||||
pwat_mm: pwat_mm,
|
||||
surface_refractivity: refractivity,
|
||||
latitude: lat_f,
|
||||
sfi: sfi_f,
|
||||
kp_max: kp_f,
|
||||
ducting_detected: ducting,
|
||||
k_index: k_idx_f,
|
||||
lifted_index: li_f,
|
||||
utc_hour: to_float(utc_hour),
|
||||
longitude: lon_f,
|
||||
month: trunc(to_float(month)),
|
||||
freq_mhz: freq_mhz
|
||||
})
|
||||
|
||||
{features, [score / 100.0]}
|
||||
end)
|
||||
end
|
||||
|
||||
# ── Helpers ───────────────────────────────────────────────────────
|
||||
# Builds a conditions map matching the shape Scorer.composite_score/2
|
||||
# expects. Sky / wind / rain default to nil → Scorer uses neutral
|
||||
# values (50, 50, 100). The ML model's feature set also omits those
|
||||
# channels, so the missing-factor contribution is constant across all
|
||||
# samples and the model learns the weather-derived residual cleanly.
|
||||
defp build_scorer_conditions(temp_c, dewpoint_c, pressure_mb, grad, hpbl_m, pwat_mm, lat, lon, month, utc_hour) do
|
||||
%{
|
||||
abs_humidity: Scorer.absolute_humidity(temp_c, dewpoint_c),
|
||||
temp_f: Scorer.c_to_f(temp_c),
|
||||
dewpoint_f: Scorer.c_to_f(dewpoint_c),
|
||||
pressure_mb: pressure_mb,
|
||||
prev_pressure_mb: nil,
|
||||
min_refractivity_gradient: grad,
|
||||
bl_depth_m: hpbl_m,
|
||||
pwat_mm: pwat_mm,
|
||||
sky_cover_pct: nil,
|
||||
wind_speed_kts: nil,
|
||||
rain_rate_mmhr: 0.0,
|
||||
utc_hour: utc_hour,
|
||||
utc_minute: 0,
|
||||
month: month,
|
||||
longitude: lon,
|
||||
latitude: lat,
|
||||
best_duct_band_ghz: nil
|
||||
}
|
||||
end
|
||||
|
||||
# ── Post-training sanity check ──────────────────────────────────────
|
||||
|
||||
# After training, evaluate the model on a synthetic set of
|
||||
# (temp, month) combinations at otherwise-identical conditions. The
|
||||
# goal is to confirm the model doesn't prefer / penalize specific
|
||||
# months independently of the weather they carry — i.e. that a
|
||||
# generic March day doesn't rank below a generic June day just
|
||||
# because the training data had fewer March contacts.
|
||||
defp check_monthly_bias(params) do
|
||||
IO.puts("\nMonthly sanity check at fixed conditions (10 GHz, 15°C, 10°C dewpoint)...")
|
||||
predict_fn = Model.compile_predict()
|
||||
|
||||
for month <- 1..12 do
|
||||
conditions = %{
|
||||
surface_temp_c: 15.0,
|
||||
surface_dewpoint_c: 10.0,
|
||||
surface_pressure_mb: 1013.0,
|
||||
min_refractivity_gradient: -70.0,
|
||||
hpbl_m: 800.0,
|
||||
pwat_mm: 20.0,
|
||||
surface_refractivity: 320.0,
|
||||
latitude: 32.5,
|
||||
longitude: -97.0,
|
||||
sfi: 120.0,
|
||||
kp_max: 2.0,
|
||||
ducting_detected: false,
|
||||
k_index: 25.0,
|
||||
lifted_index: 0.0,
|
||||
utc_hour: 12,
|
||||
month: month,
|
||||
freq_mhz: 10_000
|
||||
}
|
||||
|
||||
score = Model.predict_score(predict_fn, params, conditions)
|
||||
IO.puts(" month=#{month |> Integer.to_string() |> String.pad_leading(2)}: #{score}")
|
||||
end
|
||||
end
|
||||
|
||||
# ── Helpers ──────────────────────────────────────────────────────────
|
||||
|
||||
defp print_header(per_month, epochs, batch_size, lr) do
|
||||
"=" |> String.duplicate(70) |> IO.puts()
|
||||
IO.puts("PROPAGATION MODEL TRAINING — month-balanced HRRR → algorithm score")
|
||||
"=" |> String.duplicate(70) |> IO.puts()
|
||||
IO.puts("Samples per month: #{per_month} epochs: #{epochs} batch: #{batch_size} lr: #{lr}")
|
||||
IO.puts("Started: #{DateTime.to_string(DateTime.utc_now())}\n")
|
||||
end
|
||||
|
||||
defp print_month_counts(counts) do
|
||||
counts
|
||||
|> Enum.sort_by(fn {m, _} -> m end)
|
||||
|> Enum.each(fn {m, count} ->
|
||||
IO.puts(" month #{m |> Integer.to_string() |> String.pad_leading(2)}: #{count}")
|
||||
end)
|
||||
end
|
||||
|
||||
defp shuffle(features, targets, n) do
|
||||
key = Nx.Random.key(System.os_time())
|
||||
|
|
@ -408,14 +374,6 @@ defmodule Mix.Tasks.PropagationTrain do
|
|||
}
|
||||
end
|
||||
|
||||
defp print_band_counts(counts) do
|
||||
counts
|
||||
|> Enum.sort_by(fn {band, _} -> band end)
|
||||
|> Enum.each(fn {band, count} ->
|
||||
IO.puts(" #{div(band, 1000)} GHz: #{count}")
|
||||
end)
|
||||
end
|
||||
|
||||
defp print_loss(loss) do
|
||||
if is_float(loss) and loss == loss do
|
||||
IO.puts(" Final loss: #{Float.round(loss, 6)}")
|
||||
|
|
|
|||
Binary file not shown.
Loading…
Add table
Reference in a new issue