fix(tools): northbound flow stops serving a silent empty envelope off a dead feed

Eastmoney's kamt northbound net-flow feed died with the 2024-08-30 HKEX
disclosure change (all-zero nets, frozen cumulative, drifted payload
shape), but its HTTP calls still succeed, so get_northbound_flow
returned ok:true with null realtime and empty history and never reached
the tushare fallback (#1481).

- northbound_tool: detect the dead state (realtime AND history both
  empty/all-zero) and route it through a shared tushare-fallback
  envelope; without a usable fallback the tool now returns an honest
  ok:false naming the disclosure change instead of an empty success.
- tushare_fallbacks: drop the blanket x100 rescale. Since 2024-08-30
  tushare's hgt/sgt/north_money carry turnover in million CNY (verified
  equal to HKEX official Total Turnover to the cent on three anchor
  days); earlier rows are net buy in the same unit. Values now pass
  through unchanged with unit "CNY million" and a note documenting the
  semantics boundary.
- tests: dead-feed fall-through, honest error without tushare, healthy
  payload guard, one-sided-dead guard; the x100 pin is replaced by the
  passthrough+note pin.

Closes #1481

Signed-off-by: shadowinlife <shadowinlife@gmail.com>

agent/src/tools/northbound_tool.py

@@ -153,6 +153,32 @@ def _parse_history(payload: Any, lookback_days: int) -> list[dict[str, Any]]:
     return rows[-lookback_days:]
 
 
+def _is_dead_feed(
+    realtime: dict[str, float | None], history: list[dict[str, Any]]
+) -> bool:
+    """True when the Eastmoney payload carries no live northbound signal.
+
+    HKEX stopped publishing northbound daily net buy on 2024-08-30; the
+    Eastmoney ``kamt`` feed has returned all-zero nets with a frozen
+    cumulative ever since, and its payload shape drifted (``netBuyAmt`` →
+    ``dayNetAmtIn``, ``klines`` → per-channel arrays), so the parsed
+    envelope comes back empty or all-zero while the HTTP calls succeed
+    (issue #1481). Treat that state as provider failure so the tushare
+    fallback serves the turnover-era values instead of a silent empty
+    success. Both sides must be dead: a genuine trading halt can zero one
+    side, but a live feed always carries either a realtime figure or a
+    non-zero recent history row.
+    """
+    realtime_dead = all(
+        realtime.get(key) in (None, 0.0)
+        for key in ("shanghai_connect", "shenzhen_connect", "total")
+    )
+    history_dead = not history or all(
+        row.get("total") in (None, 0.0) for row in history
+    )
+    return realtime_dead and history_dead
+
+
 def _clamp_lookback(value: Any) -> int:
     """Clamp a requested lookback to ``[1, _MAX_LOOKBACK_DAYS]``.
 
@@ -181,8 +207,10 @@ class NorthboundFlowTool(BaseTool):
         "MARKET-WIDE Northbound (Stock-Connect / 北向) net capital flow for the "
         "whole mainland China A-share market: the aggregate net inflow from Hong "
         "Kong, split into Shanghai-Connect (沪股通) and Shenzhen-Connect (深股通) "
-        "channels (units: 10k CNY), as the latest realtime figure plus a recent "
-        "daily history. This is a market-level total, NOT per-stock flow (for a "
+        "channels, as the latest realtime figure plus a recent daily history "
+        "(units: 10k CNY from Eastmoney; the tushare fallback reports CNY "
+        "million and documents the 2024-08-30 net-buy→turnover disclosure "
+        "boundary in data.note). This is a market-level total, NOT per-stock flow (for a "
         "given symbol's order-bucket inflow use get_fund_flow). Read-only; China "
         "A-share market only. Example: get_northbound_flow(lookback_days=10)."
     )
@@ -231,35 +259,24 @@ class NorthboundFlowTool(BaseTool):
             )
         except Exception as exc:  # noqa: BLE001 - surface as error envelope
             logger.warning("northbound flow fetch failed: %s", exc)
-            try:
-                fallback_data = tushare_fallbacks.fetch_northbound_flow(
-                    lookback_days=lookback_days
-                )
-            except Exception as fallback_exc:  # noqa: BLE001 - return both provider failures
-                return json.dumps(
-                    {
-                        "ok": False,
-                        "error": f"{exc}; tushare fallback failed: {fallback_exc}",
-                    },
-                    ensure_ascii=False,
-                )
-            return json.dumps(
-                {
-                    "ok": True,
-                    "market": "China A",
-                    "source": "tushare",
-                    "warnings": [
-                        "eastmoney failed "
-                        f"({exc}); used tushare fallback with latest daily data"
-                    ],
-                    "data": fallback_data,
-                },
-                ensure_ascii=False,
+            return self._tushare_fallback_envelope(
+                lookback_days, f"eastmoney failed ({exc})"
             )
 
         realtime = _parse_realtime(realtime_payload)
         history = _parse_history(history_payload, lookback_days)
 
+        if _is_dead_feed(realtime, history):
+            logger.warning(
+                "eastmoney northbound feed is dead (empty/all-zero payload); "
+                "falling through to tushare"
+            )
+            return self._tushare_fallback_envelope(
+                lookback_days,
+                "eastmoney northbound net-flow feed is dead (empty/all-zero "
+                "since the 2024-08-30 HKEX disclosure change)",
+            )
+
         envelope = {
             "ok": True,
             "market": "China A",
@@ -272,3 +289,39 @@ class NorthboundFlowTool(BaseTool):
             },
         }
         return json.dumps(envelope, ensure_ascii=False)
+
+    def _tushare_fallback_envelope(self, lookback_days: int, reason: str) -> str:
+        """Build the tushare-fallback envelope, or an honest error without it.
+
+        Args:
+            lookback_days: Clamped history window forwarded to the fallback.
+            reason: Human-readable cause ("eastmoney failed (...)" / "feed is
+                dead (...)") embedded in the warning or error text.
+
+        Returns:
+            A JSON string: ``ok: true`` with ``source: "tushare"`` and a
+            warning when the fallback serves data, or ``ok: false`` naming
+            both failures when it cannot.
+        """
+        try:
+            fallback_data = tushare_fallbacks.fetch_northbound_flow(
+                lookback_days=lookback_days
+            )
+        except Exception as fallback_exc:  # noqa: BLE001 - both failures surface
+            return json.dumps(
+                {
+                    "ok": False,
+                    "error": f"{reason}; tushare fallback failed: {fallback_exc}",
+                },
+                ensure_ascii=False,
+            )
+        return json.dumps(
+            {
+                "ok": True,
+                "market": "China A",
+                "source": "tushare",
+                "warnings": [f"{reason}; used tushare fallback with latest daily data"],
+                "data": fallback_data,
+            },
+            ensure_ascii=False,
+        )

agent/src/tools/tushare_fallbacks.py

@@ -178,28 +178,50 @@ def fetch_dragon_tiger(trade_date: str, code: str | None) -> dict[str, Any]:
     return data
 
 
+#: HKEX stopped publishing northbound daily net buy on this date; tushare's
+#: ``hgt``/``sgt``/``north_money`` carry turnover from then on (issue #1481).
+_NORTHBOUND_SEMANTICS_BOUNDARY = "2024-08-30"
+
+_NORTHBOUND_NOTE = (
+    "HKEX stopped publishing northbound daily net buy on "
+    f"{_NORTHBOUND_SEMANTICS_BOUNDARY}. From that date onward tushare's "
+    "hgt/sgt/north_money carry northbound TURNOVER (verified equal to the "
+    "HKEX official Stock Connect Daily Statistics Total Turnover to the "
+    "cent); rows before the boundary are NET BUY. Both eras are in million "
+    "CNY and are passed through unchanged."
+)
+
+
 def fetch_northbound_flow(*, lookback_days: int) -> dict[str, Any]:
+    """Fetch northbound Stock-Connect flow via tushare ``moneyflow_hsgt``.
+
+    Values are passed through in tushare's native unit (million CNY). The
+    legacy ×100 rescale to 10k CNY assumed every row was a net buy in
+    million CNY, but the field semantics changed at the 2024-08-30 HKEX
+    disclosure reform: post-boundary rows are turnover (cross-verified
+    against HKEX official daily statistics, see issue #1481), so a single
+    constant factor cannot serve both eras. The envelope therefore keeps
+    the raw values and documents the boundary in ``note`` instead.
+    """
     start_date, end_date = _date_window(lookback_days)
     rows = _records(_pro_api().moneyflow_hsgt(start_date=start_date, end_date=end_date))
     history: list[dict[str, Any]] = []
     for row in rows:
-        shanghai = _to_float(row.get("hgt"))
-        shenzhen = _to_float(row.get("sgt"))
-        total = _to_float(row.get("north_money"))
         history.append(
             {
                 "trade_date": _dashed_date(row.get("trade_date")),
-                "shanghai_connect": shanghai * 100 if shanghai is not None else None,
-                "shenzhen_connect": shenzhen * 100 if shenzhen is not None else None,
-                "total": total * 100 if total is not None else None,
+                "shanghai_connect": _to_float(row.get("hgt")),
+                "shenzhen_connect": _to_float(row.get("sgt")),
+                "total": _to_float(row.get("north_money")),
             }
         )
     history.sort(key=lambda item: item.get("trade_date") or "")
     history = history[-lookback_days:]
     latest = history[-1] if history else {}
     return {
-        "unit": "10k CNY",
+        "unit": "CNY million",
         "lookback_days": lookback_days,
+        "note": _NORTHBOUND_NOTE,
         "realtime": {
             "shanghai_connect": latest.get("shanghai_connect"),
             "shenzhen_connect": latest.get("shenzhen_connect"),

agent/tests/test_northbound_tool.py

@@ -121,13 +121,115 @@ class TestErrorEnvelope:
         assert payload["data"]["history"][0]["trade_date"] == "2024-01-03"
         assert "used tushare fallback" in payload["warnings"][0]
 
-    def test_missing_data_block_yields_empty_history_and_null_realtime(self):
-        with patch.object(nb, "get_json", return_value={"data": None}):
+    def test_missing_data_block_is_dead_feed_and_falls_through(self):
+        """A structurally empty payload is the live post-reform state (#1481):
+        it must fall through to tushare, not pass as an empty success."""
+        fallback = {
+            "unit": "CNY million",
+            "lookback_days": 5,
+            "note": "HKEX stopped publishing northbound net buy on 2024-08-30...",
+            "realtime": {"total": 259865.83},
+            "history": [{"trade_date": "2026-09-17", "total": 259865.83}],
+        }
+        with patch.object(nb, "get_json", return_value={"data": None}), patch.object(
+            nb.tushare_fallbacks,
+            "fetch_northbound_flow",
+            return_value=fallback,
+        ) as fallback_fetch:
             text = nb.NorthboundFlowTool().execute(lookback_days=5)
+
+        fallback_fetch.assert_called_once_with(lookback_days=5)
         payload = json.loads(text)
         assert payload["ok"] is True
-        assert payload["data"]["history"] == []
-        assert payload["data"]["realtime"]["total"] is None
+        assert payload["source"] == "tushare"
+        assert "used tushare fallback" in payload["warnings"][0]
+        assert payload["data"]["unit"] == "CNY million"
+
+
+class TestDeadFeedDetection:
+    """Eastmoney's northbound net feed died at the 2024-08-30 disclosure reform.
+
+    The HTTP calls still succeed with all-zero nets / a frozen cumulative (and
+    the payload shape drifted so the parsers find nothing), which used to pass
+    as an empty ok:true envelope and short-circuit the tushare fallback.
+    """
+
+    def test_all_zero_payload_falls_through_to_tushare(self):
+        dead_realtime = {
+            "data": {
+                "hk2sh": {"netBuyAmt": 0.0},
+                "hk2sz": {"netBuyAmt": 0.0},
+            }
+        }
+        dead_history = {
+            "data": {
+                "klines": [
+                    "2026-09-16,0.0,0.0",
+                    "2026-09-17,0.0,0.0",
+                ]
+            }
+        }
+
+        def fake(url: str, *, params: dict):
+            return dead_history if "kamt.kline" in url else dead_realtime
+
+        fallback = {
+            "unit": "CNY million",
+            "lookback_days": 5,
+            "note": "...2024-08-30...",
+            "realtime": {"total": 4.5},
+            "history": [{"trade_date": "2026-09-17", "total": 4.5}],
+        }
+        with patch.object(nb, "get_json", side_effect=fake), patch.object(
+            nb.tushare_fallbacks,
+            "fetch_northbound_flow",
+            return_value=fallback,
+        ):
+            text = nb.NorthboundFlowTool().execute(lookback_days=5)
+
+        payload = json.loads(text)
+        assert payload["ok"] is True
+        assert payload["source"] == "tushare"
+        assert "2024-08-30" in payload["warnings"][0]
+
+    def test_dead_feed_without_tushare_returns_honest_error(self):
+        with patch.object(nb, "get_json", return_value={"data": None}), patch.object(
+            nb.tushare_fallbacks,
+            "fetch_northbound_flow",
+            side_effect=RuntimeError("TUSHARE_TOKEN not set"),
+        ):
+            text = nb.NorthboundFlowTool().execute(lookback_days=5)
+
+        payload = json.loads(text)
+        assert payload["ok"] is False
+        assert "2024-08-30" in payload["error"]
+        assert "tushare fallback failed" in payload["error"]
+
+    def test_healthy_payload_still_serves_eastmoney(self):
+        """Guard: detection must not swallow a live non-zero feed."""
+        with patch.object(nb, "get_json", side_effect=_fake_get_json):
+            text = nb.NorthboundFlowTool().execute(lookback_days=10)
+        payload = json.loads(text)
+        assert payload["ok"] is True
+        assert payload["source"] == "eastmoney"
+
+    def test_realtime_alive_history_dead_is_not_dead_feed(self):
+        """Both sides must be dead: a trimmed all-None history window with a
+        live realtime figure still serves the Eastmoney envelope."""
+        live_realtime = {
+            "data": {"hk2sh": {"netBuyAmt": 12.5}, "hk2sz": {"netBuyAmt": 0.0}}
+        }
+
+        def fake(url: str, *, params: dict):
+            if "kamt.kline" in url:
+                return {"data": {"klines": ["2026-09-17,0.0,0.0"]}}
+            return live_realtime
+
+        with patch.object(nb, "get_json", side_effect=fake):
+            text = nb.NorthboundFlowTool().execute(lookback_days=5)
+        payload = json.loads(text)
+        assert payload["source"] == "eastmoney"
+        assert payload["data"]["realtime"]["shanghai_connect"] == 12.5
 
 
 class TestToolMetadata:

agent/tests/test_tushare_fallbacks.py

@@ -73,7 +73,10 @@ def test_dragon_tiger_maps_top_list_and_top_inst() -> None:
     assert data["seats"][0]["net"] == 2.0e8
 
 
-def test_northbound_converts_tushare_million_yuan_to_10k_cny() -> None:
+def test_northbound_passes_through_million_yuan_with_semantics_note() -> None:
+    """No ×100 rescale: post-2024-08-30 rows are turnover in million CNY
+    (HKEX-verified, issue #1481); a constant factor cannot serve both eras,
+    so values pass through and the note documents the boundary."""
     pro = SimpleNamespace(
         moneyflow_hsgt=lambda **_: [
             {"trade_date": "20240102", "hgt": 12.0, "sgt": -2.0, "north_money": 10.0},
@@ -85,10 +88,12 @@ def test_northbound_converts_tushare_million_yuan_to_10k_cny() -> None:
     ):
         data = tf.fetch_northbound_flow(lookback_days=2)
 
-    assert data["unit"] == "10k CNY"
-    assert data["history"][0]["shanghai_connect"] == 1200.0
-    assert data["history"][0]["total"] == 1000.0
-    assert data["realtime"]["total"] == 450.0
+    assert data["unit"] == "CNY million"
+    assert data["history"][0]["shanghai_connect"] == 12.0
+    assert data["history"][0]["total"] == 10.0
+    assert data["realtime"]["total"] == 4.5
+    assert "2024-08-30" in data["note"]
+    assert "turnover" in data["note"].lower()
 
 
 def test_margin_trading_maps_and_sorts_most_recent_first() -> None: