test_algorithm_gen.py 文件源码

def test_progress(self):
        """
        Ensure the pipeline of generators are in sync, at least as far as
        their current dates.
        """
        sim_params = factory.create_simulation_parameters(
            start=datetime(2008, 1, 1, tzinfo=pytz.utc),
            end=datetime(2008, 1, 5, tzinfo=pytz.utc),
            env=self.env,
        )
        algo = TestAlgo(self, sim_params=sim_params, env=self.env)
        trade_source = factory.create_daily_trade_source(
            [8229],
            sim_params,
            env=self.env,
        )
        algo.set_sources([trade_source])

        gen = algo.get_generator()
        results = list(gen)
        self.assertEqual(results[-2]['progress'], 1.0)