# 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