From d7482904127da946504aa86c739545ee66ef1d38 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Tue, 8 Sep 2026 18:08:05 +0800 Subject: [PATCH] =?UTF-8?q?feat(factor):=20600519=E5=B8=82=E5=80=BC?= =?UTF-8?q?=E9=94=9A=E7=82=B9sanity=E8=84=9A=E6=9C=AC=E2=80=94=E2=80=94?= =?UTF-8?q?=E8=87=AA=E7=AE=97MV(close=C3=97=E8=82=A1=E6=9C=AC)vs=20valuati?= =?UTF-8?q?on=E6=80=BB=E5=B8=82=E5=80=BC=20[nas]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/factor_research/verify_mv_anchor.py | 108 ++++++++++++++++++++ tests/factor/test_verify_mv_anchor.py | 102 ++++++++++++++++++ 2 files changed, 210 insertions(+) create mode 100644 scripts/factor_research/verify_mv_anchor.py create mode 100644 tests/factor/test_verify_mv_anchor.py diff --git a/scripts/factor_research/verify_mv_anchor.py b/scripts/factor_research/verify_mv_anchor.py new file mode 100644 index 0000000..608eb34 --- /dev/null +++ b/scripts/factor_research/verify_mv_anchor.py @@ -0,0 +1,108 @@ +#!/usr/bin/env python +"""600519 市值锚点 sanity: 三表自算 MV(收盘×SHARE_CAPITAL) vs valuation.总市值. + +目的(P1 随批互评备注 3): 估值族因子市值统一 close×share_capital 自算(规避 +valuation 中文列名表,PIT 口径与报表一致)——本脚本对 600519.SH 2024 年报期 +抽样对账,自算 MV 与 valuation 域总市值的相对偏差应在个位数百分比内 +(股本口径/复权/停牌日错位是常见小幅来源;对不上先查口径再查代码)。 + +用法: + NAS 容器内/外同路径: python scripts/factor_research/verify_mv_anchor.py \ + [--static-dir /volume1/stock/sanguo_vnpy_v2/data/static] [--code 600519.SH] \ + [--report 2024-12-31] + 本机(无 NAS 数据) → no-op 早退 rc=0。 +""" +import argparse +import os +import sys +from datetime import date, datetime + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) + +import polars as pl + +DEFAULT_STATIC_DIR = "/volume1/stock/sanguo_vnpy_v2/data/static" +DEVIATION_WARN = 0.05 # 5% 超限提示口径核查(非失败) + + +def mv_deviation(close: float | None, share_capital: float | None, + total_mv: float | None) -> float | None: + """自算 MV 相对 valuation 总市值的偏差 = close×share/总市值 − 1. + + 纯函数(单测复用);任一输入缺失/总市值≤0 → None。 + """ + if close is None or share_capital is None or total_mv is None or total_mv <= 0: + return None + return close * share_capital / total_mv - 1.0 + + +def _to_date(v) -> date | None: + if v is None: + return None + return v if isinstance(v, date) else datetime.strptime(str(v)[:10], "%Y-%m-%d").date() + + +def load_share_capital(static_dir: str, file_code: str, report: str): + """balance 表该报告期行 → (share_capital, 有效披露日 notice_date).""" + path = os.path.join(static_dir, "balance", f"{file_code}_balance.parquet") + if not os.path.exists(path): + return None, None + df = pl.read_parquet(path) + rd = _to_date(report) + row = df.filter(pl.col("REPORT_DATE").cast(pl.Utf8).str.slice(0, 10) + == rd.strftime("%Y-%m-%d")) + if row.height == 0 or "SHARE_CAPITAL" not in row.columns: + return None, None + share = row["SHARE_CAPITAL"][0] + notice = row["NOTICE_DATE"][0] if "NOTICE_DATE" in row.columns else None + return (None if share is None else float(share)), _to_date(notice) + + +def load_valuation_row(static_dir: str, file_code: str, on_or_before: date): + """valuation 表 ≤ on_or_before 的最新行 → (当日收盘价, 总市值).""" + path = os.path.join(static_dir, "valuation", f"{file_code}_valuation.parquet") + if not os.path.exists(path): + return None, None + df = pl.read_parquet(path, columns=["数据日期", "当日收盘价", "总市值"]) + df = df.with_columns(pl.col("数据日期").cast(pl.Utf8).str.slice(0, 10) + .str.to_date("%Y-%m-%d", strict=False).alias("_d")) + row = (df.filter(pl.col("_d").is_not_null() & (pl.col("_d") <= on_or_before)) + .sort("_d").tail(1)) + if row.height == 0: + return None, None + close, mv = row["当日收盘价"][0], row["总市值"][0] + return (None if close is None else float(close)), (None if mv is None else float(mv)) + + +def main() -> int: + ap = argparse.ArgumentParser(description=__doc__) + ap.add_argument("--static-dir", default=DEFAULT_STATIC_DIR) + ap.add_argument("--code", default="600519.SH", help="文件名代码(带交易所后缀)") + ap.add_argument("--report", default="2024-12-31", help="报告期(年报)") + args = ap.parse_args() + + if not os.path.isdir(args.static_dir): + print(f"[skip] 静态域不存在: {args.static_dir}(本机无 NAS 数据,no-op)") + return 0 + + share, notice = load_share_capital(args.static_dir, args.code, args.report) + if share is None: + print(f"[skip] {args.code} {args.report} 无 SHARE_CAPITAL 行") + return 0 + # 对账日 = 年报披露日(报表口径与行情同日对齐;披露日非交易日则取此前最近) + anchor_day = notice or _to_date(args.report) + close, total_mv = load_valuation_row(args.static_dir, args.code, anchor_day) + dev = mv_deviation(close, share, total_mv) + if dev is None: + print(f"[skip] {args.code} valuation 行缺失或量纲异常 " + f"(close={close}, share={share}, total_mv={total_mv})") + return 0 + flag = "⚠️ 超 5%,核查口径(股本/复权/停牌错位)" if abs(dev) > DEVIATION_WARN else "OK" + print(f"[{args.code} {args.report}] 披露日={anchor_day} close={close} " + f"share_capital={share:.0f} 自算MV={close * share:.4e} " + f"valuation总市值={total_mv:.4e} 相对偏差={dev:+.4%} → {flag}") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tests/factor/test_verify_mv_anchor.py b/tests/factor/test_verify_mv_anchor.py new file mode 100644 index 0000000..d83d7fa --- /dev/null +++ b/tests/factor/test_verify_mv_anchor.py @@ -0,0 +1,102 @@ +# tests/factor/test_verify_mv_anchor.py +"""600519 市值锚点 sanity 脚本: 纯函数 + 合成 valuation/balance 自洽性. + +互评备注 3: 脚本本机跑不了 NAS(no-op 早退),但自算 MV 与 valuation 市值列的 +自洽性要在合成数据单测里锁死——夹具里 share_capital 与 valuation 总市值 +同源生成(总市值 ≡ close×share),偏差应精确为 0。 +""" +import os +import sys +from datetime import date + +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))) +sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..", "vnpy_v4.4.0"))) + +import polars as pl +import pytest + +from scripts.factor_research.verify_mv_anchor import ( + mv_deviation, load_share_capital, load_valuation_row) + + +def test_mv_deviation_pure_function(): + # 同源自洽: 总市值 = close×share → 偏差精确 0 + assert mv_deviation(12.5, 1.0e9, 12.5e9) == pytest.approx(0.0, abs=1e-12) + # 2% 高估 + assert mv_deviation(10.2, 100.0, 1000.0) == pytest.approx(0.02) + # 缺输入/非正总市值 → None + assert mv_deviation(None, 100.0, 1000.0) is None + assert mv_deviation(10.0, None, 1000.0) is None + assert mv_deviation(10.0, 100.0, None) is None + assert mv_deviation(10.0, 100.0, 0.0) is None + + +def _write_fixture(root: str, *, consistent: bool = True): + """合成 mini 静态域: balance(2024 年报 share/notice) + valuation 日频.""" + static = os.path.join(root, "static") + os.makedirs(os.path.join(static, "balance"), exist_ok=True) + os.makedirs(os.path.join(static, "valuation"), exist_ok=True) + share, notice, close = 1.256e9, date(2025, 4, 2), 1500.0 + pl.DataFrame([{ + "REPORT_DATE": "2024-12-31 00:00:00", "NOTICE_DATE": "2025-04-02 00:00:00", + "SHARE_CAPITAL": share, + }]).write_parquet(os.path.join(static, "balance", "600519.SH_balance.parquet")) + rows = [] + for d in (date(2025, 3, 28), date(2025, 3, 31), date(2025, 4, 1), + date(2025, 4, 2), date(2025, 4, 3)): + c = close * (1.0 if d <= notice else 1.01) + mv = c * share * (1.0 if consistent else 1.3) + rows.append({"数据日期": str(d), "当日收盘价": c, "总市值": mv}) + pl.DataFrame(rows).write_parquet( + os.path.join(static, "valuation", "600519.SH_valuation.parquet")) + return share, notice, close + + +def test_self_consistent_mv(tmp_path): + """同源夹具(总市值 ≡ close×share): 全链路读出的偏差 = 0.""" + share, notice, close = _write_fixture(str(tmp_path), consistent=True) + static = os.path.join(str(tmp_path), "static") + got_share, got_notice = load_share_capital(static, "600519.SH", "2024-12-31") + assert got_share == pytest.approx(share) + assert got_notice == notice + got_close, got_mv = load_valuation_row(static, "600519.SH", notice) + assert got_close == pytest.approx(close) + assert mv_deviation(got_close, got_share, got_mv) == pytest.approx(0.0, abs=1e-12) + + +def test_inconsistent_mv_detected(tmp_path): + """总市值掺 30% 水分 → 偏差应被量出(≈ −23%,检测能力下限).""" + share, notice, close = _write_fixture(str(tmp_path), consistent=False) + static = os.path.join(str(tmp_path), "static") + got_share, _ = load_share_capital(static, "600519.SH", "2024-12-31") + got_close, got_mv = load_valuation_row(static, "600519.SH", notice) + dev = mv_deviation(got_close, got_share, got_mv) + assert dev == pytest.approx(1 / 1.3 - 1) + + +def test_valuation_row_picks_latest_on_or_before(tmp_path): + """对账日取 ≤ 披露日的最新 valuation 行(披露日后涨价行不参与).""" + share, notice, close = _write_fixture(str(tmp_path), consistent=True) + static = os.path.join(str(tmp_path), "static") + got_close, _ = load_valuation_row(static, "600519.SH", notice) + assert got_close == pytest.approx(close) # 非 1.01×close 的次日行 + + +def test_missing_paths_return_none(tmp_path): + static = os.path.join(str(tmp_path), "static") + assert load_share_capital(static, "600519.SH", "2024-12-31") == (None, None) + assert load_valuation_row(static, "600519.SH", date(2025, 4, 2)) == (None, None) + + +def test_main_noop_without_nas_data(tmp_path, capsys): + """本机无 NAS 数据 → no-op 早退 rc=0(P1 任务书要求).""" + from scripts.factor_research import verify_mv_anchor as mod + old_argv = sys.argv + try: + sys.argv = ["verify_mv_anchor.py", "--static-dir", + os.path.join(str(tmp_path), "nowhere")] + rc = mod.main() + finally: + sys.argv = old_argv + assert rc == 0 + assert "[skip]" in capsys.readouterr().out