From fb7785ceb98a921c513e8c32be3528461871f1c5 Mon Sep 17 00:00:00 2001 From: shy3130 <415333856@qq.com> Date: Sun, 30 Aug 2026 19:05:32 +0800 Subject: [PATCH] =?UTF-8?q?feat(abnormal,dashboard):=20=E7=9B=98=E4=B8=AD?= =?UTF-8?q?=E5=BC=82=E5=8A=A8=E8=81=9A=E5=90=88=E3=80=81=E7=BB=B4=E5=BA=A6?= =?UTF-8?q?=20source=5Ffield=20=E4=B8=8E=E6=9D=BF=E5=9D=97=E5=88=86?= =?UTF-8?q?=E6=97=B6=E7=AB=AF=E7=82=B9?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - abnormal_moves 新增 build_intraday: enriched 七类当日信号聚合, 优先级(涨停>炸板>翘板>跌停>新高>新低>放量)+涨跌幅排序, GET /api/abnormal/intraday - 维度排名项携带 group_source 与 source_field(configId.field), 前端可精确判定概念/行业而非字符串包含 - ext_data 新增 dimension-intraday: 成分股×当日分钟分区等权聚合, prev_close 优先/首根退化基准、成分网格化 ffill、全市场对照线, 小数制涨跌幅契约、60s 进程内缓存、点击触发不预计算 - 板块分时 7 测 + 盘中异动 4 测 --- backend/app/api/abnormal.py | 18 +- backend/app/api/ext_data.py | 193 ++++++++++++++++++ backend/app/services/abnormal_moves.py | 57 ++++++ .../app/services/market_overview_builder.py | 4 + backend/tests/test_abnormal_intraday.py | 75 +++++++ backend/tests/test_dimension_intraday.py | 158 ++++++++++++++ 6 files changed, 503 insertions(+), 2 deletions(-) create mode 100644 backend/tests/test_abnormal_intraday.py create mode 100644 backend/tests/test_dimension_intraday.py diff --git a/backend/app/api/abnormal.py b/backend/app/api/abnormal.py index e4f6027..c75ac85 100644 --- a/backend/app/api/abnormal.py +++ b/backend/app/api/abnormal.py @@ -1,13 +1,27 @@ -"""异动边缘监控 API — 按交易所异动规则口径统计接近触发的个股。""" +"""异动监控 API — 竞价/盘中/偏移三类异动。 + +- /intraday: 盘中量价信号聚合 (enriched 当日信号列, 零新增采集) +- /overview: 偏移异动边缘总览 (交易所异动规则口径的接近度) +""" from __future__ import annotations from fastapi import APIRouter, Query, Request -from app.services.abnormal_moves import build_overview +from app.services.abnormal_moves import build_intraday, build_overview router = APIRouter(prefix="/api/abnormal", tags=["abnormal"]) +@router.get("/intraday") +def abnormal_intraday( + request: Request, + limit: int = Query(500, ge=1, le=2000), +): + """盘中异动: 涨停/炸板/跌停翘板/跌停/新高/新低/放量 信号命中行。""" + repo = request.app.state.repo + return build_intraday(repo, limit=limit) + + @router.get("/overview") def abnormal_overview( request: Request, diff --git a/backend/app/api/ext_data.py b/backend/app/api/ext_data.py index 6fc005c..67e81e9 100644 --- a/backend/app/api/ext_data.py +++ b/backend/app/api/ext_data.py @@ -7,6 +7,7 @@ import math import re import shutil import tempfile +import time from datetime import date, datetime from pathlib import Path from typing import Literal @@ -485,6 +486,198 @@ def dimension_members( } +# --------------------------------------------------------------------------- +# 板块分时 (dimension intraday) +# --------------------------------------------------------------------------- + +# 点击触发 + 60s 进程内缓存: 分钟分区是滚动底座 (minute_refresh / 盘后分钟同步), +# 不做后台预计算 — 板块基数大而单次聚合仅几十毫秒。 +_DIMENSION_INTRADAY_CACHE: dict[tuple[str, str, str, str | None], tuple[float, dict]] = {} +_DIMENSION_INTRADAY_CACHE_TTL_S = 60.0 +# 成分股网格化 ffill 上限: 超大板块退化为逐时间戳可得均值 (内存保护)。 +_DIMENSION_INTRADAY_FFILL_CAP = 2000 + + +def _bare_symbol_expr(col: str = "symbol") -> pl.Expr: + """'000001.SZ' → '000001'; 已是裸代码则原样。""" + return pl.col(col).cast(pl.String).str.strip_chars().str.split(".").list.first() + + +def _dimension_member_bares(matched: pl.DataFrame, config: ExtConfig) -> list[str]: + """成分股裸代码集合 (symbol 列优先级与 dimension-members 端点一致)。""" + if matched.is_empty(): + return [] + symbol_columns = ["symbol", "code", "股票代码", "代码"] + for mapping in (config.symbol_map, config.code_map): + if isinstance(mapping, dict) and mapping.get("type") == "mapped" and mapping.get("col"): + symbol_columns.append(str(mapping["col"])) + cols = [c for c in dict.fromkeys(symbol_columns) if c in matched.columns] + if not cols: + return [] + coalesced = pl.coalesce( + [pl.col(c).cast(pl.String).str.strip_chars().str.split(".").list.first() for c in cols] + ) + series = matched.select(coalesced.alias("_bare")).to_series() + return sorted({s for s in series.to_list() if s}) + + +def _prev_daily_close(data_dir: Path, target_date: str) -> pl.DataFrame | None: + """目标日前最近一个日K分区的收盘价 → (_bare, prev_close); 无则 None。""" + daily = data_dir / "kline_daily" + if not daily.exists(): + return None + dates = sorted( + d.name[5:] + for d in daily.iterdir() + if d.is_dir() and d.name.startswith("date=") and (d / "part.parquet").exists() + ) + prevs = [d for d in dates if d < target_date] + if not prevs: + return None + path = daily / f"date={prevs[-1]}" / "part.parquet" + if not path.exists(): + return None + df = pl.read_parquet(path, columns=["symbol", "close"]) + return ( + df.with_columns(_bare_symbol_expr().alias("_bare")) + .select([pl.col("_bare"), pl.col("close").cast(pl.Float64).alias("prev_close")]) + .unique(subset=["_bare"], keep="last") + ) + + +def _dimension_intraday_compute( + config: ExtConfig, + data_dir: Path, + field: str, + value: str, + snapshot_date: str | None, +) -> dict: + """板块等权分时: 成分股当日分钟K逐分钟平均涨跌幅 + 全市场对照线。 + + 口径: pct = 分钟close / ref − 1 (小数制, 与快照涨跌幅契约一致, 前端 ×100 显示), + ref 优先前一交易日日K收盘 (prev_close, 开盘跳空体现在曲线起点); + 日K缺失的标的退化为当日首根分钟close (混合基准)。 + 停牌/无成交分钟按成分股 forward-fill 后再平均, 全市场线取逐时间戳可得均值。 + """ + minute_dir = data_dir / "kline_minute" + partitions: list[str] = [] + if minute_dir.exists(): + partitions = sorted( + d.name[5:] + for d in minute_dir.iterdir() + if d.is_dir() and d.name.startswith("date=") and (d / "part.parquet").exists() + ) + if snapshot_date: + target = snapshot_date if snapshot_date in partitions else None + else: + target = partitions[-1] if partitions else None + if not target: + return {"status": "no_data", "reason": "minute_missing", "date": snapshot_date, "points": []} + + ext_df, _active = _read_ext_dataframe(config, data_dir) + if ext_df.is_empty() or field not in ext_df.columns: + return {"status": "empty", "reason": "no_members", "date": target, "points": []} + member_bares = _dimension_member_bares(_filter_dimension_member_rows(ext_df, field, value), config) + if not member_bares: + return {"status": "empty", "reason": "no_members", "date": target, "points": []} + + try: + bars = pl.read_parquet( + minute_dir / f"date={target}" / "part.parquet", + columns=["symbol", "datetime", "close"], + ) + except Exception as exc: # noqa: BLE001 + logger.warning("dimension-intraday read minute partition failed: %s", exc) + return {"status": "no_data", "reason": "minute_schema", "date": target, "points": []} + bars = bars.drop_nulls(subset=["datetime", "close"]) + if bars.is_empty(): + return {"status": "no_data", "reason": "minute_empty", "date": target, "points": []} + bars = bars.with_columns(_bare_symbol_expr().alias("_bare")) + + prev = _prev_daily_close(data_dir, target) + joined = bars.join(prev, on="_bare", how="left") if prev is not None else bars.with_columns( + pl.lit(None, dtype=pl.Float64).alias("prev_close") + ) + refs = joined.group_by("_bare").agg( + pl.col("prev_close").first().alias("_prev"), + pl.col("close").sort_by("datetime").first().alias("_first"), + ).with_columns(pl.coalesce(["_prev", "_first"]).alias("_ref")) + n_prev = refs["_prev"].is_not_null().sum() + basis = "prev_close" if n_prev == refs.height else ("first_close" if n_prev == 0 else "mixed") + joined = ( + joined.join(refs.select(["_bare", "_ref"]), on="_bare", how="left") + .with_columns((pl.col("close") / pl.col("_ref") - 1.0).alias("_pct")) + ) + + market = joined.group_by("datetime").agg(pl.col("_pct").mean().alias("_market")) + + member_bars = joined.filter(pl.col("_bare").is_in(member_bares)) + members_with_minute = member_bars["_bare"].n_unique() if not member_bars.is_empty() else 0 + if members_with_minute == 0: + return { + "status": "empty", "reason": "no_member_bars", "date": target, + "member_count": len(member_bares), "members_with_minute": 0, "points": [], + } + if members_with_minute <= _DIMENSION_INTRADAY_FFILL_CAP: + # 网格化 (成分股 × 全时间轴) + 逐股 ffill: 停牌分钟冻结在最后价而非退出均值 + grid = ( + member_bars.select(pl.col("_bare").unique()) + .join(joined.select(pl.col("datetime").unique()), how="cross") + ) + member_bars = ( + grid.join(member_bars.select(["_bare", "datetime", "_pct"]), on=["_bare", "datetime"], how="left") + .sort(["_bare", "datetime"]) + .with_columns(pl.col("_pct").forward_fill().over("_bare")) + ) + sector = member_bars.group_by("datetime").agg(pl.col("_pct").mean().alias("_sector")) + + combined = market.join(sector, on="datetime", how="left").sort("datetime") + + def _r4(v) -> float | None: + return round(float(v), 4) if v is not None and not (isinstance(v, float) and math.isnan(v)) else None + + points = [ + { + "time": row["datetime"].strftime("%H:%M"), + "sector": _r4(row["_sector"]), + "market": _r4(row["_market"]), + } + for row in combined.iter_rows(named=True) + ] + return { + "status": "ok", + "date": target, + "basis": basis, + "member_count": len(member_bares), + "members_with_minute": members_with_minute, + "points": points, + } + + +@router.get("/{config_id}/dimension-intraday") +def dimension_intraday( + request: Request, + config_id: str, + field: str = Query(..., min_length=1), + value: str = Query(..., min_length=1), + snapshot_date: str | None = Query(None, alias="date"), +): + """板块分时走势 (等权): 成分股 × 当日分钟K聚合; 60s 缓存, 点击触发不预计算。""" + config = _store(request).get(config_id) + if not config: + raise HTTPException(404, f"配置 '{config_id}' 不存在") + + cache_key = (config_id, field, value.strip(), snapshot_date) + now = time.monotonic() + hit = _DIMENSION_INTRADAY_CACHE.get(cache_key) + if hit is not None and now - hit[0] < _DIMENSION_INTRADAY_CACHE_TTL_S: + return hit[1] + + payload = _dimension_intraday_compute(config, _data_dir(request), field, value, snapshot_date) + _DIMENSION_INTRADAY_CACHE[cache_key] = (now, payload) + return payload + + # --------------------------------------------------------------------------- # 文件上传 # --------------------------------------------------------------------------- diff --git a/backend/app/services/abnormal_moves.py b/backend/app/services/abnormal_moves.py index c324570..7e583b1 100644 --- a/backend/app/services/abnormal_moves.py +++ b/backend/app/services/abnormal_moves.py @@ -223,3 +223,60 @@ def build_overview( "counts": counts, "rows": out_rows[:limit], } + + +# ================================================================ +# 盘中异动 (量价信号聚合, 异动监控「盘中」tab) +# +# 数据源: enriched 最新快照的当日消息号列 (零新增采集): +# 涨停/跌停/跌停翘板/炸板/放量(量比≥2)/创60日新高/新低。 +# 行序 = 信号优先级 (涨停 > 炸板 > 翘板 > 跌停 > 新高 > 新低 > 放量), +# 同级按 |今日涨跌| 降序; counts 供前端筛选 chips 展示各类型数量。 +# ================================================================ + +_INTRADAY_SIGNALS: tuple[tuple[str, str], ...] = ( + ("signal_limit_up", "limit_up"), + ("signal_broken_limit_up", "broken"), + ("signal_limit_down_recovery", "recovery"), + ("signal_limit_down", "limit_down"), + ("signal_n_day_high", "new_high"), + ("signal_n_day_low", "new_low"), + ("signal_volume_surge", "volume_surge"), +) +_INTRADAY_PRIORITY = {key: i for i, (_, key) in enumerate(_INTRADAY_SIGNALS)} +_INTRADAY_COLS = ("symbol", "name", "close", "change_pct", "amplitude", + "vol_ratio_5d", "turnover_rate", "consecutive_limit_ups") + + +def build_intraday(repo: Any, limit: int = 500) -> dict[str, Any]: + """enriched 最新快照 → 当日异动信号命中行 (含各类型计数)。""" + df, cache_date = repo.get_enriched_latest() + empty = {"cache_date": cache_date.isoformat() if cache_date else None, + "counts": {}, "rows": []} + if df.is_empty() or "symbol" not in df.columns: + return empty + present = [(c, k) for c, k in _INTRADAY_SIGNALS if c in df.columns] + if not present: + return empty + + hits = df.filter(pl.any_horizontal([pl.col(c).fill_null(False) for c, _ in present])) + if hits.is_empty(): + return empty + counts = {k: int(hits[c].fill_null(False).sum()) for c, k in present} + + sig_cols = {k: hits[c].fill_null(False).to_list() for c, k in present} + base_cols = [c for c in _INTRADAY_COLS if c in hits.columns] + base = hits.select(base_cols).to_dicts() + rows: list[dict[str, Any]] = [] + for i, r in enumerate(base): + signals = [k for k, flags in sig_cols.items() if flags[i]] + rows.append({ + **{c: r.get(c) for c in base_cols}, + "signals": signals, + "_prio": min((_INTRADAY_PRIORITY[s] for s in signals), default=99), + }) + rows.sort(key=lambda r: (r["_prio"], -abs(r.get("change_pct") or 0.0))) + for r in rows: + r.pop("_prio", None) + return {"cache_date": cache_date.isoformat() if cache_date else None, + "counts": counts, "rows": rows[:limit]} diff --git a/backend/app/services/market_overview_builder.py b/backend/app/services/market_overview_builder.py index 5d7f622..a6c549b 100644 --- a/backend/app/services/market_overview_builder.py +++ b/backend/app/services/market_overview_builder.py @@ -258,10 +258,12 @@ def _dimension_rank(rows: list[dict], repo, kind: str, limit: int = 5, level: in store = ExtConfigStore(repo.store.data_dir) groups: dict[str, dict[str, dict]] = {} + group_source: dict[str, str] = {} # 组名 → 首个命中的扩展字段 "configId.field" (看板成分股弹窗用) for config in store.load_all(): field = _dimension_field(config, kind) if not field: continue + source_field = f"{config.id}.{field}" for ext_row in _read_ext_rows(repo.store.data_dir, config, field): quote = None for key in _symbol_keys(ext_row, config): @@ -277,6 +279,7 @@ def _dimension_rank(rows: list[dict], repo, kind: str, limit: int = 5, level: in parts = value.split("-") value = parts[level - 1] if level <= len(parts) else parts[-1] groups.setdefault(value, {})[symbol] = quote + group_source.setdefault(value, source_field) items = [] for name, by_symbol in groups.items(): @@ -293,6 +296,7 @@ def _dimension_rank(rows: list[dict], repo, kind: str, limit: int = 5, level: in "up_count": sum(1 for v in changes if v > 0), "down_count": sum(1 for v in changes if v < 0), "amount": sum(_finite(s.get("amount")) or 0 for s in stocks), + "source_field": group_source.get(name), "leader": { "symbol": leader.get("symbol"), "name": leader.get("name"), diff --git a/backend/tests/test_abnormal_intraday.py b/backend/tests/test_abnormal_intraday.py new file mode 100644 index 0000000..64d0258 --- /dev/null +++ b/backend/tests/test_abnormal_intraday.py @@ -0,0 +1,75 @@ +"""盘中异动聚合测试 (build_intraday, 不依赖真实网络/enriched)。 + +覆盖: 信号命中过滤、counts 计数、优先级排序 (涨停 > 炸板 > …)、 +多信号行、limit 截断、空快照与缺信号列的降级。 +""" + +from __future__ import annotations + +from datetime import date + +import polars as pl + +from app.services.abnormal_moves import build_intraday + + +class _FakeRepo: + def __init__(self, df: pl.DataFrame): + self._df = df + + def get_enriched_latest(self): + return self._df, date(2026, 8, 28) + + +def _df(rows: list[dict]) -> pl.DataFrame: + cols = ["symbol", "name", "close", "change_pct", "amplitude", "vol_ratio_5d", + "turnover_rate", "consecutive_limit_ups", + "signal_limit_up", "signal_broken_limit_up", "signal_limit_down_recovery", + "signal_limit_down", "signal_n_day_high", "signal_n_day_low", + "signal_volume_surge"] + base = {c: [] for c in cols} + for r in rows: + for c in cols: + base[c].append(r.get(c)) + return pl.DataFrame(base) + + +def test_counts_filter_and_priority(): + repo = _FakeRepo(_df([ + {"symbol": "A1", "name": "甲", "close": 10.0, "change_pct": 0.1, + "signal_limit_up": True, "signal_n_day_high": True}, + {"symbol": "B1", "name": "乙", "close": 5.0, "change_pct": -0.05, + "signal_limit_down": True}, + {"symbol": "C1", "name": "丙", "close": 8.0, "change_pct": 0.02, + "signal_volume_surge": True}, + {"symbol": "D1", "name": "丁", "close": 7.0, "change_pct": None}, # 无信号 → 不出现 + ])) + out = build_intraday(repo) + assert out["cache_date"] == "2026-08-28" + assert out["counts"] == {"limit_up": 1, "broken": 0, "recovery": 0, + "limit_down": 1, "new_high": 1, "new_low": 0, + "volume_surge": 1} + syms = [r["symbol"] for r in out["rows"]] + assert syms == ["A1", "B1", "C1"] # 优先级: 涨停 > 跌停 > 放量; 无信号被过滤 + assert out["rows"][0]["signals"] == ["limit_up", "new_high"] # 多信号按优先级序 + + +def test_limit_truncates(): + repo = _FakeRepo(_df([ + {"symbol": f"S{i}", "signal_volume_surge": True, "change_pct": 0.01} for i in range(10) + ])) + out = build_intraday(repo, limit=3) + assert len(out["rows"]) == 3 + assert out["counts"]["volume_surge"] == 10 # counts 不受 limit 影响 + + +def test_empty_snapshot(): + repo = _FakeRepo(pl.DataFrame({"symbol": [], "name": []})) + out = build_intraday(repo) + assert out["rows"] == [] and out["counts"] == {} + + +def test_missing_signal_columns_degrades(): + repo = _FakeRepo(pl.DataFrame({"symbol": ["A1"], "name": ["甲"]})) + out = build_intraday(repo) + assert out["rows"] == [] and out["counts"] == {} diff --git a/backend/tests/test_dimension_intraday.py b/backend/tests/test_dimension_intraday.py new file mode 100644 index 0000000..b98d9ba --- /dev/null +++ b/backend/tests/test_dimension_intraday.py @@ -0,0 +1,158 @@ +"""板块分时 (dimension-intraday) 纯函数测试。 + +夹具: snapshot 扩展配置 (所属概念) + kline_minute/kline_daily 分区, +验证等权口径、停牌 ffill、prev_close/首根基准与各降级状态。 +""" +from __future__ import annotations + +from datetime import datetime +from pathlib import Path + +import polars as pl + +from app.api.ext_data import _dimension_intraday_compute +from app.services.ext_data import ExtConfig + + +def _mk_config() -> ExtConfig: + return ExtConfig(id="ext_gn", label="测试概念", mode="snapshot", fields=[]) + + +def _write_ext(data_dir: Path, values: dict[str, str]) -> None: + """snapshot 扩展数据: symbol → 所属概念 标签串。""" + cfg_dir = data_dir / "ext_data" / "ext_gn" + cfg_dir.mkdir(parents=True, exist_ok=True) + df = pl.DataFrame({ + "symbol": list(values.keys()), + "所属概念": list(values.values()), + }) + df.write_parquet(cfg_dir / "part.parquet") + + +def _write_minute(data_dir: Path, day: str, rows: list[tuple[str, str, float]]) -> None: + part = data_dir / "kline_minute" / f"date={day}" / "part.parquet" + part.parent.mkdir(parents=True, exist_ok=True) + df = pl.DataFrame( + { + "symbol": [r[0] for r in rows], + "datetime": [datetime.fromisoformat(r[1]) for r in rows], + "close": [r[2] for r in rows], + }, + schema_overrides={"datetime": pl.Datetime("us")}, + ) + df.write_parquet(part) + + +def _write_daily(data_dir: Path, day: str, closes: dict[str, float]) -> None: + part = data_dir / "kline_daily" / f"date={day}" / "part.parquet" + part.parent.mkdir(parents=True, exist_ok=True) + pl.DataFrame({ + "symbol": list(closes.keys()), + "close": list(closes.values()), + }).write_parquet(part) + + +def test_dimension_intraday_equal_weight_and_ffill(tmp_path: Path) -> None: + data_dir = tmp_path / "data" + _write_ext(data_dir, {"000001.SZ": "人工智能、芯片", "000002.SZ": "人工智能"}) + # 前收: 000001=10.0 (+5%/+6%/+4%), 000002=20.0, 600000 非成分股 + _write_daily(data_dir, "2026-08-27", {"000001.SZ": 10.0, "000002.SZ": 20.0, "600000.SH": 5.0}) + # 000002 在 09:32 无成交 (停牌分钟) → ffill 沿用 20.4 + _write_minute(data_dir, "2026-08-28", [ + ("000001.SZ", "2026-08-28T09:31:00", 10.5), + ("000002.SZ", "2026-08-28T09:31:00", 20.4), + ("600000.SH", "2026-08-28T09:31:00", 5.05), + ("000001.SZ", "2026-08-28T09:32:00", 10.6), + ("600000.SH", "2026-08-28T09:32:00", 5.10), + ("000001.SZ", "2026-08-28T09:33:00", 10.4), + ("000002.SZ", "2026-08-28T09:33:00", 20.8), + ("600000.SH", "2026-08-28T09:33:00", 5.20), + ]) + + payload = _dimension_intraday_compute(_mk_config(), data_dir, "所属概念", "人工智能", None) + + assert payload["status"] == "ok" + assert payload["date"] == "2026-08-28" + assert payload["basis"] == "prev_close" + assert payload["member_count"] == 2 + assert payload["members_with_minute"] == 2 + points = payload["points"] + assert [p["time"] for p in points] == ["09:31", "09:32", "09:33"] + # 09:31: 成分等权 (5% + 2%)/2 = 3.5%; 全市场 (5+2+1)/3 ≈ 2.667% (小数制) + assert points[0]["sector"] == 0.035 + assert points[0]["market"] == 0.0267 + # 09:32: 000002 ffill 20.4 → (6% + 2%)/2 = 4.0% (无 ffill 会是 6.0%) + assert points[1]["sector"] == 0.04 + assert points[1]["market"] == 0.04 + # 09:33: (4% + 4%)/2 = 4.0% + assert points[2]["sector"] == 0.04 + + +def test_dimension_intraday_tag_no_partial_match(tmp_path: Path) -> None: + data_dir = tmp_path / "data" + _write_ext(data_dir, {"000001.SZ": "人工智能体"}) + _write_minute(data_dir, "2026-08-28", [("000001.SZ", "2026-08-28T09:31:00", 10.5)]) + payload = _dimension_intraday_compute(_mk_config(), data_dir, "所属概念", "人工智能", None) + assert payload["status"] == "empty" + assert payload["reason"] == "no_members" + + +def test_dimension_intraday_no_minute_store(tmp_path: Path) -> None: + data_dir = tmp_path / "data" + _write_ext(data_dir, {"000001.SZ": "人工智能"}) + payload = _dimension_intraday_compute(_mk_config(), data_dir, "所属概念", "人工智能", None) + assert payload["status"] == "no_data" + assert payload["reason"] == "minute_missing" + + +def test_dimension_intraday_requested_date_absent(tmp_path: Path) -> None: + data_dir = tmp_path / "data" + _write_ext(data_dir, {"000001.SZ": "人工智能"}) + _write_minute(data_dir, "2026-08-28", [("000001.SZ", "2026-08-28T09:31:00", 10.5)]) + payload = _dimension_intraday_compute(_mk_config(), data_dir, "所属概念", "人工智能", "2026-08-27") + assert payload["status"] == "no_data" + assert payload["reason"] == "minute_missing" + + +def test_dimension_intraday_members_without_bars(tmp_path: Path) -> None: + """成分股全是 ETF 等无分钟数据的标的 → empty/no_member_bars。""" + data_dir = tmp_path / "data" + _write_ext(data_dir, {"510050.SH": "人工智能"}) + _write_daily(data_dir, "2026-08-27", {"510050.SH": 3.0, "000001.SZ": 10.0}) + _write_minute(data_dir, "2026-08-28", [("000001.SZ", "2026-08-28T09:31:00", 10.5)]) + payload = _dimension_intraday_compute(_mk_config(), data_dir, "所属概念", "人工智能", None) + assert payload["status"] == "empty" + assert payload["reason"] == "no_member_bars" + assert payload["member_count"] == 1 + + +def test_dimension_intraday_first_close_basis(tmp_path: Path) -> None: + """无前一交易日日K → 基准退化为当日首根 close, 曲线起点 ≈ 0。""" + data_dir = tmp_path / "data" + _write_ext(data_dir, {"000001.SZ": "人工智能"}) + _write_minute(data_dir, "2026-08-28", [ + ("000001.SZ", "2026-08-28T09:31:00", 10.0), + ("000001.SZ", "2026-08-28T09:32:00", 10.3), + ]) + payload = _dimension_intraday_compute(_mk_config(), data_dir, "所属概念", "人工智能", None) + assert payload["status"] == "ok" + assert payload["basis"] == "first_close" + assert payload["points"][0]["sector"] == 0.0 + assert payload["points"][1]["sector"] == 0.03 + + +def test_dimension_intraday_explicit_date_uses_that_partition(tmp_path: Path) -> None: + data_dir = tmp_path / "data" + _write_ext(data_dir, {"000001.SZ": "人工智能"}) + _write_daily(data_dir, "2026-08-26", {"000001.SZ": 10.0}) + _write_daily(data_dir, "2026-08-27", {"000001.SZ": 11.0}) + _write_minute(data_dir, "2026-08-27", [("000001.SZ", "2026-08-27T09:31:00", 10.5)]) + _write_minute(data_dir, "2026-08-28", [("000001.SZ", "2026-08-28T09:31:00", 12.1)]) + # 默认取最新分区 2026-08-28 → prev 为 08-27 的 11.0 → +10% + latest = _dimension_intraday_compute(_mk_config(), data_dir, "所属概念", "人工智能", None) + assert latest["date"] == "2026-08-28" + assert latest["points"][0]["sector"] == 0.1 + # 显式指定 08-27 → prev 为 08-26 的 10.0 → +5% + explicit = _dimension_intraday_compute(_mk_config(), data_dir, "所属概念", "人工智能", "2026-08-27") + assert explicit["date"] == "2026-08-27" + assert explicit["points"][0]["sector"] == 0.05