diff --git a/pyproject.toml b/pyproject.toml index bf72db5..3714710 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -19,6 +19,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" [tool.setuptools] package-dir = {"" = "src"} diff --git a/src/cn_equity_snapshot_pipelines/akshare_staging.py b/src/cn_equity_snapshot_pipelines/akshare_staging.py new file mode 100644 index 0000000..32e340a --- /dev/null +++ b/src/cn_equity_snapshot_pipelines/akshare_staging.py @@ -0,0 +1,171 @@ +from __future__ import annotations + +import argparse +from pathlib import Path + +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", +) + +DEFAULT_STAGING_SYMBOLS = ( + "600519", + "601088", + "000001", + "600036", + "601398", + "600900", + "601318", + "600028", +) + + +def _load_sample_fallback(sample_path: Path) -> pd.DataFrame: + frame = pd.read_csv(sample_path) + missing = [column for column in FACTOR_SNAPSHOT_COLUMNS if column not in frame.columns] + if missing: + missing_text = ", ".join(missing) + raise ValueError(f"sample fallback missing columns: {missing_text}") + 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 _fetch_spot_rows() -> pd.DataFrame: + import akshare as ak + + spot = ak.stock_zh_a_spot_em() + spot = spot.rename( + columns={ + "代码": "symbol", + "名称": "name", + "最新价": "close_cny", + "成交额": "turnover_cny", + "总市值": "market_cap_cny", + "市盈率-动态": "pe_ttm", + } + ) + spot["symbol"] = spot["symbol"].map(_normalize_symbol) + return spot + + +def build_factor_snapshot_from_akshare( + *, + symbols: tuple[str, ...] = DEFAULT_STAGING_SYMBOLS, + sample_fallback_path: str | Path | None = None, + min_rows: int = 4, +) -> tuple[pd.DataFrame, dict[str, object]]: + diagnostics: dict[str, object] = {"source": "akshare", "requested_symbols": list(symbols)} + try: + spot = _fetch_spot_rows() + 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)) + 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: + diagnostics["source"] = "sample_fallback" + diagnostics["akshare_error"] = f"only {len(filtered)} rows matched requested symbols" + if sample_fallback_path is None: + raise ValueError(diagnostics["akshare_error"]) + frame = _load_sample_fallback(Path(sample_fallback_path)) + 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)) + diagnostics["row_count"] = len(frame) + return frame, diagnostics + + +def write_staging_factor_snapshot( + *, + output_path: str | Path, + symbols: tuple[str, ...] = DEFAULT_STAGING_SYMBOLS, + sample_fallback_path: str | Path | None = None, +) -> dict[str, object]: + frame, diagnostics = build_factor_snapshot_from_akshare( + symbols=symbols, + sample_fallback_path=sample_fallback_path, + ) + path = Path(output_path) + path.parent.mkdir(parents=True, exist_ok=True) + frame.to_csv(path, index=False) + diagnostics["output_path"] = str(path) + return diagnostics + + +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( + "--sample-fallback", + default=str(Path(__file__).resolve().parents[2] / "examples" / "dividend_quality" / "factor_snapshot.sample.csv"), + ) + args = parser.parse_args(argv) + symbols = tuple(symbol.strip() for symbol in args.symbols.split(",") if symbol.strip()) + diagnostics = write_staging_factor_snapshot( + output_path=args.output, + symbols=symbols, + sample_fallback_path=args.sample_fallback, + ) + print(diagnostics) + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tests/test_akshare_staging.py b/tests/test_akshare_staging.py new file mode 100644 index 0000000..a66e5e1 --- /dev/null +++ b/tests/test_akshare_staging.py @@ -0,0 +1,37 @@ +from __future__ import annotations + +from pathlib import Path + +import pandas as pd + +from cn_equity_snapshot_pipelines.akshare_staging import ( + FACTOR_SNAPSHOT_COLUMNS, + build_factor_snapshot_from_akshare, + write_staging_factor_snapshot, +) + + +def test_build_factor_snapshot_falls_back_to_sample(): + sample_path = Path(__file__).resolve().parents[1] / "examples" / "dividend_quality" / "factor_snapshot.sample.csv" + frame, diagnostics = build_factor_snapshot_from_akshare( + symbols=("999999",), + sample_fallback_path=sample_path, + min_rows=1, + ) + assert diagnostics["source"] == "sample_fallback" + assert list(frame.columns) == list(FACTOR_SNAPSHOT_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, + ) + 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)