zipline.gens.AlgorithmSimulator.once_a_day() - Code Metrics - Inspection of "Q 2.0" - quantopian/zipline - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#858)

by Eddie

created 2015-12-14 15:18 UTC

zipline.gens.AlgorithmSimulator.once_a_day() A

↳ Parent: zipline.gens.AlgorithmSimulator.transform()

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	3
dl	0
loc	22
rs	9.2

#
# Copyright 2015 Quantopian, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from logbook import Logger, Processor
from pandas.tslib import normalize_date
from zipline.protocol import BarData
from zipline.utils.api_support import ZiplineAPI

from zipline.gens.sim_engine import (
    BAR,
    DAY_START,
    DAY_END,
    MINUTE_END
)

log = Logger('Trade Simulation')


class AlgorithmSimulator(object):

    EMISSION_TO_PERF_KEY_MAP = {
        'minute': 'minute_perf',
        'daily': 'daily_perf'
    }

    def __init__(self, algo, sim_params, data_portal, clock, benchmark_source):

        # ==============
        # Simulation
        # Param Setup
        # ==============
        self.sim_params = sim_params
        self.env = algo.trading_environment
        self.data_portal = data_portal

        # ==============
        # Algo Setup
        # ==============
        self.algo = algo
        self.algo_start = normalize_date(self.sim_params.first_open)

        # ==============
        # Snapshot Setup
        # ==============

        # The algorithm's data as of our most recent event.
        # We want an object that will have empty objects as default
        # values on missing keys.
        self.current_data = BarData(data_portal, self)

        # We don't have a datetime for the current snapshot until we
        # receive a message.
        self.simulation_dt = None

        self.clock = clock

        self.benchmark_source = benchmark_source

        # =============
        # Logging Setup
        # =============

        # Processor function for injecting the algo_dt into
        # user prints/logs.
        def inject_algo_dt(record):
            if 'algo_dt' not in record.extra:
                record.extra['algo_dt'] = self.simulation_dt
        self.processor = Processor(inject_algo_dt)

    def transform(self):
        """
        Main generator work loop.
        """
        algo = self.algo
        algo.data_portal = self.data_portal
        handle_data = algo.event_manager.handle_data
        current_data = self.current_data

        data_portal = self.data_portal

        # can't cache a pointer to algo.perf_tracker because we're not
        # guaranteed that the algo doesn't swap out perf trackers during
        # its lifetime.
        # likewise, we can't cache a pointer to the blotter.

        algo.perf_tracker.position_tracker.data_portal = data_portal

        def every_bar(dt_to_use):
            # called every tick (minute or day).

            self.simulation_dt = dt_to_use
            algo.on_dt_changed(dt_to_use)

            blotter = algo.blotter
            perf_tracker = algo.perf_tracker

            # handle any transactions and commissions coming out new orders
            # placed in the last bar
            new_transactions, new_commissions = \
                blotter.get_transactions(data_portal)

            for transaction in new_transactions:
                perf_tracker.process_transaction(transaction)

                # since this order was modified, record it
                order = blotter.orders[transaction.order_id]
                perf_tracker.process_order(order)

            if new_commissions:
                for commission in new_commissions:
                    perf_tracker.process_commission(commission)

            handle_data(algo, current_data, dt_to_use)

            # grab any new orders from the blotter, then clear the list.
            # this includes cancelled orders.
            new_orders = blotter.new_orders
            blotter.new_orders = []

            # if we have any new orders, record them so that we know
            # in what perf period they were placed.
            if new_orders:
                for new_order in new_orders:
                    perf_tracker.process_order(new_order)

        def once_a_day(midnight_dt):
            # set all the timestamps
            self.simulation_dt = midnight_dt
            algo.on_dt_changed(midnight_dt)
            data_portal.current_day = midnight_dt

            # call before trading start
            algo.before_trading_start(current_data)

            perf_tracker = algo.perf_tracker

            # handle any splits that impact any positions or any open orders.
            sids_we_care_about = \
                list(set(list(perf_tracker.position_tracker.positions.keys()) +
                         list(algo.blotter.open_orders.keys())))

            if len(sids_we_care_about) > 0:
                splits = data_portal.get_splits(sids_we_care_about,
                                                midnight_dt)
                if len(splits) > 0:
                    algo.blotter.process_splits(splits)
                    perf_tracker.position_tracker.handle_splits(splits)

        def handle_benchmark(date):
            algo.perf_tracker.all_benchmark_returns[date] = \
                self.benchmark_source.get_value(date)

        with self.processor, ZiplineAPI(self.algo):
            for dt, action in self.clock:
                if action == BAR:
                    every_bar(dt)
                elif action == DAY_START:
                    once_a_day(dt)
                elif action == DAY_END:
                    # End of the day.
                    handle_benchmark(dt)
                    yield self._get_daily_message(dt, algo, algo.perf_tracker)
                elif action == MINUTE_END:
                    handle_benchmark(dt)
                    minute_msg, daily_msg = \
                        self._get_minute_message(dt, algo, algo.perf_tracker)

                    yield minute_msg

                    if daily_msg:
                        yield daily_msg

        risk_message = algo.perf_tracker.handle_simulation_end()
        yield risk_message

    @staticmethod
    def _get_daily_message(dt, algo, perf_tracker):
        """
        Get a perf message for the given datetime.
        """
        perf_message = perf_tracker.handle_market_close_daily(dt)
        perf_message['daily_perf']['recorded_vars'] = algo.recorded_vars
        return perf_message

    @staticmethod
    def _get_minute_message(dt, algo, perf_tracker):
        """
        Get a perf message for the given datetime.
        """
        rvars = algo.recorded_vars

        minute_message, daily_message = perf_tracker.handle_minute_close(dt)
        minute_message['minute_perf']['recorded_vars'] = rvars

        if daily_message:
            daily_message["daily_perf"]["recorded_vars"] = rvars

        return minute_message, daily_message


1			#
2			# Copyright 2015 Quantopian, Inc.
3			#
4			# Licensed under the Apache License, Version 2.0 (the "License");
5			# you may not use this file except in compliance with the License.
6			# You may obtain a copy of the License at
7			#
8			# http://www.apache.org/licenses/LICENSE-2.0
9			#
10			# Unless required by applicable law or agreed to in writing, software
11			# distributed under the License is distributed on an "AS IS" BASIS,
12			# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13			# See the License for the specific language governing permissions and
14			# limitations under the License.
15			from logbook import Logger, Processor
16			from pandas.tslib import normalize_date
17			from zipline.protocol import BarData
18			from zipline.utils.api_support import ZiplineAPI
19
20			from zipline.gens.sim_engine import (
21			BAR,
22			DAY_START,
23			DAY_END,
24			MINUTE_END
25			)
26
27			log = Logger('Trade Simulation')
28
29
30			class AlgorithmSimulator(object):
31
32			EMISSION_TO_PERF_KEY_MAP = {
33			'minute': 'minute_perf',
34			'daily': 'daily_perf'
35			}
36
37			def __init__(self, algo, sim_params, data_portal, clock, benchmark_source):
38
39			# ==============
40			# Simulation
41			# Param Setup
42			# ==============
43			self.sim_params = sim_params
44			self.env = algo.trading_environment
45			self.data_portal = data_portal
46
47			# ==============
48			# Algo Setup
49			# ==============
50			self.algo = algo
51			self.algo_start = normalize_date(self.sim_params.first_open)
52
53			# ==============
54			# Snapshot Setup
55			# ==============
56
57			# The algorithm's data as of our most recent event.
58			# We want an object that will have empty objects as default
59			# values on missing keys.
60			self.current_data = BarData(data_portal, self)
61
62			# We don't have a datetime for the current snapshot until we
63			# receive a message.
64			self.simulation_dt = None
65
66			self.clock = clock
67
68			self.benchmark_source = benchmark_source
69
70			# =============
71			# Logging Setup
72			# =============
73
74			# Processor function for injecting the algo_dt into
75			# user prints/logs.
76			def inject_algo_dt(record):
77			if 'algo_dt' not in record.extra:
78			record.extra['algo_dt'] = self.simulation_dt
79			self.processor = Processor(inject_algo_dt)
80
81			def transform(self):
82			"""
83			Main generator work loop.
84			"""
85			algo = self.algo
86			algo.data_portal = self.data_portal
87			handle_data = algo.event_manager.handle_data
88			current_data = self.current_data
89
90			data_portal = self.data_portal
91
92			# can't cache a pointer to algo.perf_tracker because we're not
93			# guaranteed that the algo doesn't swap out perf trackers during
94			# its lifetime.
95			# likewise, we can't cache a pointer to the blotter.
96
97			algo.perf_tracker.position_tracker.data_portal = data_portal
98
99			def every_bar(dt_to_use):
100			# called every tick (minute or day).
101
102			self.simulation_dt = dt_to_use
103			algo.on_dt_changed(dt_to_use)
104
105			blotter = algo.blotter
106			perf_tracker = algo.perf_tracker
107
108			# handle any transactions and commissions coming out new orders
109			# placed in the last bar
110			new_transactions, new_commissions = \
111			blotter.get_transactions(data_portal)
112
113			for transaction in new_transactions:
114			perf_tracker.process_transaction(transaction)
115
116			# since this order was modified, record it
117			order = blotter.orders[transaction.order_id]
118			perf_tracker.process_order(order)
119
120			if new_commissions:
121			for commission in new_commissions:
122			perf_tracker.process_commission(commission)
123
124			handle_data(algo, current_data, dt_to_use)
125
126			# grab any new orders from the blotter, then clear the list.
127			# this includes cancelled orders.
128			new_orders = blotter.new_orders
129			blotter.new_orders = []
130
131			# if we have any new orders, record them so that we know
132			# in what perf period they were placed.
133			if new_orders:
134			for new_order in new_orders:
135			perf_tracker.process_order(new_order)
136
137			def once_a_day(midnight_dt):
138			# set all the timestamps
139			self.simulation_dt = midnight_dt
140			algo.on_dt_changed(midnight_dt)
141			data_portal.current_day = midnight_dt
142
143			# call before trading start
144			algo.before_trading_start(current_data)
145
146			perf_tracker = algo.perf_tracker
147
148			# handle any splits that impact any positions or any open orders.
149			sids_we_care_about = \
150			list(set(list(perf_tracker.position_tracker.positions.keys()) +
151			list(algo.blotter.open_orders.keys())))
152
153			if len(sids_we_care_about) > 0:
154			splits = data_portal.get_splits(sids_we_care_about,
155			midnight_dt)
156			if len(splits) > 0:
157			algo.blotter.process_splits(splits)
158			perf_tracker.position_tracker.handle_splits(splits)
159
160			def handle_benchmark(date):
161			algo.perf_tracker.all_benchmark_returns[date] = \
162			self.benchmark_source.get_value(date)
163
164			with self.processor, ZiplineAPI(self.algo):
165			for dt, action in self.clock:
166			if action == BAR:
167			every_bar(dt)
168			elif action == DAY_START:
169			once_a_day(dt)
170			elif action == DAY_END:
171			# End of the day.
172			handle_benchmark(dt)
173			yield self._get_daily_message(dt, algo, algo.perf_tracker)
174			elif action == MINUTE_END:
175			handle_benchmark(dt)
176			minute_msg, daily_msg = \
177			self._get_minute_message(dt, algo, algo.perf_tracker)
178
179			yield minute_msg
180
181			if daily_msg:
182			yield daily_msg
183
184			risk_message = algo.perf_tracker.handle_simulation_end()
185			yield risk_message
186
187			@staticmethod
188			def _get_daily_message(dt, algo, perf_tracker):
189			"""
190			Get a perf message for the given datetime.
191			"""
192			perf_message = perf_tracker.handle_market_close_daily(dt)
193			perf_message['daily_perf']['recorded_vars'] = algo.recorded_vars
194			return perf_message
195
196			@staticmethod
197			def _get_minute_message(dt, algo, perf_tracker):
198			"""
199			Get a perf message for the given datetime.
200			"""
201			rvars = algo.recorded_vars
202
203			minute_message, daily_message = perf_tracker.handle_minute_close(dt)
204			minute_message['minute_perf']['recorded_vars'] = rvars
205
206			if daily_message:
207			daily_message["daily_perf"]["recorded_vars"] = rvars
208
209			return minute_message, daily_message
210

quantopian / zipline

Pull Request — master (#858)

zipline.gens.AlgorithmSimulator.once_a_day() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like