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

Merge pull request #12 from quantopian/algohost

Browse Source 
merging after surviving another @sdiehl PR gauntlet
This commit is contained in:
fawce
2012-03-14 12:53:21 -07:00
parent 3c0abbf946 98c1312f16
commit d77a9fe296
15 changed files with 1435 additions and 376 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/zipline.finance.rst
+3 -3
View File
@@ -1,10 +1,10 @@
finance Package
===============
:mod:`data` Module
------------------
:mod:`performance` Module
-------------------------
.. automodule:: zipline.finance.data
.. automodule:: zipline.finance.performance
:members:
:undoc-members:
:show-inheritance:
docs/zipline.rst
+17 -8
View File
@@ -9,14 +9,6 @@ zipline Package
:undoc-members:
:show-inheritance:
:mod:`cli` Module
-----------------
.. automodule:: zipline.cli
:members:
:undoc-members:
:show-inheritance:
:mod:`component` Module
-----------------------
@@ -57,6 +49,14 @@ zipline Package
:undoc-members:
:show-inheritance:
:mod:`simulator` Module
-----------------------
.. automodule:: zipline.simulator
:members:
:undoc-members:
:show-inheritance:
:mod:`sources` Module
---------------------
@@ -65,6 +65,14 @@ zipline Package
:undoc-members:
:show-inheritance:
:mod:`topology` Module
----------------------
.. automodule:: zipline.topology
:members:
:undoc-members:
:show-inheritance:
:mod:`topos` Module
-------------------
@@ -86,6 +94,7 @@ Subpackages
.. toctree::
zipline.finance
zipline.test
zipline.transforms
docs/zipline.test.rst
+35 -3
View File
@@ -9,10 +9,18 @@ test Package
:undoc-members:
:show-inheritance:
:mod:`test_devsimulator` Module
-------------------------------
:mod:`factory` Module
---------------------
.. automodule:: zipline.test.test_devsimulator
.. automodule:: zipline.test.factory
:members:
:undoc-members:
:show-inheritance:
:mod:`test_finance` Module
--------------------------
.. automodule:: zipline.test.test_finance
:members:
:undoc-members:
:show-inheritance:
@@ -33,6 +41,22 @@ test Package
:undoc-members:
:show-inheritance:
:mod:`test_perf_tracking` Module
--------------------------------
.. automodule:: zipline.test.test_perf_tracking
:members:
:undoc-members:
:show-inheritance:
:mod:`test_risk` Module
-----------------------
.. automodule:: zipline.test.test_risk
:members:
:undoc-members:
:show-inheritance:
:mod:`test_sanity` Module
-------------------------
@@ -41,3 +65,11 @@ test Package
:undoc-members:
:show-inheritance:
:mod:`transform` Module
-----------------------
.. automodule:: zipline.test.transform
:members:
:undoc-members:
:show-inheritance:
zipline/finance/performance.py
+282 -111
View File
@@ -1,6 +1,7 @@
import datetime
import pytz
import math
import pandas
from zmq.core.poll import select
@@ -9,10 +10,9 @@ import zipline.util as qutil
import zipline.protocol as zp
import zipline.finance.risk as risk
class PortfolioClient(qmsg.Component):
class PerformanceTracker():
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
@@ -27,160 +27,293 @@ class PortfolioClient(qmsg.Component):
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)
self.txn_count = 0
self.event_count = 0
self.cumulative_performance = PerformancePeriod(
{},
capital_base,
starting_cash = capital_base
)
self.todays_performance = PerformancePeriod(
{},
capital_base,
starting_cash = capital_base
)
def to_dict(self):
"""
Creates a dictionary representing the state of this tracker.
Returns a dict object of the form:
@property
def get_id(self):
return str(zp.FINANCE_COMPONENT.PORTFOLIO_CLIENT)
+-----------------+----------------------------------------------------+
| key | value |
+=================+====================================================+
| period_start | The beginning of the period to be tracked. datetime|
| | in pytz.utc timezone. Will always be 0:00 on the |
| | date in UTC. The fact that the time may be on the |
| | prior day in the exchange's local time is ignored |
+-----------------+----------------------------------------------------+
| period_end | The end of the period to be tracked. datetime |
| | in pytz.utc timezone. Will always be 23:59 on the |
| | date in UTC. The fact that the time may be on the |
| | next day in the exchange's local time is ignored |
+-----------------+----------------------------------------------------+
| progress | percentage of test completed |
+-----------------+----------------------------------------------------+
| cumulative_capti| The net capital used (positive is spent) through |
| al_used | the course of all the events sent to this tracker |
+-----------------+----------------------------------------------------+
| max_capital_used| The maximum amount of capital deployed through the |
| | course of all the events sent to this tracker |
+-----------------+----------------------------------------------------+
| last_close | The most recent close of the market. datetime in |
| | pytz.utc timezone. Will always be 23:59 on the |
| | date in UTC. The fact that the time may be on the |
| | next day in the exchange's local time is ignored |
+-----------------+----------------------------------------------------+
| last_open | The most recent open of the market. datetime in |
| | pytz.utc timezone. Will always be 00:00 on the |
| | date in UTC. The fact that the time may be on the |
| | next day in the exchange's local time is ignored |
+-----------------+----------------------------------------------------+
| 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 |
| | :py:meth:`PerformancePeriod.to_dict` |
+-----------------+----------------------------------------------------+
| todays_perf | A dictionary representing the cumulative |
| | performance through all the events delivered to |
| | this tracker with datetime stamps between last_open|
| | and last_close. For details see the comments on |
| | :py:meth:`PerformancePeriod.to_dict` |
| | TODO: adding this because we calculate it. May be |
| | overkill. |
+-----------------+----------------------------------------------------+
| cumulative_risk | A dictionary representing the risk metrics |
| _metrics | calculated based on the positions aggregated |
| | through all the events delivered to this tracker. |
| | For details look at the comments for |
| | :py:meth:`zipline.finance.risk.RiskMetrics.to_dict`|
+-----------------+----------------------------------------------------+
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):
self.handle_simulation_end()
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:
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()
"""
returns_list = [x.to_dict() for x in self.returns]
d = {
'period_start' : self.period_start,
'period_end' : self.period_end,
'progress' : self.progress,
'cumulative_captial_used' : self.cumulative_captial_used,
'max_capital_used' : self.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_perf.to_dict(),
'todays_perf' : self.todays_perf.to_dict(),
'cumulative_risk_metrics' : self.cumulative_risk_metrics.to_dict()
}
def update(self, event_frame):
for dt, event_series in event_frame.iteritems():
data = {}
data.update(event_series)
event = zp.namedict(data)
self.process_event(event)
def process_event(self, event):
qutil.LOGGER.debug("series is " + str(event))
self.event_count += 1
if(event.dt >= self.market_close):
self.handle_market_close()
if not pandas.isnull(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)
#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
#add the return results from today to the list of daily return objects.
todays_date = self.todays_performance.period_end.replace(hour=0, minute=0, second=0)
todays_return_obj = risk.daily_return(todays_date, self.todays_performance.returns)
#add the return results from today to the list of DailyReturn objects.
todays_date = self.market_close.replace(hour=0, minute=0, second=0)
todays_return_obj = risk.DailyReturn(
todays_date,
self.todays_performance.returns
)
self.returns.append(todays_return_obj)
#calculate risk metrics for cumulative performance
self.cur_period_metrics = risk.RiskMetrics(start_date=self.cumulative_performance.period_start,
end_date=self.cumulative_performance.period_end.replace(hour=0, minute=0, second=0),
returns=self.returns,
trading_environment=self.trading_environment)
######################################################################################################
#######TODO: report/relay metrics out to qexec -- values come from self.cur_period_metrics ###########
#######TODO: report/relay position data out to qexec -- values come from self.cumulative_performance #
######################################################################################################
self.cumulative_risk_metrics = risk.RiskMetrics(
start_date=self.period_start,
end_date=self.market_close.replace(hour=0, minute=0, second=0),
returns=self.returns,
trading_environment=self.trading_environment
)
#move the market day markers forward
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("Attempt 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
#calculate progress of test
self.progress = self.day_count / self.total_days
####################################################################
#######TODO: relay the results of self.to_dict() ###########
####################################################################
#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)
self.todays_performance.calculate_performance()
self.todays_performance = PerformancePeriod(
self.todays_performance.positions,
self.todays_performance.ending_value,
self.todays_performance.ending_cash
)
def handle_simulation_end(self):
self.risk_report = risk.RiskReport(self.returns, self.trading_environment)
######################################################################################################
#######TODO: report/relay metrics out to qexec -- values come from self.risk_report ###########
######################################################################################################
self.risk_report = risk.RiskReport(
self.returns,
self.trading_environment
)
####################################################################
#######TODO: relay the results of self.risk_report.to_dict() #######
####################################################################
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
self.last_sale_price = None
self.last_sale_date = None
def update(self, txn):
if(self.sid != txn.sid):
raise NameError('attempt to update position with transaction in different sid')
raise NameError('updating position with txn for a 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)
prev_cost = self.cost_basis*self.amount
txn_cost = txn.amount*txn.price
total_cost = prev_cost + txn_cost
total_shares = self.amount + txn.amount
self.cost_basis = total_cost/total_shares
self.amount = self.amount + txn.amount
def currentValue(self):
return self.amount * self.last_sale
return self.amount * self.last_sale_price
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)
template = "sid: {sid}, amount: {amount}, cost_basis: {cost_basis}, \
last_sale_price: {last_sale_price}"
return template.format(
sid=self.sid,
amount=self.amount,
cost_basis=self.cost_basis,
last_sale_price=self.last_sale_price
)
def to_dict(self):
"""
Creates a dictionary representing the state of this position.
Returns a dict object of the form:
+-----------------+----------------------------------------------------+
| key | value |
+=================+====================================================+
| sid | the identifier for the security held in this |
| | position. |
+-----------------+----------------------------------------------------+
| amount | whole number of shares in the position |
+-----------------+----------------------------------------------------+
| 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 |
+-----------------+----------------------------------------------------+
"""
state = {
'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
}
return state
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 __init__(self, initial_positions, starting_value, starting_cash):
self.ending_value = 0.0
self.period_capital_used = 0.0
self.pnl = 0.0
#sid => position object
self.positions = initial_positions
self.starting_value = starting_value
#cash balance at start of period
self.starting_cash = starting_cash
self.ending_cash = starting_cash
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.starting_value
total_at_start = self.starting_cash + self.starting_value
self.ending_cash = self.starting_cash + self.period_capital_used
total_at_end = self.ending_cash + self.ending_value
self.pnl = total_at_end - total_at_start
if(total_at_start != 0):
self.returns = self.pnl / total_at_start
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():
@@ -188,10 +321,48 @@ class PerformancePeriod():
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
is_trade = event.type == zp.DATASOURCE_TYPE.TRADE
if self.positions.has_key(event.sid) and is_trade:
self.positions[event.sid].last_sale_price = event.price
self.positions[event.sid].last_sale_date = event.dt
def to_dict(self):
"""
Creates a dictionary representing the state of this performance period
Returns a dict object of the form:
+---------------+-----------------------------------------------------------+
| key | value |
+===============+===========================================================+
| ending_value | the total market value of the positions held at the |
| | end of the period |
+---------------+-----------------------------------------------------------+
| capital_used | the net capital consumed (positive means spent) by |
| | buying and selling securities in the period |
+---------------+-----------------------------------------------------------+
| starting_value| the total market value of the positions held at the |
| | start of the period |
+---------------+-----------------------------------------------------------+
| starting_cash | cash on hand at the beginning of the period |
+---------------+-----------------------------------------------------------+
| ending_cash | cash on hand at the end of the period |
+---------------+-----------------------------------------------------------+
| positions | a list of dicts representing positions, see |
| | :py:meth:`Position.to_dict()` |
| | for details on the contents of the dict |
+---------------+-----------------------------------------------------------+
"""
d = {
'ending_value':self.ending_value,
'capital_used':self.capital_used,
'starting_value':self.starting_value,
'starting_cash':self.starting_cash,
'ending_cash':self.ending_cash
}
position_list = []
for pos in self.positions:
position_list.append(pos.to_dict())
d['positions'] = positions_list
return d
zipline/finance/risk.py
+153 -59
View File
@@ -7,54 +7,129 @@ import zipline.util as qutil
import zipline.protocol as zp
from pymongo import ASCENDING, DESCENDING
class daily_return():
class DailyReturn():
def __init__(self, date, returns):
self.date = date
self.returns = returns
def to_dict(self):
d = {
'dt': self.date,
'returns': self.returns
}
return d
def __repr__(self):
return str(self.date) + " - " + str(self.returns)
class RiskMetrics():
def __init__(self, start_date, end_date, returns, benchmark_returns, treasury_curves, trading_calendar):
"""
:param treasury_curves: {datetime in utc -> {duration label -> interest rate}}
"""
def __init__(self, start_date, end_date, returns, trading_environment):
self.treasury_curves = treasury_curves
self.treasury_curves = trading_environment.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)
benchmark_returns = [x for x in self.trading_environment.benchmark_returns if x.date >= returns[0].date and x.date <= returns[-1].date]
self.benchmark_period_returns, self.benchmark_returns = self.calculate_period_returns(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),
algo_count=len(self.algorithm_returns),
start=start_date,
end=end_date))
message = "Mismatch between benchmark_returns ({bm_count}) and \
algorithm_returns ({algo_count}) in range {start} : {end}"
message.format(
bm_count=len(self.benchmark_returns),
algo_count=len(self.algorithm_returns),
start=start_date,
end=end_date
)
raise Exception(messge)
self.trading_days = len(self.benchmark_returns)
self.benchmark_volatility = self.calculate_volatility(self.benchmark_returns)
self.algorithm_volatility = self.calculate_volatility(self.algorithm_returns)
self.treasury_period_return = self.choose_treasury()
self.sharpe = self.calculate_sharpe()
self.beta, self.algorithm_covariance, self.benchmark_variance, self.condition_number, self.eigen_values = self.calculate_beta()
self.beta, self.algorithm_covariance, self.benchmark_variance, \
self.condition_number, self.eigen_values = self.calculate_beta()
self.alpha = self.calculate_alpha()
self.excess_return = self.algorithm_period_returns - self.treasury_period_return
self.max_drawdown = self.calculate_max_drawdown()
def to_dict(self):
"""
+-----------------+----------------------------------------------------+
| key | value |
+=================+====================================================+
| trading_days | The number of trading days between self.start_date |
| | and self.end_date |
+-----------------+----------------------------------------------------+
| benchmark_volat\| The volatility of the benchmark between |
| ility | self.start_date and self.end_date. |
+-----------------+----------------------------------------------------+
| algo_volatility | The volatility of the algo between self.start_date |
| | and self.end_date. |
+-----------------+----------------------------------------------------+
| treasury_period\| The return of treasuries over the period. Treasury |
| _return | maturity is chosen to match the duration of the |
| | test period. |
+-----------------+----------------------------------------------------+
| sharpe | The sharpe ratio based on the _algorithm_ (rather |
| | than the static portfolio) returns. |
+-----------------+----------------------------------------------------+
| beta | The _algorithm_ beta to the benchmark. |
+-----------------+----------------------------------------------------+
| alpha | The _algorithm_ alpha to the benchmark. |
+-----------------+----------------------------------------------------+
| excess_return | The excess return of the algorithm over the |
| | benchmark. |
+-----------------+----------------------------------------------------+
| max_drawdown | The largest relative peak to relative trough move |
| | for the portfolio returns between self.start_date |
| | and self.end_date. |
+-----------------+----------------------------------------------------+
"""
d = {
'trading_days' : self.trading_days,
'benchmark_volatility' : self.benchmark_volatility,
'algo_volatility' : self.algo_volatility,
'treasury_period_return': self.treasury_period_return,
'sharpe' : self.sharpe,
'beta' : self.beta,
'alpha' : self.alpha,
'excess_return' : self.excess_return,
'max_drawdown' : self.max_drawdown
}
def __repr__(self):
statements = []
for metric in ["algorithm_period_returns", "benchmark_period_returns", "excess_return", "trading_days", "benchmark_volatility", "algorithm_volatility", "sharpe", "algorithm_covariance", "benchmark_variance", "beta", "alpha", "max_drawdown", "algorithm_returns", "benchmark_returns", "condition_number", "eigen_values"]:
for metric in [
"algorithm_period_returns",
"benchmark_period_returns",
"excess_return",
"trading_days",
"benchmark_volatility",
"algorithm_volatility",
"sharpe",
"algorithm_covariance",
"benchmark_variance",
"beta",
"alpha",
"max_drawdown",
"algorithm_returns",
"benchmark_returns",
"condition_number",
"eigen_values"
]:
value = getattr(self, metric)
statements.append("{metric}:{value}".format(metric=metric, value=value))
statements.append("{m}:{v}".format(m=metric, v=value))
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)]
#qutil.LOGGER.debug("using {count} daily returns out of {total}".format(count=len(returns),total=len(daily_returns)))
#TODO: replace this with pandas.
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)]
period_returns = 1.0
for r in returns:
period_returns = period_returns * (1.0 + r)
@@ -69,8 +144,8 @@ class RiskMetrics():
return (self.algorithm_period_returns - self.treasury_period_return) / self.algorithm_volatility
def calculate_beta(self):
#qutil.LOGGER.debug("algorithm has {acount} days, benchmark has {bmcount} days".format(acount=len(self.algorithm_returns), bmcount=len(self.benchmark_returns)))
#it doesn't make much sense to calculate beta for less than two days, so return none.
#it doesn't make much sense to calculate beta for less than two days,
#so return none.
if len(self.algorithm_returns) < 2:
return 0.0, 0.0, 0.0, 0.0, []
returns_matrix = np.vstack([self.algorithm_returns, self.benchmark_returns])
@@ -80,7 +155,6 @@ class RiskMetrics():
algorithm_covariance = C[0][1]
benchmark_variance = C[1][1]
beta = C[0][1] / C[1][1]
#qutil.LOGGER.debug("bm variance is {bmv}, returns matrix is {rm}, covariance is {c}, beta is {beta}".format(rm=returns_matrix, bmv=C[1][1], c=C, beta=beta))
return beta, algorithm_covariance, benchmark_variance, condition_number, eigen_values
@@ -99,7 +173,6 @@ class RiskMetrics():
cur_return = 0.0
compounded_returns.append(cur_return)
#qutil.LOGGER.debug("compounded returns are {cr}".format(cr=compounded_returns))
cur_max = None
max_drawdown = None
for cur in compounded_returns:
@@ -110,7 +183,6 @@ class RiskMetrics():
if max_drawdown == None or drawdown < max_drawdown:
max_drawdown = drawdown
#qutil.LOGGER.debug("max drawdown is: {dd}".format(dd=max_drawdown))
if max_drawdown == None:
return 0.0
@@ -144,10 +216,10 @@ class RiskMetrics():
one_day = datetime.timedelta(days=1)
curve = None
#in case end date is not a trading day, search for the next market day for an interest rate
# 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]
break
@@ -160,55 +232,70 @@ class RiskMetrics():
if rate != None:
return rate * (td.days + 1) / 365
raise Exception("no rate for end date = {dt} and term = {term}. Using zero.".format(dt=self.end_date,
term=self.treasury_duration))
message = "no rate for end date = {dt} and term = {term}. Using zero."
message = message.format(dt=self.end_date,term=self.treasury_duration)
raise Exception(message)
class RiskReport():
def __init__(self, algorithm_returns, benchmark_returns, treasury_curves, trading_calendar):
"""algorithm_returns needs to be a list of daily_return objects sorted in date ascending order"""
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
start_date = self.algorithm_returns[0].date
end_date = self.algorithm_returns[-1].date
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,
count=len(self.bm_returns),
total=len(benchmark_returns)))
self.month_periods = self.periodsInRange(1, start_date, end_date)
self.three_month_periods = self.periodsInRange(3, start_date, end_date)
self.six_month_periods = self.periodsInRange(6, start_date, end_date)
self.year_periods = self.periodsInRange(12, start_date, end_date)
#calculate month ends
self.month_periods = self.periodsInRange(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)
#calculate 6 month ends
self.six_month_periods = self.periodsInRange(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)
#calculate 3 year ends
self.three_year_periods = self.periodsInRange(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)
def to_dict(self):
"""
RiskMetrics are calculated for rolling windows in four lengths::
- 1_month
- 3_month
- 6_month
- 12_month
The return value of this funciton is a dictionary keyed by the above
list of durations. The value of each entry is a list of RiskMetric
dicts of the same duration as denoted by the top_level key.
See :py:meth:`RiskMetrics.to_dict` for the detailed list of fields
provided for each period.
"""
d = {
'1_month' : [x.to_dict() for x in self.month_periods],
'3_month' : [x.to_dict() for x in self.three_year_periods],
'6_month' : [x.to_dict() for x in self.six_month_periods],
'12_month' : [x.to_dict() for x in self.month_periods]
}
return d
def periodsInRange(self, months_per, start, end):
one_day = datetime.timedelta(days = 1)
ends = []
cur_start = start.replace(day=1)
#ensure that we have an end at the end of a calendar month, in case the return series ends mid-month...
#ensure that we have an end at the end of a calendar month, in case
#the return series ends mid-month...
the_end = advance_by_months(end.replace(day=1),1) - one_day
while True:
cur_end = advance_by_months(cur_start, months_per) - one_day
if(cur_end > the_end):
break
#qutil.LOGGER.debug("start: {start}, end: {end}".format(start=cur_start, end=cur_end))
cur_period_metrics = RiskMetrics(start_date=cur_start,
end_date=cur_end,
returns=self.algorithm_returns,
benchmark_returns=self.bm_returns,
treasury_curves=self.treasury_curves,
trading_calendar=self.trading_calendar)
cur_period_metrics = RiskMetrics(
start_date=cur_start,
end_date=cur_end,
returns=self.algorithm_returns,
trading_environment=self.trading_environment
)
ends.append(cur_period_metrics)
cur_start = advance_by_months(cur_start, 1)
@@ -226,9 +313,10 @@ def advance_by_months(dt, jump_in_months):
years = month / 12
month = month % 12
#no remainder means that we are landing in december.
#modulo is, in a way, a zero indexed circular array.
#this is a way of converting to 1 indexed months. (in our modulo index, december is zeroth)
# no remainder means that we are landing in december.
# modulo is, in a way, a zero indexed circular array.
# this is a way of converting to 1 indexed months.
# (in our modulo index, december is zeroth)
if(month == 0):
month = 12
years = years - 1
@@ -242,12 +330,18 @@ class TradingEnvironment(object):
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
def normalize_date(self, test_date):
return datetime.datetime(year=test_date.year, month=test_date.month, day=test_date.day, tzinfo=pytz.utc)
return datetime.datetime(
year=test_date.year,
month=test_date.month,
day=test_date.day,
tzinfo=pytz.utc
)
def is_trading_day(self, test_date):
dt = self.normalize_date(test_date)
zipline/finance/trading.py
+103 -50
View File
@@ -1,6 +1,7 @@
import datetime
import pytz
import math
import pandas
from zmq.core.poll import select
@@ -10,16 +11,28 @@ import zipline.protocol as zp
class TradeSimulationClient(qmsg.Component):
def __init__(self):
def __init__(self, simulation_dt):
qmsg.Component.__init__(self)
self.received_count = 0
self.prev_dt = None
self.event_queue = []
self.event_queue = None
self.event_callbacks = []
self.txn_count = 0
self.current_dt = simulation_dt
self.last_iteration_duration = datetime.timedelta(seconds=0)
self.event_frame = None
@property
def get_id(self):
return str(zp.FINANCE_COMPONENT.TRADING_CLIENT)
def add_event_callback(self, callback):
"""
:param callable callback: must be a function with the signature
f(frame).
"""
self.event_callbacks.append(callback)
def open(self):
self.result_feed = self.connect_result()
self.order_socket = self.connect_order()
@@ -28,40 +41,77 @@ class TradeSimulationClient(qmsg.Component):
#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:
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("Client is DONE!")
self.run_callbacks()
self.signal_done()
return
event = zp.MERGE_UNFRAME(msg)
self._handle_event(event)
if(event.TRANSACTION != None):
self.txn_count += 1
#filter order flow out of the events sent to callbacks
if event.source_id != zp.FINANCE_COMPONENT.ORDER_SOURCE:
#mark the start time for client's processing of this event.
event_start = datetime.datetime.utcnow()
self.queue_event(event)
if event.dt >= self.current_dt:
self.run_callbacks()
#update time based on receipt of the order
self.last_iteration_duration = datetime.datetime.utcnow() - event_start
self.current_dt = self.current_dt + self.last_iteration_duration
#signal done to order source.
self.order_socket.send(str(zp.ORDER_PROTOCOL.BREAK))
def run_callbacks(self):
frame = self.get_frame()
for cb in self.event_callbacks:
cb(frame)
def connect_order(self):
return self.connect_push_socket(self.addresses['order_address'])
def _handle_event(self, event):
self.handle_event(event)
#signal done to order source.
self.order_socket.send(str(zp.ORDER_PROTOCOL.BREAK))
def handle_event(self, event):
raise NotImplementedError
def order(self, sid, amount):
self.order_socket.send(zp.ORDER_FRAME(sid, amount))
order = zp.namedict({
'dt':self.current_dt,
'sid':sid,
'amount':amount
})
self.order_socket.send(zp.ORDER_FRAME(order))
def signal_order_done(self):
self.order_socket.send(str(zp.ORDER_PROTOCOL.DONE))
def queue_event(self, event):
if self.event_queue == None:
self.event_queue = {}
series = event.as_series()
self.event_queue[event.dt] = series
def get_frame(self):
frame = pandas.DataFrame(self.event_queue)
self.event_queue = None
return frame
class OrderDataSource(qmsg.DataSource):
"""DataSource that relays orders from the client"""
def __init__(self, simulation_dt):
def __init__(self):
"""
:param simulation_time: datetime in UTC timezone, sets the start time of simulation. orders
:param simulation_time: datetime in UTC timezone, sets the start
time of simulation. orders
will be timestamped relative to this datetime.
event = {
'sid' : an integer for security id,
@@ -71,8 +121,6 @@ class OrderDataSource(qmsg.DataSource):
}
"""
qmsg.DataSource.__init__(self, zp.FINANCE_COMPONENT.ORDER_SOURCE)
self.simulation_dt = simulation_dt
self.last_iteration_duration = datetime.timedelta(seconds=0)
self.sent_count = 0
@property
@@ -87,57 +135,53 @@ class OrderDataSource(qmsg.DataSource):
return self.bind_pull_socket(self.addresses['order_address'])
def do_work(self):
#mark the start time for client's processing of this event.
self.event_start = datetime.datetime.utcnow()
self.simulation_dt = self.simulation_dt + self.last_iteration_duration
#TODO: if this is the first iteration, break deadlock by sending a dummy order
if(self.sent_count == 0):
self.send_dummy()
self.send(zp.namedict({}))
#pull all orders from client.
orders = []
order_dt = None
count = 0
while True:
(rlist, wlist, xlist) = select([self.order_socket],
[],
[self.order_socket],
timeout=self.heartbeat_timeout/1000) #select timeout is in sec
(rlist, wlist, xlist) = select(
[self.order_socket],
[],
[self.order_socket],
#allow half the time of a heartbeat for the order
#timeout, so we have time to signal we are done.
timeout=self.heartbeat_timeout/2000
)
#no more orders, should this be an error condition?
if len(rlist) == 0 or len(xlist) > 0:
continue
#no order message means there was a timeout above,
#and the client is done sending orders (but isn't
#telling us himself!).
self.signal_done()
return
order_msg = rlist[0].recv()
if order_msg == str(zp.ORDER_PROTOCOL.DONE):
self.signal_done()
return
if order_msg == str(zp.ORDER_PROTOCOL.BREAK):
qutil.LOGGER.info("order loop finished")
break
sid, amount = zp.ORDER_UNFRAME(order_msg)
order = zp.ORDER_UNFRAME(order_msg)
#send the order along
self.last_iteration_duration = datetime.datetime.utcnow() - self.event_start
dt = self.simulation_dt + self.last_iteration_duration
order_event = zp.namedict({"sid":sid, "amount":amount, "dt":dt})
self.send(order_event)
self.send(order)
count += 1
self.sent_count += 1
#TODO: we have to send at least one dummy order per do_work iteration or the feed will block waiting for our messages.
#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_dummy()
self.sent_count += 1
def send_dummy(self):
dt = self.simulation_dt + self.last_iteration_duration
dummy_order = zp.namedict({"sid":0, "amount":0, "dt":dt})
self.send(dummy_order)
self.send(zp.namedict({}))
@@ -147,7 +191,8 @@ class TransactionSimulator(qmsg.BaseTransform):
qmsg.BaseTransform.__init__(self, zp.TRANSFORM_TYPE.TRANSACTION)
self.open_orders = {}
self.order_count = 0
self.trade_windwo = datetime.timedelta(seconds=30)
self.txn_count = 0
self.trade_window = datetime.timedelta(seconds=30)
self.orderTTL = datetime.timedelta(days=1)
self.volume_share = 0.05
self.commission = 0.03
@@ -157,7 +202,6 @@ class TransactionSimulator(qmsg.BaseTransform):
Pulls one message from the event feed, then
loops on orders until client sends DONE message.
"""
#TODO: need a way to send a placeholder txn, to avoid blocking merge... maybe customize merge to not block on txn?
if(event.type == zp.DATASOURCE_TYPE.ORDER):
self.add_open_order(event)
self.state['value'] = None
@@ -190,7 +234,8 @@ class TransactionSimulator(qmsg.BaseTransform):
def apply_trade_to_open_orders(self, event):
if(event.volume == 0):
#there are zero volume events bc some stocks trade less frequently than once per minute.
#there are zero volume events bc some stocks trade
#less frequently than once per minute.
return self.create_dummy_txn(event.dt)
if self.open_orders.has_key(event.sid):
@@ -203,8 +248,9 @@ class TransactionSimulator(qmsg.BaseTransform):
dt = event.dt
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_windwo):
#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
@@ -224,10 +270,17 @@ class TransactionSimulator(qmsg.BaseTransform):
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)
return self.create_transaction(
event.sid,
amount,
event.price + impact,
dt.replace(tzinfo = pytz.utc),
direction
)
def create_transaction(self, sid, amount, price, dt, direction):
def create_transaction(self, sid, amount, price, dt, direction):
self.txn_count += 1
txn = {'sid' : sid,
'amount' : int(amount),
'dt' : dt,
zipline/messaging.py
+10 -4
View File
@@ -287,13 +287,18 @@ class ParallelBuffer(Component):
cur_source = None
earliest_source = None
earliest_event = None
#iterate over the queues of events from all sources (1 queue per datasource)
#iterate over the queues of events from all sources
#(1 queue per datasource)
for events in self.data_buffer.values():
if len(events) == 0:
continue
cur_source = events
first_in_list = events[0]
if first_in_list.dt == None:
#this is a filler event, discard
events.pop(0)
continue
if (earliest_event == None) or (first_in_list.dt <= earliest_event.dt):
earliest_event = first_in_list
earliest_source = cur_source
@@ -384,7 +389,8 @@ class MergedParallelBuffer(ParallelBuffer):
def append(self, event):
"""
:param event: a namedict with one entry. key is the name of the transform, value is the transformed value.
:param event: a namedict with one entry. key is the name of the
transform, value is the transformed value.
Add an event to the buffer for the source specified by
source_id.
"""
@@ -398,7 +404,7 @@ class BaseTransform(Component):
Top level execution entry point for the transform
- connects to the feed socket to subscribe to events
- connets to the result socket (most oftened bound by a TransformsMerge) to PUSH transforms
- connects to the result socket (most oftened bound by a TransformsMerge) to PUSH transforms
- processes all messages received from feed, until DONE message received
- pushes all transforms
- sends DONE to result socket, closes all sockets and context
zipline/protocol.py
+84 -57
View File
@@ -119,6 +119,7 @@ import numbers
import datetime
import pytz
import copy
import pandas
from collections import namedtuple
import zipline.util as qutil
@@ -165,7 +166,7 @@ class namedict(object):
"""
def __init__(self, dct=None):
if(dct):
if dct:
self.__dict__.update(dct)
def __setitem__(self, key, value):
@@ -205,7 +206,11 @@ class namedict(object):
def has_attr(self, name):
return self.__dict__.has_key(name)
def as_series(self):
s = pandas.Series(self.__dict__, self.__dict__.keys())
return s
# ================
# Control Protocol
# ================
@@ -295,11 +300,27 @@ def DATASOURCE_FRAME(event):
assert isinstance(event.source_id, basestring)
assert isinstance(event.type, int), 'Unexpected type %s' % (event.type)
#datasources will send sometimes send empty msgs to feel gaps
if len(event.keys()) == 2:
return msgpack.dumps(tuple([
event.type,
event.source_id,
DATASOURCE_TYPE.EMPTY
]))
if(event.type == DATASOURCE_TYPE.TRADE):
return msgpack.dumps(tuple([event.type, TRADE_FRAME(event)]))
return msgpack.dumps(tuple([
event.type,
event.source_id,
TRADE_FRAME(event)
]))
elif(event.type == DATASOURCE_TYPE.ORDER):
return msgpack.dumps(tuple([event.type, ORDER_SOURCE_FRAME(event)]))
return msgpack.dumps(tuple([
event.type,
event.source_id,
ORDER_SOURCE_FRAME(event)
]))
else:
raise INVALID_DATASOURCE_FRAME(str(event))
@@ -321,15 +342,21 @@ def DATASOURCE_UNFRAME(msg):
"""
try:
ds_type, payload = msgpack.loads(msg)
ds_type, source_id, payload = msgpack.loads(msg)
assert isinstance(ds_type, int)
if(ds_type == DATASOURCE_TYPE.TRADE):
return TRADE_UNFRAME(payload)
rval = namedict({'source_id':source_id})
if payload == DATASOURCE_TYPE.EMPTY:
child_value = namedict({'dt':None})
elif(ds_type == DATASOURCE_TYPE.TRADE):
child_value = TRADE_UNFRAME(payload)
elif(ds_type == DATASOURCE_TYPE.ORDER):
return ORDER_SOURCE_UNFRAME(payload)
child_value = ORDER_SOURCE_UNFRAME(payload)
else:
raise INVALID_DATASOURCE_FRAME(msg)
rval.merge(child_value)
return rval
except TypeError:
raise INVALID_DATASOURCE_FRAME(msg)
except ValueError:
@@ -461,7 +488,6 @@ def TRADE_FRAME(event):
"""
assert isinstance(event, namedict)
assert isinstance(event.source_id, basestring)
assert event.type == DATASOURCE_TYPE.TRADE
assert isinstance(event.sid, int)
assert isinstance(event.price, numbers.Real)
@@ -471,16 +497,14 @@ def TRADE_FRAME(event):
event.sid,
event.price,
event.volume,
event.epoch,
event.micros,
event.dt,
event.type,
event.source_id
]))
def TRADE_UNFRAME(msg):
try:
packed = msgpack.loads(msg)
sid, price, volume, epoch, micros, source_type, source_id = packed
sid, price, volume, dt, source_type = packed
assert isinstance(sid, int)
assert isinstance(price, numbers.Real)
@@ -489,10 +513,8 @@ def TRADE_UNFRAME(msg):
'sid' : sid,
'price' : price,
'volume' : volume,
'epoch' : epoch,
'micros' : micros,
'type' : source_type,
'source_id' : source_id
'dt' : dt,
'type' : source_type
})
UNPACK_DATE(rval)
return rval
@@ -505,19 +527,29 @@ def TRADE_UNFRAME(msg):
# Orders - from client to order source
# =========
def ORDER_FRAME(sid, amount):
assert isinstance(sid, int)
assert isinstance(amount, int) #no partial shares...
return msgpack.dumps(tuple([sid, amount]))
def ORDER_FRAME(order):
assert isinstance(order.sid, int)
assert isinstance(order.amount, int) #no partial shares...
PACK_DATE(order)
return msgpack.dumps(tuple([
order.sid,
order.amount,
order.dt
]))
def ORDER_UNFRAME(msg):
try:
sid, amount = msgpack.loads(msg)
sid, amount, dt = msgpack.loads(msg)
assert isinstance(sid, int)
assert isinstance(amount, int)
return sid, amount
rval = namedict({
'sid':sid,
'amount':amount,
'dt':dt
})
UNPACK_DATE(rval)
return rval
except TypeError:
raise INVALID_ORDER_FRAME(msg)
except ValueError:
@@ -540,13 +572,12 @@ def TRANSACTION_FRAME(event):
event.price,
event.amount,
event.commission,
event.epoch,
event.micros
event.dt
]))
def TRANSACTION_UNFRAME(msg):
try:
sid, price, amount, commission, epoch, micros = msgpack.loads(msg)
sid, price, amount, commission, dt = msgpack.loads(msg)
assert isinstance(sid, int)
assert isinstance(price, numbers.Real)
@@ -557,8 +588,7 @@ def TRANSACTION_UNFRAME(msg):
'price' : price,
'amount' : amount,
'commission' : commission,
'epoch' : epoch,
'micros' : micros
'dt' : dt
})
UNPACK_DATE(rval)
@@ -583,8 +613,7 @@ def ORDER_SOURCE_FRAME(event):
return msgpack.dumps(tuple([
event.sid,
event.amount,
event.epoch,
event.micros,
event.dt,
event.source_id,
event.type
]))
@@ -592,12 +621,11 @@ def ORDER_SOURCE_FRAME(event):
def ORDER_SOURCE_UNFRAME(msg):
try:
sid, amount, epoch, micros, source_id, source_type = msgpack.loads(msg)
sid, amount, dt, source_id, source_type = msgpack.loads(msg)
event = namedict({
"sid" : sid,
"amount" : amount,
"epoch" : epoch,
"micros" : micros,
"dt" : dt,
"source_id" : source_id,
"type" : source_type
})
@@ -620,9 +648,8 @@ def PACK_DATE(event):
"""
Packs the datetime property of event into msgpack'able longs.
This function should be called purely for its side effects.
The event's 'dt' property is replaced by two longs: epoch and micros.
Epoch is the unix epoch time in UTC, and micros is the microsecond
property of the original event.dt datetime object.
The event's 'dt' property is replaced by a tuple of integers::
- year, month, day, hour, minute, second, microsecond
PACK_DATE and UNPACK_DATE are inverse operations.
@@ -631,44 +658,44 @@ def PACK_DATE(event):
"""
assert isinstance(event.dt, datetime.datetime)
assert event.dt.tzinfo == pytz.utc #utc only please
epoch = long(event.dt.strftime('%s'))
event['epoch'] = epoch
event['micros'] = event.dt.microsecond
event.delete('dt')
year, month, day, hour, minute, second = event.dt.timetuple()[0:6]
micros = event.dt.microsecond
event['dt'] = tuple([year, month, day, hour, minute, second, micros])
def UNPACK_DATE(event):
"""
Unpacks the datetime property of event from msgpack'able longs.
This function should be called purely for its side effects.
The event's 'dt' property is created by reading and then combining two longs: epoch and micros.
The epoch and micros properties are removed after dt is added.
The event's 'dt' property is converted to a datetime by reading and then
combining a tuple of integers.
UNPACK_DATE and PACK_DATE are inverse operations.
:param event: event must a namedict with::
- a property named 'epoch' that is an integral representing the unix \
epoch time in UTC
- a property named 'micros' that is an integral the microsecond \
property of the original event.dt datetime object
:param tuple event: event must a namedict with::
- a property named 'dt_tuple' that is a tuple of integers
representing the date and time in UTC. dt_tumple must have year,
month, day, hour, minute, second, and microsecond
:rtype: None
"""
assert isinstance(event.epoch, numbers.Integral)
assert isinstance(event.micros, numbers.Integral)
dt = datetime.datetime.fromtimestamp(event.epoch)
dt = dt.replace(microsecond = event.micros, tzinfo = pytz.utc)
event.delete('epoch')
event.delete('micros')
assert isinstance(event.dt, tuple)
assert len(event.dt) == 7
for item in event.dt:
assert isinstance(item, numbers.Integral)
year, month, day, hour, minute, second, micros = event.dt
dt = datetime.datetime(year, month, day, hour, minute, second)
dt = dt.replace(microsecond = micros, tzinfo = pytz.utc)
event.dt = dt
DATASOURCE_TYPE = Enum(
'ORDER',
'TRADE'
'TRADE',
'EMPTY',
)
ORDER_PROTOCOL = Enum(
'DONE',
'BREAK'
'BREAK',
)
zipline/simulator.py
+2 -4
View File
@@ -61,10 +61,8 @@ class Simulator(ComponentHost):
if not self.running:
return
try:
self.controller.shutdown(context=self.context)
except:
import pdb; pdb.set_trace()
#if self.controller:
#self.controller.shutdown()
for component in self.components.itervalues():
component.shutdown()
zipline/test/client.py
+18 -11
View File
@@ -66,20 +66,27 @@ class TestClient(qmsg.Component):
return zp.MERGE_UNFRAME(msg)
class TestTradingClient(TradeSimulationClient):
class TestAlgorithm():
def __init__(self, sid, amount, order_count):
TradeSimulationClient.__init__(self)
def __init__(self, sid, amount, order_count, trading_client):
self.trading_client = trading_client
self.trading_client.add_event_callback(self.handle_frame)
self.count = order_count
self.sid = sid
self.amount = amount
self.incr = 0
self.done = False
def handle_event(self, event):
#place an order for 100 shares of sid:133
if(self.incr < self.count):
self.order(self.sid, self.amount)
self.incr += 1
else:
self.signal_order_done()
self.signal_done()
def handle_frame(self, frame):
for dt, s in frame.iteritems():
data = {}
data.update(s)
event = zp.namedict(data)
#place an order for 100 shares of sid:133
if self.incr < self.count:
if event.source_id != zp.FINANCE_COMPONENT.ORDER_SOURCE:
self.trading_client.order(self.sid, self.amount)
self.incr += 1
elif not self.done:
self.trading_client.signal_order_done()
self.done = True
zipline/test/factory.py
+27 -12
View File
@@ -11,26 +11,37 @@ def load_market_data():
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))
event_dt = datetime.datetime.fromtimestamp(epoch)
event_dt = event_dt.replace(
hour=0,
minute=0,
second=0,
tzinfo=pytz.utc
)
daily_return = risk.DailyReturn(date=event_dt, returns=returns)
bm_returns.append(daily_return)
bm_returns = sorted(bm_returns, key=lambda(x): x.date)
fp_tr = open("./zipline/test/treasury_curves.msgpack", "rb")
tr_map = msgpack.loads(fp_tr.read())
tr_curves = {}
for epoch, curve in tr_map.iteritems():
tr_curves[datetime.datetime.fromtimestamp(epoch).replace(hour=0, minute=0, second=0, tzinfo=pytz.utc)] = curve
tr_dt = datetime.datetime.fromtimestamp(epoch)
tr_dt = tr_dt.replace(hour=0, minute=0, second=0, tzinfo=pytz.utc)
tr_curves[tr_dt] = curve
return bm_returns, tr_curves
def create_trade(sid, price, amount, datetime):
row = {
row = zp.namedict({
'source_id' : "test_factory",
'type' : zp.DATASOURCE_TYPE.TRADE,
'sid' : sid,
'dt' : datetime,
'price' : price,
'volume' : amount
}
})
return row
def create_trade_history(sid, prices, amounts, start_time, interval, trading_calendar):
@@ -50,19 +61,23 @@ def create_trade_history(sid, prices, amounts, start_time, interval, trading_cal
return trades
def createTxn(sid, price, amount, datetime, btrid=None):
txn = Transaction(sid=sid, amount=amount, dt = datetime,
price=price, transaction_cost=-1*price*amount)
def create_txn(sid, price, amount, datetime, btrid=None):
txn = zp.namedict({
'sid':sid,
'amount':amount,
'dt':datetime,
'price':price,
})
return txn
def create_transaction_history(sid, priceList, amtList, startTime, interval, trading_calendar):
def create_txn_history(sid, priceList, amtList, startTime, interval, trading_calendar):
txns = []
current = startTime
for price, amount in zip(priceList, amtList):
if trading_calendar.is_trading_day(current):
txns.append(createTxn(sid, price, amount, current))
txns.append(create_txn(sid, price, amount, current))
current = current + interval
else:
@@ -78,7 +93,7 @@ def create_returns(daycount, start, trading_calendar):
one_day = datetime.timedelta(days = 1)
while i < daycount:
i += 1
r = risk.daily_return(current, random.random())
r = risk.DailyReturn(current, random.random())
test_range.append(r)
current = current + one_day
return [ x for x in test_range if(trading_calendar.is_trading_day(x.date)) ]
@@ -94,7 +109,7 @@ def create_returns_from_range(start, end, trading_calendar):
current = current + one_day
if(not trading_calendar.is_trading_day(current)):
continue
r = risk.daily_return(current, random.random())
r = risk.DailyReturn(current, random.random())
i += 1
test_range.append(r)
@@ -107,7 +122,7 @@ def create_returns_from_list(returns, start, trading_calendar):
i = 0
while len(test_range) < len(returns):
if(trading_calendar.is_trading_day(current)):
r = risk.daily_return(current, returns[i])
r = risk.DailyReturn(current, returns[i])
i += 1
test_range.append(r)
current = current + one_day
zipline/test/test_finance.py
+121 -39
View File
@@ -1,8 +1,12 @@
"""Tests for the zipline.finance package"""
import mock
import pytz
from unittest2 import TestCase
from datetime import datetime, timedelta
from collections import defaultdict
from nose.tools import timed
import zipline.test.factory as factory
import zipline.util as qutil
@@ -10,26 +14,48 @@ 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.test.client import TestAlgorithm
from zipline.sources import SpecificEquityTrades
from zipline.finance.trading import TransactionSimulator, OrderDataSource
from zipline.finance.trading import TransactionSimulator, OrderDataSource, \
TradeSimulationClient
from zipline.simulator import AddressAllocator, Simulator
from zipline.monitor import Controller
DEFAULT_TIMEOUT = 5 # seconds
allocator = AddressAllocator(1000)
class FinanceTestCase(TestCase):
leased_sockets = defaultdict(list)
def setUp(self):
qutil.configure_logging()
self.benchmark_returns, self.treasury_curves = \
factory.load_market_data()
self.trading_environment = risk.TradingEnvironment(
self.benchmark_returns,
self.benchmark_returns,
self.treasury_curves
)
self.allocator = allocator
def allocate_sockets(self, n):
"""
Allocate sockets local to this test case, track them so
we can gc after test run.
"""
assert isinstance(n, int)
assert n > 0
leased = self.allocator.lease(n)
self.leased_sockets[self.id()].extend(leased)
return leased
@timed(DEFAULT_TIMEOUT)
def test_trade_feed_protocol(self):
# TODO: Perhaps something more self-documenting for variables names?
@@ -82,21 +108,28 @@ class FinanceTestCase(TestCase):
self.assertEqual(zp.namedict(trade), event)
@timed(DEFAULT_TIMEOUT)
def test_order_protocol(self):
#client places an order
order_msg = zp.ORDER_FRAME(133, 100)
now = datetime.utcnow().replace(tzinfo=pytz.utc)
order = zp.namedict({
'dt':now,
'sid':133,
'amount':100
})
order_msg = zp.ORDER_FRAME(order)
#order datasource receives
sid, amount = zp.ORDER_UNFRAME(order_msg)
self.assertEqual(sid, 133)
self.assertEqual(amount, 100)
order = zp.ORDER_UNFRAME(order_msg)
self.assertEqual(order.sid, 133)
self.assertEqual(order.amount, 100)
self.assertEqual(order.dt, now)
#order datasource datasource frames the order
order_dt = datetime.utcnow().replace(tzinfo=pytz.utc)
order_event = zp.namedict({
"sid" : sid,
"amount" : amount,
"dt" : order_dt,
"sid" : order.sid,
"amount" : order.amount,
"dt" : order.dt,
"source_id" : zp.FINANCE_COMPONENT.ORDER_SOURCE,
"type" : zp.DATASOURCE_TYPE.ORDER
})
@@ -107,7 +140,7 @@ class FinanceTestCase(TestCase):
#transaction transform unframes
recovered_order = zp.DATASOURCE_UNFRAME(order_ds_msg)
self.assertEqual(order_dt, recovered_order.dt)
self.assertEqual(now, recovered_order.dt)
#create a transaction from the order
txn = zp.namedict({
@@ -126,14 +159,14 @@ class FinanceTestCase(TestCase):
self.assertEqual(recovered_tx.sid, 133)
self.assertEqual(recovered_tx.amount, 100)
@timed(DEFAULT_TIMEOUT)
def test_orders(self):
# Just verify sending and receiving orders.
# --------------
# Allocate sockets for the simulator components
allocator = AddressAllocator(8)
sockets = allocator.lease(8)
sockets = self.allocate_sockets(8)
addresses = {
'sync_address' : sockets[0],
@@ -160,6 +193,7 @@ class FinanceTestCase(TestCase):
price = [10.1] * 16
volume = [100] * 16
start_date = datetime.strptime("02/1/2012","%m/%d/%Y")
start_date = start_date.replace(tzinfo=pytz.utc)
trade_time_increment = timedelta(days=1)
trade_history = factory.create_trade_history(
@@ -172,15 +206,20 @@ class FinanceTestCase(TestCase):
)
set1 = SpecificEquityTrades("flat-133", trade_history)
trading_client = TradeSimulationClient(start_date)
#client will send 10 orders for 100 shares of 133
test_algo = TestAlgorithm(133, 100, 10, trading_client)
#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)
order_source = OrderDataSource()
transaction_sim = TransactionSimulator()
sim.register_components([client, order_source, transaction_sim, set1])
sim.register_components([
trading_client,
order_source,
transaction_sim,
set1
])
sim.register_controller( con )
# Simulation
@@ -188,6 +227,8 @@ class FinanceTestCase(TestCase):
sim_context = sim.simulate()
sim_context.join()
self.assertTrue(sim.ready())
self.assertFalse(sim.exception)
# TODO: Make more assertions about the final state of the components.
self.assertEqual(sim.feed.pending_messages(), 0, \
@@ -195,14 +236,14 @@ class FinanceTestCase(TestCase):
.format(n=sim.feed.pending_messages()))
def test_performance(self):
@timed(DEFAULT_TIMEOUT)
def test_performance(self):
# verify order -> transaction -> portfolio position.
# --------------
# Allocate sockets for the simulator components
allocator = AddressAllocator(8)
sockets = allocator.lease(8)
sockets = self.allocate_sockets(8)
addresses = {
'sync_address' : sockets[0],
@@ -225,10 +266,12 @@ class FinanceTestCase(TestCase):
# ---------------------
# TODO: Perhaps something more self-documenting for variables names?
trade_count = 100
sid = 133
price = [10.1] * 16
volume = [100] * 16
price = [10.1] * trade_count
volume = [100] * trade_count
start_date = datetime.strptime("02/1/2012","%m/%d/%Y")
start_date = start_date.replace(tzinfo=pytz.utc)
trade_time_increment = timedelta(days=1)
trade_history = factory.create_trade_history(
@@ -242,24 +285,25 @@ class FinanceTestCase(TestCase):
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")
ts = ts.replace(tzinfo=pytz.utc)
trading_client = TradeSimulationClient(start_date)
test_algo = TestAlgorithm(133, 100, 10, trading_client)
order_source = OrderDataSource(ts)
order_source = OrderDataSource()
transaction_sim = TransactionSimulator()
portfolio_client = perf.PortfolioClient(
perf_tracker = perf.PerformanceTracker(
trade_history[0]['dt'],
trade_history[-1]['dt'],
1000000.0,
self.trading_environment)
#register perf_tracker to receive callbacks from the client.
trading_client.add_event_callback(perf_tracker.update)
sim.register_components([
client,
trading_client,
order_source,
transaction_sim,
set1,
portfolio_client,
])
sim.register_controller( con )
@@ -268,8 +312,46 @@ class FinanceTestCase(TestCase):
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, \
self.assertEqual(
sim.feed.pending_messages(),
0,
"The feed should be drained of all messages, found {n} remaining." \
.format(n=sim.feed.pending_messages()))
.format(n=sim.feed.pending_messages())
)
self.assertEqual(
sim.merge.pending_messages(),
0,
"The merge should be drained of all messages, found {n} remaining." \
.format(n=sim.merge.pending_messages())
)
self.assertEqual(
test_algo.count,
test_algo.incr,
"The test algorithm should send as many orders as specified.")
self.assertEqual(
order_source.sent_count,
test_algo.count,
"The order source should have sent as many orders as the algo."
)
self.assertEqual(
transaction_sim.txn_count,
perf_tracker.txn_count,
"The perf tracker should handle the same number of transactions \
as the simulator emits."
)
self.assertEqual(
len(perf_tracker.cumulative_performance.positions),
1,
"Portfolio should have one position."
)
self.assertEqual(
perf_tracker.cumulative_performance.positions[133].sid,
133,
"Portfolio should have one position in 133."
)
zipline/test/test_messaging.py
+9
View File
@@ -17,6 +17,15 @@ from nose.tools import timed
# it up as a test. Its a Mixin of sorts at this point.
class SimulatorTestCase(object):
# Leased sockets is a defaultdict keyed by the test case.
# This lets you debug the sockets being allocated in the
# specific test cases and tear them down appropriately.
#
# {
# 'test_orders' : ['tcp : //127.0.0.1 : 1000', ... ],
# 'test_performance' : ['tcp : //127.0.0.1 : 1025', ... ],
# }
leased_sockets = defaultdict(list)
def setUp(self):
zipline/test/test_perf_tracking.py
+559
View File
@@ -0,0 +1,559 @@
import unittest
import copy
import random
import datetime
import pytz
import zipline.test.factory as factory
import zipline.util as qutil
import zipline.finance.performance as perf
import zipline.finance.risk as risk
import zipline.protocol as zp
from zipline.finance.trading import TradeSimulationClient
class PerformanceTestCase(unittest.TestCase):
def setUp(self):
qutil.configure_logging()
self.benchmark_returns, self.treasury_curves = \
factory.load_market_data()
self.trading_environment = risk.TradingEnvironment(
self.benchmark_returns,
self.treasury_curves
)
self.onesec = datetime.timedelta(seconds=1)
self.oneday = datetime.timedelta(days=1)
self.tradingday = datetime.timedelta(hours=6, minutes=30)
random_index = random.randint(
0,
len(self.trading_environment.trading_days)
)
self.dt = self.trading_environment.trading_days[random_index]
def tearDown(self):
pass
def test_long_position(self):
"""
verify that the performance period calculates properly for a
single buy transaction
"""
#post some trades in the market
trades = factory.create_trade_history(
1,
[10,10,10,11],
[100,100,100,100],
self.dt,
self.onesec,
self.trading_environment
)
txn = factory.create_txn(1,10.0,100,self.dt + self.onesec)
pp = perf.PerformancePeriod({}, 0.0, 1000.0)
pp.execute_transaction(txn)
for trade in trades:
pp.update_last_sale(trade)
pp.calculate_performance()
self.assertEqual(
pp.period_capital_used,
-1 * txn.price * txn.amount,
"capital used should be equal to the opposite of the transaction \
cost of sole txn in test"
)
self.assertEqual(len(pp.positions),1,"should be just one position")
self.assertEqual(
pp.positions[1].sid,
txn.sid,
"position should be in security with id 1")
self.assertEqual(
pp.positions[1].amount,
txn.amount,
"should have a position of {sharecount} shares".format(
sharecount=txn.amount
)
)
self.assertEqual(
pp.positions[1].cost_basis,
txn.price,
"should have a cost basis of 10"
)
self.assertEqual(
pp.positions[1].last_sale_price,
trades[-1]['price'],
"last sale should be same as last trade. \
expected {exp} actual {act}".format(
exp=trades[-1]['price'],
act=pp.positions[1].last_sale_price
)
)
self.assertEqual(
pp.ending_value,
1100,
"ending value should be price of last trade times number of \
shares in position"
)
self.assertEqual(pp.pnl, 100, "gain of 1 on 100 shares should be 100")
def test_short_position(self):
"""verify that the performance period calculates properly for a \
single short-sale transaction"""
trades_1 = factory.create_trade_history(
1,
[10,10,10,11],
[100,100,100,100],
self.dt,
self.onesec,
self.trading_environment
)
txn = factory.create_txn(1, 10.0, -100, self.dt + self.onesec)
pp = perf.PerformancePeriod({}, 0.0, 1000.0)
pp.execute_transaction(txn)
for trade in trades_1:
pp.update_last_sale(trade)
pp.calculate_performance()
self.assertEqual(
pp.period_capital_used,
-1 * txn.price * txn.amount,
"capital used should be equal to the opposite of the transaction\
cost of sole txn in test"
)
self.assertEqual(
len(pp.positions),
1,
"should be just one position")
self.assertEqual(
pp.positions[1].sid,
txn.sid,
"position should be in security from the transaction"
)
self.assertEqual(
pp.positions[1].amount,
-100,
"should have a position of -100 shares"
)
self.assertEqual(
pp.positions[1].cost_basis,
txn.price,
"should have a cost basis of 10"
)
self.assertEqual(
pp.positions[1].last_sale_price,
trades_1[-1]['price'],
"last sale should be price of last trade"
)
self.assertEqual(
pp.ending_value,
-1100,
"ending value should be price of last trade times number of \
shares in position"
)
self.assertEqual(pp.pnl,-100,"gain of 1 on 100 shares should be 100")
#simulate additional trades, and ensure that the position value
#reflects the new price
trades_2 = factory.create_trade_history(
1,
[10,9],
[100,100],
trades_1[-1]['dt'] + self.onesec,
self.onesec,
self.trading_environment
)
#simulate a rollover to a new period
pp2 = perf.PerformancePeriod(
pp.positions,
pp.ending_value,
pp.ending_cash
)
for trade in trades_2:
pp2.update_last_sale(trade)
pp2.calculate_performance()
self.assertEqual(
pp2.period_capital_used,
0,
"capital used should be zero, there were no transactions in \
performance period"
)
self.assertEqual(
len(pp2.positions),
1,
"should be just one position"
)
self.assertEqual(
pp2.positions[1].sid,
txn.sid,
"position should be in security from the transaction"
)
self.assertEqual(
pp2.positions[1].amount,
-100,
"should have a position of -100 shares"
)
self.assertEqual(
pp2.positions[1].cost_basis,
txn.price,
"should have a cost basis of 10"
)
self.assertEqual(
pp2.positions[1].last_sale_price,
trades_2[-1].price,
"last sale should be price of last trade"
)
self.assertEqual(
pp2.ending_value,
-900,
"ending value should be price of last trade times number of \
shares in position")
self.assertEqual(
pp2.pnl,
200,
"drop of 2 on -100 shares should be 200"
)
#now run a performance period encompassing the entire trade sample.
ppTotal = perf.PerformancePeriod({}, 0.0, 1000.0)
for trade in trades_1:
ppTotal.update_last_sale(trade)
ppTotal.execute_transaction(txn)
for trade in trades_2:
ppTotal.update_last_sale(trade)
ppTotal.calculate_performance()
self.assertEqual(
ppTotal.period_capital_used,
-1 * txn.price * txn.amount,
"capital used should be equal to the opposite of the transaction \
cost of sole txn in test"
)
self.assertEqual(
len(ppTotal.positions),
1,
"should be just one position"
)
self.assertEqual(
ppTotal.positions[1].sid,
txn.sid,
"position should be in security from the transaction"
)
self.assertEqual(
ppTotal.positions[1].amount,
-100,
"should have a position of -100 shares"
)
self.assertEqual(
ppTotal.positions[1].cost_basis,
txn.price,
"should have a cost basis of 10"
)
self.assertEqual(
ppTotal.positions[1].last_sale_price,
trades_2[-1].price,
"last sale should be price of last trade"
)
self.assertEqual(
ppTotal.ending_value,
-900,
"ending value should be price of last trade times number of \
shares in position")
self.assertEqual(
ppTotal.pnl,
100,
"drop of 1 on -100 shares should be 100"
)
def test_covering_short(self):
"""verify performance where short is bought and covered, and shares \
trade after cover"""
trades = factory.create_trade_history(
1,
[10,10,10,11,9,8,7,8,9,10],
[100,100,100,100,100,100,100,100,100,100],
self.dt,
self.onesec,
self.trading_environment
)
short_txn = factory.create_txn(
1,
10.0,
-100,
self.dt + self.onesec
)
cover_txn = factory.create_txn(1,7.0,100,self.dt + self.onesec * 6)
pp = perf.PerformancePeriod({}, 0.0, 1000.0)
pp.execute_transaction(short_txn)
pp.execute_transaction(cover_txn)
for trade in trades:
pp.update_last_sale(trade)
pp.calculate_performance()
short_txn_cost = short_txn.price * short_txn.amount
cover_txn_cost = cover_txn.price * cover_txn.amount
self.assertEqual(
pp.period_capital_used,
-1 * short_txn_cost - cover_txn_cost,
"capital used should be equal to the net transaction costs"
)
self.assertEqual(
len(pp.positions),
1,
"should be just one position"
)
self.assertEqual(
pp.positions[1].sid,
short_txn.sid,
"position should be in security from the transaction"
)
self.assertEqual(
pp.positions[1].amount,
0,
"should have a position of -100 shares"
)
self.assertEqual(
pp.positions[1].cost_basis,
0,
"a covered position should have a cost basis of 0"
)
self.assertEqual(
pp.positions[1].last_sale_price,
trades[-1].price,
"last sale should be price of last trade"
)
self.assertEqual(
pp.ending_value,
0,
"ending value should be price of last trade times number of \
shares in position"
)
self.assertEqual(
pp.pnl,
300,
"gain of 1 on 100 shares should be 300"
)
def test_cost_basis_calc(self):
trades = factory.create_trade_history(
1,
[10,11,11,12],
[100,100,100,100],
self.dt,
self.onesec,
self.trading_environment
)
transactions = factory.create_txn_history(
1,
[10,11,11,12],
[100,100,100,100],
self.dt,
self.onesec,
self.trading_environment
)
pp = perf.PerformancePeriod({}, 0.0, 1000.0)
for txn in transactions:
pp.execute_transaction(txn)
for trade in trades:
pp.update_last_sale(trade)
pp.calculate_performance()
self.assertEqual(
pp.positions[1].last_sale_price,
trades[-1].price,
"should have a last sale of 12, got {val}".format(
val=pp.positions[1].last_sale_price
)
)
self.assertEqual(
pp.positions[1].cost_basis,
11,
"should have a cost basis of 11"
)
self.assertEqual(
pp.pnl,
400
)
saleTxn = factory.create_txn(
1,
10.0,
-100,
self.dt + self.onesec * 4)
down_tick = factory.create_trade(
1,
10.0,
100,
trades[-1].dt + self.onesec)
pp2 = perf.PerformancePeriod(
copy.deepcopy(pp.positions),
pp.ending_value,
pp.ending_cash
)
pp2.execute_transaction(saleTxn)
pp2.update_last_sale(down_tick)
pp2.calculate_performance()
self.assertEqual(
pp2.positions[1].last_sale_price,
10,
"should have a last sale of 10, was {val}".format(val=pp2.positions[1].last_sale_price)
)
self.assertEqual(
round(pp2.positions[1].cost_basis,2),
11.33,
"should have a cost basis of 11.33"
)
#print "second period pnl is {pnl}".format(pnl=pp2.pnl)
self.assertEqual(pp2.pnl, -800, "this period goes from +400 to -400")
pp3 = perf.PerformancePeriod({}, 0.0, 1000.0)
transactions.append(saleTxn)
for txn in transactions:
pp3.execute_transaction(txn)
trades.append(down_tick)
for trade in trades:
pp3.update_last_sale(trade)
pp3.calculate_performance()
self.assertEqual(
pp3.positions[1].last_sale_price,
10,
"should have a last sale of 10"
)
self.assertEqual(
round(pp3.positions[1].cost_basis,2),
11.33,
"should have a cost basis of 11.33"
)
self.assertEqual(
pp3.pnl,
-400,
"should be -400 for all trades and transactions in period"
)
def test_tracker(self):
trade_count = 100
sid = 133
price = [10.1] * 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,
price,
volume,
start_date,
trade_time_increment,
self.trading_environment
)
trade_client = TradeSimulationClient(start_date)
start = trade_history[0].dt
end = trade_history[-1].dt
tracker = perf.PerformanceTracker(
start,
end,
1000.0,
self.trading_environment
)
for event in trade_history:
#create a transaction for all but
#one trade, to simulate None transaction
if(event.dt != start):
txn = zp.namedict({
'sid' : event.sid,
'amount' : -25,
'dt' : event.dt,
'price' : 10.0,
'commission' : 0.50
})
else:
txn = None
event[zp.TRANSFORM_TYPE.TRANSACTION] = txn
trade_client.queue_event(event)
df = trade_client.get_frame()
tracker.update(df)
#we skip one trade, to test case of None transaction
txn_count = len(trade_history) - 1
self.assertEqual(tracker.txn_count, txn_count)
cumulative_pos = tracker.cumulative_performance.positions[sid]
expected_size = txn_count * -25
self.assertEqual(cumulative_pos.amount, expected_size)
zipline/test/test_risk.py
+12 -15
View File
@@ -26,7 +26,7 @@ class Risk(unittest.TestCase):
start_date = datetime.datetime(year=2006, month=1, day=1, tzinfo=pytz.utc)
self.algo_returns_06 = factory.create_returns_from_list(RETURNS, start_date, self.trading_calendar)
end_date = datetime.datetime(year=2006, month=12, day=31, tzinfo=pytz.utc)
self.metrics_06 = risk.RiskReport(self.algo_returns_06, self.benchmark_returns, self.treasury_curves, self.trading_calendar)
self.metrics_06 = risk.RiskReport(self.algo_returns_06, self.trading_calendar)
def tearDown(self):
return
@@ -41,14 +41,14 @@ class Risk(unittest.TestCase):
start_date = datetime.datetime(year=2006, month=1, day=1)
returns = factory.create_returns_from_list([1.0,-0.5,0.8,.17,1.0,-0.1,-0.45], start_date, self.trading_calendar)
#200, 100, 180, 210.6, 421.2, 379.8, 208.494
metrics = risk.RiskMetrics(returns[0].date, returns[-1].date, returns, self.benchmark_returns, self.treasury_curves, self.trading_calendar)
metrics = risk.RiskMetrics(returns[0].date, returns[-1].date, returns, self.trading_calendar)
self.assertEqual(metrics.max_drawdown, 0.505)
def test_benchmark_returns_06(self):
start_date = datetime.datetime(year=2006, month=1, day=1)
end_date = datetime.datetime(year=2006, month=12, day=31)
returns = factory.create_returns_from_range(start_date, end_date, self.trading_calendar)
metrics = risk.RiskReport(returns, self.benchmark_returns, self.treasury_curves, self.trading_calendar)
metrics = risk.RiskReport(returns, self.trading_calendar)
self.assertEqual([round(x.benchmark_period_returns, 4) for x in metrics.month_periods],
[0.0255,0.0005,0.0111,0.0122,-0.0309,0.0001,0.0051,0.0213,0.0246,0.0315,0.0165,0.0126])
self.assertEqual([round(x.benchmark_period_returns, 4) for x in metrics.three_month_periods],
@@ -61,7 +61,7 @@ class Risk(unittest.TestCase):
start_date = datetime.datetime(year=2006, month=1, day=1)
end_date = datetime.datetime(year=2006, month=12, day=31)
returns = factory.create_returns_from_range(start_date, end_date, self.trading_calendar)
metrics = risk.RiskReport(returns, self.benchmark_returns, self.treasury_curves, self.trading_calendar)
metrics = risk.RiskReport(returns, self.trading_calendar)
self.assertEqual([x.trading_days for x in metrics.year_periods],[251])
self.assertEqual([x.trading_days for x in metrics.month_periods],[20,19,23,19,22,22,20,23,20,22,21,20])
@@ -69,7 +69,7 @@ class Risk(unittest.TestCase):
start_date = datetime.datetime(year=2006, month=1, day=1)
end_date = datetime.datetime(year=2006, month=12, day=31)
returns = factory.create_returns_from_range(start_date, end_date, self.trading_calendar)
metrics = risk.RiskReport(returns, self.benchmark_returns, self.treasury_curves, self.trading_calendar)
metrics = risk.RiskReport(returns, self.trading_calendar)
self.assertEqual([round(x.benchmark_volatility, 3) for x in metrics.month_periods],
[0.031,0.026,0.024,0.025,0.037,0.047,0.039,0.022,0.023,0.021,0.025,0.019])
self.assertEqual([round(x.benchmark_volatility, 3) for x in metrics.three_month_periods],
@@ -131,7 +131,7 @@ class Risk(unittest.TestCase):
start_date = datetime.datetime(year=2008, month=1, day=1)
end_date = datetime.datetime(year=2008, month=12, day=31)
returns = factory.create_returns_from_range(start_date, end_date, self.trading_calendar)
metrics = risk.RiskReport(returns, self.benchmark_returns, self.treasury_curves, self.trading_calendar)
metrics = risk.RiskReport(returns, self.trading_calendar)
self.assertEqual([round(x.benchmark_period_returns, 3) for x in metrics.month_periods],
[-0.061,-0.035,-0.006,0.048,0.011,-0.086,-0.01,0.012,-0.091,-0.169,-0.075,0.008])
self.assertEqual([round(x.benchmark_period_returns, 3) for x in metrics.three_month_periods],
@@ -144,7 +144,7 @@ class Risk(unittest.TestCase):
start_date = datetime.datetime(year=2008, month=1, day=1)
end_date = datetime.datetime(year=2008, month=12, day=31)
returns = factory.create_returns_from_range(start_date, end_date, self.trading_calendar)
metrics = risk.RiskReport(returns, self.benchmark_returns, self.treasury_curves, self.trading_calendar)
metrics = risk.RiskReport(returns, self.trading_calendar)
self.assertEqual([x.trading_days for x in metrics.year_periods],[253])
self.assertEqual([x.trading_days for x in metrics.month_periods],[21,20,20,22,21,21,22,21,21,23,19,22])
@@ -152,7 +152,7 @@ class Risk(unittest.TestCase):
start_date = datetime.datetime(year=2008, month=1, day=1)
end_date = datetime.datetime(year=2008, month=12, day=31)
returns = factory.create_returns_from_range(start_date, end_date, self.trading_calendar)
metrics = risk.RiskReport(returns, self.benchmark_returns, self.treasury_curves, self.trading_calendar)
metrics = risk.RiskReport(returns, self.trading_calendar)
self.assertEqual([round(x.benchmark_volatility, 3) for x in metrics.month_periods],
[0.07,0.058,0.082,0.054,0.041,0.057,0.068,0.06,0.157,0.244,0.195,0.145])
self.assertEqual([round(x.benchmark_volatility, 3) for x in metrics.three_month_periods],
@@ -166,7 +166,7 @@ class Risk(unittest.TestCase):
start_date = datetime.datetime(year=2006, month=1, day=1)
end_date = datetime.datetime(year=2006, month=12, day=31)
returns = factory.create_returns_from_range(start_date, end_date, self.trading_calendar)
metrics = risk.RiskReport(returns, self.benchmark_returns, self.treasury_curves, self.trading_calendar)
metrics = risk.RiskReport(returns, self.trading_calendar)
self.assertEqual([round(x.treasury_period_return, 4) for x in metrics.month_periods],
[0.0037,0.0034,0.0039,0.0038,0.0040,0.0037,0.0043,0.0043,0.0038,0.0044,0.0043,0.0041])
self.assertEqual([round(x.treasury_period_return, 4) for x in metrics.three_month_periods],
@@ -183,7 +183,7 @@ class Risk(unittest.TestCase):
start_date = datetime.datetime(year=1991, month=1, day=1)
returns = factory.create_returns(365 * 5 + 2, start_date, self.trading_calendar) #1992 and 1996 were leap years
returns = returns[:-10] #truncate the returns series to end mid-month
metrics = risk.RiskReport(returns, self.benchmark_returns, self.treasury_curves, self.trading_calendar)
metrics = risk.RiskReport(returns, self.trading_calendar)
total_months = 60
self.check_metrics(metrics, total_months, start_date)
@@ -194,7 +194,7 @@ class Risk(unittest.TestCase):
#because we may catch the leap of the last year, and i think this func is [start,end)
ld = calendar.leapdays(start_date.year, start_date.year + years + 1)
returns = factory.create_returns(365 * years + ld, start_date, self.trading_calendar)
metrics = risk.RiskReport(returns, self.benchmark_returns, self.treasury_curves, self.trading_calendar)
metrics = risk.RiskReport(returns, self.trading_calendar)
total_months = years * 12
self.check_metrics(metrics, total_months, start_date)
@@ -202,10 +202,7 @@ class Risk(unittest.TestCase):
self.assert_range_length(metrics.month_periods, total_months, 1, start_date)
self.assert_range_length(metrics.three_month_periods, total_months, 3, start_date)
self.assert_range_length(metrics.six_month_periods, total_months, 6, start_date)
self.assert_range_length(metrics.year_periods, total_months, 12, start_date)
self.assert_range_length(metrics.three_year_periods, total_months, 36, start_date)
self.assert_range_length(metrics.five_year_periods, total_months, 60, start_date)
self.assert_range_length(metrics.year_periods, total_months, 12, start_date)
def assert_last_day(self, period_end):
#30 days has september, april, june and november
if(period_end.month in [9,4,6,11]):