diff --git a/sanguo_portfolio/providers/local_unified_provider.py b/sanguo_portfolio/providers/local_unified_provider.py index 18d1975..7d08d1b 100644 --- a/sanguo_portfolio/providers/local_unified_provider.py +++ b/sanguo_portfolio/providers/local_unified_provider.py @@ -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, diff --git a/tests/portfolio/test_local_unified_provider.py b/tests/portfolio/test_local_unified_provider.py index 9155f4a..8fa14b7 100644 --- a/tests/portfolio/test_local_unified_provider.py +++ b/tests/portfolio/test_local_unified_provider.py @@ -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 价值精选), 与逐只一致。"""