历史账本：按逐笔恢复数量成本盈亏与执行进度，备份审计及事务回退（台账 020）
一、双来源历史重复污染了三票的批次、成本与父指令进度。 完整备份的 134 笔成交重建为 700、1900、1400 股；只按数量对账无法恢复盈亏。 二、真实买入唯一对应原批次，卖出按既有批次优先级重算。 33 个虚假或 RECON 批次归零留痕，61 个真实批次恢复；实际超目标成交保留。 误判已完成指令的剩余执行暂停，不自动复活旧单；活跃网格恢复真实余量。 三、写回经过备份哈希、并发原行核对和单表 SQL 事务，逐行审计并支持条件回退。 21 个新用例与完整 44 批隔离禁网回归通过。未改线上参数、原始成交、游标或历史净值； 手续费与历史峰值缺证据的部分明确保留边界。生产判收另记。
This commit is contained in:
baobao
parent
43cdc25881
commit
0871885801
|
|
@ -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、讨论稿和探针未纳入本提交。
|
||||
|
|
|
|||
|
|
@ -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))
|
||||
|
|
@ -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'])}
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
|
|
|
|||
|
|
@ -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 调整不再贡献成本或盈亏；指令真实进度不会因修复自动生成新单；原始成交、回放游标、策略阈值和明日启用门槛不变。测试覆盖金额、部分成交、漏笔、匹配歧义、并发、幂等、失败回滚及旧指令不复活。生产读回后逐条判收。
|
||||
|
|
|
|||
|
|
@ -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。
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
|
|
@ -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",
|
||||
|
|
|
|||
|
|
@ -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()
|
||||
Loading…
Reference in New Issue