fix(data): judge late newest rows as sparse; require SPY print only in the ETF panel

Tickers count as alive for 10 rows after their last bar, so a newest row
printed by a few early tickers is dropped until the rest arrive. The
benchmark-print session rule now applies only to the mixed US/HK ETF panel:
Yahoo's 510300.SS lags CN stocks by days and would stall the CN book.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

data_manager.py

@@ -34,6 +34,9 @@ REFRESH_DAYS = 30
 # A row is kept only if at least this fraction of tickers alive on that date
 # have a bar. Drops e.g. US-holiday rows created by HK-listed tickers.
 MIN_ROW_COVERAGE = 0.5
+# A ticker counts as alive for this many rows after its last bar, so a newest
+# row printed by only a few tickers (late data) is still judged sparse.
+ALIVE_GRACE_ROWS = 10
 # Relative change in the adjusted-price overlap that triggers a rescale.
 RESCALE_TOLERANCE = 1e-3
 UNAVAILABLE_RETRY_DAYS = 7
@@ -203,7 +206,7 @@ def _drop_sparse_rows(df: pd.DataFrame) -> pd.DataFrame:
     first = np.where(has_any, notna.argmax(axis=0), n)
     last = np.where(has_any, n - 1 - notna[::-1].argmax(axis=0), -1)
     pos = np.arange(n)[:, None]
-    alive = (pos >= first) & (pos <= last)
+    alive = (pos >= first) & (pos <= last + ALIVE_GRACE_ROWS)
     coverage = notna.sum(axis=1) / np.maximum(alive.sum(axis=1), 1)
     keep = coverage >= MIN_ROW_COVERAGE
     if not keep.all():

tests/test_live_pipeline.py

@@ -42,6 +42,13 @@ class PanelRefreshTests(unittest.TestCase):
         self.assertTrue((returns > 0).all(), returns)
         self.assertAlmostEqual(merged["AAA"].iloc[0], 50.0)
 
+    def test_newest_row_printed_by_few_tickers_is_dropped(self):
+        panel = _panel({
+            "A": [1, 1, 1, np.nan], "B": [1, 1, 1, np.nan],
+            "C": [1, 1, 1, np.nan], "D": [1, 1, 1, 2],
+        })
+        self.assertNotIn(panel.index[-1], data_manager._drop_sparse_rows(panel).index)
+
     def test_rows_printed_by_a_minority_of_tickers_are_dropped(self):
         panel = _panel({
             "SPY": [1, 1, np.nan, 1, 1],

trader.py

@@ -362,12 +362,13 @@ def completed_sessions(close_raw: pd.DataFrame, data_market: str, benchmark: str
                        now: datetime | None = None) -> pd.DatetimeIndex:
     """Trading days whose closing prices are final.
 
-    A day counts when the benchmark printed a bar on it. Today's bar is
-    excluded until the exchange has closed and settled, because before that
-    it is an intraday snapshot rather than a close.
+    Stored panel rows are trading days (sparse rows are dropped on update).
+    The ETF panel mixes US and HK listings, so there a day also needs a
+    benchmark (SPY) print. Today's bar is excluded until the exchange has
+    closed and settled, because before that it is an intraday snapshot.
     """
     index = close_raw.index
-    if benchmark in close_raw.columns:
+    if data_market == "etfs" and benchmark in close_raw.columns:
         index = index[close_raw[benchmark].notna().to_numpy()]
     tz_name, hour, minute = MARKET_SESSION.get(data_market, MARKET_SESSION["us"])
     now_local = (now or datetime.now(timezone.utc)).astimezone(zoneinfo.ZoneInfo(tz_name))