test_blotter.py 文件源码

def test_automatic_events_closed_pnl_mark(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", ccy="USD", margin=0, multiplier=1,
                           commission=0, isFX=False)
        blt.map_instrument("ES", "ESZ15")

        ts = pd.Timestamp("2015-08-04T11:00:00")
        blt.trade(ts, "ESZ15", 1, 2000)

        ts = pd.Timestamp("2015-08-04T12:00:00")
        hlds = blt.get_instruments()
        for instr, qty in hlds.iteritems():
            blt.trade(ts, instr, -qty, 2001)

        ts = pd.Timestamp("2015-08-05T00:00:00")
        blt.automatic_events(ts)
        pnl_history = blt.get_pnl_history()
        usd = pd.DataFrame([[1.0, 1.0, 0.0]], index=[ts],
                           columns=["pnl", "closed pnl", "open pnl"])
        pnl_history_exp = {"USD": usd}
        self.assertDictDataFrameEqual(pnl_history, pnl_history_exp)