test_assets.py 文件源码

def test_group_by_type(self):
        equities = make_simple_equity_info(
            range(5),
            start_date=pd.Timestamp('2014-01-01'),
            end_date=pd.Timestamp('2015-01-01'),
        )
        futures = make_commodity_future_info(
            first_sid=6,
            root_symbols=['CL'],
            years=[2014],
        )
        # Intersecting sid queries, to exercise loading of partially-cached
        # results.
        queries = [
            ([0, 1, 3], [6, 7]),
            ([0, 2, 3], [7, 10]),
            (list(equities.index), list(futures.index)),
        ]
        with tmp_asset_finder(equities=equities, futures=futures) as finder:
            for equity_sids, future_sids in queries:
                results = finder.group_by_type(equity_sids + future_sids)
                self.assertEqual(
                    results,
                    {'equity': set(equity_sids), 'future': set(future_sids)},
                )