from __future__ import annotations import copy from datetime import datetime from statistics import median from typing import Any from backend.data.numbers import non_nan_number as _number from backend.data.providers.tushare_client import TushareError from backend.features.market.insights_context import _display_date class MarketAuctionInsightsMixin: def auction_center( self, requested_date: str, force: bool = False, user_id: int = 0, ) -> dict[str, Any]: trade_date, previous_date = self._trade_context(requested_date) session = self._auction_session(requested_date, trade_date) phase = str(session["phase"]) ifind_ready = bool(self.ifind and self.ifind.configured) live_dynamic = phase == "observing" and ifind_ready use_ifind_snapshot = phase in {"observing", "selection", "finalized"} and ifind_ready data_date = previous_date if phase == "pending" or (phase == "observing" and not live_dynamic) else trade_date carried_forward = data_date != trade_date cache_key = data_date if not force and not live_dynamic: cached = self.database.get_data_snapshot("auction_center_v6", cache_key) if cached: result = copy.deepcopy(cached) result["meta"] = { **result.get("meta", {}), **session, "requested_date": _display_date(requested_date), "trade_date": _display_date(data_date), "carried_forward": carried_forward, "available": bool((result.get("summary") or {}).get("stock_count")), "cached": True, } return self._with_auction_watchlist(result, data_date, user_id) if use_ifind_snapshot: rows = self._dynamic_auction_rows(data_date, previous_date, user_id) else: rows = [] if not rows and not live_dynamic: try: rows = self.client.query("stk_auction", {"trade_date": data_date}) except TushareError: rows = self.database.auction_factors_for_date(data_date) if not rows: return { "meta": { **session, "requested_date": _display_date(requested_date), "trade_date": _display_date(data_date), "carried_forward": carried_forward, "available": False, "cached": False, "notice": "该交易日暂无可用竞价快照", "updated_at": datetime.now().astimezone().isoformat(timespec="seconds"), }, "summary": { "stock_count": 0, "up_count": 0, "down_count": 0, "limit_open_count": 0, "strong_open_count": 0, "median_change": 0, "amount_billion": 0, "candidate_count": 0, "focus_count": 0, "one_price_count": 0, }, "expectations": {"超预期": 0, "符合预期": 0, "低于预期": 0}, "candidate_meta": {"baseline_date": _display_date(previous_date)}, "themes": {"carry": [], "new_themes": []}, "amount_history": self._auction_amount_history(data_date), "news_feedback": {"available": False, "message": "隔夜消息反馈暂不可用"}, "focus_rows": [], "one_price_rows": [], "rows": [], "watchlist_rows": [], "watchlist_missing_count": 0, } master = self._stock_master() try: limit_rows = self.client.query( "stk_limit", {"trade_date": data_date}, "trade_date,ts_code,up_limit,down_limit", ) except TushareError: limit_rows = [] limit_map = {str(item.get("ts_code") or ""): item for item in limit_rows} normalized = [] for row in rows: ts_code = str(row.get("ts_code") or "") stock = master.get(ts_code) price = _number(row.get("price")) pre_close = _number(row.get("pre_close")) list_date = str((stock or {}).get("list_date") or "") if ( not stock or price <= 0 or pre_close <= 0 or (list_date and list_date >= data_date) ): continue change = (price / pre_close - 1) * 100 amount_million = _number(row.get("amount")) / 1_000_000 volume_ratio = _number(row.get("volume_ratio")) turnover_rate = _number(row.get("turnover_rate")) up_limit = _number((limit_map.get(ts_code) or {}).get("up_limit")) is_one_price = bool( up_limit > 0 and abs(price - up_limit) <= max(0.001, up_limit * 0.00005) ) normalized.append( { "code": str(stock.get("code") or ts_code.split(".")[0]), "ts_code": ts_code, "name": str(stock.get("name") or "--"), "sector": str(stock.get("industry") or "其他"), "price": round(price, 2), "pre_close": round(pre_close, 2), "change": round(change, 2), "volume_ten_thousand": round(_number(row.get("vol")) / 10_000, 2), "amount_million": round(amount_million, 2), "turnover_rate": round(turnover_rate, 4), "volume_ratio": round(volume_ratio, 2), "up_limit": round(up_limit, 2) if up_limit else None, "is_one_price": is_one_price, "signal": ( "竞价涨停" if change >= 9.5 else "强势高开" if change >= 3 else "高开" if change > 0.2 else "深度低开" if change <= -3 else "低开" if change < -0.2 else "平开" ), } ) normalized.sort(key=lambda item: (item["amount_million"], item["volume_ratio"]), reverse=True) self.database.upsert_auction_factors(rows) changes = [item["change"] for item in normalized] total = len(normalized) _, baseline_date = self._trade_context(data_date) candidates, candidate_meta, focus_rows = self._auction_candidates(normalized, baseline_date) candidate_map = {str(item.get("code") or ""): item for item in candidates} one_price_rows = [] for row in normalized: if not row.get("is_one_price"): continue enriched = candidate_map.get(str(row.get("code") or ""), {}) one_price_rows.append( { **row, **enriched, "attention_score": None, "expectation": "", "expected_change": None, "expectation_reason": "竞价价格封于当日涨停价，已从普通异动评分中隔离", } ) one_price_codes = {str(item.get("code") or "") for item in one_price_rows} candidates = [item for item in candidates if str(item.get("code") or "") not in one_price_codes] focus_rows = [item for item in focus_rows if str(item.get("code") or "") not in one_price_codes] one_price_rows.sort( key=lambda item: ( bool(item.get("is_market_core")), _number(item.get("prior_streak")), _number(item.get("amount_million")), ), reverse=True, ) expectations = { label: sum(item.get("expectation") == label for item in candidates) for label in ("超预期", "符合预期", "低于预期") } prior_snapshot = self.database.get_snapshot(baseline_date) or {} themes = self._auction_theme_evidence(prior_snapshot, candidates + one_price_rows) self._ensure_auction_amount_history(data_date) amount_history = self._auction_amount_history(data_date) prior_amounts = [item["amount_billion"] for item in amount_history[:-1]] current_amount = round(sum(item["amount_million"] for item in normalized) / 100, 2) previous_amount = prior_amounts[-1] if prior_amounts else 0 five_day_amounts = prior_amounts[-5:] five_day_average = sum(five_day_amounts) / len(five_day_amounts) if five_day_amounts else 0 result = { "meta": { "requested_date": _display_date(requested_date), "trade_date": _display_date(data_date), "carried_forward": carried_forward, "available": bool(normalized), **session, "cached": False, "updated_at": datetime.now().astimezone().isoformat(timespec="seconds"), }, "summary": { "stock_count": total, "up_count": sum(value > 0.2 for value in changes), "down_count": sum(value < -0.2 for value in changes), "limit_open_count": len(one_price_rows), "strong_open_count": sum(value >= 3 for value in changes), "median_change": round(median(changes), 2) if changes else 0, "amount_billion": current_amount, "amount_change_previous": round((current_amount / previous_amount - 1) * 100, 1) if previous_amount else None, "amount_change_5d": round((current_amount / five_day_average - 1) * 100, 1) if five_day_average else None, "candidate_count": len(candidates), "focus_count": len(focus_rows), "one_price_count": len(one_price_rows), }, "expectations": expectations, "candidate_meta": candidate_meta, "themes": themes, "amount_history": amount_history, "news_feedback": { "available": False, "message": "隔夜消息反馈暂不可用", "detail": "待稳定的新闻与公告数据接入后开放", }, "focus_rows": focus_rows, "one_price_rows": one_price_rows, "rows": candidates, } if not live_dynamic: self.database.save_data_snapshot("auction_center_v6", cache_key, "market", result) return self._with_auction_watchlist(result, data_date, user_id)