diff --git a/docs/market-insights-roadmap.md b/docs/market-insights-roadmap.md index b7399b6..0aa9fb2 100644 --- a/docs/market-insights-roadmap.md +++ b/docs/market-insights-roadmap.md @@ -8,9 +8,9 @@ | ④ | **风格轮动** `/styles` | 今天是大票还是小票、高股息还是成长 | 热点滚动基建 × FG 风格板块，纯复用 | ✅ 第一批（后并入热点滚动页内「风格」档，独立导航已移除） | | ⑦ | **大盘日历** `/calendar` | 全年情绪一眼扫完（红绿日历热力图） | 指数日K（`/bars/index`）现成 | ✅ 第一批（含悬停浮框 + 成交额编码方框大小） | | ② | **涨停生态 / 连板天梯** `/limitup` | 连板高度、首板/二板分布、炸板率、跌停 | 本地 vipdoc .day 文件（strength 扫描器同款读取器），close==涨停价 连续天数可回算 | ✅ 第二批 | -| ① | **市场情绪时间线** | 情绪处于冰点/回暖/高潮/退潮 | 涨跌家数、涨停跌停数逐分钟采样（新采样器 + sqlite） | ⏳ 第三批 | -| ⑨ | **市场宽度分时** | 指数新高但上涨家数背离的顶部信号 | 依赖 ① 的采样器 | ⏳ 第三批（随①） | -| ⑥ | **板块资金日历** | 哪天钱涌向了哪个板块 | board summary 主力净额逐日采样 | ⏳ 第三批（随①） | +| ① | **市场情绪时间线** `/sentiment` | 情绪处于冰点/回暖/高潮/退潮 | 涨跌家数、涨停跌停数逐分钟采样（新采样器 + sqlite） | ✅ 第三批 | +| ⑨ | **市场宽度分时** `/sentiment` | 指数新高但上涨家数背离的顶部信号 | 依赖 ① 的采样器 | ✅ 第三批（随①） | +| ⑥ | **板块资金日历** | 哪天钱涌向了哪个板块 | board summary 主力净额逐日采样 | ⏳ 后续批次 | | ⑤ | **板块相关性热力图** | 哪些板块同涨同跌（抱团 vs 分散） | 热点滚动已缓存的 60 日涨跌矩阵求两两相关 | ⏳ 第四批 | | ③ | **异动雷达时间线** | 异动密度骤增 = 盘面转折点 | `/mac/unusual` 现成，纯前端 | ⏳ 第四批 | | ⑧ | **量能仪表盘** | 放量/缩量（两市成交额 vs 5日均量带） | 指数分钟线现成 | ⏳ 第四批 | diff --git a/src/easy_tdx/screen/limitup.py b/src/easy_tdx/screen/limitup.py index 621dfb3..f163a1f 100644 --- a/src/easy_tdx/screen/limitup.py +++ b/src/easy_tdx/screen/limitup.py @@ -26,7 +26,12 @@ from easy_tdx.offline.paths import resolve_vipdoc _A_STOCK_TYPES = frozenset({"SH_A_STOCK", "SZ_A_STOCK"}) -__all__ = ["LimitUpEntry", "LimitUpEcology", "compute_limitup_ecology"] +__all__ = [ + "LimitUpEntry", + "LimitUpEcology", + "compute_limitup_ecology", + "compute_limitup_history", +] def _round_price(x: float) -> float: @@ -34,6 +39,10 @@ def _round_price(x: float) -> float: return math.floor(x * 100 + 0.5) / 100 +def _eq_price(a: float, b: float) -> bool: + return abs(a - b) < 1e-4 + + def _limit_ratio(code: str) -> float: """涨幅上限：创业板/科创板 20%，其余主板 10%（ST 由调用侧按 5% 二次判定）。""" if code.startswith(("30", "68")): @@ -245,3 +254,79 @@ def compute_limitup_ecology( eco.limit_down.sort(key=lambda e: (-e.streak, e.pct)) eco.blown.sort(key=lambda e: -e.pct) return eco + + +def compute_limitup_history( + vipdoc_path: str | Path | None = None, + *, + days: int = 60, + max_files: int = 20000, +) -> list[dict[str, int]]: + """逐日统计最近 ``days`` 个交易日的涨停/跌停家数（离线回补，无需采样积累）。 + + 与 :func:`compute_limitup_ecology` 的"只看最新交易日"不同，本函数把每只股票 + 窗口内的每一根 bar 都按同一涨停判定规则计数——历史日期上它就是当时真实的 + 涨停家数（陈旧文件在此是合法的历史数据，无污染问题）。 + + Returns: + 按 date 升序的 ``[{"date": YYYYMMDD, "limit_up": n, "limit_down": m}]``； + vipdoc 不可用时返回空列表。 + """ + try: + vipdoc = resolve_vipdoc(vipdoc_path) + except Exception: # noqa: BLE001 — 路径不存在/自动检测失败：按空数据处理 + return [] + + counts: dict[int, dict[str, int]] = {} + if not vipdoc.is_dir(): + return [] + + n_files = 0 + for exchange in ("sz", "sh"): + lday_dir = vipdoc / exchange / "lday" + if not lday_dir.is_dir(): + continue + for filepath in sorted(lday_dir.glob("*.day")): + if _detect_security_type(filepath.name) not in _A_STOCK_TYPES: + continue + code = filepath.name.lower()[2:8] + try: + bars = read_daily_bars(filepath) + except Exception: # noqa: BLE001 — 单文件损坏不阻塞整体 + continue + tail = bars[-(days + 13) :] + if len(tail) < 2: + continue + n_files += 1 + if n_files >= max_files: + break + up_ratio = _limit_ratio(code) + closes = [b.close for b in tail] + date_ints = [b.year * 10000 + b.month * 100 + b.day for b in tail] + for i in range(1, len(tail)): + p, c = closes[i - 1], closes[i] + if p <= 0: + continue + st_applicable = up_ratio == 0.10 and p >= 3.0 + d = date_ints[i] + bucket = counts.setdefault(d, {"limit_up": 0, "limit_down": 0}) + if _eq_price(c, _round_price(p * (1 + up_ratio))) or ( + st_applicable and _eq_price(c, _round_price(p * 1.05)) + ): + bucket["limit_up"] += 1 + elif _eq_price(c, _round_price(p * (1 - up_ratio))) or ( + st_applicable and _eq_price(c, _round_price(p * 0.95)) + ): + bucket["limit_down"] += 1 + if n_files >= max_files: + break + + recent = sorted(counts)[-days:] if days > 0 else [] + return [ + { + "date": d, + "limit_up": counts[d]["limit_up"], + "limit_down": counts[d]["limit_down"], + } + for d in recent + ] diff --git a/src/easy_tdx/web/app.py b/src/easy_tdx/web/app.py index fef387e..0b3af2a 100644 --- a/src/easy_tdx/web/app.py +++ b/src/easy_tdx/web/app.py @@ -173,8 +173,32 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: ex_client = None app.state.ex_client = ex_client + # --- 市场情绪采样器（交易时段每分钟落一条广度快照，供 /market/sentiment/*） --- + # 依赖标准 TDX 客户端（get_market_stat），mock 模式缩短间隔让曲线快速成形。 + app.state.sentiment_sampler = None + try: + from easy_tdx.web.sentiment_sampler import SentimentSampler + + sampler = SentimentSampler( + client.get_market_stat, + interval=5.0 if mock_mode else 60.0, + ) + sampler.start() + app.state.sentiment_sampler = sampler + logger.info("SentimentSampler 已启动") + except Exception: + logger.warning("SentimentSampler 启动失败 — 情绪采样不可用", exc_info=True) + yield + # --- 停止市场情绪采样器 --- + sampler_svc = getattr(app.state, "sentiment_sampler", None) + if sampler_svc is not None: + try: + await sampler_svc.stop() + except Exception: + logger.warning("SentimentSampler stop failed", exc_info=True) + # --- 关闭实时行情推送器 --- streamer_svc = getattr(app.state, "quote_streamer", None) if streamer_svc is not None: diff --git a/src/easy_tdx/web/routers/market.py b/src/easy_tdx/web/routers/market.py index d29b420..edba875 100644 --- a/src/easy_tdx/web/routers/market.py +++ b/src/easy_tdx/web/routers/market.py @@ -23,6 +23,8 @@ router = APIRouter(tags=["market"]) # 涨停生态结果缓存（vipdoc 盘中随通达信客户端落盘更新，60s 足够新鲜） _limitup_cache: tuple[float, dict[str, Any]] | None = None _LIMITUP_TTL = 60.0 +# 涨停逐日历史缓存（历史数据不变，10 分钟；按 days 分键） +_limitup_history_cache: dict[int, tuple[float, dict[str, Any]]] = {} def _df_response(df: Any) -> DataFrameResponse: @@ -130,6 +132,69 @@ async def limitup_ecology( return DictResponse.from_dict(payload) +@router.get("/market/sentiment/today", response_model=DictResponse) +async def sentiment_today( + date: int | None = Query(None, description="交易日 YYYYMMDD，缺省=最近有采样的日期"), +) -> DictResponse: + """当日情绪分钟曲线（上涨/下跌/涨停/跌停家数、上涨占比、总成交额）。 + + 数据来自 :class:`easy_tdx.web.sentiment_sampler.SentimentSampler` 的盘中 + 逐分钟采样——服务重启不丢（SQLite 持久化），但首次上线前无历史。 + """ + from easy_tdx.web.sentiment_store import get_sentiment_store + + store = get_sentiment_store() + d = date or store.latest_date() + if not d: + return DictResponse.from_dict({"date": 0, "count": 0, "samples": []}) + rows = store.day_samples(d) + for r in rows: + denom = max(r["up_count"] + r["down_count"], 1) + r["up_ratio"] = round(100.0 * r["up_count"] / denom, 1) + return DictResponse.from_dict({"date": d, "count": len(rows), "samples": rows}) + + +@router.get("/market/sentiment/history", response_model=DictResponse) +async def sentiment_history( + days: int = Query(60, ge=5, le=250, description="聚合天数"), +) -> DictResponse: + """逐日情绪聚合（收盘快照的上涨占比/涨跌停家数/成交额 + 涨停峰值）。 + + 同样依赖采样器的积累；涨停/跌停家数的"无采样历史"可用 + ``/market/limitup-history``（vipdoc 离线回补）替代。 + """ + from easy_tdx.web.sentiment_store import get_sentiment_store + + rows = get_sentiment_store().daily_history(days) + return DictResponse.from_dict({"count": len(rows), "days": rows}) + + +@router.get("/market/limitup-history", response_model=DictResponse) +async def limitup_history( + days: int = Query(60, ge=5, le=250, description="回补交易日数"), + vipdoc: str | None = Query(None, description="离线数据目录（默认自动检测）"), +) -> DictResponse: + """涨停/跌停家数逐日历史（本地 vipdoc 离线回补，无需采样积累）。 + + 全市场扫描约需数十秒，结果缓存 10 分钟。日期覆盖受 vipdoc 数据范围限制。 + """ + global _limitup_history_cache + now = time.monotonic() + cached = _limitup_history_cache.get(days) + if cached is not None and now - cached[0] < 600: + return DictResponse.from_dict(cached[1]) + + def _scan() -> dict[str, Any]: + from easy_tdx.screen.limitup import compute_limitup_history + + rows = compute_limitup_history(vipdoc, days=days) + return {"count": len(rows), "days": rows} + + payload = await asyncio.to_thread(_scan) + _limitup_history_cache[days] = (now, payload) + return DictResponse.from_dict(payload) + + @router.get("/fund-flow", response_model=DataFrameResponse) async def fund_flow( market: str = Query(..., description="市场: SZ, SH"), diff --git a/src/easy_tdx/web/sentiment_sampler.py b/src/easy_tdx/web/sentiment_sampler.py new file mode 100644 index 0000000..ae74494 --- /dev/null +++ b/src/easy_tdx/web/sentiment_sampler.py @@ -0,0 +1,99 @@ +"""市场情绪采样器（交易时段每分钟落一条全市场广度快照）。 + +模式对齐 :class:`easy_tdx.web.quote_streamer.QuoteStreamer`： + +- 后台 asyncio 任务，``start()`` 启动 / ``stop()`` 取消，进程生命周期由 + :mod:`easy_tdx.web.app` 的 lifespan 管理。 +- 仅在 :func:`easy_tdx.realtime.session.is_trading_time` 内采样（盘外采样 + 只会产生重复的静止快照，浪费且污染"当日分钟曲线"）。 +- 采样失败静默跳过（计数告警日志），绝不中断循环——情绪曲线缺失几个点 + 远好于采样器罢工。 +- 写入经 :class:`easy_tdx.web.sentiment_store.SentimentStore`，(date, minute) + 幂等主键，重复采样只覆盖不累积。 +""" + +from __future__ import annotations + +import asyncio +import logging +from datetime import datetime +from typing import Any + +from easy_tdx.realtime.session import is_trading_time +from easy_tdx.web.sentiment_store import SentimentStore, get_sentiment_store + +logger = logging.getLogger(__name__) + +__all__ = ["SentimentSampler"] + + +class SentimentSampler: + """交易时段全市场广度采样器。""" + + def __init__( + self, + client_get_stat: Any, + store: SentimentStore | None = None, + interval: float = 60.0, + ): + """ + Args: + client_get_stat: 异步可调用（``AsyncTdxClient.get_market_stat``）， + 返回含 up_count/limit_up_count 等列的单行 DataFrame。 + store: 情绪存储，None 则取进程级单例。 + interval: 采样间隔（秒）。E2E mock 可调小。 + """ + self._get_stat = client_get_stat + self._store = store or get_sentiment_store() + self._interval = interval + self._task: asyncio.Task | None = None + self.samples = 0 + self.failures = 0 + + def start(self) -> None: + if self._task is None or self._task.done(): + self._task = asyncio.create_task(self._run()) + + async def stop(self) -> None: + if self._task is not None: + self._task.cancel() + try: + await self._task + except asyncio.CancelledError: + pass + self._task = None + + async def _run(self) -> None: + logger.info("SentimentSampler 启动（间隔 %ss，仅交易时段）", self._interval) + while True: + try: + if is_trading_time(): + await self._sample_once() + except asyncio.CancelledError: + raise + except Exception: # noqa: BLE001 — 采样器永不退出 + self.failures += 1 + logger.warning("情绪采样失败（累计 %d 次）", self.failures, exc_info=True) + await asyncio.sleep(self._interval) + + async def _sample_once(self) -> None: + df = await self._get_stat() + if df is None or df.empty: + raise RuntimeError("get_market_stat 返回空数据") + row = df.iloc[0] + now = datetime.now() + self._store.insert( + { + "date": now.year * 10000 + now.month * 100 + now.day, + "minute": now.hour * 100 + now.minute, + "ts": int(now.timestamp()), + "up_count": int(row.get("up_count") or 0), + "down_count": int(row.get("down_count") or 0), + "neutral_count": int(row.get("neutral_count") or 0), + "total_count": int(row.get("total_count") or 0), + "limit_up_count": int(row.get("limit_up_count") or 0), + "limit_down_count": int(row.get("limit_down_count") or 0), + "total_amount": float(row.get("total_amount") or 0.0), + } + ) + self.samples += 1 diff --git a/src/easy_tdx/web/sentiment_store.py b/src/easy_tdx/web/sentiment_store.py new file mode 100644 index 0000000..01f7935 --- /dev/null +++ b/src/easy_tdx/web/sentiment_store.py @@ -0,0 +1,183 @@ +"""市场情绪采样持久化（「市场情绪」页的数据后端）。 + +设计对齐 :mod:`easy_tdx.web.watchlist_store` / :mod:`easy_tdx.web.llm_history_store`： + +- 单文件 SQLite，落在统一配置目录（``~/.easy_tdx/sentiment.db``， + 随 ``EASY_TDX_CONFIG_DIR`` 环境变量走）。 +- 短连接 + 写锁串行，跨线程/跨事件循环安全。 +- 由 :class:`easy_tdx.web.sentiment_sampler.SentimentSampler` 在交易时段每分钟 + 采一条全市场广度快照（涨/跌/平/涨停/跌停家数、总成交额），主键 (date, minute) + 幂等写入（采样器重启/重复采样不产生重复行）。 +- 查询侧供 ``/market/sentiment/today``（当日分钟曲线）与 + ``/market/sentiment/history``（逐日聚合）使用。 +""" + +from __future__ import annotations + +import os +import sqlite3 +import threading +from pathlib import Path +from typing import Any + +__all__ = ["SentimentStore", "get_sentiment_store"] + +_write_lock = threading.Lock() + + +def _config_dir() -> Path: + return Path(os.environ.get("EASY_TDX_CONFIG_DIR", str(Path.home() / ".easy_tdx"))) + + +class SentimentStore: + """情绪采样 SQLite 存储。""" + + def __init__(self, db_path: str | Path | None = None): + self._path = Path(db_path) if db_path else _config_dir() / "sentiment.db" + self._path.parent.mkdir(parents=True, exist_ok=True) + with _write_lock: + conn = self._connect() + try: + conn.execute( + """ + CREATE TABLE IF NOT EXISTS samples ( + date INTEGER NOT NULL, -- YYYYMMDD + minute INTEGER NOT NULL, -- HHMM + ts INTEGER NOT NULL, -- epoch 秒 + up_count INTEGER NOT NULL, + down_count INTEGER NOT NULL, + neutral_count INTEGER NOT NULL, + total_count INTEGER NOT NULL, + limit_up_count INTEGER NOT NULL, + limit_down_count INTEGER NOT NULL, + total_amount REAL NOT NULL, + PRIMARY KEY (date, minute) + ) + """ + ) + conn.execute("CREATE INDEX IF NOT EXISTS idx_samples_date ON samples(date)") + conn.commit() + finally: + conn.close() + + def _connect(self) -> sqlite3.Connection: + conn = sqlite3.connect(self._path, timeout=10) + conn.row_factory = sqlite3.Row + return conn + + def insert(self, sample: dict[str, Any]) -> None: + """写入/覆盖一条采样（同 minute 幂等，保留最新值）。""" + with _write_lock: + conn = self._connect() + try: + conn.execute( + """ + INSERT OR REPLACE INTO samples ( + date, minute, ts, up_count, down_count, neutral_count, + total_count, limit_up_count, limit_down_count, total_amount + ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + """, + ( + int(sample["date"]), + int(sample["minute"]), + int(sample["ts"]), + int(sample["up_count"]), + int(sample["down_count"]), + int(sample["neutral_count"]), + int(sample["total_count"]), + int(sample["limit_up_count"]), + int(sample["limit_down_count"]), + float(sample["total_amount"]), + ), + ) + conn.commit() + finally: + conn.close() + + def day_samples(self, date: int) -> list[dict[str, Any]]: + """某交易日的全部分钟采样（按时间升序）。""" + conn = self._connect() + try: + rows = conn.execute( + "SELECT * FROM samples WHERE date = ? ORDER BY minute", + (int(date),), + ).fetchall() + return [dict(r) for r in rows] + finally: + conn.close() + + def latest_date(self) -> int: + """最近有采样的交易日（YYYYMMDD），无数据返回 0。""" + conn = self._connect() + try: + row = conn.execute("SELECT MAX(date) AS d FROM samples").fetchone() + return int(row["d"] or 0) + finally: + conn.close() + + def daily_history(self, days: int = 60) -> list[dict[str, Any]]: + """逐日聚合（近 N 个有采样的交易日，升序）。 + + 每日输出：收盘快照（当日最后一条采样）的上涨占比/涨跌停家数/成交额， + 以及当日涨停家数峰值（情绪高潮探针）与样本数。 + """ + conn = self._connect() + try: + rows = conn.execute( + """ + SELECT c.date AS date, + c.n AS n, + c.limit_up_peak AS limit_up_peak, + l.up_count AS up_count, + l.down_count AS down_count, + l.limit_up_close AS limit_up_close, + l.limit_down_close AS limit_down_close, + l.amount_close AS amount_close + FROM ( + SELECT date, + COUNT(*) AS n, + MAX(limit_up_count) AS limit_up_peak + FROM samples GROUP BY date + ) c + JOIN ( + SELECT * + FROM ( + SELECT date, + up_count, + down_count, + limit_up_count AS limit_up_close, + limit_down_count AS limit_down_close, + total_amount AS amount_close, + ROW_NUMBER() OVER ( + PARTITION BY date ORDER BY minute DESC + ) AS rn + FROM samples + ) WHERE rn = 1 + ) l ON l.date = c.date + ORDER BY c.date DESC + LIMIT ? + """, + (int(days),), + ).fetchall() + out = [] + for r in reversed(rows): + d = dict(r) + denom = max(int(d["up_count"]) + int(d["down_count"]), 1) + d["up_ratio"] = round(100.0 * int(d["up_count"]) / denom, 1) + out.append(d) + return out + finally: + conn.close() + + +_store: SentimentStore | None = None +_store_lock = threading.Lock() + + +def get_sentiment_store() -> SentimentStore: + """进程级单例（测试可先 set ``sentiment_store._store = None`` 重置）。""" + global _store + with _store_lock: + if _store is None: + _store = SentimentStore() + return _store diff --git a/tests/unit/test_sentiment.py b/tests/unit/test_sentiment.py new file mode 100644 index 0000000..3b1065f --- /dev/null +++ b/tests/unit/test_sentiment.py @@ -0,0 +1,204 @@ +"""市场情绪采样（store / sampler / 端点）与涨停历史回补单测。 + +sentiment_store 用 EASY_TDX_CONFIG_DIR 指向临时目录；limitup 历史复用 +合成 .day 文件；端点侧验证 DictResponse 包装与缓存命中。 +""" + +from __future__ import annotations + +import asyncio + +import pytest + + +@pytest.fixture +def store(tmp_path, monkeypatch): + """独立配置目录 + 全新单例的 SentimentStore。""" + from easy_tdx.web import sentiment_store as ss + + monkeypatch.setenv("EASY_TDX_CONFIG_DIR", str(tmp_path / "cfg")) + ss._store = None + s = ss.get_sentiment_store() + yield s + ss._store = None + + +def _sample(date: int, minute: int, up=2000, down=2000, limit_up=50, limit_down=10, amount=8e11): + from datetime import datetime + + return { + "date": date, + "minute": minute, + "ts": int(datetime(2026, 9, 4).timestamp()), + "up_count": up, + "down_count": down, + "neutral_count": 100, + "total_count": up + down + 100, + "limit_up_count": limit_up, + "limit_down_count": limit_down, + "total_amount": amount, + } + + +def test_store_day_samples_and_idempotent(store): + store.insert(_sample(20260904, 935)) + store.insert(_sample(20260904, 930)) + # 同 (date, minute) 覆盖不累积 + store.insert(_sample(20260904, 930, limit_up=77)) + + rows = store.day_samples(20260904) + assert [r["minute"] for r in rows] == [930, 935] # 升序 + assert rows[0]["limit_up_count"] == 77 # 覆盖生效 + assert store.latest_date() == 20260904 + + +def test_store_daily_history_close_snapshot_and_peak(store): + # 收盘快照 = 当日最后一条采样；峰值 = 当日涨停最大值 + store.insert(_sample(20260903, 930, up=1500, limit_up=30, limit_down=40, amount=7e11)) + store.insert( + _sample(20260903, 1500, up=2500, down=1500, limit_up=90, limit_down=5, amount=9e11) + ) + store.insert( + _sample(20260904, 930, up=1800, down=2200, limit_up=20, limit_down=60, amount=6e11) + ) + + days = store.daily_history(10) + assert [d["date"] for d in days] == [20260903, 20260904] # 升序 + + d3 = days[0] + assert d3["limit_up_peak"] == 90 # 日内峰值（930 点只有 30，1500 点 90） + assert d3["limit_up_close"] == 90 # 收盘快照取当日最后一条 + assert d3["up_count"] == 2500 + assert d3["up_ratio"] == 62.5 # 2500 / (2500+1500) + + d4 = days[1] + assert d4["limit_up_peak"] == 20 + assert d4["up_ratio"] == 45.0 # 1800 / 4000 + + +def test_sampler_inserts_store_rows(store): + import pandas as pd + + from easy_tdx.web.sentiment_sampler import SentimentSampler + + df = pd.DataFrame( + [ + { + "up_count": 2100, + "down_count": 2300, + "neutral_count": 120, + "total_count": 4520, + "limit_up_count": 44, + "limit_down_count": 9, + "total_amount": 8.5e11, + } + ] + ) + + class FakeClient: + async def get_market_stat(self): + return df + + sampler = SentimentSampler(FakeClient().get_market_stat, store=store, interval=1.0) + asyncio.run(sampler._sample_once()) + + rows = store.day_samples(store.latest_date()) + assert len(rows) == 1 + assert rows[0]["limit_up_count"] == 44 + assert rows[0]["total_amount"] == 8.5e11 + + +@pytest.fixture +def vipdoc_factory(tmp_path): + """按 {文件名: {dates, closes}} 合成 vipdoc 目录的工厂。""" + from easy_tdx.offline.daily_bar import _DAILY_FMT + + def _day(date: int, close: float) -> bytes: + return _DAILY_FMT.pack( + date, + round((close - 0.05) * 100), + round(close * 100), + round((close - 0.10) * 100), + round(close * 100), + 5_000_000.0, + 1_000_000, + 0, + ) + + def factory(specs: dict[str, dict]) -> object: + for filename, spec in specs.items(): + exchange = filename[:2] + lday = tmp_path / exchange / "lday" + lday.mkdir(parents=True, exist_ok=True) + data = b"".join( + _day(d, c) for d, c in zip(spec["dates"], spec["closes"]) + ) + (lday / f"{filename}.day").write_bytes(data) + return tmp_path + + return factory + + +def test_limitup_history_counts(vipdoc_factory): + from easy_tdx.screen.limitup import compute_limitup_history + + v = vipdoc_factory( + # A 股票：0802、0803 连续两日涨停 + { + "sh600100": { + "dates": [20260801, 20260802, 20260803, 20260804], + "closes": [10.00, 11.00, 12.10, 12.50], + }, + # B 股票：0804 跌停 + "sz000200": { + "dates": [20260801, 20260802, 20260803, 20260804], + "closes": [10.00, 10.00, 10.00, 9.00], + }, + } + ) + rows = compute_limitup_history(v, days=10) + by_date = {r["date"]: r for r in rows} + assert by_date[20260802]["limit_up"] == 1 + assert by_date[20260803]["limit_up"] == 1 + assert by_date[20260804]["limit_down"] == 1 + assert by_date[20260804]["limit_up"] == 0 + # 升序 + dates = [r["date"] for r in rows] + assert dates == sorted(dates) + + +def test_limitup_history_endpoint_cache(vipdoc_factory, monkeypatch): + pytest.importorskip("fastapi") + from fastapi import FastAPI + from fastapi.testclient import TestClient + + from easy_tdx.screen import limitup as limitup_mod + from easy_tdx.web.errors import register_exception_handlers + from easy_tdx.web.routers import market as market_mod + + v = vipdoc_factory( + {"sh600100": {"dates": [20260801, 20260802], "closes": [10.0, 11.0]}} + ) + calls = {"n": 0} + real = limitup_mod.compute_limitup_history + + def counting(*a, **kw): + calls["n"] += 1 + return real(*a, **kw) + + monkeypatch.setattr(limitup_mod, "compute_limitup_history", counting) + + app = FastAPI() + register_exception_handlers(app) + app.include_router(market_mod.router, prefix="/api/v1") + app.state.tdx_client = object() + + with TestClient(app) as client: + r1 = client.get("/api/v1/market/limitup-history", params={"days": 10, "vipdoc": str(v)}) + assert r1.status_code == 200 + body = r1.json()["data"] + # 仅 0802 有一天涨停（0801 无前收不计数） + assert body["count"] == 1 + assert body["days"][0] == {"date": 20260802, "limit_up": 1, "limit_down": 0} + client.get("/api/v1/market/limitup-history", params={"days": 10, "vipdoc": str(v)}) + assert calls["n"] == 1 # 缓存命中 diff --git a/web-ui/src/App.vue b/web-ui/src/App.vue index bed2504..5a87b62 100644 --- a/web-ui/src/App.vue +++ b/web-ui/src/App.vue @@ -31,6 +31,7 @@ const sseLabel: Record = { 热点滚动 大盘日历 涨停生态 + 市场情绪 自选行情 期货持仓排名
分析
diff --git a/web-ui/src/api.ts b/web-ui/src/api.ts index 9f72bac..e2d8e68 100644 --- a/web-ui/src/api.ts +++ b/web-ui/src/api.ts @@ -14,6 +14,7 @@ import type { DataFrameResponse, HotspotResp, LimitUpEcologyResp, + LimitUpHistoryRow, LlmChatResponse, LlmChatContext, LlmHistoryResponse, @@ -32,6 +33,8 @@ import type { SavedStrategyCreate, SavedStrategyListResponse, SecurityQuote, + SentimentHistoryResp, + SentimentTodayResp, ServerHostInfo, ServerHostListResponse, ServerSwitchResult, @@ -843,6 +846,32 @@ export async function fetchLimitUpEcology(): Promise { return body.data } +/** 当日情绪分钟曲线（采样器逐分钟落库；date=0 表示尚无采样）。 */ +export async function fetchSentimentToday(): Promise { + const resp = await fetch(`${BASE}/market/sentiment/today`) + if (!resp.ok) await throwError(resp) + const body = (await resp.json()) as { data: SentimentTodayResp } + return body.data +} + +/** 逐日情绪聚合（收盘快照上涨占比 + 涨跌停家数，依赖采样积累）。 */ +export async function fetchSentimentHistory(days = 60): Promise { + const params = new URLSearchParams({ days: String(days) }) + const resp = await fetch(`${BASE}/market/sentiment/history?${params}`) + if (!resp.ok) await throwError(resp) + const body = (await resp.json()) as { data: SentimentHistoryResp } + return body.data +} + +/** 涨停/跌停家数逐日历史（vipdoc 离线回补，服务端缓存 10 分钟）。 */ +export async function fetchLimitUpHistory(days = 60): Promise { + const params = new URLSearchParams({ days: String(days) }) + const resp = await fetch(`${BASE}/market/limitup-history?${params}`) + if (!resp.ok) await throwError(resp) + const body = (await resp.json()) as { data: { count: number; days: LimitUpHistoryRow[] } } + return body.data.days +} + /** 中金所成交持仓排名：品种列表（含科普元数据）。 */ export async function fetchCcpmProducts(): Promise { const resp = await fetch(`${BASE}/ccpm/products`) diff --git a/web-ui/src/router.ts b/web-ui/src/router.ts index d3d94d7..6ed591f 100644 --- a/web-ui/src/router.ts +++ b/web-ui/src/router.ts @@ -12,6 +12,7 @@ import LlmHistoryView from './views/LlmHistoryView.vue' import LlmSettingsView from './views/LlmSettingsView.vue' import OptimizeView from './views/OptimizeView.vue' import PortfolioView from './views/PortfolioView.vue' +import SentimentView from './views/SentimentView.vue' import ServerSettingsView from './views/ServerSettingsView.vue' import SignalRadarView from './views/SignalRadarView.vue' import StrategiesView from './views/StrategiesView.vue' @@ -35,6 +36,8 @@ const routes = [ { path: '/watchlist', name: 'watchlist', component: WatchlistView }, // 涨停生态（连板天梯/炸板/跌停，本地 vipdoc 离线回算） { path: '/limitup', name: 'limitup', component: LimitUpView }, + // 市场情绪（宽度分时 + 涨停温度计；采样器盘中逐分钟积累） + { path: '/sentiment', name: 'sentiment', component: SentimentView }, { path: '/backtest', name: 'backtest', component: BacktestView }, { path: '/portfolio', name: 'portfolio', component: PortfolioView }, { path: '/optimize', name: 'optimize', component: OptimizeView }, diff --git a/web-ui/src/types.ts b/web-ui/src/types.ts index 8dbef04..cae5dc5 100644 --- a/web-ui/src/types.ts +++ b/web-ui/src/types.ts @@ -687,6 +687,54 @@ export interface LimitUpEcologyResp { blown: LimitUpEntry[] } +// ── 市场情绪（/market/sentiment/*，盘中逐分钟采样 + vipdoc 涨停史回补） ───── + +export interface SentimentSample { + date: number + minute: number + ts: number + up_count: number + down_count: number + neutral_count: number + total_count: number + limit_up_count: number + limit_down_count: number + total_amount: number + up_ratio: number +} + +export interface SentimentTodayResp { + /** 交易日 YYYYMMDD；0 = 尚无采样 */ + date: number + count: number + samples: SentimentSample[] +} + +export interface SentimentDay { + date: number + /** 当日样本数（<10 视为不完整交易日，曲线渲染时可忽略） */ + n: number + limit_up_peak: number + up_count: number + down_count: number + limit_up_close: number + limit_down_close: number + amount_close: number + up_ratio: number +} + +export interface SentimentHistoryResp { + count: number + days: SentimentDay[] +} + +/** vipdoc 回补的逐日涨停/跌停家数（无需采样积累）。 */ +export interface LimitUpHistoryRow { + date: number + limit_up: number + limit_down: number +} + // ── Walk-Forward 样本外验证（v1.27 POST /backtest/wf/run/async）────────────── export interface WalkForwardWindow { diff --git a/web-ui/src/views/SentimentView.vue b/web-ui/src/views/SentimentView.vue new file mode 100644 index 0000000..2e45cfc --- /dev/null +++ b/web-ui/src/views/SentimentView.vue @@ -0,0 +1,416 @@ + + + + +