Python testfixtures 模块，TempDirectory() 实例源码

def setUpClass(cls):
        cls.test_data_dir = TempDirectory()
        cls.db_path = cls.test_data_dir.getpath('adjustments.db')
        all_days = TradingEnvironment().trading_days
        cls.calendar_days = all_days[
            all_days.slice_indexer(TEST_CALENDAR_START, TEST_CALENDAR_STOP)
        ]
        daily_bar_reader = MockDailyBarSpotReader()
        writer = SQLiteAdjustmentWriter(cls.db_path, cls.calendar_days,
                                        daily_bar_reader)
        writer.write(SPLITS, MERGERS, DIVIDENDS)

        cls.assets = TEST_QUERY_ASSETS
        cls.asset_info = EQUITY_INFO
        cls.bcolz_writer = SyntheticDailyBarWriter(
            cls.asset_info,
            cls.calendar_days,
        )
        cls.bcolz_path = cls.test_data_dir.getpath('equity_pricing.bcolz')
        cls.bcolz_writer.write(cls.bcolz_path, cls.calendar_days, cls.assets)

def setUp(self):

        self.asset_info = EQUITY_INFO
        self.writer = SyntheticDailyBarWriter(
            self.asset_info,
            self.trading_days,
        )

        self.dir_ = TempDirectory()
        self.dir_.create()
        self.dest = self.dir_.getpath('daily_equity_pricing.bcolz')

def setUp(self):

        self.dir_ = TempDirectory()
        self.dir_.create()
        self.dest = self.dir_.getpath('minute_bars')
        os.makedirs(self.dest)
        self.writer = BcolzMinuteBarWriter(
            TEST_CALENDAR_START,
            self.dest,
            self.market_opens,
            self.market_closes,
            US_EQUITIES_MINUTES_PER_DAY,
        )
        self.reader = BcolzMinuteBarReader(self.dest)

def setUpClass(cls):
        cls.first_asset_start = Timestamp('2015-04-01', tz='UTC')
        cls.env = TradingEnvironment()
        cls.trading_day = day = cls.env.trading_day
        cls.calendar = date_range('2015', '2015-08', tz='UTC', freq=day)

        cls.asset_info = make_rotating_equity_info(
            num_assets=6,
            first_start=cls.first_asset_start,
            frequency=day,
            periods_between_starts=4,
            asset_lifetime=8,
        )
        cls.last_asset_end = cls.asset_info['end_date'].max()
        cls.all_asset_ids = cls.asset_info.index

        cls.env.write_data(equities_df=cls.asset_info)
        cls.finder = cls.env.asset_finder

        cls.temp_dir = TempDirectory()
        cls.temp_dir.create()

        try:
            cls.writer = SyntheticDailyBarWriter(
                asset_info=cls.asset_info[['start_date', 'end_date']],
                calendar=cls.calendar,
            )
            table = cls.writer.write(
                cls.temp_dir.getpath('testdata.bcolz'),
                cls.calendar,
                cls.all_asset_ids,
            )

            cls.pipeline_loader = USEquityPricingLoader(
                BcolzDailyBarReader(table),
                NullAdjustmentReader(),
            )
        except:
            cls.temp_dir.cleanup()
            raise

def setUpClass(cls):
        cls.AAPL = 1
        cls.MSFT = 2
        cls.BRK_A = 3
        cls.assets = [cls.AAPL, cls.MSFT, cls.BRK_A]
        asset_info = make_simple_equity_info(
            cls.assets,
            Timestamp('2014'),
            Timestamp('2015'),
            ['AAPL', 'MSFT', 'BRK_A'],
        )
        cls.env = trading.TradingEnvironment()
        cls.env.write_data(equities_df=asset_info)
        cls.tempdir = tempdir = TempDirectory()
        tempdir.create()
        try:
            cls.raw_data, bar_reader = cls.create_bar_reader(tempdir)
            adj_reader = cls.create_adjustment_reader(tempdir)
            cls.pipeline_loader = USEquityPricingLoader(
                bar_reader, adj_reader
            )
        except:
            cls.tempdir.cleanup()
            raise

        cls.dates = cls.raw_data[cls.AAPL].index.tz_localize('UTC')
        cls.AAPL_split_date = Timestamp("2014-06-09", tz='UTC')

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)

def test_algo_without_rl_violation_after_delete(self):
        sim_params = factory.create_simulation_parameters(
            start=self.extra_knowledge_date,
            num_days=4,
        )
        equities = pd.DataFrame.from_records([{
            'symbol': 'BZQ',
            'start_date': sim_params.start_session,
            'end_date': sim_params.end_session,
            'exchange': "TEST",
        }])
        with TempDirectory() as new_tempdir, \
                security_list_copy(), \
                tmp_trading_env(equities=equities,
                                load=self.make_load_function()) as env:
            # add a delete statement removing bzq
            # write a new delete statement file to disk
            add_security_data([], ['BZQ'])

            data_portal = create_data_portal(
                env.asset_finder,
                new_tempdir,
                sim_params,
                range(0, 5),
                trading_calendar=self.trading_calendar,
            )

            algo = RestrictedAlgoWithoutCheck(
                symbol='BZQ', sim_params=sim_params, env=env
            )
            algo.run(data_portal)

def setUp(self):
        self.scratch = TempDirectory()

def tempdir():
    """py.test fixture to create a temporary folder for a test."""

    with testfixtures.TempDirectory(create=True) as d:
        yield d
        d.cleanup()

def setUp(self):
        """Initialise temporary directory."""
        self.tempdir = TempDirectory(encoding='utf-8')

def setUp(self):
        self.tempdir = TempDirectory(encoding='utf-8')

def setUp(self):
        self.tempdir = TempDirectory(encoding='utf-8')

def setUp(self):
        self.temp = TempDirectory(encoding='utf-8')
        self.tempdir = self.temp.path
        self.DB_NAME = DB_NAME
        self.db_helper = DatabaseHelper(self.DB_NAME)
        self.operations = SorterOps(self.db_helper)