diff --git a/backend/data/providers/tushare_daily.py b/backend/data/providers/tushare_daily.py index 83d558d..e143e10 100644 --- a/backend/data/providers/tushare_daily.py +++ b/backend/data/providers/tushare_daily.py @@ -5,6 +5,7 @@ from typing import Any from backend.data.numbers import finite_number as _number from backend.data.providers.tushare_helpers import ( _display_time, + _optional_number, _prices_equal, calendar_is_open, ) @@ -137,7 +138,67 @@ class DailyMarketMixin: ) item["capital_trade_date"] = str(capital.get("trade_date") or "") result.append(item) - return result + return self._overlay_board_fields(result, trade_date) + + def _overlay_board_fields( + self, + rows: list[dict[str, Any]], + trade_date: str, + ) -> list[dict[str, Any]]: + if not rows: + return rows + official = self._official_board_map(trade_date) + free = self._free_board_map(trade_date) if not official else {} + merged: list[dict[str, Any]] = [] + for row in rows: + code = str(row.get("ts_code") or "") + extra = official.get(code) or free.get(code) or {} + if not extra: + merged.append(row) + continue + item = dict(row) + for key in ( + "first_time", + "last_time", + "fd_amount", + "open_times", + "limit_times", + "turnover_ratio", + ): + incoming = extra.get(key) + current = item.get(key) + if incoming in (None, "", "--"): + continue + if current in (None, "", "--", 0, 0.0): + item[key] = incoming + merged.append(item) + return merged + + def _official_board_map(self, trade_date: str) -> dict[str, dict[str, Any]]: + mapped: dict[str, dict[str, Any]] = {} + try: + for row in self._load_limit_lists(trade_date): + code = str(row.get("ts_code") or "") + if code: + mapped[code] = row + except Exception: + return {} + return mapped + + def _free_board_map(self, trade_date: str) -> dict[str, dict[str, Any]]: + aggregator = getattr(self, "realtime_aggregator", None) + loader = getattr(aggregator, "eastmoney_limit_pool", None) if aggregator else None + if not callable(loader): + return {} + try: + rows = loader(trade_date) + except Exception: + return {} + return { + str(row.get("ts_code") or ""): row + for row in rows + if row.get("ts_code") + } @staticmethod def _normalize_limit(row: dict[str, Any], status: str) -> dict[str, Any]: @@ -162,7 +223,11 @@ class DailyMarketMixin: "turnover_source": row.get("turnover_source") or "provider", "capital_trade_date": row.get("capital_trade_date") or "", "amount_billion": round(amount_billion, 2), - "seal_amount_million": round(_number(row.get("fd_amount")) / 10000, 0), + "seal_amount_million": ( + round(fd / 10000, 0) + if (fd := _optional_number(row.get("fd_amount"))) is not None + else None + ), "float_mv_billion": round(_number(row.get("float_mv")) / 100000000, 1), "status": status, } diff --git a/backend/data/providers/tushare_dashboard.py b/backend/data/providers/tushare_dashboard.py index 6599153..70f54d8 100644 --- a/backend/data/providers/tushare_dashboard.py +++ b/backend/data/providers/tushare_dashboard.py @@ -365,10 +365,9 @@ class DashboardMixin: ts_code: str, reference_date: str = "", ) -> dict[str, Any]: - rows = self.query("rt_k", {"ts_code": ts_code}) - if not rows: + row = self._realtime_quote_row(ts_code, reference_date) + if not row: raise TushareError(f"No realtime quote returned for {ts_code}") - row = rows[0] close = _number(row.get("close")) previous_close = _number(row.get("pre_close")) if close <= 0 or previous_close <= 0: @@ -452,10 +451,39 @@ class DashboardMixin: "float_share_10k": float_share, "capital_trade_date": str(capital.get("trade_date") or ""), "turnover_source": "rt_volume/latest_float_share" if float_share else "unavailable", - "data_source": "tushare", + "data_source": str(row.get("source") or "tushare"), "realtime": True, } + def _realtime_quote_row(self, ts_code: str, reference_date: str = "") -> dict[str, Any]: + hub = getattr(self, "try_quotes", None) + if callable(hub): + rows = hub([ts_code]) or [] + if rows: + return dict(rows[0]) + try: + rows = self.query("rt_k", {"ts_code": ts_code}) + if rows: + return dict(rows[0]) + except TushareError: + pass + aggregator = getattr(self, "realtime_aggregator", None) + if aggregator is None: + return {} + for loader in ( + getattr(aggregator, "eastmoney_stock_quote", None), + getattr(aggregator, "tencent_stock_quote", None), + ): + if not callable(loader): + continue + try: + quote = loader(ts_code, expected_date=reference_date) + except Exception: + continue + if quote: + return dict(quote) + return {} + def _stock_activity_metrics( self, ts_code: str, @@ -573,7 +601,7 @@ class DashboardMixin: for row in reference.get("basic_rows") or [] if row.get("ts_code") ] - quotes = self.query("rt_k", {"ts_code": ",".join(codes)}, "") + quotes, quote_source = self._load_realtime_quotes(",".join(codes), trade_date) rows = [ row for row in quotes if _number(row.get("close")) > 0 and _number(row.get("pre_close")) > 0 diff --git a/backend/data/providers/tushare_helpers.py b/backend/data/providers/tushare_helpers.py index 1a13a80..14d0430 100644 --- a/backend/data/providers/tushare_helpers.py +++ b/backend/data/providers/tushare_helpers.py @@ -23,6 +23,55 @@ def _text(value: Any) -> str: return str(value or "").strip() +def _optional_number(value: Any) -> float | None: + if value in (None, "", "-"): + return None + number = _number(value, default=float("nan")) + if number != number: + return None + return number + + +def _moneyflow_payload(flow: dict[str, Any] | None) -> dict[str, Any]: + if not flow: + return { + "available": False, + "net_million": None, + "large_million": None, + "medium_million": None, + "small_million": None, + } + net = _optional_number(flow.get("net_mf_amount")) + buy_lg = _optional_number(flow.get("buy_lg_amount")) + sell_lg = _optional_number(flow.get("sell_lg_amount")) + buy_elg = _optional_number(flow.get("buy_elg_amount")) + sell_elg = _optional_number(flow.get("sell_elg_amount")) + buy_md = _optional_number(flow.get("buy_md_amount")) + sell_md = _optional_number(flow.get("sell_md_amount")) + buy_sm = _optional_number(flow.get("buy_sm_amount")) + sell_sm = _optional_number(flow.get("sell_sm_amount")) + large = None + if None not in (buy_lg, sell_lg, buy_elg, sell_elg): + large = (buy_lg + buy_elg - sell_lg - sell_elg) + elif _optional_number(flow.get("large_amount")) is not None: + large = _optional_number(flow.get("large_amount")) + medium = None if None in (buy_md, sell_md) else (buy_md - sell_md) + if medium is None: + medium = _optional_number(flow.get("medium_amount")) + small = None if None in (buy_sm, sell_sm) else (buy_sm - sell_sm) + if small is None: + small = _optional_number(flow.get("small_amount")) + if net is None and large is None and medium is None and small is None: + return _moneyflow_payload(None) + return { + "available": True, + "net_million": None if net is None else round(net / 100, 2), + "large_million": None if large is None else round(large / 100, 2), + "medium_million": None if medium is None else round(medium / 100, 2), + "small_million": None if small is None else round(small / 100, 2), + } + + def _prices_equal(left: Any, right: Any) -> bool: if left is None or right is None: return False diff --git a/backend/data/providers/tushare_industries.py b/backend/data/providers/tushare_industries.py index c75f970..d3bbbc1 100644 --- a/backend/data/providers/tushare_industries.py +++ b/backend/data/providers/tushare_industries.py @@ -311,37 +311,37 @@ class ShenwanIndustryMixin: finalized: bool = False, ) -> dict[str, Any]: sector_code = str(industry.get("l2_code") or "") - sw_rows = self.query( - "rt_sw_k", - {"ts_code": sector_code}, - "ts_code,name,trade_time,close,pre_close,high,open,low,vol,amount,pct_change", + sw_row, outer_source, outer_error = self._sw_outer_realtime( + sector_code, + str(industry.get("l2_name") or ""), + trade_date, + finalized=finalized, ) - sw_row = sw_rows[0] if sw_rows else {} - trade_time = str(sw_row.get("trade_time") or "") - quote_date = trade_time[:10].replace("-", "") - quote_clock = trade_time[11:19] if len(trade_time) >= 19 else "" + trade_time = str(sw_row.get("trade_time") or sw_row.get("quote_time") or "") + quote_date = str(sw_row.get("quote_date") or trade_time[:10].replace("-", "")) + quote_clock = trade_time[11:19] if len(trade_time) >= 19 else str(sw_row.get("quote_clock") or "") outer_precise = bool(sw_row and quote_date == trade_date) - if finalized and (not quote_clock or quote_clock < "15:00:00"): + if finalized and quote_clock and quote_clock < "15:00:00": outer_precise = False - official_change = _number(sw_row.get("pct_change")) + official_change = _number(sw_row.get("pct_change") if sw_row.get("pct_change") not in (None, "") else sw_row.get("change")) if not official_change: - close = _number(sw_row.get("close")) - pre_close = _number(sw_row.get("pre_close")) + close = _number(sw_row.get("close") if sw_row.get("close") not in (None, "") else sw_row.get("price")) + pre_close = _number(sw_row.get("pre_close") if sw_row.get("pre_close") not in (None, "") else sw_row.get("previous_close")) official_change = (close / pre_close - 1) * 100 if close and pre_close else 0 if not outer_precise: official_change = None - outer_error = "" - if not sw_row: - outer_error = f"No Shenwan realtime index returned for {sector_code}" - elif quote_date != trade_date: - outer_error = f"Shenwan realtime index date is {quote_date or 'unknown'}, expected {trade_date}" - elif finalized and (not quote_clock or quote_clock < "15:00:00"): - outer_error = f"Shenwan realtime index is not a close snapshot ({trade_time})" + if not sw_row and not outer_error: + outer_error = f"申万行业 {sector_code} 当日外显待盘后正式数据或免费实时源" + elif quote_date and quote_date != trade_date: + outer_error = f"申万实时行业日期是 {quote_date}，期望 {trade_date}" + elif finalized and quote_clock and quote_clock < "15:00:00": + outer_error = f"申万行业尚未形成收盘快照（{trade_time}）" valid: list[dict[str, Any]] = [] codes: list[str] = [] reference: dict[str, Any] = {} inner_error = "" + inner_source = "unavailable" try: reference = self._load_realtime_reference(trade_date, previous_trade_date) active_codes = { @@ -354,18 +354,21 @@ class ShenwanIndustryMixin: for row in members if str(row.get("ts_code") or "") in active_codes ] - if codes: - quotes = self.query("rt_k", {"ts_code": ",".join(codes)}, "") - for row in quotes: - close = _number(row.get("close")) - previous_close = _number(row.get("pre_close")) - if close <= 0 or previous_close <= 0: - continue - valid.append({**row, "change": (close / previous_close - 1) * 100}) - else: + quotes, inner_source = self._load_member_realtime_quotes(codes, trade_date) + for row in quotes: + close = _number(row.get("close")) + previous_close = _number(row.get("pre_close")) + if close <= 0 or previous_close <= 0: + continue + valid.append({**row, "change": (close / previous_close - 1) * 100}) + if not codes: inner_error = f"No active Shenwan members returned for {sector_code}" + elif not quotes: + inner_error = f"申万成分实时行情暂不可用：{sector_code}" except TushareError as exc: inner_error = str(exc) + if "rt_k" in inner_error or "权限" in inner_error: + inner_error = "申万成分实时行情暂不可用，已避开无权限接口" coverage = len(valid) / max(len(codes), 1) * 100 valid_codes = {str(item.get("ts_code") or "") for item in valid} @@ -390,16 +393,16 @@ class ShenwanIndustryMixin: } equal_change = sum(item["change"] for item in valid) / len(valid) if valid else 0 amount_billion = sum(_number(item.get("amount")) for item in valid) / 100000000 + market_rows: list[dict[str, Any]] = [] try: - self._ensure_realtime_market_cache(trade_date) - with self._realtime_reference_lock: - market_rows = list( - (self._latest_realtime_market.get(trade_date) or {}).get("rows") or [] - ) + market_rows = self._ensure_realtime_market_cache(trade_date) except TushareError as exc: - market_rows = [] inner_precise = False - inner_error = inner_error or str(exc) + message = str(exc) + if "rt_k" in message or "权限" in message: + inner_error = inner_error or "全市场实时行情暂不可用，已避开无权限接口" + else: + inner_error = inner_error or message capital_map = { str(item.get("ts_code") or ""): item for item in reference.get("capital_rows") or [] @@ -408,14 +411,17 @@ class ShenwanIndustryMixin: for item in valid: capital = capital_map.get(str(item.get("ts_code") or ""), {}) float_share = _number(capital.get("float_share")) - if float_share: - sector_turnovers.append(_number(item.get("vol")) / float_share / 100) + volume = _number(item.get("vol")) + if float_share and volume: + # 免费源成交量为股；daily_basic.float_share 为万股。 + sector_turnovers.append(volume / float_share / 100) market_turnovers = [] for item in market_rows: capital = capital_map.get(str(item.get("ts_code") or ""), {}) float_share = _number(capital.get("float_share")) - if float_share: - market_turnovers.append(_number(item.get("vol")) / float_share / 100) + volume = _number(item.get("vol")) + if float_share and volume: + market_turnovers.append(volume / float_share / 100) average_turnover = sum(sector_turnovers) / len(sector_turnovers) if sector_turnovers else 0 market_turnover = sum(market_turnovers) / len(market_turnovers) if market_turnovers else 0 relative_turnover = average_turnover / market_turnover if market_turnover else 0 @@ -447,9 +453,9 @@ class ShenwanIndustryMixin: "amount_billion": round(amount_billion, 2), "count": sum(item["change"] >= 9.5 for item in valid), "max_streak": 0, - "source": "tushare_rt_sw_k+sw_members_rt_k", - "inner_source": "tushare_sw_members+rt_k", - "outer_source": "tushare_rt_sw_k", + "source": f"{outer_source or 'unavailable'}+{inner_source}", + "inner_source": inner_source, + "outer_source": outer_source or "unavailable", "taxonomy": "sw_l2", "industry": industry, "trade_date": trade_date, @@ -464,9 +470,68 @@ class ShenwanIndustryMixin: "inner_error": inner_error, "outer_error": outer_error, "schema_version": 6, - "methodology": "外显使用申万官方 rt_sw_k；内核独立使用申万成分 rt_k 宽度与相对换手聚合", + "methodology": "外显使用已发布 sw_daily 或免费申万实时；内核使用数据中枢/免费实时成分，不调用 rt_sw_k", } + def _sw_outer_realtime( + self, + sector_code: str, + sector_name: str, + trade_date: str, + finalized: bool = False, + ) -> tuple[dict[str, Any], str, str]: + aggregator = getattr(self, "realtime_aggregator", None) + loader = getattr(aggregator, "eastmoney_shenwan_quote", None) if aggregator else None + if callable(loader): + try: + row = loader(sector_code, expected_date="" if finalized else trade_date) + except Exception as exc: + message = str(exc) + if finalized: + return {}, "", f"申万行业 {sector_code} 盘后正式数据待入库" + return {}, "", f"免费申万实时暂不可用：{message[:180]}" + if row: + return dict(row), str(row.get("source") or "eastmoney_sw"), "" + if finalized: + return {}, "", f"申万行业 {sector_code} 当日盘后正式数据尚未入库" + if aggregator and sector_name: + try: + row = aggregator.eastmoney_sector(sector_name) + except Exception as exc: + return {}, "", f"免费行业实时暂不可用：{str(exc)[:180]}" + if row: + return dict(row), str(row.get("source") or "eastmoney_sector"), "" + return {}, "", f"申万行业 {sector_code} 当日外显待补充" + + def _load_member_realtime_quotes( + self, + codes: list[str], + trade_date: str, + ) -> tuple[list[dict[str, Any]], str]: + if not codes: + return [], "unavailable" + hub = getattr(self, "try_quotes", None) + if callable(hub): + rows = hub(codes) or [] + if rows: + return list(rows), "datahub" + 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 + def _filter_members_by_listing( members: list[dict[str, Any]], diff --git a/backend/data/providers/tushare_stocks.py b/backend/data/providers/tushare_stocks.py index 0d0434e..136a0f6 100644 --- a/backend/data/providers/tushare_stocks.py +++ b/backend/data/providers/tushare_stocks.py @@ -5,13 +5,14 @@ from typing import Any from backend.bootstrap.config import display_compact_date as _display_date from backend.data.numbers import finite_number as _number +from backend.data.providers.tushare_helpers import _moneyflow_payload class StockMixin: def stock_detail(self, ts_code: str, requested_date: str) -> dict[str, Any]: trade_date, _ = self.resolve_trade_context(requested_date) end = datetime.strptime(trade_date, "%Y%m%d") - start_date = (end - timedelta(days=190)).strftime("%Y%m%d") + start_date = (end - timedelta(days=400)).strftime("%Y%m%d") daily = self.query( "daily", {"ts_code": ts_code, "start_date": start_date, "end_date": trade_date}, @@ -41,7 +42,7 @@ class StockMixin: factor_map = {row["trade_date"]: _number(row.get("adj_factor"), 1) for row in factors} latest_factor = max(factor_map.values(), default=1) or 1 prices = [] - for row in sorted(daily, key=lambda item: item.get("trade_date", ""))[-90:]: + for row in sorted(daily, key=lambda item: item.get("trade_date", ""))[-250:]: factor = factor_map.get(row.get("trade_date"), latest_factor) ratio = factor / latest_factor prices.append( @@ -56,7 +57,7 @@ class StockMixin: "amount_billion": round(_number(row.get("amount")) / 100000, 2), } ) - flow = moneyflow[0] if moneyflow else {} + flow = moneyflow[0] if moneyflow else None basic = basics[0] if basics else {} daily_basic = daily_basics[0] if daily_basics else {} latest = prices[-1] if prices else {} @@ -87,22 +88,7 @@ class StockMixin: "amount_billion": latest.get("amount_billion", 0), }, "prices": prices, - "moneyflow": { - "net_million": round(_number(flow.get("net_mf_amount")) / 100, 2), - "large_million": round( - (_number(flow.get("buy_lg_amount")) + _number(flow.get("buy_elg_amount")) - - _number(flow.get("sell_lg_amount")) - _number(flow.get("sell_elg_amount"))) / 100, - 2, - ), - "medium_million": round( - (_number(flow.get("buy_md_amount")) - _number(flow.get("sell_md_amount"))) / 100, - 2, - ), - "small_million": round( - (_number(flow.get("buy_sm_amount")) - _number(flow.get("sell_sm_amount"))) / 100, - 2, - ), - }, + "moneyflow": _moneyflow_payload(flow), } def stock_intraday(self, ts_code: str, requested_date: str) -> dict[str, Any]: diff --git a/backend/data/providers/tushare_transport.py b/backend/data/providers/tushare_transport.py index 30e602e..f6bd6ef 100644 --- a/backend/data/providers/tushare_transport.py +++ b/backend/data/providers/tushare_transport.py @@ -20,6 +20,8 @@ class TushareTransportMixin: params: dict[str, Any] | None = None, fields: str = "", ) -> list[dict[str, Any]]: + if api_name == "rt_sw_k": + raise TushareError("rt_sw_k is disabled; use published sw_daily or free Shenwan realtime") payload = json.dumps( { "api_name": api_name, diff --git a/backend/data/realtime.py b/backend/data/realtime.py index 3c7d533..c935747 100644 --- a/backend/data/realtime.py +++ b/backend/data/realtime.py @@ -20,8 +20,10 @@ class RealtimeAggregateError(RuntimeError): EASTMONEY_INDEX_URL = "https://push2.eastmoney.com/api/qt/ulist.np/get" EASTMONEY_STOCK_URL = "https://push2.eastmoney.com/api/qt/stock/get" -EASTMONEY_STOCK_FIELDS = "f43,f44,f45,f46,f47,f48,f57,f58,f60,f86,f168" +EASTMONEY_STOCK_FIELDS = "f43,f44,f45,f46,f47,f48,f57,f58,f60,f86,f168,f62,f66,f72,f78,f84" EASTMONEY_SECTOR_URL = "https://push2.eastmoney.com/api/qt/clist/get" +EASTMONEY_ZT_POOL_URL = "https://push2ex.eastmoney.com/getTopicZTPool" +EASTMONEY_ZB_POOL_URL = "https://push2ex.eastmoney.com/getTopicZBPool" EASTMONEY_A_SHARE_BOARDS = ( "m:0+t:6", "m:0+t:80", @@ -321,6 +323,127 @@ class WebRealtimeAggregator: raise RealtimeAggregateError(f"Eastmoney stock quote unavailable for {ts_code}") return _require_quote_date(quote, expected_date) + def eastmoney_stock_quotes( + self, + codes: list[str], + expected_date: str = "", + ) -> list[dict[str, Any]]: + secids = [] + for code in codes: + try: + _symbol, secid, _ts = _a_share_identity(code) + except RealtimeAggregateError: + continue + secids.append(secid) + quotes: list[dict[str, Any]] = [] + for index in range(0, len(secids), 60): + payload = self._get_json( + EASTMONEY_INDEX_URL, + { + "secids": ",".join(secids[index:index + 60]), + "fltt": "2", + "invt": "2", + "fields": EASTMONEY_QUOTE_FIELDS, + }, + referer="https://quote.eastmoney.com/", + ) + for row in _diff_rows(payload.get("data") or {}): + quote = _normalize_eastmoney_quote(row) + if quote: + quotes.append(quote) + return self._filter_quotes_by_date(quotes, expected_date) + + def eastmoney_shenwan_quote( + self, + ts_code: str, + expected_date: str = "", + ) -> dict[str, Any]: + code = str(ts_code or "").split(".")[0] + if not code: + raise RealtimeAggregateError("Invalid Shenwan code") + payload = self._get_json( + EASTMONEY_INDEX_URL, + { + "secids": f"90.{code}", + "fltt": "2", + "invt": "2", + "fields": "f12,f14,f2,f3,f4,f15,f16,f17,f18,f6,f8,f104,f105,f128,f136,f140,f124", + }, + referer="https://quote.eastmoney.com/", + ) + row = next((item for item in _diff_rows(payload.get("data") or {}) if item), None) + if not row: + raise RealtimeAggregateError(f"Eastmoney Shenwan quote missing for {code}") + epoch = int(_number(row.get("f124"))) + quote_time = ( + datetime.fromtimestamp(epoch).astimezone().isoformat(timespec="seconds") + if epoch + else "" + ) + close = _number(row.get("f2")) + previous = _number(row.get("f18")) + if close <= 0 or previous <= 0: + raise RealtimeAggregateError(f"Eastmoney Shenwan quote empty for {code}") + result = { + "ts_code": f"{code}.SI", + "code": f"{code}.SI", + "name": row.get("f14") or code, + "price": close, + "close": close, + "pre_close": previous, + "previous_close": previous, + "open": _number(row.get("f17")), + "high": _number(row.get("f15")), + "low": _number(row.get("f16")), + "change": _number(row.get("f3")), + "pct_change": _number(row.get("f3")), + "amount": _number(row.get("f6")), + "leader": row.get("f128") or "--", + "leader_code": row.get("f140") or "", + "leading_pct": _number(row.get("f136")), + "up_count": int(_number(row.get("f104"))), + "down_count": int(_number(row.get("f105"))), + "quote_time": quote_time, + "trade_time": quote_time, + "quote_date": datetime.fromtimestamp(epoch).astimezone().strftime("%Y%m%d") if epoch else "", + "quote_time_epoch": epoch, + "source": "eastmoney_sw", + } + return _require_quote_date(result, expected_date) if expected_date else result + + def eastmoney_limit_pool(self, trade_date: str = "") -> list[dict[str, Any]]: + day = str(trade_date or "").replace("-", "") + rows: list[dict[str, Any]] = [] + for url, limit_type in ( + (EASTMONEY_ZT_POOL_URL, "U"), + (EASTMONEY_ZB_POOL_URL, "Z"), + ): + try: + payload = self._get_json( + url, + { + "ut": "7eea3edcaed734bea9cbfc24409ed989", + "dpt": "wz.ztzt", + "PageIndex": "0", + "PageSize": "200", + "sort": "fbt:asc", + "date": day, + }, + referer="https://quote.eastmoney.com/ztb/detail", + ) + except RealtimeAggregateError: + continue + pool = (payload.get("data") or {}).get("pool") or [] + if isinstance(pool, dict): + pool = list(pool.values()) + for item in pool: + if not isinstance(item, dict): + continue + parsed = _normalize_eastmoney_limit_row(item, limit_type) + if parsed: + rows.append(parsed) + return rows + def tencent_indices(self) -> list[dict[str, Any]]: raw, cache_age = self._get_text( TENCENT_INDEX_URL, @@ -372,11 +495,17 @@ class WebRealtimeAggregator: if not matched: raise RealtimeAggregateError(f"Eastmoney sector not found: {query}") epoch = int(_number(matched.get("f124"))) + quote_time = ( + datetime.fromtimestamp(epoch).astimezone().isoformat(timespec="seconds") + if epoch else "" + ) return { "code": matched.get("f12") or "", "name": matched.get("f14") or query, "price": _number(matched.get("f2")), + "close": _number(matched.get("f2")), "change": _number(matched.get("f3")), + "pct_change": _number(matched.get("f3")), "change_amount": _number(matched.get("f4")), "turnover_rate": _number(matched.get("f8")), "up_count": int(_number(matched.get("f104"))), @@ -385,10 +514,9 @@ class WebRealtimeAggregator: "leader_code": matched.get("f140") or "", "leading_pct": _number(matched.get("f136")), "quote_time_epoch": epoch, - "quote_time": ( - datetime.fromtimestamp(epoch).astimezone().isoformat(timespec="seconds") - if epoch else "" - ), + "quote_time": quote_time, + "trade_time": quote_time, + "quote_date": datetime.fromtimestamp(epoch).astimezone().strftime("%Y%m%d") if epoch else "", "source": "eastmoney_push2", "match_query": query, } @@ -636,6 +764,10 @@ def _normalize_eastmoney_stock_quote( "quote_date": quote_date, "quote_time_epoch": epoch, "turnover_rate": _number(row.get("f168")), + "net_mf_amount": _eastmoney_flow_wan(row.get("f62")), + "large_amount": _eastmoney_flow_wan(row.get("f62")), + "medium_amount": _eastmoney_flow_wan(row.get("f78")), + "small_amount": _eastmoney_flow_wan(row.get("f84")), "source": "eastmoney_stock", } @@ -721,6 +853,57 @@ def _normalize_eastmoney_quote(row: dict[str, Any]) -> dict[str, Any] | None: } +def _eastmoney_flow_wan(value: Any) -> float | None: + if value in (None, "", "-"): + return None + amount = _number(value, default=float("nan")) + if amount != amount: + return None + return amount / 10000 + + +def _board_clock(value: Any) -> str: + digits = "".join(character for character in str(value or "") if character.isdigit()) + if len(digits) >= 6: + return f"{digits[:2]}:{digits[2:4]}:{digits[4:6]}" + if len(digits) == 5: + digits = digits.zfill(6) + return f"{digits[:2]}:{digits[2:4]}:{digits[4:6]}" + if len(digits) == 4: + return f"{digits[:2]}:{digits[2:]}:00" + return "" + + +def _normalize_eastmoney_limit_row(row: dict[str, Any], limit_type: str) -> dict[str, Any] | None: + symbol = str(row.get("c") or row.get("code") or "").strip() + if not symbol.isdigit() or len(symbol) != 6: + return None + market = int(_number(row.get("m") if row.get("m") not in (None, "") else row.get("market"))) + 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" + first_time = _board_clock(row.get("fbt") if row.get("fbt") not in (None, "") else row.get("first_time")) + last_time = _board_clock(row.get("lbt") if row.get("lbt") not in (None, "") else row.get("last_time")) + fund = row.get("fund") + if fund in (None, ""): + fund = row.get("fd_amount") + return { + "ts_code": ts_code, + "name": row.get("n") or row.get("name") or symbol, + "limit_type": limit_type, + "first_time": first_time or None, + "last_time": last_time or None, + "open_times": int(_number(row.get("zbc") if row.get("zbc") not in (None, "") else row.get("open_times"))), + "limit_times": max(1, int(_number(row.get("lbc") if row.get("lbc") not in (None, "") else 1))), + "turnover_ratio": _number(row.get("hs") if row.get("hs") not in (None, "") else row.get("turnover_ratio")), + "fd_amount": _number(fund) if fund not in (None, "", "-") else None, + "source": "eastmoney_zt_pool", + } + + def _normalize_sector(value: Any) -> str: text = str(value or "").strip().replace(" ", "") for suffix in ("板块", "概念", "行业", "Ⅱ", "Ⅲ", "(A股)", "（A股）"): diff --git a/backend/features/heaven/market_context.py b/backend/features/heaven/market_context.py index a18c47b..30f68e9 100644 --- a/backend/features/heaven/market_context.py +++ b/backend/features/heaven/market_context.py @@ -283,9 +283,9 @@ class HeavenMarketContextMixin: ) -> dict[str, Any] | None: """Return the Shenwan L2 sector context for heaven trend. - 观势行业层只使用申万二级行业。外显盘中使用 rt_sw_k、历史使用 - sw_daily；内核独立使用目标日期成分股行情聚合。收盘过渡期在 - sw_daily 入库前接受同日15:00后的 rt_sw_k 收盘快照。 + 观势行业层只使用申万二级行业。外显优先使用已发布的 sw_daily， + 盘中及收盘过渡期使用免费申万实时行情；内核使用数据中枢或免费 + 实时成分行情。不再调用无权限的 rt_sw_k / rt_k。 """ cache_key = f"{trade_date}:{identifier.strip().lower()}" cached = self.database.get_data_snapshot("heaven_sector", cache_key) diff --git a/backend/features/market/charts.py b/backend/features/market/charts.py index f3db1d0..72adb41 100644 --- a/backend/features/market/charts.py +++ b/backend/features/market/charts.py @@ -23,6 +23,9 @@ class ChartDataError(RuntimeError): pass +DAILY_CHART_LIMIT = 250 + + TRENDS_URL = "https://push2delay.eastmoney.com/api/qt/stock/trends2/get" HIS_TRENDS_URL = "https://push2his.eastmoney.com/api/qt/stock/trends2/get" BOARD_LIST_URL = "https://push2delay.eastmoney.com/api/qt/clist/get" @@ -64,7 +67,7 @@ class MarketChartClient: except (IfindError, ChartDataError): return self.fallback.stock_intraday(normalized) - def stock_daily(self, code: str, end_date: str, limit: int = 90) -> list[dict[str, Any]]: + def stock_daily(self, code: str, end_date: str, limit: int = DAILY_CHART_LIMIT) -> list[dict[str, Any]]: normalized = str(code or "").strip() if not re.fullmatch(r"\d{6}", normalized): raise ChartDataError("Invalid stock code") @@ -73,7 +76,7 @@ class MarketChartClient: return hub_rows return self._ifind_daily(_stock_market_code(normalized), end_date, limit) - def index_daily(self, identifier: str, end_date: str, limit: int = 90) -> list[dict[str, Any]]: + def index_daily(self, identifier: str, end_date: str, limit: int = DAILY_CHART_LIMIT) -> list[dict[str, Any]]: normalized = str(identifier or "").strip().upper() if normalized not in INDEX_SECIDS: raise ChartDataError("Unsupported index") diff --git a/backend/features/market/service.py b/backend/features/market/service.py index 23041d2..1950f83 100644 --- a/backend/features/market/service.py +++ b/backend/features/market/service.py @@ -15,6 +15,7 @@ from backend.bootstrap.config import ( ) from backend.data.providers.ifind_client import IfindError from backend.data.providers.tushare_client import TushareClient, TushareError +from backend.data.providers.tushare_helpers import _moneyflow_payload, _optional_number from backend.data.realtime import RealtimeAggregateError from backend.features.market.backfill_history import ( DEFAULT_RECENT_TRADING_DAYS, @@ -27,7 +28,7 @@ from backend.features.market.backfill_history import ( select_open_trade_dates, select_open_trade_dates_in_range, ) -from backend.features.market.charts import ChartDataError +from backend.features.market.charts import ChartDataError, DAILY_CHART_LIMIT from backend.features.market.insights import MarketInsightsService from backend.features.sentiment.engine import SENTIMENT_ENGINE_VERSION @@ -677,7 +678,7 @@ class MarketServiceMixin: "index_daily", { "ts_code": basic["id"], - "start_date": (end - timedelta(days=190)).strftime("%Y%m%d"), + "start_date": (end - timedelta(days=400)).strftime("%Y%m%d"), "end_date": resolved_date, }, "ts_code,trade_date,open,high,low,close,pct_chg,vol,amount", @@ -693,10 +694,10 @@ class MarketServiceMixin: "change": float(row.get("pct_chg") or 0), "volume": float(row.get("vol") or 0), } - for row in rows[-90:] + for row in rows[-DAILY_CHART_LIMIT:] ] try: - chart_series = self.chart_data.index_daily(str(basic["id"]), resolved_date, 90) + chart_series = self.chart_data.index_daily(str(basic["id"]), resolved_date, DAILY_CHART_LIMIT) if chart_series: series = chart_series except (AttributeError, ChartDataError): @@ -804,7 +805,7 @@ class MarketServiceMixin: result = copy.deepcopy(payload) now = datetime.now().astimezone() try: - result["prices"] = self.chart_data.stock_daily(code, requested_date, 90) + result["prices"] = self.chart_data.stock_daily(code, requested_date, DAILY_CHART_LIMIT) result["meta"] = {**(result.get("meta") or {}), "chart_source": "market_chart"} except (AttributeError, ChartDataError): pass @@ -837,7 +838,7 @@ class MarketServiceMixin: **(result.get("meta") or {}), "notice": TODAY_DAILY_UNAVAILABLE_NOTICE, } - return self._enrich_stock_detail(result) + return self._enrich_stock_detail(result, requested_date) @staticmethod def _sanitize_stock_detail_prices( @@ -1012,7 +1013,7 @@ class MarketServiceMixin: else: quote_date = str(row.get("quote_date") or today) quote_time = f"{quote_date[:4]}-{quote_date[4:6]}-{quote_date[6:]}" - return { + quote = { "name": str(row.get("name") or name or "--"), "sector": sector, "price": price, @@ -1025,6 +1026,10 @@ class MarketServiceMixin: "turnover_rate": float(row.get("turnover_rate") or 0), "quote_time": quote_time, } + flow = _moneyflow_payload(row) + if flow.get("available"): + quote["moneyflow"] = flow + return quote def _intraday_realtime_stock_quote( self, code: str, today: str, payload: dict[str, Any] @@ -1100,19 +1105,24 @@ class MarketServiceMixin: prices[-1] = realtime_bar else: prices.append(realtime_bar) - payload["prices"] = prices[-90:] + payload["prices"] = prices[-DAILY_CHART_LIMIT:] stock = dict(payload.get("stock") or {}) - stock.update( - { - "name": quote["name"], - "industry": quote["sector"], - "price": quote["price"], - "change": quote["change"], - "amount_billion": quote["amount_billion"], - "turnover_rate": quote["turnover_rate"], - } - ) + updates = { + "name": quote["name"], + "industry": quote["sector"], + "price": quote["price"], + "change": quote["change"], + "amount_billion": quote["amount_billion"], + } + quote_turnover = _optional_number(quote.get("turnover_rate")) + if quote_turnover: + updates["turnover_rate"] = quote_turnover + stock.update(updates) payload["stock"] = stock + quote_flow = quote.get("moneyflow") + current_flow = payload.get("moneyflow") or {} + if isinstance(quote_flow, dict) and quote_flow.get("available") and not current_flow.get("available"): + payload["moneyflow"] = quote_flow payload["meta"] = { **(payload.get("meta") or {}), "trade_date": display_date, @@ -1403,10 +1413,40 @@ class MarketServiceMixin: return item["name"], item["sector"] or "其他" return "--", "其他" - def _enrich_stock_detail(self, payload: dict[str, Any]) -> dict[str, Any]: + def _enrich_stock_detail( + self, payload: dict[str, Any], trade_date: str = "" + ) -> dict[str, Any]: result = dict(payload) stock = dict(payload.get("stock") or {}) code = str(stock.get("code") or "") + compact_date = normalize_date( + str((payload.get("meta") or {}).get("trade_date") or trade_date) + ) + board = self._limit_event_for_stock(code, compact_date) + if board: + if not stock.get("first_time") or stock.get("first_time") == "--": + stock["first_time"] = board.get("first_time") or "--" + if not stock.get("last_time") or stock.get("last_time") == "--": + stock["last_time"] = board.get("last_time") or "--" + if not stock.get("open_times"): + stock["open_times"] = board.get("open_times") or 0 + if _optional_number(stock.get("seal_amount_million")) is None: + stock["seal_amount_million"] = board.get("seal_amount_million") + if not _optional_number(stock.get("turnover_rate")) and _optional_number(board.get("turnover_rate")): + stock["turnover_rate"] = board.get("turnover_rate") + flow = result.get("moneyflow") or {} + if not flow.get("available"): + live_flow = self._live_moneyflow_for_stock(code, compact_date) + if live_flow.get("available"): + result["moneyflow"] = live_flow + else: + result["moneyflow"] = { + "available": False, + "net_million": None, + "large_million": None, + "medium_million": None, + "small_million": None, + } watched = { item["code"]: item for item in self.database.list_watchlist(self.current_user_id) @@ -1416,6 +1456,45 @@ class MarketServiceMixin: result["notes"] = self.database.list_notes(self.current_user_id, code=code) return result + def _limit_event_for_stock(self, code: str, trade_date: str) -> dict[str, Any]: + if not code or not trade_date: + return {} + ts_code = tushare_code(code) + client = self._tushare_client() if self.configured else None + rows: list[dict[str, Any]] = [] + if client is not None: + try: + rows = client._load_limit_type(trade_date, "U") + client._load_limit_type(trade_date, "Z") + except Exception: + rows = [] + if not rows: + try: + rows = list((client._free_board_map(trade_date) or {}).values()) + except Exception: + rows = [] + match = next((row for row in rows if str(row.get("ts_code") or "") == ts_code), None) + if not match: + return {} + fd = _optional_number(match.get("fd_amount")) + return { + "first_time": match.get("first_time") or "--", + "last_time": match.get("last_time") or "--", + "open_times": match.get("open_times") or 0, + "seal_amount_million": None if fd is None else round(fd / 10000, 0), + "turnover_rate": _optional_number(match.get("turnover_ratio")), + } + + def _live_moneyflow_for_stock(self, code: str, trade_date: str) -> dict[str, Any]: + aggregator = getattr(self, "realtime_aggregator", None) + loader = getattr(aggregator, "eastmoney_stock_quote", None) if aggregator else None + if not callable(loader) or not code: + return _moneyflow_payload(None) + try: + quote = loader(tushare_code(code), expected_date=trade_date) + except Exception: + return _moneyflow_payload(None) + return _moneyflow_payload(quote) + def _with_storage(self, dashboard: dict[str, Any], cached: bool) -> dict[str, Any]: result = dict(dashboard) result["meta"] = { diff --git a/config/architecture-inventory.json b/config/architecture-inventory.json index f5d3dc9..5e12cd2 100644 --- a/config/architecture-inventory.json +++ b/config/architecture-inventory.json @@ -468,8 +468,8 @@ }, { "path": "frontend/pages/heaven/page.js", - "bytes": 97268, - "lines": 2070 + "bytes": 97770, + "lines": 2079 }, { "path": "frontend/shared/shell.css", @@ -498,8 +498,8 @@ }, { "path": "backend/data/providers/tushare_dashboard.py", - "bytes": 33603, - "lines": 784 + "bytes": 34631, + "lines": 812 }, { "path": "database.py", @@ -513,8 +513,8 @@ }, { "path": "backend/data/providers/tushare_industries.py", - "bytes": 26540, - "lines": 616 + "bytes": 29886, + "lines": 681 }, { "path": "backend/features/heaven/manual.py", @@ -573,7 +573,7 @@ }, { "path": "backend/features/heaven/market_context.py", - "bytes": 13681, + "bytes": 13687, "lines": 338 }, { @@ -616,6 +616,11 @@ "bytes": 9348, "lines": 222 }, + { + "path": "backend/data/providers/tushare_daily.py", + "bytes": 9170, + "lines": 233 + }, { "path": "frontend/pages/market/entity-detail.js", "bytes": 9119, @@ -641,11 +646,6 @@ "bytes": 6983, "lines": 146 }, - { - "path": "backend/data/providers/tushare_daily.py", - "bytes": 6949, - "lines": 168 - }, { "path": "backend/application.py", "bytes": 6751, @@ -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": "backend/data/providers/tushare_stocks.py", - "bytes": 6244, - "lines": 137 - }, { "path": "backend/features/screener/backtest.py", "bytes": 6202, @@ -681,16 +681,16 @@ "bytes": 6092, "lines": 138 }, - { - "path": "frontend/pages/market/stock-detail.js", - "bytes": 6041, - "lines": 134 - }, { "path": "frontend/pages/dragon-tiger/page.html", "bytes": 5754, "lines": 85 }, + { + "path": "backend/data/providers/tushare_stocks.py", + "bytes": 5592, + "lines": 123 + }, { "path": "frontend/pages.config.js", "bytes": 5385, @@ -721,6 +721,11 @@ "bytes": 4712, "lines": 106 }, + { + "path": "backend/data/providers/tushare_helpers.py", + "bytes": 4406, + "lines": 124 + }, { "path": "backend/features/market/routes.py", "bytes": 4276, @@ -786,11 +791,6 @@ "bytes": 2514, "lines": 63 }, - { - "path": "backend/data/providers/tushare_helpers.py", - "bytes": 2360, - "lines": 75 - }, { "path": "backend/jobs/service.py", "bytes": 2337, @@ -846,6 +846,11 @@ "bytes": 1642, "lines": 53 }, + { + "path": "backend/data/providers/tushare_transport.py", + "bytes": 1592, + "lines": 50 + }, { "path": "backend/features/market/insights.py", "bytes": 1580, @@ -856,11 +861,6 @@ "bytes": 1535, "lines": 39 }, - { - "path": "backend/data/providers/tushare_transport.py", - "bytes": 1455, - "lines": 48 - }, { "path": "backend/features/themes/routes.py", "bytes": 1337, diff --git a/frontend/pages/heaven/page.js b/frontend/pages/heaven/page.js index ff108bb..8f7cc72 100644 --- a/frontend/pages/heaven/page.js +++ b/frontend/pages/heaven/page.js @@ -113,9 +113,12 @@ async function loadHeavenSetup(force = false, sector = "", stockCode = "") { document.querySelector("#resetHeavenCalibrationButton"), ].filter(Boolean); cancelHeavenPerformance(); - heavenView?.classList.add("heaven-data-loading"); + const blocking = !state.heavenSetup; + if (blocking) heavenView?.classList.add("heaven-data-loading"); if (loadButton) loadButton.disabled = true; calibrationButtons.forEach((button) => { button.disabled = true; }); + const controller = new AbortController(); + const timeoutId = window.setTimeout(() => controller.abort(), 25_000); try { if (state.heavenSetup?.requestedKey && state.heavenSetup.requestedKey !== requestedKey) { state.personalField = null; @@ -126,7 +129,7 @@ async function loadHeavenSetup(force = false, sector = "", stockCode = "") { if (sector) query.set("sector", sector); if (stockCode) query.set("stock_code", stockCode); if (manualData) query.set("manual_data", JSON.stringify(manualData)); - const payload = await apiRequest(`/api/heaven/setup?${query}`); + const payload = await apiRequest(`/api/heaven/setup?${query}`, "GET", null, { signal: controller.signal }); if ( requestSequence !== state.heavenRequestSequence || calendarDate !== document.querySelector("#qiObservationDate")?.value @@ -152,9 +155,15 @@ async function loadHeavenSetup(force = false, sector = "", stockCode = "") { if (payload.chart.selection_notice) showHeavenNotice(payload.chart.selection_notice); } catch (error) { if (requestSequence !== state.heavenRequestSequence) return; - showHeavenNotice(error.message || "问天数据加载失败"); - showToast(error.message || "问天数据加载失败"); + const aborted = error?.payload?.aborted || /abort|超时|cancel/i.test(String(error?.message || "")); + const message = aborted + ? "问天数据仍在准备，页面可继续输入和操作" + : (error.message || "问天数据加载失败"); + showHeavenNotice(message); + if (!aborted) showToast(message); + if (!state.heavenSetup) renderHeavenWorkspace(); } finally { + window.clearTimeout(timeoutId); if (requestSequence === state.heavenRequestSequence) { heavenView?.classList.remove("heaven-data-loading"); if (loadButton) loadButton.disabled = false; diff --git a/frontend/pages/market/stock-detail.js b/frontend/pages/market/stock-detail.js index 2906238..660c586 100644 --- a/frontend/pages/market/stock-detail.js +++ b/frontend/pages/market/stock-detail.js @@ -20,17 +20,9 @@ async function openStock(code, fallback = null) { setText("detailStreak", row.status === "涨停" ? streakLabel(row.streak) : row.status || "--"); setText("detailReason", row.reason || "--"); setText("detailSector", row.sector || "其他"); - setText("detailFirst", row.first_time || "--"); - setText("detailLast", row.last_time || "--"); - setText("detailOpen", `${number(row.open_times)} 次`); - setText("detailTurnover", `${formatNumber(row.turnover_rate, 2)}%`); - setText("detailAmount", `${formatNumber(row.amount_billion, 2)} 亿`); - setText("detailSeal", `${formatNumber(row.seal_amount_million, 0)} 万`); + setStockBoardFields(row); setText("chartSource", "正在加载行情"); - setText("flowNet", "--"); - setText("flowLarge", "--"); - setText("flowMedium", "--"); - setText("flowSmall", "--"); + renderMoneyflow({}); document.querySelector("#reasonInput").value = row.reason || ""; document.querySelector("#stockNoteContent").value = ""; document.querySelector("#stockNotePlan").value = ""; @@ -48,6 +40,7 @@ async function openStock(code, fallback = null) { setText("detailName", stock.name || row.name); setText("detailPrice", formatNumber(stock.price || row.price, 2)); setText("detailChange", `${signed(stock.change ?? row.change)}%`); + setStockBoardFields({ ...row, ...stock }); renderMoneyflow(payload.moneyflow || {}); renderStockNotes(payload.notes || []); updateWatchButton(); @@ -121,6 +114,21 @@ function renderStockDetailIntraday(payload) { }); } +function setStockBoardFields(row) { + const firstTime = String(row.first_time || "").trim(); + const lastTime = String(row.last_time || "").trim(); + setText("detailFirst", firstTime && firstTime !== "--" ? firstTime : "--"); + setText("detailLast", lastTime && lastTime !== "--" ? lastTime : "--"); + setText("detailOpen", row.open_times === null || row.open_times === undefined || row.open_times === "" ? "--" : `${number(row.open_times)} 次`); + setText("detailTurnover", presentMetric(row.turnover_rate) ? `${formatNumber(row.turnover_rate, 2)}%` : "--"); + setText("detailAmount", presentMetric(row.amount_billion) ? `${formatNumber(row.amount_billion, 2)} 亿` : "--"); + setText("detailSeal", presentMetric(row.seal_amount_million) ? `${formatNumber(row.seal_amount_million, 0)} 万` : "--"); +} + +function presentMetric(value) { + return meaningfulNumber(value) && Number(value) !== 0; +} + function openActiveStockInHeaven() { const code = state.activeStock?.code; if (!/^\d{6}$/.test(String(code || ""))) return; diff --git a/frontend/pages/review/page.js b/frontend/pages/review/page.js index 025c639..2013b2d 100644 --- a/frontend/pages/review/page.js +++ b/frontend/pages/review/page.js @@ -408,8 +408,18 @@ async function saveReasonOverride(event) { } function renderMoneyflow(flow) { - for (const [id, value] of [["flowNet", flow.net_million], ["flowLarge", flow.large_million], ["flowMedium", flow.medium_million], ["flowSmall", flow.small_million]]) { + const payload = flow || {}; + const available = payload.available !== false && [ + payload.net_million, payload.large_million, payload.medium_million, payload.small_million, + ].some((value) => value !== null && value !== undefined && value !== ""); + for (const [id, value] of [["flowNet", payload.net_million], ["flowLarge", payload.large_million], ["flowMedium", payload.medium_million], ["flowSmall", payload.small_million]]) { const element = document.getElementById(id); + if (!element) continue; + if (!available || value === null || value === undefined || value === "") { + element.textContent = "--"; + element.className = ""; + continue; + } element.textContent = formatMoneyMillion(value); element.className = changeClass(value); } diff --git a/frontend/shared/api.js b/frontend/shared/api.js index 1d68232..00d274a 100644 --- a/frontend/shared/api.js +++ b/frontend/shared/api.js @@ -62,6 +62,9 @@ try { response = await fetch(url, requestOptions(method, body, options.signal)); } catch (error) { + if (error?.name === "AbortError") { + throw new ApiError("请求已取消或超时", 0, { aborted: true }); + } throw new ApiError(readableRequestError(error), 0, null); } const payload = await parseJson(response); diff --git a/tests/test_frontend_contract.py b/tests/test_frontend_contract.py index e2657b7..f29f797 100644 --- a/tests/test_frontend_contract.py +++ b/tests/test_frontend_contract.py @@ -459,6 +459,19 @@ class FrontendContractTests(unittest.TestCase): self.assertIn('payload.question_preset = state.heartQuestionPreset;', self.script) self.assertIn('payload.cast_at = state.heartCastAt;', self.script) + def test_heaven_loading_timeout_clears_dimmed_state(self): + self.assertIn("controller.abort()", self.script) + self.assertIn('heavenView?.classList.remove("heaven-data-loading")', self.script) + self.assertIn("问天数据仍在准备，页面可继续输入和操作", self.script) + self.assertIn("const blocking = !state.heavenSetup;", self.script) + self.assertIn("payload?.aborted", self.script) + + def test_stock_detail_does_not_display_missing_metrics_as_zero(self): + self.assertIn("function setStockBoardFields(row)", self.script) + self.assertIn("function presentMetric(value)", self.script) + self.assertIn("payload.available !== false", self.script) + self.assertIn('element.textContent = "--"', self.script) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_hel494_regressions.py b/tests/test_hel494_regressions.py new file mode 100644 index 0000000..1b2e274 --- /dev/null +++ b/tests/test_hel494_regressions.py @@ -0,0 +1,193 @@ +from __future__ import annotations + +import unittest +from unittest.mock import MagicMock, patch + +from backend.data.providers.tushare_client import TushareClient, TushareError +from backend.data.providers.tushare_helpers import _moneyflow_payload +from backend.data.realtime import ( + WebRealtimeAggregator, + _normalize_eastmoney_limit_row, + _normalize_eastmoney_stock_quote, +) +from backend.data.providers.tushare_daily import DailyMarketMixin + + +class MoneyflowPayloadTests(unittest.TestCase): + def test_missing_row_is_not_zero(self) -> None: + payload = _moneyflow_payload(None) + self.assertFalse(payload["available"]) + self.assertIsNone(payload["net_million"]) + self.assertIsNone(payload["large_million"]) + + def test_empty_row_is_not_zero(self) -> None: + payload = _moneyflow_payload({}) + self.assertFalse(payload["available"]) + self.assertIsNone(payload["net_million"]) + + def test_real_zero_net_is_kept_when_source_exists(self) -> None: + payload = _moneyflow_payload( + { + "net_mf_amount": 0, + "buy_lg_amount": 1, + "sell_lg_amount": 1, + "buy_elg_amount": 0, + "sell_elg_amount": 0, + "buy_md_amount": 0, + "sell_md_amount": 0, + "buy_sm_amount": 0, + "sell_sm_amount": 0, + } + ) + self.assertTrue(payload["available"]) + self.assertEqual(payload["net_million"], 0) + + +class LimitOverlayTests(unittest.TestCase): + def test_normalize_limit_keeps_missing_seal_as_none(self) -> None: + row = DailyMarketMixin._normalize_limit( + { + "ts_code": "000737.SZ", + "name": "北方铜业", + "close": 12.3, + "pct_chg": 10, + "amount": 1e8, + "amount_unit": "yuan", + }, + "涨停", + ) + self.assertIsNone(row["seal_amount_million"]) + self.assertEqual(row["first_time"], "--") + + def test_overlay_fills_board_times_from_official_list(self) -> None: + mixin = DailyMarketMixin() + mixin._load_limit_lists = lambda trade_date: [ + { + "ts_code": "000737.SZ", + "first_time": "09:31:02", + "last_time": "10:18:11", + "fd_amount": 82000000, + "open_times": 1, + "turnover_ratio": 18.4, + } + ] + mixin.realtime_aggregator = None + rows = mixin._overlay_board_fields( + [{"ts_code": "000737.SZ", "close": 12.3, "limit_type": "U"}], + "20260908", + ) + self.assertEqual(rows[0]["first_time"], "09:31:02") + self.assertEqual(rows[0]["fd_amount"], 82000000) + self.assertEqual(rows[0]["turnover_ratio"], 18.4) + + +class ShenwanRealtimeSourceTests(unittest.TestCase): + def test_transport_refuses_rt_sw_k(self) -> None: + client = TushareClient(token="demo") + with self.assertRaisesRegex(TushareError, "rt_sw_k is disabled"): + client.query("rt_sw_k", {"ts_code": "801074.SI"}) + + def test_outer_realtime_uses_eastmoney_shenwan_not_rt_sw_k(self) -> None: + client = TushareClient(token="demo") + client.query = MagicMock(side_effect=AssertionError("should not call tushare")) + client.realtime_aggregator = MagicMock() + client.realtime_aggregator.eastmoney_shenwan_quote.return_value = { + "code": "801074.SI", + "name": "工业金属", + "close": 1234.5, + "pre_close": 1200, + "change": 2.88, + "pct_change": 2.88, + "quote_date": "20260908", + "quote_time": "2026-09-08T14:50:00+08:00", + "source": "eastmoney_sw", + } + row, source, error = client._sw_outer_realtime("801074.SI", "工业金属", "20260908") + self.assertEqual(source, "eastmoney_sw") + self.assertEqual(error, "") + self.assertEqual(row["change"], 2.88) + client.query.assert_not_called() + + def test_outer_waiting_state_has_no_permission_error(self) -> None: + client = TushareClient(token="demo") + client.realtime_aggregator = None + row, source, error = client._sw_outer_realtime( + "801074.SI", "工业金属", "20260908", finalized=True + ) + self.assertEqual(row, {}) + self.assertIn("尚未入库", error) + self.assertNotIn("权限", error) + self.assertNotIn("rt_sw_k", error) + + +class EastmoneyHelperTests(unittest.TestCase): + def test_limit_pool_row_keeps_board_clock(self) -> None: + parsed = _normalize_eastmoney_limit_row( + { + "c": "000737", + "m": 0, + "n": "北方铜业", + "fbt": 93102, + "lbt": 101811, + "zbc": 1, + "lbc": 2, + "hs": 18.4, + "fund": 82000000, + }, + "U", + ) + self.assertEqual(parsed["ts_code"], "000737.SZ") + self.assertEqual(parsed["first_time"], "09:31:02") + self.assertEqual(parsed["last_time"], "10:18:11") + self.assertEqual(parsed["fd_amount"], 82000000) + + def test_stock_quote_keeps_moneyflow_when_present(self) -> None: + quote = _normalize_eastmoney_stock_quote( + { + "f43": 12.3, + "f60": 11.18, + "f46": 11.2, + "f44": 12.3, + "f45": 11.1, + "f47": 1000, + "f48": 150000000, + "f58": "北方铜业", + "f86": 0, + "f168": 8.5, + "f62": 25000000, + "f78": 3000000, + "f84": -1000000, + }, + "000737.SZ", + ) + self.assertEqual(quote["net_mf_amount"], 2500) + payload = _moneyflow_payload(quote) + self.assertTrue(payload["available"]) + self.assertEqual(payload["net_million"], 25) + + @patch.object(WebRealtimeAggregator, "_get_json") + def test_shenwan_quote_uses_eastmoney_90_prefix(self, get_json: MagicMock) -> None: + get_json.return_value = { + "rc": 0, + "data": { + "diff": [ + { + "f12": "801074", + "f14": "工业金属", + "f2": 1234.5, + "f3": 2.88, + "f18": 1200, + "f17": 1205, + "f15": 1240, + "f16": 1198, + "f6": 1, + "f124": 1757319000, + } + ] + }, + } + quote = WebRealtimeAggregator().eastmoney_shenwan_quote("801074.SI") + self.assertEqual(quote["source"], "eastmoney_sw") + self.assertAlmostEqual(quote["change"], 2.88) + params = get_json.call_args.args[1] + self.assertEqual(params["secids"], "90.801074") diff --git a/tests/test_stock_detail_realtime.py b/tests/test_stock_detail_realtime.py index 164d779..15b4a8e 100644 --- a/tests/test_stock_detail_realtime.py +++ b/tests/test_stock_detail_realtime.py @@ -319,7 +319,12 @@ class StockDetailRealtimeTests(unittest.TestCase): def test_today_detail_falls_back_to_eastmoney_then_intraday(self): today = FixedMarketDatetime.fixed_now.strftime("%Y%m%d") aggregator = FreeQuoteAggregator( - _free_quote("eastmoney_stock", ts_code="600000.SH", name="浦发银行"), + _free_quote( + "eastmoney_stock", + ts_code="600000.SH", + name="浦发银行", + net_mf_amount=12, + ), ) self.service.realtime_aggregator = aggregator DeniedRealtimeClientStub.quote_calls = 0 diff --git a/xiaobai-datahub/config/hub-quality.config.json b/xiaobai-datahub/config/hub-quality.config.json index 578dea8..2f208cd 100644 --- a/xiaobai-datahub/config/hub-quality.config.json +++ b/xiaobai-datahub/config/hub-quality.config.json @@ -14,6 +14,7 @@ "list_limit_max": 5000, "calendar_start": "20160101", "index_history_trading_days": 260, + "daily_history_trading_days": 250, "eod_retry_start": "15:15", "eod_retry_interval_minutes": 30, "eod_retry_cutoff": "23:30", diff --git a/xiaobai-datahub/datahub/admin_api.py b/xiaobai-datahub/datahub/admin_api.py index f6c86a7..3d33657 100644 --- a/xiaobai-datahub/datahub/admin_api.py +++ b/xiaobai-datahub/datahub/admin_api.py @@ -94,7 +94,7 @@ class AdminAPI: {"id": "eod_retry", "at": "15:15-23:30", "title": "盘后未出数自动重试（每 30 分钟，成功即停）"}, {"id": "eod_revise", "at": "20:00-23:20", "title": "估值发布后复核（轻量比对，有修订才整组原子追补）"}, {"id": "stocks_refresh", "at": stocks_times, "title": "股票主档刷新与正式发布（新上市/更名，无变化跳过）"}, - {"id": "history_backfill", "at": "manual", "title": "回补历史日历与指数日 K"}, + {"id": "history_backfill", "at": "manual", "title": "回补历史日历、个股日 K 与指数日 K"}, {"id": "cleanup", "at": "00:30", "title": "清理 staging / 日志"}, {"id": "backup", "at": "00:40", "title": "SQLite 备份"}, ], diff --git a/xiaobai-datahub/datahub/cli.py b/xiaobai-datahub/datahub/cli.py index b94e025..dfe400c 100644 --- a/xiaobai-datahub/datahub/cli.py +++ b/xiaobai-datahub/datahub/cli.py @@ -15,10 +15,11 @@ from datahub.timeutil import yyyymmdd def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="xiaobai-datahub CLI") sub = parser.add_subparsers(dest="command", required=True) - history = sub.add_parser("history-backfill", help="回补 2016 年起交易日历和网站所用指数日 K") + history = sub.add_parser("history-backfill", help="回补交易日历、个股日 K（默认 250 日）和网站所用指数日 K") history.add_argument("--calendar-start", default=None, help="日历起点，默认配置 calendar_start") history.add_argument("--index-days", type=int, default=None, help="指数回补交易日数量，默认 260") - history.add_argument("--force", action="store_true", help="覆盖已发布的指数日期") + history.add_argument("--daily-days", type=int, default=None, help="个股日 K 回补交易日数量，默认 250") + history.add_argument("--force", action="store_true", help="覆盖已发布的个股日 K / 指数日期") refresh = sub.add_parser("eod-refresh", help="对指定交易日补跑盘后正式数据（跳过已完整发布的一致性边界，仍走质量门禁）") refresh.add_argument("--trade-date", default=None, help="交易日 YYYYMMDD，默认今天") refresh.add_argument( @@ -46,6 +47,7 @@ def main(argv: list[str] | None = None) -> int: result = hub.pipeline.backfill_history( calendar_start=args.calendar_start, index_days=args.index_days, + daily_days=args.daily_days, force=args.force, ) json.dump(result, sys.stdout, ensure_ascii=False, indent=2, default=str) diff --git a/xiaobai-datahub/datahub/pipeline.py b/xiaobai-datahub/datahub/pipeline.py index c993c8a..0b737c4 100644 --- a/xiaobai-datahub/datahub/pipeline.py +++ b/xiaobai-datahub/datahub/pipeline.py @@ -491,24 +491,96 @@ class Pipeline: ) return sorted(str(row["cal_date"]) for row in rows) + def backfill_daily_history( + self, + end_date: str | None = None, + trading_days: int | None = None, + force: bool = False, + ) -> dict[str, Any]: + """Incrementally publish official daily bars for the website K-line window. + + Same-day EOD still uses the atomic A-group. History backfill publishes + daily (with adj_factor) first so K-line coverage cannot be blocked by + the looser moneyflow universe, then valuation and moneyflow/auction + best-effort. Complete daily dates are skipped unless ``force``. + """ + end = yyyymmdd(end_date or self.clock()) + limit = int(trading_days or self.settings.daily_history_trading_days) + open_dates = self.open_trade_dates(end, limit) + if not open_dates: + return { + "start": None, + "end": end, + "requested_days": 0, + "published": [], + "skipped": [], + "failed": [{"error": "calendar has no open dates on or before end"}], + "ok": False, + } + start = open_dates[0] + published: list[dict[str, Any]] = [] + skipped: list[str] = [] + failed: list[dict[str, Any]] = [] + for day in open_dates: + if not force and self.active_batch("daily", day): + skipped.append(day) + continue + try: + daily = self.run_dataset("daily", day) + datasets = {"daily": daily.get("state")} + try: + valuation = self.run_dataset("valuation", day) + datasets["valuation"] = valuation.get("state") + except Exception as exc: + datasets["valuation"] = f"failed:{exc}"[:180] + for name in ("moneyflow", "auction"): + try: + extra = self.run_dataset(name, day) + datasets[name] = extra.get("state") + except Exception as exc: + datasets[name] = f"failed:{exc}"[:180] + published.append({"trade_date": day, "datasets": datasets}) + except Exception as exc: + failed.append({"trade_date": day, "error": str(exc)}) + return { + "start": start, + "end": end, + "requested_days": len(open_dates), + "published": published, + "skipped": skipped, + "failed": failed, + "ok": not failed, + } + def backfill_history( self, trade_date: str | None = None, calendar_start: str | None = None, index_days: int | None = None, + daily_days: int | None = None, codes: tuple[str, ...] | None = None, force: bool = False, ) -> dict[str, Any]: - """Idempotent calendar + website-index history backfill.""" + """Idempotent calendar + stock daily + website-index history backfill.""" day = yyyymmdd(trade_date or self.clock()) calendar = self.ingest_reference(day, start=calendar_start) + daily = self.backfill_daily_history( + end_date=day, + trading_days=daily_days, + force=force, + ) index = self.backfill_index_history( end_date=day, trading_days=index_days, codes=codes, force=force, ) - return {"calendar": calendar, "index_daily": index, "ok": bool(index.get("ok"))} + return { + "calendar": calendar, + "daily": daily, + "index_daily": index, + "ok": bool(daily.get("ok")) and bool(index.get("ok")), + } def backfill_index_history( self, @@ -803,10 +875,7 @@ class Pipeline: "published_rows": len(published), "upstream_rows": 0, } - listed = self.db.fetchone( - "SELECT COUNT(*) AS n FROM stock_master WHERE list_status = 'L'", - ) - listed_n = int((listed or {}).get("n") or 0) + listed_n = self._listed_count(day) floor = float(self.settings.quality.get("daily_row_ratio") or 0.98) if listed_n and len(upstream) / listed_n < floor: return { @@ -1314,14 +1383,33 @@ class Pipeline: if isinstance(item, dict) and item.get("state") == "failed" ] + def _listed_count(self, trade_date: str = "") -> int: + """Count listed names that already existed on ``trade_date``. + + Historical daily bars must not be judged against later IPOs, or a + correct past session fails the 0.98 row-ratio gate. + """ + day = yyyymmdd(trade_date) if trade_date else "" + if day: + listed = self.db.fetchone( + """ + SELECT COUNT(*) AS n FROM stock_master + WHERE list_status = 'L' + AND (list_date IS NULL OR TRIM(list_date) = '' OR list_date <= ?) + """, + (day,), + ) + else: + listed = self.db.fetchone( + "SELECT COUNT(*) AS n FROM stock_master WHERE list_status = 'L'" + ) + return int((listed or {}).get("n") or 0) + def validate(self, dataset: str, batch_id: str, trade_date: str, rows: list[dict[str, Any]]) -> dict[str, Any]: quality = self.settings.quality errors: list[str] = [] warnings: list[str] = [] - listed = self.db.fetchone( - "SELECT COUNT(*) AS n FROM stock_master WHERE list_status = 'L'", - ) - listed_n = int((listed or {}).get("n") or 0) + listed_n = self._listed_count(trade_date) row_n = len(rows) if dataset == "limit_events": keys = [(row.get("ts_code"), row.get("trade_date"), row.get("limit_type")) for row in rows] diff --git a/xiaobai-datahub/datahub/settings.py b/xiaobai-datahub/datahub/settings.py index 6f27ef7..df25757 100644 --- a/xiaobai-datahub/datahub/settings.py +++ b/xiaobai-datahub/datahub/settings.py @@ -56,6 +56,10 @@ class Settings: def index_history_trading_days(self) -> int: return int(self.quality.get("index_history_trading_days") or 260) + @property + def daily_history_trading_days(self) -> int: + return int(self.quality.get("daily_history_trading_days") or 250) + @property def moneyflow_history_trading_days(self) -> int: return int(self.quality.get("moneyflow_history_trading_days") or 60) diff --git a/xiaobai-datahub/tests/test_history_backfill.py b/xiaobai-datahub/tests/test_history_backfill.py index d490d71..a26ab2c 100644 --- a/xiaobai-datahub/tests/test_history_backfill.py +++ b/xiaobai-datahub/tests/test_history_backfill.py @@ -5,7 +5,7 @@ from datetime import date, timedelta from datahub.coverage import calendar_coverage, point_coverage, published_range_coverage from datahub.serving import V1API -from tests.fixtures import TRADE_DATE, fake_transport +from tests.fixtures import RAW, TRADE_DATE, fake_transport from tests.test_pipeline import make_pipeline @@ -48,6 +48,8 @@ def history_transport(open_dates: list[str], extra_closed: list[str] | None = No } ) + dated_apis = ("daily", "daily_basic", "adj_factor", "moneyflow", "stk_auction") + def transport(api_name, params, fields): if api_name == "trade_cal": start = str(params.get("start_date") or "") @@ -66,6 +68,11 @@ def history_transport(open_dates: list[str], extra_closed: list[str] | None = No if end: rows = [row for row in rows if row["trade_date"] <= end] return rows + if api_name in dated_apis: + day = str(params.get("trade_date") or "") + if day not in open_set: + return [] + return [{**row, "trade_date": day} for row in RAW.get(api_name) or []] return fake_transport(api_name, params, fields) return transport @@ -221,6 +228,54 @@ class HistoryBackfillTests(unittest.TestCase): self.assertEqual(result["rows"], 1) self.assertEqual(calls["n"], before) + def test_daily_history_is_idempotent_and_covers_requested_days(self) -> None: + open_dates = consecutive_open_days(TRADE_DATE, 5) + pipe, db = make_pipeline( + quality={ + "index_history_trading_days": 5, + "daily_history_trading_days": 5, + "calendar_start": open_dates[0], + } + ) + pipe.adapter._transport = history_transport(open_dates) + first = pipe.backfill_history(TRADE_DATE, index_days=5, daily_days=5) + self.assertTrue(first["ok"]) + self.assertEqual(first["daily"]["requested_days"], 5) + self.assertEqual(len(first["daily"]["published"]), 5) + pubs = db.fetchall("SELECT trade_date FROM publications WHERE dataset='daily'") + self.assertEqual(sorted(row["trade_date"] for row in pubs), open_dates) + for day in open_dates: + rows = db.fetchall( + """ + SELECT COUNT(*) AS n FROM eod_bars + WHERE trade_date = ? AND batch_id = ( + SELECT active_batch FROM publications WHERE dataset='daily' AND trade_date = ? + ) + """, + (day, day), + ) + self.assertEqual(rows[0]["n"], 2) + + second = pipe.backfill_daily_history(end_date=TRADE_DATE, trading_days=5) + self.assertTrue(second["ok"]) + self.assertEqual(second["published"], []) + self.assertEqual(second["skipped"], open_dates) + + def test_daily_row_ratio_ignores_later_ipos(self) -> None: + pipe, db = make_pipeline() + pipe.adapter._transport = history_transport([TRADE_DATE]) + pipe.ingest_reference(TRADE_DATE, start=TRADE_DATE) + with db.write() as connection: + connection.execute( + "INSERT INTO stock_master(ts_code, symbol, name, list_status, list_date, updated_at) " + "VALUES (?,?,?,?,?,?)", + ("688001.SH", "688001", "未来上市", "L", "20250101", "2024-09-02T00:00:00+08:00"), + ) + self.assertEqual(pipe._listed_count(TRADE_DATE), 2) + result = pipe.run_eod_batch_a(TRADE_DATE) + self.assertEqual(pipe.eod_failures(result), []) + self.assertEqual(result["daily"]["state"], "published") + def test_coverage_helpers_point_and_calendar(self) -> None: pipe, db = make_pipeline() pipe.ingest_reference(TRADE_DATE)