test_algorithm.py 文件源码

def test_order_in_quiet_period(self, name, sid):
        asset = self.asset_finder.retrieve_asset(sid)

        algo_code = dedent("""
        from catalyst.api import (
            sid,
            order,
            order_value,
            order_percent,
            order_target,
            order_target_percent,
            order_target_value
        )

        def initialize(context):
            pass

        def handle_data(context, data):
            order(sid({sid}), 1)
            order_value(sid({sid}), 100)
            order_percent(sid({sid}), 0.5)
            order_target(sid({sid}), 50)
            order_target_percent(sid({sid}), 0.5)
            order_target_value(sid({sid}), 50)
        """).format(sid=sid)

        # run algo from 1/6 to 1/7
        algo = TradingAlgorithm(
            script=algo_code,
            env=self.env,
            sim_params=SimulationParameters(
                start_session=pd.Timestamp("2016-01-06", tz='UTC'),
                end_session=pd.Timestamp("2016-01-07", tz='UTC'),
                trading_calendar=self.trading_calendar,
                data_frequency="minute"
            )
        )

        with make_test_handler(self) as log_catcher:
            algo.run(self.data_portal)

            warnings = [r for r in log_catcher.records
                        if r.level == logbook.WARNING]

            # one warning per order on the second day
            self.assertEqual(6 * 390, len(warnings))

            for w in warnings:
                expected_message = (
                    'Cannot place order for ASSET{sid}, as it has de-listed. '
                    'Any existing positions for this asset will be liquidated '
                    'on {date}.'.format(sid=sid, date=asset.auto_close_date)
                )
                self.assertEqual(expected_message, w.message)