wassname/catalyst
1
0
Fork 0
mirror of https://github.com/wassname/catalyst.git synced 2026-07-13 17:42:42 +08:00
Code Issues Packages Projects Releases Wiki Activity

Resolved conflict.

Browse Source
This commit is contained in:
Thomas Wiecki
2012-04-20 09:33:57 -04:00
parent 013e23383c 1c65cc30a5
commit e2d1c580d3
16 changed files with 944 additions and 272 deletions
Download Patch File Download Diff File Expand all files Collapse all files
etc/requirements.txt
+1
View File
@@ -4,3 +4,4 @@ gevent-zeromq==0.2.2
msgpack-python==0.1.12
humanhash==0.0.1
ujson==1.18
iso8601==0.1.4
zipline/component.py
+2 -8
View File
@@ -75,7 +75,8 @@ class Component(object):
self.out_socket = None
self.killed = False
self.controller = None
self.heartbeat_timeout = 2000
# timeout after a full minute
self.heartbeat_timeout = 60 *1000
self.state_flag = COMPONENT_STATE.OK
self.error_state = COMPONENT_FAILURE.NOFAILURE
self.on_done = None
@@ -502,13 +503,6 @@ class Component(object):
"""
raise NotImplementedError
@property
def is_blocking(self):
"""
True if a zipline be held open for this component.
"""
return False
@property
def get_pure(self):
zipline/date_utils.py
+47
View File
@@ -2,6 +2,7 @@ from collections import namedtuple
import time
import pytz
import iso8601
import calendar
from dateutil import rrule
from datetime import datetime, date, timedelta
@@ -11,6 +12,50 @@ from dateutil.relativedelta import *
# --------------
d_tuple = namedtuple('dt', ['year', 'month', 'day', 'hour', 'minute', 'second', 'micros'])
# iso8061 utility
# ---------------------
def parse_iso8061(date_string):
dt = iso8601.parse_date(date_string)
dt = dt.replace(tzinfo = pytz.utc)
return dt
# Epoch utilities
# ---------------------
UNIX_EPOCH = datetime(1970, 1, 1, 0, 0, tzinfo = pytz.utc)
def EPOCH(utc_datetime):
"""
The key is to ensure all the dates you are using are in the utc timezone
before you start converting. See http://pytz.sourceforge.net/ to learn how
to do that properly. By normalizing to utc, you eliminate the ambiguity of
daylight savings transitions. Then you can safely use timedelta to calculate
distance from the unix epoch, and then convert to seconds or milliseconds.
Note that the resulting unix timestamp is itself in the UTC timezone. If you
wish to see the timestamp in a localized timezone, you will need to make
another conversion.
Also note that this will only work for dates after 1970.
"""
assert isinstance(utc_datetime, datetime)
# utc only please
assert utc_datetime.tzinfo == pytz.utc
# how long since the epoch?
delta = utc_datetime - UNIX_EPOCH
seconds = delta.total_seconds()
ms = seconds * 1000
return ms
def UN_EPOCH(ms_since_epoch):
seconds_since_epoch = ms_since_epoch / 1000
delta = timedelta(seconds = seconds_since_epoch)
dt = UNIX_EPOCH + delta
return dt
def iso8061_to_epoch(datestring):
dt = parse_iso8061(datestring)
return EPOCH(dt)
# UTC Datetime Subclasses
# -----------------------
def utcnow():
@@ -22,6 +67,8 @@ class utcdatetime(datetime):
dt = datetime.__new__(cls, *args, **kwargs)
return dt
# Datetime Calculations
# ---------------------
zipline/finance/performance.py
+73 -55
View File
@@ -38,10 +38,6 @@ Performance Tracking
+-----------------+----------------------------------------------------+
| capital_base | The initial capital assumed for this tracker. |
+-----------------+----------------------------------------------------+
| returns | List of dicts representing daily returns. See the |
| | comments for |
| | :py:meth:`zipline.finance.risk.DailyReturn.to_dict`|
+-----------------+----------------------------------------------------+
| cumulative_perf | A dictionary representing the cumulative |
| | performance through all the events delivered to |
| | this tracker. For details see the comments on |
@@ -61,8 +57,6 @@ Performance Tracking
| | For details look at the comments for |
| | :py:meth:`zipline.finance.risk.RiskMetrics.to_dict`|
+-----------------+----------------------------------------------------+
| timestamp | System time evevent occurs in zipilne |
+-----------------+----------------------------------------------------+
Position Tracking
@@ -78,12 +72,11 @@ Position Tracking
+-----------------+----------------------------------------------------+
| last_sale_price | price at last sale of the security on the exchange |
+-----------------+----------------------------------------------------+
| last_sale_date | datetime of the last trade of the position's |
| | security on the exchange |
+-----------------+----------------------------------------------------+
| timestamp | System time event occurs in zipilne |
| transactions | all the transactions that were acrued into this |
| | position. |
+-----------------+----------------------------------------------------+
Performance Period
==================
@@ -113,8 +106,7 @@ Performance Period
| returns | percentage returns for the entire portfolio over the |
| | period |
+---------------+------------------------------------------------------+
| timestamp | System time evevent occurs in zipilne |
+---------------+------------------------------------------------------+
"""
import datetime
@@ -157,14 +149,13 @@ class PerformanceTracker():
self.total_days = self.trading_environment.days_in_period
# one indexed so that we reach 100%
self.day_count = 0.0
self.cumulative_capital_used = 0.0
self.max_capital_used = 0.0
self.capital_base = self.trading_environment.capital_base
self.returns = []
self.txn_count = 0
self.event_count = 0
self.result_stream = None
self.last_dict = None
self.order_log = []
# this performance period will span the entire simulation.
self.cumulative_performance = PerformancePeriod(
@@ -183,7 +174,9 @@ class PerformanceTracker():
# initial portfolio positions have zero value
0,
# initial cash is your capital base.
starting_cash = self.capital_base
starting_cash = self.capital_base,
# save the transactions for the daily periods
keep_transactions = True
)
def get_portfolio(self):
@@ -208,57 +201,40 @@ class PerformanceTracker():
Creates a dictionary representing the state of this tracker.
Returns a dict object of the form described in header comments.
"""
returns_list = [x.to_dict() for x in self.returns]
return {
'started_at' : self.started_at,
'period_start' : self.period_start,
'period_end' : self.period_end,
'progress' : self.progress,
'cumulative_captial_used' : self.cumulative_capital_used,
'max_capital_used' : self.max_capital_used,
'cumulative_capital_used' : self.cumulative_performance.cumulative_capital_used,
'max_capital_used' : self.cumulative_performance.max_capital_used,
'last_close' : self.market_close,
'last_open' : self.market_open,
'capital_base' : self.capital_base,
'returns' : returns_list,
'cumulative_perf' : self.cumulative_performance.to_dict(),
'todays_perf' : self.todays_performance.to_dict(),
'daily_perf' : self.todays_performance.to_dict(),
'cumulative_risk_metrics' : self.cumulative_risk_metrics.to_dict(),
'timestamp' : datetime.datetime.now(),
}
def log_order(self, order):
self.order_log.append(order)
def process_event(self, event):
assert isinstance(event, zp.namedict)
self.event_count += 1
if(event.dt >= self.market_close):
self.handle_market_close()
if not pandas.isnull(event.TRANSACTION):
if event.TRANSACTION:
self.txn_count += 1
self.cumulative_performance.execute_transaction(event.TRANSACTION)
self.todays_performance.execute_transaction(event.TRANSACTION)
# we're adding a 10% cushion to the capital used,
# and then rounding to the nearest 5k
transaction_cost = event.TRANSACTION.price * event.TRANSACTION.amount
self.cumulative_capital_used += transaction_cost
if math.fabs(self.cumulative_capital_used) > self.max_capital_used:
self.max_capital_used = math.fabs(self.cumulative_capital_used)
cushioned_capital = 1.1 * self.max_capital_used
self.max_capital_used = self.round_to_nearest(
cushioned_capital,
base=5000
)
self.max_leverage = self.max_capital_used / self.capital_base
#update last sale
self.cumulative_performance.update_last_sale(event)
self.todays_performance.update_last_sale(event)
def handle_market_close(self):
#calculate performance as of last trade
@@ -281,6 +257,7 @@ class PerformanceTracker():
trading_environment=self.trading_environment
)
# increment the day counter before we move markers forward.
self.day_count += 1.0
# calculate progress of test
@@ -305,7 +282,8 @@ class PerformanceTracker():
self.todays_performance = PerformancePeriod(
self.todays_performance.positions,
self.todays_performance.ending_value,
self.todays_performance.ending_cash
self.todays_performance.ending_cash,
keep_transactions = True
)
def handle_simulation_end(self):
@@ -314,13 +292,14 @@ class PerformanceTracker():
and send it out on the result_stream.
"""
log_msg = "Simulated {n} trading days out of {m}."
qutil.LOGGER.info(log_msg.format(n=self.day_count, m=self.total_days))
qutil.LOGGER.info("first open: {d}".format(d=self.trading_environment.first_open))
# the stream will end on the last trading day, but will not trigger
# an end of day, so we trigger the final market close here.
self.handle_market_close()
log_msg = "Simulated {n} trading days out of {m}."
qutil.LOGGER.info(log_msg.format(n=self.day_count, m=self.total_days))
qutil.LOGGER.info("first open: {d}".format(d=self.trading_environment.first_open))
self.risk_report = risk.RiskReport(
self.returns,
@@ -335,9 +314,6 @@ class PerformanceTracker():
# this signals that the simulation is complete.
self.result_stream.send("DONE")
def round_to_nearest(self, x, base=5):
return int(base * round(float(x)/base))
class Position():
@@ -387,15 +363,13 @@ class Position():
'sid' : self.sid,
'amount' : self.amount,
'cost_basis' : self.cost_basis,
'last_sale_price' : self.last_sale_price,
'last_sale_date' : self.last_sale_date,
'timestamp' : datetime.datetime.now(),
'last_sale_price' : self.last_sale_price
}
class PerformancePeriod():
def __init__(self, initial_positions, starting_value, starting_cash):
def __init__(self, initial_positions, starting_value, starting_cash, keep_transactions=False):
self.ending_value = 0.0
self.period_capital_used = 0.0
self.pnl = 0.0
@@ -405,6 +379,10 @@ class PerformancePeriod():
#cash balance at start of period
self.starting_cash = starting_cash
self.ending_cash = starting_cash
self.keep_transactions = keep_transactions
self.processed_transactions = []
self.cumulative_capital_used = 0.0
self.max_capital_used = 0.0
self.calculate_performance()
@@ -422,10 +400,41 @@ class PerformancePeriod():
self.returns = 0.0
def execute_transaction(self, txn):
# Update Position
# ----------------
if(not self.positions.has_key(txn.sid)):
self.positions[txn.sid] = Position(txn.sid)
self.positions[txn.sid].update(txn)
self.period_capital_used += -1 * txn.price * txn.amount
# Max Leverage
# ---------------
# Calculate the maximum capital used and maximum leverage
transaction_cost = txn.price * txn.amount
self.cumulative_capital_used += transaction_cost
if math.fabs(self.cumulative_capital_used) > self.max_capital_used:
self.max_capital_used = math.fabs(self.cumulative_capital_used)
# We want to conveye a level, rather than a precise figure.
# round to the nearest 5,000 to keep the number easy on the eyes
self.max_capital_used = self.round_to_nearest(
self.max_capital_used,
base=5000
)
# we're adding a 10% cushion to the capital used.
self.max_leverage = 1.1 * self.max_capital_used / self.starting_cash
# add transaction to the list of processed transactions
if self.keep_transactions:
self.processed_transactions.append(txn)
def round_to_nearest(self, x, base=5):
return int(base * round(float(x)/base))
def calculate_positions_value(self):
mktValue = 0.0
@@ -444,7 +453,8 @@ class PerformancePeriod():
Creates a dictionary representing the state of this performance
period. See header comments for a detailed description.
"""
positions = self.get_positions()
positions = self.get_positions_list()
transactions = [x.as_dict() for x in self.processed_transactions]
return {
'ending_value' : self.ending_value,
@@ -454,9 +464,9 @@ class PerformancePeriod():
'ending_cash' : self.ending_cash,
'portfolio_value': self.ending_cash + self.ending_value,
'positions' : positions,
'timestamp' : datetime.datetime.now(),
'pnl' : self.pnl,
'returns' : self.returns
'returns' : self.returns,
'transactions' : transactions,
}
def to_namedict(self):
@@ -490,6 +500,14 @@ class PerformancePeriod():
return positions
#
def get_positions_list(self):
positions = []
for sid, pos in self.positions.iteritems():
cur = pos.to_dict()
positions.append(cur)
return positions
zipline/finance/risk.py
+4 -4
View File
@@ -352,10 +352,10 @@ class RiskReport():
provided for each period.
"""
return {
'1_month' : [x.to_dict() for x in self.month_periods],
'3_month' : [x.to_dict() for x in self.three_month_periods],
'6_month' : [x.to_dict() for x in self.six_month_periods],
'12_month' : [x.to_dict() for x in self.year_periods]
'one_month' : [x.to_dict() for x in self.month_periods],
'three_month' : [x.to_dict() for x in self.three_month_periods],
'six_month' : [x.to_dict() for x in self.six_month_periods],
'twelve_month' : [x.to_dict() for x in self.year_periods]
}
def periodsInRange(self, months_per, start, end):
zipline/finance/trading.py
+203 -81
View File
@@ -2,6 +2,9 @@ import datetime
import pytz
import math
import pandas
import time
from collections import Counter
# from gevent.select import select
from zmq.core.poll import select
@@ -11,6 +14,17 @@ import zipline.util as qutil
import zipline.protocol as zp
import zipline.finance.performance as perf
from zipline.protocol_utils import Enum, namedict
# the simulation style enumerates the available transaction simulation
# strategies.
SIMULATION_STYLE = Enum(
'PARTIAL_VOLUME',
'BUY_ALL',
'FIXED_SLIPPAGE',
'NOOP'
)
class TradeSimulationClient(qmsg.Component):
def __init__(self, trading_environment):
@@ -19,10 +33,13 @@ class TradeSimulationClient(qmsg.Component):
self.prev_dt = None
self.event_queue = None
self.txn_count = 0
self.order_count = 0
self.trading_environment = trading_environment
self.current_dt = trading_environment.period_start
self.last_iteration_dur = datetime.timedelta(seconds=0)
self.algorithm = None
self.max_wait = datetime.timedelta(seconds=7)
self.last_msg_dt = datetime.datetime.utcnow()
assert self.trading_environment.frame_index != None
self.event_frame = pandas.DataFrame(
@@ -47,8 +64,11 @@ class TradeSimulationClient(qmsg.Component):
def open(self):
self.result_feed = self.connect_result()
self.order_socket = self.connect_order()
# send a wake up call to the order data source.
self.order_socket.send(str(zp.ORDER_PROTOCOL.BREAK))
def do_work(self):
# poll all the sockets
socks = dict(self.poll.poll(self.heartbeat_timeout))
@@ -56,6 +76,8 @@ class TradeSimulationClient(qmsg.Component):
if self.result_feed in socks and \
socks[self.result_feed] == self.zmq.POLLIN:
self.last_msg_dt = datetime.datetime.utcnow()
# get the next message from the result feed
msg = self.result_feed.recv()
@@ -65,8 +87,7 @@ class TradeSimulationClient(qmsg.Component):
# signal the performance tracker that the simulation has
# ended. Perf will internally calculate the full risk report.
self.perf.handle_simulation_end()
# shutdown the feedback loop to the OrderDataSource
self.signal_order_done()
# signal Simulator, our ComponentHost, that this component is
# done and Simulator needn't block exit on this component.
self.signal_done()
@@ -74,13 +95,21 @@ class TradeSimulationClient(qmsg.Component):
# result_feed is a merge component, so unframe accordingly
event = zp.MERGE_UNFRAME(msg)
self.received_count += 1
# update performance and relay the event to the algorithm
self.process_event(event)
# signal done to order source.
# signal loop is done for order source.
self.order_socket.send(str(zp.ORDER_PROTOCOL.BREAK))
else:
# no events in the sock means the non-order sources are
# drained. Signal the order_source that we're done, and
# the done will cascade through the whole zipline.
# shutdown the feedback loop to the OrderDataSource
wait_time = datetime.datetime.utcnow() - self.last_msg_dt
if wait_time > self.max_wait:
self.signal_order_done()
def process_event(self, event):
# track the number of transactions, for testing purposes.
if(event.TRANSACTION != None):
@@ -107,12 +136,15 @@ class TradeSimulationClient(qmsg.Component):
# otherwise, the algorithm has fallen behind the feed
# and processing per event is longer than time between events.
if event.dt >= self.current_dt:
# compress time by moving the current_time up to the event
# time.
self.current_dt = event.dt
self.run_algorithm()
# tally the time spent on this iteration
self.last_iteration_dur = datetime.datetime.utcnow() - event_start
# move the algorithm's clock forward to include iteration time
self.current_dt = self.current_dt + self.last_iteration_dur
self.current_dt = self.current_dt + self.last_iteration_dur
def run_algorithm(self):
@@ -132,13 +164,15 @@ class TradeSimulationClient(qmsg.Component):
return self.connect_push_socket(self.addresses['order_address'])
def order(self, sid, amount):
order = zp.namedict({
'dt':self.current_dt,
'sid':sid,
'amount':amount
})
self.order_socket.send(zp.ORDER_FRAME(order))
self.order_count += 1
self.perf.log_order(order)
def signal_order_done(self):
self.order_socket.send(str(zp.ORDER_PROTOCOL.DONE))
@@ -172,19 +206,13 @@ class OrderDataSource(qmsg.DataSource):
"""
qmsg.DataSource.__init__(self, zp.FINANCE_COMPONENT.ORDER_SOURCE)
self.sent_count = 0
self.recv_count = Counter()
self.works = 0
@property
def get_type(self):
return zp.DATASOURCE_TYPE.ORDER
#
@property
def is_blocking(self):
"""
This datasource is in a loop with the TradingSimulationClient
"""
return False
def open(self):
qmsg.DataSource.open(self)
self.order_socket = self.bind_order()
@@ -194,23 +222,21 @@ class OrderDataSource(qmsg.DataSource):
def do_work(self):
#TODO: if this is the first iteration, break deadlock by sending a dummy order
if(self.sent_count == 0):
self.send(zp.namedict({}))
#pull all orders from client.
orders = []
count = 0
self.works += 1
# TODO : this can be written in a concurrency agnostic
# way... have a chat with Fawce about this ~Steve
#pull all orders from client.
count = 0
# one iteration of the client could include several orders
# so iterate until the client signals a break or a close.
while True:
# poll all the sockets
# we reduce the timeout here by a factor of 2, because we need
# to potentially receive the client's done message before the
# controller or heartbeat times out.
# this will block for timeout/2, and return an empty dict if there
# are no messages.
socks = dict(self.poll.poll(self.heartbeat_timeout/2))
# see if the poller has results for the result_feed
@@ -220,39 +246,46 @@ class OrderDataSource(qmsg.DataSource):
order_msg = self.order_socket.recv()
if order_msg == str(zp.ORDER_PROTOCOL.DONE):
qutil.LOGGER.info("order source is done")
self.signal_done()
self.recv_count['done'] += 1
return
if order_msg == str(zp.ORDER_PROTOCOL.BREAK):
# send a blank message to avoid an empty buffer
# in the feed
self.recv_count['break'] += 1
if count == 0:
self.send(namedict({}))
break
order = zp.ORDER_UNFRAME(order_msg)
self.recv_count['order'] += 1
#send the order along
self.send(order)
count += 1
self.sent_count += 1
else:
# no orders, break out
break
#TODO: we have to send at least one dummy order per do_work iteration
# or the feed will block waiting for our messages.
if(count == 0):
self.send(zp.namedict({}))
class TransactionSimulator(qmsg.BaseTransform):
def __init__(self):
def __init__(self, style=SIMULATION_STYLE.PARTIAL_VOLUME):
qmsg.BaseTransform.__init__(self, zp.TRANSFORM_TYPE.TRANSACTION)
self.open_orders = {}
self.order_count = 0
self.txn_count = 0
self.trade_window = datetime.timedelta(seconds=30)
self.trade_window = datetime.timedelta(seconds=30)
self.orderTTL = datetime.timedelta(days=1)
self.volume_share = 0.05
self.commission = 0.03
if not style or style == SIMULATION_STYLE.PARTIAL_VOLUME:
self.apply_trade_to_open_orders = self.simulate_with_partial_volume
elif style == SIMULATION_STYLE.BUY_ALL:
self.apply_trade_to_open_orders = self.simulate_buy_all
elif style == SIMULATION_STYLE.FIXED_SLIPPAGE:
self.apply_trade_to_open_orders = self.simulate_with_fixed_cost
elif style == SIMULATION_STYLE.NOOP:
self.apply_trade_to_open_orders = self.simulate_noop
def transform(self, event):
"""
@@ -267,9 +300,12 @@ class TransactionSimulator(qmsg.BaseTransform):
self.state['value'] = txn
else:
self.state['value'] = None
qutil.LOGGER.info("unexpected event type in transform: {etype}".format(etype=event.type))
log = "unexpected event type in transform: {etype}".format(
etype=event.type
)
qutil.LOGGER.info(log)
#TODO: what to do if we get another kind of datasource event.type?
return self.state
def add_open_order(self, event):
@@ -277,63 +313,141 @@ class TransactionSimulator(qmsg.BaseTransform):
Amount is explicitly converted to an int.
Orders of amount zero are ignored.
"""
self.order_count += 1
event.amount = int(event.amount)
if event.amount == 0:
qutil.LOGGER.debug("requested to trade zero shares of {sid}".format(sid=event.sid))
log = "requested to trade zero shares of {sid}".format(
sid=event.sid
)
qutil.LOGGER.debug(log)
return
self.order_count += 1
if(not self.open_orders.has_key(event.sid)):
self.open_orders[event.sid] = []
# set the filled property to zero
event.filled = 0
self.open_orders[event.sid].append(event)
def apply_trade_to_open_orders(self, event):
def simulate_buy_all(self, event):
txn = self.create_transaction(
event.sid,
event.volume,
event.price,
event.dt,
1
)
return txn
if(event.volume == 0):
#there are zero volume events bc some stocks trade
#less frequently than once per minute.
return self.create_dummy_txn(event.dt)
def simulate_noop(self, event):
return None
def simulate_with_fixed_cost(self, event):
if self.open_orders.has_key(event.sid):
orders = self.open_orders[event.sid]
orders = sorted(orders, key=lambda o: o.dt)
else:
return None
remaining_orders = []
total_order = 0
dt = event.dt
amount = 0
for order in orders:
#we're using minute bars, so allow orders within
#30 seconds of the trade
if((order.dt - event.dt) < self.trade_window):
total_order += order.amount
if(order.dt > dt):
dt = order.dt
#if the order still has time to live (TTL) keep track
elif((self.algo_time - order.dt) < self.orderTTL):
remaining_orders.append(order)
self.open_orders[event.sid] = remaining_orders
if(total_order != 0):
direction = total_order / math.fabs(total_order)
else:
direction = 1
amount += order.amount
if(amount == 0):
return
volume_share = (direction * total_order) / event.volume
if volume_share > .25:
volume_share = .25
amount = volume_share * event.volume * direction
impact = (volume_share)**2 * .1 * direction * event.price
return self.create_transaction(
event.sid,
amount,
event.price + impact,
dt.replace(tzinfo = pytz.utc),
direction
)
direction = amount / math.fabs(amount)
txn = self.create_transaction(
event.sid,
amount,
event.price + 0.10,
event.dt,
direction
)
self.open_orders[event.sid] = []
return txn
def simulate_with_partial_volume(self, event):
if(event.volume == 0):
#there are zero volume events bc some stocks trade
#less frequently than once per minute.
return None
if self.open_orders.has_key(event.sid):
orders = self.open_orders[event.sid]
orders = sorted(orders, key=lambda o: o.dt)
else:
return None
dt = event.dt
expired = []
total_order = 0
simulated_amount = 0
simulated_impact = 0.0
direction = 1.0
for order in orders:
if(order.dt < event.dt):
# orders are only good on the day they are issued
if order.dt.day < event.dt.day:
continue
open_amount = order.amount - order.filled
if(open_amount != 0):
direction = open_amount / math.fabs(open_amount)
else:
direction = 1
desired_order = total_order + open_amount
volume_share = direction * (desired_order) / event.volume
if volume_share > .25:
volume_share = .25
simulated_amount = int(volume_share * event.volume * direction)
simulated_impact = (volume_share)**2 * .1 * direction * event.price
order.filled += (simulated_amount - total_order)
total_order = simulated_amount
# we cap the volume share at 25% of a trade
if volume_share == .25:
break
orders = [ x for x in orders if abs(x.amount - x.filled) > 0 and x.dt.day >= event.dt.day]
self.open_orders[event.sid] = orders
if simulated_amount != 0:
return self.create_transaction(
event.sid,
simulated_amount,
event.price + simulated_impact,
dt.replace(tzinfo = pytz.utc),
direction
)
else:
warning = """
Calculated a zero volume transaction on trade:
{event}
for orders:
{orders}
"""
warning = warning.format(
event=str(event),
orders=str(orders)
)
qutil.LOGGER.warn(warning)
return None
def create_transaction(self, sid, amount, price, dt, direction):
@@ -445,7 +559,15 @@ class TradingEnvironment(object):
return len(self.period_trading_days)
def is_market_hours(self, test_date):
if not self.is_trading_day(test_date):
return False
mkt_open = self.set_NYSE_time(test_date, 9, 30)
#TODO: half days?
mkt_close = self.set_NYSE_time(test_date, 16, 00)
return test_date >= mkt_open and test_date <= mkt_close
def is_trading_day(self, test_date):
dt = self.normalize_date(test_date)
zipline/lines.py
+31 -16
View File
@@ -90,7 +90,7 @@ from zipline.finance.trading import TransactionSimulator, OrderDataSource, \
TradeSimulationClient
from zipline.simulator import AddressAllocator, Simulator
from zipline.monitor import Controller
from zipline.finance.trading import SIMULATION_STYLE
class SimulatedTrading(object):
@@ -125,11 +125,15 @@ class SimulatedTrading(object):
:py:class:`zipline.simulator.AddressAllocator`
- simulator_class: a :py:class:`zipline.messaging.ComponentHost`
subclass (not an instance)
- simulation_style: optional parameter that configures the
:py:class:`zipline.finance.trading.TransactionSimulator`. Expects
a SIMULATION_STYLE as defined in :py:mod:`zipline.finance.trading`
"""
assert isinstance(config, dict)
self.algorithm = config['algorithm']
self.allocator = config['allocator']
self.trading_environment = config['trading_environment']
self.sim_style = config.get('simulation_style')
self.leased_sockets = []
self.sim_context = None
@@ -169,7 +173,7 @@ class SimulatedTrading(object):
self.add_source(self.order_source)
#setup transforms
self.transaction_sim = TransactionSimulator()
self.transaction_sim = TransactionSimulator(self.sim_style)
self.transforms = {}
self.add_transform(self.transaction_sim)
@@ -191,16 +195,20 @@ class SimulatedTrading(object):
- sid - an integer, which will be used as the security ID.
- order_count - the number of orders the test algo will place,
defaults to 100
- trade_count - the number of trades to simulate, defaults to 100
- order_amount - the number of shares per order, defaults to 100
- trade_count - the number of trades to simulate, defaults to 101
to ensure all orders are processed.
- simulator_class - optional parameter that provides an alternative
subclass of ComponentHost to hold the whole zipline. Defaults to
:py:class:`zipline.simulator.Simulator`
- algorithm - optional parameter providing an algorithm. defaults
to :py:class:`zipline.test.algorithms.TestAlgorithm`
- random - optional parameter to request random trades. if present
:py:class:`zipline.sources.RandomEquityTrades` is the source. If
not :py:class:`ziplien.sources.SpecificEquityTrades` is the
source
- trade_source - optional parameter to specify trades, if present.
If not present :py:class:`ziplien.sources.SpecificEquityTrades`
is the source, with daily frequency in trades.
- simulation_style: optional parameter that configures the
:py:class:`zipline.finance.trading.TransactionSimulator`. Expects
a SIMULATION_STYLE as defined in :py:mod:`zipline.finance.trading`
"""
assert isinstance(config, dict)
@@ -219,28 +227,35 @@ class SimulatedTrading(object):
order_count = config['order_count']
else:
order_count = 100
if config.has_key('order_amount'):
order_amount = config['order_amount']
else:
order_amount = 100
if config.has_key('trade_count'):
trade_count = config['trade_count']
else:
trade_count = 100
# to ensure all orders are filled, we provide one more
# trade than order
trade_count = 101
if config.has_key('simulator_class'):
simulator_class = config['simulator_class']
else:
simulator_class = Simulator
simulation_style = config.get('simulation_style')
if not simulation_style:
simulation_style = SIMULATION_STYLE.FIXED_SLIPPAGE
#-------------------
# Trade Source
#-------------------
sids = [sid]
#-------------------
if config.has_key('random'):
trade_source = factory.create_random_trade_source(
sids,
trade_count,
trading_environment
)
if config.has_key('trade_source'):
trade_source = config['trade_source']
else:
trade_source = factory.create_daily_trade_source(
sids,
@@ -253,7 +268,6 @@ class SimulatedTrading(object):
if config.has_key('algorithm'):
test_algo = config['algorithm']
else:
order_amount = 100
test_algo = TestAlgorithm(
sid,
order_amount,
@@ -266,7 +280,8 @@ class SimulatedTrading(object):
'algorithm':test_algo,
'trading_environment':trading_environment,
'allocator':allocator,
'simulator_class':simulator_class
'simulator_class':simulator_class,
'simulation_style':simulation_style
})
#-------------------
zipline/messaging.py
+13 -17
View File
@@ -4,6 +4,8 @@ Commonly used messaging components.
import datetime
from collections import Counter
import zipline.util as qutil
from zipline.component import Component
import zipline.protocol as zp
@@ -37,7 +39,7 @@ class ComponentHost(Component):
# ----------------------
self.sync_register = {}
self.timeout = datetime.timedelta(seconds=5)
self.timeout = datetime.timedelta(seconds=60)
self.feed = Feed()
self.merge = Merge()
@@ -82,12 +84,8 @@ class ComponentHost(Component):
if isinstance(component, DataSource):
self.feed.add_source(component.get_id)
if not component.is_blocking:
self.feed.ds_finished_counter +=1
if isinstance(component, BaseTransform):
self.merge.add_source(component.get_id)
if not component.is_blocking:
self.feed.ds_finished_counter +=1
def unregister_component(self, component_id):
del self.components[component_id]
@@ -192,7 +190,11 @@ class Feed(Component):
# Depending on the size of this, might want to use a data
# structure with better asymptotics.
self.data_buffer = {}
# source_id -> integer count
self.sent_counters = Counter()
self.recv_counters = Counter()
def init(self):
pass
@@ -214,7 +216,7 @@ class Feed(Component):
def do_work(self):
# wait for synchronization reply from the host
socks = dict(self.poll.poll(self.heartbeat_timeout)) #timeout after 2 seconds.
socks = dict(self.poll.poll(self.heartbeat_timeout))
# TODO: Abstract this out, maybe on base component
if self.control_in in socks and socks[self.control_in] == self.zmq.POLLIN:
@@ -294,6 +296,7 @@ class Feed(Component):
event = self.next()
if(event != None):
self.feed_socket.send(self.frame(event), self.zmq.NOBLOCK)
self.sent_counters[event.source_id] += 1
self.sent_count += 1
def append(self, event):
@@ -302,6 +305,7 @@ class Feed(Component):
source_id.
"""
self.data_buffer[event.source_id].append(event)
self.recv_counters[event.source_id] += 1
self.received_count += 1
def next(self):
@@ -336,9 +340,9 @@ class Feed(Component):
def is_full(self):
"""
Indicates whether the buffer has messages in buffer for
all un-DONE sources.
all un-DONE, blocking sources.
"""
for events in self.data_buffer.values():
for source_id, events in self.data_buffer.iteritems():
if len(events) == 0:
return False
return True
@@ -463,10 +467,6 @@ class BaseTransform(Component):
def get_type(self):
return COMPONENT_TYPE.CONDUIT
@property
def is_blocking(self):
return True
def open(self):
"""
Establishes zmq connections.
@@ -615,10 +615,6 @@ class DataSource(Component):
@property
def get_id(self):
return self.id
@property
def is_blocking(self):
return True
@property
def get_type(self):
zipline/protocol.py
+50 -71
View File
@@ -124,6 +124,7 @@ import copy
from collections import namedtuple
from protocol_utils import Enum, FrameExceptionFactory, namedict
from date_utils import EPOCH, UN_EPOCH
#import ujson
#import ultrajson_numpy
@@ -615,96 +616,74 @@ def PERF_FRAME(perf):
"""
Frame the performance update created at the end of each simulated trading
day. The msgpack is a tuple with the first element statically set to 'PERF'.
Frames prepared by this method are sent via the same socket as
Frames prepared by RISK_FRAME. So, both methods prefix the payload with
a shorthand for their type. That way, all messages received from the socket
can be PERF_UNFRAMED(), whether they are risk or perf.
Like RISK_FRAME, this method calls BT_UPDATE_FRAME internally, so that
clients can call BT_UPDATE_UNFRAME for all messages from the backtest.
:param perf: the dictionary created by zipline.trade_client.perf
:rvalue: a msgpack string
"""
#TODO: add asserts...
assert isinstance(perf['started_at'], datetime.datetime)
assert isinstance(perf['period_start'], datetime.datetime)
assert isinstance(perf['period_end'], datetime.datetime)
assert isinstance(perf['last_close'], datetime.datetime)
assert isinstance(perf['last_open'], datetime.datetime)
#pull some special fields from the perf for easy access
date = perf['last_close']
tp = perf['todays_perf']
assert isinstance(perf['daily_perf'], dict)
assert isinstance(perf['cumulative_perf'], dict)
tp = perf['daily_perf']
cp = perf['cumulative_perf']
risk = perf['cumulative_risk_metrics']
daily_perf = {
'date' : EPOCH(date),
'returns' : tp['returns'],
'pnl' : tp['pnl'],
'market_value' : tp['ending_value'],
'portfolio_value' : tp['portfolio_value'],
'starting_cash' : tp['starting_cash'],
'ending_cash' : tp['ending_cash'],
'capital_used' : tp['capital_used']
}
cumulative_perf = {
'alpha' : risk['alpha'],
'beta' : risk['beta'],
'sharpe' : risk['sharpe'],
'volatility' : risk['algo_volatility'],
'benchmark_volatility' : risk['benchmark_volatility'],
'benchmark_returns' : risk['benchmark_period_return'],
'max_drawdown' : risk['max_drawdown'],
'total_returns' : cp['returns'],
'pnl' : cp['pnl'],
'capital_used' : cp['capital_used']
}
# nest the cumulative performance data in the daily.
daily_perf['cumulative'] = cumulative_perf
result = {
'started_at' : EPOCH(perf['started_at']),
'daily' : [daily_perf],
'percent_complete' : perf['progress'],
}
assert isinstance(tp['transactions'], list)
assert isinstance(cp['transactions'], list)
assert isinstance(tp['positions'], list)
assert isinstance(cp['positions'], list)
perf['started_at'] = EPOCH(perf['started_at'])
perf['period_start'] = EPOCH(perf['period_start'])
perf['period_end'] = EPOCH(perf['period_end'])
perf['last_close'] = EPOCH(perf['last_close'])
perf['last_open'] = EPOCH(perf['last_open'])
return msgpack.dumps(tuple(['PERF', result]))
tp['transactions'] = convert_transactions(tp['transactions'])
cp['transactions'] = convert_transactions(cp['transactions'])
return BT_UPDATE_FRAME('PERF', perf)
def convert_transactions(transactions):
results = []
for txn in transactions:
txn['date'] = EPOCH(txn['dt'])
del(txn['dt'])
results.append(txn)
return results
def RISK_FRAME(risk):
return msgpack.dumps(tuple(['RISK', risk]))
return BT_UPDATE_FRAME('RISK', risk)
def PERF_UNFRAME(msg):
prefix, payload = msgpack.loads(msg)
def BT_UPDATE_FRAME(prefix, payload):
"""
Frames prepared by RISK_FRAME and PERF_FRAME methods are sent via the same
socket. This method provides a prefix to allow for muxing the messages
onto a single socket.
"""
return msgpack.dumps(tuple([prefix, payload]))
def BT_UPDATE_UNFRAME(msg):
"""
Risk and Perf framing methods prefix the payload with
a shorthand for their type. That way, all messages received from the socket
can be PERF_FRAMED(), whether they are risk or perf.
"""
prefix, payload = msgpack.loads(msg, use_list=True)
return dict(prefix=prefix, payload=payload)
# -----------------------
# Date Helpers
# -----------------------
UNIX_EPOCH = datetime.datetime(1970, 1, 1, 0, 0, tzinfo = pytz.utc)
def EPOCH(utc_datetime):
"""
The key is to ensure all the dates you are using are in the utc timezone
before you start converting. See http://pytz.sourceforge.net/ to learn how
to do that properly. By normalizing to utc, you eliminate the ambiguity of
daylight savings transitions. Then you can safely use timedelta to calculate
distance from the unix epoch, and then convert to seconds or milliseconds.
Note that the resulting unix timestamp is itself in the UTC timezone. If you
wish to see the timestamp in a localized timezone, you will need to make
another conversion.
Also note that this will only work for dates after 1970.
"""
assert isinstance(utc_datetime, datetime.datetime)
# utc only please
assert utc_datetime.tzinfo == pytz.utc
# how long since the epoch?
delta = utc_datetime - UNIX_EPOCH
seconds = delta.total_seconds()
ms = seconds * 1000
return ms
def PACK_DATE(event):
"""
zipline/protocol_utils.py
+129 -2
View File
@@ -1,6 +1,8 @@
import copy
import pandas
from ctypes import Structure, c_ubyte
from collections import MutableMapping
from itertools import izip
def Enum(*options):
"""
@@ -28,7 +30,7 @@ def FrameExceptionFactory(name):
return InvalidFrame
class namedict(object):
class namedict(MutableMapping):
"""
Namedicts are dict like objects that have fields accessible by attribute lookup
@@ -61,6 +63,15 @@ class namedict(object):
def __getitem__(self, key):
return self.__dict__[key]
def __delitem__(self, key):
del self.__dict__[key]
def __iter__(self):
return self.__dict__.iterkeys()
def __len__(self):
return len(self.__dict__)
def keys(self):
return self.__dict__.keys()
@@ -86,8 +97,124 @@ class namedict(object):
def has_attr(self, name):
return self.__dict__.has_key(name)
def as_series(self):
s = pandas.Series(self.__dict__)
s.name = self.sid
return s
class ndict(MutableMapping):
"""
Xtreme Namedicts 2.0
Ndicts are dict like objects that have fields accessible by attribute
lookup as well as being indexable and iterable. Done right
this time.
"""
def __init__(self, dct=None):
self.__internal = dict()
self.cls = frozenset(dir(self))
if dct:
self.__internal.update(dct)
# Abstact Overloads
# -----------------
def __setitem__(self, key, value):
"""
Required for use by pymongo as_class parameter to find.
"""
if key == '_id':
self.__internal['id'] = value
else:
self.__internal[key] = value
def __getattr__(self, key):
if key in self.cls:
return self.__dict__[key]
else:
return self.__internal[key]
def __getitem__(self, key):
return self.__internal[key]
def __delitem__(self, key):
del self.__internal[key]
def __iter__(self):
return self.__internal.iterkeys()
def __len__(self):
return len(self.__internal)
# Compatability with namedicts
# ----------------------------
# for compat, not the Python way to do things though...
# Deprecated, use builtin ``del`` operator.
delete = __delitem__
def has_attr(self, key):
"""
Deprecated, use builtin ``in`` operator.
"""
return self.__contains__(key)
def has_key(self, key):
return self.__contains__(key)
# Custom Methods
# --------------
def copy(self):
return ndict(copy.copy(self.__internal))
def as_dataframe(self):
"""
Return the representation as a Pandas dataframe.
"""
d = pandas.DataFrame(self.__internal)
return d
def as_series(self):
"""
Return the representation as a Pandas time series.
"""
s = pandas.Series(self.__internal)
s.name = self.sid
return s
def as_dict(self):
"""
Return the representation as a vanilla Python dict.
"""
# shallow copy is O(n)
return copy.copy(self.__internal)
def merge(self, other_nd):
"""
Merge in place with another ndict.
"""
assert isinstance(other_nd, ndict)
self.__internal.update(other_nd.__internal)
def __repr__(self):
return "namedict: " + str(self.__internal)
# Faster dictionary comparison?
#def __eq__(self, other):
#assert isinstance(other, ndict)
#keyeq = set(self.keys()) == set(other.keys())
#if not keyeq:
#return False
#for i, j in izip(self.itervalues(), other.itervalues()):
#if i != j:
#return False
#return True
zipline/sources.py
+1
View File
@@ -94,6 +94,7 @@ class SpecificEquityTrades(TradeDataSource):
def get_type(self):
zp.COMPONENT_TYPE.SOURCE
def do_work(self):
if(len(self.event_list) == 0):
self.signal_done()
zipline/test/algorithms.py
+34 -2
View File
@@ -70,14 +70,46 @@ class TestAlgorithm():
def handle_frame(self, frame):
self.frame_count += 1
#place an order for 100 shares of sid:133
#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]
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 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.
zipline/test/factory.py
+35 -8
View File
@@ -42,13 +42,13 @@ def load_market_data():
tr_curves[tr_dt] = curve
return bm_returns, tr_curves
def create_trading_environment():
def create_trading_environment(year=2006):
"""Construct a complete environment with reasonable defaults"""
benchmark_returns, treasury_curves = load_market_data()
start = datetime(2006, 1, 1, tzinfo=pytz.utc)
end = datetime(2006, 12, 31, tzinfo=pytz.utc)
start = datetime(year, 1, 1, tzinfo=pytz.utc)
end = datetime(year, 12, 31, tzinfo=pytz.utc)
trading_environment = TradingEnvironment(
benchmark_returns,
treasury_curves,
@@ -73,7 +73,7 @@ def get_next_trading_dt(current, interval, trading_calendar):
next = current
while True:
next = next + interval
if trading_calendar.is_trading_day(next):
if trading_calendar.is_market_hours(next):
break
return next
@@ -83,10 +83,10 @@ def create_trade_history(sid, prices, amounts, interval, trading_calendar):
current = trading_calendar.first_open
for price, amount in zip(prices, amounts):
current = get_next_trading_dt(current, interval, trading_calendar)
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
@@ -172,6 +172,7 @@ def create_random_trade_source(sid, trade_count, trading_environment):
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
@@ -181,12 +182,38 @@ def create_daily_trade_source(sids, trade_count, trading_environment):
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
trade_time_increment = timedelta(days=1)
generated_trades = create_trade_history(
sid,
zipline/test/test_finance.py
+266 -4
View File
@@ -21,8 +21,12 @@ 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)
@@ -103,12 +107,21 @@ class FinanceTestCase(TestCase):
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
# ----------
zipline = SimulatedTrading.create_test_zipline(**self.zipline_test_config)
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())
@@ -118,8 +131,62 @@ class FinanceTestCase(TestCase):
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.
@@ -204,7 +271,9 @@ class FinanceTestCase(TestCase):
self.zipline_test_config['trade_count'] = 200
self.zipline_test_config['algorithm'] = test_algo
zipline = SimulatedTrading.create_test_zipline(**self.zipline_test_config)
zipline = SimulatedTrading.create_test_zipline(
**self.zipline_test_config
)
zipline.simulate(blocking=True)
#check that the algorithm received no events
@@ -214,8 +283,201 @@ class FinanceTestCase(TestCase):
"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)
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']
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
)
for i in range(order_count):
order = namedict(
{
'sid':sid,
'amount':order_amount,
'type':zp.DATASOURCE_TYPE.ORDER,
'dt' : start_date + i * order_interval
})
sim_state = trade_sim.transform(order)
# there should not be a new transaction from an order.
self.assertTrue(sim_state['name'] == trade_sim.get_id)
self.assertTrue(sim_state['value'] == None)
# 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 order in order_list:
self.assertEqual(order.sid, sid)
self.assertEqual(order.amount, order_amount)
tracker = PerformanceTracker(trading_environment)
transactions = []
for trade in generated_trades:
if trade_delay:
trade.dt = trade.dt + trade_delay
sim_state = trade_sim.transform(trade)
self.assertEqual(sim_state['name'], trade_sim.get_id)
txn = None
if sim_state['value']:
txn = sim_state['value']
transactions.append(txn)
trade[sim_state['name']] = txn
tracker.process_event(trade)
total_volume = 0
for txn in transactions:
total_volume += 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))
zipline/test/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
zipline/test/test_perf_tracking.py
+9 -4
View File
@@ -22,8 +22,15 @@ class PerformanceTestCase(unittest.TestCase):
0,
len(self.treasury_curves)
)
self.dt = self.treasury_curves.keys()[random_index]
self.end_dt = self.dt + datetime.timedelta(days=365)
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,
@@ -505,8 +512,6 @@ shares in position"
price = 10.1
price_list = [price] * trade_count
volume = [100] * trade_count
#start_date = datetime.datetime.strptime("01/01/2011","%m/%d/%Y")
#start_date = start_date.replace(tzinfo=pytz.utc)
trade_time_increment = datetime.timedelta(days=1)
trade_history = factory.create_trade_history(
sid,