feat(factor): 600519市值锚点sanity脚本——自算MV(close×股本)vs valuation总市值 [nas]
This commit is contained in:
@@ -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())
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user