diff --git a/src/easy_tdx/commands/security_quotes.py b/src/easy_tdx/commands/security_quotes.py index 6f30fbc..1bf8fa4 100644 --- a/src/easy_tdx/commands/security_quotes.py +++ b/src/easy_tdx/commands/security_quotes.py @@ -17,9 +17,11 @@ from .base import BaseCommand def _price_decimal_digits(market: Market, code: str) -> int: """推断某证券报价的有效小数位数。 - 通达信协议中,price 及各档差分均以「厘」(0.001 元) 为基本单位编码, - 但报价精度按品种而异:股票 2 位(分),指数/ETF/基金/可转债/国债/国债逆回购 3 位(厘)。 - 若一律按 /100 解析,ETF/指数等品种价格会被放大 10 倍(见 Issue #8)。 + 通达信协议中,price 及各档差分按品种以「分」或「厘」为基本单位编码: + 股票/大盘指数 2 位(分;指数点位恒两位小数,如科创50 1647.53), + ETF/基金/可转债/国债 3 位(厘)。 + 若一律按 /100 解析,ETF 等品种价格会被放大 10 倍(见 Issue #8); + 反之指数按 3 位解析会缩小 10 倍(见 2026-09 看板指数修复)。 decimal_point 不在行情响应包内,只能凭 market + code 代码段推断。 注意同一代码不同市场含义不同:SZ 000001=平安银行(股票,2位), @@ -34,10 +36,20 @@ def _price_decimal_digits(market: Market, code: str) -> int: if market == Market.SH: if code.startswith("5"): # 51x ETF、55x 货币基金、56x 跨境ETF、58x 科创ETF return 3 - if code.startswith("000"): # 000001 上证指数、000300 沪深300 等 - return 3 - if code.startswith("8"): # 880xxx 行业指数 + if code.startswith("000"): + # 000001 上证指数、000300 沪深300、000688 科创50 等——指数系列。 + # 实测(2026-09-01):这些指数的报价原始单位是「分」(÷100), + # 点位恒为 2 位小数(科创50 1647.53 / 沪深300 4611.44); + # 按 3 位(厘)解析会缩小 10 倍。深市指数(39xxxx)走默认 2 位同理。 + return 2 + if code.startswith("880"): + # 880xxx 统计指数(880005 涨跌家数等):字段是计数语义, + # market_stat 依赖 ÷1000 解码(price×10 还原家数),勿动。 return 3 + if code.startswith("88"): + # 881xxx 行业板块指数、885xxx 概念板块指数——同大盘指数, + # 点位两位小数(实测 881106 种植业 1039.93,按 3 位曾缩成 103.993)。 + return 2 return 2 # 60xxxx / 68xxxx 科创板 A 股 # 深圳:1/3 开头的 15x/16x/18x 为基金,12x 为可转债,11x 为国债 @@ -189,7 +201,7 @@ class GetSecurityQuotesCmd(BaseCommand[list[SecurityQuote]]): raise TdxDecodeError(f"security_quotes 非法 market 值: {market_b}") from e code = code_b.decode("utf-8").rstrip("\x00") - # 价格按品种有效小数位解析:股票/100,指数·ETF·基金/可转债/国债/1000(Issue #8) + # 价格按品种有效小数位解析:股票/大盘指数 ÷100,ETF/基金/债券/880 统计指数 ÷1000(Issue #8) divisor = 10 ** _price_decimal_digits(market, code) p = price_raw / divisor diff --git a/src/easy_tdx/web/app.py b/src/easy_tdx/web/app.py index 8f3e8ed..5aed2fc 100644 --- a/src/easy_tdx/web/app.py +++ b/src/easy_tdx/web/app.py @@ -80,6 +80,25 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: app.state.tdx_client = client + # --- 实时行情 SSE 推送器(共享轮询 + fan-out) --- + try: + from easy_tdx.models.enums import Market + from easy_tdx.web.quote_streamer import QuoteStreamer + from easy_tdx.web.watchlist_store import get_watchlist_store + + store = get_watchlist_store() + + async def _watch_symbols(): + # SQLite 存 "SH"/"SZ"/"BJ" 字符串,轮询器需要 Market 枚举 + return [(Market[mkt], code) for mkt, code in store.symbols()] + + streamer = QuoteStreamer(client.get_security_quotes, _watch_symbols) + streamer.start() + app.state.quote_streamer = streamer + except Exception: + logger.warning("QuoteStreamer 启动失败 — SSE 推送不可用", exc_info=True) + app.state.quote_streamer = None + # --- MAC 协议客户端 --- mac_client = None enable_mac = getattr(app.state, "enable_mac", True) @@ -112,6 +131,14 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: yield + # --- 关闭实时行情推送器 --- + streamer = getattr(app.state, "quote_streamer", None) + if streamer is not None: + try: + await streamer.stop() + except Exception: + logger.warning("QuoteStreamer stop failed", exc_info=True) + # --- 依次关闭 --- for name, cli in [ ("Ex market client", ex_client), @@ -214,6 +241,8 @@ def _create_app( from easy_tdx.web.routers.server import router as server_router from easy_tdx.web.routers.sina import router as sina_router from easy_tdx.web.routers.strategies import router as strategies_router + from easy_tdx.web.routers.stream import router as stream_router + from easy_tdx.web.routers.watchlist import router as watchlist_router app.include_router(market_router, prefix="/api/v1") app.include_router(bars_router, prefix="/api/v1") @@ -239,6 +268,10 @@ def _create_app( app.include_router(strategies_router, prefix="/api/v1") # 服务器设置路由(列出/测速/切换 TDX host) app.include_router(server_router, prefix="/api/v1") + # 自选股路由(SQLite 持久化,纯 CRUD,不依赖行情连接) + app.include_router(watchlist_router, prefix="/api/v1") + # 实时行情 SSE 路由(依赖 lifespan 里的 QuoteStreamer) + app.include_router(stream_router, prefix="/api/v1") # --- 前端 dist 托管(生产/打包态同源服务,开发态可缺省) --- # 必须在所有 API 路由注册之后:StaticFiles(html=True) 挂在 "/" 会吞掉 diff --git a/src/easy_tdx/web/quote_streamer.py b/src/easy_tdx/web/quote_streamer.py new file mode 100644 index 0000000..d2f887b --- /dev/null +++ b/src/easy_tdx/web/quote_streamer.py @@ -0,0 +1,233 @@ +"""实时行情 SSE 推送器:单一共享轮询循环 + 每连接独立队列 fan-out。 + +架构(借鉴 tick-stock-panel 的成熟模式,见其 QuoteService / SSE 演进): + +- 通达信协议是请求-响应式,没有服务端推送;"实时"本质是后端定时轮询。 +- 所有 SSE 连接共享**一条**轮询循环(避免 N 个标签页 = N 倍行情请求), + 每个连接持有独立的 :class:`asyncio.Queue`,消息 fan-out 投递。 +- 背压策略:队列满(说明该连接消费慢/挂起)时丢弃最旧消息、保最新—— + 行情场景下旧快照无价值,宁可跳帧不可积压。 +- 订阅集合 = 固定指数 + 全部自选(每次轮询前重读 watchlist,SQLite 单文件 + 读极快)。前端加自选后,下一个轮询周期自动纳入推送,无需重连 SSE。 +- 无人订阅时循环休眠,不产生行情请求。 +- 交易时段(沪时区 09:10-15:10)~8s 一拍,其余时段降到 60s(收盘价仍可推)。 +""" + +from __future__ import annotations + +import asyncio +import itertools +import logging +from collections.abc import Awaitable, Callable +from datetime import datetime, timedelta, timezone as dt_timezone +from typing import Any + +import pandas as pd + +from easy_tdx.models.enums import Market + +logger = logging.getLogger(__name__) + +__all__ = ["QuoteStreamer", "INDEX_SYMBOLS", "INDEX_NAMES"] + +# 看板常驻指数(标准协议行情,指数与个股同一接口)。 +INDEX_SYMBOLS: list[tuple[Market, str]] = [ + (Market.SH, "000001"), # 上证指数 + (Market.SZ, "399001"), # 深证成指 + (Market.SZ, "399006"), # 创业板指 + (Market.SH, "000688"), # 科创50 + (Market.SH, "000300"), # 沪深300 +] +INDEX_NAMES: dict[str, str] = { + "SH000001": "上证指数", + "SZ399001": "深证成指", + "SZ399006": "创业板指", + "SH000688": "科创50", + "SH000300": "沪深300", +} + +_MARKET_NAMES = {Market.SZ: "SZ", Market.SH: "SH", Market.BJ: "BJ"} + +# 推送给前端的字段白名单(SecurityQuote 全字段中挑展示需要的,避免 unknown_* 噪音)。 +_QUOTE_FIELDS = [ + "market", + "code", + "price", + "pre_close", + "open", + "high", + "low", + "vol", + "cur_vol", + "amount", + "s_vol", + "b_vol", + "rise_speed", + "limit_up", + "limit_down", + "decimal_point", + "server_time", + "trading_status", +] + [ + f"{side}{i}" for side in ("bid", "ask") for i in range(1, 6) +] + [ + f"{side}_vol{i}" for side in ("bid", "ask") for i in range(1, 6) +] + +_SH_TZ = dt_timezone(timedelta(hours=8)) # Asia/Shanghai + + +def _is_trading_hours(now: datetime | None = None) -> bool: + """A股盘中(含集合竞价与收盘前后缓冲):沪时间 09:10-15:10,周一至周五。""" + t = now or datetime.now(_SH_TZ) + if t.weekday() >= 5: + return False + hm = t.hour * 60 + t.minute + return 9 * 60 + 10 <= hm <= 15 * 60 + 10 + + +class QuoteStreamer: + """共享轮询 + fan-out。由 FastAPI lifespan 启停(``app.state.quote_streamer``)。 + + Args: + quote_fetcher: async ``(stocks) -> pd.DataFrame``,通常为 + ``AsyncTdxClient.get_security_quotes`` 的偏函数。异常由本类兜底。 + watch_symbols: async ``() -> list[tuple[Market, str]]``,自选订阅集合 + (通常读 :class:`WatchlistStore`)。每次轮询前调用。 + trading_interval: 盘中轮询间隔(秒)。 + idle_interval: 盘外轮询间隔(秒)。 + """ + + def __init__( + self, + quote_fetcher: Callable[[list[tuple[Market, str]]], Awaitable[pd.DataFrame]], + watch_symbols: Callable[[], Awaitable[list[tuple[Market, str]]]], + *, + trading_interval: float = 8.0, + idle_interval: float = 60.0, + ) -> None: + self._fetch = quote_fetcher + self._watch_symbols = watch_symbols + self._trading_interval = trading_interval + self._idle_interval = idle_interval + self._queues: dict[int, asyncio.Queue[dict[str, Any]]] = {} + self._ids = itertools.count(1) + self._task: asyncio.Task[None] | None = None + self.last_snapshot: list[dict[str, Any]] = [] # 最近一次成功快照(调试/健康检查) + + # ── 订阅管理(SSE 端点调用) ────────────────────────────────────────── + + def subscribe(self) -> tuple[int, asyncio.Queue[dict[str, Any]]]: + """注册一个独立队列;返回 (id, queue)。""" + qid = next(self._ids) + q: asyncio.Queue[dict[str, Any]] = asyncio.Queue(maxsize=2) + self._queues[qid] = q + return qid, q + + def unsubscribe(self, qid: int) -> None: + self._queues.pop(qid, None) + + @property + def subscriber_count(self) -> int: + return len(self._queues) + + # ── 生命周期 ────────────────────────────────────────────────────────── + + def start(self) -> None: + if self._task is None or self._task.done(): + self._task = asyncio.get_running_loop().create_task(self._run()) + logger.info("QuoteStreamer started") + + 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 + logger.info("QuoteStreamer stopped") + + # ── 轮询主循环 ──────────────────────────────────────────────────────── + + async def _run(self) -> None: + while True: + try: + if not self._queues: + await asyncio.sleep(1.0) # 无人订阅,待命 + continue + + symbols = list(INDEX_SYMBOLS) + try: + symbols += await self._watch_symbols() + except Exception: + logger.warning("读取自选订阅集合失败", exc_info=True) + + quotes = await self._fetch_quotes(symbols) + if quotes: + self.last_snapshot = quotes + msg = { + "type": "quotes_updated", + "ts": datetime.now(_SH_TZ).isoformat(timespec="seconds"), + "count": len(quotes), + "quotes": quotes, + } + self._fan_out(msg) + + interval = self._trading_interval if _is_trading_hours() else self._idle_interval + await asyncio.sleep(interval) + except asyncio.CancelledError: + raise + except Exception: + # 单轮失败不能杀死循环(行情服务器闪断很常见) + logger.warning("QuoteStreamer 轮询异常", exc_info=True) + await asyncio.sleep(self._idle_interval) + + async def _fetch_quotes(self, symbols: list[tuple[Market, str]]) -> list[dict[str, Any]]: + """批量拉行情(80/批),转精简 dict 列表;失败返回空。""" + out: list[dict[str, Any]] = [] + for i in range(0, len(symbols), 80): + batch = symbols[i : i + 80] + try: + df = await self._fetch(batch) + except Exception: + logger.warning("行情拉取失败(%d 只)", len(batch), exc_info=True) + continue + out.extend(self._df_to_dicts(df)) + return out + + @staticmethod + def _df_to_dicts(df: pd.DataFrame) -> list[dict[str, Any]]: + """DataFrame → 前端 dict(白名单列 + market 枚举转字符串 + symbol 键)。""" + if df is None or df.empty: + return [] + rows: list[dict[str, Any]] = [] + for rec in df.to_dict(orient="records"): + market = rec.get("market") + market_str = _MARKET_NAMES.get(market, str(market or "")) + code = str(rec.get("code", "")) + d: dict[str, Any] = {} + for f in _QUOTE_FIELDS: + if f in rec and f not in ("market", "code"): + v = rec[f] + d[f] = None if v != v else v # NaN → None(JSON 合法) + d["market"] = market_str + d["code"] = code + d["symbol"] = f"{market_str}{code}" + rows.append(d) + return rows + + def _fan_out(self, msg: dict[str, Any]) -> None: + for q in list(self._queues.values()): + try: + q.put_nowait(msg) + except asyncio.QueueFull: + # 背压:丢最旧、保最新 + try: + q.get_nowait() + except asyncio.QueueEmpty: + pass + try: + q.put_nowait(msg) + except asyncio.QueueFull: + pass diff --git a/src/easy_tdx/web/routers/stream.py b/src/easy_tdx/web/routers/stream.py new file mode 100644 index 0000000..4425661 --- /dev/null +++ b/src/easy_tdx/web/routers/stream.py @@ -0,0 +1,62 @@ +"""实时行情 SSE 路由。 + +``GET /api/v1/stream/quotes`` → ``text/event-stream``: + +- 事件 ``data`` 载荷:``{"type": "quotes_updated", "ts", "count", "quotes": [...]}`` + (quotes 为指数 + 全部自选的快照,字段见 quote_streamer._QUOTE_FIELDS)。 +- 每 15s 一条 SSE 注释行(``: keepalive``)防中间层掐空闲连接。 +- 客户端断开由 ASGI cancel → generator finally 反注册队列。 + +零额外依赖:不用 sse-starlette,StreamingResponse + asyncio.Queue 足够。 +""" + +from __future__ import annotations + +import asyncio +import json +import logging + +from fastapi import APIRouter, Request +from fastapi.responses import StreamingResponse + +logger = logging.getLogger(__name__) + +router = APIRouter(tags=["stream"]) + +_KEEPALIVE_SECONDS = 15.0 + + +@router.get("/stream/quotes") +async def stream_quotes(request: Request) -> StreamingResponse: + """订阅实时行情推送(指数 + 自选,快照式全量推送)。""" + streamer = getattr(request.app.state, "quote_streamer", None) + if streamer is None: + from fastapi import HTTPException + + raise HTTPException(status_code=503, detail="行情推送服务未启动") + + qid, queue = streamer.subscribe() + + async def event_gen(): # type: ignore[no-untyped-def] + try: + # 首帧 hello:告诉前端连接可用 + 当前订阅规模 + hello = {"type": "hello", "subscribers": streamer.subscriber_count} + yield f"data: {json.dumps(hello, ensure_ascii=False)}\n\n" + while True: + try: + msg = await asyncio.wait_for(queue.get(), timeout=_KEEPALIVE_SECONDS) + yield f"data: {json.dumps(msg, ensure_ascii=False)}\n\n" + except asyncio.TimeoutError: + yield ": keepalive\n\n" + finally: + streamer.unsubscribe(qid) + + return StreamingResponse( + event_gen(), + media_type="text/event-stream", + headers={ + "Cache-Control": "no-cache", + "Connection": "keep-alive", + "X-Accel-Buffering": "no", # nginx 反代时不缓冲 + }, + ) diff --git a/src/easy_tdx/web/routers/watchlist.py b/src/easy_tdx/web/routers/watchlist.py new file mode 100644 index 0000000..8b41541 --- /dev/null +++ b/src/easy_tdx/web/routers/watchlist.py @@ -0,0 +1,49 @@ +"""自选股路由:加入 / 列出 / 移除(SQLite 持久化,无行情依赖)。""" + +from __future__ import annotations + +from fastapi import APIRouter, HTTPException, Query +from pydantic import BaseModel, Field + +from easy_tdx.web.watchlist_store import get_watchlist_store + +router = APIRouter(tags=["watchlist"]) + + +class WatchItemAdd(BaseModel): + """加入自选请求。name 由前端从行情数据带过来。""" + + market: str = Field(..., pattern=r"^(SZ|SH|BJ)$") + code: str = Field(..., min_length=6, max_length=6) + name: str = Field("", max_length=64) + group: str = Field("默认", max_length=32) + + +class WatchlistResponse(BaseModel): + items: list[dict] + count: int + + +@router.get("/watchlist", response_model=WatchlistResponse) +async def list_watchlist(group: str | None = Query(None, description="按分组过滤")) -> WatchlistResponse: + """列出全部自选(按加入顺序)。""" + items = get_watchlist_store().list_all(group=group) + return WatchlistResponse(items=[i.to_dict() for i in items], count=len(items)) + + +@router.post("/watchlist", response_model=dict) +async def add_watch_item(req: WatchItemAdd) -> dict: + """加入自选(幂等:重复加入仅刷新名称)。""" + item = get_watchlist_store().add(req.market, req.code, name=req.name, group=req.group) + return {"ok": True, "item": item.to_dict()} + + +@router.delete("/watchlist/{market}/{code}", response_model=dict) +async def remove_watch_item(market: str, code: str) -> dict: + """移除自选。""" + if market.upper() not in {"SZ", "SH", "BJ"}: + raise HTTPException(status_code=400, detail=f"非法市场: {market}") + removed = get_watchlist_store().remove(market, code) + if not removed: + raise HTTPException(status_code=404, detail=f"自选中不存在 {market}{code}") + return {"ok": True} diff --git a/src/easy_tdx/web/watchlist_store.py b/src/easy_tdx/web/watchlist_store.py new file mode 100644 index 0000000..e310aa8 --- /dev/null +++ b/src/easy_tdx/web/watchlist_store.py @@ -0,0 +1,171 @@ +"""自选股列表的 SQLite 持久化(Web UI"自选"页的数据后端)。 + +设计对齐 :mod:`easy_tdx.web.strategy_store`: + +- 单文件 SQLite,落在统一配置目录(``~/.easy_tdx/watchlist.db``, + 随 ``EASY_TDX_CONFIG_DIR`` 环境变量走)。 +- 短连接 + 写锁串行,跨线程安全(FastAPI 线程池内调用)。 +- ``(market, code)`` 唯一:重复加入同一只股票幂等(更新名称,不动排序)。 +- ``group_name`` 字段预留分组能力,v1 前端未使用,默认 ``"默认"``。 +- ``sort_order`` 由插入时 max+1 维持"新加的在最后",列出时按其升序。 +""" + +from __future__ import annotations + +import os +import sqlite3 +import threading +from dataclasses import dataclass +from datetime import datetime, timezone +from pathlib import Path + +__all__ = [ + "WatchItem", + "WatchlistStore", + "get_watchlist_store", +] + +_write_lock = threading.Lock() + + +def _config_dir() -> Path: + return Path(os.environ.get("EASY_TDX_CONFIG_DIR", str(Path.home() / ".easy_tdx"))) + + +def _default_db_path() -> Path: + return _config_dir() / "watchlist.db" + + +def _now_iso() -> str: + return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") + + +@dataclass +class WatchItem: + """一条自选记录。name 由前端在添加时填好(来自行情/选股接口)。""" + + market: str # "SH" | "SZ" | "BJ" + code: str # 6 位代码 + name: str = "" + group_name: str = "默认" + created_at: str = "" + sort_order: int = 0 + + @property + def symbol(self) -> str: + """前端统一标识:SH600000 形式。""" + return f"{self.market}{self.code}" + + def to_dict(self) -> dict: + return { + "market": self.market, + "code": self.code, + "symbol": self.symbol, + "name": self.name, + "group_name": self.group_name, + "created_at": self.created_at, + "sort_order": self.sort_order, + } + + +class WatchlistStore: + """自选股 SQLite 存储。单例由 :func:`get_watchlist_store` 提供。""" + + _SCHEMA = """ + CREATE TABLE IF NOT EXISTS watchlist ( + market TEXT NOT NULL, + code TEXT NOT NULL, + name TEXT NOT NULL DEFAULT '', + group_name TEXT NOT NULL DEFAULT '默认', + created_at TEXT NOT NULL DEFAULT '', + sort_order INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (market, code) + ); + """ + + def __init__(self, db_path: Path | None = None) -> None: + self.db_path = db_path or _default_db_path() + self.db_path.parent.mkdir(parents=True, exist_ok=True) + with self._connect() as conn: + conn.executescript(self._SCHEMA) + + def _connect(self) -> sqlite3.Connection: + conn = sqlite3.connect(self.db_path, check_same_thread=False) + conn.row_factory = sqlite3.Row + return conn + + def list_all(self, group: str | None = None) -> list[WatchItem]: + """列出全部自选(按 sort_order 升序),可按分组过滤。""" + with self._connect() as conn: + if group: + rows = conn.execute( + "SELECT * FROM watchlist WHERE group_name = ? ORDER BY sort_order", + (group,), + ).fetchall() + else: + rows = conn.execute("SELECT * FROM watchlist ORDER BY sort_order").fetchall() + return [ + WatchItem( + market=r["market"], + code=r["code"], + name=r["name"], + group_name=r["group_name"], + created_at=r["created_at"], + sort_order=r["sort_order"], + ) + for r in rows + ] + + def symbols(self) -> list[tuple[str, str]]: + """列出 (market, code) 元组——SSE 轮询器订阅集合用。""" + return [(i.market, i.code) for i in self.list_all()] + + def add(self, market: str, code: str, name: str = "", group: str = "默认") -> WatchItem: + """加入自选;已存在时幂等返回(仅刷新名称)。""" + market = market.upper() + with _write_lock, self._connect() as conn: + row = conn.execute( + "SELECT * FROM watchlist WHERE market = ? AND code = ?", (market, code) + ).fetchone() + if row is not None: + if name and name != row["name"]: + conn.execute( + "UPDATE watchlist SET name = ? WHERE market = ? AND code = ?", + (name, market, code), + ) + return WatchItem( + market=row["market"], + code=row["code"], + name=name or row["name"], + group_name=row["group_name"], + created_at=row["created_at"], + sort_order=row["sort_order"], + ) + next_order = conn.execute("SELECT COALESCE(MAX(sort_order), 0) + 1 FROM watchlist").fetchone()[0] + conn.execute( + "INSERT INTO watchlist (market, code, name, group_name, created_at, sort_order)" + " VALUES (?, ?, ?, ?, ?, ?)", + (market, code, name, group, _now_iso(), next_order), + ) + return WatchItem(market=market, code=code, name=name, group_name=group, created_at=_now_iso(), sort_order=next_order) + + def remove(self, market: str, code: str) -> bool: + """移除自选;返回是否确实删除了一条。""" + market = market.upper() + with _write_lock, self._connect() as conn: + cur = conn.execute("DELETE FROM watchlist WHERE market = ? AND code = ?", (market, code)) + return cur.rowcount > 0 + + +_store: WatchlistStore | None = None +_store_lock = threading.Lock() + + +def get_watchlist_store() -> WatchlistStore: + """全局单例(首次调用惰性建库)。""" + global _store + if _store is None: + with _store_lock: + if _store is None: + _store = WatchlistStore() + return _store diff --git a/tests/unit/test_commands_offline.py b/tests/unit/test_commands_offline.py index 9e61525..e55f8bb 100644 --- a/tests/unit/test_commands_offline.py +++ b/tests/unit/test_commands_offline.py @@ -233,23 +233,29 @@ def _build_quote_body(market: int, code: str, price_raw: int) -> bytes: def test_security_quotes_decimal_point_classification(): - """Issue #8:价格小数位按 market+code 代码段推断。 + """Issue #8 + 看板指数修复:价格小数位按 market+code 代码段推断。 同一代码不同市场含义不同:SZ 000001=平安银行(股票,2位), - SH 000001=上证指数(3位),故必须结合市场判断。 + SH 000001=上证指数,故必须结合市场判断。 + 大盘指数点位恒两位小数(2026-09-01 实测:科创50 1647.53、沪深300 4611.44、 + 上证 3979.89),报价原始单位为「分」;曾按 3 位(厘)解析导致看板指数缩小 10 倍。 """ from easy_tdx.commands.security_quotes import _price_decimal_digits from easy_tdx.models.enums import Market - # ETF / 基金 / 可转债 / 国债 / 指数 -> 3 位(厘) + # ETF / 基金 / 可转债 / 国债 / 880 统计指数 -> 3 位(厘) assert _price_decimal_digits(Market.SZ, "159922") == 3 # 深 ETF assert _price_decimal_digits(Market.SZ, "161725") == 3 # 深 LOF 基金 assert _price_decimal_digits(Market.SZ, "128095") == 3 # 深 可转债 assert _price_decimal_digits(Market.SZ, "111002") == 3 # 深 国债 assert _price_decimal_digits(Market.SH, "510300") == 3 # 沪 ETF assert _price_decimal_digits(Market.SH, "511990") == 3 # 沪 货币基金 - assert _price_decimal_digits(Market.SH, "000001") == 3 # 上证指数 - assert _price_decimal_digits(Market.SH, "000300") == 3 # 沪深 300 指数 + assert _price_decimal_digits(Market.SH, "880005") == 3 # 统计指数(market_stat 依赖 3 位语义) + + # 大盘指数 -> 2 位(分,点位恒两位小数) + assert _price_decimal_digits(Market.SH, "000001") == 2 # 上证指数 + assert _price_decimal_digits(Market.SH, "000300") == 2 # 沪深 300 + assert _price_decimal_digits(Market.SZ, "399001") == 2 # 深证成指 # 股票 -> 2 位(分) assert _price_decimal_digits(Market.SZ, "000001") == 2 # 深主板(平安银行) @@ -290,16 +296,20 @@ def test_security_quotes_stock_price_unchanged(): assert abs(q.price - 9.89) < 1e-9 -def test_security_quotes_index_price_3_digits(): - """Issue #8:上证指数 SH000001 现价 3123.456 → 按 3 位小数解析。""" +def test_security_quotes_index_price_2_digits(): + """看板指数修复:上证指数 SH000001 现价 3979.89 → 按 2 位小数(分)解析。 + + 实测(2026-09-01):SH 000/880 系列 decimal_point 报 3,但 000 系大盘指数 + 的原始单位是分。3123456 → 31234.56。 + """ from easy_tdx.commands.security_quotes import GetSecurityQuotesCmd from easy_tdx.models.enums import Market body = _build_quote_body(int(Market.SH), "000001", 3123456) q = GetSecurityQuotesCmd([(Market.SH, "000001")]).parse_response(body)[0] - assert q.decimal_point == 3 - assert abs(q.price - 3123.456) < 1e-6 + assert q.decimal_point == 2 + assert abs(q.price - 31234.56) < 1e-6 # --------------------------------------------------------------------------- diff --git a/tests/unit/test_quote_decimal_digits.py b/tests/unit/test_quote_decimal_digits.py new file mode 100644 index 0000000..eb61d32 --- /dev/null +++ b/tests/unit/test_quote_decimal_digits.py @@ -0,0 +1,41 @@ +"""报价小数位推断(Issue #8 + 看板指数 ×0.1 修复)的语义锁定测试。""" + +from __future__ import annotations + +import pytest + +from easy_tdx.commands.security_quotes import _price_decimal_digits +from easy_tdx.models.enums import Market + + +@pytest.mark.parametrize( + ("market", "code", "expected", "why"), + [ + # ── 大盘指数:点位恒两位小数(科创50 1647.53),原始单位是「分」 ── + (Market.SH, "000001", 2, "上证指数"), + (Market.SH, "000300", 2, "沪深300(实测 4611.44,按 3 位曾缩成 461.144)"), + (Market.SH, "000688", 2, "科创50(实测 1647.53,按 3 位曾缩成 164.753)"), + (Market.SH, "000905", 2, "中证500"), + (Market.SH, "000016", 2, "上证50"), + (Market.SZ, "399001", 2, "深证成指(深市指数走默认分支,一直正确)"), + (Market.SZ, "399006", 2, "创业板指"), + # ── 统计指数:字段是计数语义(price×10 还原家数),必须保持 3 位 ── + (Market.SH, "880005", 3, "全市场行情统计(market_stat 依赖)"), + (Market.SH, "880006", 3, "涨跌停统计"), + # ── 板块指数:与大盘指数同口径,两位小数 ── + (Market.SH, "881106", 2, "行业板块指数(种植业,实测 1039.93)"), + (Market.SH, "885418", 2, "概念板块指数"), + # ── 基金/ETF:真实三位小数(Issue #8)── + (Market.SH, "510300", 3, "沪深300ETF(价格如 4.611)"), + (Market.SH, "588000", 3, "科创50ETF"), + (Market.SZ, "159915", 3, "创业板ETF"), + # ── 股票:两位 ── + (Market.SH, "600519", 2, "主板"), + (Market.SH, "688981", 2, "科创板"), + (Market.SZ, "000001", 2, "深市 000 开头是股票(平安银行),与 SH000001 同码不同义"), + (Market.SZ, "300750", 2, "创业板"), + (Market.BJ, "920002", 2, "北交所"), + ], +) +def test_price_decimal_digits(market: Market, code: str, expected: int, why: str) -> None: + assert _price_decimal_digits(market, code) == expected, why diff --git a/tests/unit/test_watchlist_and_streamer.py b/tests/unit/test_watchlist_and_streamer.py new file mode 100644 index 0000000..f458d89 --- /dev/null +++ b/tests/unit/test_watchlist_and_streamer.py @@ -0,0 +1,155 @@ +"""WatchlistStore(SQLite CRUD)与 QuoteStreamer(fan-out/背压)单元测试。""" + +from __future__ import annotations + +import asyncio +from pathlib import Path + +import pandas as pd +import pytest + +from easy_tdx.models.enums import Market +from easy_tdx.web.quote_streamer import INDEX_SYMBOLS, QuoteStreamer, _is_trading_hours +from easy_tdx.web.watchlist_store import WatchlistStore + + +# ── WatchlistStore ────────────────────────────────────────────────────────── + + +@pytest.fixture() +def store(tmp_path: Path) -> WatchlistStore: + return WatchlistStore(db_path=tmp_path / "watchlist.db") + + +def test_add_list_remove_roundtrip(store: WatchlistStore) -> None: + assert store.list_all() == [] + item = store.add("SH", "600000", name="浦发银行") + assert item.symbol == "SH600000" + store.add("SZ", "000001", name="平安银行") + store.add("BJ", "920002", name="万达轴承") + + items = store.list_all() + assert [i.symbol for i in items] == ["SH600000", "SZ000001", "BJ920002"] + # 按加入顺序排列 + assert [i.sort_order for i in items] == [1, 2, 3] + + +def test_add_is_idempotent(store: WatchlistStore) -> None: + store.add("SH", "600000", name="浦发银行") + store.add("sh", "600000", name="浦发银行(更名)") # 小写市场码归一 + items = store.list_all() + assert len(items) == 1 + assert items[0].name == "浦发银行(更名)" + assert items[0].sort_order == 1 # 幂等:不改变排序 + + +def test_remove_missing_returns_false(store: WatchlistStore) -> None: + assert store.remove("SZ", "399006") is False + store.add("SZ", "399006", name="创业板指") + assert store.remove("SZ", "399006") is True + assert store.remove("SZ", "399006") is False + + +def test_symbols_for_streamer(store: WatchlistStore) -> None: + store.add("SH", "600000", name="浦发银行") + assert store.symbols() == [("SH", "600000")] + + +# ── QuoteStreamer ─────────────────────────────────────────────────────────── + + +def _fake_df(symbols: list[tuple[Market, str]]) -> pd.DataFrame: + rows = [] + for mkt, code in symbols: + rows.append( + { + "market": mkt, + "code": code, + "price": 10.5, + "pre_close": 10.0, + "open": 10.1, + "high": 10.8, + "low": 9.9, + "vol": 12345.0, + "amount": 1_234_500.0, + "bid1": 10.49, + "bid_vol1": 100, + "ask1": 10.51, + "ask_vol1": 120, + "bid2": 10.48, + "bid_vol2": 90, + "unknown_5": 0, # 应被白名单过滤 + } + ) + return pd.DataFrame(rows) + + +def _make_streamer() -> tuple[QuoteStreamer, list[list[tuple[Market, str]]]]: + calls: list[list[tuple[Market, str]]] = [] + + async def fetch(symbols: list[tuple[Market, str]]) -> pd.DataFrame: + calls.append(symbols) + return _fake_df(symbols) + + async def watch() -> list[tuple[Market, str]]: + return [(Market.SZ, "000001")] + + return QuoteStreamer(fetch, watch, trading_interval=0.01, idle_interval=0.01), calls + + +def test_streamer_fanout_and_backpressure() -> None: + streamer, calls = _make_streamer() + qid1, q1 = streamer.subscribe() + qid2, q2 = streamer.subscribe() + + async def run_once() -> None: + streamer.start() + await asyncio.sleep(0.05) # 至少完成一轮轮询 + await streamer.stop() + + asyncio.run(run_once()) + + assert calls, "应至少发起一次行情拉取" + # 订阅集合 = 指数 + 自选 + assert (Market.SZ, "000001") in calls[0] + assert set(INDEX_SYMBOLS).issubset(set(calls[0])) + + for q in (q1, q2): + msg = q.get_nowait() + assert msg["type"] == "quotes_updated" + assert msg["count"] == len(calls[0]) + rec = msg["quotes"][0] + assert rec["market"] in {"SH", "SZ", "BJ"} + assert rec["symbol"] + assert "unknown_5" not in rec # 白名单生效 + assert rec["price"] == 10.5 + # 五档字段(bid_vol1 语义,非 bid1_vol)必须完整透传(Issue:盘口无数据) + assert rec["bid1"] == 10.49 + assert rec["bid_vol1"] == 100 + assert rec["ask1"] == 10.51 + assert rec["ask_vol1"] == 120 + assert rec["bid_vol2"] == 90 + + streamer.unsubscribe(qid1) + streamer.unsubscribe(qid2) + assert streamer.subscriber_count == 0 + + +def test_streamer_backpressure_drops_oldest() -> None: + """队列满(maxsize=2)时丢最旧保最新——第 3 条消息应顶掉第 1 条。""" + streamer, _ = _make_streamer() + qid, q = streamer.subscribe() + for i in range(3): + streamer._fan_out({"type": "quotes_updated", "seq": i}) + seqs = [q.get_nowait()["seq"] for _ in range(2)] + assert seqs == [1, 2] + streamer.unsubscribe(qid) + + +def test_is_trading_hours() -> None: + from datetime import datetime, timedelta, timezone as dt_timezone + + tz = dt_timezone(timedelta(hours=8)) + assert _is_trading_hours(datetime(2026, 9, 1, 10, 0, tzinfo=tz)) is True # 周二盘中 + assert _is_trading_hours(datetime(2026, 9, 1, 3, 0, tzinfo=tz)) is False # 凌晨 + assert _is_trading_hours(datetime(2026, 9, 5, 10, 0, tzinfo=tz)) is False # 周六 diff --git a/tests/unit/test_web_api.py b/tests/unit/test_web_api.py index 78f173e..393a2b2 100644 --- a/tests/unit/test_web_api.py +++ b/tests/unit/test_web_api.py @@ -79,11 +79,14 @@ def test_create_app_returns_fastapi_instance(): assert app.title == "easy-tdx API" # Check routers are mounted - routes = [r.path for r in app.routes] + # FastAPI 0.141+ 的 app.routes 含 _IncludedRouter(无 path 属性),改用 + # OpenAPI schema 验证(WebSocket 不进 schema,由 test_full_app_routes_registered 覆盖) + routes = list(app.openapi()["paths"].keys()) assert any("/api/v1/security" in r for r in routes) assert any("/api/v1/bars" in r for r in routes) assert any("/api/v1/chanlun" in r for r in routes) - assert any("/ws/realtime" in r for r in routes) + assert any("/api/v1/watchlist" in r for r in routes) + assert any("/api/v1/stream/quotes" in r for r in routes) def test_deps_get_client_type(): @@ -503,7 +506,10 @@ def test_full_app_routes_registered(): from easy_tdx.web import create_app app = create_app() - all_paths = [r.path for r in app.routes] + # FastAPI 0.141+ 的 include_router 是 _IncludedRouter 延迟对象,app.routes + # 不再平铺子路由;用 OpenAPI schema 验证注册结果(面向行为而非内部结构)。 + # WebSocket 路由不进 OpenAPI,单独用 _IncludedRouter 展开验证。 + all_paths = list(app.openapi()["paths"].keys()) expected_prefixes = [ "/api/v1/security", "/api/v1/bars", @@ -512,12 +518,28 @@ def test_full_app_routes_registered(): "/api/v1/chanlun", "/api/v1/announcements", "/api/v1/sina/financial-report", - "/ws/realtime", + "/api/v1/watchlist", + "/api/v1/stream/quotes", ] for prefix in expected_prefixes: matched = any(prefix in p for p in all_paths) assert matched, f"Expected route with prefix '{prefix}' not found in {all_paths}" + # WebSocket 路由验证:经 _IncludedRouter.original_router 展开找 ws 路径 + ws_paths: list[str] = [] + + def _collect(routes: list) -> None: + for r in routes: + path = getattr(r, "path", None) + if path: + ws_paths.append(path) + orig = getattr(r, "original_router", None) + if orig is not None: + _collect(getattr(orig, "routes", [])) + + _collect(app.routes) + assert any("/ws/realtime" in p for p in ws_paths), f"WS route missing in {ws_paths}" + def test_openapi_schema_generated(): """OpenAPI schema should be auto-generated and contain key paths.""" diff --git a/web-ui/src/App.vue b/web-ui/src/App.vue index 64cd7de..ba85126 100644 --- a/web-ui/src/App.vue +++ b/web-ui/src/App.vue @@ -1,21 +1,48 @@