diff --git a/backend/app/plugins/fuyao/client.py b/backend/app/plugins/fuyao/client.py index f13693e..e703a79 100644 --- a/backend/app/plugins/fuyao/client.py +++ b/backend/app/plugins/fuyao/client.py @@ -207,6 +207,16 @@ class FuyaoClient: rows = data.get("item") return rows if isinstance(rows, list) else [] + def trading_days(self) -> list[dict]: + """近一年 A 股交易日序列 (固定窗口 [今日-1年, 今日], 无入参)。 + + 返回 data.item 原始行: {date_ms(上海零点), date(yyyyMMdd)}。 + 供交易日探针判定「今天在列表内 ⇔ 交易日」。 + """ + data = self._get("/api/a-share/calendar/trading-days", {}) + rows = data.get("item") + return rows if isinstance(rows, list) else [] + # ---- 市场 dump ---- def dump_download_url(self, dump_kind: str) -> dict: """获取 dump 预签名下载信息(约 300s 有效)。 diff --git a/backend/app/plugins/fuyao/provider.py b/backend/app/plugins/fuyao/provider.py index 93def17..82376e0 100644 --- a/backend/app/plugins/fuyao/provider.py +++ b/backend/app/plugins/fuyao/provider.py @@ -878,6 +878,15 @@ class FuyaoProvider: rows_out.append(row) return pl.DataFrame(rows_out) if rows_out else pl.DataFrame() + def trading_days(self) -> set: + """近一年交易日集合 (供交易日探针)。失败抛 FuyaoError, 由探针兜为未知。""" + rows = self._get_client().trading_days() + return { + d + for d in (_date_of_ms(r.get("date_ms")) for r in rows) + if d is not None + } + def _derive_bps(self, symbols: list[str]) -> dict[str, float]: """估值快照 pb_mrq 与行情快照最新价同源同刻 → bps = price / pb_mrq。 diff --git a/backend/app/services/minute_refresh.py b/backend/app/services/minute_refresh.py index 2a2bd63..3b21cae 100644 --- a/backend/app/services/minute_refresh.py +++ b/backend/app/services/minute_refresh.py @@ -149,6 +149,12 @@ class MinuteRefreshService: return "capability" if not _in_continuous_session(): return "outside_trading_hours" + # 节假日 (工作日但休市): 周几门控覆盖不到, 由交易日探针剔除。 + # 未知 (None) 维持现状 — 空轮升级机制兜底 (探针失灵时的第二道防线)。 + from app.services import trading_day + + if trading_day.is_trading_day() is False: + return "holiday" return None # ------------------------------------------------------------------ diff --git a/backend/app/services/quote_service.py b/backend/app/services/quote_service.py index 1a17521..48f94f8 100644 --- a/backend/app/services/quote_service.py +++ b/backend/app/services/quote_service.py @@ -220,6 +220,7 @@ class QuoteService: # 午休/收盘最终同步状态: 到边界后必须成功拉取一版行情, 再进入休盘态。 self._final_sync_done: set[tuple[date, str]] = set() self._final_sync_failed: dict[tuple[date, str], str] = {} + self._holiday_active = False # 交易日探针当前是否判休市 (日志去重) # ================================================================ # 生命周期 @@ -907,8 +908,29 @@ class QuoteService: return (cn_today(), "close") return None + def _holiday_gate(self) -> bool: + """交易日探针门控: 确定休市 → False (停止轮询, 含 final 定版)。 + + 探针未知 (None, 未配置 fuyao 且 tickflow 不可用/开盘缓冲窗内) → True, + 维持周几近似现状行为。探针是纯读, 不落盘; 休市结论带 TTL 定期复探, + 误判自愈。首次判定变化打一条日志, 避免每拍刷屏。 + """ + from app.services import trading_day + + holiday = trading_day.is_trading_day() is False + if holiday != self._holiday_active: + self._holiday_active = holiday + if holiday: + logger.info("交易日探针判定休市, 行情轮询暂停 (30 分钟复探)") + return not holiday + def _should_poll_for_phase(self, phase: str) -> bool: - """是否处于会主动拉行情的阶段。final 阶段成功后即停止。""" + """是否处于会主动拉行情的阶段。final 阶段成功后即停止。 + + 节假日 (工作日但休市) 由交易日探针剔除 — 周几门控覆盖不到的部分。 + """ + if not self._holiday_gate(): + return False if phase in {"preopen", "morning", "pre_afternoon", "afternoon"}: return True key = self._final_sync_key(phase) diff --git a/backend/app/services/trading_day.py b/backend/app/services/trading_day.py new file mode 100644 index 0000000..bfca3f9 --- /dev/null +++ b/backend/app/services/trading_day.py @@ -0,0 +1,139 @@ +"""交易日探针 (oracle) — 回答「今天是否 A 股交易日」。 + +消费方 (实时行情轮询 / 盘中分钟增量) 在周几+时段门控之后调用, 用于把 +「工作日但休市」的节假日从轮询窗口里剔除; 返回 None (未知) 时调用方 +维持现状行为 (周几近似 + 快照新鲜度判据兜底), 不引入新依赖。 + +探测链 (按确定性排序, 先到先得): + 1. fuyao 交易日历 (已配置 fuyao 时): GET /api/a-share/calendar/trading-days, + 今天在近一年交易日列表内 ⇔ 交易日。权威日历, 无时段依赖, 无开盘缓冲问题。 + 2. tickflow 实时行情时间戳: 拉一篮流动性票快照 (单请求), max(timestamp) + 日期 == 今天 ⇔ 交易日。非交易日全市场戳停在上一交易日 (2026-08-29 周六 + 实测 5551/5551, 含停牌股 — 戳是快照定版时刻, 非最后成交时刻); + 交易日集合竞价阶段 (9:15-9:30) 戳是否已翻新未实测 → 开盘缓冲窗内 + 戳过期不作数, 保守视为未知。 + 3. 均不可用 → None: 调用方按周几近似继续。 + +安全约束: + - 周末直接返回 False (周几判断零成本, 不打任何请求)。 + - 探针是纯读: 只产出一个布尔判定, 不落盘、不进行情管道、不碰归属链路。 + - 只用于「降档」(休市不轮询); 休市结论 TTL 较短 (30 分钟) 定期复探, + 探针误判最坏损失一段快照且可自愈; 未知结论短 TTL (5 分钟) 防止 + 轮询循环每拍重打失败的探测。 +""" + +from __future__ import annotations + +import threading +import time +from dataclasses import dataclass +from datetime import datetime, time as dt_time + +from app.market_time import CN_TZ, cn_now + +# tickflow 戳探针的开盘缓冲窗: 此时刻之前戳仍是上一交易日属正常 (集合竞价), +# 不据此判休市。周一实测竞价戳翻新时机后可收紧。仅上午首个窗口需要。 +_STALE_BUFFER_UNTIL = dt_time(9, 40) + +# 一篮流动性票: 探 max(timestamp), 任一戳为今日即交易日 (OR 语义)。 +# 大盘蓝筹同日全部停牌 = 市场性事件, 与休市同处理无碍。 +_BASKET = ("000001.SZ", "600519.SH", "600036.SH", "601318.SH", "000651.SZ") + +_TTL_TRADING_S = 3600.0 # 交易日结论每小时复探 (跨日天然失效) +_TTL_HOLIDAY_S = 1800.0 # 休市结论 30 分钟复探, 误判自愈上限 +_TTL_UNKNOWN_S = 300.0 # 未知结论 5 分钟后重试探测 + +_CACHE_LOCK = threading.Lock() + + +@dataclass +class _Cache: + day: object | None = None + verdict: bool | None = None + probed_at: float = 0.0 + + +_CACHE = _Cache() + + +def reset_cache() -> None: + """清空探针缓存 (测试用)。""" + with _CACHE_LOCK: + _CACHE.day = None + _CACHE.verdict = None + _CACHE.probed_at = 0.0 + + +def _probe_fuyao(now: datetime) -> bool | None: + """fuyao 交易日历: 今天在列表内 ⇔ 交易日。未配置 fuyao / 失败 → None。""" + try: + from app.data_providers import custom as custom_sources + + if not custom_sources.is_custom_provider("fuyao"): + return None + provider = custom_sources.get_provider("fuyao") + days = provider.trading_days() + return now.date() in days if days else None + except Exception: # noqa: BLE001 — 探针失败按未知处理, 不上抛 + return None + + +def _probe_tickflow(now: datetime) -> bool | None: + """tickflow 行情时间戳: max(timestamp) 日期 == 今天 ⇔ 交易日。 + + 戳停在上一交易日: 开盘缓冲窗内 → None (可能是竞价未翻新), 窗后 → False。 + 无实时权限 / 网络失败 / 无有效戳 → None。 + """ + try: + from app.tickflow.client import get_client + + rows = get_client().quotes.get(symbols=list(_BASKET)) or [] + stamps = [r.get("timestamp") for r in rows if isinstance(r, dict)] + valid = [int(t) for t in stamps if isinstance(t, (int, float)) and t] + if not valid: + return None + latest_day = datetime.fromtimestamp(max(valid) / 1000, tz=CN_TZ).date() + if latest_day == now.date(): + return True + if now.time() < _STALE_BUFFER_UNTIL: + return None + return False + except Exception: # noqa: BLE001 — 无权限/网络失败按未知处理 + return None + + +def is_trading_day(now: datetime | None = None) -> bool | None: + """今天是否 A 股交易日。True=交易日, False=确定休市, None=未知 (维持周几近似)。 + + 周末零成本直判; 工作日走探测链 (fuyao 日历 → tickflow 时间戳), + 结论按 TTL 缓存。线程安全: 实时行情与分钟增量两个线程共用。 + """ + now = now or cn_now() + if now.weekday() >= 5: + return False + + with _CACHE_LOCK: + if ( + _CACHE.day == now.date() + and _CACHE.verdict is not None + and (time.monotonic() - _CACHE.probed_at) < _ttl_of(_CACHE.verdict) + ): + return _CACHE.verdict + + verdict = _probe_fuyao(now) + if verdict is None: + verdict = _probe_tickflow(now) + + with _CACHE_LOCK: + _CACHE.day = now.date() + _CACHE.verdict = verdict + _CACHE.probed_at = time.monotonic() + return verdict + + +def _ttl_of(verdict: bool | None) -> float: + if verdict is True: + return _TTL_TRADING_S + if verdict is False: + return _TTL_HOLIDAY_S + return _TTL_UNKNOWN_S diff --git a/backend/tests/test_minute_refresh.py b/backend/tests/test_minute_refresh.py index f3b22a6..bbe38da 100644 --- a/backend/tests/test_minute_refresh.py +++ b/backend/tests/test_minute_refresh.py @@ -79,6 +79,10 @@ def _svc(tmp_path, monkeypatch, *, enabled=True, custom_provider=False, capabili svc = MinuteRefreshService(_FakeRepo(["600000.SH"])) svc.set_app_state(_FakeAppState(capability)) monkeypatch.setattr(minute_refresh, "_in_continuous_session", lambda now=None: in_hours) + # 交易日探针默认未知 (None → 放行): 隔离真实网络探测, holiday 分支在 + # test_trading_day.py 单独覆盖 + from app.services import trading_day + monkeypatch.setattr(trading_day, "is_trading_day", lambda now=None: None) return svc diff --git a/backend/tests/test_trading_day.py b/backend/tests/test_trading_day.py new file mode 100644 index 0000000..baa6322 --- /dev/null +++ b/backend/tests/test_trading_day.py @@ -0,0 +1,257 @@ +"""交易日探针 (trading_day oracle) 与两个消费方接入的测试。 + +不依赖真实网络: 探测函数 (_probe_fuyao / _probe_tickflow) 全部 monkeypatch。 +覆盖: 周末零成本直判、探测链优先级 (fuyao 日历权威, 无开盘缓冲问题)、 +tickflow 戳的 OR 语义与开盘缓冲窗、失败/无权限 → None、TTL 缓存、 +实时行情门控与分钟增量 gate_reason 的 holiday 分支。 +""" + +from __future__ import annotations + +from datetime import date, datetime, time as dt_time, timezone, timedelta + +import pytest + +from app.services import trading_day +from app.services.trading_day import is_trading_day, reset_cache + +CN = timezone(timedelta(hours=8)) + + +@pytest.fixture(autouse=True) +def _clean_cache(): + reset_cache() + yield + reset_cache() + + +def _no_probes(monkeypatch): + """探测函数替换为爆炸 — 用于验证未被打到。""" + monkeypatch.setattr(trading_day, "_probe_fuyao", lambda now: (_ for _ in ()).throw(AssertionError("不应探测"))) + monkeypatch.setattr(trading_day, "_probe_tickflow", lambda now: (_ for _ in ()).throw(AssertionError("不应探测"))) + + +# ---- 周末零成本直判 ---- + +def test_weekend_returns_false_without_probing(monkeypatch): + sat = datetime(2026, 8, 29, 10, 0, tzinfo=CN) # 周六 + sun = datetime(2026, 8, 30, 10, 0, tzinfo=CN) # 周日 + _no_probes(monkeypatch) + assert is_trading_day(sat) is False + assert is_trading_day(sun) is False + + +# ---- 探测链优先级 ---- + +def test_fuyao_calendar_is_authoritative_even_before_open(monkeypatch): + """fuyao 日历结论无时段依赖: 开盘缓冲窗内也直接生效。""" + holiday_mon = datetime(2026, 9, 7, 9, 31, tzinfo=CN) # 周一 (缓冲窗内) + monkeypatch.setattr(trading_day, "_probe_fuyao", lambda now: False) + monkeypatch.setattr( + trading_day, "_probe_tickflow", + lambda now: (_ for _ in ()).throw(AssertionError("fuyao 已有结论不应继续探测")), + ) + assert is_trading_day(holiday_mon) is False + + +def test_chain_falls_through_to_tickflow_when_fuyao_unknown(monkeypatch): + monday = datetime(2026, 9, 7, 10, 0, tzinfo=CN) + monkeypatch.setattr(trading_day, "_probe_fuyao", lambda now: None) + monkeypatch.setattr(trading_day, "_probe_tickflow", lambda now: True) + assert is_trading_day(monday) is True + + +def test_all_probes_unknown_returns_none(monkeypatch): + monday = datetime(2026, 9, 7, 10, 0, tzinfo=CN) + monkeypatch.setattr(trading_day, "_probe_fuyao", lambda now: None) + monkeypatch.setattr(trading_day, "_probe_tickflow", lambda now: None) + assert is_trading_day(monday) is None + + +# ---- tickflow 戳语义 ---- + +def test_tickflow_stale_stamp_before_buffer_is_unknown(monkeypatch): + """开盘缓冲窗内戳停在昨日: 可能是集合竞价未翻新, 保守判未知。""" + monday_935 = datetime(2026, 9, 7, 9, 35, tzinfo=CN) + import app.tickflow.client as tf_client_mod + + class _FakeQuotes: + def get(self, symbols): + # 周一 9:35 拉到上周五 15:30 的戳 + return [{"symbol": s, "timestamp": 1787902251001} for s in symbols] + + class _FakeClient: + quotes = _FakeQuotes() + + monkeypatch.setattr(tf_client_mod, "get_client", lambda: _FakeClient()) + monkeypatch.setattr(trading_day, "_probe_fuyao", lambda now: None) + assert trading_day._probe_tickflow(monday_935) is None + + +def test_tickflow_stale_stamp_after_buffer_is_holiday(monkeypatch): + monday_1041 = datetime(2026, 9, 7, 10, 41, tzinfo=CN) + import app.tickflow.client as tf_client_mod + + class _FakeQuotes: + def get(self, symbols): + return [{"symbol": s, "timestamp": 1787902251001} for s in symbols] # 上周五 + + class _FakeClient: + quotes = _FakeQuotes() + + monkeypatch.setattr(tf_client_mod, "get_client", lambda: _FakeClient()) + assert trading_day._probe_tickflow(monday_1041) is False + + +def test_tickflow_fresh_stamp_is_trading_anytime(monkeypatch): + """缓冲窗内只要戳是今天就判交易日 (OR 语义, 任一翻新即可)。""" + monday_920 = datetime(2026, 9, 7, 9, 20, tzinfo=CN) + import app.tickflow.client as tf_client_mod + + class _FakeQuotes: + def get(self, symbols): + # 一只翻新 + 其余停在周五 → max 为今日 + fresh_ms = int(monday_920.timestamp() * 1000) + return [ + {"symbol": "000001.SZ", "timestamp": fresh_ms}, + {"symbol": "600519.SH", "timestamp": 1787902251001}, + ] + + class _FakeClient: + quotes = _FakeQuotes() + + monkeypatch.setattr(tf_client_mod, "get_client", lambda: _FakeClient()) + assert trading_day._probe_tickflow(monday_920) is True + + +def test_tickflow_error_returns_none(monkeypatch): + monday = datetime(2026, 9, 7, 10, 41, tzinfo=CN) + import app.tickflow.client as tf_client_mod + + def _boom(): + raise RuntimeError("无实时权限 (free 档)") + + monkeypatch.setattr(tf_client_mod, "get_client", _boom) + assert trading_day._probe_tickflow(monday) is None + + +# ---- TTL 缓存 ---- + +def test_verdict_cached_within_ttl(monkeypatch): + monday = datetime(2026, 9, 7, 10, 0, tzinfo=CN) + calls = {"n": 0} + + def _counting_probe(now): + calls["n"] += 1 + return True + + monkeypatch.setattr(trading_day, "_probe_fuyao", _counting_probe) + assert is_trading_day(monday) is True + assert is_trading_day(monday) is True + assert is_trading_day(monday) is True + assert calls["n"] == 1 # 命中缓存, 只探一次 + + +def test_unknown_verdict_retries_after_short_ttl(monkeypatch): + monday = datetime(2026, 9, 7, 10, 0, tzinfo=CN) + calls = {"n": 0} + + def _counting_probe(now): + calls["n"] += 1 + return None + + monkeypatch.setattr(trading_day, "_probe_fuyao", _counting_probe) + monkeypatch.setattr(trading_day, "_probe_tickflow", lambda now: None) # 隔离真实网络 + assert is_trading_day(monday) is None + # 手动把缓存时间拨回 10 分钟前 (超过 unknown TTL 300s) → 重探 + with trading_day._CACHE_LOCK: + trading_day._CACHE.probed_at -= 600 + assert is_trading_day(monday) is None + assert calls["n"] == 2 + + +# ---- 消费方 1: 实时行情门控 ---- + +def test_quote_service_holiday_gate_blocks_polling(monkeypatch): + from app.services.quote_service import QuoteService + + qs = QuoteService() + monkeypatch.setattr(trading_day, "is_trading_day", lambda now=None: False) + assert qs._should_poll_for_phase("morning") is False + assert qs._should_poll_for_phase("preopen") is False + # final 定版同样被剔除: 休市日没有需要定版的当日行情 + assert qs._should_poll_for_phase("close_final") is False + + +def test_quote_service_polls_when_trading_or_unknown(monkeypatch): + from app.services.quote_service import QuoteService + + qs = QuoteService() + monkeypatch.setattr(trading_day, "is_trading_day", lambda now=None: True) + assert qs._should_poll_for_phase("morning") is True + # 未知 → 维持现状 (周几近似) + monkeypatch.setattr(trading_day, "is_trading_day", lambda now=None: None) + assert qs._should_poll_for_phase("morning") is True + + +# ---- 消费方 2: 分钟增量 gate_reason ---- + +def _minute_service(monkeypatch): + from app.services.minute_refresh import MinuteRefreshService + + svc = MinuteRefreshService.__new__(MinuteRefreshService) + svc._app_state = None + monkeypatch.setattr( + "app.services.minute_refresh.preferences.get_minute_refresh_enabled", + lambda: True, + ) + monkeypatch.setattr(svc, "custom_provider_active", lambda: False) + monkeypatch.setattr(svc, "capability_ok", lambda: True) + return svc + + +def test_minute_refresh_gate_returns_holiday(monkeypatch): + """周几+时段门控放行 (周一盘中) 但探针判休市 → holiday。""" + svc = _minute_service(monkeypatch) + monday_1030 = datetime(2026, 9, 7, 10, 30, tzinfo=CN) + monkeypatch.setattr( + "app.services.minute_refresh._in_continuous_session", lambda now=None: True + ) + monkeypatch.setattr(trading_day, "is_trading_day", lambda now=None: False) + assert svc._gate_reason() == "holiday" + + +def test_minute_refresh_gate_passes_when_trading(monkeypatch): + svc = _minute_service(monkeypatch) + monkeypatch.setattr( + "app.services.minute_refresh._in_continuous_session", lambda now=None: True + ) + monkeypatch.setattr(trading_day, "is_trading_day", lambda now=None: True) + assert svc._gate_reason() is None + + +# ---- fuyao 日历解析 ---- + +def test_fuyao_provider_trading_days_conversion(monkeypatch): + from app.plugins.fuyao.provider import FuyaoProvider + from app.plugins.fuyao import provider as fp + + class _CalClient: + def trading_days(self): + # 上海零点戳按 provider 的 _ms_of_date 口径构造 + def _midnight_ms(d): + import calendar + return (calendar.timegm(d.timetuple()) - 28800) * 1000 + + d4, d7 = date(2026, 9, 4), date(2026, 9, 7) + return [ + {"date": "20260904", "date_ms": _midnight_ms(d4)}, + {"date": "20260907", "date_ms": _midnight_ms(d7)}, + ] + + monkeypatch.setattr( + fp, "fuyao_client", type("M", (), {"FuyaoClient": lambda **kw: _CalClient()}) + ) + monkeypatch.setattr(fp, "get_api_key", lambda: "test-key") + days = FuyaoProvider().trading_days() + assert days == {date(2026, 9, 4), date(2026, 9, 7)}