看股功能更新

backend/app/data/market_overview.py

@@ -0,0 +1,161 @@
+"""大盘行情总览（主页展示）。
+
+- A 股指数走 pro.index_daily，全球指数走 pro.index_global（均为收盘口径，晚间更新；
+  本 token 无 rt_idx_k 实时权限，故展示「最近交易日收盘」并标注日期）。
+- 两市统计走 pro.daily_info：沪市取 SH_A、深市取 SZ_MARKET 汇总出
+  总市值 / 流通市值 / 成交额（单位亿元，接口原生口径）。
+- 整包结果写 Redis 缓存（TTL 可配）；单个指数拉取失败只是跳过（errors 里注明），
+  全部失败才抛 MarketOverviewError —— 主页行情是锦上添花，不拖垮整页。
+"""
+from __future__ import annotations
+
+import asyncio
+import math
+from datetime import date, datetime, timedelta
+
+import pandas as pd
+
+from .. import cache
+from ..config import settings
+
+# (代码, 名称, 地区) —— 展示顺序即列表顺序
+MARKET_INDEXES: list[tuple[str, str, str]] = [
+    ("000001.SH", "上证指数", "cn"),
+    ("399001.SZ", "深证成指", "cn"),
+    ("399006.SZ", "创业板指", "cn"),
+    ("000688.SH", "科创50", "cn"),
+    ("HSI", "恒生指数", "hk"),
+    ("HKTECH", "恒生科技", "hk"),
+    ("DJI", "道琼斯", "us"),
+    ("IXIC", "纳斯达克", "us"),
+    ("SPX", "标普500", "us"),
+]
+
+_SPARK_DAYS = 45        # 迷你走势取最近 45 个交易日收盘
+_HISTORY_DAYS = 150     # 日历日窗口（约 100 个交易日，够取 spark）
+_CALL_INTERVAL = 0.12   # 顺序调用间隔（秒），对 tushare 控频
+_CACHE_KEY = "market_overview:v1"
+
+
+class MarketOverviewError(RuntimeError):
+    """所有指数都拉不到（token/网络故障）——接口层转 503。"""
+
+
+def _f(v) -> float | None:
+    """pandas 值 -> float；NaN/None -> None（否则 JSON 里会出现 NaN）。"""
+    if v is None:
+        return None
+    try:
+        f = float(v)
+    except (TypeError, ValueError):
+        return None
+    return None if math.isnan(f) else f
+
+
+def _d(v) -> date | None:
+    return datetime.strptime(str(v), "%Y%m%d").date() if v else None
+
+
+def _get_pro():
+    if not settings.tushare_token:
+        raise MarketOverviewError("未配置 TUSHARE_TOKEN，无法获取大盘行情（backend/.env）")
+    import tushare as ts
+
+    ts.set_token(settings.tushare_token)
+    return ts.pro_api()
+
+
+def _fetch_index_sync(pro, ts_code: str) -> pd.DataFrame:
+    start = (datetime.now() - timedelta(days=_HISTORY_DAYS)).strftime("%Y%m%d")
+    if "." in ts_code:  # A 股指数（000001.SH 形式）
+        return pro.index_daily(ts_code=ts_code, start_date=start)
+    return pro.index_global(ts_code=ts_code, start_date=start)
+
+
+def _quote_from_df(df: pd.DataFrame) -> dict | None:
+    """DataFrame -> {close, change, pct_chg, trade_date, spark, spark_dates}（旧 -> 新）。"""
+    if df is None or df.empty:
+        return None
+    df = df.sort_values("trade_date")
+    tail = df.tail(_SPARK_DAYS)
+    last = df.iloc[-1]
+    return {
+        "close": _f(last["close"]),
+        "change": _f(last.get("change")),
+        "pct_chg": _f(last.get("pct_chg")),
+        "trade_date": _d(last["trade_date"]),
+        "spark": [round(float(c), 4) for c in tail["close"]],
+        "spark_dates": [str(d) for d in tail["trade_date"]],
+    }
+
+
+def _fetch_stats_sync(pro) -> dict | None:
+    """两市市值/成交统计：沪 SH_A + 深 SZ_MARKET（同一天口径相加，亿元）。"""
+    start = (datetime.now() - timedelta(days=14)).strftime("%Y%m%d")
+    sh = pro.daily_info(exchange="SH", start_date=start)
+    sz = pro.daily_info(exchange="SZ", start_date=start)
+    if sh is None or sh.empty or sz is None or sz.empty:
+        return None
+
+    def _board(df: pd.DataFrame, code: str):
+        sub = df[df["ts_code"] == code].sort_values("trade_date")  # 接口不保证有序
+        return sub.iloc[-1] if not sub.empty else None
+
+    sh_a, sz_m = _board(sh, "SH_A"), _board(sz, "SZ_MARKET")
+    if sh_a is None or sz_m is None:
+        return None
+    # 两边各自取最新，日期不一致时以较旧一天为准凑齐口径（罕见，通常同日）
+    d = min(_d(sh_a["trade_date"]), _d(sz_m["trade_date"]))
+
+    def _sum(col: str) -> float | None:
+        a, b = _f(sh_a.get(col)), _f(sz_m.get(col))
+        return None if a is None or b is None else round(a + b, 2)
+
+    return {
+        "trade_date": d,
+        "total_mv": _sum("total_mv"),
+        "float_mv": _sum("float_mv"),
+        "amount": _sum("amount"),
+        "turnover": _f(sh_a.get("tr")),  # 换手率仅沪市有，展示口径注明沪市
+    }
+
+
+async def fetch_overview() -> dict:
+    """聚合全部指数 + 两市统计（Redis 缓存整包，TTL 内直接回）。"""
+    cached = await cache.cache_get(_CACHE_KEY)
+    if cached is not None:
+        return cached
+
+    pro = await asyncio.to_thread(_get_pro)
+
+    indexes: list[dict] = []
+    errors: list[str] = []
+    for ts_code, name, region in MARKET_INDEXES:
+        try:
+            df = await asyncio.to_thread(_fetch_index_sync, pro, ts_code)
+            q = _quote_from_df(df)
+            if q is None:
+                raise MarketOverviewError("无数据")
+            indexes.append({"code": ts_code, "name": name, "region": region, **q})
+        except Exception as e:  # noqa: BLE001 —— 单个指数失败不拖垮整包
+            errors.append(f"{name}: {str(e)[:60]}")
+        await asyncio.sleep(_CALL_INTERVAL)
+
+    if not indexes:
+        raise MarketOverviewError("大盘行情全部拉取失败: " + "; ".join(errors)[:200])
+
+    stats: dict | None = None
+    try:
+        await asyncio.sleep(_CALL_INTERVAL)
+        stats = await asyncio.to_thread(_fetch_stats_sync, pro)
+    except Exception as e:  # noqa: BLE001 —— 统计缺失时指数照常展示
+        errors.append(f"两市统计: {str(e)[:60]}")
+
+    payload = {
+        "updated_at": datetime.now(),
+        "indexes": indexes,
+        "stats": stats,
+        "errors": errors,
+    }
+    await cache.cache_set(_CACHE_KEY, payload, ttl=settings.market_overview_ttl)
+    return payload