zipline.utils.TradingDayOfWeekRule.date_func() - 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-17 20:37 UTC

zipline.utils.TradingDayOfWeekRule.date_func() A

↳ Parent: zipline.utils.TradingDayOfWeekRule

Complexity

Conditions

Size

Total Lines

Duplication

Lines	0
Ratio	0 %

Metric	Value
cc	1
dl	0
loc	3
rs	10

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


"""
Factory functions to prepare useful data.
"""
import pytz

import pandas as pd
import numpy as np
from datetime import timedelta

from zipline.protocol import Event, DATASOURCE_TYPE
from zipline.sources import (SpecificEquityTrades,
                             DataFrameSource,
                             DataPanelSource)
from zipline.finance.trading import (
    SimulationParameters, TradingEnvironment, noop_load
)
from zipline.sources.test_source import create_trade


# For backwards compatibility
from zipline.data.loader import (load_from_yahoo,
                                 load_bars_from_yahoo)

__all__ = ['load_from_yahoo', 'load_bars_from_yahoo']


def create_simulation_parameters(year=2006, start=None, end=None,
                                 capital_base=float("1.0e5"),
                                 num_days=None,
                                 data_frequency='daily',
                                 emission_rate='daily',
                                 env=None):
    if env is None:
        # Construct a complete environment with reasonable defaults
        env = TradingEnvironment(load=noop_load)
    if start is None:
        start = pd.Timestamp("{0}-01-01".format(year), tz='UTC')
    if end is None:
        if num_days:
            start_index = env.trading_days.searchsorted(start)
            end = env.trading_days[start_index + num_days - 1]
        else:
            end = pd.Timestamp("{0}-12-31".format(year), tz='UTC')
    sim_params = SimulationParameters(
        period_start=start,
        period_end=end,
        capital_base=capital_base,
        data_frequency=data_frequency,
        emission_rate=emission_rate,
        env=env,
    )

    return sim_params


def get_next_trading_dt(current, interval, env):
    next_dt = pd.Timestamp(current).tz_convert(env.exchange_tz)

    while True:
        # Convert timestamp to naive before adding day, otherwise the when
        # stepping over EDT an hour is added.
        next_dt = pd.Timestamp(next_dt.replace(tzinfo=None))
        next_dt = next_dt + interval
        next_dt = pd.Timestamp(next_dt, tz=env.exchange_tz)
        next_dt_utc = next_dt.tz_convert('UTC')
        if env.is_market_hours(next_dt_utc):
            break
        next_dt = next_dt_utc.tz_convert(env.exchange_tz)

    return next_dt_utc


def create_trade_history(sid, prices, amounts, interval, sim_params, env,
                         source_id="test_factory"):
    trades = []
    current = sim_params.first_open

    oneday = timedelta(days=1)
    use_midnight = interval >= oneday
    for price, amount in zip(prices, amounts):
        if use_midnight:
            trade_dt = current.replace(hour=0, minute=0)
        else:
            trade_dt = current
        trade = create_trade(sid, price, amount, trade_dt, source_id)
        trades.append(trade)
        current = get_next_trading_dt(current, interval, env)

    assert len(trades) == len(prices)
    return trades


def create_dividend(sid, payment, declared_date, ex_date, pay_date):
    div = Event({
        'sid': sid,
        'gross_amount': payment,
        'net_amount': payment,
        'payment_sid': None,
        'ratio': None,
        'declared_date': pd.tslib.normalize_date(declared_date),
        'ex_date': pd.tslib.normalize_date(ex_date),
        'pay_date': pd.tslib.normalize_date(pay_date),
        'type': DATASOURCE_TYPE.DIVIDEND,
        'source_id': 'MockDividendSource'
    })
    return div


def create_stock_dividend(sid, payment_sid, ratio, declared_date,
                          ex_date, pay_date):
    return Event({
        'sid': sid,
        'payment_sid': payment_sid,
        'ratio': ratio,
        'net_amount': None,
        'gross_amount': None,
        'dt': pd.tslib.normalize_date(declared_date),
        'ex_date': pd.tslib.normalize_date(ex_date),
        'pay_date': pd.tslib.normalize_date(pay_date),
        'type': DATASOURCE_TYPE.DIVIDEND,
        'source_id': 'MockDividendSource'
    })


def create_split(sid, ratio, date):
    return Event({
        'sid': sid,
        'ratio': ratio,
        'dt': date.replace(hour=0, minute=0, second=0, microsecond=0),
        'type': DATASOURCE_TYPE.SPLIT,
        'source_id': 'MockSplitSource'
    })


def create_txn(sid, price, amount, datetime):
    txn = Event({
        'sid': sid,
        'amount': amount,
        'dt': datetime,
        'price': price,
        'type': DATASOURCE_TYPE.TRANSACTION,
        'source_id': 'MockTransactionSource'
    })
    return txn


def create_commission(sid, value, datetime):
    txn = Event({
        'dt': datetime,
        'type': DATASOURCE_TYPE.COMMISSION,
        'cost': value,
        'sid': sid,
        'source_id': 'MockCommissionSource'
    })
    return txn


def create_txn_history(sid, priceList, amtList, interval, sim_params, env):
    txns = []
    current = sim_params.first_open

    for price, amount in zip(priceList, amtList):
        current = get_next_trading_dt(current, interval, env)

        txns.append(create_txn(sid, price, amount, current))
        current = current + interval
    return txns


def create_returns_from_range(sim_params):
    return pd.Series(index=sim_params.trading_days,
                     data=np.random.rand(len(sim_params.trading_days)))


def create_returns_from_list(returns, sim_params):
    return pd.Series(index=sim_params.trading_days[:len(returns)],
                     data=returns)


def create_daily_trade_source(sids, sim_params, env, concurrent=False):
    """
    creates trade_count trades for each sid in sids list.
    first trade will be on sim_params.period_start, and daily
    thereafter for each sid. Thus, two sids should result in two trades per
    day.
    """
    return create_trade_source(
        sids,
        timedelta(days=1),
        sim_params,
        env=env,
        concurrent=concurrent,
    )


def create_minutely_trade_source(sids, sim_params, env, concurrent=False):
    """
    creates trade_count trades for each sid in sids list.
    first trade will be on sim_params.period_start, and every minute
    thereafter for each sid. Thus, two sids should result in two trades per
    minute.
    """
    return create_trade_source(
        sids,
        timedelta(minutes=1),
        sim_params,
        env=env,
        concurrent=concurrent,
    )


def create_trade_source(sids, trade_time_increment, sim_params, env,
                        concurrent=False):

    # If the sim_params define an end that is during market hours, that will be
    # used as the end of the data source
    if env.is_market_hours(sim_params.period_end):
        end = sim_params.period_end
    # Otherwise, the last_close after the period_end is used as the end of the
    # data source
    else:
        end = sim_params.last_close

    args = tuple()
    kwargs = {
        'sids': sids,
        'start': sim_params.first_open,
        'end': end,
        'delta': trade_time_increment,
        'filter': sids,
        'concurrent': concurrent,
        'env': env,
    }
    source = SpecificEquityTrades(*args, **kwargs)

    return source


def create_test_df_source(sim_params=None, env=None, bars='daily'):
    if bars == 'daily':
        freq = pd.datetools.BDay()
    elif bars == 'minute':
        freq = pd.datetools.Minute()
    else:
        raise ValueError('%s bars not understood.' % bars)

    if sim_params and bars == 'daily':
        index = sim_params.trading_days
    else:
        if env is None:
            env = TradingEnvironment(load=noop_load)

        start = pd.datetime(1990, 1, 3, 0, 0, 0, 0, pytz.utc)
        end = pd.datetime(1990, 1, 8, 0, 0, 0, 0, pytz.utc)

        days = env.days_in_range(start, end)

        if bars == 'daily':
            index = days
        if bars == 'minute':
            index = pd.DatetimeIndex([], freq=freq)

            for day in days:
                day_index = env.market_minutes_for_day(day)
                index = index.append(day_index)

    x = np.arange(1, len(index) + 1)

    df = pd.DataFrame(x, index=index, columns=[0])

    return DataFrameSource(df), df


def create_test_panel_source(sim_params=None, env=None, source_type=None):
    start = sim_params.first_open \
        if sim_params else pd.datetime(1990, 1, 3, 0, 0, 0, 0, pytz.utc)

    end = sim_params.last_close \
        if sim_params else pd.datetime(1990, 1, 8, 0, 0, 0, 0, pytz.utc)

    if env is None:
        env = TradingEnvironment(load=noop_load)

    index = env.days_in_range(start, end)

    price = np.arange(0, len(index))
    volume = np.ones(len(index)) * 1000

    arbitrary = np.ones(len(index))

    df = pd.DataFrame({'price': price,
                       'volume': volume,
                       'arbitrary': arbitrary},
                      index=index)
    if source_type:
        source_types = np.full(len(index), source_type)
        df['type'] = source_types

    panel = pd.Panel.from_dict({0: df})

    return DataPanelSource(panel), panel


def create_test_panel_ohlc_source(sim_params, env):
    start = sim_params.first_open \
        if sim_params else pd.datetime(1990, 1, 3, 0, 0, 0, 0, pytz.utc)

    end = sim_params.last_close \
        if sim_params else pd.datetime(1990, 1, 8, 0, 0, 0, 0, pytz.utc)


    index = env.days_in_range(start, end)
    price = np.arange(0, len(index)) + 100
    high = price * 1.05
    low = price * 0.95
    open_ = price + .1 * (price % 2 - .5)
    volume = np.ones(len(index)) * 1000
    arbitrary = np.ones(len(index))

    df = pd.DataFrame({'price': price,
                       'high': high,
                       'low': low,
                       'open': open_,
                       'volume': volume,
                       'arbitrary': arbitrary},
                      index=index)
    panel = pd.Panel.from_dict({0: df})

    return DataPanelSource(panel), panel


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			"""
18			Factory functions to prepare useful data.
19			"""
20			import pytz
21
22			import pandas as pd
23			import numpy as np
24			from datetime import timedelta
25
26			from zipline.protocol import Event, DATASOURCE_TYPE
27			from zipline.sources import (SpecificEquityTrades,
28			DataFrameSource,
29			DataPanelSource)
30			from zipline.finance.trading import (
31			SimulationParameters, TradingEnvironment, noop_load
32			)
33			from zipline.sources.test_source import create_trade
34
35
36			# For backwards compatibility
37			from zipline.data.loader import (load_from_yahoo,
38			load_bars_from_yahoo)
39
40			__all__ = ['load_from_yahoo', 'load_bars_from_yahoo']
41
42
43			def create_simulation_parameters(year=2006, start=None, end=None,
44			capital_base=float("1.0e5"),
45			num_days=None,
46			data_frequency='daily',
47			emission_rate='daily',
48			env=None):
49			if env is None:
50			# Construct a complete environment with reasonable defaults
51			env = TradingEnvironment(load=noop_load)
52			if start is None:
53			start = pd.Timestamp("{0}-01-01".format(year), tz='UTC')
54			if end is None:
55			if num_days:
56			start_index = env.trading_days.searchsorted(start)
57			end = env.trading_days[start_index + num_days - 1]
58			else:
59			end = pd.Timestamp("{0}-12-31".format(year), tz='UTC')
60			sim_params = SimulationParameters(
61			period_start=start,
62			period_end=end,
63			capital_base=capital_base,
64			data_frequency=data_frequency,
65			emission_rate=emission_rate,
66			env=env,
67			)
68
69			return sim_params
70
71
72			def get_next_trading_dt(current, interval, env):
73			next_dt = pd.Timestamp(current).tz_convert(env.exchange_tz)
74
75			while True:
76			# Convert timestamp to naive before adding day, otherwise the when
77			# stepping over EDT an hour is added.
78			next_dt = pd.Timestamp(next_dt.replace(tzinfo=None))
79			next_dt = next_dt + interval
80			next_dt = pd.Timestamp(next_dt, tz=env.exchange_tz)
81			next_dt_utc = next_dt.tz_convert('UTC')
82			if env.is_market_hours(next_dt_utc):
83			break
84			next_dt = next_dt_utc.tz_convert(env.exchange_tz)
85
86			return next_dt_utc
87
88
89			def create_trade_history(sid, prices, amounts, interval, sim_params, env,
90			source_id="test_factory"):
91			trades = []
92			current = sim_params.first_open
93
94			oneday = timedelta(days=1)
95			use_midnight = interval >= oneday
96			for price, amount in zip(prices, amounts):
97			if use_midnight:
98			trade_dt = current.replace(hour=0, minute=0)
99			else:
100			trade_dt = current
101			trade = create_trade(sid, price, amount, trade_dt, source_id)
102			trades.append(trade)
103			current = get_next_trading_dt(current, interval, env)
104
105			assert len(trades) == len(prices)
106			return trades
107
108
109			def create_dividend(sid, payment, declared_date, ex_date, pay_date):
110			div = Event({
111			'sid': sid,
112			'gross_amount': payment,
113			'net_amount': payment,
114			'payment_sid': None,
115			'ratio': None,
116			'declared_date': pd.tslib.normalize_date(declared_date),
117			'ex_date': pd.tslib.normalize_date(ex_date),
118			'pay_date': pd.tslib.normalize_date(pay_date),
119			'type': DATASOURCE_TYPE.DIVIDEND,
120			'source_id': 'MockDividendSource'
121			})
122			return div
123
124
125			def create_stock_dividend(sid, payment_sid, ratio, declared_date,
126			ex_date, pay_date):
127			return Event({
128			'sid': sid,
129			'payment_sid': payment_sid,
130			'ratio': ratio,
131			'net_amount': None,
132			'gross_amount': None,
133			'dt': pd.tslib.normalize_date(declared_date),
134			'ex_date': pd.tslib.normalize_date(ex_date),
135			'pay_date': pd.tslib.normalize_date(pay_date),
136			'type': DATASOURCE_TYPE.DIVIDEND,
137			'source_id': 'MockDividendSource'
138			})
139
140
141			def create_split(sid, ratio, date):
142			return Event({
143			'sid': sid,
144			'ratio': ratio,
145			'dt': date.replace(hour=0, minute=0, second=0, microsecond=0),
146			'type': DATASOURCE_TYPE.SPLIT,
147			'source_id': 'MockSplitSource'
148			})
149
150
151			def create_txn(sid, price, amount, datetime):
152			txn = Event({
153			'sid': sid,
154			'amount': amount,
155			'dt': datetime,
156			'price': price,
157			'type': DATASOURCE_TYPE.TRANSACTION,
158			'source_id': 'MockTransactionSource'
159			})
160			return txn
161
162
163			def create_commission(sid, value, datetime):
164			txn = Event({
165			'dt': datetime,
166			'type': DATASOURCE_TYPE.COMMISSION,
167			'cost': value,
168			'sid': sid,
169			'source_id': 'MockCommissionSource'
170			})
171			return txn
172
173
174			def create_txn_history(sid, priceList, amtList, interval, sim_params, env):
175			txns = []
176			current = sim_params.first_open
177
178			for price, amount in zip(priceList, amtList):
179			current = get_next_trading_dt(current, interval, env)
180
181			txns.append(create_txn(sid, price, amount, current))
182			current = current + interval
183			return txns
184
185
186			def create_returns_from_range(sim_params):
187			return pd.Series(index=sim_params.trading_days,
188			data=np.random.rand(len(sim_params.trading_days)))
189
190
191			def create_returns_from_list(returns, sim_params):
192			return pd.Series(index=sim_params.trading_days[:len(returns)],
193			data=returns)
194
195
196			def create_daily_trade_source(sids, sim_params, env, concurrent=False):
197			"""
198			creates trade_count trades for each sid in sids list.
199			first trade will be on sim_params.period_start, and daily
200			thereafter for each sid. Thus, two sids should result in two trades per
201			day.
202			"""
203			return create_trade_source(
204			sids,
205			timedelta(days=1),
206			sim_params,
207			env=env,
208			concurrent=concurrent,
209			)
210
211
212			def create_minutely_trade_source(sids, sim_params, env, concurrent=False):
213			"""
214			creates trade_count trades for each sid in sids list.
215			first trade will be on sim_params.period_start, and every minute
216			thereafter for each sid. Thus, two sids should result in two trades per
217			minute.
218			"""
219			return create_trade_source(
220			sids,
221			timedelta(minutes=1),
222			sim_params,
223			env=env,
224			concurrent=concurrent,
225			)
226
227
228			def create_trade_source(sids, trade_time_increment, sim_params, env,
229			concurrent=False):
230
231			# If the sim_params define an end that is during market hours, that will be
232			# used as the end of the data source
233			if env.is_market_hours(sim_params.period_end):
234			end = sim_params.period_end
235			# Otherwise, the last_close after the period_end is used as the end of the
236			# data source
237			else:
238			end = sim_params.last_close
239
240			args = tuple()
241			kwargs = {
242			'sids': sids,
243			'start': sim_params.first_open,
244			'end': end,
245			'delta': trade_time_increment,
246			'filter': sids,
247			'concurrent': concurrent,
248			'env': env,
249			}
250			source = SpecificEquityTrades(args, *kwargs)
251
252			return source
253
254
255			def create_test_df_source(sim_params=None, env=None, bars='daily'):
256			if bars == 'daily':
257			freq = pd.datetools.BDay()
258			elif bars == 'minute':
259			freq = pd.datetools.Minute()
260			else:
261			raise ValueError('%s bars not understood.' % bars)
262
263			if sim_params and bars == 'daily':
264			index = sim_params.trading_days
265			else:
266			if env is None:
267			env = TradingEnvironment(load=noop_load)
268
269			start = pd.datetime(1990, 1, 3, 0, 0, 0, 0, pytz.utc)
270			end = pd.datetime(1990, 1, 8, 0, 0, 0, 0, pytz.utc)
271
272			days = env.days_in_range(start, end)
273
274			if bars == 'daily':
275			index = days
276			if bars == 'minute':
277			index = pd.DatetimeIndex([], freq=freq)
278
279			for day in days:
280			day_index = env.market_minutes_for_day(day)
281			index = index.append(day_index)
282
283			x = np.arange(1, len(index) + 1)
284
285			df = pd.DataFrame(x, index=index, columns=[0])
286
287			return DataFrameSource(df), df
288
289
290			def create_test_panel_source(sim_params=None, env=None, source_type=None):
291			start = sim_params.first_open \
292			if sim_params else pd.datetime(1990, 1, 3, 0, 0, 0, 0, pytz.utc)
293
294			end = sim_params.last_close \
295			if sim_params else pd.datetime(1990, 1, 8, 0, 0, 0, 0, pytz.utc)
296
297			if env is None:
298			env = TradingEnvironment(load=noop_load)
299
300			index = env.days_in_range(start, end)
301
302			price = np.arange(0, len(index))
303			volume = np.ones(len(index)) * 1000
304
305			arbitrary = np.ones(len(index))
306
307			df = pd.DataFrame({'price': price,
308			'volume': volume,
309			'arbitrary': arbitrary},
310			index=index)
311			if source_type:
312			source_types = np.full(len(index), source_type)
313			df['type'] = source_types
314
315			panel = pd.Panel.from_dict({0: df})
316
317			return DataPanelSource(panel), panel
318
319
320			def create_test_panel_ohlc_source(sim_params, env):
321			start = sim_params.first_open \
322			if sim_params else pd.datetime(1990, 1, 3, 0, 0, 0, 0, pytz.utc)
323
324			end = sim_params.last_close \
325			if sim_params else pd.datetime(1990, 1, 8, 0, 0, 0, 0, pytz.utc)
			0 ignored issues – show Duplication introduced 2015-12-17 20:39 UTC by Report Bug Copy Issue Report This code seems to be duplicated in your project. Duplicated code is one of the most pungent code smells. If you need to duplicate the same code in three or more different places, we strongly encourage you to look into extracting the code into a single class or operation. You can also find more detailed suggestions in the “Code” section of your repository. Loading history...
326
327			index = env.days_in_range(start, end)
328			price = np.arange(0, len(index)) + 100
329			high = price * 1.05
330			low = price * 0.95
331			open_ = price + .1 * (price % 2 - .5)
332			volume = np.ones(len(index)) * 1000
333			arbitrary = np.ones(len(index))
334
335			df = pd.DataFrame({'price': price,
336			'high': high,
337			'low': low,
338			'open': open_,
339			'volume': volume,
340			'arbitrary': arbitrary},
341			index=index)
342			panel = pd.Panel.from_dict({0: df})
343
344			return DataPanelSource(panel), panel
345

quantopian / zipline

Pull Request — master (#858)

zipline.utils.TradingDayOfWeekRule.date_func() A

Complexity

Size

Duplication

Duplication Side-by-Side

Filter issues like