diff --git a/sanguo_portfolio/providers/local_unified_provider.py b/sanguo_portfolio/providers/local_unified_provider.py index caf22dd..eb0a9e2 100644 --- a/sanguo_portfolio/providers/local_unified_provider.py +++ b/sanguo_portfolio/providers/local_unified_provider.py @@ -15,6 +15,7 @@ import logging import os import re import sqlite3 +import threading from concurrent.futures import ThreadPoolExecutor from datetime import datetime from typing import Any, Dict, List, Optional, Union @@ -161,16 +162,20 @@ class LocalUnifiedProvider(DataProvider): # type: ignore[misc] cfg = config or {} self.db_path: str = cfg.get("db_path", _DEFAULT_DB) self.data_dir: str = cfg.get("data_dir", _DEFAULT_DATA_DIR) - self._conn: Optional[sqlite3.Connection] = None + # thread-local:引擎跨日存活后回调线程会变(2026-09-01 live_19 事故), + # 单例连接跨线程复用直接 ProgrammingError;每线程各持一条连接 + self._tls = threading.local() self._val_bs_cache: Dict[int, pd.DataFrame] = {} # year -> valuation_baostock self._lpp_helper: Any = None def _connect(self) -> sqlite3.Connection: - """惰性连接 dbbardata sqlite(单连接复用)。""" - if self._conn is None: - self._conn = sqlite3.connect(self.db_path, timeout=30) - self._conn.execute("PRAGMA busy_timeout = 30000") - return self._conn + """惰性连接 dbbardata sqlite(thread-local 连接复用)。""" + conn = getattr(self._tls, "conn", None) + if conn is None: + conn = sqlite3.connect(self.db_path, timeout=30) + conn.execute("PRAGMA busy_timeout = 30000") + self._tls.conn = conn + return conn @staticmethod def _to_date_str(value: Optional[Union[str, datetime]]) -> Optional[str]: diff --git a/tests/portfolio/test_local_unified_provider.py b/tests/portfolio/test_local_unified_provider.py index 25edbab..70f5d91 100644 --- a/tests/portfolio/test_local_unified_provider.py +++ b/tests/portfolio/test_local_unified_provider.py @@ -12,6 +12,7 @@ Mac 本地 TDD: sqlite tmp_path + tmp parquet fixture,零 VPS 依赖,零网络 from __future__ import annotations import sqlite3 +import threading from typing import Any, Dict, List import pandas as pd @@ -863,3 +864,45 @@ class TestMixedDatetimeFormat: p = LocalUnifiedProvider({"db_path": str(db), "data_dir": str(tmp_path)}) df = p.get_price("600519.XSHG", start_date="2024-09-25", end_date="2024-09-27", fq="raw") assert len(df) == 3 # 不崩 + 返 3 行(混合格式解析 OK) + + +# ======================== Task N: _connect thread-local ======================== +class TestConnectThreadLocal: + """2026-09-01 live_19 事故:引擎首次跨日存活后回调线程变化(连接建于线程A、 + 用在线程B),单例 sqlite 连接跨线程复用直接 ProgrammingError → 牛熊分界 + 不可用跳过调仓。_connect 改 thread-local 后每线程自建连接,互不共享。""" + + def test_connect_cross_thread_usable(self, tmp_path): + db = _make_security_info_fixture(tmp_path) + p = LocalUnifiedProvider({"db_path": str(db), "data_dir": str(tmp_path)}) + conn_main = p._connect() # 主线程先建连(旧实现会缓存单例) + + errs: List[Any] = [] + + def worker(): + try: + c = p._connect() + c.execute("SELECT 1").fetchall() # 子线程直接用,不得抛错 + except Exception as exc: # noqa: BLE001 - 收集给断言看 + errs.append(exc) + + t = threading.Thread(target=worker) + t.start() + t.join() + assert not errs, f"跨线程取连接报错: {errs}" + # 主线程连接仍是原对象(thread-local 各持各的) + assert p._connect() is conn_main + + def test_two_threads_get_distinct_connections(self, tmp_path): + db = _make_security_info_fixture(tmp_path) + p = LocalUnifiedProvider({"db_path": str(db), "data_dir": str(tmp_path)}) + got: Dict[str, Any] = {} + + def worker(): + got["conn"] = p._connect() + + t = threading.Thread(target=worker) + t.start() + t.join() + assert got["conn"] is not None + assert got["conn"] is not p._connect() # 不是同一个连接对象