#!/usr/bin/env python3 """fix_daily_from_tushare.py — 用 tushare pro.daily 原值修准库内派生的"前收/涨跌"字段。 背景(2026-09-13): 库内 `raw_*` 是真 tushare 原值(逐值吻合), 但 `pre_close/change/pct_chg` 是早期"推算"残留 —— 全库 6,991,185 可判行里 **343,167 行(4.91%) 的 pre_close 与 `pct_chg` 不自洽**(既有 2dp 舍入, 也有整段尺度错: 例 300394.SZ 20210402 store 7.91 vs tushare 原值 41.38; 000001.SZ 20210105 store 14.86 vs 19.34)。 **不要改成"由 adj_factor 反推 pre_close"**: 因子只存 6dp, 反推会引入 0.005~0.009 元(≤0.1%) 的舍入误差, 且上市首日类行会错(全库 779 行反推不自洽)。正确做法 = 拉回来对齐原值。 本工具: 每只股票一次 `pro.daily(ts_code)` 全区间拉取 → 按 trade_date 对齐 → 逐字段比对; - pre_close/change/pct_chg/vol/amount: 超容差即用 tushare 原值覆盖(容差 0.005 / 1e-4 / 1 / 1) - raw_open/high/low/close: 差异 >1e-6 时**先记为 RAW 异常**(计数+明细), 同样以 tushare 为准覆盖 - ts_code/trade_date/adj_factor 不动(adj_factor 走独立因子表链路) 安全阀: API 返回 0 行 / 行数与库内不一致 / 缺 trade_date → **跳过该文件不写并计数**; 写盘前备份到 `_preclose_repair_backup/`(每池一份, 首次运行时), 变更明细落 `code/_repair_manifest_<日期>.csv`(代码,日期,字段,旧值,新值) 以便审计与回滚核对。 用法: python code/fix_daily_from_tushare.py --dry --limit 20 # 抽样试跑(不写盘) python code/fix_daily_from_tushare.py --dry # 全库试跑(不写盘) python code/fix_daily_from_tushare.py # 全库写入 --pools hs300,zz500 限定池 --codes 000001.SZ,... 限定股票 --no-backup 跳过备份(默认备份到 _preclose_repair_backup//) """ import os, sys, glob, time, shutil, datetime import numpy as np import pandas as pd import tushare as ts CODE_DIR = os.path.dirname(os.path.abspath(__file__)) BASE = os.environ.get('ZT_BASE', '/Users/xpresso/zt_app/backtest_zt_full') BAK = os.path.join(os.path.dirname(BASE), '_preclose_repair_backup') POOLS = ['hs300', 'zz500', 'zz1000', 'zz2000'] DRY = '--dry' in sys.argv NOBACKUP = '--no-backup' in sys.argv LIMIT = int(sys.argv[sys.argv.index('--limit') + 1]) if '--limit' in sys.argv else None if '--pools' in sys.argv: POOLS = sys.argv[sys.argv.index('--pools') + 1].split(',') CODES = set(sys.argv[sys.argv.index('--codes') + 1].split(',')) if '--codes' in sys.argv else None STAMP = datetime.datetime.now().strftime('%Y%m%d_%H%M') MANIFEST = f'{CODE_DIR}/_repair_manifest_{STAMP}.csv' # 字段容差: 小于此差异视为一致(不做无意义写入) TOL = {'raw_open': 1e-6, 'raw_high': 1e-6, 'raw_low': 1e-6, 'raw_close': 1e-6, 'pre_close': 0.005, 'change': 0.005, 'pct_chg': 1e-4, 'vol': 1.0, 'amount': 1.0} if '--pct-tol' in sys.argv: # 收紧 pct_chg 容差 → 连 ≤1e-4 的残留水位(0.73 vs 0.7299)也抹平 TOL['pct_chg'] = float(sys.argv[sys.argv.index('--pct-tol') + 1]) API_MAP = {'raw_open': 'open', 'raw_high': 'high', 'raw_low': 'low', 'raw_close': 'close', 'pre_close': 'pre_close', 'change': 'change', 'pct_chg': 'pct_chg', 'vol': 'vol', 'amount': 'amount'} pro = ts.pro_api() files = [] for p in POOLS: for f in sorted(glob.glob(f'{BASE}/daily_{p}/*.csv')): if CODES is None or os.path.basename(f)[:-4] in CODES: files.append((p, f)) if LIMIT: files = files[:LIMIT] print(f'待处理 {len(files)} 个文件 | 池 {POOLS} | DRY={DRY} | 备份={"否" if NOBACKUP else BAK}') manifest = [] stat = dict(files=0, changed=0, skipped=0, rows=0, rounding_only=0, patch={k: 0 for k in TOL}, raw_mismatch=0, err=0) raw_mismatch_detail = [] skip_detail = [] def fetch(code, s, e, tries=4): for i in range(tries): try: r = pro.daily(ts_code=code, start_date=s, end_date=e) return r except Exception as ex: if i == tries - 1: raise time.sleep(2 ** i) t0 = time.time() for n, (pool, fpath) in enumerate(files, 1): code = os.path.basename(fpath)[:-4] try: d = pd.read_csv(fpath, dtype={'trade_date': str}).sort_values('trade_date').reset_index(drop=True) if not len(d): stat['skipped'] += 1; skip_detail.append((code, '空文件')); continue api = fetch(code, d['trade_date'].min(), d['trade_date'].max()) if api is None or not len(api): stat['skipped'] += 1; skip_detail.append((code, 'API 0 行')); continue api = api.drop_duplicates('trade_date').set_index('trade_date') m = d['trade_date'].isin(api.index) if int(m.sum()) != len(d): stat['skipped'] += 1 skip_detail.append((code, f'交易日不齐 库{len(d)}/命中{int(m.sum())}/API{len(api)}')) continue new = d.copy() dirty = False for col, api_col in API_MAP.items(): if col not in new.columns or api_col not in api.columns: continue ref = api.loc[d['trade_date'].values, api_col].values.astype(float) cur = new[col].values.astype(float) diff = np.abs(cur - ref) mask = np.logical_and(np.isfinite(ref), np.logical_not(diff < TOL[col])) k = int(mask.sum()) if k: stat['patch'][col] += k dirty = True if col.startswith('raw_'): stat['raw_mismatch'] += k for i in np.where(mask)[0][:3]: raw_mismatch_detail.append((code, d['trade_date'].values[i], col, float(cur[i]), float(ref[i]))) for i in np.where(mask)[0]: # 明细只记"实质差异"(超出 2dp 舍入): pct_chg 由 2dp 补齐到 tushare 原值(4dp) # 的行数以 1e-4 计, 但明细只留 >0.005pp 的实质错值, 避免 manifest 爆到百万行 old_v, new_v = float(cur[i]), float(ref[i]) if col.startswith('raw_') or abs(old_v - new_v) > (0.01 if col == 'pct_chg' else 0.005): manifest.append((code, d['trade_date'].values[i], col, old_v, new_v)) else: stat['rounding_only'] += 1 new.loc[mask, col] = ref[mask] stat['files'] += 1; stat['rows'] += len(d) if dirty: stat['changed'] += 1 if not DRY: if not NOBACKUP: bdir = f'{BAK}/{pool}' os.makedirs(bdir, exist_ok=True) bdst = f'{bdir}/{code}.csv' if not os.path.exists(bdst): shutil.copy2(fpath, bdst) new.to_csv(fpath, index=False) except Exception as ex: stat['err'] += 1 skip_detail.append((code, f'异常 {type(ex).__name__}: {ex}')) if n % 200 == 0: print(f' {n}/{len(files)} | 耗时 {time.time()-t0:.0f}s | 变更文件 {stat["changed"]} | 跳过 {stat["skipped"]} | 异常 {stat["err"]}') print('\n=== 汇总 ===') print(f'处理 {stat["files"]} 个文件 / {stat["rows"]} 行 | 需改文件 {stat["changed"]} | 跳过 {stat["skipped"]} | 异常 {stat["err"]}') print('按字段需改行数:', {k: v for k, v in stat['patch'].items() if v}) print(f'其中仅 2dp 舍入补齐(非实质错值) {stat["rounding_only"]} 行; 实质错值明细行数 {len(manifest)}') print(f'raw_* 与 tushare 不一致行数(应极少): {stat["raw_mismatch"]} 例: {raw_mismatch_detail[:5]}') if skip_detail: print(f'跳过明细(前 10): {skip_detail[:10]}') if manifest and not DRY: pd.DataFrame(manifest, columns=['code', 'trade_date', 'field', 'old', 'new']).to_csv(MANIFEST, index=False) print(f'变更明细已写: {MANIFEST} ({len(manifest)} 行)') elif manifest: print(f'[DRY] 变更明细 {len(manifest)} 行(未写盘)') print('DRY 试跑, 未写盘' if DRY else f'完成, 备份: {"无" if NOBACKUP else BAK}')