"""R613:基线与额外延迟分列,逐腿记录尾部不可成交。""" PRICES = [100, 101, 103, 100, 106, 104, 108, 107] EVENTS = {'A': (0, 3), 'B': (1, 4), 'C': (5, 6)} BASE_LAG = 1 def run(candidate, extra): legs = [] for side, signal_t in zip(['buy', 'sell'], EVENTS[candidate]): fill_t = signal_t + BASE_LAG + extra assert fill_t > signal_t legs.append({'side': side, 'signal_t': signal_t, 'base_lag': BASE_LAG, 'extra_lag': extra, 'fill_t': fill_t, 'status': 'filled' if fill_t < len(PRICES) else 'out_of_range', 'price': PRICES[fill_t] if fill_t < len(PRICES) else None}) buy, sell = legs if buy['status'] != 'filled': return {'status': 'no_entry', 'legs': legs, 'closed_pnl': None, 'open_shares': 0, 'unrealized_pnl': None} if sell['status'] != 'filled': return {'status': 'open_at_end', 'legs': legs, 'closed_pnl': None, 'open_shares': 1, 'unrealized_pnl': PRICES[-1] - buy['price']} return {'status': 'closed', 'legs': legs, 'closed_pnl': sell['price'] - buy['price'], 'open_shares': 0, 'unrealized_pnl': 0} def main(): matrix = {name: [run(name, d) for d in [0, 1, 2]] for name in EVENTS} assert [r['closed_pnl'] for r in matrix['A']] == [5, 1, 8] assert [r['closed_pnl'] for r in matrix['B']] == [1, 8, 1] assert matrix['A'][0]['closed_pnl'] > matrix['B'][0]['closed_pnl'] assert matrix['A'][1]['closed_pnl'] < matrix['B'][1]['closed_pnl'] assert matrix['C'][1]['open_shares'] == 1 assert matrix['C'][1]['unrealized_pnl'] == 0 assert matrix['C'][2]['status'] == 'no_entry' assert all(leg['status'] == 'out_of_range' for leg in matrix['C'][2]['legs']) assert matrix['A'][0]['legs'][0]['fill_t'] == 1 assert matrix['A'][1]['legs'][0]['fill_t'] == 2 print('base_lag=1 extras=0,1,2 A_closed_pnl=[5, 1, 8] B_closed_pnl=[1, 8, 1]') print('C_status=' + str([r['status'] for r in matrix['C']])) print('C_extra1_legs=filled,out_of_range shares=1 unrealized=0 closed_pnl=None') print('C_extra2_legs=out_of_range,out_of_range shares=0 closed_pnl=None') if __name__ == '__main__': main()