Train ML model on real QSO-HRRR data instead of algorithm scores

Features: HRRR conditions averaged at both QSO endpoints + solar time
Target: distance_km normalized per-band (distance / p99_range, capped at 1.0)
This trains on actual propagation outcomes from 57K+ QSOs, not the
hand-tuned algorithm output.

lib/mix/tasks/propagation_train.ex

@@ -1,32 +1,45 @@
 defmodule Mix.Tasks.PropagationTrain do
-  @shortdoc "Train ML model from propagation scores + HRRR data"
+  @shortdoc "Train ML model from QSO-HRRR matched data"
 
   @moduledoc """
-  Builds training data from propagation_scores joined to hrrr_profiles,
+  Trains the propagation ML model on real QSO outcomes matched to HRRR
-  trains the Axon model, evaluates on held-out test set, and saves weights.
+  atmospheric conditions. The target is QSO distance normalized per-band
+  (0-1), so the model learns what atmospheric conditions actually produce
+  good propagation from historical contacts — not from the hand-tuned algorithm.
 
-  Uses stratified sampling across all 8 bands so the model learns
+  Features: HRRR conditions averaged across both QSO endpoints + solar time + frequency.
-  frequency-specific scoring behavior via the log_freq_mhz feature.
+  Target: distance_km / band_p99_distance (capped at 1.0).
 
   ## Usage
 
       mix propagation_train
-      mix propagation_train --epochs 100 --batch-size 512 --sample 500000
+      mix propagation_train --epochs 100 --batch-size 512
   """
 
   use Mix.Task
 
-  alias Microwaveprop.Propagation.BandConfig
   alias Microwaveprop.Propagation.Model
   alias Microwaveprop.Repo
 
   @query_timeout 600_000
 
+  # Per-band normalization: p99 distances from the QSO dataset.
+  # Distance is divided by these values and capped at 1.0.
+  @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: [epochs: :integer, batch_size: :integer, sample: :integer]
+        strict: [epochs: :integer, batch_size: :integer]
       )
 
     Application.put_env(
@@ -41,26 +54,27 @@ defmodule Mix.Tasks.PropagationTrain do
 
     Mix.Task.run("app.start")
 
-    epochs = Keyword.get(opts, :epochs, 50)
+    epochs = Keyword.get(opts, :epochs, 100)
     batch_size = Keyword.get(opts, :batch_size, 256)
-    sample_size = Keyword.get(opts, :sample, 500_000)
 
     "=" |> String.duplicate(70) |> IO.puts()
-    IO.puts("PROPAGATION MODEL TRAINING")
+    IO.puts("PROPAGATION MODEL TRAINING (QSO-HRRR)")
     "=" |> String.duplicate(70) |> IO.puts()
-    IO.puts("Config: epochs=#{epochs}, batch_size=#{batch_size}, sample=#{sample_size}")
+    IO.puts("Config: epochs=#{epochs}, batch_size=#{batch_size}")
     IO.puts("Started: #{DateTime.to_string(DateTime.utc_now())}\n")
 
-    IO.puts("Loading training data (stratified, sample=#{sample_size})...")
+    IO.puts("Loading QSO-HRRR training data...")
-    {features, targets, band_counts} = load_training_data(sample_size)
+    {features, targets, band_counts} = load_training_data()
     n = elem(Nx.shape(features), 0)
 
-    IO.puts("Loaded #{n} samples across #{map_size(band_counts)} bands:")
+    IO.puts("\nLoaded #{n} training samples from real QSOs:")
 
     band_counts
     |> Enum.sort_by(fn {band, _} -> band end)
     |> Enum.each(fn {band, count} ->
-      IO.puts("  #{div(band, 1000)} GHz: #{count} samples")
+      max_km = Map.get(@band_max_km, band)
+      suffix = if max_km, do: " (norm: #{trunc(max_km)} km)", else: " (excluded)"
+      IO.puts("  #{div(band, 1000)} GHz: #{count} QSOs#{suffix}")
     end)
 
     IO.puts("")
@@ -120,45 +134,59 @@ defmodule Mix.Tasks.PropagationTrain do
     IO.puts("Finished: #{DateTime.to_string(DateTime.utc_now())}")
   end
 
-  defp load_training_data(sample_size) do
+  defp load_training_data do
-    bands = BandConfig.all_freqs()
+    # Join QSOs to HRRR at both endpoints, average conditions.
-    per_band = div(sample_size, length(bands))
+    # Same query as the analysis task but returns raw values for encoding.
+    sql = """
+    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,
+      (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
+    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)
+    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
+    """
 
-    band_queries =
+    IO.puts("  Running QSO-HRRR join query...")
-      Enum.map_join(bands, "\nUNION ALL\n", fn band_mhz ->
+    %{rows: rows} = Repo.query!(sql, [], timeout: @query_timeout)
-        """
+    IO.puts("  Matched #{length(rows)} QSOs to HRRR conditions")
-        (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,
-          EXTRACT(HOUR FROM ps.valid_time) AS utc_hour,
-          EXTRACT(MONTH FROM ps.valid_time) AS month,
-          ps.lon
-        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
-        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, 0) end)
-
-    band_counts =
-      Enum.frequencies_by(rows, fn row -> Enum.at(row, 1) end)
 
     {feature_rows, target_rows} =
       rows
-      |> Enum.map(fn [score, band_mhz, temp_c, dewpoint_c, pressure_mb, grad, hpbl, pwat, utc_hour, month, lon] ->
+      |> Enum.filter(fn [band_mhz | _] -> Map.has_key?(@band_max_km, band_mhz) end)
+      |> Enum.map(fn [band_mhz, distance_km, utc_hour, month, lon, temp, dewpoint, pressure, grad, hpbl, pwat] ->
         features =
           Model.encode_features(%{
-            surface_temp_c: to_float(temp_c),
+            surface_temp_c: to_float(temp),
-            surface_dewpoint_c: to_float(dewpoint_c),
+            surface_dewpoint_c: to_float(dewpoint),
-            surface_pressure_mb: to_float(pressure_mb),
+            surface_pressure_mb: to_float(pressure),
             min_refractivity_gradient: to_float(grad),
             hpbl_m: to_float(hpbl),
             pwat_mm: to_float(pwat),
@@ -168,12 +196,16 @@ defmodule Mix.Tasks.PropagationTrain do
             freq_mhz: band_mhz
           })
 
-        target = [score / 100.0]
+        # Normalize distance by band's p99 range, cap at 1.0
+        max_km = Map.fetch!(@band_max_km, band_mhz)
+        target = [min(distance_km / max_km, 1.0)]
 
         {features, target}
       end)
       |> Enum.unzip()
 
+    IO.puts("  After filtering: #{length(feature_rows)} training samples")
+
     features_tensor = Nx.tensor(feature_rows, type: :f32)
     targets_tensor = Nx.tensor(target_rows, type: :f32)