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,
|
},
|
)
|
