wassname/catalyst
1
0
Fork 0
mirror of https://github.com/wassname/catalyst.git synced 2026-09-12 12:12:04 +08:00
Code Issues Packages Projects Releases Wiki Activity

Moved the test folder.

Browse Source
This commit is contained in:
Stephen Diehl
2012-05-09 13:34:13 -04:00
parent 0472384253
commit 62ec591d90
13 changed files with 0 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/__init__.py
View File
tests/algorithms.py
+140
View File
@@ -0,0 +1,140 @@
"""
Algorithm Protocol
===================
For a class to be passed as a trading algorithm to the
:py:class:`zipline.lines.SimulatedTrading` zipline
it must follow an implementation protocol. Examples of this algorithm protocol
are provided below.
The algorithm must expose methods:
- initialize: method that takes no args, no returns. Simply called to
enable the algorithm to set any internal state needed.
- get_sid_filter: method that takes no args, and returns a list
of valid sids. List must have a length between 1 and 10. If None is returned
the filter will block all events.
- handle_frame: method that accepts a :py:class:`pandas.Dataframe` of the
current state of the simulation universe. An example frame::
+-----------------+--------------+----------------+--------------------+
| | SID(133) | SID(134) | SID(135) |
+=================+==============+================+====================+
| price | $10.10 | $22.50 | $13.37 |
+-----------------+--------------+----------------+--------------------+
| volume | 10,000 | 5,000 | 50,000 |
+-----------------+--------------+----------------+--------------------+
| mvg_avg_30 | $9.97 | $22.61 | $13.37 |
+-----------------+--------------+----------------+--------------------+
| dt | 6/30/2012 | 6/30/2011 | 6/29/2012 |
+-----------------+--------------+----------------+--------------------+
- set_order: method that accepts a callable. Will be set as the value of the
order method of trading_client. An algorithm can then place orders with a
valid SID and a number of shares::
self.order(SID(133), share_count)
- set_performance: property which can be set equal to the
cumulative_trading_performance property of the trading_client. An
algorithm can then check position information with the
Portfolio object::
self.Portfolio[SID(133)]['cost_basis']
"""
import zipline.protocol as zp
class TestAlgorithm():
"""
This algorithm will send a specified number of orders, to allow unit tests
to verify the orders sent/received, transactions created, and positions
at the close of a simulation.
"""
def __init__(self, sid, amount, order_count):
self.count = order_count
self.sid = sid
self.amount = amount
self.incr = 0
self.done = False
self.order = None
self.frame_count = 0
self.portfolio = None
def initialize(self):
pass
def set_order(self, order_callable):
self.order = order_callable
def set_portfolio(self, portfolio):
self.portfolio = portfolio
def handle_frame(self, frame):
self.frame_count += 1
#place an order for 100 shares of sid
if self.incr < self.count:
self.order(self.sid, self.amount)
self.incr += 1
def get_sid_filter(self):
return [self.sid]
#
class HeavyBuyAlgorithm():
"""
This algorithm will send a specified number of orders, to allow unit tests
to verify the orders sent/received, transactions created, and positions
at the close of a simulation.
"""
def __init__(self, sid, amount):
self.sid = sid
self.amount = amount
self.incr = 0
self.done = False
self.order = None
self.frame_count = 0
self.portfolio = None
def initialize(self):
pass
def set_order(self, order_callable):
self.order = order_callable
def set_portfolio(self, portfolio):
self.portfolio = portfolio
def handle_frame(self, frame):
self.frame_count += 1
#place an order for 100 shares of sid
self.order(self.sid, self.amount)
self.incr += 1
def get_sid_filter(self):
return [self.sid]
class NoopAlgorithm(object):
"""
Dolce fa niente.
"""
def initialize(self):
pass
def set_order(self, order_callable):
pass
def set_portfolio(self, portfolio):
pass
def handle_frame(self, frame):
pass
def get_sid_filter(self):
return None
tests/client.py
+85
View File
@@ -0,0 +1,85 @@
from gevent_zeromq import zmq
import zipline.util as qutil
import zipline.messaging as qmsg
import zipline.protocol as zp
from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_TYPE
from zipline.finance.trading import TradeSimulationClient
class TestClient(qmsg.Component):
def __init__(self):
qmsg.Component.__init__(self)
self.init()
def init(self):
self.received_count = 0
self.prev_dt = None
self.result_streams = []
# Maximum outgoing result streams, really shouldn't ever
# need more than 1.
self.max_outgoing = 5
@property
def get_id(self):
return "TEST_CLIENT"
@property
def get_type(self):
return COMPONENT_TYPE.SINK
def open(self):
self.data_feed = self.connect_result()
def result_stream(self, zmq_socket, context=None):
"""
Asynchronously grab a socket to stream results out on.
"""
ctx = context or zmq.Context.instance()
sock = ctx.socket(zmq.PULL)
sock.bind(zmq_socket)
# Add
self.result_streams.append( sock )
def do_work(self):
socks = dict(self.poll.poll(self.heartbeat_timeout))
if self.control_in in socks and socks[self.control_in] == self.zmq.POLLIN:
msg = self.control_in.recv()
if self.data_feed in socks and socks[self.data_feed] == self.zmq.POLLIN:
msg = self.data_feed.recv()
#logger.info('msg:' + str(msg))
if msg == str(CONTROL_PROTOCOL.DONE):
qutil.LOGGER.info("Client is DONE!")
self.signal_done()
return
self.received_count += 1
try:
event = self.unframe(msg)
# deserialization error
except zp.INVALID_MERGE_FRAME as exc:
return self.signal_exception(exc)
if self.prev_dt != None:
if not event['dt'] >= self.prev_dt:
raise Exception(
"Message out of order: {date} after {prev}".format(
date = event['dt'], prev = self.prev_dt
)
)
else:
self.prev_dt = event.dt
if self.received_count % 100 == 0:
qutil.LOGGER.info("received {n} messages".format(n=self.received_count))
def unframe(self, msg):
return zp.MERGE_UNFRAME(msg)
tests/factory.py
+231
View File
@@ -0,0 +1,231 @@
"""
Factory functions to prepare useful data for tests.
"""
import pytz
import msgpack
import random
from datetime import datetime, timedelta
import zipline.util as qutil
import zipline.finance.risk as risk
import zipline.protocol as zp
from zipline.sources import SpecificEquityTrades, RandomEquityTrades
from zipline.finance.trading import TradingEnvironment
def load_market_data():
fp_bm = open("./zipline/test/benchmark.msgpack", "rb")
bm_list = msgpack.loads(fp_bm.read())
bm_returns = []
for packed_date, returns in bm_list:
event_dt = zp.tuple_to_date(packed_date)
#event_dt = event_dt.replace(
# hour=0,
# minute=0,
# second=0,
# tzinfo=pytz.utc
#)
daily_return = risk.DailyReturn(date=event_dt, returns=returns)
bm_returns.append(daily_return)
bm_returns = sorted(bm_returns, key=lambda(x): x.date)
fp_tr = open("./zipline/test/treasury_curves.msgpack", "rb")
tr_list = msgpack.loads(fp_tr.read())
tr_curves = {}
for packed_date, curve in tr_list:
tr_dt = zp.tuple_to_date(packed_date)
#tr_dt = tr_dt.replace(hour=0, minute=0, second=0, tzinfo=pytz.utc)
tr_curves[tr_dt] = curve
return bm_returns, tr_curves
def create_trading_environment(year=2006):
"""Construct a complete environment with reasonable defaults"""
benchmark_returns, treasury_curves = load_market_data()
start = datetime(year, 1, 1, tzinfo=pytz.utc)
end = datetime(year, 12, 31, tzinfo=pytz.utc)
trading_environment = TradingEnvironment(
benchmark_returns,
treasury_curves,
period_start = start,
period_end = end,
capital_base = 100000.0
)
return trading_environment
def create_trade(sid, price, amount, datetime):
row = zp.namedict({
'source_id' : "test_factory",
'type' : zp.DATASOURCE_TYPE.TRADE,
'sid' : sid,
'dt' : datetime,
'price' : price,
'volume' : amount
})
return row
def get_next_trading_dt(current, interval, trading_calendar):
next = current
while True:
next = next + interval
if trading_calendar.is_market_hours(next):
break
return next
def create_trade_history(sid, prices, amounts, interval, trading_calendar):
trades = []
current = trading_calendar.first_open
for price, amount in zip(prices, amounts):
trade = create_trade(sid, price, amount, current)
trades.append(trade)
current = get_next_trading_dt(current, interval, trading_calendar)
assert len(trades) == len(prices)
return trades
def create_txn(sid, price, amount, datetime, btrid=None):
txn = zp.namedict({
'sid':sid,
'amount':amount,
'dt':datetime,
'price':price,
})
return txn
def create_txn_history(sid, priceList, amtList, interval, trading_calendar):
txns = []
current = trading_calendar.first_open
for price, amount in zip(priceList, amtList):
current = get_next_trading_dt(current, interval, trading_calendar)
txns.append(create_txn(sid, price, amount, current))
current = current + interval
return txns
def create_returns(daycount, trading_calendar):
"""
For the given number of calendar (not trading) days return all the trading
days between start and start + daycount.
"""
test_range = []
current = trading_calendar.first_open
one_day = timedelta(days = 1)
for day in range(daycount):
current = current + one_day
if trading_calendar.is_trading_day(current):
r = risk.DailyReturn(current, random.random())
test_range.append(r)
return test_range
def create_returns_from_range(trading_calendar):
current = trading_calendar.first_open
end = trading_calendar.last_close
one_day = timedelta(days = 1)
test_range = []
while current <= end:
r = risk.DailyReturn(current, random.random())
test_range.append(r)
current = get_next_trading_dt(current, one_day, trading_calendar)
return test_range
def create_returns_from_list(returns, trading_calendar):
current = trading_calendar.first_open
one_day = timedelta(days = 1)
test_range = []
#sometimes the range starts with a non-trading day.
if not trading_calendar.is_trading_day(current):
current = get_next_trading_dt(current, one_day, trading_calendar)
for return_val in returns:
r = risk.DailyReturn(current, return_val)
test_range.append(r)
current = get_next_trading_dt(current, one_day, trading_calendar)
return test_range
def create_random_trade_source(sid, trade_count, trading_environment):
# create the source
source = RandomEquityTrades(sid, "rand-"+str(sid), trade_count)
# make the period_end of trading_environment match
cur = trading_environment.first_open
one_day = timedelta(days = 1)
for i in range(trade_count + 2):
cur = get_next_trading_dt(cur, one_day, trading_environment)
trading_environment.period_end = cur
return source
def create_daily_trade_source(sids, trade_count, trading_environment):
"""
creates trade_count trades for each sid in sids list.
first trade will be on trading_environment.period_start, and daily
thereafter for each sid. Thus, two sids should result in two trades per
day.
Important side-effect: trading_environment.period_end will be modified
to match the day of the final trade.
"""
return create_trade_source(
sids,
trade_count,
timedelta(days=1),
trading_environment
)
def create_minutely_trade_source(sids, trade_count, trading_environment):
"""
creates trade_count trades for each sid in sids list.
first trade will be on trading_environment.period_start, and every minute
thereafter for each sid. Thus, two sids should result in two trades per
minute.
Important side-effect: trading_environment.period_end will be modified
to match the day of the final trade.
"""
return create_trade_source(
sids,
trade_count,
timedelta(minutes=1),
trading_environment
)
def create_trade_source(sids, trade_count, trade_time_increment, trading_environment):
trade_history = []
for sid in sids:
price = [10.1] * trade_count
volume = [100] * trade_count
start_date = trading_environment.first_open
generated_trades = create_trade_history(
sid,
price,
volume,
trade_time_increment,
trading_environment
)
trade_history.extend(generated_trades)
trade_history = sorted(trade_history, key=lambda(x): x.dt)
#set the trading environment's end to same dt as the last trade in the
#history.
trading_environment.period_end = trade_history[-1].dt
source = SpecificEquityTrades("flat", trade_history)
return source
tests/test_devsimulator.py
View File
tests/test_finance.py
+514
View File
@@ -0,0 +1,514 @@
"""Tests for the zipline.finance package"""
import mock
import pytz
from unittest2 import TestCase
from datetime import datetime, timedelta
from collections import defaultdict
from nose.tools import timed
import zipline.test.factory as factory
import zipline.util as qutil
import zipline.finance.risk as risk
import zipline.protocol as zp
import zipline.finance.performance as perf
from zipline.test.algorithms import TestAlgorithm
from zipline.sources import SpecificEquityTrades
from zipline.finance.trading import TransactionSimulator, \
TradeSimulationClient, TradingEnvironment
from zipline.simulator import AddressAllocator, Simulator
from zipline.monitor import Controller
from zipline.lines import SimulatedTrading
from zipline.finance.performance import PerformanceTracker
from zipline.protocol_utils import namedict
from zipline.finance.trading import SIMULATION_STYLE
DEFAULT_TIMEOUT = 15 # seconds
EXTENDED_TIMEOUT = 90
allocator = AddressAllocator(1000)
class FinanceTestCase(TestCase):
leased_sockets = defaultdict(list)
def setUp(self):
qutil.configure_logging()
self.zipline_test_config = {
'allocator':allocator,
'sid':133
}
@timed(DEFAULT_TIMEOUT)
def test_factory(self):
trading_environment = factory.create_trading_environment()
trade_source = factory.create_daily_trade_source(
[133],
200,
trading_environment
)
prev = None
for trade in trade_source.event_list:
if prev:
self.assertTrue(trade.dt > prev.dt)
prev = trade
@timed(DEFAULT_TIMEOUT)
def test_trading_environment(self):
benchmark_returns, treasury_curves = \
factory.load_market_data()
env = TradingEnvironment(
benchmark_returns,
treasury_curves,
period_start = datetime(2008, 1, 1, tzinfo = pytz.utc),
period_end = datetime(2008, 12, 31, tzinfo = pytz.utc),
capital_base = 100000,
max_drawdown = 0.50
)
#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 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(env.is_trading_day(workday))
self.assertTrue(env.last_close.month == 12)
self.assertTrue(env.last_close.day == 31)
# The following two tests appear broken no that the order source is
# non blocking. HUNCH: The trades are streaming through before the orders
# are placed.
@timed(DEFAULT_TIMEOUT)
def test_orders(self):
# Simulation
# ----------
self.zipline_test_config['simulation_style'] = \
SIMULATION_STYLE.FIXED_SLIPPAGE
zipline = SimulatedTrading.create_test_zipline(
**self.zipline_test_config
)
zipline.simulate(blocking=True)
self.assertTrue(zipline.sim.ready())
self.assertFalse(zipline.sim.exception)
# TODO: Make more assertions about the final state of the components.
self.assertEqual(zipline.sim.feed.pending_messages(), 0, \
"The feed should be drained of all messages, found {n} remaining." \
.format(n=zipline.sim.feed.pending_messages()))
# the trading client should receive one transaction for every
# order placed.
self.assertEqual(
zipline.trading_client.txn_count,
zipline.trading_client.order_count
)
@timed(EXTENDED_TIMEOUT)
def test_aggressive_buying(self):
# Simulation
# ----------
# TODO: for some reason the orders aren't filled without an extra
# trade.
trade_count = 5001
self.zipline_test_config['order_count'] = trade_count - 1
self.zipline_test_config['trade_count'] = trade_count
self.zipline_test_config['order_amount'] = 1
# tell the simulator to fill the orders in individual transactions
# matching the order volume exactly.
self.zipline_test_config['simulation_style'] = \
SIMULATION_STYLE.FIXED_SLIPPAGE
self.zipline_test_config['environment'] = factory.create_trading_environment()
sid_list = [self.zipline_test_config['sid']]
self.zipline_test_config['trade_source'] = factory.create_minutely_trade_source(
sid_list,
trade_count,
self.zipline_test_config['environment']
)
zipline = SimulatedTrading.create_test_zipline(**self.zipline_test_config)
zipline.simulate(blocking=True)
self.assertTrue(zipline.sim.ready())
self.assertFalse(zipline.sim.exception)
self.assertEqual(zipline.sim.feed.pending_messages(), 0, \
"The feed should be drained of all messages, found {n} remaining." \
.format(n=zipline.sim.feed.pending_messages()))
#
# the trading client should receive one transaction for every
# order placed.
self.assertEqual(
zipline.trading_client.txn_count,
zipline.trading_client.order_count
)
@timed(DEFAULT_TIMEOUT)
def test_performance(self):
#provide enough trades to ensure all orders are filled.
self.zipline_test_config['order_count'] = 100
self.zipline_test_config['trade_count'] = 200
zipline = SimulatedTrading.create_test_zipline(**self.zipline_test_config)
zipline.simulate(blocking=True)
self.assertEqual(
zipline.sim.feed.pending_messages(),
0,
"The feed should be drained of all messages, found {n} remaining." \
.format(n=zipline.sim.feed.pending_messages())
)
self.assertEqual(
zipline.sim.merge.pending_messages(),
0,
"The merge should be drained of all messages, found {n} remaining." \
.format(n=zipline.sim.merge.pending_messages())
)
self.assertEqual(
zipline.algorithm.count,
zipline.algorithm.incr,
"The test algorithm should send as many orders as specified.")
transaction_sim = zipline.trading_client.txn_sim
self.assertEqual(
transaction_sim.txn_count,
zipline.trading_client.perf.txn_count,
"The perf tracker should handle the same number of transactions \
as the simulator emits."
)
self.assertEqual(
len(zipline.get_positions()),
1,
"Portfolio should have one position."
)
SID = self.zipline_test_config['sid']
self.assertEqual(
zipline.get_positions()[SID]['sid'],
SID,
"Portfolio should have one position in " + str(SID)
)
self.assertEqual(
zipline.sources['flat'].count,
self.zipline_test_config['trade_count'],
"The simulated trade source should send all trades."
)
self.assertEqual(
zipline.algorithm.frame_count,
self.zipline_test_config['trade_count'],
"The algorithm should receive all trades."
)
@timed(DEFAULT_TIMEOUT)
def test_sid_filter(self):
"""Ensure the algorithm's filter prevents events from arriving."""
# create a test algorithm whose filter will not match any of the
# trade events sourced inside the zipline.
order_amount = 100
order_count = 100
no_match_sid = 222
test_algo = TestAlgorithm(
no_match_sid,
order_amount,
order_count
)
self.zipline_test_config['trade_count'] = 200
self.zipline_test_config['algorithm'] = test_algo
zipline = SimulatedTrading.create_test_zipline(
**self.zipline_test_config
)
zipline.simulate(blocking=True)
#check that the algorithm received no events
self.assertEqual(
0,
test_algo.frame_count,
"The algorithm should not receive any events due to filtering."
)
# 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.
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':1,
'expected_txn_volume':24 * 1
}
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':1,
'expected_txn_volume':24 * -1
}
self.transaction_sim(**params2)
@timed(DEFAULT_TIMEOUT)
def test_partial_expiration_orders(self):
# create a scenario where orders expire without being filled
# entirely
params1 = {
'trade_count':100,
'trade_amount':100,
'trade_delay': timedelta(minutes=5),
'trade_interval': timedelta(days=1),
'order_count':3,
'order_amount':1000,
'order_interval': timedelta(minutes=30),
# because we placed an orders totaling less than 25% of one trade
# the simulator should produce just one transaction.
'expected_txn_count' : 1,
'expected_txn_volume' : 25
}
self.transaction_sim(**params1)
# same scenario, but short sales.
params2 = {
'trade_count' : 100,
'trade_amount' : 100,
'trade_delay' : timedelta(minutes=5),
'trade_interval' : timedelta(days=1),
'order_count' : 3,
'order_amount' :-1000,
'order_interval' : timedelta(minutes=30),
# because we placed an orders totaling less than 25% of one trade
# the simulator should produce just one transaction.
'expected_txn_count' : 1,
'expected_txn_volume' : -25
}
self.transaction_sim(**params2)
@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):
trade_count = params['trade_count']
trade_amount = params['trade_amount']
trade_interval = params['trade_interval']
trade_delay = params.get('trade_delay')
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')
trading_environment = factory.create_trading_environment()
trade_sim = TransactionSimulator()
price = [10.1] * trade_count
volume = [100] * trade_count
start_date = trading_environment.first_open
sid = 1
generated_trades = factory.create_trade_history(
sid,
price,
volume,
trade_interval,
trading_environment
)
if alternate:
alternator = -1
else:
alternator = 1
order_date = start_date
for i in xrange(order_count):
order = namedict(
{
'sid' : sid,
'amount' : order_amount * alternator**i,
'type' : zp.DATASOURCE_TYPE.ORDER,
'dt' : order_date
})
trade_sim.add_open_order(order)
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 = trade_sim.open_orders
self.assertEqual(len(oo), 1)
self.assertTrue(oo.has_key(sid))
order_list = oo[sid]
self.assertEqual(order_count, len(order_list))
for i in xrange(order_count):
order = order_list[i]
self.assertEqual(order.sid, sid)
self.assertEqual(order.amount, order_amount * alternator**i)
tracker = PerformanceTracker(trading_environment)
# this approximates the loop inside TradingSimulationClient
transactions = []
for trade in generated_trades:
if trade_delay:
trade.dt = trade.dt + trade_delay
txn = trade_sim.apply_trade_to_open_orders(trade)
if txn:
transactions.append(txn)
trade.TRANSACTION = txn
else:
trade.TRANSACTION = None
tracker.process_event(trade)
if complete_fill:
self.assertEqual(len(transactions), len(order_list))
total_volume = 0
for i in xrange(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 now be empty
oo = trade_sim.open_orders
self.assertTrue(oo.has_key(sid))
order_list = oo[sid]
self.assertEqual(0, len(order_list))
tests/test_monitor.py
View File
tests/test_ndict.py
+46
View File
@@ -0,0 +1,46 @@
from zipline.protocol_utils import ndict, namedict
def test_ndict():
nd = ndict({})
# Properties
assert len(nd) == 0
assert nd.keys() == []
assert nd.values() == []
assert list(nd.iteritems()) == []
# Accessors
nd['x'] = 1
assert nd.x == 1
assert nd['x'] == 1
assert nd.get('y') == None
assert nd.get('y', 'fizzpop') == 'fizzpop'
assert nd.has_key('x') == True
assert nd.has_key('y') == False
assert 'x' in nd
assert 'y' not in nd
# Class isolation
assert '__init__' not in nd
assert '__iter__' not in nd
assert not nd.__dict__.has_key('x')
assert nd.get('__init__') is None
# Comparison
nd2 = nd.copy()
assert id(nd2) != id(nd)
assert nd2 == nd
nd2['z'] = 3
assert nd2 != nd
class ndictlike(object):
x = 1
assert { 'x': 1 } == nd
assert ndictlike() != nd
# Deletion
del nd['x']
assert not nd.has_key('x')
assert nd.get('x') is None
tests/test_perf_tracking.py
+568
View File
@@ -0,0 +1,568 @@
import unittest
import copy
import random
import datetime
import pytz
import zipline.test.factory as factory
import zipline.test.algorithms
import zipline.util as qutil
import zipline.finance.performance as perf
import zipline.finance.risk as risk
import zipline.protocol as zp
from zipline.finance.trading import TradeSimulationClient, TradingEnvironment, \
SIMULATION_STYLE
class PerformanceTestCase(unittest.TestCase):
def setUp(self):
qutil.configure_logging()
self.benchmark_returns, self.treasury_curves = \
factory.load_market_data()
random_index = random.randint(
0,
len(self.treasury_curves)
)
for n in range(100):
self.dt = self.treasury_curves.keys()[random_index]
self.end_dt = self.dt + datetime.timedelta(days=365)
now = datetime.datetime.utcnow().replace(tzinfo=pytz.utc)
if self.end_dt <= now:
break
self.trading_environment = TradingEnvironment(
self.benchmark_returns,
self.treasury_curves,
period_start = self.dt,
period_end = self.end_dt
)
self.onesec = datetime.timedelta(seconds=1)
self.oneday = datetime.timedelta(days=1)
self.tradingday = datetime.timedelta(hours=6, minutes=30)
self.dt = self.trading_environment.trading_days[random_index]
def tearDown(self):
pass
def test_long_position(self):
"""
verify that the performance period calculates properly for a
single buy transaction
"""
#post some trades in the market
trades = factory.create_trade_history(
1,
[10,10,10,11],
[100,100,100,100],
self.onesec,
self.trading_environment
)
txn = factory.create_txn(1,10.0,100,self.dt + self.onesec)
pp = perf.PerformancePeriod({}, 0.0, 1000.0)
pp.execute_transaction(txn)
for trade in trades:
pp.update_last_sale(trade)
pp.calculate_performance()
self.assertEqual(
pp.period_capital_used,
-1 * txn.price * txn.amount,
"capital used should be equal to the opposite of the transaction \
cost of sole txn in test"
)
self.assertEqual(len(pp.positions),1,"should be just one position")
self.assertEqual(
pp.positions[1].sid,
txn.sid,
"position should be in security with id 1")
self.assertEqual(
pp.positions[1].amount,
txn.amount,
"should have a position of {sharecount} shares".format(
sharecount=txn.amount
)
)
self.assertEqual(
pp.positions[1].cost_basis,
txn.price,
"should have a cost basis of 10"
)
self.assertEqual(
pp.positions[1].last_sale_price,
trades[-1]['price'],
"last sale should be same as last trade. \
expected {exp} actual {act}".format(
exp=trades[-1]['price'],
act=pp.positions[1].last_sale_price
)
)
self.assertEqual(
pp.ending_value,
1100,
"ending value should be price of last trade times number of \
shares in position"
)
self.assertEqual(pp.pnl, 100, "gain of 1 on 100 shares should be 100")
def test_short_position(self):
"""verify that the performance period calculates properly for a \
single short-sale transaction"""
trades = factory.create_trade_history(
1,
[10,10,10,11,10,9],
[100,100,100,100,100,100],
self.onesec,
self.trading_environment
)
trades_1 = trades[:-2]
txn = factory.create_txn(1, 10.0, -100, self.dt + self.onesec)
pp = perf.PerformancePeriod({}, 0.0, 1000.0)
pp.execute_transaction(txn)
for trade in trades_1:
pp.update_last_sale(trade)
pp.calculate_performance()
self.assertEqual(
pp.period_capital_used,
-1 * txn.price * txn.amount,
"capital used should be equal to the opposite of the transaction\
cost of sole txn in test"
)
self.assertEqual(
len(pp.positions),
1,
"should be just one position")
self.assertEqual(
pp.positions[1].sid,
txn.sid,
"position should be in security from the transaction"
)
self.assertEqual(
pp.positions[1].amount,
-100,
"should have a position of -100 shares"
)
self.assertEqual(
pp.positions[1].cost_basis,
txn.price,
"should have a cost basis of 10"
)
self.assertEqual(
pp.positions[1].last_sale_price,
trades_1[-1]['price'],
"last sale should be price of last trade"
)
self.assertEqual(
pp.ending_value,
-1100,
"ending value should be price of last trade times number of \
shares in position"
)
self.assertEqual(pp.pnl,-100,"gain of 1 on 100 shares should be 100")
# simulate additional trades, and ensure that the position value
# reflects the new price
trades_2 = trades[-2:]
#simulate a rollover to a new period
pp2 = perf.PerformancePeriod(
pp.positions,
pp.ending_value,
pp.ending_cash
)
for trade in trades_2:
pp2.update_last_sale(trade)
pp2.calculate_performance()
self.assertEqual(
pp2.period_capital_used,
0,
"capital used should be zero, there were no transactions in \
performance period"
)
self.assertEqual(
len(pp2.positions),
1,
"should be just one position"
)
self.assertEqual(
pp2.positions[1].sid,
txn.sid,
"position should be in security from the transaction"
)
self.assertEqual(
pp2.positions[1].amount,
-100,
"should have a position of -100 shares"
)
self.assertEqual(
pp2.positions[1].cost_basis,
txn.price,
"should have a cost basis of 10"
)
self.assertEqual(
pp2.positions[1].last_sale_price,
trades_2[-1].price,
"last sale should be price of last trade"
)
self.assertEqual(
pp2.ending_value,
-900,
"ending value should be price of last trade times number of \
shares in position")
self.assertEqual(
pp2.pnl,
200,
"drop of 2 on -100 shares should be 200"
)
#now run a performance period encompassing the entire trade sample.
ppTotal = perf.PerformancePeriod({}, 0.0, 1000.0)
for trade in trades_1:
ppTotal.update_last_sale(trade)
ppTotal.execute_transaction(txn)
for trade in trades_2:
ppTotal.update_last_sale(trade)
ppTotal.calculate_performance()
self.assertEqual(
ppTotal.period_capital_used,
-1 * txn.price * txn.amount,
"capital used should be equal to the opposite of the transaction \
cost of sole txn in test"
)
self.assertEqual(
len(ppTotal.positions),
1,
"should be just one position"
)
self.assertEqual(
ppTotal.positions[1].sid,
txn.sid,
"position should be in security from the transaction"
)
self.assertEqual(
ppTotal.positions[1].amount,
-100,
"should have a position of -100 shares"
)
self.assertEqual(
ppTotal.positions[1].cost_basis,
txn.price,
"should have a cost basis of 10"
)
self.assertEqual(
ppTotal.positions[1].last_sale_price,
trades_2[-1].price,
"last sale should be price of last trade"
)
self.assertEqual(
ppTotal.ending_value,
-900,
"ending value should be price of last trade times number of \
shares in position")
self.assertEqual(
ppTotal.pnl,
100,
"drop of 1 on -100 shares should be 100"
)
def test_covering_short(self):
"""verify performance where short is bought and covered, and shares \
trade after cover"""
trades = factory.create_trade_history(
1,
[10,10,10,11,9,8,7,8,9,10],
[100,100,100,100,100,100,100,100,100,100],
self.onesec,
self.trading_environment
)
short_txn = factory.create_txn(
1,
10.0,
-100,
self.dt + self.onesec
)
cover_txn = factory.create_txn(1,7.0,100,self.dt + self.onesec * 6)
pp = perf.PerformancePeriod({}, 0.0, 1000.0)
pp.execute_transaction(short_txn)
pp.execute_transaction(cover_txn)
for trade in trades:
pp.update_last_sale(trade)
pp.calculate_performance()
short_txn_cost = short_txn.price * short_txn.amount
cover_txn_cost = cover_txn.price * cover_txn.amount
self.assertEqual(
pp.period_capital_used,
-1 * short_txn_cost - cover_txn_cost,
"capital used should be equal to the net transaction costs"
)
self.assertEqual(
len(pp.positions),
1,
"should be just one position"
)
self.assertEqual(
pp.positions[1].sid,
short_txn.sid,
"position should be in security from the transaction"
)
self.assertEqual(
pp.positions[1].amount,
0,
"should have a position of -100 shares"
)
self.assertEqual(
pp.positions[1].cost_basis,
0,
"a covered position should have a cost basis of 0"
)
self.assertEqual(
pp.positions[1].last_sale_price,
trades[-1].price,
"last sale should be price of last trade"
)
self.assertEqual(
pp.ending_value,
0,
"ending value should be price of last trade times number of \
shares in position"
)
self.assertEqual(
pp.pnl,
300,
"gain of 1 on 100 shares should be 300"
)
def test_cost_basis_calc(self):
trades = factory.create_trade_history(
1,
[10,11,11,12],
[100,100,100,100],
self.onesec,
self.trading_environment
)
transactions = factory.create_txn_history(
1,
[10,11,11,12],
[100,100,100,100],
self.onesec,
self.trading_environment
)
pp = perf.PerformancePeriod({}, 0.0, 1000.0)
for txn in transactions:
pp.execute_transaction(txn)
for trade in trades:
pp.update_last_sale(trade)
pp.calculate_performance()
self.assertEqual(
pp.positions[1].last_sale_price,
trades[-1].price,
"should have a last sale of 12, got {val}".format(
val=pp.positions[1].last_sale_price
)
)
self.assertEqual(
pp.positions[1].cost_basis,
11,
"should have a cost basis of 11"
)
self.assertEqual(
pp.pnl,
400
)
saleTxn = factory.create_txn(
1,
10.0,
-100,
self.dt + self.onesec * 4)
down_tick = factory.create_trade(
1,
10.0,
100,
trades[-1].dt + self.onesec)
pp2 = perf.PerformancePeriod(
copy.deepcopy(pp.positions),
pp.ending_value,
pp.ending_cash
)
pp2.execute_transaction(saleTxn)
pp2.update_last_sale(down_tick)
pp2.calculate_performance()
self.assertEqual(
pp2.positions[1].last_sale_price,
10,
"should have a last sale of 10, was {val}".format(val=pp2.positions[1].last_sale_price)
)
self.assertEqual(
round(pp2.positions[1].cost_basis,2),
11.33,
"should have a cost basis of 11.33"
)
#print "second period pnl is {pnl}".format(pnl=pp2.pnl)
self.assertEqual(pp2.pnl, -800, "this period goes from +400 to -400")
pp3 = perf.PerformancePeriod({}, 0.0, 1000.0)
transactions.append(saleTxn)
for txn in transactions:
pp3.execute_transaction(txn)
trades.append(down_tick)
for trade in trades:
pp3.update_last_sale(trade)
pp3.calculate_performance()
self.assertEqual(
pp3.positions[1].last_sale_price,
10,
"should have a last sale of 10"
)
self.assertEqual(
round(pp3.positions[1].cost_basis,2),
11.33,
"should have a cost basis of 11.33"
)
self.assertEqual(
pp3.pnl,
-400,
"should be -400 for all trades and transactions in period"
)
def test_tracker(self):
trade_count = 100
sid = 133
price = 10.1
price_list = [price] * trade_count
volume = [100] * trade_count
trade_time_increment = datetime.timedelta(days=1)
trade_history = factory.create_trade_history(
sid,
price_list,
volume,
trade_time_increment,
self.trading_environment
)
sid2 = 134
price2 = 12.12
price2_list = [price2] * trade_count
trade_history2 = factory.create_trade_history(
sid2,
price2_list,
volume,
trade_time_increment,
self.trading_environment
)
trade_history.extend(trade_history2)
self.trading_environment.period_start = trade_history[0].dt
self.trading_environment.period_end = trade_history[-1].dt
self.trading_environment.capital_base = 1000.0
self.trading_environment.frame_index = ['sid', 'volume', 'dt', \
'price', 'changed']
perf_tracker = perf.PerformanceTracker(self.trading_environment)
for event in trade_history:
#create a transaction for all but
#first trade in each sid, to simulate None transaction
if(event.dt != self.trading_environment.period_start):
txn = zp.namedict({
'sid' : event.sid,
'amount' : -25,
'dt' : event.dt,
'price' : 10.0,
'commission' : 0.50
})
else:
txn = None
event[zp.TRANSFORM_TYPE.TRANSACTION] = txn
perf_tracker.process_event(event)
#we skip two trades, to test case of None transaction
txn_count = len(trade_history) - 2
self.assertEqual(perf_tracker.txn_count, txn_count)
cumulative_pos = perf_tracker.cumulative_performance.positions[sid]
expected_size = txn_count / 2 * -25
self.assertEqual(cumulative_pos.amount, expected_size)
tests/test_protocol.py
+129
View File
@@ -0,0 +1,129 @@
"""
Test the FRAME/UNFRAME functions in the sequence expected from ziplines.
"""
import pytz
from unittest2 import TestCase
from datetime import datetime, timedelta
from collections import defaultdict
from nose.tools import timed
import zipline.test.factory as factory
import zipline.util as qutil
import zipline.protocol as zp
from zipline.sources import SpecificEquityTrades
DEFAULT_TIMEOUT = 5 # seconds
class ProtocolTestCase(TestCase):
leased_sockets = defaultdict(list)
def setUp(self):
qutil.configure_logging()
self.trading_environment = factory.create_trading_environment()
@timed(DEFAULT_TIMEOUT)
def test_trade_feed_protocol(self):
sid = 133
price = [10.0] * 4
volume = [100] * 4
start_date = datetime.strptime("02/15/2012","%m/%d/%Y")
one_day_td = timedelta(days=1)
trades = factory.create_trade_history(
sid,
price,
volume,
one_day_td,
self.trading_environment
)
for trade in trades:
#simulate data source sending frame
msg = zp.DATASOURCE_FRAME(zp.namedict(trade))
#feed unpacking frame
recovered_trade = zp.DATASOURCE_UNFRAME(msg)
#feed sending frame
feed_msg = zp.FEED_FRAME(recovered_trade)
#transform unframing
recovered_feed = zp.FEED_UNFRAME(feed_msg)
#do a transform
trans_msg = zp.TRANSFORM_FRAME('helloworld', 2345.6)
#simulate passthrough transform -- passthrough shouldn't even
# unpack the msg, just resend.
passthrough_msg = zp.TRANSFORM_FRAME(zp.TRANSFORM_TYPE.PASSTHROUGH,\
feed_msg)
#merge unframes transform and passthrough
trans_recovered = zp.TRANSFORM_UNFRAME(trans_msg)
pt_recovered = zp.TRANSFORM_UNFRAME(passthrough_msg)
#simulated merge
pt_recovered.PASSTHROUGH.merge(trans_recovered)
#frame the merged event
merged_msg = zp.MERGE_FRAME(pt_recovered.PASSTHROUGH)
#unframe the merge and validate values
event = zp.MERGE_UNFRAME(merged_msg)
#check the transformed value, should only be in event, not trade.
self.assertTrue(event.helloworld == 2345.6)
event.delete('helloworld')
self.assertEqual(zp.namedict(trade), event)
@timed(DEFAULT_TIMEOUT)
def test_order_protocol(self):
#client places an order
now = datetime.utcnow().replace(tzinfo=pytz.utc)
order = zp.namedict({
'dt':now,
'sid':133,
'amount':100
})
order_msg = zp.ORDER_FRAME(order)
#order datasource receives
order = zp.ORDER_UNFRAME(order_msg)
self.assertEqual(order.sid, 133)
self.assertEqual(order.amount, 100)
self.assertEqual(order.dt, now)
#order datasource datasource frames the order
order_event = zp.namedict({
"sid" : order.sid,
"amount" : order.amount,
"dt" : order.dt,
"source_id" : zp.FINANCE_COMPONENT.ORDER_SOURCE,
"type" : zp.DATASOURCE_TYPE.ORDER
})
order_ds_msg = zp.DATASOURCE_FRAME(order_event)
#transaction transform unframes
recovered_order = zp.DATASOURCE_UNFRAME(order_ds_msg)
self.assertEqual(now, recovered_order.dt)
#create a transaction from the order
txn = zp.namedict({
'sid' : recovered_order.sid,
'amount' : recovered_order.amount,
'dt' : recovered_order.dt,
'price' : 10.0,
'commission' : 0.50
})
#frame that transaction
txn_msg = zp.TRANSFORM_FRAME(zp.TRANSFORM_TYPE.TRANSACTION, txn)
#unframe
recovered_tx = zp.TRANSFORM_UNFRAME(txn_msg).TRANSACTION
self.assertEqual(recovered_tx.sid, 133)
self.assertEqual(recovered_tx.amount, 100)
tests/test_risk.py
+357
View File
@@ -0,0 +1,357 @@
import unittest
import copy
import datetime
import calendar
import pytz
import zipline.finance.risk as risk
import zipline.test.factory as factory
import zipline.util as qutil
from zipline.finance.trading import TradingEnvironment
class Risk(unittest.TestCase):
def setUp(self):
qutil.configure_logging()
start_date = datetime.datetime(
year=2006,
month=1,
day=1,
hour=0,
minute=0,
tzinfo=pytz.utc)
end_date = datetime.datetime(year=2006, month=12, day=31, tzinfo=pytz.utc)
self.benchmark_returns, self.treasury_curves = \
factory.load_market_data()
self.trading_env = TradingEnvironment(
self.benchmark_returns,
self.treasury_curves,
period_start = start_date,
period_end = end_date
)
self.onesec = datetime.timedelta(seconds=1)
self.oneday = datetime.timedelta(days=1)
self.tradingday = datetime.timedelta(hours=6, minutes=30)
self.dt = datetime.datetime.utcnow()
self.algo_returns_06 = factory.create_returns_from_list(
RETURNS,
self.trading_env
)
self.metrics_06 = risk.RiskReport(
self.algo_returns_06,
self.trading_env
)
start_08 = datetime.datetime(
year=2008,
month=1,
day=1,
hour=0,
minute=0,
tzinfo=pytz.utc)
end_08 = datetime.datetime(
year=2008,
month=12,
day=31,
tzinfo=pytz.utc
)
self.trading_env08 = TradingEnvironment(
self.benchmark_returns,
self.treasury_curves,
period_start = start_08,
period_end = end_08
)
def tearDown(self):
return
def test_factory(self):
returns = [0.1] * 100
r_objects = factory.create_returns_from_list(returns, self.trading_env)
self.assertTrue(r_objects[-1].date <= datetime.datetime(year=2006, month=12, day=31, tzinfo=pytz.utc))
def test_drawdown(self):
returns = factory.create_returns_from_list([1.0,-0.5,0.8,.17,1.0,-0.1,-0.45], self.trading_env)
#200, 100, 180, 210.6, 421.2, 379.8, 208.494
metrics = risk.RiskMetrics(returns[0].date, returns[-1].date, returns, self.trading_env)
self.assertEqual(metrics.max_drawdown, 0.505)
def test_benchmark_returns_06(self):
returns = factory.create_returns_from_range(self.trading_env)
metrics = risk.RiskReport(returns, self.trading_env)
self.assertEqual([round(x.benchmark_period_returns, 4) for x in metrics.month_periods],
[0.0255,0.0005,0.0111,0.0122,-0.0309,0.0001,0.0051,0.0213,0.0246,0.0315,0.0165,0.0126])
self.assertEqual([round(x.benchmark_period_returns, 4) for x in metrics.three_month_periods],
[0.0373,0.0239,-0.0083,-0.0191,-0.0259,0.0266,0.0517,0.0793,0.0743,0.0617])
self.assertEqual([round(x.benchmark_period_returns, 4) for x in metrics.six_month_periods],
[0.0176,-0.0027,0.0181,0.0316,0.0514,0.1028,0.1166])
self.assertEqual([round(x.benchmark_period_returns,4) for x in metrics.year_periods],[0.1362])
def test_trading_days_06(self):
returns = factory.create_returns_from_range(self.trading_env)
metrics = risk.RiskReport(returns, self.trading_env)
self.assertEqual([x.trading_days for x in metrics.year_periods],[251])
self.assertEqual([x.trading_days for x in metrics.month_periods],[20,19,23,19,22,22,20,23,20,22,21,20])
def test_benchmark_volatility_06(self):
returns = factory.create_returns_from_range(self.trading_env)
metrics = risk.RiskReport(returns, self.trading_env)
self.assertEqual([round(x.benchmark_volatility, 3) for x in metrics.month_periods],
[0.031,0.026,0.024,0.025,0.037,0.047,0.039,0.022,0.023,0.021,0.025,0.019])
self.assertEqual([round(x.benchmark_volatility, 3) for x in metrics.three_month_periods],
[0.047,0.042,0.050,0.064,0.070,0.064,0.049,0.037,0.039,0.037])
self.assertEqual([round(x.benchmark_volatility, 3) for x in metrics.six_month_periods],
[0.079,0.082,0.081,0.081,0.08,0.074,0.061])
self.assertEqual([round(x.benchmark_volatility, 3) for x in metrics.year_periods],[0.100])
def test_algorithm_returns_06(self):
self.assertEqual([round(x.algorithm_period_returns, 3) for x in self.metrics_06.month_periods],[0.101,-0.062,-0.041,0.092,0.135,-0.25,0.076,-0.003,-0.024,0.072,0.063,-0.071])
self.assertEqual([round(x.algorithm_period_returns, 3) for x in self.metrics_06.three_month_periods],[-0.009,-0.017,0.188,-0.071,-0.085,-0.196,0.047,0.043,0.112,0.058])
self.assertEqual([round(x.algorithm_period_returns, 3) for x in self.metrics_06.six_month_periods],[-0.08,-0.101,-0.044,-0.027,-0.045,-0.106,0.108])
self.assertEqual([round(x.algorithm_period_returns, 3) for x in self.metrics_06.year_periods],[0.02])
def test_algorithm_volatility_06(self):
self.assertEqual([round(x.algorithm_volatility, 3) for x in self.metrics_06.month_periods],[0.137,0.12,0.13,0.142,0.128,0.14,0.141,0.118,0.143,0.144,0.117,0.135])
self.assertEqual([round(x.algorithm_volatility, 3) for x in self.metrics_06.three_month_periods],[0.222,0.224,0.229,0.243,0.243,0.235,0.23,0.231,0.231,0.227])
self.assertEqual([round(x.algorithm_volatility, 3) for x in self.metrics_06.six_month_periods],[0.328,0.329,0.329,0.333,0.334,0.329,0.321])
self.assertEqual([round(x.algorithm_volatility, 3) for x in self.metrics_06.year_periods],[0.458])
def test_algorithm_sharpe_06(self):
self.assertEqual([round(x.sharpe, 3) for x in self.metrics_06.month_periods],[0.711,-0.541,-0.348,0.625,1.017,-1.809,0.508,-0.062,-0.193,0.467,0.502,-0.557])
self.assertEqual([round(x.sharpe, 3) for x in self.metrics_06.three_month_periods],[-0.094,-0.129,0.769,-0.342,-0.402,-0.888,0.153,0.131,0.432,0.2])
self.assertEqual([round(x.sharpe, 3) for x in self.metrics_06.six_month_periods],[-0.322,-0.383,-0.213,-0.156,-0.213,-0.398,0.257])
self.assertEqual([round(x.sharpe, 3) for x in self.metrics_06.year_periods],[-0.066])
def dtest_algorithm_beta_06(self):
self.assertEqual([round(x.beta, 3) for x in self.metrics_06.month_periods],[0.553,0.583,-2.168,-0.548,1.463,-0.322,-1.38,1.473,-1.315,-0.7,0.352,-2.002])
self.assertEqual([round(x.beta, 3) for x in self.metrics_06.three_month_periods],[-0.075,-0.637,0.124,0.186,-0.204,-0.497,-0.867,-0.173,-0.499,-0.563])
self.assertEqual([round(x.beta, 3) for x in self.metrics_06.six_month_periods],[-0.075,-0.637,0.124,0.186,-0.204,-0.497,-0.867,-0.173,-0.499,-0.563])
self.assertEqual([round(x.beta, 3) for x in self.metrics_06.year_periods],[-0.219])
def dtest_algorithm_alpha_06(self):
self.assertEqual([round(x.alpha, 3) for x in self.metrics_06.month_periods],[0.085,-0.063,-0.03,0.093,0.182,-0.255,0.073,-0.032,0,0.086,0.054,-0.058])
self.assertEqual([round(x.alpha, 3) for x in self.metrics_06.three_month_periods],[-0.051,-0.021,0.179,-0.077,-0.106,-0.202,0.069,0.042,0.13,0.073])
self.assertEqual([round(x.alpha, 3) for x in self.metrics_06.six_month_periods],[-0.105,-0.135,-0.072,-0.051,-0.066,-0.094,0.152])
self.assertEqual([round(x.alpha, 3) for x in self.metrics_06.year_periods],[-0.011])
#FIXME: Covariance is not matching excel precisely enough to run the test. Month 4 seems to be the problem. Variance is disabled
#just to avoid distraction - it is much closer than covariance and can probably pass with 6 significant digits instead of 7.
#re-enable variance, alpha, and beta tests once this is resolved
def dtest_algorithm_covariance_06(self):
metric = self.metrics_06.month_periods[3]
print repr(metric)
print "----"
self.assertEqual([round(x.algorithm_covariance, 7) for x in self.metrics_06.month_periods],[0.0000289,0.0000222,-0.0000554,-0.0000192,0.0000954,-0.0000333,-0.0001111,0.0000322,-0.0000349,-0.0000143,0.0000108,-0.0000386])
self.assertEqual([round(x.algorithm_covariance, 7) for x in self.metrics_06.three_month_periods],[-0.0000026,-0.0000189,0.0000049,0.0000121,-0.0000158,-0.000031,-0.0000336,-0.0000036,-0.0000119,-0.0000122])
self.assertEqual([round(x.algorithm_covariance, 7) for x in self.metrics_06.six_month_periods],[0.000005,-0.0000172,-0.0000142,-0.0000102,-0.0000089,-0.0000207,-0.0000229])
self.assertEqual([round(x.algorithm_covariance, 7) for x in self.metrics_06.year_periods],[-8.75273E-06])
def dtest_benchmark_variance_06(self):
self.assertEqual([round(x.benchmark_variance, 7) for x in self.metrics_06.month_periods],[0.0000496,0.000036,0.0000244,0.0000332,0.0000623,0.0000989,0.0000765,0.0000209,0.0000252,0.0000194,0.0000292,0.0000183])
self.assertEqual([round(x.benchmark_variance, 7) for x in self.metrics_06.three_month_periods],[0.0000351,0.0000298,0.0000395,0.0000648,0.0000773,0.0000625,0.0000387,0.0000211,0.0000238,0.0000217])
self.assertEqual([round(x.benchmark_variance, 7) for x in self.metrics_06.six_month_periods],[0.0000499,0.0000538,0.0000508,0.0000517,0.0000492,0.0000432,0.00003])
self.assertEqual([round(x.benchmark_variance, 7) for x in self.metrics_06.year_periods],[0.0000399])
def test_benchmark_returns_08(self):
returns = factory.create_returns_from_range(self.trading_env08)
metrics = risk.RiskReport(returns, self.trading_env08)
monthly = [round(x.benchmark_period_returns, 3) for x in metrics.month_periods]
self.assertEqual( monthly,
[-0.061,-0.035,-0.006,0.048,0.011,-0.086,-0.01,0.012,-0.091,-0.169,-0.075,0.008])
self.assertEqual([round(x.benchmark_period_returns, 3) for x in metrics.three_month_periods],
[-0.099,0.005,0.052,-0.032,-0.085,-0.084,-0.089,-0.236,-0.301,-0.226])
self.assertEqual([round(x.benchmark_period_returns, 3) for x in metrics.six_month_periods],
[-0.128,-0.081,-0.036,-0.118,-0.301,-0.360,-0.294])
self.assertEqual([round(x.benchmark_period_returns,3) for x in metrics.year_periods],[-0.385])
def test_trading_days_08(self):
returns = factory.create_returns_from_range(self.trading_env08)
metrics = risk.RiskReport(returns, self.trading_env08)
self.assertEqual([x.trading_days for x in metrics.year_periods],[253])
self.assertEqual([x.trading_days for x in metrics.month_periods],[21,20,20,22,21,21,22,21,21,23,19,22])
def test_benchmark_volatility_08(self):
returns = factory.create_returns_from_range(self.trading_env08)
metrics = risk.RiskReport(returns, self.trading_env08)
self.assertEqual([round(x.benchmark_volatility, 3) for x in metrics.month_periods],
[0.07,0.058,0.082,0.054,0.041,0.057,0.068,0.06,0.157,0.244,0.195,0.145])
self.assertEqual([round(x.benchmark_volatility, 3) for x in metrics.three_month_periods],
[0.120,0.113,0.105,0.09,0.098,0.107,0.179,0.293,0.344,0.340])
self.assertEqual([round(x.benchmark_volatility, 3) for x in metrics.six_month_periods],
[0.15,0.149,0.15,0.2,0.308,0.36,0.383])
#TODO: ugly, but I can't get the rounded float to match. maybe we need a different test that checks the difference between the numbers
self.assertEqual([round(x.benchmark_volatility, 3) for x in metrics.year_periods],[0.41099999999999998])
def test_treasury_returns_06(self):
returns = factory.create_returns_from_range(self.trading_env)
metrics = risk.RiskReport(returns, self.trading_env)
self.assertEqual([round(x.treasury_period_return, 4) for x in metrics.month_periods],
[0.0037,0.0034,0.0039,0.0038,0.0040,0.0037,0.0043,0.0043,0.0038,0.0044,0.0043,0.0041])
self.assertEqual([round(x.treasury_period_return, 4) for x in metrics.three_month_periods],
[0.0114,0.0118,0.0122,0.0125,0.0129,0.0127,0.0123,0.0128,0.0125,0.0128])
self.assertEqual([round(x.treasury_period_return, 4) for x in metrics.six_month_periods],
[0.0260,0.0257,0.0258,0.0252,0.0259,0.0256,0.0258])
self.assertEqual([round(x.treasury_period_return, 4) for x in metrics.year_periods],
[0.0500])
def test_benchmarkrange(self):
self.check_year_range(datetime.datetime(year=2008,month=1,day=1), 2)
def test_partial_month(self):
start = datetime.datetime(
year=1991,
month=1,
day=1,
hour=0,
minute=0,
tzinfo=pytz.utc)
#1992 and 1996 were leap years
total_days = 365 * 5 + 2
end = start + datetime.timedelta(days = total_days)
trading_env90s = TradingEnvironment(
self.benchmark_returns,
self.treasury_curves,
period_start = start,
period_end = end
)
returns = factory.create_returns(total_days, trading_env90s)
returns = returns[:-10] #truncate the returns series to end mid-month
metrics = risk.RiskReport(returns, trading_env90s)
total_months = 60
self.check_metrics(metrics, total_months, start)
def check_year_range(self, start_date, years):
if(start_date.month <= 2):
ld = calendar.leapdays(start_date.year, start_date.year + years)
else:
#because we may catch the leap of the last year, and i think this func is [start,end)
ld = calendar.leapdays(start_date.year, start_date.year + years + 1)
returns = factory.create_returns(365 * years + ld, self.trading_env08)
metrics = risk.RiskReport(returns, self.trading_env)
total_months = years * 12
self.check_metrics(metrics, total_months, start_date)
def check_metrics(self, metrics, total_months, start_date):
"""
confirm that the right number of riskmetrics were calculated for each
window length.
"""
self.assert_range_length(
metrics.month_periods,
total_months,
1,
start_date
)
self.assert_range_length(
metrics.three_month_periods,
total_months,
3,
start_date
)
self.assert_range_length(
metrics.six_month_periods,
total_months,
6,
start_date
)
self.assert_range_length(
metrics.year_periods,
total_months,
12,
start_date
)
def assert_last_day(self, period_end):
#30 days has september, april, june and november
if(period_end.month in [9,4,6,11]):
self.assertEqual(period_end.day, 30)
#all the rest have 31, except for february
elif(period_end.month != 2):
self.assertEqual(period_end.day, 31)
else:
if calendar.isleap(period_end.year):
self.assertEqual(period_end.day, 29)
else:
self.assertEqual(period_end.day, 28)
def assert_month(self, start_month, actual_end_month):
if start_month == 1:
expected_end_month = 12
else:
expected_end_month = start_month - 1
self.assertEqual(expected_end_month, actual_end_month)
def assert_range_length(self, col, total_months, period_length, start_date):
if(period_length > total_months):
self.assertEqual(len(col), 0)
else:
self.assertEqual(
len(col),
total_months - (period_length - 1),
"mismatch for total months - expected:{total_months}/actual:{actual}, period:{period_length}, start:{start_date}, calculated end:{end}".format(
total_months=total_months,
period_length=period_length,
start_date=start_date,
end=col[-1].end_date,
actual=len(col)
))
self.assert_month(start_date.month, col[-1].end_date.month)
self.assert_last_day(col[-1].end_date)
RETURNS = [
0.0093, -0.0193, 0.0351, 0.0396, 0.0338, -0.0211, 0.0389,
0.0326, -0.0137, -0.0411, -0.0032, 0.0149, 0.0133, 0.0348,
0.042 , -0.0455, 0.0262, -0.0461, 0.0021, -0.0273, -0.0429,
0.0427, -0.0104, 0.0346, -0.0311, 0.0003, 0.0211, 0.0248,
-0.0215, 0.004 , 0.0267, 0.0029, -0.0369, 0.0057, 0.0298,
-0.0179, -0.0361, -0.0401, -0.0123, -0.005 , 0.0203, -0.041 ,
0.0011, 0.0118, 0.0103, -0.0184, -0.0437, 0.0411, -0.0242,
-0.0054, -0.0039, -0.0273, -0.0075, 0.0064, -0.0376, 0.0424,
0.0399, 0.019 , 0.0236, -0.0284, -0.0341, 0.0266, 0.05 ,
0.0069, -0.0442, -0.016 , 0.0173, 0.0348, -0.0404, -0.0068,
-0.0376, 0.0356, 0.0043, -0.0481, -0.0134, 0.0257, 0.0442,
0.0234, 0.0394, 0.0376, -0.0147, -0.0098, 0.0474, -0.0102,
0.0138, 0.0286, 0.0347, 0.0279, -0.0067, 0.0462, -0.0432,
0.0247, 0.0174, -0.0305, -0.0317, -0.0068, 0.0264, -0.0257,
-0.0328, 0.0092, 0.0288, -0.002 , 0.0288, 0.028 , -0.0093,
0.0178, -0.0365, -0.0086, -0.0133, -0.0309, 0.0473, -0.0149,
0.0378, -0.0316, -0.0292, -0.0453, -0.0451, 0.0093, 0.0397,
-0.0361, -0.0168, -0.0494, -0.0143, -0.0405, -0.0349, 0.0069,
0.0378, -0.0233, -0.0492, 0.018 , -0.0386, 0.0339, 0.0119,
0.0454, 0.0118, -0.011 , -0.0254, 0.0266, -0.0366, -0.0211,
0.0399, 0.0307, 0.035 , -0.0402, 0.0304, -0.0031, 0.0256,
0.0134, -0.0019, -0.0235, -0.0058, -0.0117, 0.0051, -0.0451,
-0.0466, -0.0124, 0.0283, -0.0499, 0.0318, -0.0028, 0.0203,
0.005 , 0.0085, 0.0048, 0.0277, 0.0159, -0.0149, 0.035 ,
0.0404, -0.01 , 0.0377, 0.0302, 0.0046, -0.0328, -0.0469,
0.0071, -0.0382, -0.0214, 0.0429, 0.0145, -0.0279, -0.0172,
0.0423, 0.041 , -0.0183, 0.0137, -0.0412, -0.0348, 0.0302,
0.0248, 0.0051, -0.0298, -0.0103, -0.0333, -0.0399, 0.0485,
-0.0166, 0.0384, 0.0259, -0.0163, 0.0357, 0.0308, -0.0386,
0.0481, -0.0446, -0.0282, -0.0037, 0.0202, 0.0216, 0.0113,
0.0194, 0.0392, 0.0016, 0.0268, -0.0155, -0.027 , 0.02 ,
0.0216, -0.0009, 0.022 , 0. , 0.041 , 0.0133, -0.0382,
0.0495, -0.0221, -0.0329, -0.0033, -0.0089, -0.0129, -0.0252,
0.048 , -0.0307, -0.0357, 0.0033, -0.0412, -0.0407, 0.0455,
0.0159, -0.0051, -0.0274, -0.0213, 0.0361, 0.0051, -0.0378,
0.0084, 0.0066, -0.0103, -0.0037, 0.0478, -0.0278
]
tests/test_sanity.py
+7
View File
@@ -0,0 +1,7 @@
from unittest2 import TestCase
class TestEnviroment(TestCase):
def test_universe(self):
# first order logic is working today. Yay!
self.assertTrue(True != False)
tests/transform.py
+22
View File
@@ -0,0 +1,22 @@
from zipline.messaging import BaseTransform
from zipline.protocol import COMPONENT_TYPE
class DivideByZeroTransform(BaseTransform):
"""
A transform that fails.
"""
def __init__(self, name):
BaseTransform.__init__(self, "PASSTHROUGH")
self.state['name'] = name
self.init()
def init(self):
pass
@property
def get_type(self):
return COMPONENT_TYPE.CONDUIT
def transform(self, event):
return { 'value': 0/0 }