feat(abnormal,dashboard): 盘中异动聚合、维度 source_field 与板块分时端点

- 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 测
2026-09-12 17:54:15 +08:00 · 2026-08-30 19:05:32 +08:00
parent 3d627aba5e
commit fb7785ceb9
6 changed files with 503 additions and 2 deletions
@@ -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
+
+
 # ---------------------------------------------------------------------------
 # 文件上传
 # ---------------------------------------------------------------------------