"""增量写路径合成体检: 在副本树上真跑 update_daily(0911), 验旧行不变/新行正确/--check/负向兜底 用法: python /tmp/preflight.py # 建树 + 跑 + 校验 """ import os, sys, shutil, subprocess, glob import pandas as pd REAL = '/Users/xpresso/zt_app/backtest_zt_full' TEST = '/tmp/zt_preflight' PY = '/Users/xpresso/zt_app/zt_venv/bin/python' POOLS = ['hs300', 'zz500', 'zz1000', 'zz2000'] DATE = '20260911' # 每池取 3 只(其中优先挑当前持仓/当天有信号的股, 保证与真库可比) PREFER = ['002902.SZ', '603980.SH', '603115.SH', '603093.SH', '000567.SZ', '601058.SH', '600309.SH', '600095.SH', '002271.SZ', '002625.SZ'] PICK = {} for p in POOLS: cands = [c for c in PREFER if os.path.exists(f'{REAL}/daily_{p}/{c}.csv')] if len(cands) < 3: extra = sorted(os.path.basename(f)[:-4] for f in glob.glob(f'{REAL}/daily_{p}/*.csv') if os.path.basename(f)[:-4] not in cands) cands += extra[:3 - len(cands)] PICK[p] = sorted(cands)[:3] print("抽检:", PICK) if os.path.exists(TEST): shutil.rmtree(TEST) import pickle for p in POOLS: os.makedirs(f'{TEST}/daily_{p}', exist_ok=True) with open(f'{REAL}/{p}_members.pkl', 'rb') as fh: mem = pickle.load(fh) with open(f'{TEST}/{p}_members.pkl', 'wb') as fh: pickle.dump(mem, fh) pre_text = {} for p, codes in PICK.items(): for c in codes: src = f'{REAL}/daily_{p}/{c}.csv' if not os.path.exists(src): print('缺文件', src); continue d = pd.read_csv(src, dtype={'trade_date': str}) cut = d[d['trade_date'] <= '20260910'] dst = f'{TEST}/daily_{p}/{c}.csv' cut.to_csv(dst, index=False) pre_text[(p, c)] = open(dst).read() # 键含 pool: 同一 code 可能出现在多个池 print("测试树就绪:", sum(len(v) for v in PICK.values()), "文件") print("\n=== ① 正常路径真跑(写盘) ===") r = subprocess.run([PY, 'update_daily.py', DATE], cwd='/Users/xpresso/zt_app/code', env={**os.environ, 'ZT_BASE': TEST}, capture_output=True, text=True) print(r.stdout[-500:] or '', r.stderr[-300:] or '') print("exit=", r.returncode) # ② 旧行逐行不变? changed = [] for p, codes in PICK.items(): for c in codes: f = f'{TEST}/daily_{p}/{c}.csv' now = open(f).read() old_lines = pre_text[(p, c)].rstrip('\n').split('\n') new_lines = now.rstrip('\n').split('\n') if new_lines[:len(old_lines)] != old_lines: changed.append(c) print(f"② 旧行文本被改写: {len(changed)}/{sum(len(v) for v in PICK.values())} {changed}") # ③ 追加行列数/NaN/与真库逐值一致 real = pd.read_csv(f'{REAL}/daily_zz2000/002902.SZ.csv', dtype={'trade_date': str}) bad_cols, nan_rows, mism, n = [], [], [], 0 for p, codes in PICK.items(): for c in codes: f = f'{TEST}/daily_{p}/{c}.csv' d = pd.read_csv(f, dtype={'trade_date': str}) n += 1 if len(d.columns) != 12: bad_cols.append((c, len(d.columns))) row = d[d['trade_date'] == DATE] if not len(row): mism.append((c, '无新行')); continue row = row.iloc[0] if row.isna().any(): nan_rows.append((c, list(row[row.isna()].index))) ref_f = f'{REAL}/daily_{p}/{c}.csv' if os.path.exists(ref_f): rf = pd.read_csv(ref_f, dtype={'trade_date': str}) rr = rf[rf['trade_date'] == DATE] if len(rr): rr = rr.iloc[0] diffs = {k: (float(row[k]) - float(rr[k])) for k in ['raw_open', 'raw_high', 'raw_low', 'raw_close', 'adj_factor', 'pct_chg', 'vol'] if k in row and k in rr and pd.notna(row[k]) and pd.notna(rr[k])} if any(abs(v) > 1e-9 for v in diffs.values()): mism.append((c, {k: round(v, 8) for k, v in diffs.items() if abs(v) > 1e-9})) print(f"③ 列数≠12: {bad_cols or '无'}; 新行含 NaN: {nan_rows or '无'}; 与真库 0911 行不一致: {mism or '无'} (共 {n} 文件)") # ④ --check on copy r2 = subprocess.run([PY, 'update_daily.py', '--check'], cwd='/Users/xpresso/zt_app/code', env={**os.environ, 'ZT_BASE': TEST}, capture_output=True, text=True) print("④ --check(副本):", [l for l in r2.stdout.split('\n') if '体检' in l or '结论' in l]) # ⑤ 负向: ZT_NO_FAC=1 -> 因子沿用文件末行(0910)因子, 拒写 0 TEST2 = TEST + '_nofac' if os.path.exists(TEST2): shutil.rmtree(TEST2) shutil.copytree(TEST, TEST2) for p, codes in PICK.items(): for c in codes: f = f'{TEST2}/daily_{p}/{c}.csv' d = pd.read_csv(f, dtype={'trade_date': str}) d = d[d['trade_date'] <= '20260910'] # 回退到体检前状态 d.to_csv(f, index=False) r3 = subprocess.run([PY, 'update_daily.py', DATE], cwd='/Users/xpresso/zt_app/code', env={**os.environ, 'ZT_BASE': TEST2, 'ZT_NO_FAC': '1'}, capture_output=True, text=True) print("⑤ ZT_NO_FAC=1:", [l for l in r3.stdout.split('\n') if '汇总' in l or '因子回退' in l]) ok = 0 for p, codes in PICK.items(): for c in codes: d = pd.read_csv(f'{TEST2}/daily_{p}/{c}.csv', dtype={'trade_date': str}) last_prev = d[d['trade_date'] == '20260910'] row = d[d['trade_date'] == DATE] if len(last_prev) and len(row): if abs(float(row.iloc[0]['adj_factor']) - float(last_prev.iloc[0]['adj_factor'])) < 1e-9: ok += 1 print(f"⑤ 沿用值 == 文件末行(0910)因子: {ok}/{n}") # ⑥ 混 schema 负向: 遗留 19 列文件 -> 应"拒写且文件不变"(而非静默写成 20 列+NaN) LEGACY_BAK = '/Users/xpresso/zt_app/_hfq_ohlc_backup_20260912' TEST3 = TEST + '_legacy' if os.path.exists(TEST3): shutil.rmtree(TEST3) shutil.copytree(TEST, TEST3) for p, codes in PICK.items(): for c in codes: f = f'{TEST3}/daily_{p}/{c}.csv' d = pd.read_csv(f, dtype={'trade_date': str}) d = d[d['trade_date'] <= '20260910'] d.to_csv(f, index=False) legacy_code = None for p, codes in PICK.items(): for c in codes: lb = f'{LEGACY_BAK}/{c}.csv' if os.path.exists(lb): dl = pd.read_csv(lb, dtype={'trade_date': str}) dl = dl[dl['trade_date'] <= '20260910'] if len(dl): dl.to_csv(f'{TEST3}/daily_{p}/{c}.csv', index=False) legacy_code, legacy_pool = c, p break if legacy_code: break print(f"\n⑥ 混 schema 负向: 把 {legacy_pool}/{legacy_code} 换成遗留 {len(dl.columns)} 列文件") before = pd.read_csv(f'{TEST3}/daily_{legacy_pool}/{legacy_code}.csv', dtype={'trade_date': str}) r4 = subprocess.run([PY, 'update_daily.py', DATE], cwd='/Users/xpresso/zt_app/code', env={**os.environ, 'ZT_BASE': TEST3}, capture_output=True, text=True) print(" ", [l for l in r4.stdout.split('\n') if '汇总' in l or 'schema 不符' in l]) after = pd.read_csv(f'{TEST3}/daily_{legacy_pool}/{legacy_code}.csv', dtype={'trade_date': str}) after2 = pd.read_csv(f'{TEST3}/daily_{legacy_pool}/{legacy_code}.csv', dtype={'trade_date': str}) print(f" 遗留文件列数 {len(before.columns)} → {len(after.columns)}; 行数 {len(before)} → {len(after)} " f"(应均为 19 列/不变); 被静默写成 20 列? {'是 ❌' if len(after.columns) == 20 else '否 ✅'}") # ⑦ 除权守卫负向(2026-09-13, B-only 判据): ②因子兜底 × 当日除权 → 必须拒写。 # 构造法: 把某股 0910 的 raw_close 抬高(模拟除权前价), 则 0911 的 # B=raw_close/前日raw_close 与 pct_chg 失配 → 判除权(判据单点 code/exdiv_judge.py)。 TEST4 = TEST + '_exdiv' if os.path.exists(TEST4): shutil.rmtree(TEST4) shutil.copytree(TEST, TEST4) for p, codes in PICK.items(): for c in codes: f = f'{TEST4}/daily_{p}/{c}.csv' d = pd.read_csv(f, dtype={'trade_date': str}) d.to_csv(f, index=False) tgt_pool, tgt_code = sorted(PICK.items())[0][0], sorted(PICK.items())[0][1][0] f4 = f'{TEST4}/daily_{tgt_pool}/{tgt_code}.csv' d4 = pd.read_csv(f4, dtype={'trade_date': str}) d4 = d4[d4['trade_date'] <= '20260910'] m = d4['trade_date'] == '20260910' d4.loc[m, 'raw_close'] = round(float(d4.loc[m, 'raw_close'].iloc[0]) * 1.5, 3) # 模拟除权前价 d4.to_csv(f4, index=False) ctrl4 = f'{TEST4}/daily_{tgt_pool}/{sorted(PICK[tgt_pool])[-1]}.csv' print(f"\n⑦ 除权守卫负向: 抬 {tgt_pool}/{tgt_code} 的 0910 raw_close ×1.5, 对照 {sorted(PICK[tgt_pool])[-1]}") r5 = subprocess.run([PY, 'update_daily.py', DATE], cwd='/Users/xpresso/zt_app/code', env={**os.environ, 'ZT_BASE': TEST4, 'ZT_NO_FAC': '1'}, capture_output=True, text=True) print(" ", [l for l in r5.stdout.split('\n') if '汇总' in l or '除权守卫' in l]) rows_t = pd.read_csv(f4, dtype={'trade_date': str}) rows_c = pd.read_csv(ctrl4, dtype={'trade_date': str}) hit = len(rows_t[rows_t['trade_date'] == DATE]) == 0 ctl = len(rows_c[rows_c['trade_date'] == DATE]) == 1 print(f" 被拦股写入新行? {'是 ❌' if not hit else '否 ✅(拒写)'} | " f"正常股写入新行? {'是 ✅' if ctl else '否 ❌'}") # ⑧ 判据自检: B-only 定稿用例(含大比例分红 000338.SZ 20241018 —— A∧B 会漏的那种) r6 = subprocess.run([PY, 'exdiv_judge.py'], cwd='/Users/xpresso/zt_app/code', capture_output=True, text=True) print("\n⑧ 判据自检 exdiv_judge.py:", r6.stdout.strip().split('\n')[-1]) print(" 全部通过?", '是 ✅' if '全部通过' in r6.stdout else '否 ❌')