diff --git a/etc/requirements.txt b/etc/requirements.txt index 318942c9..679b1315 100644 --- a/etc/requirements.txt +++ b/etc/requirements.txt @@ -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 \ No newline at end of file diff --git a/zipline/component.py b/zipline/component.py index 0004051f..e4af443c 100644 --- a/zipline/component.py +++ b/zipline/component.py @@ -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): diff --git a/zipline/date_utils.py b/zipline/date_utils.py index 966525c5..14a88571 100644 --- a/zipline/date_utils.py +++ b/zipline/date_utils.py @@ -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 # --------------------- diff --git a/zipline/finance/performance.py b/zipline/finance/performance.py index f62d4f0d..8fd3d3d6 100644 --- a/zipline/finance/performance.py +++ b/zipline/finance/performance.py @@ -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 + diff --git a/zipline/finance/risk.py b/zipline/finance/risk.py index 7fa1d103..fc4bb076 100644 --- a/zipline/finance/risk.py +++ b/zipline/finance/risk.py @@ -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): diff --git a/zipline/finance/trading.py b/zipline/finance/trading.py index 76ad2a19..9a950e61 100644 --- a/zipline/finance/trading.py +++ b/zipline/finance/trading.py @@ -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) diff --git a/zipline/lines.py b/zipline/lines.py index 2123cfdf..43c2eac0 100644 --- a/zipline/lines.py +++ b/zipline/lines.py @@ -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 }) #------------------- diff --git a/zipline/messaging.py b/zipline/messaging.py index 2c4caebc..03764807 100644 --- a/zipline/messaging.py +++ b/zipline/messaging.py @@ -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): diff --git a/zipline/protocol.py b/zipline/protocol.py index f8db09bc..471238a4 100644 --- a/zipline/protocol.py +++ b/zipline/protocol.py @@ -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): """ diff --git a/zipline/protocol_utils.py b/zipline/protocol_utils.py index ba805b3b..8d21b1a0 100644 --- a/zipline/protocol_utils.py +++ b/zipline/protocol_utils.py @@ -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 diff --git a/zipline/sources.py b/zipline/sources.py index b1385754..bf08644c 100644 --- a/zipline/sources.py +++ b/zipline/sources.py @@ -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() diff --git a/zipline/test/algorithms.py b/zipline/test/algorithms.py index 49ec15c0..17873d27 100644 --- a/zipline/test/algorithms.py +++ b/zipline/test/algorithms.py @@ -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. diff --git a/zipline/test/factory.py b/zipline/test/factory.py index b8d0cb04..e0fb89e0 100644 --- a/zipline/test/factory.py +++ b/zipline/test/factory.py @@ -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, diff --git a/zipline/test/test_finance.py b/zipline/test/test_finance.py index 52538f6a..a85f220d 100644 --- a/zipline/test/test_finance.py +++ b/zipline/test/test_finance.py @@ -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)) + + + \ No newline at end of file diff --git a/zipline/test/test_ndict.py b/zipline/test/test_ndict.py new file mode 100644 index 00000000..e2cb8a84 --- /dev/null +++ b/zipline/test/test_ndict.py @@ -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 diff --git a/zipline/test/test_perf_tracking.py b/zipline/test/test_perf_tracking.py index bd6efe7c..be7c9a25 100644 --- a/zipline/test/test_perf_tracking.py +++ b/zipline/test/test_perf_tracking.py @@ -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,