# -*- coding: utf-8 -*- """逐笔历史修复的离线编译器。只有完整证据才生成写回计划,不猜基线。""" from collections import defaultdict from copy import deepcopy from datetime import datetime, timezone, timedelta from decimal import Decimal, ROUND_HALF_UP import hashlib import json from app.core.cushion import sell_allocation, cushion_state D = lambda x: Decimal(str(x or 0)) TERMINAL = {'FILLED', 'CANCELLED', 'EXPIRED', 'REJECTED', 'ABORTED', 'SEND_FAILED'} def canonical(value): return json.dumps(value, ensure_ascii=False, sort_keys=True, separators=(',', ':'), default=str) def digest(value): return hashlib.sha256(canonical(value).encode()).hexdigest() def rounded(value, places=3): return str(D(value).quantize(Decimal(10) ** -places, rounding=ROUND_HALF_UP)) def net_lot_cost(lots): """核销均价只有三位小数;盈亏金额保留了逐笔核销金额,不能反乘均价丢精度。""" return sum((D(l['qty']) * D(l['open_price']) - (D(l['realized_pnl']) if l.get('realized_pnl') is not None else D(l.get('closed_qty')) * (D(l.get('close_avg_price')) - D(l['open_price'])))) for l in lots) def need(condition, message): if not condition: raise ValueError(message) def compile_repair(backup, codes, *, scale=2000000, solid=0.03): """返回原行、目标列、逐笔分配和不能还原事项。输入、配置相同则结果完全一致。""" need(backup.get('instance') == '155-sim', '仅接受已确认的 155 模拟仓备份') need(codes and len(set(codes)) == len(codes), '股票范围为空或重复') tables = backup['tables'] run_id = 'HIST_' + digest({'backup': backup, 'codes': sorted(codes), 'scale': scale, 'solid': solid})[:24] out = {'run_id': run_id, 'backup_digest': digest(backup), 'codes': sorted(codes), 'changes': [], 'summaries': [], 'allocations': [], 'daily': {}, 'unknowns': [ '手续费原文为空时无法恢复实际净盈亏,以下盈亏均为不含费用的批次盈亏。', '缺少完整盘中行情及当时账本,安全垫历史峰值保留原值,不声明已还原。', '历史日报与全组合净值保留原始记录,受污染期间不得作为已修复的真实绩效。']} snapshots = {x['ts_code']: x for x in backup['position_snapshot']['payload']['items']} all_trades = [] for r in tables['pms_qmt_inbox']: p = json.loads(r['payload_json']) if p.get('ts_code') in codes: all_trades.append(dict(r, payload=p)) by_no = set() for r in all_trades: p = r['payload'] need(p.get('trade_no') and p['trade_no'] not in by_no, '重复或缺失成交号') by_no.add(p['trade_no']) def change(table, row, fields): out['changes'].append({'table': table, 'key': {'id': row['id']}, 'before': row, 'fields': fields}) for code in sorted(codes): positions = [x for x in tables['pms_position'] if x['ts_code'] == code] need(len(positions) == 1 and code in snapshots, f'{code} 持仓事实缺失') pos, fact = positions[0], snapshots[code] lots = [x for x in tables['pms_lot'] if x['ts_code'] == code] need(lots, f'{code} 无可追溯批次基线') start = min(str(l['open_date']) for l in lots) orders = {x['instruction_id']: x for x in tables['pms_qmt_order'] if x['ts_code'] == code} instructions = {x['instruction_id']: x for x in tables['pms_instruction'] if x['ts_code'] == code} trades = sorted([x for x in all_trades if x['payload']['ts_code'] == code], key=lambda x: (int(x['payload']['trade_time']), int(x['seq']))) before_qty = 0 current = [] for r in trades: p = r['payload'] day = datetime.fromtimestamp(int(p['trade_time']) / 1000, timezone(timedelta(hours=8))).date().isoformat() if day < start: before_qty += int(p['qty']) * (1 if p['side'] == 'buy' else -1) else: current.append(r) need(before_qty == 0, f'{code} 起点前未平仓,不能假设零基线') need(current, f'{code} 缺少逐笔历史') parents = defaultdict(list) for iid, o in orders.items(): if str(o['created_at'])[:10] < start: continue need(o['status'] in TERMINAL, f'{iid} 仍有在途委托') ts = [x for x in current if x['payload']['instruction_id'] == iid] need(sum(int(x['payload']['qty']) for x in ts) == int(o['cum_qty']), f'{iid} 漏笔或累计量冲突') for r in current: p = r['payload']; o = orders.get(p['instruction_id']) need(o and o['parent_id'] in instructions, f"{p['trade_no']} 缺出口或父指令") need(o['side'] == p['side'] and o['broker_order_id'] == p['broker_order_id'], '成交身份冲突') need(int(r['processed']) == 1 and int(p['qty']) > 0 and D(p['price']) > 0, '未完成原始入账或坏成交') need(D(p.get('amount')) == int(p['qty']) * D(p['price']), '逐笔金额不一致') parents[o['parent_id']].append(r) fills = [x for x in backup['table_fills'] if x['ts_code'] == code and x['done_time'][:10] >= start] for f in fills: owners = [o for o in orders.values() if o['instruction_id'] == f['order_id'] or (f.get('raw', {}).get('qmt_order_id') and o['broker_order_id'] == f['raw']['qmt_order_id'])] need(len(owners) == 1 and owners[0]['side'] == f['side'], '存在无法解释的外部成交') used, state, matched = set(), {}, {} for r in current: p = r['payload']; o = orders[p['instruction_id']] if p['side'] != 'buy': continue matches = [l for l in lots if l['instruction_id'] == o['parent_id'] and int(l['qty']) + int(l['closed_qty']) == int(p['qty']) and D(l['open_price']) == D(p['price']) and str(l['created_at']) == str(r['processed_at'])] need(len(matches) == 1 and matches[0]['id'] not in used, f"{p['trade_no']} 原批次无法唯一对应") l = matches[0]; used.add(l['id']); matched[r['seq']] = l['id'] state[l['id']] = dict(l, qty=0, closed_qty=0, close_avg_price=None, realized_pnl=Decimal(0), sell_amount=Decimal(0)) groups = defaultdict(list); void = [] for l in lots: if l['id'] in used: continue void.append(l['id']) if l['lot_type'] != 'RECON': groups[(str(l['created_at']), str(l['open_price']))].append(l) matched_fills = set() for (created, price), group in groups.items(): qty = sum(int(l['qty']) + int(l['closed_qty']) for l in group) fs = [f for f in fills if f['side'] == 'buy' and f['qty'] == qty and D(f['price']) == D(price) and f['done_time'][:10] == created[:10] and all( not l['instruction_id'] or l['instruction_id'] == orders[f['order_id']]['parent_id'] for l in group)] need(len(fs) == 1 and fs[0]['order_id'] not in matched_fills, '非真实买入批次缺重复汇总单证据') matched_fills.add(fs[0]['order_id']) daily = {}; buy_cash = D(0); sell_cash = D(0) for r in current: p = r['payload']; qty = int(p['qty']); px = D(p['price']) if p['side'] == 'buy': state[matched[r['seq']]]['qty'] = qty buy_cash += qty * px else: active = sorted((l for l in state.values() if l['qty'] > 0), key=lambda l: (str(l['open_date']), l['id'])) allocations = sell_allocation([dict(lot_id=l['id'], lot_type=l['lot_type'], qty=l['qty'], open_date=int(str(l['open_date']).replace('-', ''))) for l in active], qty) for a in allocations: l = state[a['lot_id']]; q = a['qty'] l['qty'] -= q; l['closed_qty'] += q; l['sell_amount'] += q * px l['realized_pnl'] += q * (px - D(l['open_price'])) out['allocations'].append(dict(ts_code=code, seq=r['seq'], trade_no=p['trade_no'], lot_id=l['id'], qty=q, price=str(px), realized_pnl=str(q * (px - D(l['open_price']))))) sell_cash += qty * px day = str(r['received_at'])[:10] q = sum(l['qty'] for l in state.values()) daily[day] = {'qty': q, 'buy_amount': str(buy_cash), 'sell_amount': str(sell_cash), 'realized_pnl': str(sum(l['realized_pnl'] for l in state.values())), 'avg_cost': rounded((buy_cash-sell_cash)/q) if q else None} total = sum(l['qty'] for l in state.values()) need(total == int(fact['total_qty']), f'{code} 重建数量与 QMT 不符') need(net_lot_cost(list(state.values())) == buy_cash - sell_cash, '成本与逐笔资金不守恒') pnl = sum(l['realized_pnl'] for l in state.values()) for old in lots: l = state.get(old['id']) fields = {'qty': 0, 'closed_qty': 0, 'close_avg_price': None, 'realized_pnl': '0.00', 'status': 'CLOSED', 'note': f'{run_id} 原重复/RECON批次作废;原值见备份'} if l: fields.update(qty=l['qty'], closed_qty=l['closed_qty'], close_avg_price=rounded(l['sell_amount']/l['closed_qty']) if l['closed_qty'] else None, realized_pnl=rounded(l['realized_pnl'], 2), status='OPEN' if l['qty'] else 'CLOSED', note=f'{run_id} 按逐笔重建;原值见备份') change('pms_lot', old, fields) by_type = defaultdict(int) for l in state.values(): by_type[l['lot_type']] += l['qty'] avg = max(D(0), (buy_cash - sell_cash)/total) if total else None px = D(fact['market_price']); cp = px/avg-1 if avg else None # 保留原有 T+1 限制:当日买入不能因模拟 QMT 可用数过宽而全部放开。 today = backup['at'][:10] day_open = daily[max((d for d in daily if d < today), default=min(daily))]['qty'] if min(daily) < today else 0 today_sell = sum(int(x['payload']['qty']) for x in current if x['received_at'][:10] == today and x['payload']['side']=='sell') avail = min(int(fact['avail_qty']), max(0, day_open-today_sell), total) fields = dict(total_qty=total, avail_qty=avail, base_qty=by_type['BASE']+by_type['RECON'], add_qty=by_type['ADD'], dca_qty=by_type['DCA'], fill_qty=by_type['FILL'], t0_qty=by_type['T0'], avg_cost=rounded(avg) if avg else None, realized_t_profit=rounded(sum(l['realized_pnl'] for l in state.values() if l['lot_type']=='T0'),2), cushion_pct=rounded(cp,4) if cp is not None else None, cushion_state=cushion_state(float(cp) if cp is not None else None,solid), pct_of_scale=rounded(total*px/D(scale),4) if scale else None, status=('HOLDING' if pos['status']=='CLOSED' and total else pos['status'])) change('pms_position', pos, fields) progress_rows = [] for pid, ts in parents.items(): ins = instructions[pid]; qty = sum(x['payload']['qty'] for x in ts) amount = sum(D(x['payload']['amount']) for x in ts) progress = json.loads(ins['progress_json'] or '{}') evidence = {'run_id': run_id, 'old_exec_qty': ins['exec_qty'], 'old_status': ins['status'], 'exec_qty': qty, 'exec_amount': str(amount), 'remaining_qty': max(0, ins['qty']-qty), 'overfill_qty': max(0,qty-ins['qty']), 'seqs': [x['seq'] for x in ts]} status = ins['status'] if status == 'CONFIRMED' and qty < ins['qty']: status = 'CANCELLED' evidence['hold_reason'] = '历史修复暂停剩余执行;子单均已终态,不自动续单' progress['historical_repair'] = evidence for child in progress.get('children', []): child_ts = [x for x in ts if x['payload']['instruction_id'] == child.get('ref')] child['actual_exec_qty'] = sum(x['payload']['qty'] for x in child_ts) child['actual_exec_amount'] = str(sum(D(x['payload']['amount']) for x in child_ts)) change('pms_instruction', ins, dict(exec_qty=qty, exec_avg_price=rounded(amount/qty), status=status, progress_json=canonical(progress))) progress_rows.append(dict(instruction_id=pid, **evidence, new_status=status)) need(not [x for x in tables['pms_plan'] if x['ts_code']==code], '命令方案须另作完整重建,当前工具不猜命令进度') # 活跃网格账的剩余档位来自同一批真实批次,已撤销策略不恢复。 for strategy in tables['pms_strategy']: if strategy['ts_code'] != code or strategy['status'] != 'ACTIVE': continue need(strategy['type']=='GRID', '活跃非网格策略需要专门核验') st = json.loads(strategy['state_json'] or '{}') need(not st.get('pending'), '网格策略仍有待结算指令') levels = defaultdict(lambda: [0,D(0)]) for l in state.values(): ins = instructions[l['instruction_id']] if ins.get('origin_id') != strategy['strategy_id'] or not l['qty']: continue prog = json.loads(ins['progress_json'] or '{}'); leg = prog.get('leg','') need(leg.startswith('grid_buy:'), '策略批次缺网格档位来源') level = leg.split(':')[1]; levels[level][0] += l['qty']; levels[level][1] += l['qty']*D(l['open_price']) st['filled_levels'] = {k:dict(qty=v[0],price=float(v[1]/v[0])) for k,v in levels.items()} st['invested'] = float(sum(v[1] for v in levels.values())) st['historical_repair'] = {'run_id':run_id,'source':'真实剩余批次'} change('pms_strategy',strategy,dict(state_json=canonical(st))) out['daily'][code] = daily out['summaries'].append(dict(ts_code=code,start=start,trades=len(current),real_lots=len(used), void_lots=void,old_qty=pos['total_qty'],qty=total,avg_cost=fields['avg_cost'],avail_qty=avail, old_avg_cost=pos['avg_cost'],realized_pnl=rounded(pnl,2),buy_amount=str(buy_cash),sell_amount=str(sell_cash), instructions=progress_rows)) return json.loads(canonical(out))