diff --git a/backend/data/datahub/bridge.py b/backend/data/datahub/bridge.py index 771f195..8cc1967 100644 --- a/backend/data/datahub/bridge.py +++ b/backend/data/datahub/bridge.py @@ -148,7 +148,7 @@ class DatahubBridge: cleaned = [str(item or "").strip() for item in codes if str(item or "").strip()] if not cleaned: return None - return self._try_quote_rows("quotes", {"codes": ",".join(cleaned[:60])}, minimum=1) + return self._try_quote_rows("quotes", {"codes": ",".join(cleaned)}, minimum=1) def try_index_quotes(self) -> list[dict[str, Any]] | None: flags = self.settings.flags("index_quotes") @@ -208,7 +208,7 @@ class DatahubBridge: if not rows: raise DatahubError("EMPTY", f"{dataset} chart empty") self._record_route(dataset, "datahub", str((response.meta or {}).get("source") or "datahub")) - return rows[-max(20, min(180, int(limit))):] + return rows[-max(1, int(limit)):] except Exception as exc: self._log_failure(dataset, exc) return None diff --git a/backend/data/providers/tushare_industries.py b/backend/data/providers/tushare_industries.py index d3bbbc1..67680a8 100644 --- a/backend/data/providers/tushare_industries.py +++ b/backend/data/providers/tushare_industries.py @@ -132,21 +132,63 @@ class ShenwanIndustryMixin: actual_trade_date = str(daily.get("trade_date") or "") outer_precise = actual_trade_date == trade_date outer_error = "" if outer_precise else ( - f"No Shenwan daily returned for {sector_code} on {trade_date}" + f"申万行业 {sector_code} 当日盘后正式数据尚未入库" ) + outer_source = "tushare_sw_daily" if outer_precise else "unavailable" if not outer_precise and allow_realtime_close: - try: - return self._sw_realtime_sector_snapshot( - industry, - members, + inner_ok = bool(member_rows) and not coverage_issue + if inner_ok: + sw_row, rt_source, rt_error = self._sw_outer_realtime( + sector_code, + str(industry.get("l2_name") or ""), trade_date, - previous_trade_date, finalized=True, ) - except TushareError as exc: - outer_error = f"{outer_error}; realtime close fallback failed: {exc}" + if sw_row: + daily = sw_row + actual_trade_date = str( + sw_row.get("quote_date") or sw_row.get("trade_date") or "" + ) + trade_time = str(sw_row.get("trade_time") or sw_row.get("quote_time") or "") + quote_clock = ( + trade_time[11:19] + if len(trade_time) >= 19 + else str(sw_row.get("quote_clock") or "") + ) + outer_precise = actual_trade_date == trade_date + if quote_clock and quote_clock < "15:00:00": + outer_precise = False + outer_source = rt_source or "eastmoney_sw" + outer_error = "" if outer_precise else ( + rt_error or f"申万行业 {sector_code} 免费实时尚未形成收盘快照" + ) + else: + outer_error = rt_error or outer_error + else: + try: + snapshot = self._sw_realtime_sector_snapshot( + industry, + members, + trade_date, + previous_trade_date, + finalized=True, + ) + snapshot.update({ + "raw_member_count": raw_member_count, + "excluded_member_count": len(excluded_members), + "excluded_members": excluded_members, + }) + return snapshot + except TushareError: + outer_error = f"{outer_error}; 免费实时成分暂不可用" - official_change = _number(daily.get("pct_change")) if outer_precise else None + official_change = None + if outer_precise: + official_change = _number( + daily.get("pct_change") + if daily.get("pct_change") not in (None, "") + else daily.get("change") + ) return { "code": sector_code, "name": industry.get("l2_name") or daily.get("name") or sector_code, @@ -173,9 +215,9 @@ class ShenwanIndustryMixin: "amount_billion": round(amount_billion, 2), "count": 0, "max_streak": 0, - "source": "tushare_sw_daily+member_daily" if outer_precise else "tushare_member_daily", + "source": f"{outer_source}+tushare_member_daily" if outer_precise else "tushare_member_daily", "inner_source": "tushare_member_daily", - "outer_source": "tushare_sw_daily" if outer_precise else "unavailable", + "outer_source": outer_source, "taxonomy": "sw_l2", "industry": industry, "trade_date": trade_date, @@ -189,7 +231,7 @@ class ShenwanIndustryMixin: "inner_error": inner_error, "outer_error": outer_error, "schema_version": 6, - "methodology": "外显使用申万二级行业官方日线;内核独立使用当日成分日线宽度与等权涨跌聚合", + "methodology": "外显使用已发布 sw_daily 或免费申万实时;内核优先使用当日成分日线,不调用 rt_sw_k", } def _sw_sector_members( @@ -508,29 +550,72 @@ class ShenwanIndustryMixin: codes: list[str], trade_date: str, ) -> tuple[list[dict[str, Any]], str]: - if not codes: + wanted = [str(code).strip() for code in codes if str(code or "").strip()] + if not wanted: return [], "unavailable" + best_rows: list[dict[str, Any]] = [] + best_source = "unavailable" + + def consider(rows: list[dict[str, Any]] | None, source: str) -> list[dict[str, Any]]: + nonlocal best_rows, best_source + filtered = _filter_quotes_for_codes(rows, wanted) + if len(filtered) > len(best_rows): + best_rows = filtered + best_source = source + return filtered + + hub_market = getattr(self, "try_market_quotes", None) + if callable(hub_market): + filtered = consider(hub_market(trade_date) or [], "datahub") + if len(filtered) >= max(1, int(len(wanted) * 0.9)): + return filtered, "datahub" + hub = getattr(self, "try_quotes", None) if callable(hub): - rows = hub(codes) or [] - if rows: - return list(rows), "datahub" + collected: list[dict[str, Any]] = [] + for index in range(0, len(wanted), _QUOTE_BATCH): + collected.extend(hub(wanted[index:index + _QUOTE_BATCH]) or []) + filtered = consider(collected, "datahub") + if len(filtered) >= max(1, int(len(wanted) * 0.9)): + return filtered, "datahub" + + aggregator = getattr(self, "realtime_aggregator", None) + loader = getattr(aggregator, "eastmoney_stock_quotes", None) if aggregator else None + if callable(loader): + try: + filtered = consider(loader(wanted, expected_date=trade_date) or [], "eastmoney_ulist") + if len(filtered) >= max(1, int(len(wanted) * 0.9)): + return filtered, "eastmoney_ulist" + except Exception: + pass + try: - quotes, source = self._load_realtime_quotes(",".join(codes), trade_date) - return quotes, source - except TushareError as exc: - message = str(exc) - if "rt_k" in message or "权限" in message: - aggregator = getattr(self, "realtime_aggregator", None) - loader = getattr(aggregator, "eastmoney_stock_quotes", None) if aggregator else None - if callable(loader): - try: - rows = loader(codes, expected_date=trade_date) - if rows: - return list(rows), "eastmoney_ulist" - except Exception: - pass - raise + quotes, source = self._free_realtime_quotes(trade_date, "") + consider(quotes, source) + except TushareError: + pass + + if best_rows: + return best_rows, best_source + return [], "unavailable" + + +_QUOTE_BATCH = 60 + + +def _filter_quotes_for_codes( + rows: list[dict[str, Any]] | None, + codes: list[str], +) -> list[dict[str, Any]]: + wanted = {str(code) for code in codes if code} + filtered: list[dict[str, Any]] = [] + seen: set[str] = set() + for row in rows or []: + ts_code = str(row.get("ts_code") or "") + if ts_code in wanted and ts_code not in seen: + seen.add(ts_code) + filtered.append(row) + return filtered def _filter_members_by_listing( diff --git a/backend/features/market/charts.py b/backend/features/market/charts.py index 72adb41..61afe09 100644 --- a/backend/features/market/charts.py +++ b/backend/features/market/charts.py @@ -23,7 +23,7 @@ class ChartDataError(RuntimeError): pass -DAILY_CHART_LIMIT = 250 +DAILY_CHART_LIMIT = 45 TRENDS_URL = "https://push2delay.eastmoney.com/api/qt/stock/trends2/get" @@ -377,7 +377,7 @@ class MarketChartClient: pass if not normalized: raise ChartDataError("No iFinD daily chart data returned") - return normalized[-max(20, min(180, int(limit))):] + return normalized[-max(1, int(limit)):] def _previous_close(self, code: str, trade_date: str, fallback: float) -> float: today = datetime.now().astimezone().date().isoformat() diff --git a/backend/features/market/service.py b/backend/features/market/service.py index 1950f83..51391b8 100644 --- a/backend/features/market/service.py +++ b/backend/features/market/service.py @@ -1161,7 +1161,7 @@ class MarketServiceMixin: intraday_status = "unavailable" intraday_notice = "分时行情暂不可用,请稍后重试。" - prices = list(detail.get("prices") or [])[-60:] + prices = list(detail.get("prices") or [])[-DAILY_CHART_LIMIT:] stock = dict(detail.get("stock") or {"code": code}) realtime = bool(detail_meta.get("realtime")) return { diff --git a/config/architecture-inventory.json b/config/architecture-inventory.json index 5e12cd2..1e639c6 100644 --- a/config/architecture-inventory.json +++ b/config/architecture-inventory.json @@ -501,6 +501,11 @@ "bytes": 34631, "lines": 812 }, + { + "path": "backend/data/providers/tushare_industries.py", + "bytes": 33324, + "lines": 766 + }, { "path": "database.py", "bytes": 32073, @@ -511,11 +516,6 @@ "bytes": 31756, "lines": 562 }, - { - "path": "backend/data/providers/tushare_industries.py", - "bytes": 29886, - "lines": 681 - }, { "path": "backend/features/heaven/manual.py", "bytes": 24521, @@ -533,8 +533,8 @@ }, { "path": "frontend/pages/market/preview.js", - "bytes": 18339, - "lines": 450 + "bytes": 18230, + "lines": 447 }, { "path": "backend/features/heaven/trend.py", @@ -548,8 +548,8 @@ }, { "path": "frontend/pages/market/charts.js", - "bytes": 15311, - "lines": 387 + "bytes": 15743, + "lines": 401 }, { "path": "frontend/shared/admin.js", @@ -623,7 +623,7 @@ }, { "path": "frontend/pages/market/entity-detail.js", - "bytes": 9119, + "bytes": 9139, "lines": 199 }, { @@ -661,16 +661,16 @@ "bytes": 6547, "lines": 220 }, - { - "path": "frontend/pages/market/stock-detail.js", - "bytes": 6540, - "lines": 142 - }, { "path": "frontend/pages/sentiment/page.html", "bytes": 6488, "lines": 81 }, + { + "path": "frontend/pages/market/stock-detail.js", + "bytes": 6325, + "lines": 137 + }, { "path": "backend/features/screener/backtest.py", "bytes": 6202, diff --git a/frontend/m/js/pages.js b/frontend/m/js/pages.js index af02d59..9d105b2 100644 --- a/frontend/m/js/pages.js +++ b/frontend/m/js/pages.js @@ -3402,9 +3402,9 @@ const payload = detail && detail.payload ? detail.payload : {}; const meta = payload.meta || {}; if (tab === "daily") { - const bars = (payload.prices || []).slice(-48); + const bars = (payload.prices || []).slice(-45); const last = bars.length ? bars[bars.length - 1].trade_date : ""; - return "日线 · 近48根 · 至 " + (displayCompactDate(last) || "--"); + return "日线 · 近45根 · 至 " + (displayCompactDate(last) || "--"); } const d = displayCompactDate(meta.intraday_trade_date) || displayCompactDate(meta.trade_date); return "分时 · " + (d || "--"); @@ -3702,7 +3702,7 @@ const W = 360, H = 240, padL = 8, padR = 52, padT = 10, padB = 22; const pw = W - padL - padR; const ph = H - padT - padB; - const prices = (payload.prices || []).slice(-48); + const prices = (payload.prices || []).slice(-45); if (prices.length < 2) return emptyChart("日线数据暂不可用"); diff --git a/frontend/pages.config.js b/frontend/pages.config.js index 1a29586..6e0c38d 100644 --- a/frontend/pages.config.js +++ b/frontend/pages.config.js @@ -68,10 +68,10 @@ "/pages/sentiment/page.js?v=20260729-1", "/pages/pools/page.js?v=20260820-1", "/pages/market/breadth.js?v=20260803-1", - "/pages/market/charts.js?v=20260803-1", - "/pages/market/entity-detail.js?v=20260803-1", - "/pages/market/stock-detail.js?v=20260803-1", - "/pages/market/preview.js?v=20260806-1", + "/pages/market/charts.js?v=20260908-1", + "/pages/market/entity-detail.js?v=20260908-1", + "/pages/market/stock-detail.js?v=20260908-1", + "/pages/market/preview.js?v=20260908-1", "/pages/market/search.js?v=20260803-1", "/pages/market/bindings.js?v=20260803-1", "/pages/ladder/page.js?v=20260820-1", diff --git a/frontend/pages/market/charts.js b/frontend/pages/market/charts.js index 184c7ea..1f5de99 100644 --- a/frontend/pages/market/charts.js +++ b/frontend/pages/market/charts.js @@ -1,3 +1,16 @@ +const DAILY_CHART_BARS = 45; + +function visibleDailyPrices(prices) { + return (prices || []).slice(-DAILY_CHART_BARS); +} + +function dailyChartSourceLabel(prices, notice) { + const count = visibleDailyPrices(prices).length; + const base = `日 K 行情 · ${count} 个交易日`; + const text = String(notice || "").trim(); + return text ? `${base} · ${text}` : base; +} + function currentChartPalette() { const style = getComputedStyle(document.documentElement); const color = (token, fallback) => style.getPropertyValue(token).trim() || fallback; @@ -56,7 +69,8 @@ function drawCandlestick(context, x, item, priceY, candleWidth, palette = curren function drawPriceChart(prices) { const canvas = elements.priceChart; - if (!prices?.length) { + const visible = visibleDailyPrices(prices); + if (!visible.length) { clearPriceChart("暂无日 K 数据"); return; } @@ -81,15 +95,15 @@ function drawPriceChart(prices) { const gap = 12; const priceBottom = height - bottom - volumeHeight - gap; const plotWidth = width - left - right; - const highs = prices.map((item) => number(item.high)); - const lows = prices.map((item) => number(item.low)); + const highs = visible.map((item) => number(item.high)); + const lows = visible.map((item) => number(item.low)); const maximum = Math.max(...highs); const minimum = Math.min(...lows); const range = Math.max(maximum - minimum, maximum * 0.01, 0.01); - const volumes = prices.map((item) => number(item.volume)); + const volumes = visible.map((item) => number(item.volume)); const maxVolume = Math.max(...volumes, 1); const priceY = (value) => top + (maximum - value) / range * (priceBottom - top); - const step = plotWidth / prices.length; + const step = plotWidth / visible.length; const candleWidth = clamp(step * 0.62, 2, 8); context.strokeStyle = palette.grid; @@ -105,7 +119,7 @@ function drawPriceChart(prices) { context.fillText((maximum - range * line / 4).toFixed(2), left - 5, y + 4); } - prices.forEach((item, index) => { + visible.forEach((item, index) => { const x = left + step * index + step / 2; const color = drawCandlestick(context, x, item, priceY, candleWidth, palette); const volumeBarHeight = number(item.volume) / maxVolume * volumeHeight; @@ -117,10 +131,10 @@ function drawPriceChart(prices) { context.textAlign = "center"; context.fillStyle = palette.axis; - const labelIndexes = [0, Math.floor((prices.length - 1) / 2), prices.length - 1]; + const labelIndexes = [0, Math.floor((visible.length - 1) / 2), visible.length - 1]; labelIndexes.forEach((index) => { const x = left + step * index + step / 2; - context.fillText(String(prices[index].trade_date).slice(5), x, height - 5); + context.fillText(String(visible[index].trade_date).slice(5), x, height - 5); }); } @@ -301,7 +315,7 @@ function drawIntradayPreviewChart(points, dailyPrices, referenceClose = 0) { function drawDailyPreviewChart(prices) { const { context, width, height, palette } = prepareStockPreviewCanvas(); - const visible = prices.slice(-45); + const visible = visibleDailyPrices(prices); const visibleStart = prices.length - visible.length; const left = 45; const right = 10; diff --git a/frontend/pages/market/entity-detail.js b/frontend/pages/market/entity-detail.js index 4c293f6..89ac9ef 100644 --- a/frontend/pages/market/entity-detail.js +++ b/frontend/pages/market/entity-detail.js @@ -113,13 +113,13 @@ function renderEntityDetailMetrics(metrics) { } function drawEntityDetailChart(series, canvas = elements.entityDetailChart) { - const candles = (series || []).filter((item) => number(item.close) > 0).map((item) => { + const candles = visibleDailyPrices((series || []).filter((item) => number(item.close) > 0).map((item) => { const close = number(item.close); const open = number(item.open) || close; const high = Math.max(number(item.high) || close, open, close); const low = Math.min(number(item.low) || close, open, close); return { ...item, open, high, low, close }; - }); + })); if (!candles.length) { clearEntityDetailChart("暂无日 K 数据", canvas); return; diff --git a/frontend/pages/market/preview.js b/frontend/pages/market/preview.js index 4142f68..d52ebee 100644 --- a/frontend/pages/market/preview.js +++ b/frontend/pages/market/preview.js @@ -368,10 +368,7 @@ function selectStockPreviewChart(chart) { } else if ((payload.prices || []).length) { setText("stockPreviewDate", payload.meta?.trade_date || "最新行情"); const notice = String(payload.meta?.notice || "").trim(); - setText( - "stockPreviewSource", - notice ? `日 K 行情 · ${payload.prices.length} 个交易日 · ${notice}` : `日 K 行情 · ${payload.prices.length} 个交易日`, - ); + setText("stockPreviewSource", dailyChartSourceLabel(payload.prices, notice)); drawDailyPreviewChart(payload.prices); } else { setText("stockPreviewDate", payload.meta?.trade_date || "最新行情"); diff --git a/frontend/pages/market/stock-detail.js b/frontend/pages/market/stock-detail.js index 660c586..78b5a40 100644 --- a/frontend/pages/market/stock-detail.js +++ b/frontend/pages/market/stock-detail.js @@ -46,10 +46,7 @@ async function openStock(code, fallback = null) { updateWatchButton(); if (state.stockDetailChartMode === "daily") { const notice = String(payload.meta?.notice || "").trim(); - setText( - "chartSource", - notice ? `日 K 行情 · ${payload.prices.length} 个交易日 · ${notice}` : `日 K 行情 · ${payload.prices.length} 个交易日`, - ); + setText("chartSource", dailyChartSourceLabel(payload.prices, notice)); requestAnimationFrame(() => drawPriceChart(payload.prices || [])); } } catch (error) { @@ -69,9 +66,7 @@ async function selectStockDetailChart(mode) { const notice = String(state.stockDetail?.meta?.notice || "").trim(); setText( "chartSource", - prices.length - ? (notice ? `日 K 行情 · ${prices.length} 个交易日 · ${notice}` : `日 K 行情 · ${prices.length} 个交易日`) - : "正在加载行情", + prices.length ? dailyChartSourceLabel(prices, notice) : "正在加载行情", ); if (prices.length) requestAnimationFrame(() => drawPriceChart(prices)); else clearPriceChart("正在加载日 K 数据"); diff --git a/tests/test_global_search.py b/tests/test_global_search.py index 624da3e..9d02bef 100644 --- a/tests/test_global_search.py +++ b/tests/test_global_search.py @@ -94,7 +94,7 @@ class GlobalSearchTests(unittest.TestCase): self.assertIn('event.key.toLowerCase() !== "k"', script) self.assertIn('openStock(item.id, { code: item.code', script) self.assertNotIn('include_notes', script) - self.assertIn('const candles = (series || [])', script) + self.assertIn('const candles = visibleDailyPrices((series || [])', script) self.assertIn('renderStockNotes(payload.notes || [])', script) diff --git a/tests/test_hel494_regressions.py b/tests/test_hel494_regressions.py index 1b2e274..7b2ee27 100644 --- a/tests/test_hel494_regressions.py +++ b/tests/test_hel494_regressions.py @@ -191,3 +191,88 @@ class EastmoneyHelperTests(unittest.TestCase): self.assertAlmostEqual(quote["change"], 2.88) params = get_json.call_args.args[1] self.assertEqual(params["secids"], "90.801074") + + +class ChartWindowTests(unittest.TestCase): + def test_display_window_is_45_not_250(self) -> None: + from backend.features.market.charts import DAILY_CHART_LIMIT + + self.assertEqual(DAILY_CHART_LIMIT, 45) + + +class MemberQuoteCoverageTests(unittest.TestCase): + def test_prefers_full_hub_market_over_truncated_named_quotes(self) -> None: + client = TushareClient(token="demo") + wanted = [f"{index:06d}.SZ" for index in range(205)] + market = [ + {"ts_code": code, "close": 10.0, "pre_close": 9.0} + for code in wanted + ] + client.try_market_quotes = MagicMock(return_value=market) + client.try_quotes = MagicMock(return_value=market[:60]) + client.realtime_aggregator = MagicMock() + rows, source = client._load_member_realtime_quotes(wanted, "20260908") + self.assertEqual(len(rows), 205) + self.assertEqual(source, "datahub") + client.try_quotes.assert_not_called() + + def test_ignores_non_member_quotes_from_market_snapshot(self) -> None: + client = TushareClient(token="demo") + client.try_market_quotes = MagicMock( + return_value=[ + {"ts_code": "000737.SZ", "close": 12.3, "pre_close": 11.2}, + {"ts_code": "600000.SH", "close": 10.0, "pre_close": 9.9}, + ] + ) + client.try_quotes = MagicMock(return_value=[]) + client._free_realtime_quotes = MagicMock(return_value=([], "empty")) + rows, _source = client._load_member_realtime_quotes( + ["000737.SZ", "000630.SZ"], "20260908" + ) + self.assertEqual([row["ts_code"] for row in rows], ["000737.SZ"]) + + def test_closed_keeps_daily_inner_when_sw_daily_missing(self) -> None: + client = TushareClient(token="demo") + client.resolve_trade_context = lambda _date: ("20260908", "20260907") + client.sw_stock_industry = MagicMock( + return_value={"l2_code": "801074.SI", "l2_name": "工业金属"} + ) + client._sw_sector_members = MagicMock( + return_value=[ + {"ts_code": "000737.SZ", "name": "北方铜业"}, + {"ts_code": "000630.SZ", "name": "铜陵有色"}, + ] + ) + client._stock_listing_reference = MagicMock(return_value={}) + client._load_daily = MagicMock( + return_value=[ + {"ts_code": "000737.SZ", "name": "北方铜业", "pct_chg": 2, "amount": 1e8}, + {"ts_code": "000630.SZ", "name": "铜陵有色", "pct_chg": 1, "amount": 1e8}, + ] + ) + client._confirmed_suspended_members = MagicMock(return_value=[]) + client.query = MagicMock(return_value=[]) + client._sw_realtime_sector_snapshot = MagicMock( + side_effect=AssertionError("daily inner should be kept") + ) + client.realtime_aggregator = MagicMock() + client.realtime_aggregator.eastmoney_shenwan_quote.return_value = { + "code": "801074.SI", + "name": "工业金属", + "change": 1.5, + "pct_change": 1.5, + "quote_date": "20260908", + "quote_time": "2026-09-08T15:00:00+08:00", + "source": "eastmoney_sw", + } + snapshot = client.sw_sector_snapshot( + "000737.SZ", "20260908", allow_realtime_close=True + ) + self.assertEqual(snapshot["quote_count"], 2) + self.assertEqual(snapshot["member_count"], 2) + self.assertTrue(snapshot["inner_precise"]) + self.assertTrue(snapshot["outer_precise"]) + self.assertEqual(snapshot["inner_source"], "tushare_member_daily") + self.assertEqual(snapshot["change"], 1.5) + self.assertNotIn("权限", snapshot.get("outer_error") or "") + self.assertNotIn("rt_sw_k", snapshot.get("outer_error") or "") diff --git a/xiaobai-datahub/datahub/adapters/eastmoney.py b/xiaobai-datahub/datahub/adapters/eastmoney.py index fa52155..63c8f04 100644 --- a/xiaobai-datahub/datahub/adapters/eastmoney.py +++ b/xiaobai-datahub/datahub/adapters/eastmoney.py @@ -126,7 +126,7 @@ class EastmoneyAdapter(MarketAdapter): return result def fetch_quotes(self, codes: list[str]) -> list[dict[str, Any]]: - # Eastmoney clist does not accept arbitrary code lists well; use ulist.np for batches. + # Eastmoney ulist.np accepts ~60 secids per request; page remaining codes. secids = [] for code in codes: ts = str(code or "").upper() @@ -137,47 +137,59 @@ class EastmoneyAdapter(MarketAdapter): secids.append(f"0.{symbol}") if not secids: return [] - payload = self._get_json( - EASTMONEY_INDEX_URL, - { - "secids": ",".join(secids[:60]), - "fltt": "2", - "invt": "2", - "fields": "f12,f14,f2,f3,f4,f15,f16,f17,f18,f5,f6,f8,f124", - }, - referer="https://quote.eastmoney.com/", - ) - rows = list((payload.get("data") or {}).get("diff") or []) - result = [] - for row in rows: - symbol = str(row.get("f12") or "") - if not symbol: - continue - ts_code = f"{symbol}.SH" if symbol.startswith(("5", "6", "9")) else f"{symbol}.SZ" - epoch = int(finite_number(row.get("f124")) or 0) - result.append( + result: list[dict[str, Any]] = [] + for index in range(0, len(secids), 60): + payload = self._get_json( + EASTMONEY_INDEX_URL, { - "ts_code": ts_code, - "name": row.get("f14") or symbol, - "price": round4(finite_number(row.get("f2"))), - "pct_chg": round4(finite_number(row.get("f3"))), - "change_amount": round4(finite_number(row.get("f4"))), - "open": round4(finite_number(row.get("f17"))), - "high": round4(finite_number(row.get("f15"))), - "low": round4(finite_number(row.get("f16"))), - "previous_close": round4(finite_number(row.get("f18"))), - "volume": round4(finite_number(row.get("f5"))), - "amount": round4(finite_number(row.get("f6"))), - "turnover_rate": round4(finite_number(row.get("f8"))), - "quote_time_epoch": epoch, - "quote_time": ( - datetime.fromtimestamp(epoch).astimezone().isoformat(timespec="seconds") - if epoch - else "" - ), - "source": "eastmoney_push2", - } + "secids": ",".join(secids[index:index + 60]), + "fltt": "2", + "invt": "2", + "fields": "f12,f14,f2,f3,f4,f15,f16,f17,f18,f5,f6,f8,f124", + }, + referer="https://quote.eastmoney.com/", ) + rows = list((payload.get("data") or {}).get("diff") or []) + for row in rows: + symbol = str(row.get("f12") or "") + if not symbol: + continue + ts_code = f"{symbol}.SH" if symbol.startswith(("5", "6", "9")) else f"{symbol}.SZ" + epoch = int(finite_number(row.get("f124")) or 0) + close = round4(finite_number(row.get("f2"))) + previous = round4(finite_number(row.get("f18"))) + quote_date = ( + datetime.fromtimestamp(epoch).astimezone().strftime("%Y%m%d") + if epoch + else "" + ) + result.append( + { + "ts_code": ts_code, + "name": row.get("f14") or symbol, + "price": close, + "close": close, + "pct_chg": round4(finite_number(row.get("f3"))), + "change_amount": round4(finite_number(row.get("f4"))), + "open": round4(finite_number(row.get("f17"))), + "high": round4(finite_number(row.get("f15"))), + "low": round4(finite_number(row.get("f16"))), + "pre_close": previous, + "previous_close": previous, + "volume": round4(finite_number(row.get("f5"))), + "vol": round4(finite_number(row.get("f5")) * 100), + "amount": round4(finite_number(row.get("f6"))), + "turnover_rate": round4(finite_number(row.get("f8"))), + "quote_date": quote_date, + "quote_time_epoch": epoch, + "quote_time": ( + datetime.fromtimestamp(epoch).astimezone().isoformat(timespec="seconds") + if epoch + else "" + ), + "source": "eastmoney_push2", + } + ) return result def fetch_market_quotes(self) -> list[dict[str, Any]]: diff --git a/xiaobai-datahub/datahub/realtime_serve.py b/xiaobai-datahub/datahub/realtime_serve.py index 01257b8..17ec21e 100644 --- a/xiaobai-datahub/datahub/realtime_serve.py +++ b/xiaobai-datahub/datahub/realtime_serve.py @@ -5,6 +5,7 @@ Free sources only. Never writes official eod_* tables. Uses rt_cache + LKG. from __future__ import annotations +import hashlib import json import time from datetime import datetime @@ -20,6 +21,7 @@ from datahub.timeutil import isoformat, now_shanghai, yyyymmdd QUOTE_TTL = 60 INDEX_TTL = 60 INTRADAY_TTL = 20 +QUOTE_BATCH = 60 class RealtimeApiError(RuntimeError): @@ -95,20 +97,25 @@ def fetch_quotes(db: HubDB, codes: list[str]) -> dict[str, Any]: if not codes: return fetch_market_quotes(db) resolved: list[str] = [] - for code in codes[:60]: + seen: set[str] = set() + for code in codes: item = resolve_code(db, code) or _guess_ts_code(code) - if item: + if item and item not in seen: + seen.add(item) resolved.append(item) if not resolved: raise RealtimeApiError("INVALID_ARGUMENT", "no resolvable codes") - cache_key = "quotes:" + ",".join(sorted(resolved)) + digest = hashlib.sha1(",".join(sorted(resolved)).encode("utf-8")).hexdigest() + cache_key = f"quotes:{digest}:{len(resolved)}" cached = _read_cache(db, cache_key) if cached is not None: return cached adapter = EastmoneyAdapter() try: - rows = adapter.fetch_quotes(resolved) - source = "eastmoney:clist" + rows: list[dict[str, Any]] = [] + for index in range(0, len(resolved), QUOTE_BATCH): + rows.extend(adapter.fetch_quotes(resolved[index:index + QUOTE_BATCH])) + source = "eastmoney:ulist" except Exception as exc: raise RealtimeApiError("SOURCE_UNAVAILABLE", f"quotes unavailable: {exc}") from exc payload = _envelope( diff --git a/xiaobai-datahub/datahub/scheduler.py b/xiaobai-datahub/datahub/scheduler.py index 90e24ac..f9d67eb 100644 --- a/xiaobai-datahub/datahub/scheduler.py +++ b/xiaobai-datahub/datahub/scheduler.py @@ -110,7 +110,7 @@ class Scheduler: ("eod_b", time(15, 10)), ("eod_c", time(16, 40)), ("eod_d", time(16, 45)), - ("eod_e", time(18, 5)), + ("eod_e", time(15, 20)), ("eod_f", time(22, 40)), ("cleanup", time(0, 30)), ("backup", time(0, 40)), @@ -124,6 +124,9 @@ class Scheduler: key = (job_id, day, at.strftime("%H%M")) if key in self._fired: continue + if job_id not in self.jobs: + self._fired.add(key) + continue if job_id in {"eod_a", "eod_b", "eod_c", "eod_d", "eod_e", "eod_f", "stocks_refresh"} and not open_day: self._fired.add(key) continue @@ -203,6 +206,12 @@ class Scheduler: LOGGER.warning("eod retry failed for %s", day, exc_info=True) ran.append("eod_retry") self._settle_eod(day) + if "eod_e" in self.jobs and not self.pipeline.active_batch("sector_daily", day): + try: + self.run_job("eod_e", day) + ran.append("eod_e") + except Exception: + LOGGER.exception("sector_daily retry failed for %s", day) return ran def _settle_eod(self, day: str) -> None: diff --git a/xiaobai-datahub/tests/test_realtime_intraday.py b/xiaobai-datahub/tests/test_realtime_intraday.py index 0a4f703..9eee3d9 100644 --- a/xiaobai-datahub/tests/test_realtime_intraday.py +++ b/xiaobai-datahub/tests/test_realtime_intraday.py @@ -219,6 +219,18 @@ class MarketQuotesTests(unittest.TestCase): payload = self.api.handle("/v1/quotes/latest", {"codes": ["600000.SH"]}) mocked.return_value.fetch_market_quotes.assert_not_called() self.assertEqual(payload["data"][0]["ts_code"], "600000.SH") + self.assertEqual(payload["meta"]["source"], "eastmoney:ulist") + + def test_named_quotes_page_beyond_sixty_codes(self) -> None: + codes = [f"{index:06d}.SZ" for index in range(70)] + def fake_fetch(chunk): + return [{"ts_code": code, "close": 10, "pre_close": 9} for code in chunk] + + with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked: + mocked.return_value.fetch_quotes.side_effect = fake_fetch + payload = self.api.handle("/v1/quotes/latest", {"codes": [",".join(codes)]}) + self.assertEqual(mocked.return_value.fetch_quotes.call_count, 2) + self.assertEqual(len(payload["data"]), 70) def test_market_unavailable_stays_source_error(self) -> None: with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked: