#!/usr/bin/env python3
"""键集合差集定位（Mac 侧）：本机「当前数据重跑」(5786) vs 「重建版 88438d5」(5776) vs 「记录版 88438d5^」(5837)

输出:
  --keys-out  本次重跑的全部键 (code,pool,buy_date,tb) —— 供对端逐键 diff
  --diff-out  差集明细: 归属(only_rerun / only_rebuild / only_record) + 该键的约束余量(脆弱度)
  --report    摘要 md

用法: python chk_keyset_diff.py --keys-out /tmp/keys_mac_20260914.csv --diff-out /tmp/keyset_diff.csv --report /tmp/keyset_diff.md
"""
import argparse, io, os, subprocess, sys
import numpy as np
import pandas as pd

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
import grid_or as g
import ideal_engine as ie
from scan_daily_tb import COMBO, SELL_COMBO
from chk_key_fragility import margins

APP = '/Users/xpresso/zt_app'


def baseline_keys(ref, cut='20260911'):
    r = subprocess.run(['git', '-C', APP, 'show', f'{ref}:backtest_zt_full/strength_baseline.csv'], capture_output=True)
    df = pd.read_csv(io.BytesIO(r.stdout), dtype={'buy_date': str})
    df = df[df['buy_date'] <= cut]
    return set(zip(df['code'], df['buy_date']))


def main():
    ap = argparse.ArgumentParser()
    ap.add_argument('--keys-out', default='/tmp/keys_mac_20260914.csv')
    ap.add_argument('--diff-out', default='/tmp/keyset_diff.csv')
    ap.add_argument('--report', default='/tmp/keyset_diff.md')
    a = ap.parse_args()
    rows = []
    for pool in COMBO:
        bp = COMBO[pool]; sc = SELL_COMBO[pool]
        members, dailies = g.load_pool(pool)
        sig = g.build_sig(dailies, bp)
        trades = ie.ideal_backtest(pool, members, dailies, sig, sc['profit'], sc['dd'],
                                   mode=sc['mode'], stop=sc['stop'], hold=sc['hold'], start='20240201')
        for t in trades:
            d = dailies[t['code']].sort_values('trade_date').reset_index(drop=True)
            t0i, tai, tbi = chain_for_buy_safe(d, t['code'], bp, t['buy_date'])
            if t0i is None:
                continue
            m = margins(dailies[t['code']], t['code'], bp, t['buy_date']) or {}
            rows.append({'code': t['code'], 'pool': pool, 'buy_date': str(t['buy_date']),
                         'tb': str(d['trade_date'].iloc[tbi]),
                         'm_t0': m.get('m_t0'), 'm_tb': m.get('m_tb'), 'm_ta': m.get('m_ta'),
                         'm_vol': m.get('m_vol'), 'm_min': m.get('m_min')})
        print(f'{pool}: {len(rows)} 键', flush=True)
    df = pd.DataFrame(rows)
    df.to_csv(a.keys_out, index=False)
    now = set(zip(df['code'], df['buy_date']))
    reb = baseline_keys('88438d5')
    rec = baseline_keys('88438d5^')
    print(f'重跑 {len(now)} | 重建版 {len(reb)} | 记录版 {len(rec)}')
    only_r = sorted(now - reb)
    only_b = sorted(reb - now)
    only_rec = sorted(rec - now)
    det = df.set_index(['code', 'buy_date'])
    out = []
    for tag, ks in [('only_rerun(↔重建版)', only_r), ('only_rebuild', only_b), ('only_record(↔记录版)', only_rec)]:
        for c, bd in ks:
            r = {'归属': tag, 'code': c, 'buy_date': bd}
            if (c, bd) in det.index:
                r.update({k: v for k, v in det.loc[(c, bd)].to_dict().items() if k != 'code'})
            out.append(r)
    od = pd.DataFrame(out)
    od.to_csv(a.diff_out, index=False)
    L = ['# 键集合差集定位（Mac 侧，本机复跑）\n']
    L.append(f'- 本机「当前(修准后)数据重跑」: **{len(now)} 键**（与对端 5786 同数）；重建版 88438d5: {len(reb)}；记录版 88438d5^: {len(rec)}')
    L.append(f'- 重跑 − 重建版 = **{len(only_r)} 键**（仅重跑有；= 对端所说"11 键差"）；重建版 − 重跑 = **{len(only_b)} 键**（仅重建版有）')
    L.append(f'- 重跑 − 记录版 = {len(only_rec)} 键（对端口径 264+213 的另一半，此处仅列 %d 行）\n' % len(only_rec))
    L.append('## 12 个待归因键（11 仅重跑有 + 1 仅重建版有）\n')
    sub = od[od['归属'] != 'only_record(↔记录版)']
    L.append(sub.to_string(index=False))
    L.append('\n## 归因线索（本端）\n')
    L.append('- `m_t0/m_tb/m_ta/m_vol/m_min` = 该键各约束的归一化余量（越小越易因数据微动而增减键）')
    L.append('- 若 11 键的 m_min 明显大于 1 键的，说明两侧不同因（前者=修准新增信号、后者=修准抹掉的边缘信号）')
    L.append(f'- 键级余量明细全表: `{a.diff_out}`；本端全量键: `{a.keys_out}`（供对端逐键 diff）')
    open(a.report, 'w').write('\n'.join(L) + '\n')
    print('\n'.join(L))
    print(f'\n报告: {a.report}')


def chain_for_buy_safe(d, code, bp, bd):
    from plot_ideal_top10 import chain_for_buy
    try:
        return chain_for_buy(d, code, bp, str(bd))
    except Exception:
        return (None, None, None)


if __name__ == '__main__':
    main()