356 lines
15 KiB
Python
356 lines
15 KiB
Python
"""大盘行情总览（主页展示）——两层结构。
|
||
|
||
- live 层：腾讯免费实时行情（qt.gtimg.cn，一次 GET 拿全部指数现价/涨跌幅），
|
||
进程内缓存 30s（成功）/ 15s 负缓存（失败，防止接口抖动持续拖慢请求）。
|
||
- EOD 层：tushare 收盘数据 —— A 股指数 pro.index_daily、全球指数 pro.index_global
|
||
（45 日 spark 走势）+ 两市统计/成交额历史 pro.daily_info。收盘数据一天一变，
|
||
进程内新鲜期 4h + Redis 兜底 24h，过期走 SWR（先返旧值，后台刷新，永不阻塞用户）。
|
||
- merge：实时价覆盖 close/change/pct_chg（realtime=True），拿不到实时值的指数回退
|
||
收盘口径；今日为交易日且 EOD 尚未含今日时，用腾讯全市口径成交额追加盘中 bar。
|
||
|
||
口径说明（daily_info 板块行，实测 2026-09）：
|
||
- 沪市 SH_MARKET = 主板A + 科创板(SH_STAR) + B股，不含基金(SH_FUND)；
|
||
旧口径 SH_A 漏科创板（日均 ~2500 亿），是成交额偏小的根因。
|
||
- 深市 SZ_MARKET = 主板 + 创业板，全部为股票。
|
||
- 任何单个来源失败只是跳过（errors 里注明），全部失败才抛 MarketOverviewError。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import time
|
||
from datetime import date, datetime, timedelta
|
||
|
||
import httpx
|
||
import pandas as pd
|
||
|
||
from .. import cache
|
||
from ..config import settings
|
||
from .sync_utils import d8_iso, f_clean
|
||
|
||
# (tushare代码, 名称, 地区, 腾讯符号) —— 展示顺序即列表顺序
|
||
# 首页聚焦中美（港股/国际指数在 /indexes 国际指数页）；标普500 腾讯符号是 s_usINX（不是 s_usSPX）
|
||
MARKET_INDEXES: list[tuple[str, str, str, str]] = [
|
||
("000001.SH", "上证指数", "cn", "s_sh000001"),
|
||
("399001.SZ", "深证成指", "cn", "s_sz399001"),
|
||
("399006.SZ", "创业板指", "cn", "s_sz399006"),
|
||
("000688.SH", "科创50", "cn", "s_sh000688"),
|
||
("DJI", "道琼斯", "us", "s_usDJI"),
|
||
("IXIC", "纳斯达克", "us", "s_usIXIC"),
|
||
("SPX", "标普500", "us", "s_usINX"),
|
||
]
|
||
|
||
# 深证综指：不展示，仅取其 f[7]（深市全市成交额，万元）
|
||
_TENCENT_SZ_TOTAL = "s_sz399106"
|
||
_TENCENT_MAP = {ts_code: sym for ts_code, _, _, sym in MARKET_INDEXES}
|
||
_TENCENT_URL = "http://qt.gtimg.cn/q=" + ",".join([*_TENCENT_MAP.values(), _TENCENT_SZ_TOTAL])
|
||
|
||
_SPARK_DAYS = 45 # 迷你走势取最近 45 个交易日收盘
|
||
_HISTORY_DAYS = 150 # 日历日窗口（约 100 个交易日，够取 spark）
|
||
_CALL_INTERVAL = 0.12 # 顺序调用间隔（秒），对 tushare 控频
|
||
_EOD_KEY = "market_overview:eod:v3" # v3：首页列表移除港股（移入国际指数页），与 v2 隔离
|
||
_AMOUNT_HIST_CAL_DAYS = 190 # 成交额历史的日历日窗口（≈128 交易日）
|
||
_AMOUNT_HIST_BARS = 120 # 输出的柱数（取尾部）
|
||
_LIVE_FAIL_TTL = 15.0 # live 层失败负缓存（秒）
|
||
|
||
|
||
class MarketOverviewError(RuntimeError):
|
||
"""所有指数都拉不到（token/网络故障）——接口层转 503。"""
|
||
|
||
|
||
def _get_pro():
|
||
if not settings.tushare_token:
|
||
raise MarketOverviewError("未配置 TUSHARE_TOKEN，无法获取大盘行情（backend/.env）")
|
||
# 走统一入口：15000 积分档 token 只认 quicksync 镜像（直连 api.tushare.pro 会 40101）
|
||
from .tushare_provider import get_pro
|
||
|
||
return get_pro()
|
||
|
||
|
||
# ======================= live 层：腾讯实时行情 =======================
|
||
|
||
_live: dict = {"at": 0.0, "quotes": None} # 进程内缓存（monotonic 时钟）
|
||
_live_error: str | None = None # 最近一次实时拉取失败的原因（成功后清空）
|
||
|
||
|
||
async def _fetch_live_http() -> dict[str, dict]:
|
||
"""一次 GET 拿全部符号。响应 GBK，行如 v_s_sh000001="1~上证指数~000001~3930.12~-11.97~-0.30~537286161~93825519~~";
|
||
|
||
字段序：f[1]名称 / f[3]现价 / f[4]涨跌 / f[5]涨跌% / f[7]成交额。
|
||
单位陷阱：仅 s_sh000001 与 s_sz399106 的 f[7] 是「万元、全市口径」，可算两市成交额；
|
||
港股行的 f[7] 是手数、美股行非人民币金额，s_sz399001（深证成指）是成分股口径——都不能用。
|
||
"""
|
||
async with httpx.AsyncClient(timeout=settings.tencent_quote_timeout) as client:
|
||
resp = await client.get(_TENCENT_URL)
|
||
resp.raise_for_status()
|
||
text = resp.content.decode("gbk", errors="replace") # 响应头 charset 不可靠，显式解码
|
||
|
||
quotes: dict[str, dict] = {}
|
||
for line in text.splitlines():
|
||
if "=" not in line:
|
||
continue
|
||
head, _, body = line.partition("=")
|
||
sym = head.strip().removeprefix("v_")
|
||
fields = body.strip().strip(';"').split("~")
|
||
if not sym or len(fields) < 8:
|
||
continue
|
||
|
||
def _num(i: int) -> float | None:
|
||
try:
|
||
return float(fields[i])
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
quotes[sym] = {
|
||
"name": fields[1],
|
||
"price": _num(3),
|
||
"change": _num(4),
|
||
"pct": _num(5),
|
||
"amount_wan": _num(7),
|
||
}
|
||
if not quotes:
|
||
raise RuntimeError("腾讯行情响应为空或无法解析")
|
||
return quotes
|
||
|
||
|
||
async def _fetch_live() -> dict[str, dict] | None:
|
||
"""实时行情（进程内缓存）；任何失败返回 None，上层降级 EOD。"""
|
||
global _live_error
|
||
now = time.monotonic()
|
||
age = now - _live["at"]
|
||
if _live["quotes"] is not None and age < settings.market_live_ttl:
|
||
return _live["quotes"]
|
||
if _live["quotes"] is None and _live["at"] > 0 and age < _LIVE_FAIL_TTL:
|
||
return None # 负缓存：刚失败过，短时间内不再打腾讯
|
||
try:
|
||
quotes = await _fetch_live_http()
|
||
except Exception as e: # noqa: BLE001 —— 实时层是锦上添花，失败不拖垮整包
|
||
_live.update(at=now, quotes=None)
|
||
_live_error = f"实时行情: {str(e)[:60]}"
|
||
return None
|
||
_live.update(at=now, quotes=quotes)
|
||
_live_error = None
|
||
return quotes
|
||
|
||
|
||
# ======================= EOD 层：tushare 收盘数据 =======================
|
||
|
||
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_clean(last["close"]),
|
||
"change": f_clean(last.get("change")),
|
||
"pct_chg": f_clean(last.get("pct_chg")),
|
||
"trade_date": d8_iso(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_MARKET（主板A+科创+B，不含基金）+ 深 SZ_MARKET（全部股票）。
|
||
|
||
口径与本地 candles 全市场 sum(amount) 吻合（candles 另含北交所，约 +70 亿）。
|
||
"""
|
||
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_m, sz_m = _board(sh, "SH_MARKET"), _board(sz, "SZ_MARKET")
|
||
if sh_m is None or sz_m is None:
|
||
return None
|
||
# 两边各自取最新，日期不一致时以较旧一天为准凑齐口径（罕见，通常同日）
|
||
d = min(d8_iso(sh_m["trade_date"]), d8_iso(sz_m["trade_date"]))
|
||
|
||
def _sum(col: str) -> float | None:
|
||
a, b = f_clean(sh_m.get(col)), f_clean(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_clean(sh_m.get("tr")), # 换手率仅沪市有，展示口径注明沪市
|
||
}
|
||
|
||
|
||
def _fetch_amount_history_sync(pro) -> list[dict]:
|
||
"""两市成交额历史：daily_info 范围查询一次拉多日（SH 6 个月实测 0.09s），
|
||
沪 SH_MARKET + 深 SZ_MARKET 按日对齐相加（接口原生亿元），升序取尾部 120 根。"""
|
||
start = (datetime.now() - timedelta(days=_AMOUNT_HIST_CAL_DAYS)).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 []
|
||
sh_m = sh[sh["ts_code"] == "SH_MARKET"].set_index("trade_date")["amount"]
|
||
sz_m = sz[sz["ts_code"] == "SZ_MARKET"].set_index("trade_date")["amount"]
|
||
common = sh_m.index.intersection(sz_m.index) # 内连接：两市都有数据的交易日
|
||
if len(common) == 0:
|
||
return []
|
||
total = (sh_m[common] + sz_m[common]).sort_index()
|
||
return [{"date": d8_iso(d), "amount": round(float(v), 2)} for d, v in total.tail(_AMOUNT_HIST_BARS).items()]
|
||
|
||
|
||
# ---- EOD 的 SWR（stale-while-revalidate）：新鲜期内直返；过期先返旧值后台刷新 ----
|
||
|
||
_eod_state: dict = {"payload": None} # 进程内新鲜/陈旧兜底（payload 自带 fetched_ts 墙钟）
|
||
_eod_refreshing = False # 后台刷新防重入标志
|
||
_eod_refresh_error: str | None = None # 最近一次后台刷新失败的原因
|
||
_bg_tasks: set[asyncio.Task] = set() # 持引用防 GC
|
||
|
||
|
||
async def _refresh_eod() -> dict:
|
||
"""拉全量 EOD（9 指数 + 统计 + 成交额历史，顺序控频），写进程内 state + Redis。"""
|
||
pro = await asyncio.to_thread(_get_pro)
|
||
|
||
indexes: list[dict] = []
|
||
errors: list[str] = []
|
||
for ts_code, name, region, _sym 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]}")
|
||
|
||
amount_history: list[dict] = []
|
||
try:
|
||
await asyncio.sleep(_CALL_INTERVAL)
|
||
amount_history = await asyncio.to_thread(_fetch_amount_history_sync, pro)
|
||
except Exception as e: # noqa: BLE001 —— 历史图缺数据时其余照常
|
||
errors.append(f"成交额历史: {str(e)[:60]}")
|
||
|
||
payload = {
|
||
"fetched_at": datetime.now().isoformat(),
|
||
"fetched_ts": time.time(), # 墙钟（epoch float）：跨进程（Redis）判新鲜度用
|
||
"indexes": indexes,
|
||
"stats": stats,
|
||
"amount_history": amount_history,
|
||
"errors": errors,
|
||
}
|
||
_eod_state["payload"] = payload
|
||
await cache.cache_set(_EOD_KEY, payload, ttl=settings.market_eod_redis_ttl)
|
||
return payload
|
||
|
||
|
||
async def _refresh_eod_wrapped() -> None:
|
||
"""后台刷新的主体：失败静默保留旧值并记录原因（下次请求并入 errors 便于排查）。"""
|
||
global _eod_refresh_error, _eod_refreshing
|
||
try:
|
||
await _refresh_eod()
|
||
_eod_refresh_error = None
|
||
except Exception as e: # noqa: BLE001
|
||
_eod_refresh_error = f"EOD后台刷新: {str(e)[:60]}"
|
||
finally:
|
||
_eod_refreshing = False
|
||
|
||
|
||
def _spawn_eod_refresh() -> None:
|
||
global _eod_refreshing
|
||
if _eod_refreshing:
|
||
return
|
||
_eod_refreshing = True
|
||
task = asyncio.create_task(_refresh_eod_wrapped())
|
||
_bg_tasks.add(task)
|
||
task.add_done_callback(_bg_tasks.discard)
|
||
|
||
|
||
async def _get_eod() -> dict:
|
||
"""读 EOD：内存新鲜直返（0 RTT）→ Redis 回填 → 有旧值先返 + SWR 后台刷新 → 真冷启动同步拉。"""
|
||
p = _eod_state["payload"]
|
||
if p is not None and time.time() - p["fetched_ts"] < settings.market_eod_fresh_ttl:
|
||
return p
|
||
if p is None:
|
||
cached = await cache.cache_get(_EOD_KEY)
|
||
if cached:
|
||
p = cached
|
||
_eod_state["payload"] = p
|
||
if p is not None:
|
||
_spawn_eod_refresh() # 陈旧但可用：立即返回，后台拉新
|
||
return p
|
||
return await _refresh_eod() # 首次访问：同步等（~3.5s，与旧行为一致）
|
||
|
||
|
||
# ======================= merge：实时叠加收盘 =======================
|
||
|
||
async def fetch_overview(is_trading_day: bool | None = None) -> dict:
|
||
"""聚合 live + EOD。实时价覆盖 close/change/pct_chg（realtime=True），今日实时成交额
|
||
（腾讯全市口径）在 EOD 尚未含今日时追加为盘中 bar。响应不再整包缓存：两层各有
|
||
进程内缓存，合并是 O(10) 操作，热路径 0 外部 RTT。"""
|
||
eod = await _get_eod()
|
||
live = await _fetch_live()
|
||
today_iso = date.today().isoformat()
|
||
|
||
indexes: list[dict] = []
|
||
for it in eod["indexes"]:
|
||
out = dict(it)
|
||
sym = _TENCENT_MAP.get(it["code"])
|
||
q = live.get(sym) if (live and sym) else None
|
||
if q and q.get("price") is not None:
|
||
# spark 永远来自 EOD（末点是上一收盘点，与实时价并存是已知的装饰性差异，不改历史序列）
|
||
out.update(close=q["price"], change=q["change"], pct_chg=q["pct"],
|
||
trade_date=today_iso, realtime=True)
|
||
else:
|
||
out["realtime"] = False
|
||
indexes.append(out)
|
||
|
||
stats = dict(eod["stats"]) if eod.get("stats") else None
|
||
history = list(eod.get("amount_history") or [])
|
||
|
||
# 今日实时两市成交额：沪深全市口径（万元->亿）。EOD 已含今日、非交易日、金额缺失时不追加。
|
||
if live and stats and stats.get("trade_date") != today_iso:
|
||
if is_trading_day is None:
|
||
is_trading_day = datetime.now().weekday() < 5 # 日历判定不可用时的降级启发式
|
||
if is_trading_day:
|
||
sh_amt = (live.get("s_sh000001") or {}).get("amount_wan")
|
||
sz_amt = (live.get(_TENCENT_SZ_TOTAL) or {}).get("amount_wan")
|
||
if sh_amt is not None and sz_amt is not None:
|
||
amt = round((sh_amt + sz_amt) / 10000, 2)
|
||
stats["amount_today"] = amt
|
||
history.append({"date": today_iso, "amount": amt, "intraday": True})
|
||
|
||
errors = list(eod.get("errors") or [])
|
||
if _eod_refresh_error:
|
||
errors.append(_eod_refresh_error)
|
||
if _live_error:
|
||
errors.append(_live_error)
|
||
|
||
return {
|
||
"updated_at": datetime.now().isoformat(),
|
||
"indexes": indexes,
|
||
"stats": stats,
|
||
"amount_history": history,
|
||
"errors": errors,
|
||
}
|