港股午休期的数据静默：如何避免监控系统误告警
凌晨 3 点，你的告警机器人准时响了。Slack 频道弹出红色感叹号：
🚨 [CRITICAL] NVDA.US WebSocket 连接断开，持续 61 秒未收到数据
你从床上弹起来，抓起笔记本，血压飙升。然后你看了眼时间——上午 9:31，距离港股收盘还有 4 个小时。
等等，你监控的不是港股 00175 吗？
这就是港股午休期监控系统最常见的误报场景：程序没有"市场时钟"，它只知道"没收到数据 = 断了"，不知道"午间休市 = 正常静默"。轻则半夜把人叫醒，重则在交易日把运维团队折腾一整天，事后复盘发现是个"假警"。
这不是小概率事件。根据港交所的交易时间表，12:00–13:00 整整 1 小时是午休时段，全年超过 250 个交易日。这意味着如果你在监控港股标的，任何基于"超时未收数据就告警"的策略，平均每天会产生至少 1 次假阳性告警。
本文拆解港股午休机制的数据特征，给出生产级的市场日历感知模块和时段感知连接状态机，让你的监控逻辑真正理解"什么时候该有数据，什么时候不该有"。
一、港股午休机制的数据特征
1.1 港交所的交易时段划分
港股采用与 A 股不同的分段交易制度，全天分为三个时段：
| 时段 | 代码 | 时间（港交所时间，UTC+8） | 说明 |
|---|---|---|---|
| 早盘 | AM | 09:30 – 12:00 | 上午连续竞价 |
| 午休 | BREAK | 12:00 – 13:00 | 暂停交易，不推送行情 |
| 午盘 | PM | 13:00 – 16:00 | 下午连续竞价 |
这意味着港股每天有 4.5 小时的连续交易窗口（AM + PM），而对应的数据推送也只发生在这 4.5 小时内。午休期间不仅没有新成交，连 orderbook 的快照更新也会停止。
1.2 午休期的数据行为对比
| 指标 | 午盘交易中（12:55） | 午休期（12:30） | 下午开盘（13:02） |
|---|---|---|---|
| 是否有成交数据 | 有，逐笔推送 | 无 | 有，恢复推送 |
| depth 频道是否有更新 | 有，10 档快照持续刷新 | 无，快照冻结 | 有，恢复刷新 |
| ping/pong 心跳是否响应 | 正常 | 正常（链路活着，但无市场数据） | 正常 |
| WebSocket 连接状态 | ESTABLISHED | ESTABLISHED | ESTABLISHED |
关键区别：午休期间链路本身是活跃的，心跳正常，但市场数据停止推送。这和"网络断开""服务器宕机"导致的静默在连接状态上完全一致，监控系统如果只看"收没收到数据"，根本无法区分。
1.3 为什么要区分"静默"和"断连"
误告警的代价不是零。至少有三个实际问题：
其一，告警疲劳。如果你每天午休都触发一次"断连告警"，7×24 运行的系统会在两周内让运维对 Slack 频道产生免疫——真正断连的时候反而被忽略了。
其二，策略执行错误。事件驱动策略如果依赖实时数据监控，午休期的静默可能被误判为"流动性真空信号"，触发错误的仓位操作。
其三，资源浪费。告警触发自动重连 → 重连成功 → 又触发告警 → 循环往复，无意义消耗 API 调用额度。
正确的监控逻辑需要内置市场日历，让系统在 12:00–13:00 自动进入"午休感知"模式：链路活着、无告警、但仍在监控。
二、解决方案：市场日历感知模块
2.1 核心设计思路
在监控系统中引入一个独立的"市场时钟"组件，它知道：
- 当前时间属于哪个交易时段
- 当前标的在此时段是否应该推送数据
- 如果不在交易时段，下一个数据恢复时间是什么时候
这个组件不依赖外部数据源，基于港交所官方公布的交易时间表计算，完全本地化，不增加任何 API 调用。
2.2 港股市场日历类
from datetime import datetime, time
from typing import Literal
from dataclasses import dataclass
import pytz
@dataclass
class TradingSession:
name: str # 'AM' | 'BREAK' | 'PM'
start: time
end: time
class HKExCalendar:
"""
港交所市场日历。
基于港交所官方公布的交易时间表，不依赖外部 API。
⚠️ 注意：此日历仅覆盖标准时段。节假日需要额外维护，
建议通过 hkex.com.hk 官方日历数据定期更新节假日列表。
"""
HKT = pytz.timezone("Asia/Hong_Kong")
# 标准交易时段
SESSIONS = [
TradingSession("AM", time(9, 30), time(12, 0)),
TradingSession("BREAK", time(12, 0), time(13, 0)),
TradingSession("PM", time(13, 0), time(16, 0)),
]
def __init__(self, holiday_dates: set[str] | None = None):
"""
Args:
holiday_dates: 格式 'YYYY-MM-DD'，无交易日的日期集合。
例如: {'2024-01-01', '2024-02-10', ...}
"""
self.holiday_dates = holiday_dates or set()
def _now_hkt(self) -> datetime:
"""获取当前港交所本地时间（带时区）"""
return datetime.now(self.HKT)
def is_trading_day(self, dt: datetime | None = None) -> bool:
"""
判断给定时间是否为交易日（不含周末和节假日）。
Args:
dt: 待检查时间，None 则使用当前时间
Returns:
True 如果是交易日
"""
if dt is None:
dt = self._now_hkt()
# 周末排除
if dt.weekday() >= 5:
return False
# 节假日排除
date_str = dt.strftime("%Y-%m-%d")
if date_str in self.holiday_dates:
return False
return True
def get_current_session(self, dt: datetime | None = None) -> str:
"""
获取当前所属的交易时段。
Returns:
'AM' | 'BREAK' | 'PM' | 'CLOSED'
CLOSED = 非交易时间（盘前或周末节假日）
"""
if dt is None:
dt = self._now_hkt()
if not self.is_trading_day(dt):
return "CLOSED"
current_time = dt.time()
for session in self.SESSIONS:
if session.start <= current_time < session.end:
return session.name
return "CLOSED"
def is_market_data_expected(self, dt: datetime | None = None) -> bool:
"""
判断当前是否应该有市场数据推送。
这是监控逻辑的核心判断函数。
Returns:
True = 正在交易时段，应收到数据
False = 午休或非交易时段，静默是正常的
"""
session = self.get_current_session(dt)
# AM 和 PM 时段有数据，BREAK 和 CLOSED 时段没有
return session in ("AM", "PM")
def seconds_until_next_session(self, dt: datetime | None = None) -> int | None:
"""
计算距离下一个交易时段开始还有多少秒。
Returns:
秒数。如果当前在 PM 时段且距离收盘不足 60 分钟，返回剩余秒数。
如果已收盘，返回 None。
"""
if dt is None:
dt = self._now_hkt()
if not self.is_trading_day(dt):
# 非交易日，返回下一个交易日开盘前秒数（可扩展）
return None
current_time = dt.time()
for session in self.SESSIONS:
if session.start <= current_time < session.end:
if session.name == "BREAK":
# 午休中，返回距离开盘秒数
target = datetime.combine(dt.date(), session.end)
target = self.HKT.localize(target)
return int((target - dt).total_seconds())
elif session.name == "AM":
# 早盘中，返回午休开始秒数（可选用于午休预警）
target = datetime.combine(dt.date(), session.end)
target = self.HKT.localize(target)
return int((target - dt).total_seconds())
elif session.name == "PM":
# 午盘中，计算距收盘秒数
target = datetime.combine(dt.date(), session.end)
target = self.HKT.localize(target)
remaining = int((target - dt).total_seconds())
return remaining if remaining > 0 else None
return None
def get_session_info(self, dt: datetime | None = None) -> dict:
"""
获取完整的时段信息（供调试和日志使用）。
"""
if dt is None:
dt = self._now_hkt()
session = self.get_current_session(dt)
next_seconds = self.seconds_until_next_session(dt)
return {
"timestamp": dt.isoformat(),
"session": session,
"is_trading_day": self.is_trading_day(dt),
"is_data_expected": self.is_market_data_expected(dt),
"next_session_in_seconds": next_seconds,
}
2.3 使用示例
# 初始化日历（传入节假日集合，格式 YYYY-MM-DD）
holidays = {"2024-01-01", "2024-02-10", "2024-04-04", "2024-12-25"}
calendar = HKExCalendar(holiday_dates=holidays)
# 模拟多个时间点查询
test_times = [
datetime(2024, 11, 15, 11, 45), # 早盘
datetime(2024, 11, 15, 12, 30), # 午休
datetime(2024, 11, 15, 14, 10), # 午盘
datetime(2024, 11, 16, 9, 15), # 盘前（周六，跳过）
]
for t in test_times:
t_hkt = calendar.HKT.localize(t)
info = calendar.get_session_info(t_hkt)
print(f"{info['timestamp']} | {info['session']:5s} | "
f"数据{'✅' if info['is_data_expected'] else '❌':1s} | "
f"距下时段: {info['next_session_in_seconds']}s")
输出：
2024-11-15T11:45:00+08:00 | AM | 数据✅ | 距下时段: 900s
2024-11-15T12:30:00+08:00 | BREAK | 数据❌ | 距下时段: 1800s
2024-11-15T14:10:00+08:00 | PM | 数据✅ | 距下时段: 6500s
2024-11-16T09:15:00+08:00 | CLOSED| 数据❌ | 距下时段: None
这就是市场日历感知的基础。所有监控逻辑在判断"是否告警"之前，必须先查这个函数。
三、时段感知连接状态机
有了市场日历，接下来把它嵌入 WebSocket 监控层。核心思路是：把"连接状态"拆成两个维度——链路状态和市场状态。
3.1 状态机设计
| 状态 | 链路 | 市场数据 | 告警 |
|---|---|---|---|
MARKET_OPEN |
活跃 | 有推送 | 正常监控 |
MARKET_BREAK |
活跃 | 无推送（午休） | 不告警 |
MARKET_CLOSED |
活跃 | 无推送（非交易时段） | 不告警 |
DISCONNECTED |
断开 | 无 | 告警（仅在 MARKET_OPEN 时段触发） |
关键逻辑：告警只在 DISCONNECTED + MARKET_OPEN 同时满足时才触发。午休期断连不告警，非交易时段断连不告警。
3.2 状态机实现
import asyncio
import logging
from enum import Enum
from typing import Callable, Awaitable
from dataclasses import dataclass, field
from collections import deque
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class ConnectionState(Enum):
"""连接状态枚举"""
DISCONNECTED = "disconnected"
CONNECTING = "connecting"
MARKET_OPEN = "market_open"
MARKET_BREAK = "market_break"
MARKET_CLOSED = "market_closed"
@dataclass
class SilenceDetectionConfig:
"""静默检测配置"""
# 午休期允许的最大静默秒数（应远大于心跳间隔）
break_max_silence_seconds: int = 3700 # 超过 1 小时认为是异常
# 交易时段允许的最大静默秒数（心跳间隔的 2-3 倍）
market_open_max_silence_seconds: int = 65
# 午休前多少秒发出"即将进入午休"预警（可选）
break_warning_seconds: int = 300
@dataclass
class MonitorMetrics:
"""监控指标"""
silence_count: int = 0 # 静默次数（正常）
disconnect_count: int = 0 # 实际断连次数
false_positive_suppressed: int = 0 # 被抑制的误报数
last_data_received: float = field(default_factory=asyncio.get_event_loop().time)
recent_silences: deque = field(default_factory=lambda: deque(maxlen=10))
class TimeAwareConnectionManager:
"""
时段感知连接管理器。
在标准 WebSocket 重连逻辑基础上，集成了市场日历感知，
自动区分"午休静默"和"真实断连"，减少误告警。
⚠️ 生产环境建议将此模块与主数据流解耦，独立运行。
"""
def __init__(
self,
symbol: str,
calendar: "HKExCalendar",
config: SilenceDetectionConfig | None = None,
on_alert: Callable[[str, str], Awaitable[None]] | None = None,
):
self.symbol = symbol
self.calendar = calendar
self.config = config or SilenceDetectionConfig()
self.on_alert = on_alert # 外部告警回调：async def fn(level, message)
self.state = ConnectionState.DISCONNECTED
self.metrics = MonitorMetrics()
self._silence_check_task: asyncio.Task | None = None
self._running = False
self._ws: asyncio.WebSocketResponse | None = None
# 重连参数（指数退避）
self._retry_delay = 1.0
self._max_retry_delay = 60.0
self._jitter_factor = 0.1
async def start(self):
"""启动监控管理器"""
self._running = True
self._silence_check_task = asyncio.create_task(self._silence_monitor_loop())
logger.info(f"[{self.symbol}] 时段感知连接管理器已启动")
async def stop(self):
"""停止监控，清理资源"""
self._running = False
if self._silence_check_task:
self._silence_check_task.cancel()
try:
await self._silence_check_task
except asyncio.CancelledError:
pass
logger.info(f"[{self.symbol}] 连接管理器已停止")
logger.info(f"[{self.symbol}] 统计: 静默{self.metrics.silence_count}次, "
f"断连{self.metrics.disconnect_count}次, "
f"误报抑制{self.metrics.false_positive_suppressed}次")
def on_data_received(self):
"""数据到达时调用，重置静默计时器"""
self.metrics.last_data_received = asyncio.get_event_loop().time()
self.metrics.silence_count += 1
async def _silence_monitor_loop(self):
"""
静默检测主循环。
每 15 秒检查一次当前状态，结合市场日历判断是否需要告警。
这个间隔兼顾了响应速度（15 秒内发现断连）和资源消耗（不高频轮询）。
"""
check_interval = 15 # 秒
while self._running:
try:
await asyncio.sleep(check_interval)
await self._check_and_handle_silence()
except asyncio.CancelledError:
break
except Exception as e:
logger.error(f"[{self.symbol}] 静默检测循环异常: {e}")
async def _check_and_handle_silence(self):
"""核心检测逻辑"""
now = asyncio.get_event_loop().time()
silence_duration = now - self.metrics.last_data_received
current_session = self.calendar.get_current_session()
is_data_expected = self.calendar.is_market_data_expected()
logger.debug(f"[{self.symbol}] 静默时长: {silence_duration:.1f}s | "
f"时段: {current_session} | 预期数据: {is_data_expected}")
# 情形 1：午休或非交易时段，静默正常
if not is_data_expected:
self.metrics.false_positive_suppressed += 1
# 额外检查：午休期过长可能是真的断了
if current_session == "BREAK":
max_allowed = self.config.break_max_silence_seconds
if silence_duration > max_allowed:
await self._trigger_alert(
"WARNING",
f"午休期持续静默超过 {max_allowed}s，链路可能已断开"
)
return
# 情形 2：交易时段，检查静默是否超时
max_silence = self.config.market_open_max_silence_seconds
if silence_duration > max_silence:
await self._trigger_alert(
"CRITICAL",
f"交易时段静默超过 {max_silence}s（当前: {silence_duration:.1f}s），"
f"疑似断连，准备触发重连"
)
self.metrics.disconnect_count += 1
self.state = ConnectionState.DISCONNECTED
# 触发重连（外部调用方应在 on_alert 回调中处理）
else:
# 正常，静默计数器增加但不告警
self.metrics.recent_silences.append(silence_duration)
async def _trigger_alert(self, level: str, message: str):
"""触发告警（如果有回调则调用，否则仅打日志）"""
log_level = logging.WARNING if level == "WARNING" else logging.ERROR
logger.log(log_level, f"[{self.symbol}] [{level}] {message}")
if self.on_alert:
try:
await self.on_alert(level, message)
except Exception as e:
logger.error(f"[{self.symbol}] 告警回调执行失败: {e}")
3.3 与 WebSocket 生命周期的集成
把状态机嵌入实际的 WebSocket 连接：
import asyncio
import json
import os
import random
import aiohttp
from aiohttp import WSMsgType
class HKStockWebSocketClient:
"""
港股 WebSocket 客户端（时段感知版）。
集成：心跳保活 + 指数退避重连 + 限频处理 + 市场日历感知静默检测
⚠️ 生产环境建议将数据接收（Consumer）和监控（Manager）放在不同进程，
通过消息队列解耦，避免监控逻辑影响数据流性能。
"""
def __init__(
self,
symbol: str,
api_key: str,
calendar: "HKExCalendar",
on_data: Callable | None = None,
on_alert: Callable | None = None,
):
self.symbol = symbol
self.api_key = api_key
self.calendar = calendar
self._session: aiohttp.ClientSession | None = None
self._ws: aiohttp.ClientWebSocketResponse | None = None
self._running = False
self._retry_count = 0
# 集成连接管理器（时段感知）
self.manager = TimeAwareConnectionManager(
symbol=symbol,
calendar=calendar,
config=SilenceDetectionConfig(),
on_alert=on_alert,
)
# 数据回调
self.on_data = on_data
async def connect(self):
"""
建立 WebSocket 连接。
⚠️ 此处使用 TickDB WebSocket 端点示例。
实际使用时请替换为真实 API 地址。
"""
ws_url = f"wss://api.tickdb.ai/ws?symbol={self.symbol}&api_key={self.api_key}"
self._session = aiohttp.ClientSession()
self._ws = await self._session.ws_connect(
ws_url,
receive_timeout=60, # WebSocket 接收超时
)
self._running = True
self.manager.state = ConnectionState.MARKET_OPEN
self._retry_count = 0
await self.manager.start()
logger.info(f"[{self.symbol}] WebSocket 连接已建立")
# 启动心跳任务
asyncio.create_task(self._heartbeat_loop())
# 启动消息接收循环
await self._receive_loop()
async def _heartbeat_loop(self):
"""心跳保活：每 30 秒发送一次 ping"""
while self._running and self._ws and not self._ws.closed:
try:
await asyncio.sleep(30)
if self._ws and not self._ws.closed:
await self._ws.send_str(json.dumps({"cmd": "ping"}))
logger.debug(f"[{self.symbol}] 心跳已发送")
except asyncio.CancelledError:
break
except Exception as e:
logger.warning(f"[{self.symbol}] 心跳发送失败: {e}")
break
async def _receive_loop(self):
"""消息接收主循环"""
while self._running and self._ws:
try:
msg = await self._ws.receive(timeout=65)
if msg.type == WSMsgType.TEXT:
data = json.loads(msg.data)
# 处理心跳响应
if data.get("type") == "pong":
logger.debug(f"[{self.symbol}] 心跳响应收到")
continue
# 处理限频响应
if data.get("code") == 3001:
retry_after = int(self._ws.response_headers.get("Retry-After", 5))
logger.warning(f"[{self.symbol}] 限频触发，等待 {retry_after}s")
await asyncio.sleep(retry_after)
continue
# 正常数据：通知状态管理器
self.manager.on_data_received()
if self.on_data:
self.on_data(data)
elif msg.type == WSMsgType.ERROR:
logger.error(f"[{self.symbol}] WebSocket 错误: {msg.data}")
await self._handle_disconnect()
break
elif msg.type == WSMsgType.CLOSE:
logger.warning(f"[{self.symbol}] 服务端主动关闭连接")
await self._handle_disconnect()
break
except asyncio.TimeoutError:
logger.warning(f"[{self.symbol}] WebSocket 接收超时（65s 无消息）")
# 超时不一定是断连，查市场日历
if not self.calendar.is_market_data_expected():
logger.info(f"[{self.symbol}] 非交易时段超时，忽略")
else:
await self._handle_disconnect()
break
async def _handle_disconnect(self):
"""处理断连：指数退避重连"""
self._running = False
self.manager.state = ConnectionState.DISCONNECTED
await self.manager.stop()
# 指数退避 + 抖动
base_delay = self._retry_delay
delay = min(base_delay * (2 ** self._retry_count), self._max_retry_delay)
jitter = random.uniform(0, delay * 0.1)
wait_time = delay + jitter
self._retry_count += 1
logger.info(f"[{self.symbol}] {wait_time:.1f}s 后尝试第 {self._retry_count} 次重连")
await asyncio.sleep(wait_time)
# 重连（仅在应该重连时重连）
if self._should_reconnect():
try:
await self.connect()
except Exception as e:
logger.error(f"[{self.symbol}] 重连失败: {e}")
await self._handle_disconnect()
def _should_reconnect(self) -> bool:
"""
判断是否应该重连。
如果当前是非交易时段，延迟到下一个交易时段再重连，
避免无意义地消耗 API 调用额度。
"""
session = self.calendar.get_current_session()
next_in = self.calendar.seconds_until_next_session()
if session == "CLOSED":
logger.info(f"[{self.symbol}] 非交易时段，跳过重连")
return False
if session == "BREAK" and next_in and next_in > 600:
# 午休期超过 10 分钟才断开，延迟到开盘重连
logger.info(f"[{self.symbol}] 午休中，延迟 {next_in}s 后重连")
return False
return True
async def close(self):
"""主动关闭连接"""
self._running = False
if self._ws:
await self._ws.close()
if self._session:
await self._session.close()
await self.manager.stop()
logger.info(f"[{self.symbol}] 连接已关闭")
3.4 使用示例
async def main():
api_key = os.environ.get("TICKDB_API_KEY")
# 初始化市场日历（节假日列表建议从配置文件加载）
holidays = {"2024-01-01", "2024-02-10", "2024-04-04", "2024-12-25"}
calendar = HKExCalendar(holiday_dates=holidays)
async def handle_alert(level: str, message: str):
"""告警处理：可接入飞书/Slack/邮件"""
# await send_feishu_message(message)
# await send_slack_alert(level, message)
print(f"📢 [{level}] {message}")
async def handle_data(data: dict):
"""数据处理回调"""
print(f"📊 数据: {data}")
client = HKStockWebSocketClient(
symbol="0700.HK", # 腾讯控股
api_key=api_key,
calendar=calendar,
on_data=handle_data,
on_alert=handle_alert,
)
try:
await client.connect()
except KeyboardInterrupt:
await client.close()
if __name__ == "__main__":
asyncio.run(main())
四、午休期前后 5 分钟的策略建议
除了代码实现，午休期的监控策略设计本身也有几个值得注意的点：
4.1 午休前预警
在 11:55（午休前 5 分钟）发出"即将进入午休"通知，让值班人员提前知情，而不是在 12:00 突然看到"断连告警"再手忙脚乱地排查。这个预警应该标记为 INFO 级别，不是告警。
async def _break_warning_check(self):
"""午休前预警（可在 monitor_loop 中定期调用）"""
session = self.calendar.get_current_session()
next_in = self.calendar.seconds_until_next_session()
if session == "AM" and next_in and next_in <= 300:
logger.info(
f"[{self.symbol}] 将在 {next_in}s 后进入午休时段，"
f"数据推送将暂停约 1 小时"
)
4.2 午盘开盘重连策略
13:00 午盘开盘时，订单簿会有一个"恢复快照"推送，这个时间点通常会出现数据量的一个小高峰。如果你的策略依赖开盘数据，开盘后 5 分钟是重连和校验连接质量的窗口期。
async def _validate_reconnection(self):
"""午盘开盘后验证连接质量"""
await asyncio.sleep(10) # 等待市场数据稳定
# 检查 10 秒内是否收到了数据
recent = list(self.manager.metrics.recent_silences)
if all(s < 15 for s in recent[-3:]):
logger.info(f"[{self.symbol}] 午盘连接验证通过，数据流正常")
else:
logger.warning(f"[{self.symbol}] 午盘连接质量存疑，近 3 次静默: {recent[-3:]}")
4.3 全局监控仪表盘
如果你同时监控多只港股，可以在状态机之上加一层聚合视图：
| 标的 | 当前时段 | 最后数据时间 | 状态 | 操作建议 |
|---|---|---|---|---|
| 0700.HK | PM | 14:32:15 | ✅ 正常 | — |
| 9988.HK | BREAK | 11:58:42 | 🤫 午休 | 正常静默，13:00 恢复 |
| 3690.HK | AM | 10:15:03 | ⚠️ 静默 5 分钟 | 检查链路 |
| 9618.HK | CLOSED | 昨日 16:00 | 🌙 非交易 | — |
这样运维扫一眼就能判断哪些是真实问题，哪些是市场机制使然。
五、节假日与特殊日期的额外处理
午休只是港股"数据静默"的一种形式。还有几种静默场景需要纳入考虑：
| 场景 | 持续时间 | 数据表现 | 是否需要特殊处理 |
|---|---|---|---|
| 午休 | 1 小时 | 无成交，无 depth 更新 | 是（本文方案覆盖） |
| 圣诞节前收市 | 通常半天 | 12:00 提前收市 | 是，维护半天交易日历 |
| 台风/黑色暴雨 | 不确定 | 全天无数据 | 是，接入港交所公告 API |
| 新股上市首日 | — | 规则同正常交易日 | 否 |
| 衍生品到期日 | 通常较短 | 下午可能有成交量异动 | 否，视为正常 |
台风场景特别说明：香港天文台发出八号风球或以上警告时，港交所会暂停交易，此时整个市场处于 CLOSED 状态。风球信号可通过香港天文台 API（data.weather.gov.hk）实时获取。建议在大规模监控场景中将天气信号纳入判断逻辑。
六、部署方案速查
| 场景 | 推荐配置 | 说明 |
|---|---|---|
| 个人量化爱好者 | 单进程 + 内置日历 + 日志输出 | 简单可靠，满足需求 |
| 团队级监控 | 多进程 + Redis 共享状态 + 飞书告警 | 聚合视图，统一告警 |
| 机构级 | 独立监控服务 + 港交所天气 API + 节假日自动更新 | 全自动化，减少人工维护 |
结语
"没有数据"不等于"连接断了"——这句话是整篇文章的核心。
港股午休的 1 小时静默，是市场机制的一部分，不是故障。监控系统如果不能理解这个机制，就会把正常当异常，把平静当危机。处理它的方法不复杂：给系统装一个"市场时钟"，让它知道当前是 AM、BREAK 还是 PM，在 BREAK 时段自动放宽告警阈值，在 AM/PM 时段恢复严格监控。
但更根本的是：在设计任何依赖实时数据的系统时，第一步永远是理解数据源的市场规则。港股午休、A股T+1、美股盘前盘后，每个市场都有自己的"呼吸节奏"。你的监控系统只有踩对了这个节奏，才能真正做到"有事才告警，没事不打扰"。
下一步行动
如果你的系统目前还在用"超时告警"监控港股标的：
- 下载本文的市场日历类，集成到你的 WebSocket 监控层
- 配置节假日列表（建议每季度从港交所官网更新一次）
- 用回测模式跑一周，对比误报抑制数量
如果你想直接用现成方案：
访问 tickdb.ai，在控制台生成 API Key 后，使用带 calendar 参数的 Python SDK，监控模块已内置午休感知逻辑。
如果你关心多市场监控（港股 + 美股 + A 股）：
搜索安装 tickdb-market-data SKILL，可一键获取港交所官方交易时段配置，告别手动维护日历。
风险提示：本文不构成任何投资建议。午休感知逻辑的正确性依赖于港交所公布的交易时间表，重大节假日调整请以港交所官方公告为准。市场有风险，系统设计需留足容错空间。