feat(portfolio): LocalUnifiedProvider get_price+前复权(Task1)

2026-07-23 08:06:45 +08:00
parent e12c66acd5
commit 4b4dca2f44
2 changed files with 247 additions and 0 deletions
@@ -118,3 +118,113 @@ class LocalUnifiedProvider(DataProvider):  # type: ignore[misc]
        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
+        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
+
+    @staticmethod
+    def _to_date_str(value: Optional[Union[str, datetime]]) -> Optional[str]:
+        if value is None:
+            return None
+        if isinstance(value, str):
+            return value[:10]
+        try:
+            return value.strftime("%Y-%m-%d")
+        except AttributeError:
+            return str(value)[:10]
+
+    # ==================== get_price ====================
+    def get_price(
+        self,
+        security: Union[str, List[str]],
+        start_date: Optional[Union[str, datetime]] = None,
+        end_date: Optional[Union[str, datetime]] = None,
+        frequency: str = "daily",
+        fields: Optional[List[str]] = None,
+        skip_paused: bool = False,
+        fq: str = "raw",
+        count: Optional[int] = None,
+        panel: bool = True,
+        fill_paused: bool = True,
+        **kwargs: Any,
+    ) -> pd.DataFrame:
+        """读 ``dbbardata('d')`` raw 日线,按需 ``bs_adjust_factor`` 算前复权。
+
+        策略契约(all_weather 实证):
+        - ``panel=False`` 返长表含 ``time`` + ``code`` 列(供 pivot)
+        - ``fields`` 里缺失列(如 ``high_limit``)补 NaN(降级)
+        - ``frequency`` 非 day/1d/d → 返空 DataFrame(1m 数据层无)
+        """
+        freq = str(frequency or "").lower()
+        if freq not in ("daily", "day", "1d", "d"):
+            return pd.DataFrame()
+        secs: List[str] = [security] if isinstance(security, str) else list(security or [])
+        if not secs:
+            return pd.DataFrame()
+        conn = self._connect()
+        start_str = self._to_date_str(start_date) or "1990-01-01"
+        end_str = self._to_date_str(end_date) or datetime.now().strftime("%Y-%m-%d")
+
+        frames: Dict[str, pd.DataFrame] = {}
+        for jq_code in secs:
+            sym, exc = jq_to_dbbardata(jq_code)
+            q = (
+                "SELECT datetime, open_price, high_price, low_price, close_price, "
+                "volume, turnover FROM dbbardata WHERE symbol=? AND exchange=? "
+                "AND interval='d' AND datetime>=? AND datetime<=? ORDER BY datetime"
+            )
+            df = pd.read_sql(
+                q, conn,
+                params=(sym, exc, start_str + " 00:00:00", end_str + " 23:59:59"),
+            )
+            if df.empty:
+                frames[jq_code] = df
+                continue
+            df["datetime"] = pd.to_datetime(df["datetime"])
+            df = df.set_index("datetime")
+            df.index.name = None
+            if count:
+                df = df.tail(count)
+            # 前复权
+            if fq in ("qfq", "pre", "前复权") and not df.empty:
+                factor = _build_qfq_factor(_jq_to_bs_code(jq_code), conn, pd.Series(df.index))
+                for col in ("open_price", "high_price", "low_price", "close_price"):
+                    df[col] = df[col].values * factor.values
+            # jq 风格字段重命名
+            df = df.rename(columns={
+                "open_price": "open", "high_price": "high",
+                "low_price": "low", "close_price": "close",
+            })
+            # 缺失字段(如 high_limit)补 NaN
+            if fields:
+                for f in fields:
+                    if f not in df.columns:
+                        df[f] = float("nan")
+                df = df[[f for f in fields if f in df.columns]]
+            frames[jq_code] = df
+
+        if not frames or all(f.empty for f in frames.values()):
+            return pd.DataFrame()
+        if not panel:
+            parts: List[pd.DataFrame] = []
+            for jq_code, df in frames.items():
+                if df.empty:
+                    continue
+                d = df.reset_index()
+                # index.name=None 时 reset_index 出 'index' 列; 统一改名 'time'
+                if "index" in d.columns and "time" not in d.columns:
+                    d = d.rename(columns={"index": "time"})
+                elif "datetime" in d.columns:
+                    d = d.rename(columns={"datetime": "time"})
+                d.insert(0, "code", jq_code)
+                parts.append(d)
+            return pd.concat(parts, ignore_index=True) if parts else pd.DataFrame()
+        if len(frames) == 1:
+            return next(iter(frames.values()))
+        return pd.concat(frames, axis=1)