From 59227c727af4ce9d32d815004f503038d12f9988 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 27 Jun 2026 23:27:14 +0800 Subject: [PATCH 1/2] Enrich AkShare staging with real dividend and price factors. Uses fhps batch dividend yield, price-history vol/momentum, financial ROE, and dividend stability with sample fallback and a new ETF market-history staging CLI. Co-authored-by: Cursor --- README.md | 7 + pyproject.toml | 3 +- .../akshare_enrichment.py | 203 ++++++++++++++++++ .../akshare_market_history.py | 117 ++++++++++ .../akshare_staging.py | 203 +++++++++++------- tests/test_akshare_enrichment.py | 88 ++++++++ tests/test_akshare_staging.py | 92 ++++++-- 7 files changed, 620 insertions(+), 93 deletions(-) create mode 100644 src/cn_equity_snapshot_pipelines/akshare_enrichment.py create mode 100644 src/cn_equity_snapshot_pipelines/akshare_market_history.py create mode 100644 tests/test_akshare_enrichment.py diff --git a/README.md b/README.md index 8258478..5023a3c 100644 --- a/README.md +++ b/README.md @@ -29,6 +29,13 @@ Build a sample artifact pack locally: PYTHONPATH=src python scripts/build_dividend_quality_sample.py ``` +Stage real factor inputs via AkShare (falls back to sample CSV on failure): + +```bash +cneq-stage-akshare-dividend-quality --output data/staging/dividend_quality/factor_snapshot.latest.csv +cneq-stage-akshare-market-history --output data/staging/market_history/etf_universe.latest.csv +``` + Or use the installed entrypoint: ```bash diff --git a/pyproject.toml b/pyproject.toml index 3714710..05052e0 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -10,7 +10,7 @@ readme = "README.md" requires-python = ">=3.11" dependencies = [ "pandas>=2.0", - "cn-equity-strategies @ git+https://github.com/QuantStrategyLab/CnEquityStrategies.git@a911cf05c58d1fd14c48a41a0295e50e03638ec2", + "cn-equity-strategies @ git+https://github.com/QuantStrategyLab/CnEquityStrategies.git@92d3cfea1e979ab072718f6f5415b20f8d952351", ] [project.optional-dependencies] @@ -20,6 +20,7 @@ public-data = ["akshare>=1.14"] [project.scripts] cneq-build-dividend-quality-snapshot = "cn_equity_snapshot_pipelines.dividend_quality:main" cneq-stage-akshare-dividend-quality = "cn_equity_snapshot_pipelines.akshare_staging:main" +cneq-stage-akshare-market-history = "cn_equity_snapshot_pipelines.akshare_market_history:main" [tool.setuptools] package-dir = {"" = "src"} diff --git a/src/cn_equity_snapshot_pipelines/akshare_enrichment.py b/src/cn_equity_snapshot_pipelines/akshare_enrichment.py new file mode 100644 index 0000000..5530bab --- /dev/null +++ b/src/cn_equity_snapshot_pipelines/akshare_enrichment.py @@ -0,0 +1,203 @@ +from __future__ import annotations + +from datetime import date, datetime, timezone +from typing import Any + +import pandas as pd + +FACTOR_SNAPSHOT_COLUMNS = ( + "symbol", + "sector", + "close_cny", + "adv20_cny", + "market_cap_cny", + "dividend_yield_ttm", + "dividend_stability_3y", + "earnings_positive", + "payout_ratio", + "roe_ttm", + "roe_stability_3y", + "realized_vol_126", + "mom_12_1", + "sma200_gap", + "suspension_days_63", + "is_st", + "list_days", +) + +FHPS_CANDIDATE_DATES = ( + "20241231", + "20231231", + "20221231", + "20240630", + "20230630", +) + + +def normalize_symbol(value: object) -> str: + text = str(value or "").strip().upper() + if text.endswith(".SH") or text.endswith(".SZ"): + text = text.split(".", 1)[0] + return text.zfill(6) if text.isdigit() else text + + +def _coerce_float(value: Any, default: float = 0.0) -> float: + try: + resolved = float(value) + except (TypeError, ValueError): + return default + if pd.isna(resolved): + return default + return resolved + + +def compute_price_features(hist: pd.DataFrame) -> dict[str, float | int]: + if hist.empty: + raise ValueError("history must not be empty") + frame = hist.sort_values("日期").copy() + close = pd.to_numeric(frame["收盘"], errors="coerce") + turnover = pd.to_numeric(frame["成交额"], errors="coerce") + volume = pd.to_numeric(frame["成交量"], errors="coerce") + if close.dropna().empty: + raise ValueError("history close series is empty") + + latest_close = float(close.dropna().iloc[-1]) + adv20 = float(turnover.tail(20).mean()) if turnover.tail(20).notna().any() else 0.0 + returns = close.pct_change().dropna() + if len(returns) >= 126: + realized_vol_126 = float(returns.tail(126).std(ddof=0) * (252**0.5)) + else: + realized_vol_126 = float(returns.std(ddof=0) * (252**0.5)) if len(returns) >= 2 else 0.18 + + if len(close.dropna()) >= 252: + mom_12_1 = float(close.iloc[-21] / close.iloc[-252] - 1.0) + elif len(close.dropna()) >= 22: + mom_12_1 = float(close.iloc[-1] / close.iloc[-22] - 1.0) + else: + mom_12_1 = 0.0 + + if len(close.dropna()) >= 200: + sma200_gap = float(latest_close / close.tail(200).mean() - 1.0) + else: + sma200_gap = 0.0 + + suspension_days_63 = int((volume.tail(63).fillna(0) <= 0).sum()) + first_date = pd.to_datetime(frame["日期"].iloc[0], errors="coerce") + list_days = 2000 + if pd.notna(first_date): + list_days = max(int((pd.Timestamp(datetime.now(timezone.utc).date()) - first_date.normalize()).days), 1) + + return { + "close_cny": latest_close, + "adv20_cny": max(adv20, 1.0), + "realized_vol_126": max(realized_vol_126, 0.01), + "mom_12_1": mom_12_1, + "sma200_gap": sma200_gap, + "suspension_days_63": suspension_days_63, + "list_days": list_days, + } + + +def compute_dividend_stability(dividends: pd.DataFrame, *, years: int = 3) -> float: + if dividends.empty or "派息" not in dividends.columns: + return 0.0 + frame = dividends.copy() + date_column = "除权除息日" if "除权除息日" in frame.columns else "股权登记日" + frame["event_date"] = pd.to_datetime(frame[date_column], errors="coerce") + frame["payout_per10"] = pd.to_numeric(frame["派息"], errors="coerce").fillna(0.0) + frame = frame.loc[frame["payout_per10"] > 0.0].dropna(subset=["event_date"]) + if frame.empty: + return 0.0 + + latest_year = int(frame["event_date"].dt.year.max()) + recent = frame.loc[frame["event_date"].dt.year >= latest_year - years + 1] + years_with_dividend = int(recent["event_date"].dt.year.nunique()) + coverage = years_with_dividend / float(years) + payout_std = float(recent.groupby(recent["event_date"].dt.year)["payout_per10"].sum().std(ddof=0) or 0.0) + payout_mean = float(recent.groupby(recent["event_date"].dt.year)["payout_per10"].sum().mean() or 0.0) + variability = payout_std / payout_mean if payout_mean > 0 else 1.0 + consistency = max(0.0, 1.0 - min(variability, 1.0)) + return max(0.0, min(coverage * (0.5 + 0.5 * consistency), 1.0)) + + +def compute_financial_features(financials: pd.DataFrame) -> dict[str, float | bool]: + if financials.empty: + return { + "roe_ttm": 0.0, + "roe_stability_3y": 0.0, + "earnings_positive": False, + } + frame = financials.sort_values("日期").copy() + roe = pd.to_numeric(frame.get("净资产报酬率(%)"), errors="coerce") / 100.0 + eps = pd.to_numeric(frame.get("摊薄每股收益(元)"), errors="coerce") + roe = roe.dropna() + eps = eps.dropna() + latest_roe = float(roe.iloc[-1]) if not roe.empty else 0.0 + earnings_positive = bool(float(eps.iloc[-1]) > 0.0) if not eps.empty else False + tail = roe.tail(12) + if len(tail) >= 4 and abs(float(tail.mean())) > 1e-9: + roe_stability_3y = max(0.0, 1.0 - min(float(tail.std(ddof=0) / abs(float(tail.mean()))), 1.0)) + else: + roe_stability_3y = 0.5 + return { + "roe_ttm": latest_roe, + "roe_stability_3y": roe_stability_3y, + "earnings_positive": earnings_positive, + } + + +def extract_fhps_features(row: pd.Series, *, close_cny: float) -> dict[str, float | bool]: + dividend_yield_ttm = _coerce_float(row.get("现金分红-股息率"), 0.0) + cash_div_per10 = _coerce_float(row.get("现金分红-现金分红比例"), 0.0) + eps = _coerce_float(row.get("每股收益"), 0.0) + total_shares = _coerce_float(row.get("总股本"), 0.0) + payout_ratio = 0.0 + if eps > 0 and cash_div_per10 > 0: + payout_ratio = min((cash_div_per10 / 10.0) / eps, 2.0) + market_cap_cny = total_shares * close_cny if total_shares > 0 and close_cny > 0 else 0.0 + name = str(row.get("名称") or "") + return { + "dividend_yield_ttm": max(dividend_yield_ttm, 0.0), + "payout_ratio": payout_ratio, + "market_cap_cny": market_cap_cny, + "earnings_positive": eps > 0, + "is_st": "ST" in name.upper(), + } + + +def merge_factor_row( + *, + symbol: str, + price: dict[str, float | int], + fhps: dict[str, float | bool] | None, + financials: dict[str, float | bool], + dividend_stability_3y: float, + sector: str, +) -> dict[str, object]: + merged: dict[str, object] = { + "symbol": normalize_symbol(symbol), + "sector": sector or "unknown", + "dividend_stability_3y": float(dividend_stability_3y), + "suspension_days_63": int(price["suspension_days_63"]), + "is_st": bool(fhps.get("is_st")) if fhps else False, + } + merged.update(price) + merged.update(financials) + if fhps: + merged.update(fhps) + if float(merged.get("market_cap_cny") or 0.0) <= 0: + merged["market_cap_cny"] = float(price["close_cny"]) * 1_000_000_000.0 + else: + merged.setdefault("dividend_yield_ttm", 0.0) + merged.setdefault("payout_ratio", 0.0) + merged.setdefault("market_cap_cny", float(price["close_cny"]) * 1_000_000_000.0) + return {column: merged.get(column) for column in FACTOR_SNAPSHOT_COLUMNS} + + +def stamp_as_of(frame: pd.DataFrame, *, as_of: str | date | None = None) -> pd.DataFrame: + output = frame.copy() + if "as_of" in output.columns or "snapshot_date" in output.columns: + return output + stamp = as_of or datetime.now(timezone.utc).date().isoformat() + output.insert(0, "as_of", str(stamp)) + return output diff --git a/src/cn_equity_snapshot_pipelines/akshare_market_history.py b/src/cn_equity_snapshot_pipelines/akshare_market_history.py new file mode 100644 index 0000000..9e30fbb --- /dev/null +++ b/src/cn_equity_snapshot_pipelines/akshare_market_history.py @@ -0,0 +1,117 @@ +from __future__ import annotations + +import argparse +from datetime import datetime, timezone +from pathlib import Path + +import pandas as pd + +DEFAULT_ETF_SYMBOLS = ( + "510300", + "510500", + "159915", + "588000", + "512100", + "512170", + "515030", + "512760", + "518880", + "513100", + "511880", + "511260", +) + + +def normalize_symbol(value: object) -> str: + text = str(value or "").strip().upper() + if text.endswith(".SH") or text.endswith(".SZ"): + text = text.split(".", 1)[0] + return text.zfill(6) if text.isdigit() else text + + +def _import_akshare(): + import akshare as ak + + return ak + + +def fetch_etf_history(symbol: str, *, ak=None, start_date: str = "20200101") -> pd.DataFrame: + ak_module = ak or _import_akshare() + end_date = datetime.now(timezone.utc).strftime("%Y%m%d") + frame = ak_module.fund_etf_hist_em( + symbol=normalize_symbol(symbol), + period="daily", + start_date=start_date, + end_date=end_date, + adjust="qfq", + ) + if frame.empty: + raise ValueError(f"empty ETF history for {symbol}") + output = pd.DataFrame( + { + "date": pd.to_datetime(frame["日期"], errors="coerce").dt.date.astype(str), + "symbol": normalize_symbol(symbol), + "close": pd.to_numeric(frame["收盘"], errors="coerce"), + } + ) + return output.dropna(subset=["date", "close"]) + + +def build_market_history_frame( + symbols: tuple[str, ...], + *, + ak=None, + start_date: str = "20200101", +) -> pd.DataFrame: + frames: list[pd.DataFrame] = [] + errors: dict[str, str] = {} + for symbol in symbols: + try: + frames.append(fetch_etf_history(symbol, ak=ak, start_date=start_date)) + except Exception as exc: + errors[normalize_symbol(symbol)] = str(exc) + if not frames: + missing = ", ".join(sorted(errors)) + raise RuntimeError(f"failed to fetch ETF histories: {missing}") + history = pd.concat(frames, ignore_index=True) + history = history.sort_values(["symbol", "date"]).reset_index(drop=True) + return history + + +def write_market_history_csv( + *, + output_path: str | Path, + symbols: tuple[str, ...] = DEFAULT_ETF_SYMBOLS, + start_date: str = "20200101", +) -> dict[str, object]: + ak = _import_akshare() + frame = build_market_history_frame(symbols, ak=ak, start_date=start_date) + path = Path(output_path) + path.parent.mkdir(parents=True, exist_ok=True) + frame.to_csv(path, index=False) + return { + "output_path": str(path), + "row_count": int(len(frame)), + "symbols": [normalize_symbol(symbol) for symbol in symbols], + "start_date": start_date, + } + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(description="Stage ETF market history CSV for cn_index_etf_tactical_rotation.") + parser.add_argument("--output", default="data/staging/market_history/etf_universe.latest.csv") + parser.add_argument("--symbols", default=",".join(DEFAULT_ETF_SYMBOLS)) + parser.add_argument("--start-date", default="20200101") + args = parser.parse_args(argv) + symbols = tuple(symbol.strip() for symbol in args.symbols.split(",") if symbol.strip()) + diagnostics = write_market_history_csv( + output_path=args.output, + symbols=symbols, + start_date=args.start_date, + ) + print(diagnostics) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/src/cn_equity_snapshot_pipelines/akshare_staging.py b/src/cn_equity_snapshot_pipelines/akshare_staging.py index 33a2fd4..3ef5e69 100644 --- a/src/cn_equity_snapshot_pipelines/akshare_staging.py +++ b/src/cn_equity_snapshot_pipelines/akshare_staging.py @@ -3,27 +3,20 @@ import argparse from datetime import datetime, timezone from pathlib import Path +from typing import Callable import pandas as pd -FACTOR_SNAPSHOT_COLUMNS = ( - "symbol", - "sector", - "close_cny", - "adv20_cny", - "market_cap_cny", - "dividend_yield_ttm", - "dividend_stability_3y", - "earnings_positive", - "payout_ratio", - "roe_ttm", - "roe_stability_3y", - "realized_vol_126", - "mom_12_1", - "sma200_gap", - "suspension_days_63", - "is_st", - "list_days", +from .akshare_enrichment import ( + FACTOR_SNAPSHOT_COLUMNS, + FHPS_CANDIDATE_DATES, + compute_dividend_stability, + compute_financial_features, + compute_price_features, + extract_fhps_features, + merge_factor_row, + normalize_symbol, + stamp_as_of, ) DEFAULT_STAGING_SYMBOLS = ( @@ -47,29 +40,96 @@ def _load_sample_fallback(sample_path: Path) -> pd.DataFrame: return frame.loc[:, FACTOR_SNAPSHOT_COLUMNS].copy() -def _normalize_symbol(value: object) -> str: - text = str(value or "").strip().upper() - if text.endswith(".SH") or text.endswith(".SZ"): - text = text.split(".", 1)[0] - return text.zfill(6) if text.isdigit() else text +def _import_akshare(): + import akshare as ak + + return ak + + +def _fetch_fhps_table(ak) -> pd.DataFrame: + last_error: Exception | None = None + for report_date in FHPS_CANDIDATE_DATES: + try: + frame = ak.stock_fhps_em(date=report_date) + if frame is not None and not frame.empty: + frame = frame.copy() + frame["symbol"] = frame["代码"].map(normalize_symbol) + return frame + except Exception as exc: + last_error = exc + continue + if last_error is not None: + raise last_error + raise RuntimeError("stock_fhps_em returned no data for candidate report dates") + + +def _fetch_history(ak, symbol: str) -> pd.DataFrame: + end_date = datetime.now(timezone.utc).strftime("%Y%m%d") + return ak.stock_zh_a_hist( + symbol=normalize_symbol(symbol), + period="daily", + start_date="20180101", + end_date=end_date, + adjust="qfq", + ) -def _fetch_spot_rows() -> pd.DataFrame: - import akshare as ak +def _fetch_financials(ak, symbol: str) -> pd.DataFrame: + start_year = str(datetime.now(timezone.utc).year - 4) + return ak.stock_financial_analysis_indicator(symbol=normalize_symbol(symbol), start_year=start_year) + + +def _fetch_dividends(ak, symbol: str) -> pd.DataFrame: + return ak.stock_history_dividend_detail(symbol=normalize_symbol(symbol), indicator="分红") + - spot = ak.stock_zh_a_spot_em() - spot = spot.rename( - columns={ - "代码": "symbol", - "名称": "name", - "最新价": "close_cny", - "成交额": "turnover_cny", - "总市值": "market_cap_cny", - "市盈率-动态": "pe_ttm", - } +def _fetch_sector(ak, symbol: str) -> str: + try: + profile = ak.stock_profile_cninfo(symbol=normalize_symbol(symbol)) + if not profile.empty and "所属行业" in profile.columns: + sector = str(profile.iloc[0]["所属行业"]).strip() + return sector or "unknown" + except Exception: + return "unknown" + return "unknown" + + +def build_factor_row_from_akshare( + symbol: str, + *, + ak=None, + fhps_table: pd.DataFrame | None = None, + fetch_history: Callable[[str], pd.DataFrame] | None = None, + fetch_financials: Callable[[str], pd.DataFrame] | None = None, + fetch_dividends: Callable[[str], pd.DataFrame] | None = None, + fetch_sector: Callable[[str], str] | None = None, +) -> dict[str, object]: + ak_module = ak or _import_akshare() + history_loader = fetch_history or (lambda item: _fetch_history(ak_module, item)) + financial_loader = fetch_financials or (lambda item: _fetch_financials(ak_module, item)) + dividend_loader = fetch_dividends or (lambda item: _fetch_dividends(ak_module, item)) + sector_loader = fetch_sector or (lambda item: _fetch_sector(ak_module, item)) + + normalized = normalize_symbol(symbol) + price = compute_price_features(history_loader(normalized)) + financials = compute_financial_features(financial_loader(normalized)) + dividend_stability_3y = compute_dividend_stability(dividend_loader(normalized)) + + fhps_features = None + if fhps_table is not None and not fhps_table.empty: + matched = fhps_table.loc[fhps_table["symbol"] == normalized] + if not matched.empty: + fhps_features = extract_fhps_features(matched.iloc[0], close_cny=float(price["close_cny"])) + + sector = sector_loader(normalized) + return merge_factor_row( + symbol=normalized, + price=price, + fhps=fhps_features, + financials=financials, + dividend_stability_3y=dividend_stability_3y, + sector=sector, ) - spot["symbol"] = spot["symbol"].map(_normalize_symbol) - return spot def build_factor_snapshot_from_akshare( @@ -77,61 +137,50 @@ def build_factor_snapshot_from_akshare( symbols: tuple[str, ...] = DEFAULT_STAGING_SYMBOLS, sample_fallback_path: str | Path | None = None, min_rows: int = 4, + as_of: str | None = None, ) -> tuple[pd.DataFrame, dict[str, object]]: - diagnostics: dict[str, object] = {"source": "akshare", "requested_symbols": list(symbols)} + diagnostics: dict[str, object] = { + "source": "akshare", + "requested_symbols": list(symbols), + "symbol_errors": {}, + } try: - spot = _fetch_spot_rows() + ak = _import_akshare() + fhps_table = _fetch_fhps_table(ak) except Exception as exc: diagnostics["source"] = "sample_fallback" diagnostics["akshare_error"] = str(exc) if sample_fallback_path is None: raise - frame = _load_sample_fallback(Path(sample_fallback_path)) + frame = stamp_as_of(_load_sample_fallback(Path(sample_fallback_path)), as_of=as_of) diagnostics["row_count"] = len(frame) return frame, diagnostics - normalized_symbols = {_normalize_symbol(symbol) for symbol in symbols} - filtered = spot.loc[spot["symbol"].isin(normalized_symbols)].copy() - if len(filtered) < min_rows: + rows: list[dict[str, object]] = [] + for symbol in symbols: + try: + rows.append( + build_factor_row_from_akshare( + symbol, + ak=ak, + fhps_table=fhps_table, + ) + ) + except Exception as exc: + diagnostics["symbol_errors"][normalize_symbol(symbol)] = str(exc) + + if len(rows) < min_rows: diagnostics["source"] = "sample_fallback" - diagnostics["akshare_error"] = f"only {len(filtered)} rows matched requested symbols" + diagnostics["akshare_error"] = f"only {len(rows)} symbols enriched successfully" if sample_fallback_path is None: raise ValueError(diagnostics["akshare_error"]) - frame = _load_sample_fallback(Path(sample_fallback_path)) + frame = stamp_as_of(_load_sample_fallback(Path(sample_fallback_path)), as_of=as_of) diagnostics["row_count"] = len(frame) return frame, diagnostics - rows: list[dict[str, object]] = [] - for _, item in filtered.iterrows(): - close_cny = float(item.get("close_cny") or 0.0) - turnover_cny = float(item.get("turnover_cny") or 0.0) - market_cap_cny = float(item.get("market_cap_cny") or 0.0) - rows.append( - { - "symbol": item["symbol"], - "sector": "unknown", - "close_cny": close_cny, - "adv20_cny": max(turnover_cny, 1.0), - "market_cap_cny": market_cap_cny, - "dividend_yield_ttm": 0.04, - "dividend_stability_3y": 0.70, - "earnings_positive": True, - "payout_ratio": 0.40, - "roe_ttm": 0.12, - "roe_stability_3y": 0.65, - "realized_vol_126": 0.18, - "mom_12_1": 0.05, - "sma200_gap": 0.02, - "suspension_days_63": 0, - "is_st": False, - "list_days": 2000, - } - ) - frame = pd.DataFrame(rows, columns=list(FACTOR_SNAPSHOT_COLUMNS)) - if "as_of" not in frame.columns and "snapshot_date" not in frame.columns: - stamp = datetime.now(timezone.utc).date().isoformat() - frame.insert(0, "as_of", stamp) + frame = stamp_as_of(pd.DataFrame(rows, columns=list(FACTOR_SNAPSHOT_COLUMNS)), as_of=as_of) diagnostics["row_count"] = len(frame) + diagnostics["fhps_rows"] = int(len(fhps_table)) return frame, diagnostics @@ -140,10 +189,12 @@ def write_staging_factor_snapshot( output_path: str | Path, symbols: tuple[str, ...] = DEFAULT_STAGING_SYMBOLS, sample_fallback_path: str | Path | None = None, + as_of: str | None = None, ) -> dict[str, object]: frame, diagnostics = build_factor_snapshot_from_akshare( symbols=symbols, sample_fallback_path=sample_fallback_path, + as_of=as_of, ) path = Path(output_path) path.parent.mkdir(parents=True, exist_ok=True) @@ -156,6 +207,7 @@ def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser(description="Stage cn_dividend_quality_snapshot factor CSV via AkShare.") parser.add_argument("--output", default="data/staging/dividend_quality/factor_snapshot.latest.csv") parser.add_argument("--symbols", default=",".join(DEFAULT_STAGING_SYMBOLS)) + parser.add_argument("--as-of", default=None, help="Optional as_of date (YYYY-MM-DD). Defaults to UTC today.") parser.add_argument( "--sample-fallback", default=str(Path(__file__).resolve().parents[2] / "examples" / "dividend_quality" / "factor_snapshot.sample.csv"), @@ -166,6 +218,7 @@ def main(argv: list[str] | None = None) -> int: output_path=args.output, symbols=symbols, sample_fallback_path=args.sample_fallback, + as_of=args.as_of, ) print(diagnostics) return 0 diff --git a/tests/test_akshare_enrichment.py b/tests/test_akshare_enrichment.py new file mode 100644 index 0000000..a3573ff --- /dev/null +++ b/tests/test_akshare_enrichment.py @@ -0,0 +1,88 @@ +from __future__ import annotations + +import pandas as pd + +from cn_equity_snapshot_pipelines.akshare_enrichment import ( + compute_dividend_stability, + compute_financial_features, + compute_price_features, + extract_fhps_features, + merge_factor_row, +) + + +def _sample_history(rows: int = 280) -> pd.DataFrame: + dates = pd.bdate_range("2024-01-02", periods=rows) + close = [100 + idx * 0.05 for idx in range(rows)] + return pd.DataFrame( + { + "日期": dates, + "收盘": close, + "成交额": [50_000_000.0 + idx * 1000 for idx in range(rows)], + "成交量": [1_000_000 + idx for idx in range(rows)], + } + ) + + +def test_compute_price_features_from_history(): + features = compute_price_features(_sample_history()) + assert features["close_cny"] > 100 + assert features["adv20_cny"] > 0 + assert features["realized_vol_126"] > 0 + assert features["list_days"] > 200 + + +def test_compute_financial_features_uses_latest_roe(): + financials = pd.DataFrame( + { + "日期": ["2024-03-31", "2024-06-30", "2024-09-30"], + "净资产报酬率(%)": [8.0, 9.0, 10.0], + "摊薄每股收益(元)": [1.0, 1.1, 1.2], + } + ) + features = compute_financial_features(financials) + assert features["roe_ttm"] == 0.10 + assert features["earnings_positive"] is True + assert 0.0 <= features["roe_stability_3y"] <= 1.0 + + +def test_compute_dividend_stability_from_history(): + dividends = pd.DataFrame( + { + "除权除息日": ["2024-06-20", "2025-06-20", "2026-06-20"], + "派息": [200.0, 210.0, 220.0], + } + ) + stability = compute_dividend_stability(dividends, years=3) + assert stability >= 0.8 + + +def test_merge_factor_row_prefers_fhps_dividend_fields(): + fhps = pd.Series( + { + "名称": "贵州茅台", + "现金分红-股息率": 0.03, + "现金分红-现金分红比例": 276.0, + "每股收益": 68.0, + "总股本": 1_256_197_800, + } + ) + row = merge_factor_row( + symbol="600519", + price={ + "close_cny": 1000.0, + "adv20_cny": 100_000_000.0, + "realized_vol_126": 0.2, + "mom_12_1": 0.05, + "sma200_gap": 0.03, + "suspension_days_63": 0, + "list_days": 3000, + }, + fhps=extract_fhps_features(fhps, close_cny=1000.0), + financials={"roe_ttm": 0.12, "roe_stability_3y": 0.7, "earnings_positive": True}, + dividend_stability_3y=0.8, + sector="白酒", + ) + assert row["dividend_yield_ttm"] == 0.03 + assert row["sector"] == "白酒" + assert float(row["market_cap_cny"]) > 0 diff --git a/tests/test_akshare_staging.py b/tests/test_akshare_staging.py index a66e5e1..f49b0ce 100644 --- a/tests/test_akshare_staging.py +++ b/tests/test_akshare_staging.py @@ -3,12 +3,10 @@ from pathlib import Path import pandas as pd +import pytest -from cn_equity_snapshot_pipelines.akshare_staging import ( - FACTOR_SNAPSHOT_COLUMNS, - build_factor_snapshot_from_akshare, - write_staging_factor_snapshot, -) +from cn_equity_snapshot_pipelines.akshare_market_history import build_market_history_frame, normalize_symbol +from cn_equity_snapshot_pipelines.akshare_staging import build_factor_row_from_akshare, build_factor_snapshot_from_akshare def test_build_factor_snapshot_falls_back_to_sample(): @@ -17,21 +15,81 @@ def test_build_factor_snapshot_falls_back_to_sample(): symbols=("999999",), sample_fallback_path=sample_path, min_rows=1, + as_of="2026-06-27", ) assert diagnostics["source"] == "sample_fallback" - assert list(frame.columns) == list(FACTOR_SNAPSHOT_COLUMNS) + assert "as_of" in frame.columns assert len(frame) >= 1 -def test_write_staging_factor_snapshot(tmp_path): - sample_path = Path(__file__).resolve().parents[1] / "examples" / "dividend_quality" / "factor_snapshot.sample.csv" - output_path = tmp_path / "factor_snapshot.latest.csv" - diagnostics = write_staging_factor_snapshot( - output_path=output_path, - symbols=("999999",), - sample_fallback_path=sample_path, +def test_build_factor_row_from_mocked_akshare_sources(): + history = pd.DataFrame( + { + "日期": pd.bdate_range("2024-01-02", periods=260), + "收盘": [100 + idx * 0.1 for idx in range(260)], + "成交额": [80_000_000.0] * 260, + "成交量": [900_000] * 260, + } + ) + financials = pd.DataFrame( + { + "日期": ["2025-03-31", "2025-06-30"], + "净资产报酬率(%)": [10.0, 11.0], + "摊薄每股收益(元)": [2.0, 2.2], + } + ) + dividends = pd.DataFrame({"除权除息日": ["2025-06-20"], "派息": [250.0]}) + fhps = pd.DataFrame( + [ + { + "代码": "600519", + "symbol": "600519", + "名称": "贵州茅台", + "现金分红-股息率": 0.028, + "现金分红-现金分红比例": 250.0, + "每股收益": 50.0, + "总股本": 1_000_000_000, + } + ] + ) + + row = build_factor_row_from_akshare( + "600519", + fhps_table=fhps, + fetch_history=lambda _symbol: history, + fetch_financials=lambda _symbol: financials, + fetch_dividends=lambda _symbol: dividends, + fetch_sector=lambda _symbol: "白酒", + ) + assert row["symbol"] == "600519" + assert row["sector"] == "白酒" + assert float(row["dividend_yield_ttm"]) == pytest.approx(0.028) + assert float(row["roe_ttm"]) == pytest.approx(0.11) + assert float(row["realized_vol_126"]) > 0 + + +def test_build_market_history_frame_from_mocked_fetchers(): + def _fetch(symbol: str) -> pd.DataFrame: + return pd.DataFrame( + { + "日期": pd.bdate_range("2024-01-02", periods=3), + "收盘": [10.0, 10.1, 10.2], + } + ) + + from cn_equity_snapshot_pipelines import akshare_market_history as module + + original = module.fetch_etf_history + module.fetch_etf_history = lambda symbol, **kwargs: pd.DataFrame( + { + "date": ["2024-01-02", "2024-01-03", "2024-01-04"], + "symbol": normalize_symbol(symbol), + "close": [10.0, 10.1, 10.2], + } ) - assert output_path.exists() - frame = pd.read_csv(output_path) - assert list(frame.columns) == list(FACTOR_SNAPSHOT_COLUMNS) - assert diagnostics["output_path"] == str(output_path) + try: + frame = build_market_history_frame(("510300", "510500"), ak=object()) + assert set(frame["symbol"]) == {"510300", "510500"} + assert len(frame) == 6 + finally: + module.fetch_etf_history = original From 3132bab92fbe781fd31a0cc60bfa99d9c68b4c83 Mon Sep 17 00:00:00 2001 From: Pigbibi <20649888+Pigbibi@users.noreply.github.com> Date: Sat, 27 Jun 2026 23:29:28 +0800 Subject: [PATCH 2/2] Fix mocked AkShare row build without importing akshare. Co-authored-by: Cursor --- src/cn_equity_snapshot_pipelines/akshare_staging.py | 10 +++++++++- 1 file changed, 9 insertions(+), 1 deletion(-) diff --git a/src/cn_equity_snapshot_pipelines/akshare_staging.py b/src/cn_equity_snapshot_pipelines/akshare_staging.py index 3ef5e69..ff0c440 100644 --- a/src/cn_equity_snapshot_pipelines/akshare_staging.py +++ b/src/cn_equity_snapshot_pipelines/akshare_staging.py @@ -104,7 +104,15 @@ def build_factor_row_from_akshare( fetch_dividends: Callable[[str], pd.DataFrame] | None = None, fetch_sector: Callable[[str], str] | None = None, ) -> dict[str, object]: - ak_module = ak or _import_akshare() + needs_akshare = any( + item is None + for item in (fetch_history, fetch_financials, fetch_dividends, fetch_sector, fhps_table) + ) + ak_module = ak + if needs_akshare and ak_module is None: + ak_module = _import_akshare() + if fhps_table is None and ak_module is not None: + fhps_table = _fetch_fhps_table(ak_module) history_loader = fetch_history or (lambda item: _fetch_history(ak_module, item)) financial_loader = fetch_financials or (lambda item: _fetch_financials(ak_module, item)) dividend_loader = fetch_dividends or (lambda item: _fetch_dividends(ak_module, item))