96 lines
5.8 KiB
Python
96 lines
5.8 KiB
Python
# -*- 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()
|