test_algorithm.py 文件源码

def test_minute_data(self, algo_class):
        start_session = pd.Timestamp('2002-1-2', tz='UTC')
        period_end = pd.Timestamp('2002-1-4', tz='UTC')
        equities = pd.DataFrame([{
            'start_date': start_session,
            'end_date': period_end + timedelta(days=1),
            'exchange': "TEST",
        }] * 2)
        equities['symbol'] = ['A', 'B']
        with TempDirectory() as tempdir, \
                tmp_trading_env(equities=equities,
                                load=self.make_load_function()) as env:
            sim_params = SimulationParameters(
                start_session=start_session,
                end_session=period_end,
                capital_base=1.0e5,
                data_frequency='minute',
                trading_calendar=self.trading_calendar,
            )

            data_portal = create_data_portal(
                env.asset_finder,
                tempdir,
                sim_params,
                equities.index,
                self.trading_calendar,
            )
            algo = algo_class(sim_params=sim_params, env=env)
            algo.run(data_portal)