tests.BlotterTestCase.test_order_hold() - 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-11 16:10 UTC

tests.BlotterTestCase.test_order_hold() B

↳ Parent: tests.BlotterTestCase

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	4
dl	0
loc	59
rs	8.9847

How to fix Long Method

#
# 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.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
        )

        cls.data_portal = DataPortal(
            cls.env,
            daily_equities_path=path,
            sim_params=cls.sim_params
        )

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

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

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

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

quantopian / zipline

Pull Request — master (#858)

tests.BlotterTestCase.test_order_hold() B

Complexity

Size

Duplication

How to fix Long Method

Long Method

Duplication Side-by-Side

Filter issues like