tests.FinanceTestCase.setUp() - Code Metrics - quantopian/zipline - Measure and Improve Code Quality continuously with Scrutinizer

tests.FinanceTestCase.setUp() A
last analyzed 2016-01-08 22:03 UTC

↳ Parent: tests.FinanceTestCase

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	1
dl	0
loc	6
rs	9.4286

#
# Copyright 2013 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.

"""
Tests for the zipline.finance package
"""
import itertools
import operator

import pytz

from unittest import TestCase
from datetime import datetime, timedelta

import numpy as np

from nose.tools import timed

from six.moves import range

import zipline.protocol
from zipline.protocol import Event, DATASOURCE_TYPE

import zipline.utils.factory as factory
import zipline.utils.simfactory as simfactory

from zipline.finance.blotter import Blotter
from zipline.gens.composites import date_sorted_sources

from zipline.finance.trading import TradingEnvironment
from zipline.finance.execution import MarketOrder, LimitOrder
from zipline.finance.trading import SimulationParameters

from zipline.finance.performance import PerformanceTracker
from zipline.utils.test_utils import(
    setup_logger,
    teardown_logger,
    assert_single_position
)

DEFAULT_TIMEOUT = 15  # seconds
EXTENDED_TIMEOUT = 90

_multiprocess_can_split_ = False


class FinanceTestCase(TestCase):

    @classmethod
    def setUpClass(cls):
        cls.env = TradingEnvironment()
        cls.env.write_data(equities_identifiers=[1, 133])

    @classmethod
    def tearDownClass(cls):
        del cls.env

    def setUp(self):
        self.zipline_test_config = {
            'sid': 133,
        }

        setup_logger(self)

    def tearDown(self):
        teardown_logger(self)

    @timed(DEFAULT_TIMEOUT)
    def test_factory_daily(self):
        sim_params = factory.create_simulation_parameters()
        trade_source = factory.create_daily_trade_source(
            [133],
            sim_params,
            env=self.env,
        )
        prev = None
        for trade in trade_source:
            if prev:
                self.assertTrue(trade.dt > prev.dt)
            prev = trade

    @timed(EXTENDED_TIMEOUT)
    def test_full_zipline(self):
        # provide enough trades to ensure all orders are filled.
        self.zipline_test_config['order_count'] = 100
        # making a small order amount, so that each order is filled
        # in a single transaction, and txn_count == order_count.
        self.zipline_test_config['order_amount'] = 25
        # No transactions can be filled on the first trade, so
        # we have one extra trade to ensure all orders are filled.
        self.zipline_test_config['trade_count'] = 101
        full_zipline = simfactory.create_test_zipline(
            **self.zipline_test_config)
        assert_single_position(self, full_zipline)

    # TODO: write tests for short sales
    # TODO: write a test to do massive buying or shorting.

    @timed(DEFAULT_TIMEOUT)
    def test_partially_filled_orders(self):

        # create a scenario where order size and trade size are equal
        # so that orders must be spread out over several trades.
        params = {
            'trade_count': 360,
            'trade_amount': 100,
            'trade_interval': timedelta(minutes=1),
            'order_count': 2,
            'order_amount': 100,
            'order_interval': timedelta(minutes=1),
            # because we placed an order for 100 shares, and the volume
            # of each trade is 100, the simulator should spread the order
            # into 4 trades of 25 shares per order.
            'expected_txn_count': 8,
            'expected_txn_volume': 2 * 100
        }

        self.transaction_sim(**params)

        # same scenario, but with short sales
        params2 = {
            'trade_count': 360,
            'trade_amount': 100,
            'trade_interval': timedelta(minutes=1),
            'order_count': 2,
            'order_amount': -100,
            'order_interval': timedelta(minutes=1),
            'expected_txn_count': 8,
            'expected_txn_volume': 2 * -100
        }

        self.transaction_sim(**params2)

    @timed(DEFAULT_TIMEOUT)
    def test_collapsing_orders(self):
        # create a scenario where order.amount <<< trade.volume
        # to test that several orders can be covered properly by one trade,
        # but are represented by multiple transactions.
        params1 = {
            'trade_count': 6,
            'trade_amount': 100,
            'trade_interval': timedelta(hours=1),
            'order_count': 24,
            'order_amount': 1,
            'order_interval': timedelta(minutes=1),
            # because we placed an orders totaling less than 25% of one trade
            # the simulator should produce just one transaction.
            'expected_txn_count': 24,
            'expected_txn_volume': 24
        }
        self.transaction_sim(**params1)

        # second verse, same as the first. except short!
        params2 = {
            'trade_count': 6,
            'trade_amount': 100,
            'trade_interval': timedelta(hours=1),
            'order_count': 24,
            'order_amount': -1,
            'order_interval': timedelta(minutes=1),
            'expected_txn_count': 24,
            'expected_txn_volume': -24
        }
        self.transaction_sim(**params2)

        # Runs the collapsed trades over daily trade intervals.
        # Ensuring that our delay works for daily intervals as well.
        params3 = {
            'trade_count': 6,
            'trade_amount': 100,
            'trade_interval': timedelta(days=1),
            'order_count': 24,
            'order_amount': 1,
            'order_interval': timedelta(minutes=1),
            'expected_txn_count': 24,
            'expected_txn_volume': 24
        }
        self.transaction_sim(**params3)

    @timed(DEFAULT_TIMEOUT)
    def test_alternating_long_short(self):
        # create a scenario where we alternate buys and sells
        params1 = {
            'trade_count': int(6.5 * 60 * 4),
            'trade_amount': 100,
            'trade_interval': timedelta(minutes=1),
            'order_count': 4,
            'order_amount': 10,
            'order_interval': timedelta(hours=24),
            'alternate': True,
            'complete_fill': True,
            'expected_txn_count': 4,
            'expected_txn_volume': 0  # equal buys and sells
        }
        self.transaction_sim(**params1)

    def transaction_sim(self, **params):
        """ This is a utility method that asserts expected
        results for conversion of orders to transactions given a
        trade history"""

        trade_count = params['trade_count']
        trade_interval = params['trade_interval']
        order_count = params['order_count']
        order_amount = params['order_amount']
        order_interval = params['order_interval']
        expected_txn_count = params['expected_txn_count']
        expected_txn_volume = params['expected_txn_volume']
        # optional parameters
        # ---------------------
        # if present, alternate between long and short sales
        alternate = params.get('alternate')
        # if present, expect transaction amounts to match orders exactly.
        complete_fill = params.get('complete_fill')

        sid = 1
        sim_params = factory.create_simulation_parameters()
        blotter = Blotter()
        price = [10.1] * trade_count
        volume = [100] * trade_count
        start_date = sim_params.first_open

        generated_trades = factory.create_trade_history(
            sid,
            price,
            volume,
            trade_interval,
            sim_params,
            env=self.env,
        )

        if alternate:
            alternator = -1
        else:
            alternator = 1

        order_date = start_date
        for i in range(order_count):

            blotter.set_date(order_date)
            blotter.order(sid, order_amount * alternator ** i, MarketOrder())

            order_date = order_date + order_interval
            # move after market orders to just after market next
            # market open.
            if order_date.hour >= 21:
                if order_date.minute >= 00:
                    order_date = order_date + timedelta(days=1)
                    order_date = order_date.replace(hour=14, minute=30)

        # there should now be one open order list stored under the sid
        oo = blotter.open_orders
        self.assertEqual(len(oo), 1)
        self.assertTrue(sid in oo)
        order_list = oo[sid][:]  # make copy
        self.assertEqual(order_count, len(order_list))

        for i in range(order_count):
            order = order_list[i]
            self.assertEqual(order.sid, sid)
            self.assertEqual(order.amount, order_amount * alternator ** i)

        tracker = PerformanceTracker(sim_params, env=self.env)

        benchmark_returns = [
            Event({'dt': dt,
                   'returns': ret,
                   'type':
                   zipline.protocol.DATASOURCE_TYPE.BENCHMARK,
                   'source_id': 'benchmarks'})
            for dt, ret in self.env.benchmark_returns.iteritems()
            if dt.date() >= sim_params.period_start.date() and
            dt.date() <= sim_params.period_end.date()
        ]

        generated_events = date_sorted_sources(generated_trades,
                                               benchmark_returns)

        # this approximates the loop inside TradingSimulationClient
        transactions = []
        for dt, events in itertools.groupby(generated_events,
                                            operator.attrgetter('dt')):
            for event in events:
                if event.type == DATASOURCE_TYPE.TRADE:

                    for txn, order in blotter.process_trade(event):
                        transactions.append(txn)
                        tracker.process_transaction(txn)
                elif event.type == DATASOURCE_TYPE.BENCHMARK:
                    tracker.process_benchmark(event)
                elif event.type == DATASOURCE_TYPE.TRADE:
                    tracker.process_trade(event)

        if complete_fill:
            self.assertEqual(len(transactions), len(order_list))

        total_volume = 0
        for i in range(len(transactions)):
            txn = transactions[i]
            total_volume += txn.amount
            if complete_fill:
                order = order_list[i]
                self.assertEqual(order.amount, txn.amount)

        self.assertEqual(total_volume, expected_txn_volume)
        self.assertEqual(len(transactions), expected_txn_count)

        cumulative_pos = tracker.cumulative_performance.positions[sid]
        self.assertEqual(total_volume, cumulative_pos.amount)

        # the open orders should not contain sid.
        oo = blotter.open_orders
        self.assertNotIn(sid, oo, "Entry is removed when no open orders")

    def test_blotter_processes_splits(self):
        sim_params = factory.create_simulation_parameters()
        blotter = Blotter()
        blotter.set_date(sim_params.period_start)

        # set up two open limit orders with very low limit prices,
        # one for sid 1 and one for sid 2
        blotter.order(1, 100, LimitOrder(10))
        blotter.order(2, 100, LimitOrder(10))

        # send in a split for sid 2
        split_event = factory.create_split(2, 0.33333,
                                           sim_params.period_start +
                                           timedelta(days=1))

        blotter.process_split(split_event)

        for sid in [1, 2]:
            order_lists = blotter.open_orders[sid]
            self.assertIsNotNone(order_lists)
            self.assertEqual(1, len(order_lists))

        aapl_order = blotter.open_orders[1][0].to_dict()
        fls_order = blotter.open_orders[2][0].to_dict()

        # make sure the aapl order didn't change
        self.assertEqual(100, aapl_order['amount'])
        self.assertEqual(10, aapl_order['limit'])
        self.assertEqual(1, aapl_order['sid'])

        # make sure the fls order did change
        # to 300 shares at 3.33
        self.assertEqual(300, fls_order['amount'])
        self.assertEqual(3.33, fls_order['limit'])
        self.assertEqual(2, fls_order['sid'])


class TradingEnvironmentTestCase(TestCase):
    """
    Tests for date management utilities in zipline.finance.trading.
    """

    def setUp(self):
        setup_logger(self)

    def tearDown(self):
        teardown_logger(self)

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

    @classmethod
    def tearDownClass(cls):
        del cls.env

    @timed(DEFAULT_TIMEOUT)
    def test_is_trading_day(self):
        # holidays taken from: http://www.nyse.com/press/1191407641943.html
        new_years = datetime(2008, 1, 1, tzinfo=pytz.utc)
        mlk_day = datetime(2008, 1, 21, tzinfo=pytz.utc)
        presidents = datetime(2008, 2, 18, tzinfo=pytz.utc)
        good_friday = datetime(2008, 3, 21, tzinfo=pytz.utc)
        memorial_day = datetime(2008, 5, 26, tzinfo=pytz.utc)
        july_4th = datetime(2008, 7, 4, tzinfo=pytz.utc)
        labor_day = datetime(2008, 9, 1, tzinfo=pytz.utc)
        tgiving = datetime(2008, 11, 27, tzinfo=pytz.utc)
        christmas = datetime(2008, 5, 25, tzinfo=pytz.utc)
        a_saturday = datetime(2008, 8, 2, tzinfo=pytz.utc)
        a_sunday = datetime(2008, 10, 12, tzinfo=pytz.utc)
        holidays = [
            new_years,
            mlk_day,
            presidents,
            good_friday,
            memorial_day,
            july_4th,
            labor_day,
            tgiving,
            christmas,
            a_saturday,
            a_sunday
        ]

        for holiday in holidays:
            self.assertTrue(not self.env.is_trading_day(holiday))

        first_trading_day = datetime(2008, 1, 2, tzinfo=pytz.utc)
        last_trading_day = datetime(2008, 12, 31, tzinfo=pytz.utc)
        workdays = [first_trading_day, last_trading_day]

        for workday in workdays:
            self.assertTrue(self.env.is_trading_day(workday))

    def test_simulation_parameters(self):
        env = SimulationParameters(
            period_start=datetime(2008, 1, 1, tzinfo=pytz.utc),
            period_end=datetime(2008, 12, 31, tzinfo=pytz.utc),
            capital_base=100000,
            env=self.env,
        )

        self.assertTrue(env.last_close.month == 12)
        self.assertTrue(env.last_close.day == 31)

    @timed(DEFAULT_TIMEOUT)
    def test_sim_params_days_in_period(self):

        #     January 2008
        #  Su Mo Tu We Th Fr Sa
        #         1  2  3  4  5
        #   6  7  8  9 10 11 12
        #  13 14 15 16 17 18 19
        #  20 21 22 23 24 25 26
        #  27 28 29 30 31

        params = SimulationParameters(
            period_start=datetime(2007, 12, 31, tzinfo=pytz.utc),
            period_end=datetime(2008, 1, 7, tzinfo=pytz.utc),
            capital_base=100000,
            env=self.env,
        )

        expected_trading_days = (
            datetime(2007, 12, 31, tzinfo=pytz.utc),
            # Skip new years
            # holidays taken from: http://www.nyse.com/press/1191407641943.html
            datetime(2008, 1, 2, tzinfo=pytz.utc),
            datetime(2008, 1, 3, tzinfo=pytz.utc),
            datetime(2008, 1, 4, tzinfo=pytz.utc),
            # Skip Saturday
            # Skip Sunday
            datetime(2008, 1, 7, tzinfo=pytz.utc)
        )

        num_expected_trading_days = 5
        self.assertEquals(num_expected_trading_days, params.days_in_period)
        np.testing.assert_array_equal(expected_trading_days,
                                      params.trading_days.tolist())

    @timed(DEFAULT_TIMEOUT)
    def test_market_minute_window(self):

        #     January 2008
        #  Su Mo Tu We Th Fr Sa
        #         1  2  3  4  5
        #   6  7  8  9 10 11 12
        #  13 14 15 16 17 18 19
        #  20 21 22 23 24 25 26
        #  27 28 29 30 31

        us_east = pytz.timezone('US/Eastern')
        utc = pytz.utc

        # 10:01 AM Eastern on January 7th..
        start = us_east.localize(datetime(2008, 1, 7, 10, 1))
        utc_start = start.astimezone(utc)

        # Get the next 10 minutes
        minutes = self.env.market_minute_window(
            utc_start, 10,
        )
        self.assertEqual(len(minutes), 10)
        for i in range(10):
            self.assertEqual(minutes[i], utc_start + timedelta(minutes=i))

        # Get the previous 10 minutes.
        minutes = self.env.market_minute_window(
            utc_start, 10, step=-1,
        )
        self.assertEqual(len(minutes), 10)
        for i in range(10):
            self.assertEqual(minutes[i], utc_start + timedelta(minutes=-i))

        # Get the next 900 minutes, including utc_start, rolling over into the
        # next two days.
        # Should include:
        # Today:    10:01 AM  ->  4:00 PM  (360 minutes)
        # Tomorrow: 9:31  AM  ->  4:00 PM  (390 minutes, 750 total)
        # Last Day: 9:31  AM  -> 12:00 PM  (150 minutes, 900 total)
        minutes = self.env.market_minute_window(
            utc_start, 900,
        )
        today = self.env.market_minutes_for_day(start)[30:]
        tomorrow = self.env.market_minutes_for_day(
            start + timedelta(days=1)
        )
        last_day = self.env.market_minutes_for_day(
            start + timedelta(days=2))[:150]

        self.assertEqual(len(minutes), 900)
        self.assertEqual(minutes[0], utc_start)
        self.assertTrue(all(today == minutes[:360]))
        self.assertTrue(all(tomorrow == minutes[360:750]))
        self.assertTrue(all(last_day == minutes[750:]))

        # Get the previous 801 minutes, including utc_start, rolling over into
        # Friday the 4th and Thursday the 3rd.
        # Should include:
        # Today:    10:01 AM -> 9:31 AM (31 minutes)
        # Friday:   4:00 PM  -> 9:31 AM (390 minutes, 421 total)
        # Thursday: 4:00 PM  -> 9:41 AM (380 minutes, 801 total)
        minutes = self.env.market_minute_window(
            utc_start, 801, step=-1,
        )

        today = self.env.market_minutes_for_day(start)[30::-1]
        # minus an extra two days from each of these to account for the two
        # weekend days we skipped
        friday = self.env.market_minutes_for_day(
            start + timedelta(days=-3),
        )[::-1]
        thursday = self.env.market_minutes_for_day(
            start + timedelta(days=-4),
        )[:9:-1]

        self.assertEqual(len(minutes), 801)
        self.assertEqual(minutes[0], utc_start)
        self.assertTrue(all(today == minutes[:31]))
        self.assertTrue(all(friday == minutes[31:421]))
        self.assertTrue(all(thursday == minutes[421:]))

    def test_max_date(self):
        max_date = datetime(2008, 8, 1, tzinfo=pytz.utc)
        env = TradingEnvironment(max_date=max_date)

        self.assertLessEqual(env.last_trading_day, max_date)
        self.assertLessEqual(env.treasury_curves.index[-1],
                             max_date)


1			#
2			# Copyright 2013 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			"""
17			Tests for the zipline.finance package
18			"""
19			import itertools
20			import operator
21
22			import pytz
23
24			from unittest import TestCase
25			from datetime import datetime, timedelta
26
27			import numpy as np
28
29			from nose.tools import timed
30
31			from six.moves import range
32
33			import zipline.protocol
34			from zipline.protocol import Event, DATASOURCE_TYPE
35
36			import zipline.utils.factory as factory
37			import zipline.utils.simfactory as simfactory
38
39			from zipline.finance.blotter import Blotter
40			from zipline.gens.composites import date_sorted_sources
41
42			from zipline.finance.trading import TradingEnvironment
43			from zipline.finance.execution import MarketOrder, LimitOrder
44			from zipline.finance.trading import SimulationParameters
45
46			from zipline.finance.performance import PerformanceTracker
47			from zipline.utils.test_utils import(
48			setup_logger,
49			teardown_logger,
50			assert_single_position
51			)
52
53			DEFAULT_TIMEOUT = 15 # seconds
54			EXTENDED_TIMEOUT = 90
55
56			_multiprocess_can_split_ = False
57
58
59			class FinanceTestCase(TestCase):
60
61			@classmethod
62			def setUpClass(cls):
63			cls.env = TradingEnvironment()
64			cls.env.write_data(equities_identifiers=[1, 133])
65
66			@classmethod
67			def tearDownClass(cls):
68			del cls.env
69
70			def setUp(self):
71			self.zipline_test_config = {
72			'sid': 133,
73			}
74
75			setup_logger(self)
76
77			def tearDown(self):
78			teardown_logger(self)
79
80			@timed(DEFAULT_TIMEOUT)
81			def test_factory_daily(self):
82			sim_params = factory.create_simulation_parameters()
83			trade_source = factory.create_daily_trade_source(
84			[133],
85			sim_params,
86			env=self.env,
87			)
88			prev = None
89			for trade in trade_source:
90			if prev:
91			self.assertTrue(trade.dt > prev.dt)
92			prev = trade
93
94			@timed(EXTENDED_TIMEOUT)
95			def test_full_zipline(self):
96			# provide enough trades to ensure all orders are filled.
97			self.zipline_test_config['order_count'] = 100
98			# making a small order amount, so that each order is filled
99			# in a single transaction, and txn_count == order_count.
100			self.zipline_test_config['order_amount'] = 25
101			# No transactions can be filled on the first trade, so
102			# we have one extra trade to ensure all orders are filled.
103			self.zipline_test_config['trade_count'] = 101
104			full_zipline = simfactory.create_test_zipline(
105			**self.zipline_test_config)
106			assert_single_position(self, full_zipline)
107
108			# TODO: write tests for short sales
109			# TODO: write a test to do massive buying or shorting.
110
111			@timed(DEFAULT_TIMEOUT)
112			def test_partially_filled_orders(self):
113
114			# create a scenario where order size and trade size are equal
115			# so that orders must be spread out over several trades.
116			params = {
117			'trade_count': 360,
118			'trade_amount': 100,
119			'trade_interval': timedelta(minutes=1),
120			'order_count': 2,
121			'order_amount': 100,
122			'order_interval': timedelta(minutes=1),
123			# because we placed an order for 100 shares, and the volume
124			# of each trade is 100, the simulator should spread the order
125			# into 4 trades of 25 shares per order.
126			'expected_txn_count': 8,
127			'expected_txn_volume': 2 * 100
128			}
129
130			self.transaction_sim(**params)
131
132			# same scenario, but with short sales
133			params2 = {
134			'trade_count': 360,
135			'trade_amount': 100,
136			'trade_interval': timedelta(minutes=1),
137			'order_count': 2,
138			'order_amount': -100,
139			'order_interval': timedelta(minutes=1),
140			'expected_txn_count': 8,
141			'expected_txn_volume': 2 * -100
142			}
143
144			self.transaction_sim(**params2)
145
146			@timed(DEFAULT_TIMEOUT)
147			def test_collapsing_orders(self):
148			# create a scenario where order.amount <<< trade.volume
149			# to test that several orders can be covered properly by one trade,
150			# but are represented by multiple transactions.
151			params1 = {
152			'trade_count': 6,
153			'trade_amount': 100,
154			'trade_interval': timedelta(hours=1),
155			'order_count': 24,
156			'order_amount': 1,
157			'order_interval': timedelta(minutes=1),
158			# because we placed an orders totaling less than 25% of one trade
159			# the simulator should produce just one transaction.
160			'expected_txn_count': 24,
161			'expected_txn_volume': 24
162			}
163			self.transaction_sim(**params1)
164
165			# second verse, same as the first. except short!
166			params2 = {
167			'trade_count': 6,
168			'trade_amount': 100,
169			'trade_interval': timedelta(hours=1),
170			'order_count': 24,
171			'order_amount': -1,
172			'order_interval': timedelta(minutes=1),
173			'expected_txn_count': 24,
174			'expected_txn_volume': -24
175			}
176			self.transaction_sim(**params2)
177
178			# Runs the collapsed trades over daily trade intervals.
179			# Ensuring that our delay works for daily intervals as well.
180			params3 = {
181			'trade_count': 6,
182			'trade_amount': 100,
183			'trade_interval': timedelta(days=1),
184			'order_count': 24,
185			'order_amount': 1,
186			'order_interval': timedelta(minutes=1),
187			'expected_txn_count': 24,
188			'expected_txn_volume': 24
189			}
190			self.transaction_sim(**params3)
191
192			@timed(DEFAULT_TIMEOUT)
193			def test_alternating_long_short(self):
194			# create a scenario where we alternate buys and sells
195			params1 = {
196			'trade_count': int(6.5 * 60 * 4),
197			'trade_amount': 100,
198			'trade_interval': timedelta(minutes=1),
199			'order_count': 4,
200			'order_amount': 10,
201			'order_interval': timedelta(hours=24),
202			'alternate': True,
203			'complete_fill': True,
204			'expected_txn_count': 4,
205			'expected_txn_volume': 0 # equal buys and sells
206			}
207			self.transaction_sim(**params1)
208
209			def transaction_sim(self, **params):
210			""" This is a utility method that asserts expected
211			results for conversion of orders to transactions given a
212			trade history"""
213
214			trade_count = params['trade_count']
215			trade_interval = params['trade_interval']
216			order_count = params['order_count']
217			order_amount = params['order_amount']
218			order_interval = params['order_interval']
219			expected_txn_count = params['expected_txn_count']
220			expected_txn_volume = params['expected_txn_volume']
221			# optional parameters
222			# ---------------------
223			# if present, alternate between long and short sales
224			alternate = params.get('alternate')
225			# if present, expect transaction amounts to match orders exactly.
226			complete_fill = params.get('complete_fill')
227
228			sid = 1
229			sim_params = factory.create_simulation_parameters()
230			blotter = Blotter()
231			price = [10.1] * trade_count
232			volume = [100] * trade_count
233			start_date = sim_params.first_open
234
235			generated_trades = factory.create_trade_history(
236			sid,
237			price,
238			volume,
239			trade_interval,
240			sim_params,
241			env=self.env,
242			)
243
244			if alternate:
245			alternator = -1
246			else:
247			alternator = 1
248
249			order_date = start_date
250			for i in range(order_count):
251
252			blotter.set_date(order_date)
253			blotter.order(sid, order_amount * alternator ** i, MarketOrder())
254
255			order_date = order_date + order_interval
256			# move after market orders to just after market next
257			# market open.
258			if order_date.hour >= 21:
259			if order_date.minute >= 00:
260			order_date = order_date + timedelta(days=1)
261			order_date = order_date.replace(hour=14, minute=30)
262
263			# there should now be one open order list stored under the sid
264			oo = blotter.open_orders
265			self.assertEqual(len(oo), 1)
266			self.assertTrue(sid in oo)
267			order_list = oo[sid][:] # make copy
268			self.assertEqual(order_count, len(order_list))
269
270			for i in range(order_count):
271			order = order_list[i]
272			self.assertEqual(order.sid, sid)
273			self.assertEqual(order.amount, order_amount * alternator ** i)
274
275			tracker = PerformanceTracker(sim_params, env=self.env)
276
277			benchmark_returns = [
278			Event({'dt': dt,
279			'returns': ret,
280			'type':
281			zipline.protocol.DATASOURCE_TYPE.BENCHMARK,
282			'source_id': 'benchmarks'})
283			for dt, ret in self.env.benchmark_returns.iteritems()
284			if dt.date() >= sim_params.period_start.date() and
285			dt.date() <= sim_params.period_end.date()
286			]
287
288			generated_events = date_sorted_sources(generated_trades,
289			benchmark_returns)
290
291			# this approximates the loop inside TradingSimulationClient
292			transactions = []
293			for dt, events in itertools.groupby(generated_events,
294			operator.attrgetter('dt')):
295			for event in events:
296			if event.type == DATASOURCE_TYPE.TRADE:
297
298			for txn, order in blotter.process_trade(event):
299			transactions.append(txn)
300			tracker.process_transaction(txn)
301			elif event.type == DATASOURCE_TYPE.BENCHMARK:
302			tracker.process_benchmark(event)
303			elif event.type == DATASOURCE_TYPE.TRADE:
304			tracker.process_trade(event)
305
306			if complete_fill:
307			self.assertEqual(len(transactions), len(order_list))
308
309			total_volume = 0
310			for i in range(len(transactions)):
311			txn = transactions[i]
312			total_volume += txn.amount
313			if complete_fill:
314			order = order_list[i]
315			self.assertEqual(order.amount, txn.amount)
316
317			self.assertEqual(total_volume, expected_txn_volume)
318			self.assertEqual(len(transactions), expected_txn_count)
319
320			cumulative_pos = tracker.cumulative_performance.positions[sid]
321			self.assertEqual(total_volume, cumulative_pos.amount)
322
323			# the open orders should not contain sid.
324			oo = blotter.open_orders
325			self.assertNotIn(sid, oo, "Entry is removed when no open orders")
326
327			def test_blotter_processes_splits(self):
328			sim_params = factory.create_simulation_parameters()
329			blotter = Blotter()
330			blotter.set_date(sim_params.period_start)
331
332			# set up two open limit orders with very low limit prices,
333			# one for sid 1 and one for sid 2
334			blotter.order(1, 100, LimitOrder(10))
335			blotter.order(2, 100, LimitOrder(10))
336
337			# send in a split for sid 2
338			split_event = factory.create_split(2, 0.33333,
339			sim_params.period_start +
340			timedelta(days=1))
341
342			blotter.process_split(split_event)
343
344			for sid in [1, 2]:
345			order_lists = blotter.open_orders[sid]
346			self.assertIsNotNone(order_lists)
347			self.assertEqual(1, len(order_lists))
348
349			aapl_order = blotter.open_orders[1][0].to_dict()
350			fls_order = blotter.open_orders[2][0].to_dict()
351
352			# make sure the aapl order didn't change
353			self.assertEqual(100, aapl_order['amount'])
354			self.assertEqual(10, aapl_order['limit'])
355			self.assertEqual(1, aapl_order['sid'])
356
357			# make sure the fls order did change
358			# to 300 shares at 3.33
359			self.assertEqual(300, fls_order['amount'])
360			self.assertEqual(3.33, fls_order['limit'])
361			self.assertEqual(2, fls_order['sid'])
362
363
364			class TradingEnvironmentTestCase(TestCase):
365			"""
366			Tests for date management utilities in zipline.finance.trading.
367			"""
368
369			def setUp(self):
370			setup_logger(self)
371
372			def tearDown(self):
373			teardown_logger(self)
374
375			@classmethod
376			def setUpClass(cls):
377			cls.env = TradingEnvironment()
378
379			@classmethod
380			def tearDownClass(cls):
381			del cls.env
382
383			@timed(DEFAULT_TIMEOUT)
384			def test_is_trading_day(self):
385			# holidays taken from: http://www.nyse.com/press/1191407641943.html
386			new_years = datetime(2008, 1, 1, tzinfo=pytz.utc)
387			mlk_day = datetime(2008, 1, 21, tzinfo=pytz.utc)
388			presidents = datetime(2008, 2, 18, tzinfo=pytz.utc)
389			good_friday = datetime(2008, 3, 21, tzinfo=pytz.utc)
390			memorial_day = datetime(2008, 5, 26, tzinfo=pytz.utc)
391			july_4th = datetime(2008, 7, 4, tzinfo=pytz.utc)
392			labor_day = datetime(2008, 9, 1, tzinfo=pytz.utc)
393			tgiving = datetime(2008, 11, 27, tzinfo=pytz.utc)
394			christmas = datetime(2008, 5, 25, tzinfo=pytz.utc)
395			a_saturday = datetime(2008, 8, 2, tzinfo=pytz.utc)
396			a_sunday = datetime(2008, 10, 12, tzinfo=pytz.utc)
397			holidays = [
398			new_years,
399			mlk_day,
400			presidents,
401			good_friday,
402			memorial_day,
403			july_4th,
404			labor_day,
405			tgiving,
406			christmas,
407			a_saturday,
408			a_sunday
409			]
410
411			for holiday in holidays:
412			self.assertTrue(not self.env.is_trading_day(holiday))
413
414			first_trading_day = datetime(2008, 1, 2, tzinfo=pytz.utc)
415			last_trading_day = datetime(2008, 12, 31, tzinfo=pytz.utc)
416			workdays = [first_trading_day, last_trading_day]
417
418			for workday in workdays:
419			self.assertTrue(self.env.is_trading_day(workday))
420
421			def test_simulation_parameters(self):
422			env = SimulationParameters(
423			period_start=datetime(2008, 1, 1, tzinfo=pytz.utc),
424			period_end=datetime(2008, 12, 31, tzinfo=pytz.utc),
425			capital_base=100000,
426			env=self.env,
427			)
428
429			self.assertTrue(env.last_close.month == 12)
430			self.assertTrue(env.last_close.day == 31)
431
432			@timed(DEFAULT_TIMEOUT)
433			def test_sim_params_days_in_period(self):
434
435			# January 2008
436			# Su Mo Tu We Th Fr Sa
437			# 1 2 3 4 5
438			# 6 7 8 9 10 11 12
439			# 13 14 15 16 17 18 19
440			# 20 21 22 23 24 25 26
441			# 27 28 29 30 31
442
443			params = SimulationParameters(
444			period_start=datetime(2007, 12, 31, tzinfo=pytz.utc),
445			period_end=datetime(2008, 1, 7, tzinfo=pytz.utc),
446			capital_base=100000,
447			env=self.env,
448			)
449
450			expected_trading_days = (
451			datetime(2007, 12, 31, tzinfo=pytz.utc),
452			# Skip new years
453			# holidays taken from: http://www.nyse.com/press/1191407641943.html
454			datetime(2008, 1, 2, tzinfo=pytz.utc),
455			datetime(2008, 1, 3, tzinfo=pytz.utc),
456			datetime(2008, 1, 4, tzinfo=pytz.utc),
457			# Skip Saturday
458			# Skip Sunday
459			datetime(2008, 1, 7, tzinfo=pytz.utc)
460			)
461
462			num_expected_trading_days = 5
463			self.assertEquals(num_expected_trading_days, params.days_in_period)
464			np.testing.assert_array_equal(expected_trading_days,
465			params.trading_days.tolist())
466
467			@timed(DEFAULT_TIMEOUT)
468			def test_market_minute_window(self):
469
470			# January 2008
471			# Su Mo Tu We Th Fr Sa
472			# 1 2 3 4 5
473			# 6 7 8 9 10 11 12
474			# 13 14 15 16 17 18 19
475			# 20 21 22 23 24 25 26
476			# 27 28 29 30 31
477
478			us_east = pytz.timezone('US/Eastern')
479			utc = pytz.utc
480
481			# 10:01 AM Eastern on January 7th..
482			start = us_east.localize(datetime(2008, 1, 7, 10, 1))
483			utc_start = start.astimezone(utc)
484
485			# Get the next 10 minutes
486			minutes = self.env.market_minute_window(
487			utc_start, 10,
488			)
489			self.assertEqual(len(minutes), 10)
490			for i in range(10):
491			self.assertEqual(minutes[i], utc_start + timedelta(minutes=i))
492
493			# Get the previous 10 minutes.
494			minutes = self.env.market_minute_window(
495			utc_start, 10, step=-1,
496			)
497			self.assertEqual(len(minutes), 10)
498			for i in range(10):
499			self.assertEqual(minutes[i], utc_start + timedelta(minutes=-i))
500
501			# Get the next 900 minutes, including utc_start, rolling over into the
502			# next two days.
503			# Should include:
504			# Today: 10:01 AM -> 4:00 PM (360 minutes)
505			# Tomorrow: 9:31 AM -> 4:00 PM (390 minutes, 750 total)
506			# Last Day: 9:31 AM -> 12:00 PM (150 minutes, 900 total)
507			minutes = self.env.market_minute_window(
508			utc_start, 900,
509			)
510			today = self.env.market_minutes_for_day(start)[30:]
511			tomorrow = self.env.market_minutes_for_day(
512			start + timedelta(days=1)
513			)
514			last_day = self.env.market_minutes_for_day(
515			start + timedelta(days=2))[:150]
516
517			self.assertEqual(len(minutes), 900)
518			self.assertEqual(minutes[0], utc_start)
519			self.assertTrue(all(today == minutes[:360]))
520			self.assertTrue(all(tomorrow == minutes[360:750]))
521			self.assertTrue(all(last_day == minutes[750:]))
522
523			# Get the previous 801 minutes, including utc_start, rolling over into
524			# Friday the 4th and Thursday the 3rd.
525			# Should include:
526			# Today: 10:01 AM -> 9:31 AM (31 minutes)
527			# Friday: 4:00 PM -> 9:31 AM (390 minutes, 421 total)
528			# Thursday: 4:00 PM -> 9:41 AM (380 minutes, 801 total)
529			minutes = self.env.market_minute_window(
530			utc_start, 801, step=-1,
531			)
532
533			today = self.env.market_minutes_for_day(start)[30::-1]
534			# minus an extra two days from each of these to account for the two
535			# weekend days we skipped
536			friday = self.env.market_minutes_for_day(
537			start + timedelta(days=-3),
538			)[::-1]
539			thursday = self.env.market_minutes_for_day(
540			start + timedelta(days=-4),
541			)[:9:-1]
542
543			self.assertEqual(len(minutes), 801)
544			self.assertEqual(minutes[0], utc_start)
545			self.assertTrue(all(today == minutes[:31]))
546			self.assertTrue(all(friday == minutes[31:421]))
547			self.assertTrue(all(thursday == minutes[421:]))
548
549			def test_max_date(self):
550			max_date = datetime(2008, 8, 1, tzinfo=pytz.utc)
551			env = TradingEnvironment(max_date=max_date)
552
553			self.assertLessEqual(env.last_trading_day, max_date)
554			self.assertLessEqual(env.treasury_curves.index[-1],
555			max_date)
556

quantopian / zipline

tests.FinanceTestCase.setUp() A last analyzed 2016-01-08 22:03 UTC

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like

tests.FinanceTestCase.setUp() A
last analyzed 2016-01-08 22:03 UTC