test_perf_tracking.py 文件源码

def test_close_position_event(self):
        pt = perf.PositionTracker(asset_finder=self.env.asset_finder)
        dt = pd.Timestamp("1984/03/06 3:00PM")
        pos1 = perf.Position(1, amount=np.float64(120.0),
                             last_sale_date=dt, last_sale_price=3.4)
        pos2 = perf.Position(2, amount=np.float64(-100.0),
                             last_sale_date=dt, last_sale_price=3.4)
        pt.update_positions({1: pos1, 2: pos2})

        event_type = DATASOURCE_TYPE.CLOSE_POSITION
        index = [dt + timedelta(days=1)]
        pan = pd.Panel({1: pd.DataFrame({'price': 1, 'volume': 0,
                                         'type': event_type}, index=index),
                        2: pd.DataFrame({'price': 1, 'volume': 0,
                                         'type': event_type}, index=index),
                        3: pd.DataFrame({'price': 1, 'volume': 0,
                                         'type': event_type}, index=index)})

        source = DataPanelSource(pan)
        for i, event in enumerate(source):
            txn = pt.maybe_create_close_position_transaction(event)
            if event.sid == 1:
                # Test owned long
                self.assertEqual(-120, txn.amount)
            elif event.sid == 2:
                # Test owned short
                self.assertEqual(100, txn.amount)
            elif event.sid == 3:
                # Test not-owned SID
                self.assertIsNone(txn)