zipline.finance.Blotter.process_trade() - 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-18 16:20 UTC

zipline.finance.Blotter.process_trade() F

↳ Parent: zipline.finance.Blotter

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	9
dl	0
loc	33
rs	3

#
# 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.
import math

from logbook import Logger
from collections import defaultdict

import pandas as pd
from six import iteritems

from zipline.finance.order import Order

from zipline.finance.slippage import VolumeShareSlippage
from zipline.finance.commission import PerShare

from zipline.utils.serialization_utils import (
    VERSION_LABEL
)

log = Logger('Blotter')


class Blotter(object):
    def __init__(self, data_frequency,
                 slippage_func=None, commission=None):
        # these orders are aggregated by sid
        self.open_orders = defaultdict(list)

        # keep a dict of orders by their own id
        self.orders = {}

        # holding orders that have come in since the last event.
        self.new_orders = []
        self.current_dt = None

        self.max_shares = int(1e+11)

        self.slippage_func = slippage_func or VolumeShareSlippage()
        self.commission = commission or PerShare()

        self.data_frequency = data_frequency

    def __repr__(self):
        return """
{class_name}(
    slippage={slippage_func},
    commission={commission},
    open_orders={open_orders},
    orders={orders},
    new_orders={new_orders},
    current_dt={current_dt})
""".strip().format(class_name=self.__class__.__name__,
                   slippage_func=self.slippage_func,
                   commission=self.commission,
                   open_orders=self.open_orders,
                   orders=self.orders,
                   new_orders=self.new_orders,
                   current_dt=self.current_dt)

    def set_date(self, dt):
        self.current_dt = dt

    def order(self, sid, amount, style, order_id=None):
        # something could be done with amount to further divide
        # between buy by share count OR buy shares up to a dollar amount
        # numeric == share count  AND  "$dollar.cents" == cost amount

        """
        amount > 0 :: Buy/Cover
        amount < 0 :: Sell/Short
        Market order:    order(sid, amount)
        Limit order:     order(sid, amount, style=LimitOrder(limit_price))
        Stop order:      order(sid, amount, style=StopOrder(stop_price))
        StopLimit order: order(sid, amount, style=StopLimitOrder(limit_price,
                               stop_price))
        """
        if amount == 0:
            # Don't bother placing orders for 0 shares.
            return
        elif amount > self.max_shares:
            # Arbitrary limit of 100 billion (US) shares will never be
            # exceeded except by a buggy algorithm.
            raise OverflowError("Can't order more than %d shares" %
                                self.max_shares)

        is_buy = (amount > 0)
        order = Order(
            dt=self.current_dt,
            sid=sid,
            amount=amount,
            stop=style.get_stop_price(is_buy),
            limit=style.get_limit_price(is_buy),
            id=order_id
        )

        self.open_orders[order.sid].append(order)
        self.orders[order.id] = order
        self.new_orders.append(order)

        return order.id

    def cancel(self, order_id):
        if order_id not in self.orders:
            return

        cur_order = self.orders[order_id]

        if cur_order.open:
            order_list = self.open_orders[cur_order.sid]
            if cur_order in order_list:
                order_list.remove(cur_order)

            if cur_order in self.new_orders:
                self.new_orders.remove(cur_order)
            cur_order.cancel()
            cur_order.dt = self.current_dt
            # we want this order's new status to be relayed out
            # along with newly placed orders.
            self.new_orders.append(cur_order)

    def reject(self, order_id, reason=''):
        """
        Mark the given order as 'rejected', which is functionally similar to
        cancelled. The distinction is that rejections are involuntary (and
        usually include a message from a broker indicating why the order was
        rejected) while cancels are typically user-driven.
        """
        if order_id not in self.orders:
            return

        cur_order = self.orders[order_id]

        order_list = self.open_orders[cur_order.sid]
        if cur_order in order_list:
            order_list.remove(cur_order)

        if cur_order in self.new_orders:
            self.new_orders.remove(cur_order)
        cur_order.reject(reason=reason)
        cur_order.dt = self.current_dt
        # we want this order's new status to be relayed out
        # along with newly placed orders.
        self.new_orders.append(cur_order)

    def hold(self, order_id, reason=''):
        """
        Mark the order with order_id as 'held'. Held is functionally similar
        to 'open'. When a fill (full or partial) arrives, the status
        will automatically change back to open/filled as necessary.
        """
        if order_id not in self.orders:
            return

        cur_order = self.orders[order_id]
        if cur_order.open:
            if cur_order in self.new_orders:
                self.new_orders.remove(cur_order)
            cur_order.hold(reason=reason)
            cur_order.dt = self.current_dt
            # we want this order's new status to be relayed out
            # along with newly placed orders.
            self.new_orders.append(cur_order)

    def process_splits(self, splits):
        """
        Processes a list of splits by modifying any open orders as needed.

        Parameters
        ----------
        splits: list
            A list of splits.  Each split is a tuple of (sid, ratio).

        Returns
        -------
        None
        """
        for split in splits:
            sid = split[0]
            if sid not in self.open_orders:
                return

            orders_to_modify = self.open_orders[sid]
            for order in orders_to_modify:
                order.handle_split(split[1])

    def get_transactions(self, data_portal):
        """
        Creates a list of transactions based on the current open orders,
        slippage model, and commission model.

        Parameters
        ----------
        data_portal: zipline.data.DataPortal
            The data portal to use for getting price and volume information
            when calculating slippage.

        Notes
        -----
        This method book-keeps the blotter's open_orders dictionary, so that
         it is accurate by the time we're done processing open orders.

        Returns
        -------
        transactions_list: List
            transactions_list: list of transactions resulting from the current
            open orders.  If there were no open orders, an empty list is
            returned.

        commissions_list: List
            commissions_list: list of commissions resulting from filling the
            open orders.  A commission is an object with "sid" and "cost"
            parameters.  If there are no commission events (because, for
            example, Zipline models the commission cost into the fill price
            of the transaction), then this is None.
        """
        closed_orders = []
        transactions = []

        for asset, asset_orders in iteritems(self.open_orders):
            price = data_portal.get_spot_value(
                asset, 'close', self.current_dt, self.data_frequency)

            volume = data_portal.get_spot_value(
                asset, 'volume', self.current_dt, self.data_frequency)

            for order, txn in self.slippage_func(asset_orders, self.current_dt,
                                                 price, volume):
                direction = math.copysign(1, txn.amount)
                per_share, total_commission = self.commission.calculate(txn)
                txn.price += per_share * direction
                txn.commission = total_commission
                order.filled += txn.amount

                if txn.commission is not None:
                    order.commission = (order.commission or 0.0) + \
                        txn.commission

                txn.dt = pd.Timestamp(txn.dt, tz='UTC')
                order.dt = txn.dt

                transactions.append(txn)

                if not order.open:
                    closed_orders.append(order)

        # remove all closed orders from our open_orders dict
        for order in closed_orders:
            sid = order.sid
            try:
                sid_orders = self.open_orders[sid]
                sid_orders.remove(order)
            except KeyError:
                continue

        # now clear out the sids from our open_orders dict that have
        # zero open orders
        for sid in list(self.open_orders.keys()):
            if len(self.open_orders[sid]) == 0:
                del self.open_orders[sid]

        # FIXME this API doesn't feel right (returning two things here)
        return transactions, None

    def __getstate__(self):

        state_to_save = ['new_orders', 'orders', '_status', 'data_frequency']

        state_dict = {k: self.__dict__[k] for k in state_to_save
                      if k in self.__dict__}

        # Have to handle defaultdicts specially
        state_dict['open_orders'] = dict(self.open_orders)

        STATE_VERSION = 1
        state_dict[VERSION_LABEL] = STATE_VERSION

        return state_dict

    def __setstate__(self, state):

        self.__init__(state.pop('data_frequency'))

        OLDEST_SUPPORTED_STATE = 1
        version = state.pop(VERSION_LABEL)

        if version < OLDEST_SUPPORTED_STATE:
            raise BaseException("Blotter saved is state too old.")

        open_orders = defaultdict(list)
        open_orders.update(state.pop('open_orders'))
        self.open_orders = open_orders

        self.__dict__.update(state)


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			import math
16
17			from logbook import Logger
18			from collections import defaultdict
19
20			import pandas as pd
21			from six import iteritems
22
23			from zipline.finance.order import Order
24
25			from zipline.finance.slippage import VolumeShareSlippage
26			from zipline.finance.commission import PerShare
27
28			from zipline.utils.serialization_utils import (
29			VERSION_LABEL
30			)
31
32			log = Logger('Blotter')
33
34
35			class Blotter(object):
36			def __init__(self, data_frequency,
37			slippage_func=None, commission=None):
38			# these orders are aggregated by sid
39			self.open_orders = defaultdict(list)
40
41			# keep a dict of orders by their own id
42			self.orders = {}
43
44			# holding orders that have come in since the last event.
45			self.new_orders = []
46			self.current_dt = None
47
48			self.max_shares = int(1e+11)
49
50			self.slippage_func = slippage_func or VolumeShareSlippage()
51			self.commission = commission or PerShare()
52
53			self.data_frequency = data_frequency
54
55			def __repr__(self):
56			return """
57			{class_name}(
58			slippage={slippage_func},
59			commission={commission},
60			open_orders={open_orders},
61			orders={orders},
62			new_orders={new_orders},
63			current_dt={current_dt})
64			""".strip().format(class_name=self.__class__.__name__,
65			slippage_func=self.slippage_func,
66			commission=self.commission,
67			open_orders=self.open_orders,
68			orders=self.orders,
69			new_orders=self.new_orders,
70			current_dt=self.current_dt)
71
72			def set_date(self, dt):
73			self.current_dt = dt
74
75			def order(self, sid, amount, style, order_id=None):
76			# something could be done with amount to further divide
77			# between buy by share count OR buy shares up to a dollar amount
78			# numeric == share count AND "$dollar.cents" == cost amount
79
80			"""
81			amount > 0 :: Buy/Cover
82			amount < 0 :: Sell/Short
83			Market order: order(sid, amount)
84			Limit order: order(sid, amount, style=LimitOrder(limit_price))
85			Stop order: order(sid, amount, style=StopOrder(stop_price))
86			StopLimit order: order(sid, amount, style=StopLimitOrder(limit_price,
87			stop_price))
88			"""
89			if amount == 0:
90			# Don't bother placing orders for 0 shares.
91			return
92			elif amount > self.max_shares:
93			# Arbitrary limit of 100 billion (US) shares will never be
94			# exceeded except by a buggy algorithm.
95			raise OverflowError("Can't order more than %d shares" %
96			self.max_shares)
97
98			is_buy = (amount > 0)
99			order = Order(
100			dt=self.current_dt,
101			sid=sid,
102			amount=amount,
103			stop=style.get_stop_price(is_buy),
104			limit=style.get_limit_price(is_buy),
105			id=order_id
106			)
107
108			self.open_orders[order.sid].append(order)
109			self.orders[order.id] = order
110			self.new_orders.append(order)
111
112			return order.id
113
114			def cancel(self, order_id):
115			if order_id not in self.orders:
116			return
117
118			cur_order = self.orders[order_id]
119
120			if cur_order.open:
121			order_list = self.open_orders[cur_order.sid]
122			if cur_order in order_list:
123			order_list.remove(cur_order)
124
125			if cur_order in self.new_orders:
126			self.new_orders.remove(cur_order)
127			cur_order.cancel()
128			cur_order.dt = self.current_dt
129			# we want this order's new status to be relayed out
130			# along with newly placed orders.
131			self.new_orders.append(cur_order)
132
133			def reject(self, order_id, reason=''):
134			"""
135			Mark the given order as 'rejected', which is functionally similar to
136			cancelled. The distinction is that rejections are involuntary (and
137			usually include a message from a broker indicating why the order was
138			rejected) while cancels are typically user-driven.
139			"""
140			if order_id not in self.orders:
141			return
142
143			cur_order = self.orders[order_id]
144
145			order_list = self.open_orders[cur_order.sid]
146			if cur_order in order_list:
147			order_list.remove(cur_order)
148
149			if cur_order in self.new_orders:
150			self.new_orders.remove(cur_order)
151			cur_order.reject(reason=reason)
152			cur_order.dt = self.current_dt
153			# we want this order's new status to be relayed out
154			# along with newly placed orders.
155			self.new_orders.append(cur_order)
156
157			def hold(self, order_id, reason=''):
158			"""
159			Mark the order with order_id as 'held'. Held is functionally similar
160			to 'open'. When a fill (full or partial) arrives, the status
161			will automatically change back to open/filled as necessary.
162			"""
163			if order_id not in self.orders:
164			return
165
166			cur_order = self.orders[order_id]
167			if cur_order.open:
168			if cur_order in self.new_orders:
169			self.new_orders.remove(cur_order)
170			cur_order.hold(reason=reason)
171			cur_order.dt = self.current_dt
172			# we want this order's new status to be relayed out
173			# along with newly placed orders.
174			self.new_orders.append(cur_order)
175
176			def process_splits(self, splits):
177			"""
178			Processes a list of splits by modifying any open orders as needed.
179
180			Parameters
181			----------
182			splits: list
183			A list of splits. Each split is a tuple of (sid, ratio).
184
185			Returns
186			-------
187			None
188			"""
189			for split in splits:
190			sid = split[0]
191			if sid not in self.open_orders:
192			return
193
194			orders_to_modify = self.open_orders[sid]
195			for order in orders_to_modify:
196			order.handle_split(split[1])
197
198			def get_transactions(self, data_portal):
199			"""
200			Creates a list of transactions based on the current open orders,
201			slippage model, and commission model.
202
203			Parameters
204			----------
205			data_portal: zipline.data.DataPortal
206			The data portal to use for getting price and volume information
207			when calculating slippage.
208
209			Notes
210			-----
211			This method book-keeps the blotter's open_orders dictionary, so that
212			it is accurate by the time we're done processing open orders.
213
214			Returns
215			-------
216			transactions_list: List
217			transactions_list: list of transactions resulting from the current
218			open orders. If there were no open orders, an empty list is
219			returned.
220
221			commissions_list: List
222			commissions_list: list of commissions resulting from filling the
223			open orders. A commission is an object with "sid" and "cost"
224			parameters. If there are no commission events (because, for
225			example, Zipline models the commission cost into the fill price
226			of the transaction), then this is None.
227			"""
228			closed_orders = []
229			transactions = []
230
231			for asset, asset_orders in iteritems(self.open_orders):
232			price = data_portal.get_spot_value(
233			asset, 'close', self.current_dt, self.data_frequency)
234
235			volume = data_portal.get_spot_value(
236			asset, 'volume', self.current_dt, self.data_frequency)
237
238			for order, txn in self.slippage_func(asset_orders, self.current_dt,
239			price, volume):
240			direction = math.copysign(1, txn.amount)
241			per_share, total_commission = self.commission.calculate(txn)
242			txn.price += per_share * direction
243			txn.commission = total_commission
244			order.filled += txn.amount
245
246			if txn.commission is not None:
247			order.commission = (order.commission or 0.0) + \
248			txn.commission
249
250			txn.dt = pd.Timestamp(txn.dt, tz='UTC')
251			order.dt = txn.dt
252
253			transactions.append(txn)
254
255			if not order.open:
256			closed_orders.append(order)
257
258			# remove all closed orders from our open_orders dict
259			for order in closed_orders:
260			sid = order.sid
261			try:
262			sid_orders = self.open_orders[sid]
263			sid_orders.remove(order)
264			except KeyError:
265			continue
266
267			# now clear out the sids from our open_orders dict that have
268			# zero open orders
269			for sid in list(self.open_orders.keys()):
270			if len(self.open_orders[sid]) == 0:
271			del self.open_orders[sid]
272
273			# FIXME this API doesn't feel right (returning two things here)
274			return transactions, None
275
276			def __getstate__(self):
277
278			state_to_save = ['new_orders', 'orders', '_status', 'data_frequency']
279
280			state_dict = {k: self.__dict__[k] for k in state_to_save
281			if k in self.__dict__}
282
283			# Have to handle defaultdicts specially
284			state_dict['open_orders'] = dict(self.open_orders)
285
286			STATE_VERSION = 1
287			state_dict[VERSION_LABEL] = STATE_VERSION
288
289			return state_dict
290
291			def __setstate__(self, state):
292
293			self.__init__(state.pop('data_frequency'))
294
295			OLDEST_SUPPORTED_STATE = 1
296			version = state.pop(VERSION_LABEL)
297
298			if version < OLDEST_SUPPORTED_STATE:
299			raise BaseException("Blotter saved is state too old.")
300
301			open_orders = defaultdict(list)
302			open_orders.update(state.pop('open_orders'))
303			self.open_orders = open_orders
304
305			self.__dict__.update(state)
306

quantopian / zipline

Pull Request — master (#858)

zipline.finance.Blotter.process_trade() F

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like