diff --git a/zipline/finance/performance.py b/zipline/finance/performance.py new file mode 100644 index 00000000..df17fe17 --- /dev/null +++ b/zipline/finance/performance.py @@ -0,0 +1,183 @@ +import datetime +import pytz +import math + +from zmq.core.poll import select + +import zipline.messaging as qmsg +import zipline.util as qutil +import zipline.protocol as zp +import zipline.finance.risk as risk + +class PortfolioClient(qmsg.Component): + + def __init__(self, period_start, period_end, capital_base, trading_environment): + qmsg.Component.__init__(self) + self.trading_day = datetime.timedelta(hours=6, minutes=30) + self.calendar_day = datetime.timedelta(hours=24) + self.period_start = period_start + self.period_end = period_end + self.market_open = self.period_start + self.market_close = self.market_open + self.trading_day + self.progress = 0.0 + self.total_days = (self.period_end - self.period_start).days + self.day_count = 0 + self.cumulative_capital_used= 0.0 + self.max_capital_used = 0.0 + self.capital_base = capital_base + self.trading_environment = trading_environment + self.returns = [] + self.cumulative_performance = PerformancePeriod(self.period_start, self.period_end, {}, 0, capital_base = capital_base) + self.todays_performance = PerformancePeriod(self.market_open, self.market_close, {}, 0, capital_base = capital_base) + + @property + def get_id(self): + return str(zp.FINANCE_COMPONENT.PORTFOLIO_CLIENT) + + def open(self): + self.result_feed = self.connect_result() + + def do_work(self): + #next feed event + socks = dict(self.poll.poll(self.heartbeat_timeout)) + + if self.result_feed in socks and socks[self.result_feed] == self.zmq.POLLIN: + msg = self.result_feed.recv() + + if msg == str(zp.CONTROL_PROTOCOL.DONE): + qutil.LOGGER.info("Portfolio Client is DONE!") + self.signal_done() + return + + event = zp.MERGE_UNFRAME(msg) + + if(event.dt >= self.market_close): + self.handle_market_close() + + if event.TRANSACTION != None: + 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 + self.cumulative_capital_used += event.TRANSACTION.price * event.TRANSACTION.amount + if(math.fabs(self.cumulative_capital_used) > self.max_capital_used): + self.max_capital_used = math.fabs(self.cumulative_capital_used) + self.max_capital_used = self.round_to_nearest(1.1 * self.max_capital_used, 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) + + #calculate performance as of last trade + self.cumulative_performance.calculate_performance() + self.todays_performance.calculate_performance() + + + # + def handle_market_close(self): + self.market_open = self.market_open + self.calendar_day + while not self.trading_environment.is_trading_day(self.market_open): + if self.market_open > self.trading_environment.trading_days[-1]: + raise Exception("Attempting to backtest beyond available history.") + self.market_open = self.market_open + self.calendar_day + self.market_close = self.market_open + self.trading_day + self.day_count += 1.0 + self.progress = self.day_count / self.total_days + self.returns.append(risk.daily_return(self.todays_performance.period_end.replace(hour=0, minute=0, second=0), self.todays_performance.returns)) + self.cur_period_metrics = risk.periodmetrics(start_date=self.period_start, + end_date=self.todays_performance.period_end.replace(hour=0, minute=0, second=0), + returns=self.returns, + trading_environment=self.trading_environment) + ############################################### + #######TODO: report/relay metrics here######### + ############################################### + #roll over positions to current day. + self.todays_performance = PerformancePeriod(self.market_open, + self.market_close, + self.todays_performance.positions, + self.todays_performance.ending_value, + self.capital_base) + + # + def round_to_nearest(self, x, base=5): + return int(base * round(float(x)/base)) + +class Position(): + sid = None + amount = None + cost_basis = None + last_sale = None + last_date = None + + def __init__(self, sid): + self.sid = sid + self.amount = 0 + self.cost_basis = 0.0 ##per share + + def update(self, txn): + if(self.sid != txn.sid): + raise NameError('attempt to update position with transaction in different sid') + #throw exception + + if(self.amount + txn.amount == 0): #we're covering a short or closing a position + self.cost_basis = 0.0 + self.amount = 0 + else: + self.cost_basis = (self.cost_basis*self.amount + (txn.amount*txn.price))/(self.amount + txn.amount) + self.amount = self.amount + txn.amount + + def currentValue(self): + return self.amount * self.last_sale + + + def __repr__(self): + return "sid: {sid}, amount: {amount}, cost_basis: {cost_basis}, last_sale: {last_sale}".format( + sid=self.sid, amount=self.amount, cost_basis=self.cost_basis, last_sale=self.last_sale) + +class PerformancePeriod(): + + def __init__(self, period_start, period_end, initial_positions, initial_value, capital_base = None): + self.ending_value = 0.0 + self.period_capital_used = 0.0 + self.period_start = period_start + self.period_end = period_end + self.positions = initial_positions #sid => position object + self.starting_value = initial_value + if(capital_base != None): + self.capital_base = capital_base + else: + self.capital_base = 0 + + def calculate_performance(self): + self.ending_value = self.calculate_positions_value() + self.pnl = (self.ending_value - self.starting_value) - self.period_capital_used + if(self.capital_base != 0): + self.returns = self.pnl / self.capital_base + else: + self.returns = 0.0 + + def execute_transaction(self, txn): + if(txn.dt > self.period_end): + raise Exception("transaction dated {dt} attempted for period ending {ending}". + format(dt=txn.dt, ending=self.period_end)) + 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 + + + def calculate_positions_value(self): + mktValue = 0.0 + for key,pos in self.positions.iteritems(): + mktValue += pos.currentValue() + return mktValue + + def update_last_sale(self, event): + if self.positions.has_key(event.sid): + self.positions[event.sid].last_sale = event.price + self.positions[event.sid].last_date = event.dt + + + + \ No newline at end of file diff --git a/zipline/finance/risk.py b/zipline/finance/risk.py index 1dcc844f..b1959187 100644 --- a/zipline/finance/risk.py +++ b/zipline/finance/risk.py @@ -17,17 +17,16 @@ class daily_return(): return str(self.date) + " - " + str(self.returns) class periodmetrics(): - def __init__(self, start_date, end_date, returns, benchmark_returns, treasury_curves, trading_calendar): + def __init__(self, start_date, end_date, returns, trading_environment): """ :param treasury_curves: {datetime in utc -> {duration label -> interest rate}} """ - self.treasury_curves = treasury_curves self.start_date = start_date self.end_date = end_date - self.trading_calendar = trading_calendar + self.trading_environment = trading_environment self.algorithm_period_returns, self.algorithm_returns = self.calculate_period_returns(returns) - self.benchmark_period_returns, self.benchmark_returns = self.calculate_period_returns(benchmark_returns) + self.benchmark_period_returns, self.benchmark_returns = self.calculate_period_returns(trading_environment.benchmark_returns) if(len(self.benchmark_returns) != len(self.algorithm_returns)): raise Exception("Mismatch between benchmark_returns ({bm_count}) and algorithm_returns ({algo_count}) in range {start} : {end}".format( bm_count=len(self.benchmark_returns), @@ -53,7 +52,7 @@ class periodmetrics(): return '\n'.join(statements) def calculate_period_returns(self, daily_returns): - returns = [x.returns for x in daily_returns if x.date >= self.start_date and x.date <= self.end_date and self.trading_calendar.is_trading_day(x.date)] + returns = [x.returns for x in daily_returns if x.date >= self.start_date and x.date <= self.end_date and self.trading_environment.is_trading_day(x.date)] #qutil.LOGGER.debug("using {count} daily returns out of {total}".format(count=len(returns),total=len(daily_returns))) period_returns = 1.0 for r in returns: @@ -146,9 +145,8 @@ class periodmetrics(): curve = None #in case end date is not a trading day, search for the next market day for an interest rate for i in range(7): - if(self.treasury_curves.has_key(self.end_date + i * one_day)): - #qutil.LOGGER.info(self.treasury_curves[self.end_date + i * one_day]) - curve = self.treasury_curves[self.end_date + i * one_day] + if(self.trading_environment.treasury_curves.has_key(self.end_date + i * one_day)): + curve = self.trading_environment.treasury_curves[self.end_date + i * one_day] break if curve: @@ -165,13 +163,14 @@ class periodmetrics(): class riskmetrics(): - def __init__(self, algorithm_returns, benchmark_returns, treasury_curves, trading_calendar): + def __init__(self, algorithm_returns, trading_environment): """algorithm_returns needs to be a list of daily_return objects sorted in date ascending order""" self.algorithm_returns = algorithm_returns - self.bm_returns = [x for x in benchmark_returns if x.date >= self.algorithm_returns[0].date and x.date <= self.algorithm_returns[-1].date] - self.treasury_curves = treasury_curves - self.trading_calendar = trading_calendar + self.trading_environment = trading_environment + self.bm_returns = [x for x in self.trading_environmentself.benchmark_returns if x.date >= self.algorithm_returns[0].date and x.date <= self.algorithm_returns[-1].date] + self.treasury_curves = self.trading_environment.treasury_curves + qutil.LOGGER.debug("#### {start} thru {end} with {count} trading_days of {total} possible".format(start=self.algorithm_returns[0].date, end=self.algorithm_returns[-1].date, @@ -179,20 +178,20 @@ class riskmetrics(): total=len(benchmark_returns))) #calculate month ends - self.month_periods = self.periodsInRange(1, self.algorithm_returns[0].date, self.algorithm_returns[-1].date) + self.month_periods = self.periods_in_range(1, self.algorithm_returns[0].date, self.algorithm_returns[-1].date) #calculate 3 month ends - self.three_month_periods = self.periodsInRange(3, self.algorithm_returns[0].date, self.algorithm_returns[-1].date) + self.three_month_periods = self.periods_in_range(3, self.algorithm_returns[0].date, self.algorithm_returns[-1].date) #calculate 6 month ends - self.six_month_periods = self.periodsInRange(6, self.algorithm_returns[0].date, self.algorithm_returns[-1].date) + self.six_month_periods = self.periods_in_range(6, self.algorithm_returns[0].date, self.algorithm_returns[-1].date) #calculate 1 year ends - self.year_periods = self.periodsInRange(12, self.algorithm_returns[0].date, self.algorithm_returns[-1].date) + self.year_periods = self.periods_in_range(12, self.algorithm_returns[0].date, self.algorithm_returns[-1].date) #calculate 3 year ends - self.three_year_periods = self.periodsInRange(36, self.algorithm_returns[0].date, self.algorithm_returns[-1].date) + self.three_year_periods = self.periods_in_range(36, self.algorithm_returns[0].date, self.algorithm_returns[-1].date) #calculate 5 year ends - self.five_year_periods = self.periodsInRange(60, self.algorithm_returns[0].date, self.algorithm_returns[-1].date) + self.five_year_periods = self.periods_in_range(60, self.algorithm_returns[0].date, self.algorithm_returns[-1].date) - def periodsInRange(self, months_per, start, end): + def periods_in_range(self, months_per, start, end): one_day = datetime.timedelta(days = 1) ends = [] cur_start = start.replace(day=1) @@ -208,7 +207,7 @@ class riskmetrics(): returns=self.algorithm_returns, benchmark_returns=self.bm_returns, treasury_curves=self.treasury_curves, - trading_calendar=self.trading_calendar) + trading_environment=self.trading_environment) ends.append(cur_period_metrics) cur_start = advance_by_months(cur_start, 1) @@ -236,12 +235,13 @@ def advance_by_months(dt, jump_in_months): r = dt.replace(year = dt.year + years, month = month) return r -class TradingCalendar(object): +class TradingEnvironment(object): def __init__(self, benchmark_returns, treasury_curves): self.trading_days = [] self.trading_day_map = {} self.treasury_curves = treasury_curves + self.benchmark_returns = benchmark_returns for bm in benchmark_returns: self.trading_days.append(bm.date) self.trading_day_map[bm.date] = bm diff --git a/zipline/test/factory.py b/zipline/test/factory.py index 98406461..68ed322b 100644 --- a/zipline/test/factory.py +++ b/zipline/test/factory.py @@ -7,13 +7,13 @@ import zipline.finance.risk as risk import zipline.protocol as zp def load_market_data(): - fp_bm = open("./etc/benchmark.msgpack", "rb") + fp_bm = open("./zipline/test/benchmark.msgpack", "rb") bm_map = msgpack.loads(fp_bm.read()) bm_returns = [] for epoch, returns in bm_map.iteritems(): bm_returns.append(risk.daily_return(date=datetime.datetime.fromtimestamp(epoch).replace(hour=0, minute=0, second=0, tzinfo=pytz.utc), returns=returns)) bm_returns = sorted(bm_returns, key=lambda(x): x.date) - fp_tr = open("./etc/treasury_curves.msgpack", "rb") + fp_tr = open("./zipline/test/treasury_curves.msgpack", "rb") tr_map = msgpack.loads(fp_tr.read()) tr_curves = {} for epoch, curve in tr_map.iteritems(): diff --git a/zipline/test/test_finance.py b/zipline/test/test_finance.py index 7c6771ec..4a1b5da7 100644 --- a/zipline/test/test_finance.py +++ b/zipline/test/test_finance.py @@ -8,6 +8,7 @@ import zipline.test.factory as factory import zipline.util as qutil import zipline.finance.risk as risk import zipline.protocol as zp +import zipline.finance.performance as perf from zipline.test.client import TestTradingClient from zipline.sources import SpecificEquityTrades @@ -20,13 +21,11 @@ class FinanceTestCase(TestCase): def setUp(self): qutil.configure_logging() - self.benchmark_returns, self.treasury_curves = factory.load_market_data() - self.trading_calendar = risk.TradingCalendar(self.benchmark_returns, self.treasury_curves) + benchmark_returns, treasury_curves = factory.load_market_data() + self.trading_env = risk.TradingEnvironment(benchmark_returns, treasury_curves) def test_trade_feed_protocol(self): - - # TODO: Perhaps something more self-documenting for variables names? sid = 133 price = [10.0] * 4 volume = [100] * 4 @@ -34,7 +33,7 @@ class FinanceTestCase(TestCase): start_date = datetime.strptime("02/15/2012","%m/%d/%Y") one_day_td = timedelta(days=1) - trades = factory.create_trade_history(sid, price, volume, start_date, one_day_td, self.trading_calendar) + trades = factory.create_trade_history(sid, price, volume, start_date, one_day_td, self.trading_env) for trade in trades: #simulate data source sending frame @@ -149,7 +148,7 @@ class FinanceTestCase(TestCase): start_date = datetime.strptime("02/1/2012","%m/%d/%Y") trade_time_increment = timedelta(days=1) - trade_history = factory.create_trade_history( sid, price, volume, start_date, trade_time_increment, self.trading_calendar ) + trade_history = factory.create_trade_history( sid, price, volume, start_date, trade_time_increment, self.trading_env ) set1 = SpecificEquityTrades("flat-133", trade_history) @@ -173,3 +172,66 @@ class FinanceTestCase(TestCase): self.assertEqual(sim.feed.pending_messages(), 0, \ "The feed should be drained of all messages, found {n} remaining." \ .format(n=sim.feed.pending_messages())) + + + def test_performance(self): + + # verify order -> transaction -> portfolio position. + # -------------- + + # Allocate sockets for the simulator components + allocator = AddressAllocator(8) + sockets = allocator.lease(8) + + addresses = { + 'sync_address' : sockets[0], + 'data_address' : sockets[1], + 'feed_address' : sockets[2], + 'merge_address' : sockets[3], + 'result_address' : sockets[4], + 'order_address' : sockets[5] + } + + con = Controller( + sockets[6], + sockets[7], + logging = qutil.LOGGER + ) + + sim = Simulator(addresses) + + # Simulation Components + # --------------------- + + # TODO: Perhaps something more self-documenting for variables names? + sid = 133 + price = [10.1] * 16 + volume = [100] * 16 + start_date = datetime.strptime("02/1/2012","%m/%d/%Y") + trade_time_increment = timedelta(days=1) + + trade_history = factory.create_trade_history( sid, price, volume, start_date, trade_time_increment, self.trading_env ) + + set1 = SpecificEquityTrades("flat-133", trade_history) + + #client sill send 10 orders for 100 shares of 133 + client = TestTradingClient(133, 100, 10) + ts = datetime.strptime("02/1/2012","%m/%d/%Y").replace(tzinfo=pytz.utc) + + order_source = OrderDataSource(ts) + transaction_sim = TransactionSimulator() + portfolio_client = perf.PortfolioClient(trade_history[0]['dt'], trade_history[-1]['dt'], 1000000.0, self.trading_env) + + sim.register_components([client, order_source, transaction_sim, set1, portfolio_client]) + sim.register_controller( con ) + + # Simulation + # ---------- + sim_context = sim.simulate() + sim_context.join() + + + # TODO: Make more assertions about the final state of the components. + self.assertEqual(sim.feed.pending_messages(), 0, \ + "The feed should be drained of all messages, found {n} remaining." \ + .format(n=sim.feed.pending_messages())) \ No newline at end of file