from __future__ import annotations import json from datetime import datetime, time from typing import Any from zoneinfo import ZoneInfo from backend.data.contracts import SnapshotState from backend.data.gateway import DataGateway, MarketDataUnavailable from backend.data.providers.base import ProviderError from backend.data.repository import MarketRepository from backend.database.connection import Database from backend.features.market.insights.auction import build_auction, build_watchlist_rows from backend.features.market.insights.dragon import build_dragon_list from backend.features.market.insights.popularity import build_popularity from backend.features.market.insights.support import ( auction_phase as _auction_phase, ) from backend.features.market.insights.support import ( clock as _clock, ) from backend.features.market.insights.support import ( decorate as _decorate, ) from backend.features.market.insights.support import ( empty_auction as _empty_auction, ) from backend.features.market.insights.support import ( empty_standard as _empty_standard, ) from backend.features.market.insights.support import ( number as _number, ) from backend.features.market.insights.support import ( result as _result, ) from backend.features.market.insights.support import ( rows as _rows, ) from backend.features.market.insights.support import ( serialized_rows as _serialized_rows, ) from backend.features.market.insights.support import ( standard as _standard, ) from backend.features.market.insights.support import ( tuple_or_none as _tuple_or_none, ) from backend.features.market.insights.support import ( valid_date as _date, ) from backend.features.market.insights.themes import build_theme_detail, build_theme_library SHANGHAI = ZoneInfo("Asia/Shanghai") class MarketInsightError(RuntimeError): pass class MarketInsightService: def __init__( self, database: Database, repository: MarketRepository, gateway: DataGateway ) -> None: self._database = database self._repository = repository self._gateway = gateway def workspace( self, key: str, requested_date: str | None = None, *, user_id: int, force: bool = False, now: datetime | None = None, ) -> dict[str, Any]: if key == "auction": return self.auction(requested_date, user_id=user_id, force=force, now=now) if key == "themes": return self.themes(requested_date, force=force) if key == "popularity": return self.popularity(requested_date, force=force) if key == "dragon-list": return self.dragon_list(requested_date, force=force) raise MarketInsightError("不支持的市场洞察工作区") def auction( self, requested_date: str | None, *, user_id: int, force: bool = False, now: datetime | None = None, ) -> dict[str, Any]: clock = _clock(now) requested, trade_date, previous = self._trade_dates(requested_date, clock) phase = _auction_phase(requested, trade_date, clock) target = previous if phase == "pending" else trade_date baseline = self._previous_date(target) cached = self._snapshot("auction", target) if cached and not force and phase not in {"observing", "selection"}: return _decorate( self._personalize_auction(cached, user_id), requested, phase, carried_forward=target != requested, message=( "今日竞价尚未开始，显示前一交易日归档" if phase == "pending" else "" ), ) inputs = self._gateway.insight_inputs("auction", target, baseline) raw = _rows(inputs.get("auction")) dynamic = False observed_at = clock if phase in {"observing", "selection"}: identifiers = self._auction_universe(baseline, inputs) end = time(9, 25) if phase == "selection" else clock.time().replace(tzinfo=None) try: live = self._gateway.dynamic_auction( identifiers, f"{target} 09:15:00", f"{target} {end.strftime('%H:%M:%S')}", ) raw = live.rows observed_at = live.metadata.observed_at dynamic = True except (MarketDataUnavailable, ProviderError): if phase == "observing": prior = self._snapshot("auction", previous) if prior: return _decorate( self._personalize_auction(prior, user_id), requested, phase, carried_forward=True, message="今日动态竞价暂不可用，当前显示前一交易日归档", current_available=False, ) return _empty_auction( requested, previous, phase, "今日动态竞价暂不可用，且没有历史归档", ) if not raw: if cached: return _decorate( self._personalize_auction(cached, user_id), requested, phase, False, "当前读取失败，保留真实归档", ) return _empty_auction(requested, target, phase, "该交易日暂无可用竞价快照") payload = build_auction( trade_date=target, raw_rows=raw, price_limits=_rows(inputs.get("price_limits")), directory=self._gateway.stock_directory(), prior_snapshot=self._market_snapshot(baseline), ths_hot=_rows(inputs.get("ths_hot")), dc_hot=_rows(inputs.get("dc_hot")), history=self._auction_history(target), dynamic=dynamic, ) minimum = 0.8 if phase == "observing" else 0.9 if float(payload["coverage"]) < minimum: if cached: return _decorate( self._personalize_auction(cached, user_id), requested, phase, False, f"竞价覆盖率不足{minimum * 100:.0f}%，保留原有真实归档", ) return _empty_auction( requested, target, phase, f"竞价覆盖率不足{minimum * 100:.0f}%，未形成正式结果", ) state = ( SnapshotState.REALTIME if phase == "observing" else SnapshotState.FINAL if target == clock.date().isoformat() else SnapshotState.ARCHIVE ) payload["observed_at"] = observed_at.isoformat(timespec="seconds") payload["state"] = state.value if phase != "observing": source = "ifind" if dynamic else "tushare" self._save("auction", target, "", payload, state, source, payload["coverage"]) return _decorate( self._personalize_auction(payload, user_id), requested, phase, target != requested, "", ) def _personalize_auction( self, payload: dict[str, Any], user_id: int ) -> dict[str, Any]: result = {**payload} market_rows = list(result.pop("_market_rows", ())) with self._database.read() as connection: watchlist = tuple( dict(row) for row in self._repository.watchlist(connection, user_id) ) result["watchlist_rows"] = build_watchlist_rows( market_rows, list(result.get("rows") or ()), list(result.get("one_price_rows") or ()), watchlist, ) result["watchlist_ready"] = bool(market_rows) or not watchlist return result def themes(self, requested_date: str | None, *, force: bool = False) -> dict[str, Any]: requested, trade_date, _ = self._trade_dates(requested_date) cached = self._snapshot("themes", trade_date) if cached and not force: return _standard(cached, requested) inputs = self._gateway.insight_inputs("themes", trade_date) directory = _result(inputs.get("directory")) daily = _result(inputs.get("daily")) hot = _result(inputs.get("hot")) if directory is None: fallback = self._latest_snapshot("themes", trade_date) if fallback: return _standard(fallback, requested, "当前题材目录暂不可用，显示最近有效榜单") raise MarketInsightError("题材目录暂不可用") payload = build_theme_library( trade_date, directory.rows, daily.rows if daily else (), hot.rows if hot else (), ) payload["observed_at"] = directory.metadata.observed_at.isoformat(timespec="seconds") payload["state"] = SnapshotState.ARCHIVE.value payload["message"] = "" if daily and daily.rows else "该交易日暂无题材行情" with self._database.transaction() as connection: self._repository.replace_themes( connection, list(payload["items"]), directory.metadata.source.value, payload["observed_at"], ) self._save("themes", trade_date, "", payload, SnapshotState.ARCHIVE, "tushare", 1) return _standard(payload, requested) def theme_detail(self, identifier: str, requested_date: str | None) -> dict[str, Any]: library = self.themes(requested_date) code = identifier.strip().upper() theme = next((item for item in library["items"] if item["code"] == code), None) if theme is None: raise MarketInsightError("未找到该题材") trade_date = str(library["trade_date"]) cached = self._snapshot("themes", trade_date, code) if cached: return cached inputs = self._gateway.insight_inputs("theme-detail", trade_date, identifier=code) members = _result(inputs.get("members")) daily = _result(inputs.get("daily")) payload = build_theme_detail( trade_date, theme, members.rows if members else (), daily.rows if daily else (), ) payload["message"] = "" if members and members.rows else "该题材暂无可核验成分股" payload["observed_at"] = ( members.metadata.observed_at if members else datetime.now(SHANGHAI) ).isoformat(timespec="seconds") payload["state"] = SnapshotState.ARCHIVE.value if members is not None: self._save( "themes", trade_date, code, payload, SnapshotState.ARCHIVE, "tushare", 1 ) return payload def popularity( self, requested_date: str | None, *, force: bool = False ) -> dict[str, Any]: requested, trade_date, previous = self._trade_dates(requested_date) cached = self._snapshot("popularity", trade_date) if cached and not force: return _standard(cached, requested) inputs = self._gateway.insight_inputs("popularity", trade_date, previous) ths = _result(inputs.get("ths")) dc = _result(inputs.get("dc")) if not ((ths and ths.rows) or (dc and dc.rows)): fallback = self._latest_snapshot("popularity", previous) if fallback: return _standard(fallback, requested, "当日榜单尚未生成，显示最近有效榜单") return _empty_standard(requested, trade_date, "该交易日暂无可用人气榜") payload = build_popularity( trade_date, ths.rows if ths else (), dc.rows if dc else (), _rows(inputs.get("previous_ths")), _rows(inputs.get("previous_dc")), ) payload["observed_at"] = datetime.now(SHANGHAI).isoformat(timespec="seconds") payload["state"] = SnapshotState.ARCHIVE.value missing = [] if ths is None: missing.append("同花顺榜单暂不可用") if dc is None: missing.append("东方财富榜单暂不可用") payload["message"] = "；".join(missing) coverage = (int(ths is not None) + int(dc is not None)) / 2 self._save( "popularity", trade_date, "", payload, SnapshotState.ARCHIVE, "tushare", coverage, ) return _standard(payload, requested) def dragon_list( self, requested_date: str | None, *, force: bool = False ) -> dict[str, Any]: requested, trade_date, previous = self._trade_dates(requested_date) raw = self._snapshot("dragon-list", trade_date) if raw is None or force: inputs = self._gateway.insight_inputs("dragon-list", trade_date) raw = { "trade_date": trade_date, "observed_at": datetime.now(SHANGHAI).isoformat(timespec="seconds"), "state": SnapshotState.ARCHIVE.value, "official": _serialized_rows(inputs.get("official")), "profiles": _serialized_rows(inputs.get("profiles")), "stocks": _serialized_rows(inputs.get("stocks")), "seats": _serialized_rows(inputs.get("seats")), } coverage = sum(value is not None for value in raw.values() if isinstance(value, list)) self._save( "dragon-list", trade_date, "", raw, SnapshotState.ARCHIVE, "tushare", min(coverage / 4, 1), ) with self._database.read() as connection: aliases = self._repository.seat_aliases(connection) result = build_dragon_list( trade_date=trade_date, official_rows=_tuple_or_none(raw.get("official")), profile_rows=_tuple_or_none(raw.get("profiles")), stock_rows=_tuple_or_none(raw.get("stocks")), seat_rows=_tuple_or_none(raw.get("seats")), aliases=aliases, ) result.update( { "requested_date": requested, "previous_date": previous, "observed_at": raw.get("observed_at"), "state": SnapshotState.ARCHIVE.value, "carried_forward": False, } ) return result def save_seat_alias(self, seat_name: str, alias_name: str, user_id: int) -> dict[str, str]: seat = " ".join(seat_name.split()) alias = " ".join(alias_name.split()) if not seat or not alias: raise MarketInsightError("营业部和游资名称不能为空") with self._database.transaction() as connection: self._repository.save_seat_alias( connection, seat, alias, datetime.now(SHANGHAI).isoformat(timespec="seconds"), user_id, ) return {"seat_name": seat, "alias_name": alias} def _trade_dates( self, requested_date: str | None, clock: datetime | None = None ) -> tuple[str, str, str]: try: requested = _date( requested_date or (clock or datetime.now(SHANGHAI)).date().isoformat() ) except ValueError as exc: raise MarketInsightError("日期格式无效") from exc dates = self._gateway.trading_dates(requested, 2) if len(dates) < 2: raise MarketInsightError("请先同步完整交易日历") return requested, dates[0], dates[1] def _previous_date(self, trade_date: str) -> str: dates = self._gateway.trading_dates(trade_date, 2) if len(dates) < 2: raise MarketInsightError("缺少前一交易日") return dates[1] def _auction_universe( self, baseline: str, inputs: dict[str, Any] ) -> tuple[str, ...]: snapshot = self._market_snapshot(baseline) codes = { str(item.get("identifier") or "") for key in ("limits", "broken") for item in snapshot.get(key) or [] } for key, data_type in (("ths_hot", "热股"), ("dc_hot", "A股市场")): for row in _rows(inputs.get(key)): valid_type = str(row.get("data_type") or "") == data_type top_twenty = int(_number(row.get("rank"), 9999)) <= 20 if valid_type and top_twenty: codes.add(str(row.get("ts_code") or "")) return tuple(sorted(code for code in codes if code)) def _market_snapshot(self, trade_date: str) -> dict[str, Any]: with self._database.read() as connection: row = self._repository.latest_summary(connection, trade_date) if row is None or str(row["trade_date"]) != trade_date: return {} return json.loads(str(row["payload_json"])) def _auction_history(self, trade_date: str) -> list[dict[str, Any]]: with self._database.read() as connection: rows = self._repository.insight_snapshots(connection, "auction", trade_date, 10) result = [] for row in rows: payload = json.loads(str(row["payload_json"])) summary = payload.get("summary") or {} result.append( { "trade_date": str(row["trade_date"]), "amount_billion": _number(summary.get("amount_billion")), "stock_count": int(summary.get("stock_count") or 0), } ) return result def _snapshot( self, kind: str, trade_date: str, entity_key: str = "" ) -> dict[str, Any] | None: with self._database.read() as connection: row = self._repository.insight_snapshot(connection, kind, trade_date, entity_key) return json.loads(str(row["payload_json"])) if row else None def _latest_snapshot( self, kind: str, through: str, entity_key: str = "" ) -> dict[str, Any] | None: with self._database.read() as connection: row = self._repository.latest_insight_snapshot(connection, kind, through, entity_key) return json.loads(str(row["payload_json"])) if row else None def _save( self, kind: str, trade_date: str, entity_key: str, payload: dict[str, Any], state: SnapshotState, source: str, coverage: float, ) -> None: with self._database.transaction() as connection: self._repository.save_insight_snapshot( connection, kind=kind, trade_date=trade_date, entity_key=entity_key, observed_at=str( payload.get("observed_at") or datetime.now(SHANGHAI).isoformat(timespec="seconds") ), state=state.value, source=source, coverage=max(0, min(coverage, 1)), payload=payload, )