#!/usr/bin/env python3
"""Mac 实操侧 price-schema-v1 迁移验收: ①文件数/异常数 ②无损性最大偏差"""
import glob, os, json
import pandas as pd

BASE = os.path.expanduser('~/zt_app/backtest_zt_full')
COLS = ['raw_open','raw_high','raw_low','raw_close','adj_factor',
        'pre_close','change','pct_chg','vol','amount','ts_code','trade_date']
COLS_SET = set(COLS)
POOLS = ['hs300','zz500','zz1000','zz2000']

# ---------- ① 列头审计 ----------
from collections import Counter
dist = Counter(); bad = []; per_pool = {}; order_bad = []
for p in POOLS:
    files = sorted(glob.glob(f'{BASE}/daily_{p}/*.csv'))
    per_pool[p] = len(files)
    for f in files:
        with open(f) as fh:
            cols = fh.readline().strip().split(',')
        dist[len(cols)] += 1
        if set(cols) != COLS_SET or cols != COLS:
            if set(cols) != COLS_SET:
                bad.append((p, os.path.basename(f), len(cols), cols))
            else:
                order_bad.append((p, os.path.basename(f), cols))
print('① 列头审计')
print('   每池文件数:', per_pool, '合计', sum(per_pool.values()))
print('   列数分布:', dict(dist))
print('   列集合不符:', len(bad), bad[:5])
print('   列集合符但顺序不符:', len(order_bad), order_bad[:3])

# ---------- ② 无损性: 旧落盘 hfq_close/qfq_close vs raw×factor ----------
BAK = os.path.expanduser('~/zt_app/_hfq_ohlc_backup_20260912')
baks = sorted(glob.glob(f'{BAK}/*.csv'))
cur = {}
for p in POOLS:
    for f in glob.glob(f'{BASE}/daily_{p}/*.csv'):
        cur[os.path.basename(f)[:-4]] = (p, f)
print(f'\n② 无损性: 备份 {len(baks)} 文件, 当前库 {len(cur)} 文件')
hit = 0; miss = []
maxabs_hfq = 0.0; maxabs_hfq_where = None
maxrel_qfq = 0.0; maxrel_qfq_where = None
maxabs_raw = 0.0; maxabs_raw_where = None
nrows = 0; exdiv_files = 0
for bf in baks:
    code = os.path.basename(bf)[:-4]
    if code not in cur:
        miss.append(code); continue
    d = pd.read_csv(bf, dtype={'trade_date': str})
    if 'hfq_close' not in d.columns or 'raw_close' not in d.columns:
        miss.append(code); continue
    c = pd.read_csv(cur[code][1], dtype={'trade_date': str})
    m = d.merge(c, on='trade_date', suffixes=('_old','_cur'))
    if not len(m):
        miss.append(code); continue
    hit += 1; nrows += len(m)
    rc = m['raw_close_cur'].astype(float); fac = m['adj_factor'].astype(float)
    # raw 不变
    dr = (m['raw_close_old'].astype(float) - rc).abs()
    i = dr.idxmax()
    if dr.max() > maxabs_raw:
        maxabs_raw = dr.max(); maxabs_raw_where = (code, m.loc[i,'trade_date'])
    # hfq 无损: 旧落盘 vs raw×factor
    dh = (m['hfq_close'].astype(float) - rc*fac).abs()
    i = dh.idxmax()
    if dh.max() > maxabs_hfq:
        maxabs_hfq = dh.max(); maxabs_hfq_where = (code, m.loc[i,'trade_date'],
            float(m.loc[i,'hfq_close']), float(rc[i]*fac[i]))
    # qfq 无损: 旧落盘 vs raw×factor/末行factor
    if 'qfq_close' in m.columns:
        latest = fac.iloc[-1]
        dq = (m['qfq_close'].astype(float) - rc*fac/latest).abs()
        rel = dq / m['qfq_close'].astype(float).abs().clip(lower=1e-9)
        i = rel.idxmax()
        if rel.max() > maxrel_qfq:
            maxrel_qfq = rel.max(); maxrel_qfq_where = (code, m.loc[i,'trade_date'],
                float(m.loc[i,'qfq_close']), float(rc[i]*fac[i]/latest))
print(f'   可比对文件: {hit}, 逐行 {nrows}, 未命中: {len(miss)} {miss[:6]}')
print(f'   raw_close 最大绝对偏差: {maxabs_raw:.3e} @ {maxabs_raw_where}')
print(f'   旧落盘 hfq_close vs raw×factor 最大绝对偏差: {maxabs_hfq:.6e} @ {maxabs_hfq_where}')
print(f'   旧落盘 qfq_close vs raw×factor/末行factor 最大相对偏差: {maxrel_qfq:.6e} @ {maxrel_qfq_where}')