from __future__ import annotations import copy from datetime import datetime, timedelta 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 MarketThemeInsightsMixin: 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