From 0871885801794161ff39e768bcba5f6b3582b709 Mon Sep 17 00:00:00 2001 From: baobao Date: Thu, 15 Oct 2026 12:34:06 +0800 Subject: [PATCH] =?UTF-8?q?=E5=8E=86=E5=8F=B2=E8=B4=A6=E6=9C=AC=EF=BC=9A?= =?UTF-8?q?=E6=8C=89=E9=80=90=E7=AC=94=E6=81=A2=E5=A4=8D=E6=95=B0=E9=87=8F?= =?UTF-8?q?=E6=88=90=E6=9C=AC=E7=9B=88=E4=BA=8F=E4=B8=8E=E6=89=A7=E8=A1=8C?= =?UTF-8?q?=E8=BF=9B=E5=BA=A6=EF=BC=8C=E5=A4=87=E4=BB=BD=E5=AE=A1=E8=AE=A1?= =?UTF-8?q?=E5=8F=8A=E4=BA=8B=E5=8A=A1=E5=9B=9E=E9=80=80=EF=BC=88=E5=8F=B0?= =?UTF-8?q?=E8=B4=A6=20020=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 一、双来源历史重复污染了三票的批次、成本与父指令进度。 完整备份的 134 笔成交重建为 700、1900、1400 股；只按数量对账无法恢复盈亏。 二、真实买入唯一对应原批次，卖出按既有批次优先级重算。 33 个虚假或 RECON 批次归零留痕，61 个真实批次恢复；实际超目标成交保留。 误判已完成指令的剩余执行暂停，不自动复活旧单；活跃网格恢复真实余量。 三、写回经过备份哈希、并发原行核对和单表 SQL 事务，逐行审计并支持条件回退。 21 个新用例与完整 44 批隔离禁网回归通过。未改线上参数、原始成交、游标或历史净值； 手续费与历史峰值缺证据的部分明确保留边界。生产判收另记。 --- DEVLOG.md | 7 + app/core/ledger_repair.py | 236 +++++++++++++++++++++++++ app/db/session.py | 55 ++++++ app/services/ledger_service.py | 9 +- docs/PMS三项对账差异排查_2026-10-15.md | 15 ++ docs/复盘决定台账.md | 11 ++ scripts/repair_ledger_history.py | 95 ++++++++++ scripts/run_tests.py | 3 +- scripts/test_batch40_units.py | 164 +++++++++++++++++ 9 files changed, 589 insertions(+), 6 deletions(-) create mode 100644 app/core/ledger_repair.py create mode 100644 scripts/repair_ledger_history.py create mode 100644 scripts/test_batch40_units.py diff --git a/DEVLOG.md b/DEVLOG.md index 35736dc..20e459a 100644 --- a/DEVLOG.md +++ b/DEVLOG.md @@ -1857,3 +1857,10 @@ docs/交接_2026-09-20.md（新）、docs/dev_preview/mock_preview.py 与 sample 12:02:54 四服务重建，镜像 c17af17aeb09d151bbe932395b879add1ed336b16739703f9acc594bee30b3f1，Python 指纹 cc988cd762bf。 12:04:59 健康正常、QMT ONLINE、待入账及在途出口委托均 0，业务参数保持。午休无自然新成交，逐笔实际回放效果待事件判收。 三项历史数量差异仍在，未强制对账或回写成本与指令。完整证据与回退信息见同日每日计划交接末节。 + + +## 2026-10-15 · 三票历史逐笔修复工具与预演 + +用户已授权备份后恢复数量、成本、盈亏和执行进度。实现 app/core/ledger_repair.py、scripts/repair_ledger_history.py 和 db.session 的受控事务入口；原行比较、全部逐笔与运行参数守卫、行级审计、条件回退均在同一事务边界内。ledger_service 的成本重算改用剩余成本减已实现金额，保持相同摊薄口径并消除三位均价反乘误差。 + +本地配置隔离且禁网：test_batch40 的 21 例通过，全量 44 批通过。生产备份和完整预判见 docs/PMS三项对账差异排查_2026-10-15.md；真实读数仅作离线预演，尚未写账，部署和判收后续追加。原未提交台账 017、讨论稿和探针未纳入本提交。 diff --git a/app/core/ledger_repair.py b/app/core/ledger_repair.py new file mode 100644 index 0000000..a375468 --- /dev/null +++ b/app/core/ledger_repair.py @@ -0,0 +1,236 @@ +# -*- 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)) diff --git a/app/db/session.py b/app/db/session.py index 77f76db..4baad3e 100644 --- a/app/db/session.py +++ b/app/db/session.py @@ -236,3 +236,58 @@ def execute_many(sql: str, seq_params: list, source: str = "proxy") -> int: except Exception as e: _mark(source, False) raise DBUnavailable(f"批量写入异常({source}): {type(e).__name__}: {e}") from e + + +def apply_audited_repair(plan: dict, guards: list, *, rollback=False) -> dict: + """维护专用：核验原行、逐条单表写入和审计在同一事务。绝不调下单或回放服务。""" + import json + from datetime import datetime + from app.core.ledger_repair import canonical + allowed = { + 'pms_lot': {'qty','closed_qty','close_avg_price','realized_pnl','status','note'}, + 'pms_position': {'total_qty','avail_qty','base_qty','fill_qty','add_qty','dca_qty','t0_qty', + 'avg_cost','realized_t_profit','cushion_pct','cushion_state','pct_of_scale','status'}, + 'pms_instruction': {'exec_qty','exec_avg_price','status','progress_json'}, + 'pms_strategy': {'state_json'}, + } + if TABLE_PREFIX: + raise ValueError('历史修复仅允许模拟仓') + run_id = plan['run_id'] + ('_ROLLBACK' if rollback else '') + _check_cooldown('proxy') + with get_engine('proxy').begin() as c: + def query(sql, params=None): + sql = map_tables(sql) + assert_single_table(sql) + return c.execute(text(sql), params or {}) + done = query("SELECT id FROM pms_op_log WHERE op='history_repair_summary' AND ref=:r FOR UPDATE", + {'r':run_id}).mappings().fetchall() + if done: + return {'ok':True, 'already_applied':True, 'run_id':run_id} + for g in guards: + rows = [dict(r) for r in query(g['sql'],g.get('params')).mappings().fetchall()] + if canonical(rows) != canonical(g['expected']): + raise ValueError('修复前置数据已变化，拒绝写入: '+g['name']) + for item in plan['changes']: + table, fields = item['table'], item['fields'] + if table not in allowed or set(fields) - allowed[table] - {'updated_at'}: + raise ValueError('修复列越界') + current = [dict(r) for r in query(f'SELECT * FROM {table} WHERE id=:id FOR UPDATE',item['key']).mappings().fetchall()] + if len(current) != 1 or canonical(current[0]) != canonical(item['before']): + raise ValueError('目标行已变: '+table+':'+str(item['key'])) + now = datetime.now().replace(microsecond=0) + for item in plan['changes']: + fields = dict(item['fields']) + fields.setdefault('updated_at',now) + query(f"UPDATE {item['table']} SET "+', '.join(f'{k}=:{k}' for k in fields)+' WHERE id=:row_id', + dict(fields,row_id=item['key']['id'])) + after = dict(item['before'],**fields) + query('INSERT INTO pms_op_log (at,op,by_user,ts_code,result,reason,params_json,ref) ' + 'VALUES (:at,:op,:by,:code,:result,:reason,:params,:ref)', + dict(at=now,op='history_repair_row',by='codex',code=after.get('ts_code'),result='OK', + reason='逐笔账本修复'+('回退' if rollback else ''), + params=canonical(dict(item,after=after)),ref=run_id)) + query('INSERT INTO pms_op_log (at,op,by_user,result,reason,params_json,ref) ' + 'VALUES (:at,:op,:by,:result,:reason,:params,:ref)', + dict(at=now,op='history_repair_summary',by='codex',result='OK',reason='历史修复事务完成', + params=canonical({k:v for k,v in plan.items() if k not in ('changes','allocations','daily')}),ref=run_id)) + return {'ok':True,'run_id':run_id,'changed_rows':len(plan['changes'])} diff --git a/app/services/ledger_service.py b/app/services/ledger_service.py index e45cf0d..e6c5f18 100644 --- a/app/services/ledger_service.py +++ b/app/services/ledger_service.py @@ -6,7 +6,7 @@ 摊薄成本口径 (与 core/cushion.PositionCost 等价, 但数据来自批次表): cum_buy = Σ (剩余数量 + 已核销数量) × 开仓价 - cum_sell = Σ 已核销数量 × 核销均价 + cum_sell = Σ 已核销数量 × 开仓价 + 已实现盈亏（金额，避免核销均价舍入误差） avg_cost = max(0, (cum_buy − cum_sell) / 当前持股数) 做T利润通过 T0 批次的买卖流水自然摊入上式, 故 **不再另行扣减** realized_t_profit (该列仅作展示统计, 重复扣减会把成本做低两次)。 @@ -22,6 +22,7 @@ from app.core import command_spec as cs from app.core import cushion as cu from app.core import rebuild_check as rbc from app.core import recon as rc +from app.core.ledger_repair import net_lot_cost from app.core import tradedays as td from app.core import ws_codec as wsc from app.repo import downstream_repo, pms_repo, qmt_repo @@ -627,15 +628,13 @@ def recompute_position(ts_code: str) -> dict: """由批次表重算持仓数量/摊薄成本/安全垫/垫子峰值。""" lots = pms_repo.list_lots(ts_code, status=None, limit=2000) qty = sum(int(l["qty"] or 0) for l in lots) - cum_buy = sum((int(l["qty"] or 0) + int(l["closed_qty"] or 0)) * float(l["open_price"] or 0) - for l in lots) - cum_sell = sum(int(l["closed_qty"] or 0) * float(l["close_avg_price"] or 0) for l in lots) + net_cost = float(net_lot_cost(lots)) by_type = {} for l in lots: if int(l["qty"] or 0) > 0: by_type[l["lot_type"]] = by_type.get(l["lot_type"], 0) + int(l["qty"]) - avg_cost = max(0.0, (cum_buy - cum_sell) / qty) if qty > 0 else None + avg_cost = max(0.0, net_cost / qty) if qty > 0 else None px = market.get_price(ts_code) or avg_cost or 0 cp = (px / avg_cost - 1.0) if (avg_cost and avg_cost > 0 and px) else None solid = param_store.get_float("PMS_CUSHION_SOLID", 0.03) diff --git a/docs/PMS三项对账差异排查_2026-10-15.md b/docs/PMS三项对账差异排查_2026-10-15.md index 68fc695..90a135c 100644 --- a/docs/PMS三项对账差异排查_2026-10-15.md +++ b/docs/PMS三项对账差异排查_2026-10-15.md @@ -45,3 +45,18 @@ WebSocket 逐笔回报均 processed=1，汇总委托全部已进入 PMS_REPLAY_C 当前为午休，修复后自然新成交回放尚待事件验证；三项历史差异仍保留，不报告为清零。 原始证据文件 ~/pms_logs/reconciliation-evidence-20261015-before-fix-v2.json 的 SHA256 为 8e0badf4d771fbf8a3227250c6cc2c2bf3c8234dda2fc33a31f0ab00920f5501。 原始交易持仓表额外出现 562590.SH 与 688825.SH 属另一项事实源覆盖差异，继续按既有 ws_first 取实际持仓，不混同三项账本差异。 + + +## 历史修复开发规格与预判（2026-10-15 12:30，北京时间） + +用户已明确授权先备份，再逐笔恢复数量、成本、盈亏和指令执行进度。修复限定 155 模拟仓 002050.SZ、002979.SZ、688100.SH。188 不在范围内。首份完整备份为主机 pms_logs/historical-repair-backup-20261015-1225.json，SHA256 为 81cfc59600d7ca07de7fc4b83b7c5b153719d0ea248be21fd9d910bcfc32e636。包含全部持仓、批次、指令、方案、命令、出口单、手续费、策略、提议、日报、净值，三票动作账本，完整留存逐笔成交及下游成交表。写前还要重新备份并校验并发变化。 + +实现采用纯函数重建与单表 SQL 事务写回。逐笔成交必须能唯一对应出口子单，代码、方向、委托号、累计数量一致；同成交号冲突、漏笔、卖超、外部成交无法解释、来源读取失败均阻断。每笔真实买入以父指令、数量、价格、原入账时间唯一对应原批次。重复买入批次须能聚合对应同一条下游汇总单，RECON 调整须单列；不能直接按数量差删除。真实卖出按既有批次优先级逐笔重算核销量与盈亏。原批次 ID 保留，虚假批次归零并注明备份与修复批号，不删除原证据。 + +成本使用成交金额或等价的“剩余批次成本减已实现盈亏”，避免三位小数核销均价再乘数量带来的舍入误差。费用不摊进成本；缺失费用保持未知。父指令按真实逐笔量与金额重建 exec_qty/exec_avg_price，同时在 progress 中保存原值、子单证据和剩余量。原误判完成但未足量的指令改为 CANCELLED 并记录“历史修复暂停剩余执行”，前提是所有子单均已终态，绝不恢复自动续单；原在执行的指令保持生命周期。真实超目标成交如 002979 的 1,300 股网格指令成交 2,600 股要如实保留并标异常。方案、命令当前为空，无需伪造。 + +完整性边界按当前留存批次的起始日验证，之前成交净数量必须为零。002050 从 09-22、002979 从 09-21、688100 从 10-14 重建。688100 的 08-27 至 09-01 已闭合旧轮次缺父指令及原批次，单列成交金额，不混入当前账期。历史安全垫峰值、手续费、过往日报与全组合净值如缺原始材料不宣称恢复；保存原值和修复附注。 + +写回只在受影响子单无在途、无未处理逐笔时进行。短暂停止 pms-beat 与 pms-worker，保留通道接收能力；所有快照比较和更新位于同一数据库事务，每条 SQL 仍经过表名前缀与单表守卫。事务写入审计记录；备份指纹不符、数据已变或中途失败均不提交。重复执行同一批号返回已完成。回退必须先比较修复后字段，若已有新成交则拒绝覆盖。写完只读核对，再恢复调度。 + +预判：重建后数量分别为 700、1,900、1,400；全部相关子单逐笔量等于累计量；虚假买入批次和 RECON 调整不再贡献成本或盈亏；指令真实进度不会因修复自动生成新单；原始成交、回放游标、策略阈值和明日启用门槛不变。测试覆盖金额、部分成交、漏笔、匹配歧义、并发、幂等、失败回滚及旧指令不复活。生产读回后逐条判收。 diff --git a/docs/复盘决定台账.md b/docs/复盘决定台账.md index 25bca4a..501611a 100644 --- a/docs/复盘决定台账.md +++ b/docs/复盘决定台账.md @@ -274,3 +274,14 @@ **预期。** 新成交不再双记数量、成本与指令进度，不改变交易参数、候选及执行规则。历史成本和盈亏不由数量对账冒充修复；原始证据已保存在 155 宿主日志目录。 **复核日期。** 2026-10-15 部署后核验身份分类与自然回放；2026-10-16 晨版核验同步检查。12 个新增用例及全量 43 批离线回归通过；部署结果追加到 DEVLOG。 + + +## 020 · 2026-10-15 · 三票历史逐笔账本恢复 + +**改动。** 用户明确批准先备份后修复 155 模拟仓的三项错账。新增只读备份、离线预演、事务写入、原值审计和条件回退入口。逐笔成交唯一对应原批次，重复汇总批次与 RECON 调整归零留痕；真实卖出按原核销规则重建盈亏。成本从金额计算，避免反乘舍入均价；父指令恢复真实进度，原误判完成的剩余执行暂停。活跃网格恢复实际剩余档位与成本。 + +**依据。** 完整留存 134 笔成交、61 个真实买入批次；33 个重复或 RECON 批次有明确来源。离线重建数量 700、1900、1400 与 QMT 相同。002979 的一条网格指令真实成交 2600、目标 1300，保留超额事实，不冒充重复删除。 + +**预期。** 数量、金额、批次盈亏和指令执行量可逐笔核验。任何漏笔、歧义、并发变化、中断不产生半套账；已有终态指令不因修账续单。手续费、历史峰值和原历史绩效不假装已还原。方案、排序、仓位及明日启用门槛不变。 + +**复核日期。** 2026-10-15 写回前后按备份核验；2026-10-16 晨版核验继续检查自然回放。21 个新回归和全部 44 批隔离离线测试通过。部署与生产判收另记 DEVLOG。 diff --git a/scripts/repair_ledger_history.py b/scripts/repair_ledger_history.py new file mode 100644 index 0000000..426ef51 --- /dev/null +++ b/scripts/repair_ledger_history.py @@ -0,0 +1,95 @@ +# -*- coding: utf-8 -*- +"""155 容器：python scripts/repair_ledger_history.py backup --codes ... > 主机备份.json +预演：python scripts/repair_ledger_history.py preview --codes ... < 备份.json +写入：停止 worker/beat 后，apply --backup-sha256 <文件哈希> --codes ... < 同一备份 +回退：rollback --run-id HIST_xxx，仅原值未再变化时可回退。审计和原证据均保留。 +脚本无模型调用、无交易下发，所有 SQL 经单表守卫；不得在开发机连接数据库。 +""" +import argparse +import hashlib +import json +import sys +from datetime import datetime +from pathlib import Path +sys.path.insert(0,str(Path(__file__).resolve().parents[1])) +from app.core.ledger_repair import compile_repair, canonical + + +def backup(codes): + from app.db.session import fetch_all, TABLE_PREFIX + from app.repo import qmt_repo as q, pms_repo as p, downstream_repo as ds + from app.services import param_store as ps + if TABLE_PREFIX: + raise ValueError('仅允许模拟仓命名空间') + out = dict(schema=1,at=datetime.now().isoformat()+'+08:00',instance='155-sim',tables={}) + for table in ['pms_position','pms_lot','pms_instruction','pms_plan','pms_command','pms_qmt_order', + 'pms_cash_flow','pms_strategy','pms_proposal','pms_daily_report','pms_nav_daily']: + out['tables'][table] = fetch_all('SELECT * FROM '+table+' ORDER BY '+('ymd' if table=='pms_nav_daily' else 'id')) + params={f'c{i}':c for i,c in enumerate(codes)}; where='ts_code IN ('+','.join(':'+k for k in params)+')' + for table in ['pms_action_ledger','pms_op_log']: + out['tables'][table]=fetch_all('SELECT * FROM '+table+' WHERE '+where+' ORDER BY id',params) + rows=[];seq=0 + while True: + page=fetch_all("SELECT * FROM pms_qmt_inbox WHERE msg_type='trade' AND seq>:s ORDER BY seq LIMIT 1000",{'s':seq}) + if not page:break + rows.extend(page);seq=page[-1]['seq'] + out['tables']['pms_qmt_inbox']=rows + out['position_snapshot']=q.latest_snapshot('positions') + out['funds_snapshot']=q.latest_snapshot('funds') + out['table_fills']=ds.fetch_filled_orders(since_time='2000-01-01 00:00:00',limit=100000) + if len(out['table_fills'])>=100000:raise ValueError('成交表截断，拒绝备份') + out['replay_cursor']=p.get_param('PMS_REPLAY_CURSOR') + out['runtime_rows']=fetch_all("SELECT * FROM pms_runtime_param WHERE param_key IN ('PMS_TOTAL_SCALE','PMS_CUSHION_SOLID','PMS_DISPATCH_MODE','PMS_PLAN_VERSION_RECHECK','PMS_REPLAY_CURSOR') ORDER BY param_key") + out['effective']={k:ps.get(k) for k in ['PMS_TOTAL_SCALE','PMS_CUSHION_SOLID','PMS_DISPATCH_MODE','PMS_PLAN_VERSION_RECHECK']} + return out + + +def guards_for(b,codes): + guards=[] + params={f'c{i}':c for i,c in enumerate(codes)};where='ts_code IN ('+','.join(':'+k for k in params)+')' + for table in ['pms_position','pms_lot','pms_instruction','pms_qmt_order','pms_strategy','pms_plan']: + guards.append(dict(name=table,sql='SELECT * FROM '+table+' WHERE '+where+' ORDER BY id FOR UPDATE', + params=params,expected=[x for x in b['tables'][table] if x['ts_code'] in codes])) + guards.append(dict(name='全部逐笔证据',sql="SELECT * FROM pms_qmt_inbox WHERE msg_type='trade' ORDER BY seq FOR UPDATE", + expected=b['tables']['pms_qmt_inbox'])) + guards.append(dict(name='运行参数及回放游标',sql="SELECT * FROM pms_runtime_param WHERE param_key IN ('PMS_TOTAL_SCALE','PMS_CUSHION_SOLID','PMS_DISPATCH_MODE','PMS_PLAN_VERSION_RECHECK','PMS_REPLAY_CURSOR') ORDER BY param_key FOR UPDATE", expected=b['runtime_rows'])) + return guards + + +def main(): + ap=argparse.ArgumentParser(description=__doc__) + ap.add_argument('mode',choices=['backup','preview','apply','rollback']) + ap.add_argument('--codes',nargs='+');ap.add_argument('--backup-sha256');ap.add_argument('--run-id') + args=ap.parse_args() + if args.mode=='backup': + if not args.codes:raise ValueError('缺股票范围') + print(canonical(backup(args.codes)));return + if args.mode=='rollback': + from app.db.session import fetch_all, apply_audited_repair, TABLE_PREFIX + if TABLE_PREFIX or not args.run_id:raise ValueError('回退实例或批号错误') + rows=fetch_all("SELECT * FROM pms_op_log WHERE ref=:r AND op='history_repair_row' ORDER BY id",{'r':args.run_id}) + if not rows:raise ValueError('无修复审计') + changes=[] + for row in rows: + item=json.loads(row['params_json']) + changes.append(dict(table=item['table'],key=item['key'],before=item['after'], + fields={k:item['before'][k] for k in set(item['fields'])|{'updated_at'}})) + print(canonical(apply_audited_repair(dict(run_id=args.run_id,changes=changes),[],rollback=True)));return + raw=sys.stdin.buffer.read();b=json.loads(raw) + if not args.codes:raise ValueError('缺股票范围') + effective=b.get('effective',{}) + plan=compile_repair(b,args.codes,scale=float(effective.get('PMS_TOTAL_SCALE',2000000)), + solid=float(effective.get('PMS_CUSHION_SOLID',0.03))) + if args.mode=='preview':print(canonical(plan));return + if not args.backup_sha256 or hashlib.sha256(raw).hexdigest()!=args.backup_sha256: + raise ValueError('完整备份文件哈希不符') + from app.db.session import apply_audited_repair, TABLE_PREFIX + if TABLE_PREFIX:raise ValueError('拒绝真实仓') + age=(datetime.now()-datetime.fromisoformat(b['at']).replace(tzinfo=None)).total_seconds() + if not 0<=age<900:raise ValueError('备份超过15分钟，须重新备份和预演') + if not b['position_snapshot'] or b['position_snapshot']['age_sec']>900: + raise ValueError('QMT持仓快照过旧') + print(canonical(apply_audited_repair(plan,guards_for(b,args.codes)))) + + +if __name__=='__main__':main() diff --git a/scripts/run_tests.py b/scripts/run_tests.py index 7811d89..0dec6d9 100644 --- a/scripts/run_tests.py +++ b/scripts/run_tests.py @@ -5,6 +5,7 @@ 运行: docker compose run --rm pms-web python scripts/run_tests.py 包含 (例数按 2026-10-12 隔离回归实跑校正): + test_batch40_units.py 逐笔历史修复、进度暂停、并发与事务回滚 (21 例) test_batch39_units.py 成交双来源归属、迟到、部分成交与失败守成 (12 例) test_batch38_units.py 每日计划顺序、版本核验、重新确认、并发采纳、历史保留 (15 例) test_core_units.py 仓位规划器 / 安全垫与成本账 (14 例) @@ -189,7 +190,7 @@ SUITES = ["test_core_units.py", "test_batch2_units.py", "test_batch3_units.py", "test_batch35_units.py", "test_batch36_units.py", "test_batch37_units.py", - "test_batch38_units.py", "test_batch39_units.py", + "test_batch38_units.py", "test_batch39_units.py", "test_batch40_units.py", "test_page_copy_guard.py", "test_page_enum_guard.py", "test_page_wiring_guard.py", diff --git a/scripts/test_batch40_units.py b/scripts/test_batch40_units.py new file mode 100644 index 0000000..cdbef73 --- /dev/null +++ b/scripts/test_batch40_units.py @@ -0,0 +1,164 @@ +# -*- coding: utf-8 -*- +"""本地：python scripts/unit_test_guard.py scripts/test_batch40_units.py。历史修复离线回归。""" +import unit_test_guard # noqa +import sys +from pathlib import Path +sys.path.insert(0,str(Path(__file__).resolve().parents[1])) +import json +import unittest +from copy import deepcopy +from contextlib import contextmanager +from unittest.mock import patch +from app.core.ledger_repair import compile_repair, net_lot_cost, D +from app.db import session + +CODE='600001.SH' + + +def fixture(): + ins=dict(id=1,instruction_id='P',ts_code=CODE,qty=200,exec_qty=200,status='CONFIRMED', + progress_json='{"children":[{"ref":"P_D01"}]}',origin_id=None,updated_at='2026-10-15 10:00:00') + order=dict(instruction_id='P_D01',parent_id='P',ts_code=CODE,side='buy',broker_order_id='B', + status='FILLED',cum_qty=100,created_at='2026-10-15 10:00:00') + lot=dict(id=1,ts_code=CODE,instruction_id='P',qty=100,closed_qty=0,open_price='10.000', + open_date='2026-10-15',created_at='2026-10-15 10:01:00',lot_type='BASE',realized_pnl='0.00') + pos=dict(id=1,ts_code=CODE,total_qty=200,avg_cost='10.000',status='HOLDING') + payload=dict(ts_code=CODE,instruction_id='P_D01',trade_no='T',broker_order_id='B',side='buy', + qty=100,price=10,amount=1000,trade_time=1792029600000) + trade=dict(seq=1,payload_json=json.dumps(payload),processed=1,received_at='2026-10-15 10:00:00', + processed_at='2026-10-15 10:01:00') + tables=dict(pms_position=[pos],pms_instruction=[ins],pms_qmt_order=[order],pms_lot=[lot,dict(lot,id=2, + instruction_id=None,created_at='2026-10-15 10:02:00')],pms_qmt_inbox=[trade],pms_plan=[],pms_strategy=[]) + return dict(instance='155-sim',at='2026-10-15T12:00:00+08:00',tables=tables, + position_snapshot={'payload':{'items':[dict(ts_code=CODE,total_qty=100,avail_qty=100,market_price=11)]}}, + table_fills=[dict(order_id='P_D01',ts_code=CODE,side='buy',qty=100,price=10,done_time='2026-10-15 10:00:00')]) + + +class CoreTests(unittest.TestCase): + def plan(self,b=None):return compile_repair(b or fixture(),[CODE]) + def test_quantity_cost_progress_and_terminal_hold(self): + p=self.plan();s=p['summaries'][0] + self.assertEqual((s['qty'],s['avg_cost'],s['avail_qty']),(100,'10.000',0)) + self.assertEqual(s['void_lots'],[2]);self.assertEqual(s['instructions'][0]['new_status'],'CANCELLED') + self.assertEqual(s['instructions'][0]['exec_qty'],100) + def test_determinism(self):self.assertEqual(self.plan(),self.plan()) + def test_missing_trade_blocks(self): + b=fixture();b['tables']['pms_qmt_inbox']=[] + with self.assertRaises(ValueError):self.plan(b) + def test_duplicate_trade_blocks(self): + b=fixture();b['tables']['pms_qmt_inbox']*=2 + with self.assertRaises(ValueError):self.plan(b) + def test_wrong_broker_blocks(self): + b=fixture();b['tables']['pms_qmt_order'][0]['broker_order_id']='X' + with self.assertRaises(ValueError):self.plan(b) + def test_missing_summary_evidence_blocks(self): + b=fixture();b['table_fills']=[] + with self.assertRaises(ValueError):self.plan(b) + def test_ambiguous_lot_blocks(self): + b=fixture();b['tables']['pms_lot'][1].update(instruction_id='P',created_at='2026-10-15 10:01:00') + with self.assertRaises(ValueError):self.plan(b) + def test_position_disagreement_blocks(self): + b=fixture();b['position_snapshot']['payload']['items'][0]['total_qty']=200 + with self.assertRaises(ValueError):self.plan(b) + def test_pending_order_blocks(self): + b=fixture();b['tables']['pms_qmt_order'][0]['status']='PARTIAL' + with self.assertRaises(ValueError):self.plan(b) + def test_active_grid_cost_and_qty_follow_actual_lots(self): + b=fixture();t=b['tables'];t['pms_instruction'][0].update(origin_id='GRID',progress_json='{"leg":"grid_buy:3"}') + t['pms_lot'][0]['lot_type']='ADD' + t['pms_strategy']=[dict(id=1,ts_code=CODE,strategy_id='GRID',type='GRID',status='ACTIVE', + state_json='{"pending":null,"filled_levels":{"3":{"qty":200,"price":10}},"invested":2000}')] + p=self.plan(b);row=next(c for c in p['changes'] if c['table']=='pms_strategy') + st=json.loads(row['fields']['state_json']) + self.assertEqual(st['filled_levels']['3']['qty'],100);self.assertEqual(st['invested'],1000) + def test_missing_baseline_is_not_assumed_zero(self): + b=fixture();p=json.loads(b['tables']['pms_qmt_inbox'][0]['payload_json']) + p.update(trade_no='OLD',trade_time=p['trade_time']-86400000) + b['tables']['pms_qmt_inbox'].append(dict(b['tables']['pms_qmt_inbox'][0],seq=0,payload_json=json.dumps(p))) + with self.assertRaises(ValueError):self.plan(b) + def test_existing_live_parent_stays_live(self): + b=fixture();b['tables']['pms_instruction'][0]['status']='DISPATCHED' + self.assertEqual(self.plan(b)['summaries'][0]['instructions'][0]['new_status'],'DISPATCHED') + def test_real_overfill_is_not_erased(self): + b=fixture();b['tables']['pms_instruction'][0]['qty']=50 + s=self.plan(b)['summaries'][0]['instructions'][0] + self.assertEqual((s['exec_qty'],s['overfill_qty']),(100,50)) + def test_money_precision_does_not_multiply_rounded_average(self): + lots=[dict(qty=1,closed_qty=3,open_price='10.000',close_avg_price='10.003',realized_pnl='0.01')] + self.assertEqual(net_lot_cost(lots),D('9.99')) + def test_sell_rebuilds_pnl_and_remaining_cost(self): + b=fixture();t=b['tables'];o=dict(t['pms_qmt_order'][0],instruction_id='S_D01',parent_id='S',side='sell',cum_qty=40) + t['pms_qmt_order'].append(o);t['pms_instruction'].append(dict(t['pms_instruction'][0],id=2,instruction_id='S',qty=40)) + p=json.loads(t['pms_qmt_inbox'][0]['payload_json']);p.update(instruction_id='S_D01',trade_no='T2',side='sell',qty=40,price=11,amount=440,trade_time=p['trade_time']+60000) + t['pms_qmt_inbox'].append(dict(t['pms_qmt_inbox'][0],seq=2,payload_json=json.dumps(p))) + b['position_snapshot']['payload']['items'][0]['total_qty']=60 + s=self.plan(b)['summaries'][0] + self.assertEqual((s['qty'],s['realized_pnl'],s['avg_cost']),(60,'40.00','9.333')) + + +class Result: + def __init__(self,rows):self.rows=rows + def mappings(self):return self + def fetchall(self):return self.rows + + +class Engine: + def __init__(self):self.rows={1:{'id':1,'qty':100,'ts_code':CODE},2:{'id':2,'qty':200,'ts_code':CODE}};self.audit=[];self.fail=False + @contextmanager + def begin(self): + old=deepcopy((self.rows,self.audit)) + try:yield self + except Exception: + self.rows,self.audit=old + raise + def execute(self,sql,params): + s=str(sql);session.assert_single_table(s) + if s.startswith('SELECT id FROM pms_op_log'): + return Result([{'id':1}] if any(x['op']=='history_repair_summary' and x['ref']==params['r'] for x in self.audit) else []) + if s.startswith('SELECT * FROM pms_lot'):return Result([deepcopy(self.rows[params['id']])]) + if s.startswith('UPDATE'): + if self.fail and params['row_id']==2:raise RuntimeError('interrupted') + self.rows[params['row_id']].update({k:v for k,v in params.items() if k!='row_id'}) + return Result([]) + if s.startswith('INSERT INTO pms_op_log'):self.audit.append(deepcopy(params));return Result([]) + raise AssertionError(s) + + +class TransactionTests(unittest.TestCase): + def setUp(self): + self.engine=Engine();self.plan=dict(run_id='TEST',changes=[dict(table='pms_lot',key={'id':i},before=deepcopy(r),fields={'qty':0}) for i,r in self.engine.rows.items()]) + self.p=patch.object(session,'get_engine',return_value=self.engine);self.p.start();self.addCleanup(self.p.stop) + def test_success_and_idempotency(self): + self.assertEqual(session.apply_audited_repair(self.plan,[])['changed_rows'],2) + self.assertTrue(session.apply_audited_repair(self.plan,[])['already_applied']) + self.assertEqual(len(self.engine.audit),3) + def test_failure_rolls_back_all_rows_and_audit(self): + self.engine.fail=True + with self.assertRaises(RuntimeError):session.apply_audited_repair(self.plan,[]) + self.assertEqual(self.engine.rows[1]['qty'],100);self.assertEqual(self.engine.audit,[]) + def test_changed_before_image_rejects(self): + self.engine.rows[2]['qty']=201 + with self.assertRaises(ValueError):session.apply_audited_repair(self.plan,[]) + self.assertEqual(self.engine.rows[1]['qty'],100) + def test_forbidden_table_rejected(self): + self.plan['changes'][0]['table']='real_pms_lot' + with self.assertRaises(ValueError):session.apply_audited_repair(self.plan,[]) + def test_audited_inverse_restores_original_values(self): + for row in self.engine.rows.values():row['updated_at']='2026-10-15 09:00:00' + for item in self.plan['changes']:item['before']=deepcopy(self.engine.rows[item['key']['id']]) + original=deepcopy(self.engine.rows) + session.apply_audited_repair(self.plan,[]) + changes=[] + for row in self.engine.audit: + if row['op']!='history_repair_row':continue + item=json.loads(row['params']) + changes.append(dict(table=item['table'],key=item['key'],before=item['after'], + fields={k:item['before'][k] for k in set(item['fields'])|{'updated_at'}})) + session.apply_audited_repair(dict(run_id='TEST',changes=changes),[],rollback=True) + self.assertEqual(self.engine.rows,original) + def test_single_table_guard_still_applies(self): + with self.assertRaises(session.MultiTableSQL): + session.apply_audited_repair(self.plan,[dict(name='bad',sql='SELECT * FROM pms_lot JOIN pms_position ON 1=1',expected=[])]) + + +if __name__=='__main__':unittest.main()