From f84cb671ef89c63b3152f1982e189bbf06b7d56c Mon Sep 17 00:00:00 2001 From: shy3130 <415333856@qq.com> Date: Tue, 1 Sep 2026 13:03:54 +0800 Subject: [PATCH 1/2] =?UTF-8?q?feat(data):=20=E5=85=A8=E9=87=8F=E5=88=86?= =?UTF-8?q?=E9=92=9F=E5=81=A5=E5=BA=B7=E6=97=B6=E7=AD=96=E7=95=A5=E9=A1=B5?= =?UTF-8?q?=E4=B8=8E=E7=9B=91=E6=8E=A7=E6=B3=A8=E5=85=A5=E5=88=87=E8=AF=BB?= =?UTF-8?q?=E6=9C=AC=E5=9C=B0=E5=88=86=E6=97=B6?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 策略页: healthy 时解除分时列 100 只截断并 prefer_local 读本地分区 - 监控信号注入: 股票 healthy 时直接读本地当日分区, 免每分钟 bucket 一次的全量 API 拉取; ETF 不在服务 universe, 不健康/读失败回落原路径 - minute-refresh status 增加 healthy 读侧新鲜度字段 (前端共享判断) - 测试: 注入本地读 3 用例 (健康读本地/不健康回落/ETF 恒走 API) --- backend/app/services/minute_refresh.py | 3 +- backend/app/services/quote_service.py | 21 +++++- backend/tests/test_custom_provider_indices.py | 75 +++++++++++++++++++ frontend/src/lib/api.ts | 2 + frontend/src/pages/Screener.tsx | 16 +++- 5 files changed, 108 insertions(+), 9 deletions(-) diff --git a/backend/app/services/minute_refresh.py b/backend/app/services/minute_refresh.py index cc809d8..ce3b97f 100644 --- a/backend/app/services/minute_refresh.py +++ b/backend/app/services/minute_refresh.py @@ -298,7 +298,7 @@ class MinuteRefreshService: def is_healthy(self) -> bool: """读侧 freshness 判定: 本地 kline_minute 分区是否正被服务持续写入。 - 条件: 偏好开启 + 轮询线程存活 + 最近一轮距现在 ≤ max(2×间隔, 30s)。 + 条件: 偏好开启 + 轮询线程存活 + 最近一轮距现在 ≤ max(2*间隔, 30s)。 健康时消费方 (自选分时等) 可信任本地分区、跳过批量补拉; 连续失败轮不更新 last_round_at → 自动超时判不健康, 无需单独盯 last_error。 """ @@ -325,6 +325,7 @@ class MinuteRefreshService: return { "enabled": enabled, "running": running, + "healthy": self.is_healthy(), "interval_seconds": preferences.get_minute_refresh_interval(), "capability_ok": self.capability_ok(), "custom_provider_active": self.custom_provider_active(), diff --git a/backend/app/services/quote_service.py b/backend/app/services/quote_service.py index 3a43d13..812cfbb 100644 --- a/backend/app/services/quote_service.py +++ b/backend/app/services/quote_service.py @@ -1241,11 +1241,24 @@ class QuoteService: ) capset = getattr(self._app_state, "capabilities", None) - support = intraday_monitor_support(capset) - if not support["available"] or len(symbols) > int(support["max_symbols"]): - return self._intraday_signal_evaluator.inject(enriched, []) - minute_df = fetch_intraday_monitor_batch(sorted(symbols), capset, now=now) + # 全量分钟健康时股票读本地分区 (服务按间隔持续落盘, 与 API 同一列契约), + # 免去每分钟 bucket 一次的全量 API 拉取; ETF 不在服务 universe 内, + # 本地读空/异常回落原 API 路径 (含能力与上限检查) + minute_df = pl.DataFrame() + if asset_type == "stock": + svc = getattr(self._app_state, "minute_refresh", None) if self._app_state else None + if svc is not None and svc.is_healthy() and self._repo is not None: + try: + minute_df = self._repo.get_minute_batch(sorted(symbols), cn_today()) + except Exception as e: # 本地读异常回落 API + logger.warning("分时信号本地读失败, 回退 API 路径: %s", e) + minute_df = pl.DataFrame() + if minute_df.is_empty(): + support = intraday_monitor_support(capset) + if not support["available"] or len(symbols) > int(support["max_symbols"]): + return self._intraday_signal_evaluator.inject(enriched, []) + minute_df = fetch_intraday_monitor_batch(sorted(symbols), capset, now=now) prev_close: dict[str, float] = {} available_cols = set(enriched.columns) for row in enriched.filter(pl.col("symbol").is_in(sorted(symbols))).iter_rows(named=True): diff --git a/backend/tests/test_custom_provider_indices.py b/backend/tests/test_custom_provider_indices.py index 99fec24..1805539 100644 --- a/backend/tests/test_custom_provider_indices.py +++ b/backend/tests/test_custom_provider_indices.py @@ -11,6 +11,8 @@ from __future__ import annotations from types import SimpleNamespace from typing import ClassVar +import polars as pl + from app.services import quote_service as qs from app.services.index_const import CORE_INDEX_SYMBOLS @@ -114,3 +116,76 @@ def test_custom_provider_index_fetch_error_is_soft(monkeypatch): service, captured = _service_with_provider(monkeypatch, _Boom()) service._fetch_full_market_quotes() assert len(captured) == 1 and captured[0][0]["symbol"] == "600519.SH" + + +# ---- 监控分时注入: 全量分钟健康时股票读本地分区 ---- + + +def _injection_env(monkeypatch, *, healthy, local_df, asset_type="stock", symbols=None): + """构造 _inject_intraday_signals 最小环境, 捕获传入 evaluator 的 minute_df。""" + symbols = symbols or {"600519.SH"} + service = qs.QuoteService() + service._repo = SimpleNamespace( + get_minute_batch=lambda syms, d: local_df, + ) + engine = SimpleNamespace( + intraday_signal_symbols=lambda at: set(symbols) if at == asset_type else set(), + ) + minute_svc = SimpleNamespace(is_healthy=lambda: healthy) + service._app_state = SimpleNamespace(minute_refresh=minute_svc) + + import app.services.quote_service as qsm + api_calls: list[list[str]] = [] + monkeypatch.setattr( + qsm, "_noop", qsm.__dict__.get("_noop", None), raising=False) # 占位无操作 + from app.services.kline_sync import intraday_monitor_support + monkeypatch.setattr( + "app.services.quote_service.logger", qsm.logger, raising=False) + # 打桩 API 拉取路径 (健康时不应被调) + import app.services.kline_sync as ks + monkeypatch.setattr( + ks, "fetch_intraday_monitor_batch", + lambda symbols, capset, *, now=None: (api_calls.append(list(symbols)), local_df)[1]) + monkeypatch.setattr( + ks, "intraday_monitor_support", + lambda capset: {"available": True, "max_symbols": 200, "source": "minute_batch"}) + + evaluator = SimpleNamespace( + evaluate=lambda minute_df, **kw: (captured.append(minute_df), [])[1], + inject=lambda enriched, signals: enriched, + ) + captured: list = [] + service._intraday_signal_evaluator = evaluator + service._intraday_signal_bucket = {} + return service, engine, api_calls, captured + + +def test_intraday_signals_read_local_when_healthy(monkeypatch): + """健康时股票读本地分区, 不触发 API 拉取。""" + local = pl.DataFrame({ + "symbol": ["600519.SH"], "datetime": ["2026-01-15 09:31:00"], + "open": [100.0], "high": [101.0], "low": [99.0], "close": [100.5], + "volume": [1000.0], "amount": [100500.0], + }) + service, engine, api_calls, captured = _injection_env(monkeypatch, healthy=True, local_df=local) + enriched = pl.DataFrame({"symbol": ["600519.SH"], "close": [100.5]}) + service._inject_intraday_signals(enriched, engine, asset_type="stock") + assert api_calls == [] # 未走 API + assert captured and captured[0].height == 1 # evaluator 拿到本地数据 + + +def test_intraday_signals_fall_back_to_api_when_unhealthy(monkeypatch): + """不健康 (服务关/挂) 时回落原 API 拉取路径。""" + service, engine, api_calls, captured = _injection_env(monkeypatch, healthy=False, local_df=pl.DataFrame()) + enriched = pl.DataFrame({"symbol": ["600519.SH"], "close": [100.5]}) + service._inject_intraday_signals(enriched, engine, asset_type="stock") + assert api_calls == [["600519.SH"]] # 走了 API + + +def test_intraday_signals_etf_never_reads_local(monkeypatch): + """ETF 不在全量分钟 universe: 即使健康也走 API 路径。""" + service, engine, api_calls, captured = _injection_env( + monkeypatch, healthy=True, local_df=pl.DataFrame(), asset_type="etf", symbols={"510300.SH"}) + enriched = pl.DataFrame({"symbol": ["510300.SH"], "close": [4.0]}) + service._inject_intraday_signals(enriched, engine, asset_type="etf") + assert api_calls == [["510300.SH"]] diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index 03dd8b8..f7f703e 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -1803,6 +1803,8 @@ export const api = { available: boolean enabled?: boolean running?: boolean + /** 读侧 freshness: 本地分区正被服务持续写入 (前端据此切本地读/解除截断) */ + healthy?: boolean interval_seconds?: number capability_ok?: boolean custom_provider_active?: boolean diff --git a/frontend/src/pages/Screener.tsx b/frontend/src/pages/Screener.tsx index 4fa37a1..c7f09ee 100644 --- a/frontend/src/pages/Screener.tsx +++ b/frontend/src/pages/Screener.tsx @@ -438,6 +438,14 @@ export function Screener() { ) // 分时图依赖分钟K批量数据 (kline.minute.batch), 无数据时开了列也不拉 const caps = useCapabilities() + // 全量分钟服务健康 (freshness 契约): 健康时本地分区按配置间隔持续落盘, + // 分时读本地不受批量上限约束 → 不截断 + prefer_local; 与监控设置页共享缓存 + const refreshStatus = useQuery({ + queryKey: ['minute-refresh-status'], + queryFn: api.minuteRefreshStatus, + refetchInterval: 15000, + }) + const fullMinuteHealthy = !!refreshStatus.data?.healthy const hasMinuteBatch = !!caps.data?.capabilities?.['kline.minute.batch'] const intradayVisible = !!intradayColumn && hasMinuteBatch && intradayChartVisible @@ -456,11 +464,11 @@ export function Screener() { () => displayRows.map((r: any) => r.symbol), [displayRows], ) - const intradayTruncated = intradayVisible && allIntradaySymbols.length > minuteBatchCap - // 截断到 batch 上限, 一次请求 = 一次数据源调用 + // 拉模型 (走批量接口) 才截断到 batch 上限; 全量分钟健康时读本地分区无上限 + const intradayTruncated = intradayVisible && !fullMinuteHealthy && allIntradaySymbols.length > minuteBatchCap const intradaySymbols = useMemo( () => intradayTruncated ? allIntradaySymbols.slice(0, minuteBatchCap) : allIntradaySymbols, - [allIntradaySymbols, intradayTruncated, minuteBatchCap], + [allIntradaySymbols, intradayTruncated, minuteBatchCap, fullMinuteHealthy], ) const intradayRequestSymbols = useMemo( () => [...new Set(intradaySymbols)].sort(), @@ -471,7 +479,7 @@ export function Screener() { const minuteBatch = useQuery({ queryKey: QK.minuteBatch(intradaySymbolsKey), // 增量轮询: 读缓存以最后一根为 since 只拉新增, 本地合并为完整序列 - queryFn: () => fetchMinuteBatchIncremental(qc, QK.minuteBatch(intradaySymbolsKey), intradayRequestSymbols), + queryFn: () => fetchMinuteBatchIncremental(qc, QK.minuteBatch(intradaySymbolsKey), intradayRequestSymbols, fullMinuteHealthy), enabled: intradayVisible && intradayRequestSymbols.length > 0, staleTime: 10_000, placeholderData: previousData => previousData, From bc9461da9ad9a09df33d574c3d7c132fb2132766 Mon Sep 17 00:00:00 2001 From: shy3130 <415333856@qq.com> Date: Tue, 1 Sep 2026 13:04:00 +0800 Subject: [PATCH 2/2] =?UTF-8?q?feat(net):=20=E5=88=86=E6=97=B6/=E6=97=A5K?= =?UTF-8?q?=E6=89=B9=E9=87=8F=E5=93=8D=E5=BA=94=E5=8F=AF=E9=80=89=20gzip?= =?UTF-8?q?=20=E5=8E=8B=E7=BC=A9=E4=B8=8E=E7=BD=91=E7=BB=9C=E8=AE=BE?= =?UTF-8?q?=E7=BD=AE?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - _gzip_payload: 偏好开启 + Accept-Encoding 协商 + 超 1KB 才压 (level 6); datetime 序列化与非压缩路径一致, 保证 since 增量 字符串字典序合并不受影响 - minute-batch / daily-batch 端点各自独立偏好键, 逐请求即时生效; 实测分时 9.1MB→1.28MB (86%), 日K 736KB→156KB (79%) - 超时设置 tab 更名网络设置, 卡内新增压缩区: 大开关 OR 联动 (任一子开即亮, 全关才灭, 点击全开/全关) + 分时/日K两个子开关 - preferences.save() 加线程锁: 修复并行 PUT read-modify-write 互相覆盖 (总开关批量关不齐的根因), 存量偏好写入一并受益 - 测试: 双端点压缩/关闭直通/无协商头直通/偏好默认 + 并行 save 竞态 --- backend/app/api/kline.py | 57 +++++- backend/app/api/settings.py | 26 +++ backend/app/services/preferences.py | 37 +++- backend/tests/test_minute_routing.py | 170 ++++++++++++++++++ .../src/components/data/ActiveJobCard.tsx | 2 +- frontend/src/lib/api.ts | 17 ++ frontend/src/pages/Settings.tsx | 2 +- .../src/pages/settings/JobTimeoutCard.tsx | 112 +++++++++++- 8 files changed, 405 insertions(+), 18 deletions(-) diff --git a/backend/app/api/kline.py b/backend/app/api/kline.py index 0379a6b..eb4fb64 100644 --- a/backend/app/api/kline.py +++ b/backend/app/api/kline.py @@ -1,6 +1,8 @@ """K 线 / 同步 API。""" from __future__ import annotations +import gzip +import json import logging import math from datetime import date, timedelta @@ -9,7 +11,7 @@ from zoneinfo import ZoneInfo from functools import lru_cache from typing import Optional -from fastapi import APIRouter, HTTPException, Query, Request +from fastapi import APIRouter, HTTPException, Query, Request, Response from app.indicators.pipeline import compute_enriched, compute_enriched_single from app.market_time import cn_now, cn_today, in_continuous_session @@ -22,6 +24,41 @@ logger = logging.getLogger(__name__) router = APIRouter(prefix="/api/kline", tags=["kline"]) +def _gzip_payload(request: Request, payload: dict, *, pref_key: str) -> dict | Response: + """大 JSON 响应的传输压缩: 偏好开启 + 客户端接受 gzip + 响应超阈值才压。 + + 分时/日K批量各自独立偏好键 (网络设置里大开关批量、子开关单独控制)。 + level 6 实测 13MB ≈ 290ms CPU 压掉 87%; level 9 要 2.5s 不可用。 + datetime → isoformat, 与 FastAPI jsonable_encoder 输出一致 + (前端 since 增量按字符串字典序比较, 格式必须与非压缩路径相同)。 + """ + from app.services import preferences as _prefs + _getters = { + "minute_batch_compress": _prefs.get_minute_batch_compress, + "daily_batch_compress": _prefs.get_daily_batch_compress, + } + getter = _getters.get(pref_key) + compress_on = False + if getter is not None: + try: + compress_on = bool(getter()) + except Exception: # 偏好读取异常按不压缩返回原样 + compress_on = False + headers = getattr(request, "headers", None) or {} + if compress_on and "gzip" in (headers.get("accept-encoding") or ""): + raw = json.dumps( + payload, ensure_ascii=False, separators=(",", ":"), allow_nan=True, + default=lambda o: o.isoformat() if hasattr(o, "isoformat") else str(o), + ).encode() + if len(raw) > 1024: + return Response( + content=gzip.compress(raw, 6), + media_type="application/json", + headers={"Content-Encoding": "gzip", "Vary": "Accept-Encoding"}, + ) + return payload + + def _minute_allowed(capset) -> bool: """是否有分钟K权限 (TickFlow Pro+ 或 custom minute 源)。""" from app.tickflow.capabilities import Cap @@ -567,7 +604,8 @@ def get_daily_batch(request: Request, body: dict): if not sub.is_empty(): result[sub["symbol"][0]] = sub.to_dicts() - return {"data": result} + # 日K批量同为大响应端点 (千只自选 MB 级), 与分时各自独立压缩开关 + return _gzip_payload(request, {"data": result}, pref_key="daily_batch_compress") @router.post("/minute-batch") @@ -779,12 +817,15 @@ def get_minute_batch(request: Request, body: dict): } result = {sym: rows for sym, rows in result.items() if rows} - # full_minute_local: 本轮 prefer_local 生效 (本地分区由全量分钟服务供给, 股票未做补拉) - return { - "data": result, - "full_minute_local": full_minute_healthy, - "incremental": since_dt is not None, - } + return _gzip_payload( + request, + { + "data": result, + "full_minute_local": full_minute_healthy, + "incremental": since_dt is not None, + }, + pref_key="minute_batch_compress", + ) @router.get("/minute-range") diff --git a/backend/app/api/settings.py b/backend/app/api/settings.py index a78ed30..6a7cb8f 100644 --- a/backend/app/api/settings.py +++ b/backend/app/api/settings.py @@ -427,6 +427,14 @@ class DataSourceJobTimeoutPrefs(BaseModel): data_source_long_job_timeout_s: int = Field(ge=60) +class MinuteBatchCompressPrefs(BaseModel): + minute_batch_compress: bool + + +class DailyBatchCompressPrefs(BaseModel): + daily_batch_compress: bool + + class DatasetFieldMapItem(BaseModel): source: str target: str @@ -504,6 +512,8 @@ def get_preferences() -> dict: "financial_data_provider": preferences.get_financial_provider(), "data_source_job_timeout_s": preferences.get_data_source_job_timeout_s(), "data_source_long_job_timeout_s": preferences.get_data_source_long_job_timeout_s(), + "minute_batch_compress": preferences.get_minute_batch_compress(), + "daily_batch_compress": preferences.get_daily_batch_compress(), **preferences.get_realtime_quote_scope(), "pipeline_pull_a_share": preferences.get_pipeline_pull_a_share(), "pipeline_pull_etf": preferences.get_pipeline_pull_etf(), @@ -790,6 +800,22 @@ def update_data_source_job_timeouts(req: DataSourceJobTimeoutPrefs) -> dict: return req.model_dump() +@router.put("/preferences/minute-batch-compress") +def update_minute_batch_compress(req: MinuteBatchCompressPrefs) -> dict: + """保存分时批量响应的 gzip 传输压缩开关。逐请求即时读取, 保存后立即生效。""" + from app.services import preferences + preferences.save({"minute_batch_compress": req.minute_batch_compress}) + return {"minute_batch_compress": preferences.get_minute_batch_compress()} + + +@router.put("/preferences/daily-batch-compress") +def update_daily_batch_compress(req: DailyBatchCompressPrefs) -> dict: + """保存日K批量响应的 gzip 传输压缩开关 (与分时独立)。逐请求即时读取。""" + from app.services import preferences + preferences.save({"daily_batch_compress": req.daily_batch_compress}) + return {"daily_batch_compress": preferences.get_daily_batch_compress()} + + @router.put("/preferences/mining-schedule") def update_mining_schedule(req: MiningSchedulePrefs) -> dict: """一次更新周度自动 mining 配置。""" diff --git a/backend/app/services/preferences.py b/backend/app/services/preferences.py index f5daaa6..4bbc7c9 100644 --- a/backend/app/services/preferences.py +++ b/backend/app/services/preferences.py @@ -9,6 +9,7 @@ import copy import json import logging import re +import threading from pathlib import Path logger = logging.getLogger(__name__) @@ -54,14 +55,22 @@ def load() -> dict: return copy.deepcopy(_cache) +_SAVE_LOCK = threading.Lock() + + def save(updates: dict) -> dict: - """合并写入。返回新内容。""" - current = load() - current.update(updates) - _path().write_text( - json.dumps(current, indent=2, ensure_ascii=False), encoding="utf-8", - ) - _invalidate_cache() + """合并写入。返回新内容。 + + 锁内 read-modify-write: FastAPI 同步端点跑线程池, 并行 PUT 各自基于旧快照 + 写盘会互相覆盖 (实测: 压缩总开关并行写分时/日K两键, 后写者把先写者覆盖)。 + """ + with _SAVE_LOCK: + current = load() + current.update(updates) + _path().write_text( + json.dumps(current, indent=2, ensure_ascii=False), encoding="utf-8", + ) + _invalidate_cache() return current @@ -238,6 +247,20 @@ def get_data_source_long_job_timeout_s() -> int: return max(DATA_SOURCE_JOB_TIMEOUT_MIN_S, timeout_s) +def get_minute_batch_compress() -> bool: + """分时批量响应是否启用 gzip 传输压缩。默认开启 (公网部署传输是大头); + 本机/内网可关闭省服务端 CPU。每次请求即时读取, 开关保存后立即生效。 + """ + raw = load().get("minute_batch_compress", True) + return bool(raw) + + +def get_daily_batch_compress() -> bool: + """日K批量响应是否启用 gzip 传输压缩 (与分时各自独立配置)。默认开启。""" + raw = load().get("daily_batch_compress", True) + return bool(raw) + + def _allowed_data_providers() -> set[str]: try: from app.data_providers import custom as custom_sources diff --git a/backend/tests/test_minute_routing.py b/backend/tests/test_minute_routing.py index 5e4f399..1145a4f 100644 --- a/backend/tests/test_minute_routing.py +++ b/backend/tests/test_minute_routing.py @@ -1011,3 +1011,173 @@ def test_minute_refresh_is_healthy_requires_recent_round(monkeypatch): monkeypatch.setattr(mr.preferences, "get_minute_refresh_enabled", lambda: False) svc._state.last_round_at = time_mod.time() - 10 assert svc.is_healthy() is False + + +# ---------- 测试: 分时批量传输压缩 (网络设置开关) ---------- + + +def test_minute_batch_compress_preference_default_and_toggle(monkeypatch): + """偏好默认开启; 关闭后 getter 立即反映 (逐请求读取, 无缓存)。""" + from app.services import preferences as prefs + + monkeypatch.setattr(prefs, "load", lambda: {}) + assert prefs.get_minute_batch_compress() is True + monkeypatch.setattr(prefs, "load", lambda: {"minute_batch_compress": False}) + assert prefs.get_minute_batch_compress() is False + + +def _compress_mock_env(monkeypatch, *, compress_on, accept="gzip, deflate"): + """构造 get_minute_batch 压缩路径的最小 mock 环境, 返回 (mock_request, sync_spy)。""" + from app.api import kline as kline_api + from app.services import preferences as prefs + + monkeypatch.setattr(prefs, "get_minute_batch_compress", lambda: compress_on) + + sync_spy = MagicMock(return_value=_mock_minute_df()) + monkeypatch.setattr(kline_api.kline_sync, "sync_minute_batch", sync_spy) + + mock_repo = MagicMock() + mock_repo.get_etf_symbol_set.return_value = set() + mock_repo.get_minute_batch.return_value = _mock_minute_rows("600519.SH", 100) + + mock_capset = MagicMock() + mock_capset.has.return_value = True + mock_capset.limits.return_value = None + + mock_request = MagicMock() + mock_request.app.state.repo = mock_repo + mock_request.app.state.capabilities = mock_capset + mock_request.headers = {"accept-encoding": accept} if accept else {} + return mock_request, sync_spy + + +def test_get_minute_batch_gzip_response_when_enabled(monkeypatch): + """开关开 + 客户端接受 gzip + 响应超阈值 → 返回 gzip Response, 解压后 JSON 完整。""" + import gzip as gzip_mod + from app.api import kline as kline_api + from fastapi import Response + + mock_request, _ = _compress_mock_env(monkeypatch, compress_on=True) + result = kline_api.get_minute_batch( + mock_request, {"symbols": ["600519.SH"], "date": "2026-01-15"} + ) + assert isinstance(result, Response) + assert result.headers["content-encoding"] == "gzip" + import json as json_mod + payload = json_mod.loads(gzip_mod.decompress(result.body)) + assert payload["full_minute_local"] is False + assert len(payload["data"]["600519.SH"]) > 0 + + +def test_get_minute_batch_plain_when_disabled(monkeypatch): + """开关关 → 恒返回普通 dict, 不做压缩。""" + from app.api import kline as kline_api + + mock_request, _ = _compress_mock_env(monkeypatch, compress_on=False) + result = kline_api.get_minute_batch( + mock_request, {"symbols": ["600519.SH"], "date": "2026-01-15"} + ) + assert isinstance(result, dict) and "600519.SH" in result["data"] + + +def test_get_minute_batch_plain_without_accept_encoding(monkeypatch): + """开关开但客户端未声明 gzip (如裸 curl) → 尊重协商, 原样返回。""" + from app.api import kline as kline_api + + mock_request, _ = _compress_mock_env(monkeypatch, compress_on=True, accept=None) + result = kline_api.get_minute_batch( + mock_request, {"symbols": ["600519.SH"], "date": "2026-01-15"} + ) + assert isinstance(result, dict) + + +# ---------- 测试: 日K批量传输压缩 (与分时独立开关) ---------- + + +def _daily_mock_env(monkeypatch, *, compress_on, accept="gzip, deflate"): + """构造 get_daily_batch 压缩路径的最小 mock。""" + from app.api import kline as kline_api + from app.services import preferences as prefs + + monkeypatch.setattr(prefs, "get_daily_batch_compress", lambda: compress_on) + + # 20 根日K (date 列), 足以过 1KB 阈值 + n = 20 + daily_df = pl.DataFrame({ + "symbol": ["600519.SH"] * n, + "date": [date(2026, 1, 1) + timedelta(days=i) for i in range(n)], + "open": [100.0] * n, "high": [101.0] * n, + "low": [99.0] * n, "close": [100.5] * n, "volume": [1000.0] * n, + }) + mock_repo = MagicMock() + mock_repo.resolve_asset_type.return_value = "stock" + mock_repo.get_daily_batch.return_value = daily_df + + mock_request = MagicMock() + mock_request.app.state.repo = mock_repo + mock_request.headers = {"accept-encoding": accept} if accept else {} + return mock_request + + +def test_daily_batch_gzip_when_enabled(monkeypatch): + """日K压缩开 + 接受 gzip → 压缩 Response, 解压 JSON 完整。""" + import gzip as gzip_mod + import json as json_mod + from app.api import kline as kline_api + from fastapi import Response + + req = _daily_mock_env(monkeypatch, compress_on=True) + result = kline_api.get_daily_batch(req, {"symbols": ["600519.SH"], "days": 20}) + assert isinstance(result, Response) + assert result.headers["content-encoding"] == "gzip" + payload = json_mod.loads(gzip_mod.decompress(result.body)) + assert len(payload["data"]["600519.SH"]) == 20 + + +def test_daily_batch_plain_when_disabled(monkeypatch): + from app.api import kline as kline_api + + req = _daily_mock_env(monkeypatch, compress_on=False) + result = kline_api.get_daily_batch(req, {"symbols": ["600519.SH"], "days": 20}) + assert isinstance(result, dict) and "600519.SH" in result["data"] + + +def test_daily_batch_independent_from_minute_switch(monkeypatch): + """日K与分时独立: 分时关、日K开 → 日K仍压缩 (helper 按 pref_key 走各自 getter)。""" + import gzip as gzip_mod + from app.api import kline as kline_api + from app.services import preferences as prefs + from fastapi import Response + + monkeypatch.setattr(prefs, "get_minute_batch_compress", lambda: False) + req = _daily_mock_env(monkeypatch, compress_on=True) + result = kline_api.get_daily_batch(req, {"symbols": ["600519.SH"], "days": 20}) + assert isinstance(result, Response) and gzip_mod.decompress(result.body) + + +def test_preferences_parallel_saves_do_not_lose_each_other(tmp_path, monkeypatch): + """回归: 并行 save 不同键不得互相覆盖 (压缩总开关并行 PUT 两键的竞态)。 + + save 是 read-modify-write, 无锁时两线程同时基于旧快照写盘, + 后写者会把先写者的更新覆盖掉。 + """ + import threading + from app.services import preferences as prefs + + monkeypatch.setattr(prefs, "_path", lambda: tmp_path / "preferences.json") + prefs._invalidate_cache() + prefs.save({"minute_batch_compress": True}) + + barrier = threading.Barrier(2) + + def write_key(key: str) -> None: + barrier.wait() # 尽量同时进入 save + prefs.save({key: False}) + + t1 = threading.Thread(target=write_key, args=("minute_batch_compress",)) + t2 = threading.Thread(target=write_key, args=("daily_batch_compress",)) + t1.start(); t2.start(); t1.join(); t2.join() + + final = prefs.load() + assert final["minute_batch_compress"] is False + assert final["daily_batch_compress"] is False diff --git a/frontend/src/components/data/ActiveJobCard.tsx b/frontend/src/components/data/ActiveJobCard.tsx index 7971d70..72e2527 100644 --- a/frontend/src/components/data/ActiveJobCard.tsx +++ b/frontend/src/components/data/ActiveJobCard.tsx @@ -131,7 +131,7 @@ export function ActiveJobCard({ job }: { job: PipelineJob }) { {job.error.includes('超时自动取消') && (
判定依据是「无进度」而非总时长, 任务只要仍在推进就不会被中断; - 若网络环境较慢可在 设置 → 超时设置 中调大停滞阈值。 + 若网络环境较慢可在 设置 → 网络设置 中调大停滞阈值。
)} diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index f7f703e..07f163e 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -1611,6 +1611,8 @@ export interface Preferences { financial_data_provider?: string data_source_job_timeout_s: number data_source_long_job_timeout_s: number + minute_batch_compress: boolean + daily_batch_compress: boolean realtime_pull_stock?: boolean realtime_pull_etf?: boolean pipeline_pull_a_share: boolean @@ -1785,8 +1787,23 @@ export const api = { data_source_job_timeout_s: dataSourceJobTimeoutS, data_source_long_job_timeout_s: dataSourceLongJobTimeoutS, }), + }, ), + + /** 分时批量响应 gzip 压缩开关 (网络设置) — 逐请求即时生效 */ + updateMinuteBatchCompress: (enabled: boolean) => + request>('/api/settings/preferences/minute-batch-compress', { + method: 'PUT', + body: JSON.stringify({ minute_batch_compress: enabled }), + }), + + /** 日K批量响应 gzip 压缩开关 (与分时独立) — 逐请求即时生效 */ + updateDailyBatchCompress: (enabled: boolean) => + request>('/api/settings/preferences/daily-batch-compress', { + method: 'PUT', + body: JSON.stringify({ daily_batch_compress: enabled }), + }), updateMinuteSync: (enabled: boolean, days: number, segmentDays?: number) => request('/api/settings/preferences/minute-sync', { method: 'PUT', diff --git a/frontend/src/pages/Settings.tsx b/frontend/src/pages/Settings.tsx index 4493733..f0ccb05 100644 --- a/frontend/src/pages/Settings.tsx +++ b/frontend/src/pages/Settings.tsx @@ -36,7 +36,7 @@ const TABS: readonly TabDef[] = [ { key: 'monitoring', label: '实时监控', icon: Radio, panel: SettingsMonitoringPanel }, { key: 'ext-pages', label: '扩展页面', icon: BarChart3, panel: SettingsExtPagesPanel }, { key: 'signals', label: '信号库', icon: Zap, panel: SettingsCustomSignalsPanel }, - { key: 'timeout', label: '超时设置', icon: Clock3, panel: SettingsTimeoutPanel }, + { key: 'timeout', label: '网络设置', icon: Clock3, panel: SettingsTimeoutPanel }, { key: 'menus', label: '菜单设置', icon: SlidersHorizontal, panel: SettingsMenuSettingsPanel }, { key: 'system', label: '系统设置', icon: Settings2, panel: SettingsSystemPanel }, ] diff --git a/frontend/src/pages/settings/JobTimeoutCard.tsx b/frontend/src/pages/settings/JobTimeoutCard.tsx index 2356772..5aac5cd 100644 --- a/frontend/src/pages/settings/JobTimeoutCard.tsx +++ b/frontend/src/pages/settings/JobTimeoutCard.tsx @@ -1,5 +1,6 @@ /** - * 数据任务超时配置卡片 — 从 DataSources 抽出, 放在系统设置页。 + * 网络设置面板内容 — 任务停滞超时配置 + 分时批量传输压缩开关。 + * 菜单/Tab 名为「网络设置」(Settings.tsx), 卡片内超时区块标题保持「超时设置」。 */ import { useState } from 'react' import { useMutation, useQueryClient } from '@tanstack/react-query' @@ -54,6 +55,47 @@ export function JobTimeoutCard() { const timeoutValuesChanged = regularTimeout !== currentRegularTimeout || longTimeout !== currentLongTimeout + const minuteBatchCompress = prefs.data?.minute_batch_compress ?? true + const dailyBatchCompress = prefs.data?.daily_batch_compress ?? true + // 总开关显示: 任一子开即亮, 全关才灭; 点击 = 全开/全关 (批量写两个子项) + const compressAnyOn = minuteBatchCompress || dailyBatchCompress + const toggleCompress = useMutation({ + mutationFn: (enabled: boolean) => api.updateMinuteBatchCompress(enabled), + onSuccess: (saved) => { + qc.setQueryData(QK.preferences, current => ( + current ? { ...current, ...saved } : current + )) + toast(saved.minute_batch_compress ? '分时压缩已开启' : '分时压缩已关闭', 'success') + }, + onError: (e: Error) => toast(`保存失败: ${e.message}`, 'error'), + }) + const toggleDailyCompress = useMutation({ + mutationFn: (enabled: boolean) => api.updateDailyBatchCompress(enabled), + onSuccess: (saved) => { + qc.setQueryData(QK.preferences, current => ( + current ? { ...current, ...saved } : current + )) + toast(saved.daily_batch_compress ? '日K压缩已开启' : '日K压缩已关闭', 'success') + }, + onError: (e: Error) => toast(`保存失败: ${e.message}`, 'error'), + }) + const toggleAllCompress = useMutation({ + mutationFn: async (enabled: boolean) => { + const [a, b] = await Promise.all([ + api.updateMinuteBatchCompress(enabled), + api.updateDailyBatchCompress(enabled), + ]) + return { ...a, ...b } + }, + onSuccess: (saved) => { + qc.setQueryData(QK.preferences, current => ( + current ? { ...current, ...saved } : current + )) + toast(saved.minute_batch_compress ? '传输压缩已全部开启' : '传输压缩已全部关闭', 'success') + }, + onError: (e: Error) => toast(`保存失败: ${e.message}`, 'error'), + }) + const saveJobTimeouts = useMutation({ mutationFn: () => api.updateDataSourceJobTimeouts(regularTimeout, longTimeout), onSuccess: (saved) => { @@ -152,6 +194,74 @@ export function JobTimeoutCard() { 默认 30 分钟无进度，最小 1 分钟 + +
+
+
+ 数据传输压缩 + + 大数据接口（分时、日K）启用 gzip 压缩，响应可缩至约 1/8，公网访问明显更快；本机或内网可关闭以节省服务端 CPU。任一子项开启时总开关为开，点击总开关一键全开/全关，子项可单独微调，立即生效。 + +
+ +
+
+ toggleCompress.mutate(!minuteBatchCompress)} + /> + toggleDailyCompress.mutate(!dailyBatchCompress)} + /> +
+
) } + +function CompressToggleRow({ label, desc, enabled, pending, onToggle }: { + label: string + desc: string + enabled: boolean + pending: boolean + onToggle: () => void +}) { + return ( +
+
+ {label} + {desc} +
+ +
+ ) +}