fix(HEL-487): 盘中当天看板在 rt_k 无权限时降级到免费实时源

rt_k 失败、无权限、超时或空结果时改用东财全市场快照,再失败则用腾讯批量行情;两者都失败仍不退回昨天。

Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: multica-agent <github@multica.ai>
This commit is contained in:
总工
2026-09-08 10:52:10 +08:00
co-authored by Cursor multica-agent
parent a043bc9eb1
commit dd89a09643
11 changed files with 663 additions and 18 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)
+81 -3
View File
@@ -128,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}")
@@ -186,12 +186,28 @@ class DashboardMixin:
previous_sectors = _build_sectors(previous_limits) previous_sectors = _build_sectors(previous_limits)
now = self._now() 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,
@@ -199,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,
@@ -213,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,
+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股)"):
+8 -1
View File
@@ -63,7 +63,11 @@ 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: def _now(self) -> datetime:
clock = getattr(self, "clock", None) clock = getattr(self, "clock", None)
@@ -289,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(
+9 -9
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": 28327, "bytes": 31361,
"lines": 654 "lines": 732
}, },
{ {
"path": "backend/data/providers/tushare_industries.py", "path": "backend/data/providers/tushare_industries.py",
@@ -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,
@@ -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,
+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": [
+48
View File
@@ -150,6 +150,32 @@ class FakeRealtimeTodayClient:
return requested, "20260907" 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, clock=None): def __init__(self, client, latest=None, clock=None):
self.configured = True self.configured = True
@@ -204,6 +230,28 @@ class DashboardFreshnessTests(unittest.TestCase):
self.assertNotIn("今日数据正在准备", meta.get("display_notice") or "") self.assertNotIn("今日数据正在准备", meta.get("display_notice") or "")
self.assertEqual(harness.database.saved[0][0], today) 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): def test_intraday_missing_quotes_do_not_carry_yesterday(self):
today = TRADE_DAY.strftime("%Y%m%d") today = TRADE_DAY.strftime("%Y%m%d")
latest = { latest = {
+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()
+180
View File
@@ -5,6 +5,12 @@ 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_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):
@@ -83,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()
@@ -184,6 +249,121 @@ class RealtimeDashboardTests(unittest.TestCase):
self.assertEqual(dashboard["meta"]["quote_count"], 3) self.assertEqual(dashboard["meta"]["quote_count"], 3)
self.assertEqual(dashboard["overview"]["limit_up_count"], 0) 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"},