tradingSystem/scripts/test_batch40_units.py

# -*- 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_new_fill_invalidates_fixed_historical_average(self):
        import sqlite3
        from app.repo import pms_repo
        with sqlite3.connect(':memory:') as db:
            db.execute('CREATE TABLE pms_instruction (instruction_id TEXT, exec_qty INTEGER, exec_avg_price REAL, updated_at TEXT, progress_json TEXT)')
            db.execute("INSERT INTO pms_instruction VALUES ('P',100,10,'old','fixed evidence')")
            with patch.object(pms_repo,'execute',side_effect=lambda sql,params:db.execute(sql,params).rowcount):
                pms_repo.add_instruction_exec('P',100)
            row=db.execute('SELECT exec_qty,exec_avg_price,progress_json FROM pms_instruction').fetchone()
            self.assertEqual(row,(200,None,'fixed evidence'))
    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()