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-03 19:55 UTC

zipline.finance.Blotter.process_trade() F

↳ Parent: zipline.finance.Blotter.get_transactions()

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, 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.max_shares = int(1e+11)

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

        self.current_dt = None

    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)

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

            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']

        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__()

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

quantopian / zipline

Pull Request — master (#858)

zipline.finance.Blotter.process_trade() F

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like