From 011c9d4ddbfdfc3108cb387e3e8f3aecad242096 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Thu, 23 Jul 2026 08:11:35 +0800 Subject: [PATCH] =?UTF-8?q?feat(portfolio):=20LocalUnifiedProvider=20funda?= =?UTF-8?q?mentals=20baostock=E4=BC=B0=E5=80=BC+akshare=E5=B8=82=E5=80=BC(?= =?UTF-8?q?Task3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../providers/local_unified_provider.py | 100 +++++++++++++ .../portfolio/test_local_unified_provider.py | 132 ++++++++++++++++++ 2 files changed, 232 insertions(+) diff --git a/sanguo_portfolio/providers/local_unified_provider.py b/sanguo_portfolio/providers/local_unified_provider.py index f42fdd4..7f31636 100644 --- a/sanguo_portfolio/providers/local_unified_provider.py +++ b/sanguo_portfolio/providers/local_unified_provider.py @@ -270,3 +270,103 @@ class LocalUnifiedProvider(DataProvider): # type: ignore[misc] ) -> List[str]: """spec §6 语义别名 = ``get_index_stocks``。""" return self.get_index_stocks(index, date) + + # ==================== get_fundamentals_df ==================== + def _get_lpp_helper(self) -> Any: + """惰性创建 LocalParquetProvider 委托读 static akshare(三表/市值)。DRY。""" + if self._lpp_helper is None: + from .local_parquet_provider import LocalParquetProvider + self._lpp_helper = LocalParquetProvider({"data_dir": self.data_dir}) + return self._lpp_helper + + def _read_valuation_baostock(self, year: int) -> pd.DataFrame: + """读 ``valuation_baostock/.parquet``(baostock 权威: pe/pb/ps/pcf)。""" + if year in self._val_bs_cache: + return self._val_bs_cache[year] + p = os.path.join(self.data_dir, "valuation_baostock", f"{year}.parquet") + if not os.path.exists(p): + self._val_bs_cache[year] = pd.DataFrame() + return pd.DataFrame() + try: + df = pd.read_parquet(p) + self._val_bs_cache[year] = df + return df + except Exception as exc: + logger.warning("读 valuation_baostock/%s 失败: %s", year, exc) + self._val_bs_cache[year] = pd.DataFrame() + return pd.DataFrame() + + def get_fundamentals_df( + self, + stocks: List[str], + date: Optional[Union[str, datetime]] = None, + ) -> pd.DataFrame: + """合并多股 fundamentals, 列对齐 ``_FUNDAMENTAL_COLUMNS``。 + + 数据源路由: + - ``pe/pb/ps/pcf`` ← ``valuation_baostock/.parquet`` (baostock 权威, 覆盖 akshare) + - ``market_cap`` / ``circulating_market_cap`` ← ``static/valuation`` akshare + (baostock valuation 无市值列) + - 三表(eps/yoy/total_liability/...) ← ``static/{income,balance}`` akshare + (委托 LocalParquetProvider 读, DRY) + """ + from .local_parquet_provider import _FUNDAMENTAL_COLUMNS, jq_to_file_code + if not stocks: + return pd.DataFrame(columns=_FUNDAMENTAL_COLUMNS) + date_str = self._to_date_str(date) or datetime.now().strftime("%Y-%m-%d") + rows: List[Dict[str, Any]] = [ + self._build_fundamental_row(s, date_str) for s in stocks + ] + df = pd.DataFrame(rows, columns=_FUNDAMENTAL_COLUMNS) + if "code" in df.columns: + df = df.set_index("code", drop=False) + return df + + def _build_fundamental_row( + self, jq_code: str, date_str: str, + ) -> Dict[str, Any]: + """单股 fundamentals 行: baostock 估值覆盖 akshare pe/pb, 市值+三表用 LocalParquetProvider。""" + from .local_parquet_provider import ( + jq_to_file_code, _to_float, _or_nan, _pct_to_decimal, + ) + from ..factors.valuation import to_yi + + sym, _ = jq_to_dbbardata(jq_code) + fc = jq_to_file_code(jq_code) + + # 先委托 LocalParquetProvider 取整行(akshare 全套) + lpp = self._get_lpp_helper() + ak_row = lpp._build_fundamental_row(jq_code, date_str) + + # 复制 akshare 行(市值/eps/三表/yoy/...), 然后用 baostock 覆盖 pe/pb/ps/pcf + row: Dict[str, Any] = dict(ak_row) + row["code"] = jq_code + + # baostock 估值覆盖 pe/pb/ps/pcf + year = int(date_str[:4]) + vbs = self._read_valuation_baostock(year) + vrow = None + if not vbs.empty: + sub = vbs[ + (vbs["symbol"].astype(str) == sym) + & (vbs["date"].astype(str) <= date_str) + ] + vrow = sub.iloc[-1] if not sub.empty else None + + def gbs(k: str) -> Optional[float]: + return _to_float(vrow.get(k)) if vrow is not None else None + + bs_pe = gbs("peTTM") + bs_pb = gbs("pbMRQ") + bs_ps = gbs("psTTM") + bs_pcf = gbs("pcfNcfTTM") + # baostock 有该日数据则覆盖; 否则保留 akshare pe/pb + if bs_pe is not None: + row["pe_ratio"] = float(bs_pe) + if bs_pb is not None: + row["pb_ratio"] = float(bs_pb) + if bs_ps is not None: + row["ps_ratio"] = float(bs_ps) + if bs_pcf is not None: + row["pcf_ratio"] = float(bs_pcf) + return row diff --git a/tests/portfolio/test_local_unified_provider.py b/tests/portfolio/test_local_unified_provider.py index 5c3e3a3..4db6f65 100644 --- a/tests/portfolio/test_local_unified_provider.py +++ b/tests/portfolio/test_local_unified_provider.py @@ -340,3 +340,135 @@ class TestGetIndexStocks: p = LocalUnifiedProvider({"db_path": str(db), "data_dir": str(tmp_path)}) stocks = p.get_index_stocks("000300", "2020-01-01") assert len(stocks) == 3 + + +# ======================== Task 3: get_fundamentals_df fixture ======================== +def _make_fundamentals_fixture(tmp_path): + """造 valuation_baostock + static/valuation + static/income + static/balance 样本。""" + # 1. valuation_baostock/2024.parquet(baostock 权威: pe/pb/ps/pcf) + vdir = tmp_path / "valuation_baostock" + vdir.mkdir() + pd.DataFrame({ + "symbol": ["600519"], + "exchange": ["SH"], + "date": ["2024-09-30"], + "peTTM": [25.0], + "psTTM": [15.0], + "pcfNcfTTM": [20.0], + "pbMRQ": [7.5], + "turn": [0.1], + "pctChg": [1.0], + "isST": [0], + }).to_parquet(vdir / "2024.parquet") + + # 2. static/valuation akshare(市值/股本) + sdir = tmp_path / "static" / "valuation" + sdir.mkdir(parents=True) + pd.DataFrame({ + "数据日期": ["2024-09-30"], + "总市值": [2e12], + "流通市值": [1.5e12], + "总股本": [1.256e9], + "PE(TTM)": [25.0], + "市净率": [7.5], + }).to_parquet(sdir / "600519.SH_valuation.parquet") + + # 3. static/income akshare(eps + yoy + net_profit) + idir = tmp_path / "static" / "income" + idir.mkdir(parents=True) + pd.DataFrame({ + "SECUCODE": ["600519.SH"], + "REPORT_DATE": ["2024-09-30"], + "REPORT_TYPE": ["Q3"], + "BASIC_EPS": [41.0], + "OPERATE_INCOME": [3.7e10], + "PARENT_NETPROFIT": [9.5e9], + "OPERATE_INCOME_YOY": [15.0], + "OPERATE_PROFIT_YOY": [14.0], + }).to_parquet(idir / "600519.SH_income.parquet") + + # 4. static/balance akshare(资产/负债/权益) + bdir = tmp_path / "static" / "balance" + bdir.mkdir(parents=True) + pd.DataFrame({ + "SECUCODE": ["600519.SH"], + "REPORT_DATE": ["2024-09-30"], + "REPORT_TYPE": ["Q3"], + "TOTAL_ASSETS": [2.5e11], + "TOTAL_LIABILITIES": [5.4e10], + "TOTAL_PARENT_EQUITY": [2.2e11], + "SURPLUS_RESERVE": [8e10], + "UNASSIGN_RPOFIT": [7e10], + }).to_parquet(bdir / "600519.SH_balance.parquet") + + db = tmp_path / "t.db" + c = sqlite3.connect(str(db)) + c.execute("CREATE TABLE dbbardata(symbol TEXT, exchange TEXT, datetime TEXT, interval TEXT)") + c.execute("INSERT INTO dbbardata VALUES('600519','SSE','2024-09-30 00:00:00','d')") + c.commit() + c.close() + return db + + +# ======================== Task 3: get_fundamentals_df ======================== +class TestGetFundamentals: + def test_pe_pb_from_baostock(self, tmp_path): + db = _make_fundamentals_fixture(tmp_path) + p = LocalUnifiedProvider({"db_path": str(db), "data_dir": str(tmp_path)}) + df = p.get_fundamentals_df(["600519.XSHG"], date="2024-09-30") + assert not df.empty + # baostock 权威: peTTM=25 / pbMRQ=7.5 + assert abs(df.loc["600519.XSHG", "pe_ratio"] - 25.0) < 1e-6 + assert abs(df.loc["600519.XSHG", "pb_ratio"] - 7.5) < 1e-6 + assert abs(df.loc["600519.XSHG", "ps_ratio"] - 15.0) < 1e-6 + assert abs(df.loc["600519.XSHG", "pcf_ratio"] - 20.0) < 1e-6 + + def test_market_cap_from_akshare(self, tmp_path): + # baostock valuation 无市值列 → 从 static/valuation akshare 补 + db = _make_fundamentals_fixture(tmp_path) + p = LocalUnifiedProvider({"db_path": str(db), "data_dir": str(tmp_path)}) + df = p.get_fundamentals_df(["600519.XSHG"], date="2024-09-30") + # 2e12 元 → 2e4 亿 + assert abs(df.loc["600519.XSHG", "market_cap"] - 2e4) < 1 + # 1.5e12 元 → 1.5e4 亿 + assert abs(df.loc["600519.XSHG", "circulating_market_cap"] - 1.5e4) < 1 + + def test_three_tables_delegated(self, tmp_path): + # 三表(income/balance)从 static akshare 读(委托 LocalParquetProvider) + db = _make_fundamentals_fixture(tmp_path) + p = LocalUnifiedProvider({"db_path": str(db), "data_dir": str(tmp_path)}) + df = p.get_fundamentals_df(["600519.XSHG"], date="2024-09-30") + row = df.loc["600519.XSHG"] + # eps 来自 income.BASIC_EPS + assert abs(row["eps"] - 41.0) < 1e-6 + # 总负债 5.4e10 元 → 540 亿 + assert abs(row["total_liability"] - 540.0) < 1 + # 归母权益 2.2e11 元 → 2200 亿 + assert abs(row["total_sheet_owner_equities"] - 2200.0) < 1 + # 留存收益 = 盈余公积 8e10 + 未分配利润 7e10 = 1.5e11 元 → 1500 亿 + assert abs(row["retained_profit"] - 1500.0) < 1 + # OPERATE_INCOME_YOY 15.0% → 0.15 + assert abs(row["inc_revenue_year_on_year"] - 0.15) < 1e-6 + + def test_required_columns_present(self, tmp_path): + db = _make_fundamentals_fixture(tmp_path) + p = LocalUnifiedProvider({"db_path": str(db), "data_dir": str(tmp_path)}) + df = p.get_fundamentals_df(["600519.XSHG"], date="2024-09-30") + # _FUNDAMENTAL_COLUMNS(对齐策略 all_weather) + expected = [ + "code", "market_cap", "circulating_market_cap", + "pe_ratio", "pb_ratio", "ps_ratio", "pcf_ratio", + "roe", "roa", "eps", "gross_profit_margin", "net_profit_margin", + "inc_revenue_year_on_year", "inc_operation_profit_year_on_year", + "inc_total_revenue_year_on_year", + "total_liability", "total_sheet_owner_equities", "retained_profit", + "roic", + ] + for col in expected: + assert col in df.columns, f"missing col: {col}" + + def test_empty_stocks_returns_empty(self, tmp_path): + db = _make_fundamentals_fixture(tmp_path) + p = LocalUnifiedProvider({"db_path": str(db), "data_dir": str(tmp_path)}) + df = p.get_fundamentals_df([], date="2024-09-30") + assert df.empty