Files
sanguo_vnpy_v2/tests/factor/test_daily_section.py
T
claude_dev 45bc8df46f
CI/CD / test (push) Successful in 29s
CI/CD / nas-deploy (push) Successful in 13s
CI/CD / nas-verify (push) Successful in 11s
[nas] fix(factor): universe 缓存补跑截数据防线——purge_cache 单点+daily_section 跑批前必清 [no-doc]
NAS 实证: 09-22 晚生成的 universe_cache_0214_0922 止于库内 09-21, 次晨库已
同步 09-22 但同窗补跑命中旧缓存 → 截面缺 09-22 行(section_sanity fail-loud
守卫立功)。生产场景每日新窗=新 key 天然免疫, 补跑场景必中。修=universe.
cache_path_for 公式单点 + purge_cache 删除;daily_section 跑批前必清同窗
缓存(vnpy_db=None 时走 cfg 解析同款路径)。

Co-Authored-By: Claude Code <notify@anthropic.com>
2026-09-24 08:04:45 +08:00

182 lines
7.2 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
# tests/factor/test_daily_section.py
"""每日截面增量生产器(影子盘数据契约): 幂等 merge + sanity + 窗推导."""
import os
import sys
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "..")))
import pandas as pd
import pytest
from sanguo_factor import daily_section
def _wide(days, base=1.0):
"""days(list[str]) × 3 股宽表, 值=base 递增可区分."""
cols = ["600000.SSE", "000001.SZSE", "300001.SZSE"]
idx = pd.to_datetime(days)
return pd.DataFrame({c: [base + i for _ in range(len(days))]
for i, c in enumerate(cols)}, index=idx)
def test_merge_appends_new_date(tmp_path):
main_p = tmp_path / "main.parquet"
new_p = tmp_path / "new.parquet"
_wide(["2026-09-10", "2026-09-11"]).to_parquet(main_p)
_wide(["2026-09-14"], base=9.0).to_parquet(new_p)
out = daily_section.merge_section(str(main_p), str(new_p))
got = pd.read_parquet(main_p)
assert list(got.index.strftime("%Y-%m-%d")) == \
["2026-09-10", "2026-09-11", "2026-09-14"]
assert (got.loc["2026-09-14"].to_numpy() == [9.0, 10.0, 11.0]).all()
assert out["added_dates"] == 1 and out["replaced_dates"] == 0
def test_merge_replaces_existing_date(tmp_path):
main_p = tmp_path / "main.parquet"
_wide(["2026-09-10", "2026-09-11"]).to_parquet(main_p)
_wide(["2026-09-11"], base=7.0).to_parquet(tmp_path / "new.parquet")
out = daily_section.merge_section(str(main_p), str(tmp_path / "new.parquet"))
got = pd.read_parquet(main_p)
assert len(got) == 2 # 不增行
assert (got.loc["2026-09-11"].to_numpy() == [7.0, 8.0, 9.0]).all()
assert out["added_dates"] == 0 and out["replaced_dates"] == 1
def test_merge_idempotent_rerun_same_day(tmp_path):
"""同日重跑: 第二次 merge 无新增日期(added=0, replaced=同日覆盖计数),
值不变(幂等)."""
main_p = tmp_path / "main.parquet"
new_p = tmp_path / "new.parquet"
_wide(["2026-09-10"]).to_parquet(main_p)
_wide(["2026-09-11"], base=5.0).to_parquet(new_p)
daily_section.merge_section(str(main_p), str(new_p))
snap = pd.read_parquet(main_p)
out2 = daily_section.merge_section(str(main_p), str(new_p))
assert out2["added_dates"] == 0 and out2["rows_after"] == len(snap)
pd.testing.assert_frame_equal(pd.read_parquet(main_p), snap)
def test_sanity_counts_and_turnover(tmp_path):
"""sanity: 当日行非空数 + 与前一行的进出股票数."""
main_p = tmp_path / "main.parquet"
d1 = _wide(["2026-09-10"])
d1["600000.SSE"] = float("nan") # 当日缺值
d1["688001.SSE"] = 1.0 # 新上市列
d2 = _wide(["2026-09-11"])
d2 = d2.drop(columns=["300001.SZSE"]) # 退市消失
pd.concat([d1, d2]).to_parquet(main_p)
s = daily_section.section_sanity(str(main_p), "2026-09-11")
assert s["date"] == "2026-09-11"
assert s["non_null"] == 2 # 600000/000001
# vs 09-10 的非空集: 688001(昨有今无)+300001(退市消失)出,
# 600000 昨 NaN 今有值 → 视为进(在场=非空口径)
assert set(s["entered"]) == {"600000.SSE"}
assert set(s["exited"]) == {"688001.SSE", "300001.SZSE"}
def test_lookback_window_dates():
"""窗推导: end 往前 lookback 历史日, 覆盖 WARMUP_BARS(120 交易日)预热."""
start, end = daily_section.lookback_window("2026-09-22", lookback_days=220)
assert end == "2026-09-22"
span = (pd.Timestamp(end) - pd.Timestamp(start)).days
assert 218 <= span <= 222
# 默认 lookback 足够 120 交易日预热(~170 历史日)
assert span >= 170
def test_purge_cache_formula_and_delete(tmp_path):
"""universe 缓存: 公式单点 + purge 删除(补跑截数据防线, NAS 09-22 实证)."""
from sanguo_factor import universe
db = str(tmp_path / "qt.db")
p = universe.cache_path_for(db, "2026-02-14", "2026-09-22")
assert p.endswith("universe_cache_2026-02-14_2026-09-22_all.parquet")
assert universe.purge_cache(db, "2026-02-14", "2026-09-22") is False # 无文件
from pathlib import Path
Path(p).write_bytes(b"x")
assert universe.purge_cache(db, "2026-02-14", "2026-09-22") is True
assert not Path(p).exists()
def test_run_daily_section_purges_universe_cache(db_stock_small, tmp_path, monkeypatch):
"""跑批前必清同窗 universe 缓存(旧缓存截新数据,09-22 NAS 补跑实证)."""
from sanguo_factor import universe
called: list[tuple[str, str]] = []
monkeypatch.setattr(universe, "purge_cache",
lambda db, s, e: called.append((s, e)) or True)
main_p = tmp_path / "sections" / "vma_60.parquet"
main_p.parent.mkdir(parents=True)
_wide(["2026-07-09"]).to_parquet(main_p)
daily_section.run_daily_section(
"vma_60", end_date="2026-07-10", out_dir=str(tmp_path / "sections"),
vnpy_db=db_stock_small, eval_db=str(tmp_path / "e.db"),
tmp_dir=str(tmp_path / "tmpcs"))
start, end = daily_section.lookback_window("2026-07-10")
assert (start, end) in called
def test_run_daily_section_end_to_end(db_stock_small, tmp_path, monkeypatch):
"""端到端: 合成 5 股 sqlite → 短窗跑批导出 → merge 主 parquet → sanity 齐."""
from sanguo_factor.batch_eval import run_batch_eval # noqa: F401 导入链可用性
main_p = tmp_path / "sections" / "vma_60.parquet"
main_p.parent.mkdir(parents=True)
_wide(["2026-07-09"]).to_parquet(main_p) # 既有主截面(生产日前一交易日)
# 合成库 bar 终点 ~2026-07(400 个工作日), 截面日取 bar 内最后交易日
out = daily_section.run_daily_section(
"vma_60", end_date="2026-07-10", out_dir=str(tmp_path / "sections"),
vnpy_db=db_stock_small, eval_db=str(tmp_path / "e.db"),
tmp_dir=str(tmp_path / "tmpcs"))
assert out["ok"] is True
got = pd.read_parquet(main_p)
dates = list(got.index.strftime("%Y-%m-%d"))
assert dates[-1] == "2026-07-10" and len(dates) > 1
assert out["sanity"]["date"] == "2026-07-10"
assert out["sanity"]["non_null"] > 0
# ---------------- 合成小库: 5 股 × ~400 bar(覆盖 warmup) ----------------
@pytest.fixture(scope="module")
def db_stock_small(tmp_path_factory):
import numpy as np
import sqlite3
ddl = """
CREATE TABLE dbbardata(
symbol TEXT, exchange TEXT, datetime TEXT, interval TEXT,
volume REAL, turnover REAL, open_interest REAL,
open_price REAL, high_price REAL, low_price REAL, close_price REAL)
"""
p = tmp_path_factory.mktemp("dsec")
db = str(p / "qt.db")
conn = sqlite3.connect(db)
conn.execute(ddl)
days = pd.bdate_range("2025-01-01", periods=400)
rng = np.random.default_rng(11)
for j, (sym, ex) in enumerate([("600000", "SSE"), ("000001", "SZSE"),
("300001", "SZSE"), ("600004", "SSE"),
("000333", "SZSE")]):
px = 10.0 + j
for day in days:
px = max(px * (1 + rng.normal(0, 0.015)), 0.5)
conn.execute(
"INSERT INTO dbbardata VALUES(?,?,?,?,?,?,?,?,?,?,?)",
(sym, ex, f"{day.date()} 00:00:00", "d", float(rng.integers(1e5, 9e5)),
px * 1e5, 0, px, px * 1.01, px * 0.99, px))
conn.commit()
conn.close()
return db