tests.BlotterTestCase.tearDownClass() - Code Metrics - Inspection of "Q 2.0" - quantopian/zipline - Measure and Improve Code Quality continuously with Scrutinizer

Completed

Pull Request — master (#858)

unknown

created 2015-12-21 19:26 UTC

tests.BlotterTestCase.tearDownClass() A

↳ Parent: tests.BlotterTestCase

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	1
dl	0
loc	5
rs	9.4286

#
# Copyright 2014 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 os
from nose_parameterized import parameterized
from unittest import TestCase
from testfixtures import TempDirectory
import pandas as pd

import zipline.utils.factory as factory

from zipline.finance import trading
from zipline.finance.blotter import Blotter
from zipline.finance.order import ORDER_STATUS
from zipline.finance.execution import (
    LimitOrder,
    MarketOrder,
    StopLimitOrder,
    StopOrder,
)

from zipline.utils.test_utils import(
    setup_logger,
    teardown_logger,
)
from zipline.finance.slippage import DEFAULT_VOLUME_SLIPPAGE_BAR_LIMIT, \
    FixedSlippage
from .utils.daily_bar_writer import DailyBarWriterFromDataFrames
from zipline.data.us_equity_pricing import BcolzDailyBarReader
from zipline.data.data_portal import DataPortal


class BlotterTestCase(TestCase):

    @classmethod
    def setUpClass(cls):
        setup_logger(cls)
        cls.env = trading.TradingEnvironment()

        cls.sim_params = factory.create_simulation_parameters(
            start=pd.Timestamp("2006-01-05", tz='UTC'),
            end=pd.Timestamp("2006-01-06", tz='UTC')
        )

        cls.env.write_data(equities_data={
            24: {
                'start_date': cls.sim_params.trading_days[0],
                'end_date': cls.sim_params.trading_days[-1]
            }
        })

        cls.tempdir = TempDirectory()

        assets = {
            24: pd.DataFrame({
                "open": [50, 50],
                "high": [50, 50],
                "low": [50, 50],
                "close": [50, 50],
                "volume": [100, 400],
                "day": [day.value for day in cls.sim_params.trading_days]
            })
        }

        path = os.path.join(cls.tempdir.path, "tempdata.bcolz")

        DailyBarWriterFromDataFrames(assets).write(
            path,
            cls.sim_params.trading_days,
            assets
        )

        equity_daily_reader = BcolzDailyBarReader(path)

        cls.data_portal = DataPortal(
            cls.env,
            equity_daily_reader=equity_daily_reader,
        )

    @classmethod
    def tearDownClass(cls):
        del cls.env
        cls.tempdir.cleanup()
        teardown_logger(cls)

    @parameterized.expand([(MarketOrder(), None, None),
                           (LimitOrder(10), 10, None),
                           (StopOrder(10), None, 10),
                           (StopLimitOrder(10, 20), 10, 20)])
    def test_blotter_order_types(self, style_obj, expected_lmt, expected_stp):

        blotter = Blotter('daily')

        blotter.order(24, 100, style_obj)
        result = blotter.open_orders[24][0]

        self.assertEqual(result.limit, expected_lmt)
        self.assertEqual(result.stop, expected_stp)

    def test_order_rejection(self):
        blotter = Blotter(self.sim_params.data_frequency)

        # Reject a nonexistent order -> no order appears in new_order,
        # no exceptions raised out
        blotter.reject(56)
        self.assertEqual(blotter.new_orders, [])

        # Basic tests of open order behavior
        open_order_id = blotter.order(24, 100, MarketOrder())
        second_order_id = blotter.order(24, 50, MarketOrder())
        self.assertEqual(len(blotter.open_orders[24]), 2)
        open_order = blotter.open_orders[24][0]
        self.assertEqual(open_order.status, ORDER_STATUS.OPEN)
        self.assertEqual(open_order.id, open_order_id)
        self.assertIn(open_order, blotter.new_orders)

        # Reject that order immediately (same bar, i.e. still in new_orders)
        blotter.reject(open_order_id)
        self.assertEqual(len(blotter.new_orders), 2)
        self.assertEqual(len(blotter.open_orders[24]), 1)
        still_open_order = blotter.new_orders[0]
        self.assertEqual(still_open_order.id, second_order_id)
        self.assertEqual(still_open_order.status, ORDER_STATUS.OPEN)
        rejected_order = blotter.new_orders[1]
        self.assertEqual(rejected_order.status, ORDER_STATUS.REJECTED)
        self.assertEqual(rejected_order.reason, '')

        # Do it again, but reject it at a later time (after tradesimulation
        # pulls it from new_orders)
        blotter = Blotter(self.sim_params.data_frequency)

        new_open_id = blotter.order(24, 10, MarketOrder())
        new_open_order = blotter.open_orders[24][0]
        self.assertEqual(new_open_id, new_open_order.id)
        # Pretend that the trade simulation did this.
        blotter.new_orders = []

        rejection_reason = "Not enough cash on hand."
        blotter.reject(new_open_id, reason=rejection_reason)
        rejected_order = blotter.new_orders[0]
        self.assertEqual(rejected_order.id, new_open_id)
        self.assertEqual(rejected_order.status, ORDER_STATUS.REJECTED)
        self.assertEqual(rejected_order.reason, rejection_reason)

        # You can't reject a filled order.
        # Reset for paranoia
        blotter = Blotter(self.sim_params.data_frequency)
        blotter.slippage_func = FixedSlippage()
        filled_id = blotter.order(24, 100, MarketOrder())
        filled_order = None
        blotter.current_dt = self.sim_params.trading_days[-1]
        txns, _ = blotter.get_transactions(self.data_portal)
        for txn in txns:
            filled_order = blotter.orders[txn.order_id]

        self.assertEqual(filled_order.id, filled_id)
        self.assertIn(filled_order, blotter.new_orders)
        self.assertEqual(filled_order.status, ORDER_STATUS.FILLED)
        self.assertNotIn(filled_order, blotter.open_orders[24])

        blotter.reject(filled_id)
        updated_order = blotter.orders[filled_id]
        self.assertEqual(updated_order.status, ORDER_STATUS.FILLED)

    def test_order_hold(self):
        """
        Held orders act almost identically to open orders, except for the
        status indication. When a fill happens, the order should switch
        status to OPEN/FILLED as necessary
        """
        blotter = Blotter(self.sim_params.data_frequency)
        # Nothing happens on held of a non-existent order
        blotter.hold(56)
        self.assertEqual(blotter.new_orders, [])

        open_id = blotter.order(24, 100, MarketOrder())
        open_order = blotter.open_orders[24][0]
        self.assertEqual(open_order.id, open_id)

        blotter.hold(open_id)
        self.assertEqual(len(blotter.new_orders), 1)
        self.assertEqual(len(blotter.open_orders[24]), 1)
        held_order = blotter.new_orders[0]
        self.assertEqual(held_order.status, ORDER_STATUS.HELD)
        self.assertEqual(held_order.reason, '')

        blotter.cancel(held_order.id)
        self.assertEqual(len(blotter.new_orders), 1)
        self.assertEqual(len(blotter.open_orders[24]), 0)
        cancelled_order = blotter.new_orders[0]
        self.assertEqual(cancelled_order.id, held_order.id)
        self.assertEqual(cancelled_order.status, ORDER_STATUS.CANCELLED)

        for data in ([100, self.sim_params.trading_days[0]],
                     [400, self.sim_params.trading_days[1]]):
            # Verify that incoming fills will change the order status.
            trade_amt = data[0]
            dt = data[1]

            order_size = 100
            expected_filled = int(trade_amt *
                                  DEFAULT_VOLUME_SLIPPAGE_BAR_LIMIT)
            expected_open = order_size - expected_filled
            expected_status = ORDER_STATUS.OPEN if expected_open else \
                ORDER_STATUS.FILLED

            blotter = Blotter(self.sim_params.data_frequency)
            open_id = blotter.order(24, order_size, MarketOrder())
            open_order = blotter.open_orders[24][0]
            self.assertEqual(open_id, open_order.id)
            blotter.hold(open_id)
            held_order = blotter.new_orders[0]

            filled_order = None
            blotter.current_dt = dt
            txns, _ = blotter.get_transactions(self.data_portal)
            for txn in txns:
                filled_order = blotter.orders[txn.order_id]

            self.assertEqual(filled_order.id, held_order.id)
            self.assertEqual(filled_order.status, expected_status)
            self.assertEqual(filled_order.filled, expected_filled)
            self.assertEqual(filled_order.open_amount, expected_open)


1			#
2			# Copyright 2014 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
16			import os
17			from nose_parameterized import parameterized
18			from unittest import TestCase
19			from testfixtures import TempDirectory
20			import pandas as pd
21
22			import zipline.utils.factory as factory
23
24			from zipline.finance import trading
25			from zipline.finance.blotter import Blotter
26			from zipline.finance.order import ORDER_STATUS
27			from zipline.finance.execution import (
28			LimitOrder,
29			MarketOrder,
30			StopLimitOrder,
31			StopOrder,
32			)
33
34			from zipline.utils.test_utils import(
35			setup_logger,
36			teardown_logger,
37			)
38			from zipline.finance.slippage import DEFAULT_VOLUME_SLIPPAGE_BAR_LIMIT, \
39			FixedSlippage
40			from .utils.daily_bar_writer import DailyBarWriterFromDataFrames
41			from zipline.data.us_equity_pricing import BcolzDailyBarReader
42			from zipline.data.data_portal import DataPortal
43
44
45			class BlotterTestCase(TestCase):
46
47			@classmethod
48			def setUpClass(cls):
49			setup_logger(cls)
50			cls.env = trading.TradingEnvironment()
51
52			cls.sim_params = factory.create_simulation_parameters(
53			start=pd.Timestamp("2006-01-05", tz='UTC'),
54			end=pd.Timestamp("2006-01-06", tz='UTC')
55			)
56
57			cls.env.write_data(equities_data={
58			24: {
59			'start_date': cls.sim_params.trading_days[0],
60			'end_date': cls.sim_params.trading_days[-1]
61			}
62			})
63
64			cls.tempdir = TempDirectory()
65
66			assets = {
67			24: pd.DataFrame({
68			"open": [50, 50],
69			"high": [50, 50],
70			"low": [50, 50],
71			"close": [50, 50],
72			"volume": [100, 400],
73			"day": [day.value for day in cls.sim_params.trading_days]
74			})
75			}
76
77			path = os.path.join(cls.tempdir.path, "tempdata.bcolz")
78
79			DailyBarWriterFromDataFrames(assets).write(
80			path,
81			cls.sim_params.trading_days,
82			assets
83			)
84
85			equity_daily_reader = BcolzDailyBarReader(path)
86
87			cls.data_portal = DataPortal(
88			cls.env,
89			equity_daily_reader=equity_daily_reader,
90			)
91
92			@classmethod
93			def tearDownClass(cls):
94			del cls.env
95			cls.tempdir.cleanup()
96			teardown_logger(cls)
97
98			@parameterized.expand([(MarketOrder(), None, None),
99			(LimitOrder(10), 10, None),
100			(StopOrder(10), None, 10),
101			(StopLimitOrder(10, 20), 10, 20)])
102			def test_blotter_order_types(self, style_obj, expected_lmt, expected_stp):
103
104			blotter = Blotter('daily')
105
106			blotter.order(24, 100, style_obj)
107			result = blotter.open_orders[24][0]
108
109			self.assertEqual(result.limit, expected_lmt)
110			self.assertEqual(result.stop, expected_stp)
111
112			def test_order_rejection(self):
113			blotter = Blotter(self.sim_params.data_frequency)
114
115			# Reject a nonexistent order -> no order appears in new_order,
116			# no exceptions raised out
117			blotter.reject(56)
118			self.assertEqual(blotter.new_orders, [])
119
120			# Basic tests of open order behavior
121			open_order_id = blotter.order(24, 100, MarketOrder())
122			second_order_id = blotter.order(24, 50, MarketOrder())
123			self.assertEqual(len(blotter.open_orders[24]), 2)
124			open_order = blotter.open_orders[24][0]
125			self.assertEqual(open_order.status, ORDER_STATUS.OPEN)
126			self.assertEqual(open_order.id, open_order_id)
127			self.assertIn(open_order, blotter.new_orders)
128
129			# Reject that order immediately (same bar, i.e. still in new_orders)
130			blotter.reject(open_order_id)
131			self.assertEqual(len(blotter.new_orders), 2)
132			self.assertEqual(len(blotter.open_orders[24]), 1)
133			still_open_order = blotter.new_orders[0]
134			self.assertEqual(still_open_order.id, second_order_id)
135			self.assertEqual(still_open_order.status, ORDER_STATUS.OPEN)
136			rejected_order = blotter.new_orders[1]
137			self.assertEqual(rejected_order.status, ORDER_STATUS.REJECTED)
138			self.assertEqual(rejected_order.reason, '')
139
140			# Do it again, but reject it at a later time (after tradesimulation
141			# pulls it from new_orders)
142			blotter = Blotter(self.sim_params.data_frequency)
143
144			new_open_id = blotter.order(24, 10, MarketOrder())
145			new_open_order = blotter.open_orders[24][0]
146			self.assertEqual(new_open_id, new_open_order.id)
147			# Pretend that the trade simulation did this.
148			blotter.new_orders = []
149
150			rejection_reason = "Not enough cash on hand."
151			blotter.reject(new_open_id, reason=rejection_reason)
152			rejected_order = blotter.new_orders[0]
153			self.assertEqual(rejected_order.id, new_open_id)
154			self.assertEqual(rejected_order.status, ORDER_STATUS.REJECTED)
155			self.assertEqual(rejected_order.reason, rejection_reason)
156
157			# You can't reject a filled order.
158			# Reset for paranoia
159			blotter = Blotter(self.sim_params.data_frequency)
160			blotter.slippage_func = FixedSlippage()
161			filled_id = blotter.order(24, 100, MarketOrder())
162			filled_order = None
163			blotter.current_dt = self.sim_params.trading_days[-1]
164			txns, _ = blotter.get_transactions(self.data_portal)
165			for txn in txns:
166			filled_order = blotter.orders[txn.order_id]
167
168			self.assertEqual(filled_order.id, filled_id)
169			self.assertIn(filled_order, blotter.new_orders)
170			self.assertEqual(filled_order.status, ORDER_STATUS.FILLED)
171			self.assertNotIn(filled_order, blotter.open_orders[24])
172
173			blotter.reject(filled_id)
174			updated_order = blotter.orders[filled_id]
175			self.assertEqual(updated_order.status, ORDER_STATUS.FILLED)
176
177			def test_order_hold(self):
178			"""
179			Held orders act almost identically to open orders, except for the
180			status indication. When a fill happens, the order should switch
181			status to OPEN/FILLED as necessary
182			"""
183			blotter = Blotter(self.sim_params.data_frequency)
184			# Nothing happens on held of a non-existent order
185			blotter.hold(56)
186			self.assertEqual(blotter.new_orders, [])
187
188			open_id = blotter.order(24, 100, MarketOrder())
189			open_order = blotter.open_orders[24][0]
190			self.assertEqual(open_order.id, open_id)
191
192			blotter.hold(open_id)
193			self.assertEqual(len(blotter.new_orders), 1)
194			self.assertEqual(len(blotter.open_orders[24]), 1)
195			held_order = blotter.new_orders[0]
196			self.assertEqual(held_order.status, ORDER_STATUS.HELD)
197			self.assertEqual(held_order.reason, '')
198
199			blotter.cancel(held_order.id)
200			self.assertEqual(len(blotter.new_orders), 1)
201			self.assertEqual(len(blotter.open_orders[24]), 0)
202			cancelled_order = blotter.new_orders[0]
203			self.assertEqual(cancelled_order.id, held_order.id)
204			self.assertEqual(cancelled_order.status, ORDER_STATUS.CANCELLED)
205
206			for data in ([100, self.sim_params.trading_days[0]],
207			[400, self.sim_params.trading_days[1]]):
208			# Verify that incoming fills will change the order status.
209			trade_amt = data[0]
210			dt = data[1]
211
212			order_size = 100
213			expected_filled = int(trade_amt *
214			DEFAULT_VOLUME_SLIPPAGE_BAR_LIMIT)
215			expected_open = order_size - expected_filled
216			expected_status = ORDER_STATUS.OPEN if expected_open else \
217			ORDER_STATUS.FILLED
218
219			blotter = Blotter(self.sim_params.data_frequency)
220			open_id = blotter.order(24, order_size, MarketOrder())
221			open_order = blotter.open_orders[24][0]
222			self.assertEqual(open_id, open_order.id)
223			blotter.hold(open_id)
224			held_order = blotter.new_orders[0]
225
226			filled_order = None
227			blotter.current_dt = dt
228			txns, _ = blotter.get_transactions(self.data_portal)
229			for txn in txns:
230			filled_order = blotter.orders[txn.order_id]
231
232			self.assertEqual(filled_order.id, held_order.id)
233			self.assertEqual(filled_order.status, expected_status)
234			self.assertEqual(filled_order.filled, expected_filled)
235			self.assertEqual(filled_order.open_amount, expected_open)
236

quantopian / zipline

Pull Request — master (#858)

tests.BlotterTestCase.tearDownClass() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like