from __future__ import annotations import copy import json from datetime import datetime, time as dt_time, timedelta, timezone from statistics import median from typing import Any, Callable from database import ReviewDatabase from ifind_client import IfindError, IfindHttpClient from tushare_client import TushareClient, TushareError CHINA_TIMEZONE = timezone(timedelta(hours=8)) def _number(value: Any, default: float = 0.0) -> float: try: number = float(value) return number if number == number else default except (TypeError, ValueError): return default def _display_date(value: str) -> str: text = str(value or "").replace("-", "") if len(text) != 8: return str(value or "") return f"{text[:4]}-{text[4:6]}-{text[6:]}" class MarketInsightsService: """Read-only market features backed by Tushare and shared SQLite caches.""" def __init__( self, database: ReviewDatabase, client: TushareClient, now_provider: Callable[[], datetime] | None = None, ifind: IfindHttpClient | None = None, ) -> None: self.database = database self.client = client self._now_provider = now_provider or (lambda: datetime.now(CHINA_TIMEZONE)) self.ifind = ifind def _trade_context(self, requested_date: str) -> tuple[str, str]: """Resolve trading dates without making cached feature pages depend on Tushare uptime.""" requested = str(requested_date or "").replace("-", "") try: return self.client.resolve_trade_context(requested) except TushareError: latest = self.database.get_latest_real_snapshot(requested) or {} trade_date = str( (latest.get("meta") or {}).get("trade_date") or latest.get("_snapshot_date") or requested ).replace("-", "") previous = self.database.get_latest_real_snapshot(trade_date, strictly_before=True) or {} previous_date = str( (previous.get("meta") or {}).get("trade_date") or previous.get("_snapshot_date") or "" ).replace("-", "") return trade_date, previous_date def _latest_feature_snapshot(self, kind: str, trade_date: str) -> dict[str, Any] | None: return self.database.get_latest_data_snapshot(kind, "", trade_date) def _auction_session(self, requested_date: str, trade_date: str) -> dict[str, Any]: now = self._now_provider() if now.tzinfo is None: now = now.replace(tzinfo=CHINA_TIMEZONE) else: now = now.astimezone(CHINA_TIMEZONE) requested = str(requested_date or "").replace("-", "") today = now.strftime("%Y%m%d") if requested != today or trade_date != today: return { "phase": "archive", "actionable": False, "next_transition_at": "", } local_time = now.time().replace(tzinfo=None) transitions = ( (dt_time(9, 15), "pending", dt_time(9, 15)), (dt_time(9, 25), "observing", dt_time(9, 25)), (dt_time(9, 30), "selection", dt_time(9, 30)), ) for boundary, phase, next_boundary in transitions: if local_time < boundary: transition = now.replace( hour=next_boundary.hour, minute=next_boundary.minute, second=0, microsecond=0, ) return { "phase": phase, "actionable": phase == "selection", "next_transition_at": transition.isoformat(timespec="seconds"), } return { "phase": "finalized", "actionable": False, "next_transition_at": "", } def _stock_master(self) -> dict[str, dict[str, Any]]: rows = self.database.list_stock_master() if not rows: rows = self.client.query( "stock_basic", {"list_status": "L"}, "ts_code,name,industry,market,list_date", ) self.database.upsert_stock_master(rows) rows = self.database.list_stock_master() return {str(row.get("ts_code") or ""): row for row in rows} @staticmethod def _expectation_label(actual_strength: float, expected_change: float) -> str: difference = actual_strength - expected_change if difference >= 1.5: return "超预期" if difference <= -1.5: return "低于预期" return "符合预期" @staticmethod def _auction_confirmation(row: dict[str, Any]) -> float: volume_ratio = _number(row.get("volume_ratio")) turnover_rate = _number(row.get("turnover_rate")) amount_million = _number(row.get("amount_million")) return ( (0.6 if volume_ratio >= 2 else 0.3 if volume_ratio >= 1.2 else -0.5 if volume_ratio < 0.6 else 0) + (0.25 if turnover_rate >= 0.15 else -0.25 if turnover_rate < 0.03 else 0) + (0.3 if amount_million >= 20 else 0.15 if amount_million >= 5 else -0.3 if amount_million < 1 else 0) ) @staticmethod def _attention_score( row: dict[str, Any], expected_change: float, core_tags: list[str], sources: list[str], prior_streak: int, strong_sector: bool, ) -> float: if core_tags: identity_score = 35.0 elif prior_streak >= 2: identity_score = 27.0 elif any(source in {"昨日涨停", "昨日炸板"} for source in sources): identity_score = 21.0 else: identity_score = 14.0 deviation_score = min(30.0, abs(_number(row.get("change")) - expected_change) * 5) volume_score = min(10.0, max(0.0, _number(row.get("volume_ratio"))) / 2 * 10) amount_score = min(6.0, max(0.0, _number(row.get("amount_million"))) / 10 * 6) turnover_score = min(4.0, max(0.0, _number(row.get("turnover_rate"))) / 0.2 * 4) theme_score = 15.0 if strong_sector else 7.0 if row.get("concepts") else 0.0 return round(min(100.0, identity_score + deviation_score + volume_score + amount_score + turnover_score + theme_score), 1) def _auction_candidates( self, rows: list[dict[str, Any]], baseline_date: str, ) -> tuple[list[dict[str, Any]], dict[str, Any], list[dict[str, Any]]]: """Build a narrow, explainable universe from prior limits, breaks and top-20 hot lists.""" snapshot = self.database.get_snapshot(baseline_date) or {} prior_limits = list(snapshot.get("limits") or []) prior_broken = list(snapshot.get("broken") or []) prior_sectors = list(snapshot.get("sectors") or []) strong_sector_names = { str(item.get("name") or "") for item in prior_sectors[:5] if item.get("name") } ths_rows, dc_rows, errors = self._hot_rows(baseline_date) candidates: dict[str, dict[str, Any]] = {} core_tags: dict[str, set[str]] = {} def ensure_candidate(item: dict[str, Any]) -> dict[str, Any] | None: code = str(item.get("code") or str(item.get("ts_code") or "").split(".")[0]) if not code: return None return candidates.setdefault( code, { "sources": [], "streak": 0, "sector": str(item.get("sector") or "其他"), "name": str(item.get("name") or item.get("ts_name") or "--"), "concepts": [], "ths_rank": None, "dc_rank": None, }, ) for item in prior_limits: candidate = ensure_candidate(item) if candidate is None: continue candidate["sources"].append("昨日涨停") candidate["streak"] = max(1, int(_number(item.get("streak"), 1))) for item in prior_broken: candidate = ensure_candidate(item) if candidate is not None and "昨日炸板" not in candidate["sources"]: candidate["sources"].append("昨日炸板") limit_streaks = [max(1, int(_number(item.get("streak"), 1))) for item in prior_limits] highest_streak = max(limit_streaks, default=0) for item in prior_limits: code = str(item.get("code") or "") streak = max(1, int(_number(item.get("streak"), 1))) if streak >= 3: core_tags.setdefault(code, set()).add("三板以上") if highest_streak and streak == highest_streak: core_tags.setdefault(code, set()).add("市场最高板") for sector in prior_sectors[:5]: name = str(sector.get("name") or "") members = [item for item in prior_limits if str(item.get("sector") or "其他") == name] if not members: continue leader = max( members, key=lambda item: ( int(_number(item.get("streak"), 1)), _number(item.get("amount_billion")), -_number(item.get("open_times")), ), ) core_tags.setdefault(str(leader.get("code") or ""), set()).add("题材核心") leadership = sorted( prior_limits, key=lambda item: ( int(_number(item.get("streak"), 1)), str(item.get("sector") or "") in strong_sector_names, _number(item.get("amount_billion")), ), reverse=True, ) if leadership: core_tags.setdefault(str(leadership[0].get("code") or ""), set()).add("市场领涨") hot_records: dict[str, dict[str, Any]] = {} for source, hot_rows, data_type in ( ("同花顺热榜", ths_rows, "热股"), ("东方财富热榜", dc_rows, "A股市场"), ): for item in hot_rows: if str(item.get("data_type") or "") != data_type: continue ts_code = str(item.get("ts_code") or "") code = ts_code.split(".")[0] rank = max(1, int(_number(item.get("rank"), 9999))) if not code or rank > 20: continue hot = hot_records.setdefault( code, { "name": str(item.get("ts_name") or "--"), "concepts": [], "ths_rank": None, "dc_rank": None, }, ) hot["ths_rank" if source == "同花顺热榜" else "dc_rank"] = rank if source == "同花顺热榜": hot["concepts"] = self._parse_concepts(item.get("concept")) ranked_hot = sorted( hot_records.items(), key=lambda pair: ( ((21 - (pair[1].get("ths_rank") or 21)) / 20) + ((21 - (pair[1].get("dc_rank") or 21)) / 20) + (0.35 if pair[1].get("ths_rank") and pair[1].get("dc_rank") else 0) ), reverse=True, ) for code, _ in ranked_hot[:5]: core_tags.setdefault(code, set()).add("人气前5") for code, hot in hot_records.items(): ranks = [rank for rank in (hot.get("ths_rank"), hot.get("dc_rank")) if isinstance(rank, int)] dual = len(ranks) == 2 if not ranks or (min(ranks) > 10 and not dual and code not in candidates and code not in core_tags): continue candidate = candidates.setdefault( code, { "sources": [], "streak": 0, "sector": "其他", "name": hot["name"], "concepts": [], "ths_rank": None, "dc_rank": None, }, ) candidate["ths_rank"] = hot.get("ths_rank") candidate["dc_rank"] = hot.get("dc_rank") candidate["concepts"] = hot.get("concepts") or [] if hot.get("ths_rank") and "同花顺热榜" not in candidate["sources"]: candidate["sources"].append("同花顺热榜") if hot.get("dc_rank") and "东方财富热榜" not in candidate["sources"]: candidate["sources"].append("东方财富热榜") normalized = [] for row in rows: candidate = candidates.get(str(row.get("code") or "")) if not candidate: continue streak = int(candidate["streak"]) expected_change = {1: 1.5, 2: 3.0, 3: 4.0}.get(streak, 5.0 if streak else 0.5) ranks = [ rank for rank in (candidate.get("ths_rank"), candidate.get("dc_rank")) if isinstance(rank, int) ] if len(ranks) == 2: expected_change += 0.8 elif ranks: best_rank = min(ranks) expected_change += 0.7 if best_rank <= 10 else 0.4 if best_rank <= 30 else 0.2 expected_change = min(expected_change, 6.5) volume_ratio = _number(row.get("volume_ratio")) turnover_rate = _number(row.get("turnover_rate")) amount_million = _number(row.get("amount_million")) confirmation = self._auction_confirmation(row) actual_strength = _number(row.get("change")) + confirmation label = self._expectation_label(actual_strength, expected_change) is_broken = "昨日炸板" in candidate["sources"] and "昨日涨停" not in candidate["sources"] identity = f"昨日{streak}板" if streak > 1 else "昨日首板" if streak == 1 else "昨日炸板" if is_broken else "人气榜标的" popularity = "，双榜共识" if len(ranks) == 2 else "，热榜靠前" if ranks and min(ranks) <= 10 else "" difference = _number(row.get("change")) - expected_change direction = "高于" if difference > 0 else "低于" if difference < 0 else "贴合" reason = ( f"{identity}{popularity}；竞价涨幅{direction}预期中枢" f"{abs(difference):.1f}个百分点，量比{volume_ratio:.2f}" ) tags = sorted(core_tags.get(str(row.get("code") or ""), set())) scored_row = { **row, "concepts": candidate["concepts"], } attention_score = self._attention_score( scored_row, expected_change, tags, candidate["sources"], streak, str(candidate.get("sector") or row.get("sector") or "") in strong_sector_names, ) normalized.append( { **scored_row, "sector": candidate["sector"] if candidate["sector"] != "其他" else row.get("sector", "其他"), "candidate_sources": candidate["sources"], "source_label": " · ".join(candidate["sources"]), "prior_streak": streak, "concepts": candidate["concepts"], "expected_change": round(expected_change, 2), "actual_strength": round(actual_strength, 2), "expectation": label, "attention_score": attention_score, "core_tags": tags, "is_market_core": bool(tags), "expectation_reason": reason, } ) normalized.sort(key=lambda item: (_number(item.get("attention_score")), _number(item.get("amount_million"))), reverse=True) matched_top = { str(item.get("code") or "") for item in sorted( (item for item in normalized if item.get("expectation") == "符合预期"), key=lambda item: _number(item.get("attention_score")), reverse=True, )[:20] } focus_candidates = [ item for item in normalized if item.get("is_market_core") or (_number(item.get("attention_score")) >= 55 and item.get("expectation") != "符合预期") or str(item.get("code") or "") in matched_top ] mandatory = [item for item in focus_candidates if item.get("is_market_core")] mandatory_codes = {str(item.get("code") or "") for item in mandatory} optional = [item for item in focus_candidates if str(item.get("code") or "") not in mandatory_codes] focus_rows = sorted(mandatory, key=lambda item: _number(item.get("attention_score")), reverse=True) focus_rows.extend(optional[:max(0, 30 - len(focus_rows))]) focus_rows.sort(key=lambda item: _number(item.get("attention_score")), reverse=True) return normalized, { "baseline_date": _display_date(baseline_date), "prior_limit_count": len(prior_limits), "prior_broken_count": len(prior_broken), "hot_candidate_count": sum( any(source in {"同花顺热榜", "东方财富热榜"} for source in item["sources"]) for item in candidates.values() ), "core_count": sum(bool(item.get("is_market_core")) for item in normalized), "notice": "；".join(errors), }, focus_rows @staticmethod def _auction_theme_evidence( prior_snapshot: dict[str, Any], candidate_rows: list[dict[str, Any]], ) -> dict[str, list[dict[str, Any]]]: prior_sectors = list(prior_snapshot.get("sectors") or []) carry = [] for sector in prior_sectors[:10]: name = str(sector.get("name") or "其他") matched = [row for row in candidate_rows if str(row.get("sector") or "其他") == name] changes = [_number(row.get("change")) for row in matched] middle = median(changes) if changes else -10.0 positive_rate = sum(value > 0.2 for value in changes) / len(changes) * 100 if changes else 0.0 if middle >= 2 and positive_rate >= 60: status = "强承接" elif middle >= 0 and positive_rate >= 50: status = "有承接" elif middle > -2: status = "分歧" else: status = "承接弱" carry.append( { "name": name, "status": status, "prior_limit_count": int(_number(sector.get("count"))), "leader": str(sector.get("leader") or "--"), "matched_count": len(matched), "median_change": round(middle, 2) if matched else None, "positive_rate": round(positive_rate, 1), "amount_million": round(sum(_number(row.get("amount_million")) for row in matched), 2), } ) concept_groups: dict[str, list[dict[str, Any]]] = {} prior_names = {str(item.get("name") or "") for item in prior_sectors} for row in candidate_rows: for concept in row.get("concepts") or []: if concept and concept not in prior_names: concept_groups.setdefault(str(concept), []).append(row) new_themes = [] for name, members in concept_groups.items(): unique = {str(item.get("code") or ""): item for item in members} values = list(unique.values()) changes = [_number(item.get("change")) for item in values] if len(values) < 2 or median(changes) < 2 or sum(value > 0.2 for value in changes) / len(values) < 0.67: continue new_themes.append( { "name": name, "stock_count": len(values), "median_change": round(median(changes), 2), "amount_million": round(sum(_number(item.get("amount_million")) for item in values), 2), "leaders": [str(item.get("name") or "--") for item in sorted(values, key=lambda value: _number(value.get("change")), reverse=True)[:3]], } ) new_themes.sort(key=lambda item: (item["stock_count"], item["median_change"], item["amount_million"]), reverse=True) return {"carry": carry, "new_themes": new_themes[:8]} def _auction_amount_history(self, trade_date: str) -> list[dict[str, Any]]: dates = self.database.auction_factor_dates(trade_date, 10) stock_list_dates = { str(item.get("ts_code") or ""): str(item.get("list_date") or "") for item in self.database.list_stock_master() if item.get("ts_code") } history = [] for current_date in dates: rows = [ row for row in self.database.auction_factors_for_date(current_date) if ( str(row.get("ts_code") or "") in stock_list_dates and ( not stock_list_dates[str(row.get("ts_code") or "")] or stock_list_dates[str(row.get("ts_code") or "")] < current_date ) ) ] history.append( { "trade_date": _display_date(current_date), "amount_billion": round(sum(_number(row.get("amount")) for row in rows) / 100_000_000, 2), "stock_count": len(rows), } ) return history def _ensure_auction_amount_history(self, trade_date: str, target_days: int = 10) -> None: existing = set(self.database.auction_factor_dates(trade_date, target_days + 5)) if len(existing) >= target_days: return end = datetime.strptime(trade_date, "%Y%m%d") start = (end - timedelta(days=35)).strftime("%Y%m%d") try: calendar = self.client.query( "trade_cal", { "exchange": "SSE", "start_date": start, "end_date": trade_date, "is_open": 1, }, "cal_date,is_open", ) except TushareError: return dates = sorted( str(item.get("cal_date") or "") for item in calendar if int(_number(item.get("is_open"))) == 1 and item.get("cal_date") )[-target_days:] for current_date in dates: if current_date in existing: continue try: rows = self.client.query( "stk_auction", {"trade_date": current_date}, "ts_code,trade_date,vol,price,amount,pre_close,turnover_rate,volume_ratio,float_share", ) except TushareError: break if rows: self.database.upsert_auction_factors(rows) existing.add(current_date) def _with_auction_watchlist( self, result: dict[str, Any], trade_date: str, user_id: int, ) -> dict[str, Any]: personalized = copy.deepcopy(result) if not user_id: personalized["watchlist_rows"] = [] personalized["watchlist_missing_count"] = 0 return personalized watched = self.database.list_watchlist(user_id) if not watched: personalized["watchlist_rows"] = [] personalized["watchlist_missing_count"] = 0 return personalized public_rows = { str(item.get("code") or ""): item for item in ( list(personalized.get("rows") or []) + list(personalized.get("one_price_rows") or []) ) } factors = { str(item.get("ts_code") or "").split(".")[0]: item for item in self.database.auction_factors_for_date(trade_date) } master = { str(item.get("ts_code") or "").split(".")[0]: item for item in self.database.list_stock_master() } rows = [] missing = 0 for item in watched: code = str(item.get("code") or "") if code in public_rows: rows.append({**public_rows[code], "is_watchlist": True}) continue factor = factors.get(code) if not factor: missing += 1 rows.append( { "code": code, "name": str(item.get("name") or "--"), "sector": str(item.get("sector") or "其他"), "available": False, "is_watchlist": True, } ) continue stock = master.get(code, {}) price = _number(factor.get("price")) pre_close = _number(factor.get("pre_close")) change = (price / pre_close - 1) * 100 if price > 0 and pre_close > 0 else 0 row = { "code": code, "ts_code": str(factor.get("ts_code") or ""), "name": str(item.get("name") or stock.get("name") or "--"), "sector": str(item.get("sector") or stock.get("industry") or "其他"), "price": round(price, 2), "pre_close": round(pre_close, 2), "change": round(change, 2), "amount_million": round(_number(factor.get("amount")) / 1_000_000, 2), "turnover_rate": round(_number(factor.get("turnover_rate")), 4), "volume_ratio": round(_number(factor.get("volume_ratio")), 2), "candidate_sources": ["我的自选"], "source_label": "我的自选", "prior_streak": 0, "concepts": [], "expected_change": 0.0, "core_tags": [], "is_market_core": False, "is_watchlist": True, "available": True, } actual_strength = change + self._auction_confirmation(row) row["actual_strength"] = round(actual_strength, 2) row["expectation"] = self._expectation_label(actual_strength, 0.0) row["attention_score"] = self._attention_score(row, 0.0, [], ["我的自选"], 0, False) direction = "高于" if change > 0 else "低于" if change < 0 else "贴合" row["expectation_reason"] = f"自选观察；竞价涨幅{direction}个人观察基准{abs(change):.1f}个百分点，量比{row['volume_ratio']:.2f}" rows.append(row) rows.sort( key=lambda row: (bool(row.get("available", True)), _number(row.get("attention_score"))), reverse=True, ) personalized["watchlist_rows"] = rows personalized["watchlist_missing_count"] = missing return personalized def _dynamic_auction_rows( self, trade_date: str, baseline_date: str, user_id: int, ) -> list[dict[str, Any]]: if not self.ifind or not self.ifind.configured: return [] master = self._stock_master() placeholders = [ { "code": str(item.get("code") or ts_code.split(".")[0]), "ts_code": ts_code, "name": str(item.get("name") or "--"), "sector": str(item.get("industry") or "其他"), } for ts_code, item in master.items() ] candidates, _, _ = self._auction_candidates(placeholders, baseline_date) selected_codes = { str(item.get("ts_code") or "") for item in candidates if item.get("ts_code") } if user_id: watched = {str(item.get("code") or "") for item in self.database.list_watchlist(user_id)} selected_codes.update( ts_code for ts_code in master if ts_code.split(".")[0] in watched ) selected_codes.discard("") if not selected_codes: return [] display_date = _display_date(trade_date) now = self._now_provider() if now.tzinfo is None: now = now.replace(tzinfo=CHINA_TIMEZONE) else: now = now.astimezone(CHINA_TIMEZONE) end_time = min(now.time().replace(tzinfo=None), dt_time(9, 25)) end_stamp = f"{display_date} {end_time.strftime('%H:%M:%S')}" start_stamp = f"{display_date} 09:15:00" snapshot_rows: list[dict[str, Any]] = [] ordered_codes = sorted(selected_codes) for index in range(0, len(ordered_codes), 80): try: snapshot_rows.extend( self.ifind.snapshots( ordered_codes[index:index + 80], [ "latest", "volume", "amount", "preClose", "bid1", "bidSize1", "ask1", "askSize1", ], start_stamp, end_stamp, cache_ttl=8, ) ) except IfindError: continue latest: dict[str, dict[str, Any]] = {} for row in snapshot_rows: ts_code = str(row.get("thscode") or "") previous = latest.get(ts_code) or {} if ( ts_code and _number(row.get("latest")) > 0 and str(row.get("time") or "") >= str(previous.get("time") or "") ): latest[ts_code] = row prior_factors = { str(item.get("ts_code") or ""): item for item in self.database.auction_factors_for_date(baseline_date) } normalized = [] for ts_code, row in latest.items(): price = _number(row.get("latest")) pre_close = _number(row.get("preClose")) volume = _number(row.get("volume")) bid_size = _number(row.get("bidSize1")) ask_size = _number(row.get("askSize1")) if volume <= 0 and bid_size > 0 and ask_size > 0: volume = min(bid_size, ask_size) amount = _number(row.get("amount")) if amount <= 0 and price > 0 and volume > 0: amount = price * volume prior_volume = _number((prior_factors.get(ts_code) or {}).get("vol")) normalized.append( { "ts_code": ts_code, "trade_date": trade_date, "vol": volume, "price": price, "amount": amount, "pre_close": pre_close, "turnover_rate": 0, "volume_ratio": volume / prior_volume if prior_volume > 0 else 0, "float_share": 0, "bid_size1": bid_size, "ask_size1": ask_size, "snapshot_time": str(row.get("time") or ""), "dynamic": True, } ) return normalized 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) def _theme_directory(self) -> list[dict[str, Any]]: cached = self.database.get_data_snapshot("theme_directory_v1", "ths") or {} if cached.get("items"): return list(cached["items"]) rows = self.client.query( "ths_index", {}, "ts_code,name,count,exchange,list_date,type" ) items = [ { "code": str(row.get("ts_code") or ""), "name": str(row.get("name") or ""), "member_count": int(_number(row.get("count"))), "list_date": str(row.get("list_date") or ""), } for row in rows if str(row.get("type") or "").upper() == "N" and str(row.get("exchange") or "").upper() == "A" and row.get("ts_code") and row.get("name") ] self.database.save_data_snapshot( "theme_directory_v1", "ths", "market", {"items": items} ) return items def theme_library(self, requested_date: str, force: bool = False) -> dict[str, Any]: trade_date, previous_date = self._trade_context(requested_date) if not force: cached = self.database.get_data_snapshot("theme_library_v1", trade_date) if cached: result = copy.deepcopy(cached) result["meta"] = {**result.get("meta", {}), "cached": True} return result try: daily = self.client.query( "ths_daily", {"trade_date": trade_date}, "ts_code,trade_date,open,high,low,close,pre_close,pct_change,vol,turnover_rate", ) except TushareError: fallback = self._latest_feature_snapshot("theme_library_v1", trade_date) if fallback: result = copy.deepcopy(fallback) result["meta"] = { **result.get("meta", {}), "requested_date": _display_date(requested_date), "carried_forward": True, "cached": True, "notice": "当前题材行情暂不可用，展示最近有效快照", } return result daily = [] actual_date = trade_date carried_forward = False if not daily and previous_date: try: daily = self.client.query( "ths_daily", {"trade_date": previous_date}, "ts_code,trade_date,open,high,low,close,pre_close,pct_change,vol,turnover_rate", ) except TushareError: daily = [] actual_date = previous_date carried_forward = bool(daily) daily_map = {str(row.get("ts_code") or ""): row for row in daily} try: hot_rows = self.client.query("ths_hot", {"trade_date": actual_date}) except TushareError: hot_rows = [] hot_map = { str(row.get("ts_code") or ""): int(_number(row.get("rank"))) for row in hot_rows if str(row.get("data_type") or "") == "概念板块" } items = [] for item in self._theme_directory(): quote = daily_map.get(item["code"], {}) items.append( { **item, "change": round(_number(quote.get("pct_change")), 2), "close": round(_number(quote.get("close")), 3), "turnover_rate": round(_number(quote.get("turnover_rate")), 2), "volume": round(_number(quote.get("vol")), 2), "hot_rank": hot_map.get(item["code"]), "has_quote": bool(quote), } ) items.sort( key=lambda item: ( item["has_quote"], item["hot_rank"] is not None, -(item["hot_rank"] or 9999), item["change"], ), reverse=True, ) quoted = [item for item in items if item["has_quote"]] result = { "meta": { "requested_date": _display_date(requested_date), "trade_date": _display_date(actual_date), "carried_forward": carried_forward, "cached": False, "notice": "" if quoted else "该交易日暂无题材行情，已保留题材目录", "updated_at": datetime.now().astimezone().isoformat(timespec="seconds"), }, "summary": { "theme_count": len(items), "quoted_count": len(quoted), "up_count": sum(item["change"] > 0 for item in quoted), "down_count": sum(item["change"] < 0 for item in quoted), "hot_count": len(hot_map), }, "items": items, } self.database.save_data_snapshot("theme_library_v1", trade_date, "market", result) return result def theme_detail(self, code: str, requested_date: str) -> dict[str, Any]: code = str(code or "").strip().upper() library = self.theme_library(requested_date) theme = next((item for item in library["items"] if item["code"] == code), None) if not theme: raise ValueError("未找到对应题材。") actual_date = str(library["meta"]["trade_date"]).replace("-", "") detail_key = f"{actual_date}:{code}" cached_detail = self.database.get_data_snapshot("theme_detail_v1", detail_key) if cached_detail: return cached_detail try: members = self.client.query( "ths_member", {"ts_code": code, "is_new": "Y"}, "ts_code,con_code,con_name" ) except TushareError: members = [] bars = self.database.daily_bars_for_date(actual_date) if not bars: bars = self.client.query( "daily", {"trade_date": actual_date}, "ts_code,trade_date,open,high,low,close,pct_chg,vol,amount", ) self.database.upsert_daily_bars(bars) bar_map = {str(row.get("ts_code") or ""): row for row in bars} normalized_members = [] for member in members: ts_code = str(member.get("con_code") or "") quote = bar_map.get(ts_code, {}) normalized_members.append( { "code": ts_code.split(".")[0], "ts_code": ts_code, "name": str(member.get("con_name") or "--"), "price": round(_number(quote.get("close")), 2), "change": round(_number(quote.get("pct_chg")), 2), "amount_billion": round(_number(quote.get("amount")) / 100_000, 2), "has_quote": bool(quote), } ) normalized_members.sort( key=lambda item: (item["has_quote"], item["change"], item["amount_billion"]), reverse=True, ) end = datetime.strptime(actual_date, "%Y%m%d") try: history = self.client.query( "ths_daily", { "ts_code": code, "start_date": (end - timedelta(days=190)).strftime("%Y%m%d"), "end_date": actual_date, }, "ts_code,trade_date,open,high,low,close,pct_change,vol,turnover_rate", ) except TushareError: history = [] history.sort(key=lambda row: str(row.get("trade_date") or "")) series = [ { "trade_date": _display_date(str(row.get("trade_date") or "")), "open": _number(row.get("open")), "high": _number(row.get("high")), "low": _number(row.get("low")), "close": _number(row.get("close")), "change": _number(row.get("pct_change")), "volume": _number(row.get("vol")), } for row in history[-90:] ] result = { "meta": { "trade_date": _display_date(actual_date), "notice": "" if members or history else "题材成分与走势暂不可用", }, "theme": theme, "series": series, "members": normalized_members, "summary": { "member_count": len(normalized_members), "up_count": sum(item["change"] > 0 for item in normalized_members if item["has_quote"]), "down_count": sum(item["change"] < 0 for item in normalized_members if item["has_quote"]), "quoted_count": sum(item["has_quote"] for item in normalized_members), }, } if members or history: self.database.save_data_snapshot("theme_detail_v1", detail_key, "market", result) return result @staticmethod def _parse_concepts(value: Any) -> list[str]: if isinstance(value, list): return [str(item) for item in value if str(item).strip()] text = str(value or "").strip() if not text: return [] try: parsed = json.loads(text) if isinstance(parsed, list): return [str(item) for item in parsed if str(item).strip()] except json.JSONDecodeError: pass return [part.strip() for part in text.split(",") if part.strip()] def popularity(self, requested_date: str, force: bool = False) -> dict[str, Any]: trade_date, previous_date = self._trade_context(requested_date) if not force: cached = self.database.get_data_snapshot("popularity_v1", trade_date) if cached: result = copy.deepcopy(cached) result["meta"] = {**result.get("meta", {}), "cached": True} return result ths_rows, dc_rows, errors = self._hot_rows(trade_date) actual_date = trade_date carried_forward = False if not ths_rows and not dc_rows and previous_date: ths_rows, dc_rows, errors = self._hot_rows(previous_date) actual_date = previous_date carried_forward = bool(ths_rows or dc_rows) if not ths_rows and not dc_rows: fallback = self._latest_feature_snapshot("popularity_v1", trade_date) if fallback: result = copy.deepcopy(fallback) result["meta"] = { **result.get("meta", {}), "requested_date": _display_date(requested_date), "carried_forward": True, "cached": True, "notice": "当前榜单暂不可用，展示最近有效快照", } return result return { "meta": { "requested_date": _display_date(requested_date), "trade_date": _display_date(trade_date), "previous_trade_date": _display_date(previous_date), "carried_forward": False, "cached": False, "notice": "该交易日暂无可用人气榜", "updated_at": datetime.now().astimezone().isoformat(timespec="seconds"), }, "summary": {"ths_count": 0, "dc_count": 0, "dual_count": 0}, "combined": [], "ths": [], "dc": [], } prior_request = (datetime.strptime(actual_date, "%Y%m%d") - timedelta(days=1)).strftime("%Y%m%d") prior_date, _ = self._trade_context(prior_request) previous_ths, previous_dc, _ = self._hot_rows(prior_date) ths = self._normalize_hot(ths_rows, "热股", previous_ths) dc = self._normalize_hot(dc_rows, "A股市场", previous_dc) ths_map = {item["ts_code"]: item for item in ths} dc_map = {item["ts_code"]: item for item in dc} combined = [] for ts_code in set(ths_map) | set(dc_map): ths_item = ths_map.get(ts_code) dc_item = dc_map.get(ts_code) base = ths_item or dc_item or {} ths_rank = int(ths_item["rank"]) if ths_item else None dc_rank = int(dc_item["rank"]) if dc_item else None score = ( (101 - (ths_rank or 101)) * 0.5 + (201 - (dc_rank or 201)) * 0.25 ) combined.append( { **base, "ths_rank": ths_rank, "dc_rank": dc_rank, "score": round(score, 2), "dual_source": bool(ths_item and dc_item), "concepts": (ths_item or {}).get("concepts") or [], } ) combined.sort(key=lambda item: (item["dual_source"], item["score"]), reverse=True) for index, item in enumerate(combined, 1): item["rank"] = index result = { "meta": { "requested_date": _display_date(requested_date), "trade_date": _display_date(actual_date), "previous_trade_date": _display_date(prior_date), "carried_forward": carried_forward, "cached": False, "updated_at": datetime.now().astimezone().isoformat(timespec="seconds"), "notice": "；".join(errors), }, "summary": { "ths_count": len(ths), "dc_count": len(dc), "dual_count": sum(item["dual_source"] for item in combined), }, "combined": combined[:200], "ths": ths, "dc": dc, } self.database.save_data_snapshot("popularity_v1", trade_date, "market", result) return result def _hot_rows(self, trade_date: str) -> tuple[list[dict[str, Any]], list[dict[str, Any]], list[str]]: errors = [] try: ths = self.client.query("ths_hot", {"trade_date": trade_date}) except TushareError: ths = [] errors.append("同花顺榜单暂不可用") try: dc = self.client.query("dc_hot", {"trade_date": trade_date}) except TushareError: dc = [] errors.append("东方财富榜单暂不可用") return ths, dc, errors def _normalize_hot( self, rows: list[dict[str, Any]], data_type: str, previous_rows: list[dict[str, Any]], ) -> list[dict[str, Any]]: previous = { str(row.get("ts_code") or ""): int(_number(row.get("rank"))) for row in previous_rows if str(row.get("data_type") or "") == data_type } items = [] for row in rows: if str(row.get("data_type") or "") != data_type: continue rank = int(_number(row.get("rank"))) ts_code = str(row.get("ts_code") or "") prior_rank = previous.get(ts_code) items.append( { "rank": rank, "ts_code": ts_code, "code": ts_code.split(".")[0], "name": str(row.get("ts_name") or "--"), "change": round(_number(row.get("pct_change")), 2), "price": round(_number(row.get("current_price")), 2), "hot": round(_number(row.get("hot")), 1), "rank_change": (prior_rank - rank) if prior_rank else None, "concepts": self._parse_concepts(row.get("concept")), "reason": str(row.get("rank_reason") or ""), "rank_time": str(row.get("rank_time") or ""), } ) items.sort(key=lambda item: item["rank"]) return items