from __future__ import annotations import math import time from datetime import date, datetime, time as wall_time, timedelta, timezone from decimal import Decimal from typing import Any, Mapping, Sequence import mysql.connector from .core import LedgerError, PriceRecord from .database import MySQLSettings, connect_market_readonly, safe_mysql_error SHANGHAI = timezone(timedelta(hours=8)) FINAL_LOAD_TIME = wall_time(15, 5) MARKET_SOURCE_ID = "trading_xuntou.cn_stock_kline_1d_front:xtquant:front" REQUIRED_COLUMNS = { "cn_stock_kline_1d_front": { "id", "symbol", "trade_date", "close", "source", "updated_at" }, "formal_trading_calendar_days": { "market", "trade_date", "is_open", "source_kind", "source_fetch_time", "source_version", "asset_version", }, "formal_trading_calendar_assets": { "asset_version", "market", "window_end", "generated_at", "status", "is_current", }, } def _market_schema_contract(connection, database: str) -> dict[str, Any]: cursor = connection.cursor(dictionary=True) try: cursor.execute( "SELECT TABLE_NAME, COLUMN_NAME, IS_NULLABLE, COLUMN_DEFAULT, COLUMN_COMMENT " "FROM information_schema.COLUMNS WHERE TABLE_SCHEMA=%s AND TABLE_NAME IN " "('cn_stock_kline_1d_front','formal_trading_calendar_days'," "'formal_trading_calendar_assets')", (database,), ) rows = cursor.fetchall() finally: cursor.close() actual = {name: set() for name in REQUIRED_COLUMNS} for row in rows: table = row["TABLE_NAME"] if table in actual: actual[table].add(row["COLUMN_NAME"]) missing = { table: sorted(columns - actual[table]) for table, columns in REQUIRED_COLUMNS.items() if columns - actual[table] } if missing: raise LedgerError("E_MARKET_SCHEMA", f"行情 schema 缺少必需字段：{missing}") return { "price_table": "cn_stock_kline_1d_front", "adjustment_semantics": "front_adjusted_dedicated_table", "source_id": MARKET_SOURCE_ID, "calendar_market": "SH", "final_load_time": FINAL_LOAD_TIME.isoformat(timespec="minutes"), } def _as_datetime(value: Any, field: str) -> datetime: if isinstance(value, datetime): return value if isinstance(value, str): try: return datetime.fromisoformat(value.replace("Z", "+00:00")) except ValueError as exc: raise LedgerError("E_MARKET_ROW", f"{field} 不是有效时间") from exc raise LedgerError("E_MARKET_ROW", f"{field} 缺失") def _as_date(value: Any, field: str) -> date: if isinstance(value, datetime): return value.date() if isinstance(value, date): return value if isinstance(value, str): try: return date.fromisoformat(value[:10]) except ValueError as exc: raise LedgerError("E_MARKET_ROW", f"{field} 不是有效日期") from exc raise LedgerError("E_MARKET_ROW", f"{field} 缺失") def _record_from_row( row: Mapping[str, Any], security: Mapping[str, str], as_of: date, process_start: datetime, ) -> PriceRecord: ticker = str(security["ticker"]) symbol = str(row.get("symbol") or "") if symbol != ticker: raise LedgerError("E_MARKET_IDENTITY", "行情代码与证券代码不一致") suffix = ticker.rsplit(".", 1)[-1] if suffix not in {"SH", "SZ", "BJ"}: raise LedgerError("E_MARKET_UNSUPPORTED", "trading_xuntou 首版只覆盖 A 股") if security.get("currency") != "CNY": raise LedgerError("E_CURRENCY", "A 股行情必须使用 CNY") trade_date = _as_date(row.get("trade_date"), "trade_date") if trade_date > as_of: raise LedgerError("E_ASOF", "行情日期晚于请求 as-of") try: close = float(Decimal(str(row.get("close")))) except (ValueError, TypeError): close = math.nan if not math.isfinite(close) or close <= 0: raise LedgerError("E_PRICE", "收盘价必须是有限正数") if row.get("is_open") not in (1, True): raise LedgerError("E_INCOMPLETE_TRADING_DAY", "正式交易日历未证明该日开市") if row.get("asset_status") != "current" or row.get("asset_is_current") not in (1, True): raise LedgerError("E_INCOMPLETE_TRADING_DAY", "正式交易日历资产不是 current") window_end = _as_date(row.get("window_end"), "calendar.window_end") if window_end < as_of: raise LedgerError("E_INCOMPLETE_TRADING_DAY", "正式交易日历水位未覆盖请求 as-of") source_time = _as_datetime(row.get("updated_at"), "updated_at") if source_time.tzinfo is None: source_time = source_time.replace(tzinfo=SHANGHAI) else: source_time = source_time.astimezone(SHANGHAI) completed_after = datetime.combine(trade_date, FINAL_LOAD_TIME, SHANGHAI) process_local = process_start.astimezone(SHANGHAI) if source_time < completed_after or ( trade_date == process_local.date() and process_local < completed_after ): raise LedgerError("E_INCOMPLETE_TRADING_DAY", "日 K 尚无收盘后最终装载水位") source = str(row.get("source") or "") calendar_source = str(row.get("calendar_source_kind") or "").strip() if source != "xtquant" or not calendar_source: raise LedgerError("E_SOURCE", "前复权行情来源必须为 xtquant 且交易日历来源不得缺失") return PriceRecord( ticker=ticker, trade_date=trade_date.isoformat(), close=close, currency="CNY", source_id=MARKET_SOURCE_ID, source_timestamp=source_time.isoformat(), is_complete_trading_day=True, ) def fetch_trading_closes( settings: MySQLSettings, securities: Sequence[dict[str, str]], as_of: str, *, process_start: datetime | None = None, ) -> tuple[list[PriceRecord], list[dict[str, str]], dict[str, Any]]: cutoff = date.fromisoformat(as_of) process_start = process_start or datetime.now().astimezone() if process_start.tzinfo is None: raise ValueError("process_start 必须带时区") started = time.monotonic() failures: list[dict[str, str]] = [] supported: list[dict[str, str]] = [] for item in securities: if str(item["ticker"]).endswith((".SH", ".SZ", ".BJ")): supported.append(item) else: failures.append( {"ticker": item["ticker"], "code": "E_MARKET_UNSUPPORTED", "message": "trading_xuntou 首版只覆盖 A 股"} ) connection = connect_market_readonly(settings) cursor = connection.cursor(dictionary=True) try: contract = _market_schema_contract(connection, settings.market_database) rows: list[dict[str, Any]] = [] if supported: placeholders = ",".join(["%s"] * len(supported)) query = f""" WITH expected AS ( SELECT MAX(c.trade_date) AS trade_date FROM formal_trading_calendar_assets a JOIN formal_trading_calendar_days c ON c.asset_version=a.asset_version AND c.market=a.market WHERE a.market='SH' AND a.status='current' AND a.is_current=1 AND a.window_end >= %s AND c.is_open=1 AND c.trade_date<=%s ), ranked AS ( SELECT k.symbol, k.trade_date, k.close, k.source, k.updated_at, ROW_NUMBER() OVER ( PARTITION BY k.symbol ORDER BY k.trade_date DESC, k.updated_at DESC, k.id DESC ) AS rn FROM cn_stock_kline_1d_front k WHERE k.symbol IN ({placeholders}) AND k.trade_date <= %s ) SELECT r.symbol, r.trade_date, r.close, r.source, r.updated_at, c.is_open, c.source_kind AS calendar_source_kind, a.window_end, a.status AS asset_status, a.is_current AS asset_is_current FROM ranked r JOIN formal_trading_calendar_assets a ON a.market='SH' AND a.status='current' AND a.is_current=1 AND a.window_end >= %s JOIN formal_trading_calendar_days c ON c.asset_version=a.asset_version AND c.market=a.market AND c.trade_date=r.trade_date CROSS JOIN expected e WHERE r.rn=1 AND r.trade_date=e.trade_date """ cursor.execute( query, [cutoff, cutoff] + [item["ticker"] for item in supported] + [cutoff, cutoff], ) rows = cursor.fetchall() by_ticker = {str(row["symbol"]): row for row in rows} prices: list[PriceRecord] = [] for security in supported: ticker = str(security["ticker"]) row = by_ticker.get(ticker) if row is None: failures.append( {"ticker": ticker, "code": "E_MARKET_NO_PROVEN_ROW", "message": "没有匹配请求日历水位且证据完整的前复权日 K"} ) continue try: prices.append(_record_from_row(row, security, cutoff, process_start)) except LedgerError as exc: failures.append({"ticker": ticker, "code": exc.code, "message": str(exc)}) connection.rollback() except mysql.connector.Error as exc: connection.rollback() raise LedgerError("E_MARKET_DATABASE", safe_mysql_error(exc)) from exc finally: cursor.close() connection.close() return ( sorted(prices, key=lambda item: item.ticker), sorted(failures, key=lambda item: item["ticker"]), { "provider": MARKET_SOURCE_ID, "requested": len(securities), "succeeded": len(prices), "failed": len(failures), "elapsed_seconds": time.monotonic() - started, "contract": contract, "select_only": True, }, )