fix: require monthly shadow build and deterministic tiebreak

.github/workflows/monthly_publish.yml

@@ -52,8 +52,12 @@ jobs:
       - name: Download Or Update Raw History
         run: python scripts/download_history.py --top-liquid "${DOWNLOAD_TOP_LIQUID}" --force-exchange-info
 
-      - name: Build Production v1 Live Pool
-        run: python scripts/build_live_pool.py --universe-mode "${PUBLISH_MODE}"
+      - name: Build Production v1 Live Pool And Shadow Tracks
+        run: |
+          python scripts/run_monthly_shadow_build.py \
+            --universe-mode "${PUBLISH_MODE}" \
+            --shadow-universe-mode "${PUBLISH_MODE}" \
+            --skip-publish-dry-run
 
       - name: Publish Production v1 Release
         run: python scripts/publish_release.py --mode "${PUBLISH_MODE}"

README.md

@@ -427,6 +427,8 @@ The monthly operator workflow is now:
 2. run the baseline publish dry-run check
 3. refresh the dual-track shadow candidate histories
 
+The GitHub monthly publish workflow now runs this shadow-build wrapper before the real publish step, so the monthly report and AI review always receive same-cycle `official_baseline` and `challenger_topk_60` coverage.
+
 Canonical command:
 
 ```bash
@@ -463,6 +465,14 @@ Track identity fields to rely on:
 
 Baseline remains the official production reference. `challenger_topk_60` remains shadow-only.
 
+Monthly ranking tie-break rule for `core_major` live exports:
+
+1. `final_score` descending
+2. `confidence` descending
+3. `liquidity_stability` descending
+4. `avg_quote_vol_180` descending
+5. `symbol` ascending
+
 ## Monthly Build Telegram Notify
 
 Optional short build/publish health notification:

docs/operator_runbook.md

@@ -61,6 +61,7 @@ Operator-facing summary entrypoints:
 - `scripts/run_monthly_build_telegram.py` for the optional short Telegram health notification or local preview text
 - `scripts/run_monthly_report_bundle.py` for the standard monthly report bundle used by Actions artifacts and AI review handoff
 - `scripts/write_release_heartbeat.py` for the lightweight logs-branch heartbeat record
+- Monthly live-pool ordering uses a deterministic tie-break: `final_score`, then `confidence`, then `liquidity_stability`, then `avg_quote_vol_180`, then `symbol`
 
 Boundary rules:
 

src/export.py

@@ -4,6 +4,7 @@
 
 import pandas as pd
 
+from .ranking import sort_ranking_snapshot
 from .utils import date_to_str, write_json
 
 
@@ -20,6 +21,7 @@ def export_latest_ranking(panel: pd.DataFrame, output_dir: str | Any, as_of_date
     """Export the latest ranking cross section to CSV."""
     snapshot = panel.xs(as_of_date, level="date").copy()
     snapshot = snapshot.loc[snapshot["in_universe"] | snapshot["selected_flag"]].copy()
+    snapshot = sort_ranking_snapshot(snapshot)
     snapshot["as_of_date"] = date_to_str(as_of_date)
     snapshot["symbol"] = snapshot.index
     columns = [
@@ -34,7 +36,7 @@ def export_latest_ranking(panel: pd.DataFrame, output_dir: str | Any, as_of_date
         "selected_flag",
         "current_rank",
     ]
-    exported = snapshot[columns].sort_values("final_score", ascending=False).reset_index(drop=True)
+    exported = snapshot[columns].reset_index(drop=True)
     exported.to_csv(output_dir / "latest_ranking.csv", index=False)
     return exported
 
@@ -62,7 +64,7 @@ def build_live_pool_payload(
     selection_meta_fields: list[str] | None = None,
 ) -> tuple[dict[str, Any], dict[str, Any]]:
     """Build additive live-pool payloads without performing I/O."""
-    selected = ranking_snapshot.sort_values("final_score", ascending=False).head(pool_size).copy()
+    selected = sort_ranking_snapshot(ranking_snapshot).head(pool_size).copy()
     symbols = selected.index.tolist()
     metadata_indexed = metadata.set_index("symbol")
     as_of_date_str = date_to_str(as_of_date)

src/ranking.py

@@ -9,6 +9,25 @@
 from .utils import normalize_component_by_date
 
 
+def sort_ranking_snapshot(snapshot: pd.DataFrame) -> pd.DataFrame:
+    """Apply a deterministic ranking order with explicit tie-breaks."""
+    ordered = snapshot.copy()
+    added_columns: list[str] = []
+    for column in ("confidence", "liquidity_stability", "avg_quote_vol_180"):
+        if column not in ordered.columns:
+            ordered[column] = np.nan
+            added_columns.append(column)
+
+    ordered["_sort_symbol"] = pd.Index(ordered.index).astype(str).str.upper()
+    ordered = ordered.sort_values(
+        ["final_score", "confidence", "liquidity_stability", "avg_quote_vol_180", "_sort_symbol"],
+        ascending=[False, False, False, False, True],
+        na_position="last",
+        kind="mergesort",
+    )
+    return ordered.drop(columns=["_sort_symbol", *added_columns], errors="ignore")
+
+
 def merge_predictions(panel: pd.DataFrame, predictions: pd.DataFrame) -> pd.DataFrame:
     """Attach model prediction columns to the main panel."""
     if predictions.empty:
@@ -68,9 +87,9 @@ def build_final_scores(panel: pd.DataFrame, config: dict[str, Any]) -> pd.DataFr
         eligible = group.loc[group["in_universe"] & group["final_score"].notna()].copy()
         if eligible.empty:
             continue
-        ranks = eligible["final_score"].rank(ascending=False, method="first")
-        panel.loc[ranks.index, "current_rank"] = ranks
-        selected = eligible["final_score"].nlargest(pool_size)
+        ordered = sort_ranking_snapshot(eligible)
+        panel.loc[ordered.index, "current_rank"] = np.arange(1, len(ordered) + 1, dtype=float)
+        selected = ordered.head(pool_size)
         panel.loc[selected.index, "selected_flag"] = True
 
     if "prediction_window_count" in panel.columns:
@@ -87,5 +106,5 @@ def latest_ranking_snapshot(panel: pd.DataFrame, as_of_date: pd.Timestamp | str)
     """Return one date slice sorted by the current final score."""
     snapshot = panel.xs(pd.Timestamp(as_of_date), level="date").copy()
     if "final_score" in snapshot.columns:
-        snapshot = snapshot.sort_values("final_score", ascending=False)
+        snapshot = sort_ranking_snapshot(snapshot)
     return snapshot

tests/test_monthly_publish_workflow_config.py

@@ -27,6 +27,9 @@ def test_monthly_review_issue_creation_does_not_require_gh_cli(self) -> None:
         self.assertNotIn("gh label create", workflow)
         self.assertNotIn("gh issue create", workflow)
         self.assertNotIn("gh workflow run", workflow)
+        self.assertIn("run_monthly_shadow_build.py", workflow)
+        self.assertIn("--skip-publish-dry-run", workflow)
+        self.assertIn("--shadow-universe-mode", workflow)
         self.assertIn("https://api.github.com/repos/{repository}", workflow)
         self.assertIn('GITHUB_TOKEN: ${{ secrets.GITHUB_TOKEN }}', workflow)
         self.assertIn("issue_number=", workflow)

tests/test_ranking.py

@@ -0,0 +1,79 @@
+from __future__ import annotations
+
+import unittest
+from unittest.mock import patch
+
+import pandas as pd
+
+from src.export import build_live_pool_payload
+from src.ranking import build_final_scores, latest_ranking_snapshot
+
+
+class RankingTieBreakTests(unittest.TestCase):
+    def test_tie_break_prefers_confidence_then_liquidity_then_symbol(self) -> None:
+        as_of_date = pd.Timestamp("2026-04-01")
+        index = pd.MultiIndex.from_tuples(
+            [
+                (as_of_date, "AAAUSDT"),
+                (as_of_date, "BBBUSDT"),
+                (as_of_date, "CCCUSDT"),
+            ],
+            names=["date", "symbol"],
+        )
+        panel = pd.DataFrame(
+            {
+                "in_universe": [True, True, True],
+                "rule_score": [1.0, 0.5, 0.5],
+                "linear_score_raw": [0.0, 0.5, 0.5],
+                "ml_score_raw": [0.5, 0.5, 0.5],
+                "regime": ["risk_off", "risk_off", "risk_off"],
+                "liquidity_stability": [0.70, 0.90, 0.80],
+                "avg_quote_vol_180": [20_000_000.0, 30_000_000.0, 30_000_000.0],
+            },
+            index=index,
+        )
+        config = {
+            "ensemble": {"default_weights": {"rule_score": 1.0, "linear_score": 1.0, "ml_score": 1.0}},
+            "regime_weights": {},
+            "ranking": {"selected_pool_size": 2},
+        }
+
+        with patch("src.ranking.normalize_component_by_date", side_effect=lambda frame, column, mask: frame[column]):
+            scored = build_final_scores(panel, config)
+
+        snapshot = latest_ranking_snapshot(scored, as_of_date)
+        self.assertEqual(snapshot.index.tolist(), ["BBBUSDT", "CCCUSDT", "AAAUSDT"])
+        self.assertEqual(snapshot["current_rank"].tolist(), [1.0, 2.0, 3.0])
+        self.assertEqual(snapshot.loc[snapshot["selected_flag"]].index.tolist(), ["BBBUSDT", "CCCUSDT"])
+
+    def test_live_pool_payload_uses_same_deterministic_tie_break(self) -> None:
+        as_of_date = pd.Timestamp("2026-04-01")
+        ranking_snapshot = pd.DataFrame(
+            {
+                "final_score": [0.5, 0.5, 0.5],
+                "confidence": [0.6, 0.6, 0.6],
+                "liquidity_stability": [0.80, 0.80, 0.80],
+                "avg_quote_vol_180": [15_000_000.0, 25_000_000.0, 25_000_000.0],
+            },
+            index=pd.Index(["CCCUSDT", "BBBUSDT", "AAAUSDT"], name="symbol"),
+        )
+        metadata = pd.DataFrame(
+            {
+                "symbol": ["AAAUSDT", "BBBUSDT", "CCCUSDT"],
+                "base_asset": ["AAA", "BBB", "CCC"],
+            }
+        )
+
+        payload, _ = build_live_pool_payload(
+            ranking_snapshot=ranking_snapshot,
+            metadata=metadata,
+            as_of_date=as_of_date,
+            pool_size=2,
+            mode="core_major",
+        )
+
+        self.assertEqual(payload["symbols"], ["AAAUSDT", "BBBUSDT"])
+
+
+if __name__ == "__main__":
+    unittest.main()