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]

2026-08-24 13:49:24 +08:00
parent 7a6e9c8c52
commit 80fc517531
2 changed files with 132 additions and 7 deletions
@@ -36,6 +36,23 @@ logger = logging.getLogger(__name__)
 _DEFAULT_DB = r"C:\sanguo_vnpy_v2\data\quant_trading.db"
 _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_SUFFIX_TO_EXC = {"XSHG": "SSE", "XSHE": "SZSE", "SH": "SSE", "SZ": "SZSE"}
 _EXC_TO_JQ_SUFFIX = {"SSE": "XSHG", "SZSE": "XSHE"}
@@ -595,6 +612,44 @@ class LocalUnifiedProvider(DataProvider):  # type: ignore[misc]
            "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(
        self,
        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)})
        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] = {}
        for i in range(0, len(syms), 400):
            chunk = syms[i:i + 400]
-            ph = ",".join("?" * len(chunk))
-            for r in conn.execute(
-                f"SELECT symbol, exchange, MIN(datetime), MAX(datetime) FROM dbbardata "
-                f"WHERE interval='d' AND symbol IN ({ph}) GROUP BY symbol, exchange",
-                chunk,
-            ):
+            sql = self._minmax_sql(conn, len(chunk))
+            for r in conn.execute(sql, chunk):
                minmax[(r[0], r[1])] = (
                    r[2][:10] if r[2] else None,
                    r[3][:10] if r[3] else None,
@@ -602,6 +602,79 @@ class TestGetSecurityInfoBatch:
        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:
    """get_value_metrics_batch: ThreadPool 并发逐只(策略01 价值精选), 与逐只一致。"""