Validate live pool ranking order

scripts/run_release_status_summary.py

@@ -74,6 +74,21 @@ def _safe_int(value: Any, default: int = 0) -> int:
         return default
 
 
+def _ranking_preview(ranking: pd.DataFrame, size: int) -> pd.DataFrame:
+    preview_size = max(0, int(size))
+    if preview_size == 0:
+        return ranking.head(0)
+
+    if "current_rank" not in ranking.columns:
+        return ranking.head(preview_size)
+
+    ordered = ranking.copy()
+    ordered["_current_rank_numeric"] = pd.to_numeric(ordered["current_rank"], errors="coerce")
+    if ordered["_current_rank_numeric"].notna().any():
+        ordered = ordered.sort_values("_current_rank_numeric", na_position="last", kind="mergesort")
+    return ordered.drop(columns=["_current_rank_numeric"], errors="ignore").head(preview_size)
+
+
 def build_release_status_payload(
     output_dir: Path | str,
     *,
@@ -103,7 +118,7 @@ def build_release_status_payload(
 
     selected_mask = ranking["selected_flag"].map(_coerce_bool) if "selected_flag" in ranking.columns else pd.Series(dtype=bool)
     ranking_preview_rows = []
-    preview = ranking.head(max(0, int(ranking_preview_size)))
+    preview = _ranking_preview(ranking, ranking_preview_size)
     for _, row in preview.iterrows():
         ranking_preview_rows.append(
             {

src/release_contract.py

@@ -174,6 +174,43 @@ def _coerce_selected_flag(series: pd.Series) -> pd.Series:
     )
 
 
+def _selected_symbols_ordered_by_rank(
+    latest_ranking: pd.DataFrame,
+    selected_mask: pd.Series,
+    errors: list[str],
+) -> list[str]:
+    selected_rows = latest_ranking.loc[selected_mask, ["symbol", "current_rank"]].copy()
+    if selected_rows.empty:
+        return []
+
+    selected_rows["_symbol_normalized"] = selected_rows["symbol"].astype(str).str.strip().str.upper()
+    selected_rows["_current_rank_numeric"] = pd.to_numeric(selected_rows["current_rank"], errors="coerce")
+
+    invalid_rank_symbols = selected_rows.loc[
+        selected_rows["_current_rank_numeric"].isna(),
+        "_symbol_normalized",
+    ].tolist()
+    if invalid_rank_symbols:
+        errors.append(
+            "latest_ranking.csv selected rows must have numeric current_rank values: "
+            + ", ".join(invalid_rank_symbols)
+        )
+        return []
+
+    duplicated_rank_symbols = selected_rows.loc[
+        selected_rows["_current_rank_numeric"].duplicated(keep=False),
+        "_symbol_normalized",
+    ].tolist()
+    if duplicated_rank_symbols:
+        errors.append(
+            "latest_ranking.csv selected current_rank values must be unique: "
+            + ", ".join(duplicated_rank_symbols)
+        )
+
+    ordered = selected_rows.sort_values("_current_rank_numeric", kind="mergesort")
+    return ordered["_symbol_normalized"].tolist()
+
+
 def validate_release_outputs(
     output_dir: Path | str,
     *,
@@ -399,10 +436,17 @@ def validate_release_outputs(
                 errors,
             )
         )
+        selected_symbols_by_rank = _selected_symbols_ordered_by_rank(latest_ranking, selected_mask, errors)
         if live_pool_symbols and not set(live_pool_symbols).issubset(set(ranking_symbols)):
             errors.append("live_pool.json symbols must all be present in latest_ranking.csv")
         if live_pool_symbols and not set(live_pool_symbols).issubset(selected_symbols):
             errors.append("live_pool.json symbols must all be selected in latest_ranking.csv")
+        if live_pool_symbols and selected_symbols_by_rank:
+            expected_live_pool_symbols = selected_symbols_by_rank[: len(live_pool_symbols)]
+            if live_pool_symbols != expected_live_pool_symbols:
+                errors.append(
+                    "live_pool.json symbols must match selected latest_ranking.csv symbols ordered by current_rank"
+                )
 
     if manifest_present:
         manifest_mode = str(manifest.get("mode", "")).strip()

tests/test_release_contract.py

@@ -217,6 +217,37 @@ def test_validate_release_outputs_rejects_mismatched_manifest_payload(self) -> N
             validation["errors"],
         )
 
+    def test_validate_release_outputs_rejects_live_pool_order_mismatch(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            root = Path(tmp_dir)
+            self.build_outputs(root)
+            output_dir = root / "data" / "output"
+            live_pool_path = output_dir / "live_pool.json"
+            artifact_manifest_path = output_dir / "artifact_manifest.json"
+
+            live_pool = json.loads(live_pool_path.read_text(encoding="utf-8"))
+            live_pool["symbols"] = [
+                "TRXUSDT",
+                "ETHUSDT",
+                "NEARUSDT",
+                "BCHUSDT",
+                "SOLUSDT",
+            ]
+            write_json(live_pool_path, live_pool)
+
+            artifact_manifest = json.loads(artifact_manifest_path.read_text(encoding="utf-8"))
+            artifact_manifest["symbols"] = live_pool["symbols"]
+            artifact_manifest["artifacts"]["live_pool"]["sha256"] = sha256_file(live_pool_path)
+            write_json(artifact_manifest_path, artifact_manifest)
+
+            validation = validate_release_outputs(root / "data" / "output", require_artifact_manifest=True)
+
+        self.assertFalse(validation["ok"])
+        self.assertIn(
+            "live_pool.json symbols must match selected latest_ranking.csv symbols ordered by current_rank",
+            validation["errors"],
+        )
+
     def test_validate_release_outputs_rejects_stale_outputs_when_required(self) -> None:
         with tempfile.TemporaryDirectory() as tmp_dir:
             root = Path(tmp_dir)

tests/test_release_status_summary.py

@@ -176,6 +176,31 @@ def test_build_release_status_payload_reports_ok_for_consistent_release(self) ->
         self.assertEqual(len(payload["artifact_summary"]["ranking_preview"]), 3)
         self.assertTrue(payload["validation"]["ok"])
 
+    def test_build_release_status_payload_orders_preview_by_current_rank(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp_dir:
+            output_dir = self.write_outputs(Path(tmp_dir))
+            ranking_path = output_dir / "latest_ranking.csv"
+            ranking = pd.read_csv(ranking_path)
+            ranking = ranking.iloc[[2, 0, 1, 3, 4]]
+            ranking.to_csv(ranking_path, index=False)
+            artifact_manifest_path = output_dir / "artifact_manifest.json"
+            artifact_manifest = json.loads(artifact_manifest_path.read_text(encoding="utf-8"))
+            artifact_manifest["artifacts"]["latest_ranking"]["sha256"] = sha256_file(ranking_path)
+            write_json(artifact_manifest_path, artifact_manifest)
+
+            payload = MODULE.build_release_status_payload(
+                output_dir,
+                max_age_days=45,
+                require_freshness=False,
+                ranking_preview_size=3,
+                reference_date="2026-04-01",
+            )
+
+        self.assertEqual(
+            [row["symbol"] for row in payload["artifact_summary"]["ranking_preview"]],
+            ["TRXUSDT", "ETHUSDT", "BCHUSDT"],
+        )
+
     def test_build_release_status_payload_reports_error_when_manifest_missing(self) -> None:
         with tempfile.TemporaryDirectory() as tmp_dir:
             output_dir = self.write_outputs(Path(tmp_dir), include_manifest=False)