cirry/stock
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

提交

Browse Source
This commit is contained in:
cirry
2026-09-04 19:05:23 +08:00
parent 76b422320b
commit 991d1bf343
10 changed files with 449 additions and 195 deletions
Download Patch File Download Diff File Expand all files Collapse all files

277
backend/app/data/market_overview.py
View File

@@ -1,40 +1,58 @@
"""大盘行情总览（主页展示）。
"""大盘行情总览（主页展示）——两层结构。
- A 股指数走 pro.index_daily，全球指数走 pro.index_global（均为收盘口径，晚间更新；
本 token 无 rt_idx_k 实时权限，故展示「最近交易日收盘」并标注日期）。
- 两市统计走 pro.daily_info：沪市取 SH_A、深市取 SZ_MARKET 汇总出
总市值 / 流通市值 / 成交额（单位亿元，接口原生口径）。
- 整包结果写 Redis 缓存（TTL 可配）；单个指数拉取失败只是跳过（errors 里注明），
全部失败才抛 MarketOverviewError —— 主页行情是锦上添花，不拖垮整页。
- 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 math
import time
from datetime import date, datetime, timedelta
import httpx
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"),
# (tushare代码, 名称, 地区, 腾讯符号) —— 展示顺序即列表顺序
# 标普500 腾讯符号是 s_usINX（不是 s_usSPX）；恒生科技是 s_hkHSTECH（不是 HSTECH）
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"),
("HSI", "恒生指数", "hk", "s_hkHSI"),
("HKTECH", "恒生科技", "hk", "s_hkHSTECH"),
("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 控频
_CACHE_KEY = "market_overview:v1"
_EOD_KEY = "market_overview:eod:v2" # v2：与旧整包缓存 v1 的 payload 形状不同，天然隔离
_AMOUNT_HIST_CAL_DAYS = 190 # 成交额历史的日历日窗口（≈128 交易日）
_AMOUNT_HIST_BARS = 120 # 输出的柱数（取尾部）
_LIVE_FAIL_TTL = 15.0 # live 层失败负缓存（秒）
class MarketOverviewError(RuntimeError):
@@ -66,6 +84,74 @@ def _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 形式）
@@ -91,7 +177,10 @@ def _quote_from_df(df: pd.DataFrame) -> dict | None:
def _fetch_stats_sync(pro) -> dict | None:
"""两市市值/成交统计：沪 SH_A + 深 SZ_MARKET（同一天口径相加，亿元）。"""
"""两市市值/成交统计：沪 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)
@@ -102,14 +191,14 @@ def _fetch_stats_sync(pro) -> dict | None:
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:
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(_d(sh_a["trade_date"]), _d(sz_m["trade_date"]))
d = min(_d(sh_m["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))
a, b = _f(sh_m.get(col)), _f(sz_m.get(col))
return None if a is None or b is None else round(a + b, 2)
return {
@@ -117,21 +206,42 @@ def _fetch_stats_sync(pro) -> dict | None:
"total_mv": _sum("total_mv"),
"float_mv": _sum("float_mv"),
"amount": _sum("amount"),
"turnover": _f(sh_a.get("tr")), # 换手率仅沪市有，展示口径注明沪市
"turnover": _f(sh_m.get("tr")), # 换手率仅沪市有，展示口径注明沪市
}
async def fetch_overview() -> dict:
"""聚合全部指数 + 两市统计（Redis 缓存整包，TTL 内直接回）。"""
cached = await cache.cache_get(_CACHE_KEY)
if cached is not None:
return cached
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": _d(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 in MARKET_INDEXES:
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)
@@ -152,11 +262,112 @@ async def fetch_overview() -> dict:
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 = {
"updated_at": datetime.now().isoformat(), # 字符串：cache_set 走 json.dumps
"fetched_at": datetime.now().isoformat(),
"fetched_ts": time.time(), # 墙钟（epoch float）：跨进程（Redis）判新鲜度用
"indexes": indexes,
"stats": stats,
"amount_history": amount_history,
"errors": errors,
}
await cache.cache_set(_CACHE_KEY, payload, ttl=settings.market_overview_ttl)
_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,
}