Compare commits

..
Author SHA1 Message Date
dd89a09643 fix(HEL-487): 盘中当天看板在 rt_k 无权限时降级到免费实时源
rt_k 失败、无权限、超时或空结果时改用东财全市场快照,再失败则用腾讯批量行情;两者都失败仍不退回昨天。

Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: multica-agent <github@multica.ai>
2026-09-08 10:52:10 +08:00
a043bc9eb1 fix(HEL-485): 盘中选择当天不再整页退回昨天
交易时段缺少盘后正式数据时继续展示当天盘中行情,只有开盘前、周末和历史日期才沿用最近收盘结果。

Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: multica-agent <github@multica.ai>
2026-09-08 10:16:52 +08:00
15 changed files with 1037 additions and 70 deletions
+3 -1
View File
@@ -37,7 +37,9 @@ class DataGateway:
) -> TushareClient: ) -> TushareClient:
if dataset_id: if dataset_id:
self.policy.assert_allowed(dataset_id, "tushare", usage) self.policy.assert_allowed(dataset_id, "tushare", usage)
return DatahubAwareTushareClient(self.tushare_provider.client(), self.datahub) legacy = self.tushare_provider.client()
legacy.realtime_aggregator = self.realtime_observer
return DatahubAwareTushareClient(legacy, self.datahub)
def dataset_status(self, trade_date: str) -> list[dict[str, Any]] | None: def dataset_status(self, trade_date: str) -> list[dict[str, Any]] | None:
return self.datahub.dataset_status(trade_date) return self.datahub.dataset_status(trade_date)
+10 -2
View File
@@ -3,7 +3,11 @@ from __future__ import annotations
from typing import Any from typing import Any
from backend.data.numbers import finite_number as _number from backend.data.numbers import finite_number as _number
from backend.data.providers.tushare_helpers import _display_time, _prices_equal from backend.data.providers.tushare_helpers import (
_display_time,
_prices_equal,
calendar_is_open,
)
class DailyMarketMixin: class DailyMarketMixin:
@@ -17,7 +21,11 @@ class DailyMarketMixin:
trade_date = requested trade_date = requested
else: else:
row = requested_rows[0] row = requested_rows[0]
trade_date = row["cal_date"] if row.get("is_open") == 1 else row.get("pretrade_date", requested) trade_date = (
row["cal_date"]
if calendar_is_open(row.get("is_open"))
else row.get("pretrade_date", requested)
)
resolved_rows = self.query( resolved_rows = self.query(
"trade_cal", "trade_cal",
+96 -12
View File
@@ -16,6 +16,12 @@ from backend.data.providers.tushare_transport import TushareError
class DashboardMixin: class DashboardMixin:
def _now(self) -> datetime:
clock = getattr(self, "clock", None)
if callable(clock):
return clock()
return datetime.now().astimezone()
def dashboard(self, requested_date: str) -> dict[str, Any]: def dashboard(self, requested_date: str) -> dict[str, Any]:
trade_date, previous_trade_date = self.resolve_trade_context(requested_date) trade_date, previous_trade_date = self.resolve_trade_context(requested_date)
if self.should_use_realtime(requested_date, trade_date): if self.should_use_realtime(requested_date, trade_date):
@@ -26,11 +32,12 @@ class DashboardMixin:
) )
daily = self._load_daily(trade_date) daily = self._load_daily(trade_date)
now = self._now()
if ( if (
not daily not daily
and requested_date == datetime.now().astimezone().strftime("%Y%m%d") and requested_date == now.strftime("%Y%m%d")
and trade_date == requested_date and trade_date == requested_date
and datetime.now().astimezone().time().replace(tzinfo=None) >= dt_time(9, 15) and now.time().replace(tzinfo=None) >= dt_time(9, 15)
): ):
return self._realtime_dashboard( return self._realtime_dashboard(
requested_date, requested_date,
@@ -98,15 +105,14 @@ class DashboardMixin:
} }
return apply_sentiment_to_dashboard(dashboard) return apply_sentiment_to_dashboard(dashboard)
@staticmethod def should_use_realtime(self, requested_date: str, trade_date: str) -> bool:
def should_use_realtime(requested_date: str, trade_date: str) -> bool: """Use live quotes for today's open session until official daily settles."""
"""Use rt_k for today's open market until end-of-day datasets settle.""" now = self._now()
now = datetime.now().astimezone()
today = now.strftime("%Y%m%d") today = now.strftime("%Y%m%d")
return ( return (
requested_date == today requested_date == today
and trade_date == today and trade_date == today
and dt_time(9, 15) <= now.time().replace(tzinfo=None) < dt_time(16, 30) and dt_time(9, 15) <= now.time().replace(tzinfo=None) < dt_time(15, 5)
) )
def _realtime_dashboard( def _realtime_dashboard(
@@ -122,7 +128,7 @@ class DashboardMixin:
) )
if not codes: if not codes:
raise TushareError("No active stock codes available for rt_k") raise TushareError("No active stock codes available for rt_k")
quotes = self.query("rt_k", {"ts_code": codes}) quotes, quote_source = self._load_realtime_quotes(codes, trade_date)
if not quotes: if not quotes:
raise TushareError(f"No realtime data returned for {trade_date}") raise TushareError(f"No realtime data returned for {trade_date}")
@@ -178,14 +184,30 @@ class DashboardMixin:
) )
sectors = _build_sectors(limits) sectors = _build_sectors(limits)
previous_sectors = _build_sectors(previous_limits) previous_sectors = _build_sectors(previous_limits)
now = datetime.now().astimezone() now = self._now()
market_status = _realtime_market_status(now.time().replace(tzinfo=None)) market_status = _realtime_market_status(now.time().replace(tzinfo=None))
if quote_source == "eastmoney_clist":
notice = (
"盘中行情由东财免费实时快照计算;涨停原因、封板时间和开板次数以盘后榜单校正为准。"
)
source_name = "eastmoney"
elif quote_source == "tencent_qt":
notice = (
"盘中行情由腾讯免费实时行情计算;涨停原因、封板时间和开板次数以盘后榜单校正为准。"
)
source_name = "tencent"
else:
notice = (
"盘中行情由 Tushare rt_k 实时计算;涨停原因、封板时间和开板次数以盘后榜单校正为准。"
)
source_name = "tushare"
dashboard = { dashboard = {
"meta": { "meta": {
"requested_date": _display_date(requested_date), "requested_date": _display_date(requested_date),
"trade_date": _display_date(trade_date), "trade_date": _display_date(trade_date),
"previous_trade_date": _display_date(previous_trade_date), "previous_trade_date": _display_date(previous_trade_date),
"source": "tushare", "source": source_name,
"quote_source": quote_source,
"mode": "realtime", "mode": "realtime",
"realtime": True, "realtime": True,
"market_status": market_status, "market_status": market_status,
@@ -193,7 +215,8 @@ class DashboardMixin:
"auto_refresh": False, "auto_refresh": False,
"quote_count": len(daily), "quote_count": len(daily),
"updated_at": now.isoformat(timespec="seconds"), "updated_at": now.isoformat(timespec="seconds"),
"notice": "盘中行情由 Tushare rt_k 实时计算;涨停原因、封板时间和开板次数以盘后榜单校正为准。", "notice": notice,
"indices": self._free_realtime_indices() if quote_source != "tushare_rt_k" else [],
}, },
"overview": _build_overview(daily, up_rows, down_rows, broken_rows), "overview": _build_overview(daily, up_rows, down_rows, broken_rows),
"limits": limits, "limits": limits,
@@ -207,6 +230,67 @@ class DashboardMixin:
} }
return apply_sentiment_to_dashboard(dashboard) return apply_sentiment_to_dashboard(dashboard)
def _realtime_aggregator(self):
aggregator = getattr(self, "realtime_aggregator", None)
if aggregator is None:
raise TushareError("免费实时源未配置")
return aggregator
def _load_realtime_quotes(
self,
codes: str,
trade_date: str,
) -> tuple[list[dict[str, Any]], str]:
rt_error = ""
try:
quotes = self.query("rt_k", {"ts_code": codes})
if quotes:
return list(quotes), "tushare_rt_k"
rt_error = f"No realtime data returned for {trade_date}"
except TushareError as exc:
rt_error = str(exc)
try:
quotes, quote_source = self._free_realtime_quotes(trade_date, codes)
except Exception as exc:
raise TushareError(
f"当天盘中实时行情不可用:rt_k={rt_error};免费源={exc}"
) from exc
if not quotes:
raise TushareError(
f"当天盘中实时行情不可用:rt_k={rt_error};免费源=empty"
)
return quotes, quote_source
def _free_realtime_quotes(
self,
trade_date: str,
codes: str = "",
) -> tuple[list[dict[str, Any]], str]:
aggregator = self._realtime_aggregator()
last_error = ""
try:
quotes = aggregator.eastmoney_market_quotes(expected_date=trade_date)
if quotes:
return quotes, "eastmoney_clist"
except Exception as exc:
last_error = str(exc)
code_list = [item for item in str(codes or "").split(",") if item]
try:
quotes = aggregator.tencent_market_quotes(code_list, expected_date=trade_date)
except Exception as exc:
raise TushareError(
f"eastmoney={last_error or 'empty'}tencent={exc}"
) from exc
if not quotes:
raise TushareError(f"eastmoney={last_error or 'empty'}tencent=empty")
return quotes, "tencent_qt"
def _free_realtime_indices(self) -> list[dict[str, Any]]:
try:
return self._realtime_aggregator().eastmoney_indices()
except Exception:
return []
def _load_realtime_reference( def _load_realtime_reference(
self, self,
trade_date: str, trade_date: str,
@@ -234,7 +318,7 @@ class DashboardMixin:
{"trade_date": previous_trade_date}, {"trade_date": previous_trade_date},
"ts_code,trade_date,total_share,float_share,free_share,total_mv,circ_mv", "ts_code,trade_date,total_share,float_share,free_share,total_mv,circ_mv",
) )
if not basic_rows or not price_limits: if not basic_rows:
raise TushareError(f"Realtime reference data is incomplete for {trade_date}") raise TushareError(f"Realtime reference data is incomplete for {trade_date}")
result = { result = {
"basic_rows": basic_rows, "basic_rows": basic_rows,
+11
View File
@@ -6,6 +6,17 @@ from typing import Any
from backend.data.numbers import finite_number as _number from backend.data.numbers import finite_number as _number
def calendar_is_open(value: Any) -> bool:
if value in (True, 1, "1", "Y", "y"):
return True
if value in (False, 0, "0", "N", "n", None, ""):
return False
try:
return int(value) == 1
except (TypeError, ValueError):
return False
def _text(value: Any) -> str: def _text(value: Any) -> str:
if isinstance(value, (list, tuple, set)): if isinstance(value, (list, tuple, set)):
return "".join(str(item).strip() for item in value if str(item).strip()) return "".join(str(item).strip() for item in value if str(item).strip())
+55
View File
@@ -59,6 +59,12 @@ class IndexMixin:
} }
def realtime_market_indices(self, requested_date: str) -> dict[str, Any]: def realtime_market_indices(self, requested_date: str) -> dict[str, Any]:
try:
return self._tushare_realtime_market_indices(requested_date)
except TushareError:
return self._free_realtime_market_indices(requested_date)
def _tushare_realtime_market_indices(self, requested_date: str) -> dict[str, Any]:
trade_date, _ = self.resolve_trade_context(requested_date) trade_date, _ = self.resolve_trade_context(requested_date)
index_names = { index_names = {
"000001.SH": "上证指数", "000001.SH": "上证指数",
@@ -116,3 +122,52 @@ class IndexMixin:
"average_return_20d": 0, "average_return_20d": 0,
}, },
} }
def _free_realtime_market_indices(self, requested_date: str) -> dict[str, Any]:
trade_date, _ = self.resolve_trade_context(requested_date)
aggregator = getattr(self, "realtime_aggregator", None)
if aggregator is None:
raise TushareError("免费实时源未配置")
quotes = aggregator.eastmoney_indices()
index_names = {
"000001": ("000001.SH", "上证指数"),
"399001": ("399001.SZ", "深证成指"),
"399006": ("399006.SZ", "创业板指"),
}
indices = []
for quote in quotes:
mapped = index_names.get(str(quote.get("code") or ""))
if not mapped:
continue
ts_code, name = mapped
close = _number(quote.get("price"))
previous_close = _number(quote.get("previous_close"))
if close <= 0 or previous_close <= 0:
continue
indices.append(
{
"ts_code": ts_code,
"name": str(quote.get("name") or name).strip(),
"trade_date": trade_date,
"close": close,
"pct_chg": round(_number(quote.get("change")) or (close / previous_close - 1) * 100, 3),
"return_5d": 0,
"amount_billion": round(_number(quote.get("amount_billion")), 2),
"quote_time": quote.get("quote_time") or "",
"source": quote.get("source") or "eastmoney_push2",
}
)
if len(indices) != 3:
raise TushareError("Realtime index quotes are incomplete")
return {
"trade_date": trade_date,
"source": "eastmoney_push2",
"realtime": True,
"precise": True,
"indices": indices,
"aggregate": {
"average_pct_chg": round(sum(item["pct_chg"] for item in indices) / len(indices), 3),
"average_return_5d": 0,
"average_return_20d": 0,
},
}
+237
View File
@@ -20,7 +20,17 @@ class RealtimeAggregateError(RuntimeError):
EASTMONEY_INDEX_URL = "https://push2.eastmoney.com/api/qt/ulist.np/get" EASTMONEY_INDEX_URL = "https://push2.eastmoney.com/api/qt/ulist.np/get"
EASTMONEY_SECTOR_URL = "https://push2.eastmoney.com/api/qt/clist/get" EASTMONEY_SECTOR_URL = "https://push2.eastmoney.com/api/qt/clist/get"
EASTMONEY_A_SHARE_BOARDS = (
"m:0+t:6",
"m:0+t:80",
"m:1+t:2",
"m:1+t:23",
"m:0+t:81",
)
EASTMONEY_QUOTE_FIELDS = "f12,f13,f14,f2,f3,f4,f5,f6,f15,f16,f17,f18,f8,f124"
EASTMONEY_MARKET_PAGE_SIZE = 100
TENCENT_INDEX_URL = "https://qt.gtimg.cn/q=sh000001,sz399001,sz399006" TENCENT_INDEX_URL = "https://qt.gtimg.cn/q=sh000001,sz399001,sz399006"
TENCENT_QUOTE_URL = "https://qt.gtimg.cn/q="
THS_LIMIT_URL = "https://data.10jqka.com.cn/dataapi/limit_up/limit_up_pool" THS_LIMIT_URL = "https://data.10jqka.com.cn/dataapi/limit_up/limit_up_pool"
XGB_POOL_URL = "https://flash-api.xuangubao.cn/api/pool/detail" XGB_POOL_URL = "https://flash-api.xuangubao.cn/api/pool/detail"
BROWSER_USER_AGENT = ( BROWSER_USER_AGENT = (
@@ -134,6 +144,145 @@ class WebRealtimeAggregator:
raise RealtimeAggregateError(f"Eastmoney returned {len(result)}/3 indices") raise RealtimeAggregateError(f"Eastmoney returned {len(result)}/3 indices")
return result return result
def eastmoney_market_quotes(self, expected_date: str = "") -> list[dict[str, Any]]:
"""Full A-share snapshot via Eastmoney clist, used when Tushare rt_k is unavailable."""
now = time.time()
cache_key = "assembled:eastmoney_market"
with self._response_cache_lock:
cached = self._response_cache.get(cache_key)
cache_age = now - float((cached or {}).get("created_at") or 0)
if cached and cache_age <= min(20, self.response_cache_ttl_seconds):
quotes = list(cached.get("payload") or [])
return self._filter_quotes_by_date(quotes, expected_date)
rows: list[dict[str, Any]] = []
board_errors: list[str] = []
for board in EASTMONEY_A_SHARE_BOARDS:
try:
rows.extend(self._eastmoney_board_quotes(board))
except Exception as exc:
board_errors.append(f"{board}:{exc}")
quotes = []
seen: set[str] = set()
for row in rows:
quote = _normalize_eastmoney_quote(row)
ts_code = str((quote or {}).get("ts_code") or "")
if not quote or ts_code in seen:
continue
seen.add(ts_code)
quotes.append(quote)
if len(quotes) < 200:
detail = f"{'; '.join(board_errors)}" if board_errors else ""
raise RealtimeAggregateError(
f"Eastmoney market snapshot too small: {len(quotes)}{detail}"
)
quotes = self._filter_quotes_by_date(quotes, expected_date)
with self._response_cache_lock:
self._response_cache[cache_key] = {"created_at": now, "payload": quotes}
return quotes
def _eastmoney_board_quotes(self, board: str) -> list[dict[str, Any]]:
first = self._eastmoney_market_page(board, 1)
data = first.get("data") or {}
rows = _diff_rows(data)
total = int(_number(data.get("total")))
page_count = 1
if total > 0:
page_count = max(1, (total + EASTMONEY_MARKET_PAGE_SIZE - 1) // EASTMONEY_MARKET_PAGE_SIZE)
for page in range(2, min(page_count, 40) + 1):
payload = self._eastmoney_market_page(board, page)
rows.extend(_diff_rows(payload.get("data") or {}))
return rows
def _eastmoney_market_page(self, board: str, page: int) -> dict[str, Any]:
return self._get_json(
EASTMONEY_SECTOR_URL,
{
"pn": str(page),
"pz": str(EASTMONEY_MARKET_PAGE_SIZE),
"po": "1",
"np": "1",
"fltt": "2",
"invt": "2",
"fid": "f12",
"fs": board,
"fields": EASTMONEY_QUOTE_FIELDS,
},
referer="https://quote.eastmoney.com/center/gridlist.html",
)
def _filter_quotes_by_date(
self,
quotes: list[dict[str, Any]],
expected_date: str,
) -> list[dict[str, Any]]:
want = str(expected_date or "").replace("-", "")
if not want or not quotes:
return quotes
dated = [item for item in quotes if str(item.get("quote_date") or "") == want]
if dated and len(dated) >= max(100, int(len(quotes) * 0.2)):
return dated
if dated:
return dated
if all(not item.get("quote_date") for item in quotes):
return quotes
raise RealtimeAggregateError(f"Eastmoney quotes are not for {want}")
def tencent_market_quotes(
self,
codes: list[str],
expected_date: str = "",
) -> list[dict[str, Any]]:
symbols: list[str] = []
seen: set[str] = set()
for raw in codes:
ts = str(raw or "").strip().upper()
if not ts:
continue
symbol = ts.split(".")[0]
if not symbol.isdigit() or len(symbol) != 6 or symbol in seen:
continue
seen.add(symbol)
if ts.endswith(".SH") or symbol.startswith(("5", "6", "9")):
symbols.append(f"sh{symbol}")
elif ts.endswith(".BJ") or symbol.startswith(("4", "8")):
symbols.append(f"bj{symbol}")
else:
symbols.append(f"sz{symbol}")
if not symbols:
raise RealtimeAggregateError("No stock codes available for Tencent quotes")
quotes: list[dict[str, Any]] = []
batch_size = 80
def load_batch(batch: list[str]) -> list[dict[str, Any]]:
raw, _cache_age = self._get_text(
f"{TENCENT_QUOTE_URL}{','.join(batch)}",
referer="https://gu.qq.com/",
encoding="gb18030",
)
return [
quote
for line in raw.splitlines()
if (quote := _parse_tencent_stock_quote(line))
]
batches = [symbols[index:index + batch_size] for index in range(0, len(symbols), batch_size)]
errors: list[str] = []
with ThreadPoolExecutor(max_workers=4) as executor:
for result in executor.map(self._capture, [lambda batch=batch: load_batch(batch) for batch in batches]):
rows, status = result
if status.get("ok") and rows:
quotes.extend(rows)
elif not status.get("ok"):
errors.append(str(status.get("error") or "batch failed"))
if len(quotes) < 200:
detail = f"{'; '.join(errors[:3])}" if errors else ""
raise RealtimeAggregateError(
f"Tencent market snapshot too small: {len(quotes)}{detail}"
)
return self._filter_quotes_by_date(quotes, expected_date)
def tencent_indices(self) -> list[dict[str, Any]]: def tencent_indices(self) -> list[dict[str, Any]]:
raw, cache_age = self._get_text( raw, cache_age = self._get_text(
TENCENT_INDEX_URL, TENCENT_INDEX_URL,
@@ -397,6 +546,94 @@ class WebRealtimeAggregator:
) from last_error ) from last_error
def _diff_rows(data: dict[str, Any]) -> list[dict[str, Any]]:
diff = data.get("diff") or []
if isinstance(diff, dict):
return [row for row in diff.values() if isinstance(row, dict)]
return [row for row in diff if isinstance(row, dict)]
def _parse_tencent_stock_quote(line: str) -> dict[str, Any] | None:
if '="' not in line:
return None
prefix, payload = line.split('="', 1)
fields = payload.rsplit('";', 1)[0].split("~")
if len(fields) < 38:
return None
symbol = fields[2]
if not symbol.isdigit() or len(symbol) != 6:
return None
close = _number(fields[3])
previous_close = _number(fields[4])
if close <= 0 or previous_close <= 0:
return None
marker = prefix.lower()
if "sh" in marker:
ts_code = f"{symbol}.SH"
elif "bj" in marker:
ts_code = f"{symbol}.BJ"
else:
ts_code = f"{symbol}.SZ"
try:
quote_time = datetime.strptime(fields[30], "%Y%m%d%H%M%S")
quote_date = quote_time.strftime("%Y%m%d")
epoch = int(quote_time.timestamp())
except ValueError:
quote_date = ""
epoch = 0
return {
"ts_code": ts_code,
"name": fields[1] or symbol,
"pre_close": previous_close,
"open": _number(fields[5]),
"high": _number(fields[33]),
"low": _number(fields[34]),
"close": close,
"vol": _number(fields[6]) * 100,
"amount": _number(fields[37]) * 10000,
"num": 0,
"quote_date": quote_date,
"quote_time_epoch": epoch,
"source": "tencent_qt",
}
def _normalize_eastmoney_quote(row: dict[str, Any]) -> dict[str, Any] | None:
symbol = str(row.get("f12") or "").strip()
if not symbol.isdigit() or len(symbol) != 6:
return None
close = _number(row.get("f2"))
previous_close = _number(row.get("f18"))
if close <= 0 or previous_close <= 0:
return None
market = int(_number(row.get("f13")))
if market == 1 or symbol.startswith(("5", "6", "9")):
ts_code = f"{symbol}.SH"
elif symbol.startswith(("4", "8")):
ts_code = f"{symbol}.BJ"
else:
ts_code = f"{symbol}.SZ"
epoch = int(_number(row.get("f124")))
quote_date = ""
if epoch > 0:
quote_date = datetime.fromtimestamp(epoch).astimezone().strftime("%Y%m%d")
return {
"ts_code": ts_code,
"name": row.get("f14") or symbol,
"pre_close": previous_close,
"open": _number(row.get("f17")),
"high": _number(row.get("f15")),
"low": _number(row.get("f16")),
"close": close,
"vol": _number(row.get("f5")) * 100,
"amount": _number(row.get("f6")),
"num": 0,
"quote_date": quote_date,
"quote_time_epoch": epoch,
"source": "eastmoney_clist",
}
def _normalize_sector(value: Any) -> str: def _normalize_sector(value: Any) -> str:
text = str(value or "").strip().replace(" ", "") text = str(value or "").strip().replace(" ", "")
for suffix in ("板块", "概念", "行业", "", "", "(A股)", "A股)"): for suffix in ("板块", "概念", "行业", "", "", "(A股)", "A股)"):
+72 -10
View File
@@ -63,11 +63,37 @@ class MarketServiceMixin:
if gateway is not None: if gateway is not None:
return gateway.tushare() return gateway.tushare()
# Compatibility for isolated legacy unit-test service stubs. # Compatibility for isolated legacy unit-test service stubs.
return TushareClient(self.token) client = TushareClient(self.token)
aggregator = getattr(self, "realtime_aggregator", None)
if aggregator is not None:
client.realtime_aggregator = aggregator
return client
def _now(self) -> datetime:
clock = getattr(self, "clock", None)
if callable(clock):
return clock()
return datetime.now().astimezone()
def _is_requested_open_session(self, requested_date: str) -> bool:
now = self._now()
if requested_date != now.strftime("%Y%m%d"):
return False
if now.time().replace(tzinfo=None) < dt_time(9, 15):
return False
client = self._tushare_client() if self.configured else None
resolve = getattr(client, "resolve_trade_context", None) if client else None
if resolve is None:
return now.weekday() < 5
try:
trade_date, _ = resolve(requested_date)
except Exception:
return now.weekday() < 5
return str(trade_date or "") == requested_date
def get_dashboard(self, trade_date: str, force: bool = False) -> dict[str, Any]: def get_dashboard(self, trade_date: str, force: bool = False) -> dict[str, Any]:
normalized_date = normalize_date(trade_date) normalized_date = normalize_date(trade_date)
now = datetime.now().astimezone() now = self._now()
if ( if (
normalized_date == now.strftime("%Y%m%d") normalized_date == now.strftime("%Y%m%d")
and now.time().replace(tzinfo=None) < datetime.strptime("09:15", "%H:%M").time() and now.time().replace(tzinfo=None) < datetime.strptime("09:15", "%H:%M").time()
@@ -174,14 +200,14 @@ class MarketServiceMixin:
def _should_retry_incomplete_snapshot( def _should_retry_incomplete_snapshot(
self, snapshot: dict[str, Any], requested_date: str self, snapshot: dict[str, Any], requested_date: str
) -> bool: ) -> bool:
if requested_date != date.today().strftime("%Y%m%d"): if requested_date != self._now().strftime("%Y%m%d"):
return False return False
meta = snapshot.get("meta") or {} meta = snapshot.get("meta") or {}
incomplete = ( actual = str(meta.get("trade_date") or "").replace("-", "")
meta.get("limit_data_source") == "derived" stale_carry = bool(meta.get("carried_forward") or actual != requested_date)
or bool(meta.get("carried_forward")) if stale_carry and self._is_requested_open_session(requested_date):
or str(meta.get("trade_date") or "").replace("-", "") != requested_date return True
) incomplete = meta.get("limit_data_source") == "derived" or stale_carry
return incomplete and self._snapshot_age_seconds(meta) >= 60 return incomplete and self._snapshot_age_seconds(meta) >= 60
def _annotate_data_status(self, dashboard: dict[str, Any]) -> dict[str, Any]: def _annotate_data_status(self, dashboard: dict[str, Any]) -> dict[str, Any]:
@@ -199,6 +225,9 @@ class MarketServiceMixin:
else: else:
meta["data_status"] = "preparing" meta["data_status"] = "preparing"
meta["display_notice"] = self._preparing_display_notice(actual, requested) meta["display_notice"] = self._preparing_display_notice(actual, requested)
elif meta.get("realtime"):
meta["data_status"] = "intraday"
meta.setdefault("display_notice", "")
else: else:
meta["data_status"] = "official" meta["data_status"] = "official"
meta.setdefault("display_notice", "") meta.setdefault("display_notice", "")
@@ -225,9 +254,9 @@ class MarketServiceMixin:
normalized_date: str, normalized_date: str,
snapshot: dict[str, Any], snapshot: dict[str, Any],
) -> bool: ) -> bool:
if not self.configured or normalized_date != date.today().strftime("%Y%m%d"): if not self.configured or normalized_date != self._now().strftime("%Y%m%d"):
return False return False
now = datetime.now().astimezone() now = self._now()
local_time = now.time().replace(tzinfo=None) local_time = now.time().replace(tzinfo=None)
realtime_start = datetime.strptime("09:15", "%H:%M").time() realtime_start = datetime.strptime("09:15", "%H:%M").time()
morning_end = datetime.strptime("11:35", "%H:%M").time() morning_end = datetime.strptime("11:35", "%H:%M").time()
@@ -264,7 +293,10 @@ class MarketServiceMixin:
raise TushareError("公共行情尚未配置") raise TushareError("公共行情尚未配置")
dashboard = self._tushare_client().dashboard(normalized_date) dashboard = self._tushare_client().dashboard(normalized_date)
meta = dashboard.setdefault("meta", {}) meta = dashboard.setdefault("meta", {})
quote_source = str(meta.get("quote_source") or "")
meta["source"] = source meta["source"] = source
if quote_source:
meta["quote_source"] = quote_source
meta["requested_date"] = self._display_compact_date(normalized_date) meta["requested_date"] = self._display_compact_date(normalized_date)
if meta.get("limit_data_source") == "derived": if meta.get("limit_data_source") == "derived":
meta.setdefault( meta.setdefault(
@@ -276,6 +308,12 @@ class MarketServiceMixin:
actual_date = normalize_date( actual_date = normalize_date(
str(dashboard.get("meta", {}).get("trade_date") or normalized_date) str(dashboard.get("meta", {}).get("trade_date") or normalized_date)
) )
if actual_date != normalized_date and self._is_requested_open_session(
normalized_date
):
raise TushareError(
f"Intraday dashboard resolved {actual_date} instead of {normalized_date}"
)
self.database.save_snapshot(actual_date, source, dashboard) self.database.save_snapshot(actual_date, source, dashboard)
if actual_date != normalized_date: if actual_date != normalized_date:
dashboard.setdefault("meta", {}).update( dashboard.setdefault("meta", {}).update(
@@ -297,6 +335,30 @@ class MarketServiceMixin:
) )
return self._apply_reason_overrides(self._with_storage(dashboard, cached=False)) return self._apply_reason_overrides(self._with_storage(dashboard, cached=False))
except TushareError as exc: except TushareError as exc:
if self._is_requested_open_session(normalized_date):
existing = self.database.get_snapshot(normalized_date)
existing_date = str(
((existing or {}).get("meta") or {}).get("trade_date") or ""
).replace("-", "")
if existing and existing_date == normalized_date:
kept = copy.deepcopy(existing)
kept.setdefault("meta", {}).update(
{
"requested_date": self._display_compact_date(normalized_date),
}
)
self.database.finish_sync(
sync_id,
"fallback",
self._record_count(kept),
str(exc),
"tushare",
)
return self._apply_reason_overrides(
self._with_storage(kept, cached=True)
)
self.database.finish_sync(sync_id, "failed", message=str(exc))
raise ValueError("当天盘中行情暂时不可用,请稍后重试。") from exc
fallback = self.database.get_latest_real_snapshot(normalized_date) fallback = self.database.get_latest_real_snapshot(normalized_date)
if fallback: if fallback:
actual = str((fallback.get("meta") or {}).get("trade_date") or "") actual = str((fallback.get("meta") or {}).get("trade_date") or "")
+2
View File
@@ -41,6 +41,8 @@ def official_catchup_due(today: str, snapshot: dict[str, object]) -> bool:
actual == today actual == today
and meta.get("limit_data_source") != "derived" and meta.get("limit_data_source") != "derived"
and not meta.get("carried_forward") and not meta.get("carried_forward")
and not meta.get("realtime")
and meta.get("mode") != "realtime"
): ):
return False return False
return True return True
+26 -26
View File
@@ -222,12 +222,12 @@
{ {
"provider": "eastmoney", "provider": "eastmoney",
"path": "backend/data/realtime.py", "path": "backend/data/realtime.py",
"runtime_role": "isolated realtime observation" "runtime_role": "isolated realtime observation and intraday dashboard fallback"
}, },
{ {
"provider": "tencent", "provider": "tencent",
"path": "backend/data/realtime.py", "path": "backend/data/realtime.py",
"runtime_role": "index observation fallback" "runtime_role": "index observation and intraday quote fallback"
} }
], ],
"provider_domains": [ "provider_domains": [
@@ -508,8 +508,8 @@
}, },
{ {
"path": "backend/data/providers/tushare_dashboard.py", "path": "backend/data/providers/tushare_dashboard.py",
"bytes": 28234, "bytes": 31361,
"lines": 648 "lines": 732
}, },
{ {
"path": "backend/data/providers/tushare_industries.py", "path": "backend/data/providers/tushare_industries.py",
@@ -551,6 +551,11 @@
"bytes": 15311, "bytes": 15311,
"lines": 387 "lines": 387
}, },
{
"path": "frontend/shared/dashboard.js",
"bytes": 15063,
"lines": 321
},
{ {
"path": "frontend/pages/pools/page.html", "path": "frontend/pages/pools/page.html",
"bytes": 14942, "bytes": 14942,
@@ -561,11 +566,6 @@
"bytes": 14743, "bytes": 14743,
"lines": 342 "lines": 342
}, },
{
"path": "frontend/shared/dashboard.js",
"bytes": 14740,
"lines": 316
},
{ {
"path": "frontend/shared/admin.js", "path": "frontend/shared/admin.js",
"bytes": 14410, "bytes": 14410,
@@ -631,6 +631,11 @@
"bytes": 8357, "bytes": 8357,
"lines": 116 "lines": 116
}, },
{
"path": "backend/data/providers/tushare_indices.py",
"bytes": 7823,
"lines": 173
},
{ {
"path": "backend/features/screener/formula.py", "path": "backend/features/screener/formula.py",
"bytes": 6983, "bytes": 6983,
@@ -638,8 +643,8 @@
}, },
{ {
"path": "backend/data/providers/tushare_daily.py", "path": "backend/data/providers/tushare_daily.py",
"bytes": 6837, "bytes": 6949,
"lines": 160 "lines": 168
}, },
{ {
"path": "backend/application.py", "path": "backend/application.py",
@@ -686,11 +691,6 @@
"bytes": 5690, "bytes": 5690,
"lines": 124 "lines": 124
}, },
{
"path": "backend/data/providers/tushare_indices.py",
"bytes": 5451,
"lines": 118
},
{ {
"path": "frontend/pages.config.js", "path": "frontend/pages.config.js",
"bytes": 5385, "bytes": 5385,
@@ -786,6 +786,11 @@
"bytes": 2514, "bytes": 2514,
"lines": 63 "lines": 63
}, },
{
"path": "backend/data/providers/tushare_helpers.py",
"bytes": 2360,
"lines": 75
},
{ {
"path": "backend/jobs/service.py", "path": "backend/jobs/service.py",
"bytes": 2337, "bytes": 2337,
@@ -811,11 +816,6 @@
"bytes": 2165, "bytes": 2165,
"lines": 35 "lines": 35
}, },
{
"path": "backend/data/providers/tushare_helpers.py",
"bytes": 2083,
"lines": 64
},
{ {
"path": "frontend/pages/market/breadth.js", "path": "frontend/pages/market/breadth.js",
"bytes": 2071, "bytes": 2071,
@@ -827,13 +827,13 @@
"lines": 45 "lines": 45
}, },
{ {
"path": "backend/features/system/routes.py", "path": "backend/jobs/refresh.py",
"bytes": 1791, "bytes": 1808,
"lines": 46 "lines": 48
}, },
{ {
"path": "backend/jobs/refresh.py", "path": "backend/features/system/routes.py",
"bytes": 1728, "bytes": 1791,
"lines": 46 "lines": 46
}, },
{ {
+2 -2
View File
@@ -213,12 +213,12 @@
{ {
"provider": "eastmoney", "provider": "eastmoney",
"path": "realtime_aggregator.py", "path": "realtime_aggregator.py",
"runtime_role": "isolated realtime observation" "runtime_role": "isolated realtime observation and intraday dashboard fallback"
}, },
{ {
"provider": "tencent", "provider": "tencent",
"path": "realtime_aggregator.py", "path": "realtime_aggregator.py",
"runtime_role": "index observation fallback" "runtime_role": "index observation and intraday quote fallback"
} }
], ],
"llm_entrypoints": [ "llm_entrypoints": [
+5
View File
@@ -67,6 +67,11 @@ async function startAdminRefresh() {
const actualCompact = actualDate.replaceAll("-", ""); const actualCompact = actualDate.replaceAll("-", "");
const updated = formatTimestamp(meta.updated_at); const updated = formatTimestamp(meta.updated_at);
const freshness = dashboardFreshnessMessage(meta); const freshness = dashboardFreshnessMessage(meta);
if (meta.realtime && actualCompact === requestedCompact && !meta.carried_forward) {
setAdminRefreshStatus("success", `刷新成功:已获取 ${actualDate} 的盘中行情,更新时间 ${updated}`, "circle-check");
showToast(`刷新成功:已获取 ${actualDate} 的盘中行情`);
return;
}
if (freshness || actualCompact !== requestedCompact || meta.carried_forward || meta.limit_data_source === "derived") { if (freshness || actualCompact !== requestedCompact || meta.carried_forward || meta.limit_data_source === "derived") {
setAdminRefreshStatus("warning", freshness || `部分正式数据尚未到齐,当前展示 ${actualDate || "最近可用数据"}`, "triangle-alert"); setAdminRefreshStatus("warning", freshness || `部分正式数据尚未到齐,当前展示 ${actualDate || "最近可用数据"}`, "triangle-alert");
setStatus(freshness || "部分正式数据尚未到齐,当前展示最近可用数据"); setStatus(freshness || "部分正式数据尚未到齐,当前展示最近可用数据");
+244 -15
View File
@@ -3,7 +3,8 @@ from __future__ import annotations
import copy import copy
import threading import threading
import unittest import unittest
from datetime import date, datetime, timedelta, timezone from datetime import date, datetime, timedelta, timezone, time as dt_time
from unittest.mock import patch
from pathlib import Path from pathlib import Path
from backend.features.market.service import MarketServiceMixin from backend.features.market.service import MarketServiceMixin
@@ -105,18 +106,84 @@ class FakeDerivedClient:
} }
SHANGHAI = timezone(timedelta(hours=8))
TRADE_DAY = date(2026, 9, 8)
def at_clock(hour: int, minute: int, day: date = TRADE_DAY) -> datetime:
return datetime(day.year, day.month, day.day, hour, minute, tzinfo=SHANGHAI)
class FakeMissingDailyClient: class FakeMissingDailyClient:
def __init__(self, open_today: bool = True):
self.open_today = open_today
def dashboard(self, trade_date: str): def dashboard(self, trade_date: str):
raise TushareError(f"No daily data returned for {trade_date}") raise TushareError(f"No daily data returned for {trade_date}")
def resolve_trade_context(self, requested: str):
if self.open_today:
return requested, "20260907"
return "20260907", "20260904"
class FakeRealtimeTodayClient:
def dashboard(self, trade_date: str):
return {
"meta": {
"trade_date": f"{trade_date[:4]}-{trade_date[4:6]}-{trade_date[6:8]}",
"requested_date": f"{trade_date[:4]}-{trade_date[4:6]}-{trade_date[6:8]}",
"realtime": True,
"mode": "realtime",
"market_status": "trading",
"notice": "盘中行情由 Tushare rt_k 实时计算;涨停原因、封板时间和开板次数以盘后榜单校正为准。",
"updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
},
"overview": {"limit_up_count": 15},
"limits": [{"code": "000001"}],
"broken": [],
"down_limits": [],
"yesterday_limits": [],
}
def resolve_trade_context(self, requested: str):
return requested, "20260907"
class FakeFreeRealtimeTodayClient:
def dashboard(self, trade_date: str):
return {
"meta": {
"trade_date": f"{trade_date[:4]}-{trade_date[4:6]}-{trade_date[6:8]}",
"requested_date": f"{trade_date[:4]}-{trade_date[4:6]}-{trade_date[6:8]}",
"realtime": True,
"mode": "realtime",
"quote_source": "eastmoney_clist",
"source": "eastmoney",
"market_status": "trading",
"notice": "盘中行情由东财免费实时快照计算;涨停原因、封板时间和开板次数以盘后榜单校正为准。",
"updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
"indices": [{"code": "000001", "price": 3800.1, "change": 0.5}],
},
"overview": {"limit_up_count": 18, "up_count": 2100, "amount_billion": 12345.6},
"limits": [{"code": "000001"}],
"broken": [],
"down_limits": [],
"yesterday_limits": [],
}
def resolve_trade_context(self, requested: str):
return requested, "20260907"
class SyncHarness(MarketServiceMixin): class SyncHarness(MarketServiceMixin):
def __init__(self, client, latest=None): def __init__(self, client, latest=None, clock=None):
self.configured = True self.configured = True
self.sync_lock = threading.Lock() self.sync_lock = threading.Lock()
self.database = FakeSyncDatabase(latest) self.database = FakeSyncDatabase(latest)
self._client = client self._client = client
self.current_user_id = 1 self.current_user_id = 1
self.clock = clock
def _tushare_client(self): def _tushare_client(self):
return self._client return self._client
@@ -142,23 +209,161 @@ class DashboardFreshnessTests(unittest.TestCase):
self.assertEqual(harness.database.finished[0][0][1], "success") self.assertEqual(harness.database.finished[0][0][1], "success")
self.assertEqual(verified_dashboard_result(payload), payload) self.assertEqual(verified_dashboard_result(payload), payload)
def test_missing_official_data_keeps_previous_day_with_preparing_notice(self): def test_intraday_refresh_keeps_today_and_does_not_fall_back_to_yesterday(self):
today = date.today() today = TRADE_DAY.strftime("%Y%m%d")
previous = (today - timedelta(days=1)).strftime("%Y-%m-%d")
latest = { latest = {
"meta": {"trade_date": previous, "source": "tushare"}, "meta": {"trade_date": "2026-09-07", "source": "tushare"},
"overview": {"limit_up_count": 20}, "overview": {"limit_up_count": 20},
} }
harness = SyncHarness(FakeMissingDailyClient(), latest) harness = SyncHarness(
payload = harness.sync_dashboard(today.strftime("%Y%m%d")) FakeRealtimeTodayClient(),
latest,
clock=lambda: at_clock(10, 5),
)
payload = harness.sync_dashboard(today)
meta = payload["meta"] meta = payload["meta"]
self.assertTrue(meta["carried_forward"]) self.assertFalse(meta.get("carried_forward"))
self.assertEqual(meta["data_status"], "preparing") self.assertTrue(meta["realtime"])
self.assertIn("今日数据正在准备,当前展示", meta["display_notice"]) self.assertEqual(meta["data_status"], "intraday")
self.assertIn("", meta["display_notice"]) self.assertEqual(str(meta["trade_date"]).replace("-", ""), today)
self.assertNotIn("No daily data", meta["display_notice"]) self.assertNotIn("今日数据正在准备", meta.get("display_notice") or "")
self.assertNotEqual(verified_dashboard_result(payload).get("status"), "failed") self.assertEqual(harness.database.saved[0][0], today)
def test_intraday_free_source_keeps_today_and_indices(self):
today = TRADE_DAY.strftime("%Y%m%d")
latest = {
"meta": {"trade_date": "2026-09-07", "source": "tushare"},
"overview": {"limit_up_count": 20},
}
harness = SyncHarness(
FakeFreeRealtimeTodayClient(),
latest,
clock=lambda: at_clock(10, 5),
)
payload = harness.sync_dashboard(today)
meta = payload["meta"]
self.assertFalse(meta.get("carried_forward"))
self.assertTrue(meta["realtime"])
self.assertEqual(meta["data_status"], "intraday")
self.assertEqual(str(meta["trade_date"]).replace("-", ""), today)
self.assertEqual(meta["quote_source"], "eastmoney_clist")
self.assertEqual(payload["overview"]["amount_billion"], 12345.6)
self.assertEqual(meta["indices"][0]["price"], 3800.1)
self.assertEqual(harness.database.saved[0][0], today)
def test_intraday_missing_quotes_do_not_carry_yesterday(self):
today = TRADE_DAY.strftime("%Y%m%d")
latest = {
"meta": {"trade_date": "2026-09-07", "source": "tushare"},
"overview": {"limit_up_count": 20},
}
harness = SyncHarness(
FakeMissingDailyClient(),
latest,
clock=lambda: at_clock(10, 5),
)
with self.assertRaises(ValueError) as ctx:
harness.sync_dashboard(today)
self.assertIn("当天盘中行情", str(ctx.exception))
self.assertFalse(harness.database.saved)
def test_intraday_keeps_existing_today_snapshot_when_refresh_fails(self):
today = TRADE_DAY.strftime("%Y%m%d")
existing = {
"meta": {
"trade_date": "2026-09-08",
"realtime": True,
"mode": "realtime",
"source": "tushare",
},
"overview": {"limit_up_count": 11},
"limits": [{"code": "600000"}],
"broken": [],
"down_limits": [],
"yesterday_limits": [],
}
harness = SyncHarness(
FakeMissingDailyClient(),
clock=lambda: at_clock(10, 5),
)
harness.database.get_snapshot = lambda *_args, **_kwargs: copy.deepcopy(existing)
payload = harness.sync_dashboard(today)
meta = payload["meta"]
self.assertEqual(str(meta["trade_date"]).replace("-", ""), today)
self.assertTrue(meta["realtime"])
self.assertEqual(meta["data_status"], "intraday")
self.assertFalse(meta.get("carried_forward"))
def test_lunch_and_after_hours_keep_today_until_official_arrives(self):
today = TRADE_DAY.strftime("%Y%m%d")
for clock in (lambda: at_clock(12, 0), lambda: at_clock(16, 10)):
harness = SyncHarness(
FakeRealtimeTodayClient(),
clock=clock,
)
payload = harness.sync_dashboard(today)
self.assertEqual(str(payload["meta"]["trade_date"]).replace("-", ""), today)
self.assertFalse(payload["meta"].get("carried_forward"))
def test_preopen_and_weekend_still_carry_last_session(self):
latest = {
"meta": {"trade_date": "2026-09-07", "source": "tushare"},
"overview": {"limit_up_count": 20},
}
preopen = SyncHarness(
FakeMissingDailyClient(),
latest,
clock=lambda: at_clock(8, 30),
)
preopen_payload = preopen.sync_dashboard(TRADE_DAY.strftime("%Y%m%d"))
self.assertTrue(preopen_payload["meta"]["carried_forward"])
self.assertEqual(preopen_payload["meta"]["data_status"], "preparing")
self.assertIn("今日数据正在准备,当前展示", preopen_payload["meta"]["display_notice"])
weekend = SyncHarness(
FakeMissingDailyClient(open_today=False),
latest,
clock=lambda: at_clock(10, 5, date(2026, 9, 5)),
)
weekend_payload = weekend.sync_dashboard("20260905")
self.assertTrue(weekend_payload["meta"]["carried_forward"])
def test_history_date_still_uses_official_or_preparing_notice(self):
latest = {
"meta": {"trade_date": "2026-09-01", "source": "tushare"},
"overview": {"limit_up_count": 8},
}
harness = SyncHarness(
FakeMissingDailyClient(),
latest,
clock=lambda: at_clock(10, 5),
)
payload = harness.sync_dashboard("20260902")
self.assertTrue(payload["meta"]["carried_forward"])
self.assertIn("所选日期数据尚未到齐", payload["meta"]["display_notice"])
def test_carried_today_snapshot_is_retried_immediately_in_session(self):
today = TRADE_DAY.strftime("%Y%m%d")
snapshot = {
"meta": {
"source": "tushare",
"trade_date": "2026-09-07",
"carried_forward": True,
"requested_date": "2026-09-08",
"updated_at": at_clock(10, 0).isoformat(),
},
"overview": {"limit_up_count": 1},
}
harness = SyncHarness(
FakeRealtimeTodayClient(),
clock=lambda: at_clock(10, 5),
)
harness.database.get_snapshot = lambda *_args, **_kwargs: copy.deepcopy(snapshot)
payload = harness.get_dashboard(today)
self.assertEqual(str(payload["meta"]["trade_date"]).replace("-", ""), today)
self.assertEqual(payload["meta"]["data_status"], "intraday")
self.assertTrue(harness.database.saved)
def test_weekend_carry_is_not_labeled_as_preparing(self): def test_weekend_carry_is_not_labeled_as_preparing(self):
snapshot = { snapshot = {
@@ -200,19 +405,43 @@ class DashboardFreshnessTests(unittest.TestCase):
{"meta": {"trade_date": iso, "limit_data_source": "derived"}}, {"meta": {"trade_date": iso, "limit_data_source": "derived"}},
) )
now = datetime.now().astimezone().time().replace(tzinfo=None) now = datetime.now().astimezone().time().replace(tzinfo=None)
if datetime.strptime("15:05", "%H:%M").time() <= now < datetime.strptime("22:00", "%H:%M").time(): if dt_time(15, 5) <= now < dt_time(22, 0):
self.assertFalse(due) self.assertFalse(due)
self.assertTrue(derived_due) self.assertTrue(derived_due)
else: else:
self.assertFalse(due) self.assertFalse(due)
self.assertFalse(derived_due) self.assertFalse(derived_due)
def test_official_catchup_is_due_for_intraday_snapshot_after_close(self):
today = TRADE_DAY.strftime("%Y%m%d")
snapshot = {
"meta": {
"trade_date": "2026-09-08",
"realtime": True,
"mode": "realtime",
}
}
with patch("backend.jobs.refresh.datetime") as mocked:
mocked.now.return_value = at_clock(16, 10)
mocked.strptime = datetime.strptime
self.assertTrue(official_catchup_due(today, snapshot))
official = {
"meta": {
"trade_date": "2026-09-08",
"limit_data_source": "official",
"realtime": False,
}
}
self.assertFalse(official_catchup_due(today, official))
class FrontendRefreshCopyTests(unittest.TestCase): class FrontendRefreshCopyTests(unittest.TestCase):
def test_dashboard_script_distinguishes_partial_from_failure(self): def test_dashboard_script_distinguishes_partial_from_failure(self):
script = (Path(__file__).resolve().parents[1] / "frontend" / "shared" / "dashboard.js").read_text(encoding="utf-8") script = (Path(__file__).resolve().parents[1] / "frontend" / "shared" / "dashboard.js").read_text(encoding="utf-8")
self.assertIn("今日数据正在准备,当前展示", script) self.assertIn("今日数据正在准备,当前展示", script)
self.assertIn("部分正式数据尚未到齐", script) self.assertIn("部分正式数据尚未到齐", script)
self.assertIn("盘中行情", script)
self.assertIn("meta.realtime && actualCompact === requestedCompact", script)
self.assertIn('job.status === "failed"', script) self.assertIn('job.status === "failed"', script)
failed_block = script.split("if (job.status === \"failed\")", 1)[1].split("const query", 1)[0] failed_block = script.split("if (job.status === \"failed\")", 1)[1].split("const query", 1)[0]
self.assertIn("后台刷新失败", failed_block) self.assertIn("后台刷新失败", failed_block)
+38
View File
@@ -3,6 +3,7 @@ from __future__ import annotations
import http.client import http.client
import json import json
import unittest import unittest
from datetime import datetime
from unittest.mock import MagicMock, patch from unittest.mock import MagicMock, patch
from backend.data.realtime import WebRealtimeAggregator from backend.data.realtime import WebRealtimeAggregator
@@ -377,6 +378,43 @@ class RealtimeAggregatorTests(unittest.TestCase):
self.assertEqual(rows[0]["quote_time"][:10], "2026-07-20") self.assertEqual(rows[0]["quote_time"][:10], "2026-07-20")
self.assertAlmostEqual(rows[0]["amount_billion"], 12946.52) self.assertAlmostEqual(rows[0]["amount_billion"], 12946.52)
@patch.object(WebRealtimeAggregator, "_get_json")
def test_eastmoney_market_quotes_normalize_and_keep_expected_date(self, get_json: MagicMock):
epoch = datetime(2026, 7, 20, 10, 5).timestamp()
rows = []
for index in range(200):
sz = index < 100
rows.append(
{
"f12": f"{index:06d}" if sz else f"{600000 + index - 100:06d}",
"f13": 0 if sz else 1,
"f14": f"股票{index}",
"f2": 11.2,
"f3": 2.0,
"f5": 10,
"f6": 50000000,
"f15": 11.3,
"f16": 11.0,
"f17": 11.1,
"f18": 11.0,
"f124": epoch,
}
)
def fake_get_json(_url, params, referer=""):
page = int(params.get("pn") or 1)
start = (page - 1) * 100
return {"rc": 0, "data": {"total": 200, "diff": rows[start:start + 100]}}
get_json.side_effect = fake_get_json
aggregator = WebRealtimeAggregator()
aggregator._response_cache.clear()
quotes = aggregator.eastmoney_market_quotes("20260720")
self.assertEqual(len(quotes), 200)
self.assertEqual(quotes[0]["ts_code"], "000000.SZ")
self.assertTrue(quotes[100]["ts_code"].endswith(".SH"))
self.assertEqual(quotes[0]["vol"], 1000)
self.assertEqual(quotes[0]["quote_date"], "20260720")
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+234
View File
@@ -1,8 +1,16 @@
from __future__ import annotations from __future__ import annotations
import unittest import unittest
from datetime import datetime, timedelta, timezone
from backend.data.providers.tushare_client import TushareClient from backend.data.providers.tushare_client import TushareClient
from backend.data.providers.tushare_helpers import calendar_is_open
from backend.data.providers.tushare_transport import TushareError
from backend.data.realtime import (
RealtimeAggregateError,
_normalize_eastmoney_quote,
_parse_tencent_stock_quote,
)
class FakeRealtimeClient(TushareClient): class FakeRealtimeClient(TushareClient):
@@ -81,6 +89,65 @@ class FakeRealtimeClient(TushareClient):
raise AssertionError(f"Unexpected API call: {api_name} {params}") raise AssertionError(f"Unexpected API call: {api_name} {params}")
FREE_QUOTES = [
{
"ts_code": "000001.SZ", "name": "", "pre_close": 10.0,
"open": 10.1, "high": 11.0, "low": 10.0, "close": 11.0,
"vol": 1000, "amount": 100000000, "num": 10,
"quote_date": "20260720",
},
{
"ts_code": "000002.SZ", "name": "", "pre_close": 20.0,
"open": 19.5, "high": 20.0, "low": 18.0, "close": 18.0,
"vol": 2000, "amount": 200000000, "num": 20,
"quote_date": "20260720",
},
{
"ts_code": "000003.SZ", "name": "", "pre_close": 30.0,
"open": 31.0, "high": 33.0, "low": 30.0, "close": 32.0,
"vol": 3000, "amount": 300000000, "num": 30,
"quote_date": "20260720",
},
]
class FakeFreeAggregator:
def __init__(self, quotes=None, fail=False):
self.quotes = list(quotes if quotes is not None else FREE_QUOTES)
self.fail = fail
self.calls = 0
def eastmoney_market_quotes(self, expected_date=""):
self.calls += 1
if self.fail:
raise RealtimeAggregateError("eastmoney down")
if expected_date and self.quotes:
dated = [
row for row in self.quotes
if str(row.get("quote_date") or "") == str(expected_date).replace("-", "")
]
if dated:
return dated
return list(self.quotes)
def tencent_market_quotes(self, codes, expected_date=""):
return self.eastmoney_market_quotes(expected_date)
def eastmoney_indices(self):
return [
{
"code": "000001",
"name": "上证指数",
"price": 3800.12,
"change": 0.85,
"previous_close": 3768.0,
"amount_billion": 4200.5,
"quote_time": "2026-07-20T10:05:00+08:00",
"source": "eastmoney_push2",
}
]
class RealtimeDashboardTests(unittest.TestCase): class RealtimeDashboardTests(unittest.TestCase):
def setUp(self): def setUp(self):
TushareClient._realtime_reference_cache.clear() TushareClient._realtime_reference_cache.clear()
@@ -130,6 +197,173 @@ class RealtimeDashboardTests(unittest.TestCase):
self.assertEqual(dashboard["meta"]["limit_data_source"], "derived") self.assertEqual(dashboard["meta"]["limit_data_source"], "derived")
self.assertIn("日线数据推算", dashboard["meta"]["notice"]) self.assertIn("日线数据推算", dashboard["meta"]["notice"])
def test_calendar_open_flag_accepts_string_and_bool(self):
self.assertTrue(calendar_is_open(1))
self.assertTrue(calendar_is_open("1"))
self.assertTrue(calendar_is_open(True))
self.assertFalse(calendar_is_open(0))
self.assertFalse(calendar_is_open("0"))
self.assertFalse(calendar_is_open(False))
original_query = self.client.query
def query(api_name, params=None, fields=""):
if api_name == "trade_cal":
return [
{
"cal_date": params.get("start_date"),
"is_open": "1",
"pretrade_date": "20260907",
}
]
return original_query(api_name, params, fields)
self.client.query = query
trade_date, previous = self.client.resolve_trade_context("20260908")
self.assertEqual(trade_date, "20260908")
self.assertEqual(previous, "20260907")
def test_session_clock_uses_realtime_until_official_window(self):
today = "20260908"
self.client.clock = lambda: datetime(
2026, 9, 8, 10, 5, tzinfo=timezone(timedelta(hours=8))
)
self.assertTrue(self.client.should_use_realtime(today, today))
self.client.clock = lambda: datetime(
2026, 9, 8, 16, 10, tzinfo=timezone(timedelta(hours=8))
)
self.assertFalse(self.client.should_use_realtime(today, today))
def test_realtime_dashboard_survives_missing_limit_table(self):
original_query = self.client.query
def query(api_name, params=None, fields=""):
if api_name == "stk_limit":
return []
return original_query(api_name, params, fields)
self.client.query = query
TushareClient._realtime_reference_cache.clear()
dashboard = self.client._realtime_dashboard("20260720", "20260720", "20260717")
self.assertTrue(dashboard["meta"]["realtime"])
self.assertEqual(dashboard["meta"]["quote_count"], 3)
self.assertEqual(dashboard["overview"]["limit_up_count"], 0)
def test_rt_k_permission_error_falls_back_to_free_quotes(self):
original_query = self.client.query
def query(api_name, params=None, fields=""):
if api_name == "rt_k":
raise TushareError("没有接口访问权限")
return original_query(api_name, params, fields)
self.client.query = query
self.client.realtime_aggregator = FakeFreeAggregator()
TushareClient._realtime_reference_cache.clear()
dashboard = self.client._realtime_dashboard("20260720", "20260720", "20260717")
self.assertTrue(dashboard["meta"]["realtime"])
self.assertEqual(dashboard["meta"]["quote_source"], "eastmoney_clist")
self.assertEqual(dashboard["meta"]["trade_date"], "2026-07-20")
self.assertEqual(dashboard["meta"]["quote_count"], 3)
self.assertEqual(dashboard["overview"]["limit_up_count"], 1)
self.assertEqual(dashboard["overview"]["limit_down_count"], 1)
self.assertEqual(dashboard["overview"]["amount_billion"], 6.0)
self.assertIn("东财免费实时", dashboard["meta"]["notice"])
self.assertEqual(dashboard["meta"]["indices"][0]["price"], 3800.12)
def test_rt_k_empty_result_falls_back_to_free_quotes(self):
original_query = self.client.query
def query(api_name, params=None, fields=""):
if api_name == "rt_k":
return []
return original_query(api_name, params, fields)
self.client.query = query
self.client.realtime_aggregator = FakeFreeAggregator()
TushareClient._realtime_reference_cache.clear()
dashboard = self.client._realtime_dashboard("20260720", "20260720", "20260717")
self.assertEqual(dashboard["meta"]["quote_source"], "eastmoney_clist")
self.assertEqual(str(dashboard["meta"]["trade_date"]).replace("-", ""), "20260720")
def test_rt_k_and_free_source_failure_keeps_today_error(self):
original_query = self.client.query
def query(api_name, params=None, fields=""):
if api_name == "rt_k":
raise TushareError("没有接口访问权限")
return original_query(api_name, params, fields)
self.client.query = query
self.client.realtime_aggregator = FakeFreeAggregator(fail=True)
TushareClient._realtime_reference_cache.clear()
with self.assertRaises(TushareError) as ctx:
self.client._realtime_dashboard("20260720", "20260720", "20260717")
self.assertIn("当天盘中实时行情不可用", str(ctx.exception))
self.assertIn("没有接口访问权限", str(ctx.exception))
def test_rt_k_and_eastmoney_failure_falls_back_to_tencent(self):
original_query = self.client.query
def query(api_name, params=None, fields=""):
if api_name == "rt_k":
raise TushareError("没有接口访问权限")
return original_query(api_name, params, fields)
class TencentOnlyAggregator(FakeFreeAggregator):
def eastmoney_market_quotes(self, expected_date=""):
raise RealtimeAggregateError("eastmoney blocked")
def tencent_market_quotes(self, codes, expected_date=""):
return list(FREE_QUOTES)
self.client.query = query
self.client.realtime_aggregator = TencentOnlyAggregator()
TushareClient._realtime_reference_cache.clear()
dashboard = self.client._realtime_dashboard("20260720", "20260720", "20260717")
self.assertEqual(dashboard["meta"]["quote_source"], "tencent_qt")
self.assertEqual(str(dashboard["meta"]["trade_date"]).replace("-", ""), "20260720")
self.assertIn("腾讯免费实时", dashboard["meta"]["notice"])
self.assertEqual(dashboard["overview"]["amount_billion"], 6.0)
def test_normalize_eastmoney_quote_maps_units_and_exchange(self):
quote = _normalize_eastmoney_quote(
{
"f12": "600000",
"f13": 1,
"f14": "浦发银行",
"f2": 10.5,
"f5": 12.0,
"f6": 200000000,
"f15": 10.8,
"f16": 10.2,
"f17": 10.3,
"f18": 10.0,
"f124": 1752986700,
}
)
self.assertEqual(quote["ts_code"], "600000.SH")
self.assertEqual(quote["vol"], 1200)
self.assertEqual(quote["close"], 10.5)
self.assertEqual(quote["pre_close"], 10.0)
self.assertEqual(quote["source"], "eastmoney_clist")
def test_parse_tencent_stock_quote_keeps_today_and_units(self):
line = (
'v_sz000001="51~平安银行~000001~11.73~11.70~11.66~346232~0~0~0~0~0~0~0~0~0~0~0~0~0~0~0~0~0~0~0~0~0~0~'
'~20260720100500~0.03~0.26~11.79~11.65~11.73/346232/406045563~346232~40605~0.18~5.24~~11.79~11.65~1.20~'
'2276.29~2276.31~0.49~12.87~10.53~0.95~-3076~11.73~4.43~5.34~~~0.18~40604.5563~0.0000~0~";'
)
quote = _parse_tencent_stock_quote(line)
self.assertEqual(quote["ts_code"], "000001.SZ")
self.assertEqual(quote["quote_date"], "20260720")
self.assertEqual(quote["close"], 11.73)
self.assertEqual(quote["pre_close"], 11.70)
self.assertEqual(quote["vol"], 34623200)
self.assertEqual(quote["amount"], 406050000)
self.assertEqual(quote["source"], "tencent_qt")
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+2 -2
View File
@@ -221,8 +221,8 @@ def build() -> dict[str, Any]:
{"provider": "datahub", "path": "backend/data/datahub/client.py", "runtime_role": "optional official EOD read path behind per-dataset flags"}, {"provider": "datahub", "path": "backend/data/datahub/client.py", "runtime_role": "optional official EOD read path behind per-dataset flags"},
{"provider": "ifind", "path": "backend/data/providers/ifind_client.py", "runtime_role": "realtime, charts, snapshots, enrichment"}, {"provider": "ifind", "path": "backend/data/providers/ifind_client.py", "runtime_role": "realtime, charts, snapshots, enrichment"},
{"provider": "eastmoney", "path": "backend/features/market/charts.py", "runtime_role": "display chart fallback"}, {"provider": "eastmoney", "path": "backend/features/market/charts.py", "runtime_role": "display chart fallback"},
{"provider": "eastmoney", "path": "backend/data/realtime.py", "runtime_role": "isolated realtime observation"}, {"provider": "eastmoney", "path": "backend/data/realtime.py", "runtime_role": "isolated realtime observation and intraday dashboard fallback"},
{"provider": "tencent", "path": "backend/data/realtime.py", "runtime_role": "index observation fallback"}, {"provider": "tencent", "path": "backend/data/realtime.py", "runtime_role": "index observation and intraday quote fallback"},
], ],
"provider_domains": [ "provider_domains": [
{"provider": "tushare", "path": "backend/data/providers/tushare_transport.py", "responsibility": "HTTP transport and provider errors"}, {"provider": "tushare", "path": "backend/data/providers/tushare_transport.py", "responsibility": "HTTP transport and provider errors"},