feat(runtime): harden candle readiness and recovery

backend_api_python/app/commands/strategy_dispatcher_worker.py

@@ -10,6 +10,7 @@ def main() -> None:
     os.environ["QD_PROCESS_ROLE"] = "strategy-dispatcher"
 
     from app import create_app
+    from app.events.bar_recovery import StrategyBarRecoveryPublisher
     from app.events.dispatcher import StrategyDueDispatcher
     from app.events.kafka import KafkaEventConsumer
     from app.events.protocol import MarketBarClosedV1
@@ -26,6 +27,7 @@ def main() -> None:
     heartbeat_repository = StrategyCommandRepository()
     worker_id = build_worker_id()
     dispatcher = StrategyDueDispatcher()
+    recovery = StrategyBarRecoveryPublisher()
     consumer = KafkaEventConsumer(
         topics=[topic_for_event(MarketBarClosedV1.event_type)],
         group_id=os.getenv(
@@ -34,23 +36,46 @@ def main() -> None:
         ),
     )
     last_heartbeat = 0.0
+    last_recovery = 0.0
+    try:
+        recovery_interval = max(
+            1.0,
+            float(os.getenv("STRATEGY_BAR_RECOVERY_INTERVAL_SEC", "5")),
+        )
+    except (TypeError, ValueError):
+        recovery_interval = 5.0
+
+    def dispatch(event):
+        recovery.observe(event)
+        return dispatcher.dispatch(event)
 
     with app.app_context():
         try:
             logger.info("Strategy dispatcher worker started: %s", worker_id)
             while not shutdown.event.is_set():
-                consumer.poll_once(dispatcher.dispatch, timeout=1.0)
+                consumer.poll_once(dispatch, timeout=1.0)
                 now = time.monotonic()
+                if now - last_recovery >= recovery_interval:
+                    try:
+                        recovery.publish_due()
+                    except Exception:
+                        logger.exception("Strategy bar recovery cycle failed")
+                    last_recovery = now
                 if now - last_heartbeat >= 10.0:
                     heartbeat_repository.record_worker_heartbeat(
                         worker_id=worker_id,
                         role="strategy-dispatcher",
-                        metadata={**consumer.snapshot(), **dispatcher.snapshot()},
+                        metadata={
+                            **consumer.snapshot(),
+                            **dispatcher.snapshot(),
+                            **recovery.snapshot(),
+                        },
                     )
                     last_heartbeat = now
         finally:
             consumer.close()
             dispatcher.close()
+            recovery.close()
             heartbeat_repository.mark_worker_stopped(worker_id)
             logger.info("Strategy dispatcher worker stopped: %s", worker_id)
 

backend_api_python/app/data_sources/asia_stock_kline.py

@@ -145,7 +145,7 @@ def _get_twelve_data_api_key() -> str:
     "15m": "15min",
     "30m": "30min",
     "1H": "1h",
-    "4H": "4h",
+    "4H": "1h",
     "1D": "1day",
     "1W": "1week",
 }
@@ -231,6 +231,8 @@ def fetch_twelvedata_klines(
 
     symbol, exchange = _td_symbol_and_exchange(tencent_code, is_hk)
     merge_factor = _MERGE_FACTOR_MAP.get(timeframe, 1)
+    if is_hk and timeframe in {"1H", "4H"}:
+        interval, merge_factor = "30min", 2 if timeframe == "1H" else 8
     params: Dict[str, Any] = {
         "symbol": symbol,
         "exchange": exchange,
@@ -242,6 +244,8 @@ def fetch_twelvedata_klines(
     }
     if before_time:
         end_dt = datetime.fromtimestamp(int(before_time))
+        if is_hk:
+            end_dt = pd.Timestamp(int(before_time), unit="s", tz="UTC").tz_convert("Asia/Hong_Kong")
         params["end_date"] = end_dt.strftime("%Y-%m-%d %H:%M:%S")
 
     url = "https://api.twelvedata.com/time_series"
@@ -290,7 +294,10 @@ def fetch_twelvedata_klines(
             dt_str = v.get("datetime", "")
             for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d"):
                 try:
-                    ts = int(datetime.strptime(dt_str, fmt).timestamp())
+                    parsed_time = datetime.strptime(dt_str, fmt)
+                    if is_hk:
+                        parsed_time = pd.Timestamp(parsed_time).tz_localize("Asia/Hong_Kong")
+                    ts = int(parsed_time.timestamp())
                     break
                 except ValueError:
                     continue
@@ -316,7 +323,7 @@ def fetch_twelvedata_klines(
 
     out.sort(key=lambda x: x["time"])
     if merge_factor > 1 and out:
-        out = _merge_every_n_sorted_bars(out, merge_factor)
+        out = _merge_session_bars(out, timeframe, is_hk, source_seconds=1800 if is_hk and timeframe in {"1H", "4H"} else None)
     logger.debug("TwelveData returned %d bars for %s/%s tf=%s", len(out), symbol, exchange, timeframe)
     return out
 
@@ -483,7 +490,7 @@ def fetch_yfinance_klines(
 
     bars = _bars_from_yfinance_df(df)
     if merge_factor > 1 and bars:
-        bars = _merge_every_n_sorted_bars(bars, merge_factor)
+        bars = _merge_session_bars(bars, timeframe, is_hk)
     logger.debug("yfinance returned %d bars for %s tf=%s", len(bars), yf_sym, timeframe)
     return bars
 
@@ -513,7 +520,7 @@ def _min_bar_window(
     return start.strftime(fmt), end.strftime(fmt)
 
 
-def _bars_from_ak_min_df(df: Any) -> List[Dict[str, Any]]:
+def _bars_from_ak_min_df(df: Any, *, is_hk: bool = False, period_seconds: int = 0) -> List[Dict[str, Any]]:
     if df is None or getattr(df, "empty", True):
         return []
     cols = [str(x) for x in df.columns]
@@ -536,6 +543,9 @@ def _pick(name_zh: str, idx: int) -> str:
     for _, row in df.iterrows():
         try:
             t = pd.Timestamp(row[time_c])
+            if is_hk:
+                t = t.tz_localize("Asia/Hong_Kong") if t.tzinfo is None else t
+                t -= pd.Timedelta(seconds=period_seconds)
             ts = int(t.timestamp())
             o, c, h, low, v = float(row[c_open]), float(row[c_close]), float(row[c_high]), float(row[c_low]), float(row[c_vol])
             out.append({
@@ -552,6 +562,15 @@ def _pick(name_zh: str, idx: int) -> str:
     return out
 
 
+def _merge_session_bars(bars, timeframe, is_hk, *, source_seconds=None):
+    if not is_hk:
+        return _merge_every_n_sorted_bars(bars, _MERGE_FACTOR_MAP.get(timeframe, 1))
+    from app.services.market_schedule import resample_equity_klines
+    source_seconds = source_seconds or (3600 if timeframe == "4H" else 60)
+    target_seconds = {"1H": 3600, "4H": 14400, "3m": 180}[timeframe]
+    return resample_equity_klines(bars, "HKStock", source_seconds, target_seconds)
+
+
 def _merge_every_n_sorted_bars(bars: List[Dict[str, Any]], n: int) -> List[Dict[str, Any]]:
     if n <= 1 or len(bars) < n:
         return bars
@@ -587,6 +606,8 @@ def fetch_akshare_minute_klines(
         return []
 
     p = _minute_period_str(timeframe)
+    if is_hk and timeframe in {"1H", "4H"}:
+        p = "30"
     if p is None:
         return []
     try:
@@ -597,6 +618,11 @@ def fetch_akshare_minute_klines(
 
     sym = ak_hk_code_from_tencent(tencent_code) if is_hk else ak_a_code_from_tencent(tencent_code)
     sd, ed = _min_bar_window(timeframe, limit, before_time, after_time)
+    if is_hk:
+        if before_time is not None:
+            ed = pd.Timestamp(before_time, unit="s", tz="UTC").tz_convert("Asia/Hong_Kong").strftime("%Y-%m-%d %H:%M:%S")
+        if after_time is not None:
+            sd = pd.Timestamp(after_time, unit="s", tz="UTC").tz_convert("Asia/Hong_Kong").strftime("%Y-%m-%d %H:%M:%S")
     adj = "" if p == "1" else "qfq"
 
     df: Any = None
@@ -625,10 +651,12 @@ def fetch_akshare_minute_klines(
             logger.warning("AkShare minute K-line failed %s tf=%s sym=%s: %s", tencent_code, timeframe, sym, e)
             return []
 
-    bars = _bars_from_ak_min_df(df)
+    bars = _bars_from_ak_min_df(df, is_hk=is_hk, period_seconds=int(p) * 60)
     merge_factor = _MERGE_FACTOR_MAP.get(timeframe, 1)
+    if is_hk and timeframe in {"1H", "4H"}:
+        merge_factor = 2 if timeframe == "1H" else 8
     if merge_factor > 1 and bars:
-        bars = _merge_every_n_sorted_bars(bars, merge_factor)
+        bars = _merge_session_bars(bars, timeframe, is_hk, source_seconds=int(p) * 60)
     return bars
 
 

backend_api_python/app/data_sources/candle_calendar.py

@@ -0,0 +1,24 @@
+"""Native crypto candle boundaries, independent of strategy and transport code."""
+
+from app.data_sources.base import timeframe_seconds
+
+
+def candle_origin(timeframe: str, venue: str = "", market_type: str = "spot") -> int:
+    seconds = timeframe_seconds(timeframe)
+    venue = str(venue or "").lower()
+    offset = -8 * 3600 if venue in {"okx", "bitget", "htx", "huobi"} else 0
+    if seconds == 604800:
+        if venue == "gate" and market_type == "swap":
+            return 0
+        offset += 4 * 86400
+    return offset % seconds
+
+
+def candle_open(timestamp: float, timeframe: str, venue: str = "", market_type: str = "spot") -> int:
+    seconds = timeframe_seconds(timeframe)
+    origin = candle_origin(timeframe, venue, market_type)
+    return int((timestamp - origin) // seconds) * seconds + origin
+
+
+def completed_candle_open(timestamp: float, timeframe: str, venue: str = "", market_type: str = "spot") -> int:
+    return candle_open(timestamp, timeframe, venue, market_type) - timeframe_seconds(timeframe)

backend_api_python/app/data_sources/crypto.py

@@ -501,11 +501,14 @@ def get_kline(
         limit: int,
         before_time: Optional[int] = None,
         after_time: Optional[int] = None,
+        *,
+        closed_only: bool = False,
     ) -> List[Dict[str, Any]]:
         """获取加密货币K线数据"""
         self._clear_last_failure()
         klines = []
         symbol_pair = ""
+        candle_options = {"closed_only": True} if closed_only else {}
 
         # A public source is cached and reused by chart/backtest/signal
         # runtimes. Once its configured venue has failed and another venue has
@@ -530,7 +533,7 @@ def get_kline(
             preferred_rows = type(self).for_exchange(
                 preferred_exchange,
                 fallback_market_type,
-            ).get_kline(symbol, timeframe, limit, before_time, after_time)
+            ).get_kline(symbol, timeframe, limit, before_time, after_time, **candle_options)
             if preferred_rows:
                 return preferred_rows
             # The promoted provider has also become unavailable. Clear it and
@@ -591,6 +594,7 @@ def get_kline(
             ohlcv = self._fetch_ohlcv(
                 symbol_pair, fetch_exchange_timeframe, fetch_limit,
                 before_time, fetch_qd_timeframe, after_time,
+                **candle_options,
             )
 
             if not ohlcv:
@@ -728,6 +732,7 @@ def get_kline(
                         limit,
                         before_time,
                         after_time,
+                        **candle_options,
                     )
                     if fallback_rows:
                         self._preferred_public_exchange_id = exchange_id
@@ -825,6 +830,8 @@ def _fetch_ohlcv(
         before_time: Optional[int],
         timeframe: str,
         after_time: Optional[int] = None,
+        *,
+        closed_only: bool = False,
     ) -> List:
         """获取OHLCV数据（支持分页获取完整数据）"""
         try:
@@ -949,6 +956,7 @@ def _fetch_ohlcv(
                                     since=current_since,
                                     limit=request_limit,
                                     until=end_ms,
+                                    **({"closed_only": True} if closed_only else {}),
                                 )
                             else:
                                 batch = self.exchange.fetch_ohlcv(
@@ -1000,7 +1008,8 @@ def _fetch_ohlcv(
                 ohlcv = sorted(by_ts.values(), key=lambda r: r[0])
                 if not ohlcv:
                     return self._fetch_ohlcv_fallback(
-                        symbol_pair, exchange_timeframe, limit, before_time, timeframe, after_time
+                        symbol_pair, exchange_timeframe, limit, before_time, timeframe, after_time,
+                        **({"closed_only": True} if closed_only else {}),
                     )
                 # Never attach an old partial page across an unreported gap. Only
                 # validate the returned chart tail, not its extra warmup buffer.
@@ -1046,7 +1055,10 @@ def _fetch_ohlcv(
                 # rejecting the request outright or silently truncating, which
                 # downstream code then interprets as "empty data".
                 safe_limit = min(int(limit), self._SINGLE_FETCH_HARD_CAP)
-                ohlcv = self.exchange.fetch_ohlcv(symbol_pair, exchange_timeframe, limit=safe_limit)
+                ohlcv = self.exchange.fetch_ohlcv(
+                    symbol_pair, exchange_timeframe, limit=safe_limit,
+                    **({"closed_only": True} if closed_only and isinstance(self.exchange, NativeCryptoPublicClient) else {}),
+                )
 
             return ohlcv
 
@@ -1077,7 +1089,8 @@ def _fetch_ohlcv(
             logger.warning(f"Exchange fetch_ohlcv failed: {str(e)}; trying fallback")
             self._set_last_failure(e, symbol=symbol_pair, timeframe=timeframe)
             return self._fetch_ohlcv_fallback(
-                symbol_pair, exchange_timeframe, limit, before_time, timeframe, after_time
+                symbol_pair, exchange_timeframe, limit, before_time, timeframe, after_time,
+                **({"closed_only": True} if closed_only else {}),
             )
 
     def _fetch_ohlcv_fallback(
@@ -1088,6 +1101,8 @@ def _fetch_ohlcv_fallback(
         before_time: Optional[int],
         timeframe: str,
         after_time: Optional[int] = None,
+        *,
+        closed_only: bool = False,
     ) -> List:
         """备用获取方法"""
         # An explicit historical window must never be replaced with a recent
@@ -1121,7 +1136,10 @@ def _fetch_ohlcv_fallback(
             # exactly when it mattered. Cap it so we at least return one valid
             # page of data, which downstream callers can then handle gracefully.
             safe_limit = min(int(limit), self._SINGLE_FETCH_HARD_CAP)
-            ohlcv = self.exchange.fetch_ohlcv(symbol_pair, exchange_timeframe, since=since, limit=safe_limit)
+            ohlcv = self.exchange.fetch_ohlcv(
+                symbol_pair, exchange_timeframe, since=since, limit=safe_limit,
+                **({"closed_only": True} if closed_only and isinstance(self.exchange, NativeCryptoPublicClient) else {}),
+            )
             if ohlcv:
                 return ohlcv
         except Exception as e:
@@ -1137,6 +1155,7 @@ def _fetch_ohlcv_fallback(
                 symbol_pair,
                 exchange_timeframe,
                 limit=min(int(limit), self._SINGLE_FETCH_HARD_CAP),
+                **({"closed_only": True} if closed_only and isinstance(self.exchange, NativeCryptoPublicClient) else {}),
             )
             if recent:
                 logger.warning(

backend_api_python/app/data_sources/factory.py

@@ -268,10 +268,16 @@ def get_kline_with_diagnostics(
         try:
             assert_fd_available(f"market-data kline {m}:{symbol}")
             source = cls._resolve_source(m, exchange_id=exchange_id, market_type=market_type)
-            klines = source.get_kline(symbol, timeframe, limit, before_time, after_time)
+            from app.data_sources.crypto import CryptoDataSource
+            candle_options = {"closed_only": True} if closed_only and isinstance(source, CryptoDataSource) else {}
+            klines = source.get_kline(symbol, timeframe, limit, before_time, after_time, **candle_options)
 
             if closed_only:
-                klines = filter_closed_klines(klines, timeframe)
+                if m in {"USStock", "HKStock"}:
+                    from app.services.market_schedule import filter_closed_equity_klines
+                    klines = filter_closed_equity_klines(klines, timeframe, m)
+                else:
+                    klines = filter_closed_klines(klines, timeframe)
 
             klines.sort(key=lambda x: x['time'])
             failure = None