fix(data): get_security_info_batch SQL计划劣化治本——INDEXED BY钉死复合索引(运行时探测索引名)——策略session 08-24移交P0:VPS生产EXPLAIN实锤min/max批量SQL被选去单列idx_dbbardata_interval整扫~21M日线行+USE TEMP B-TREE(interval IN列表作废),30只冷跑101s=开盘选股38-51min主根因;复合唯一索引存在但无统计信息计划不选。修=SQL加INDEXED BY钉死(symbol,exchange,interval,datetime);索引名跨环境不同(ensure_schema长名/NAS副本merge_increment建uq_dbbardata)故sqlite_master运行时探测+实例缓存(哨兵区分None),探不到回退裸查询不劣于现状;GROUP BY(symbol,exchange)=索引前缀→TEMP B-TREE消失(测试EXPLAIN断言钉死);+5测试(双索引名/回退/计划签名/hint下batch==逐只等价);628绿。配套:今晚VPS/NAS库ANALYZE灌统计(建议a一并落地) [vps]
This commit is contained in:
@@ -36,6 +36,23 @@ logger = logging.getLogger(__name__)
|
|||||||
_DEFAULT_DB = r"C:\sanguo_vnpy_v2\data\quant_trading.db"
|
_DEFAULT_DB = r"C:\sanguo_vnpy_v2\data\quant_trading.db"
|
||||||
_DEFAULT_DATA_DIR = r"C:\sanguo_vnpy_v2\data"
|
_DEFAULT_DATA_DIR = r"C:\sanguo_vnpy_v2\data"
|
||||||
|
|
||||||
|
# 实例缓存哨兵(区分「未探测」与「探测结果=None」) + CREATE INDEX 列清单解析
|
||||||
|
_UNSET = object()
|
||||||
|
_INDEX_COLS_RE = re.compile(r"\(([^)]*)\)")
|
||||||
|
|
||||||
|
|
||||||
|
def _sqlite_index_columns(index_sql: str) -> List[str]:
|
||||||
|
"""CREATE INDEX 语句 → 小写列名列表(去引号/ASC/DESC 修饰); 解析不了返 []。"""
|
||||||
|
m = _INDEX_COLS_RE.search(index_sql or "")
|
||||||
|
if not m:
|
||||||
|
return []
|
||||||
|
cols: List[str] = []
|
||||||
|
for part in m.group(1).split(","):
|
||||||
|
token = part.strip()
|
||||||
|
if token:
|
||||||
|
cols.append(token.split()[0].strip('"`[]').lower())
|
||||||
|
return cols
|
||||||
|
|
||||||
# jq 后缀 ↔ dbbardata exchange
|
# jq 后缀 ↔ dbbardata exchange
|
||||||
_JQ_SUFFIX_TO_EXC = {"XSHG": "SSE", "XSHE": "SZSE", "SH": "SSE", "SZ": "SZSE"}
|
_JQ_SUFFIX_TO_EXC = {"XSHG": "SSE", "XSHE": "SZSE", "SH": "SSE", "SZ": "SZSE"}
|
||||||
_EXC_TO_JQ_SUFFIX = {"SSE": "XSHG", "SZSE": "XSHE"}
|
_EXC_TO_JQ_SUFFIX = {"SSE": "XSHG", "SZSE": "XSHE"}
|
||||||
@@ -595,6 +612,44 @@ class LocalUnifiedProvider(DataProvider): # type: ignore[misc]
|
|||||||
"type": "stock",
|
"type": "stock",
|
||||||
}
|
}
|
||||||
|
|
||||||
|
def _composite_minmax_index(self, conn: sqlite3.Connection) -> Optional[str]:
|
||||||
|
"""dbbardata (symbol,exchange,interval,datetime) 前缀唯一索引名; 探不到 None。
|
||||||
|
|
||||||
|
2026-08-24 计划劣化治本: 无统计信息时 planner 会选单列 interval 索引整扫
|
||||||
|
~21M 日线行 + USE TEMP B-TREE(VPS 生产 EXPLAIN 实锤; 30 只冷跑 101s,
|
||||||
|
开盘选股 38-51min 主根因)。复合索引名跨环境不同(ensure_schema 建长名 /
|
||||||
|
NAS 副本 merge_increment 建 uq_dbbardata), 故运行时探测 + 实例缓存;
|
||||||
|
探不到回退裸查询(不劣于现状)。
|
||||||
|
"""
|
||||||
|
cached = getattr(self, "_cached_minmax_index", _UNSET)
|
||||||
|
if cached is not _UNSET:
|
||||||
|
return cached # type: ignore[return-value]
|
||||||
|
name: Optional[str] = None
|
||||||
|
try:
|
||||||
|
for iname, isql in conn.execute(
|
||||||
|
"SELECT name, sql FROM sqlite_master "
|
||||||
|
"WHERE type='index' AND tbl_name='dbbardata' AND sql IS NOT NULL"
|
||||||
|
).fetchall():
|
||||||
|
if _sqlite_index_columns(isql)[:4] == [
|
||||||
|
"symbol", "exchange", "interval", "datetime",
|
||||||
|
]:
|
||||||
|
name = iname
|
||||||
|
break
|
||||||
|
except sqlite3.Error:
|
||||||
|
name = None
|
||||||
|
self._cached_minmax_index = name
|
||||||
|
return name
|
||||||
|
|
||||||
|
def _minmax_sql(self, conn: sqlite3.Connection, n_symbols: int) -> str:
|
||||||
|
"""min/max 批量 SQL; 复合索引可用时 INDEXED BY 钉死(防计划劣化)。"""
|
||||||
|
ph = ",".join("?" * n_symbols)
|
||||||
|
idx = self._composite_minmax_index(conn)
|
||||||
|
hint = f" INDEXED BY {idx}" if idx else ""
|
||||||
|
return (
|
||||||
|
f"SELECT symbol, exchange, MIN(datetime), MAX(datetime) FROM dbbardata{hint} "
|
||||||
|
f"WHERE interval='d' AND symbol IN ({ph}) GROUP BY symbol, exchange"
|
||||||
|
)
|
||||||
|
|
||||||
def get_security_info_batch(
|
def get_security_info_batch(
|
||||||
self,
|
self,
|
||||||
securities: List[str],
|
securities: List[str],
|
||||||
@@ -615,16 +670,13 @@ class LocalUnifiedProvider(DataProvider): # type: ignore[misc]
|
|||||||
syms = list({sym for sym, _ in (p[1] for p in pairs)})
|
syms = list({sym for sym, _ in (p[1] for p in pairs)})
|
||||||
conn = self._connect()
|
conn = self._connect()
|
||||||
|
|
||||||
# 1. dbbardata min/max per (symbol, exchange); chunk 防 >999 参数
|
# 1. dbbardata min/max per (symbol, exchange); chunk 防 >999 参数;
|
||||||
|
# INDEXED BY 钉复合索引(2026-08-24 计划劣化治本, 见 _minmax_sql)
|
||||||
minmax: Dict[tuple, tuple] = {}
|
minmax: Dict[tuple, tuple] = {}
|
||||||
for i in range(0, len(syms), 400):
|
for i in range(0, len(syms), 400):
|
||||||
chunk = syms[i:i + 400]
|
chunk = syms[i:i + 400]
|
||||||
ph = ",".join("?" * len(chunk))
|
sql = self._minmax_sql(conn, len(chunk))
|
||||||
for r in conn.execute(
|
for r in conn.execute(sql, chunk):
|
||||||
f"SELECT symbol, exchange, MIN(datetime), MAX(datetime) FROM dbbardata "
|
|
||||||
f"WHERE interval='d' AND symbol IN ({ph}) GROUP BY symbol, exchange",
|
|
||||||
chunk,
|
|
||||||
):
|
|
||||||
minmax[(r[0], r[1])] = (
|
minmax[(r[0], r[1])] = (
|
||||||
r[2][:10] if r[2] else None,
|
r[2][:10] if r[2] else None,
|
||||||
r[3][:10] if r[3] else None,
|
r[3][:10] if r[3] else None,
|
||||||
|
|||||||
@@ -602,6 +602,79 @@ class TestGetSecurityInfoBatch:
|
|||||||
assert p.get_security_info_batch([]) == {}
|
assert p.get_security_info_batch([]) == {}
|
||||||
|
|
||||||
|
|
||||||
|
class TestSecurityInfoBatchIndexHint:
|
||||||
|
"""2026-08-24 P0 计划劣化治本: VPS 生产 EXPLAIN 实锤 min/max 批量 SQL 被选去
|
||||||
|
单列 interval 索引, 整扫 ~21M 日线行 + USE TEMP B-TREE(30 只冷跑 101s,
|
||||||
|
开盘选股 38-51min 主根因)。修 = INDEXED BY 钉死 (symbol,exchange,interval,
|
||||||
|
datetime) 复合唯一索引; 索引名跨环境不同(ensure_schema 长名 / NAS 副本
|
||||||
|
merge_increment 建 uq_dbbardata)→运行时探测+实例缓存, 探不到回退裸查询。"""
|
||||||
|
|
||||||
|
@staticmethod
|
||||||
|
def _db_with_index(tmp_path, index_name):
|
||||||
|
db = _make_security_info_fixture(tmp_path)
|
||||||
|
c = sqlite3.connect(str(db))
|
||||||
|
c.execute(
|
||||||
|
f"CREATE UNIQUE INDEX {index_name} ON dbbardata "
|
||||||
|
"(symbol, exchange, interval, datetime)"
|
||||||
|
)
|
||||||
|
c.commit()
|
||||||
|
c.close()
|
||||||
|
return db
|
||||||
|
|
||||||
|
def test_minmax_sql_pins_composite_index(self, tmp_path):
|
||||||
|
db = self._db_with_index(tmp_path, "dbbardata_symbol_exchange_interval_datetime")
|
||||||
|
p = LocalUnifiedProvider({"db_path": str(db), "data_dir": str(tmp_path)})
|
||||||
|
conn = p._connect()
|
||||||
|
sql = p._minmax_sql(conn, 3)
|
||||||
|
assert "INDEXED BY dbbardata_symbol_exchange_interval_datetime" in sql
|
||||||
|
# 探测结果实例缓存(None 与未探测用哨兵区分, 见 _composite_minmax_index)
|
||||||
|
assert p._composite_minmax_index(conn) == "dbbardata_symbol_exchange_interval_datetime"
|
||||||
|
|
||||||
|
def test_minmax_sql_detects_nas_replica_index_name(self, tmp_path):
|
||||||
|
"""NAS 副本复合索引叫 uq_dbbardata(硬编码索引名在此环境必炸)。"""
|
||||||
|
db = self._db_with_index(tmp_path, "uq_dbbardata")
|
||||||
|
p = LocalUnifiedProvider({"db_path": str(db), "data_dir": str(tmp_path)})
|
||||||
|
sql = p._minmax_sql(p._connect(), 3)
|
||||||
|
assert "INDEXED BY uq_dbbardata" in sql
|
||||||
|
|
||||||
|
def test_minmax_sql_fallback_without_composite_index(self, tmp_path):
|
||||||
|
db = _make_security_info_fixture(tmp_path)
|
||||||
|
p = LocalUnifiedProvider({"db_path": str(db), "data_dir": str(tmp_path)})
|
||||||
|
conn = p._connect()
|
||||||
|
sql = p._minmax_sql(conn, 3)
|
||||||
|
assert "INDEXED BY" not in sql # 探不到回退裸查询(不劣于现状)
|
||||||
|
rows = list(conn.execute(sql, ("600519", "000001", "000002")))
|
||||||
|
assert rows # 裸查询仍可执行
|
||||||
|
|
||||||
|
def test_pinned_plan_avoids_temp_btree(self, tmp_path):
|
||||||
|
"""钉死复合索引后 GROUP BY (symbol,exchange) 走索引序, 无 TEMP B-TREE
|
||||||
|
(劣化计划的特征签名); 单列 interval 索引同库共存也不受其扰。"""
|
||||||
|
db = self._db_with_index(tmp_path, "dbbardata_symbol_exchange_interval_datetime")
|
||||||
|
c = sqlite3.connect(str(db))
|
||||||
|
c.execute("CREATE INDEX idx_dbbardata_interval ON dbbardata (interval)")
|
||||||
|
c.commit()
|
||||||
|
c.close()
|
||||||
|
p = LocalUnifiedProvider({"db_path": str(db), "data_dir": str(tmp_path)})
|
||||||
|
conn = p._connect()
|
||||||
|
sql = p._minmax_sql(conn, 3)
|
||||||
|
plan = "\n".join(
|
||||||
|
r[3] for r in conn.execute(
|
||||||
|
"EXPLAIN QUERY PLAN " + sql, ("600519", "000001", "000002"))
|
||||||
|
)
|
||||||
|
assert "TEMP B-TREE" not in plan
|
||||||
|
assert "dbbardata_symbol_exchange_interval_datetime" in plan
|
||||||
|
|
||||||
|
def test_batch_equivalence_with_index_hint(self, tmp_path):
|
||||||
|
"""带复合索引(=生产形态)下 batch 与逐只结果一致(hint 不改语义)。"""
|
||||||
|
db = self._db_with_index(tmp_path, "uq_dbbardata")
|
||||||
|
p = LocalUnifiedProvider({"db_path": str(db), "data_dir": str(tmp_path)})
|
||||||
|
codes = ["600519.XSHG", "000001.XSHE", "000002.XSHE", "999999.XSHG"]
|
||||||
|
batch = p.get_security_info_batch(codes)
|
||||||
|
for code in codes:
|
||||||
|
assert batch[code] == p.get_security_info(code), f"mismatch {code}"
|
||||||
|
assert batch["600519.XSHG"]["start_date"] == "2024-06-18"
|
||||||
|
|
||||||
|
|
||||||
class TestGetValueMetricsBatch:
|
class TestGetValueMetricsBatch:
|
||||||
"""get_value_metrics_batch: ThreadPool 并发逐只(策略01 价值精选), 与逐只一致。"""
|
"""get_value_metrics_batch: ThreadPool 并发逐只(策略01 价值精选), 与逐只一致。"""
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user