test_blotter.py 文件源码

def test_automatic_events_future_type_creation(self):
        blt = blotter.Blotter(self.prices, self.rates,
                              accrual_time=pd.Timedelta(0, unit='h'),
                              eod_time=pd.Timedelta(0, unit='h'),
                              sweep_time=pd.Timedelta(0, unit='h'))
        blt.connect_market_data()
        blt.define_generic("ES", "USD", 0.1, 100, 2.50)
        blt.map_instrument("ES", "ESZ15")
        ts = pd.Timestamp('2015-08-04T10:00:00')
        number_instr = 1
        blt._trade(ts, "ESZ15", number_instr, 2000)
        blt.automatic_events(pd.Timestamp('2015-08-05T10:00:00'))

        ev_types = []
        for ev in blt.event_log:
            ev_types.append(ev.split("|")[0])

        ev_types_exp = ["TRADE", "INTEREST", "PNL"]
        self.assertEqual(ev_types, ev_types_exp)