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

Merge pull request #13 from quantopian/reframing

Browse Source 
Reframing
This commit is contained in:
fawce
2012-03-15 16:32:53 -07:00
parent d77a9fe296 c42ffb5e1a
commit 5ec72350d3
15 changed files with 921 additions and 419 deletions
Download Patch File Download Diff File Expand all files Collapse all files
etc/jenkins.sh
+1 -1
View File
@@ -29,7 +29,7 @@ pip freeze
paver apidocs html
#run all the tests in test. see setup.cfg for flags.
nosetests
nosetests --config=jenkins_setup.cfg
#run pylint checks
cp ./pylint.rcfile /mnt/jenkins/.pylintrc #default location for config file...
jenkins_setup.cfg
+12
View File
@@ -0,0 +1,12 @@
[nosetests]
verbosity=2
detailed-errors=1
with-xcoverage=1
cover-package=zipline
cover-erase=1
cover-html=1
cover-html-dir=docs/_build/html/cover
with-xunit=1
notebooks/Experimenting with Frames.ipynb
+352
View File
@@ -0,0 +1,352 @@
{
"metadata": {
"name": "Experimenting with Frames"
},
"nbformat": 3,
"worksheets": [
{
"cells": [
{
"cell_type": "heading",
"source": [
"Performance Tracking"
]
},
{
"cell_type": "code",
"collapsed": false,
"input": [
"import datetime",
"import pandas",
"import pytz",
"",
"import zipline.test.factory as factory",
"import zipline.finance.performance as perf",
"import zipline.protocol as zp",
"import zipline.finance.risk as risk",
"import zipline.finance.trading as trading"
],
"language": "python",
"outputs": [],
"prompt_number": 38
},
{
"cell_type": "heading",
"source": [
"Create a simulated trade history using the test factory"
]
},
{
"cell_type": "markdown",
"source": [
"For any backtesting, zipline relies on a TradingEnvironment object. Trading environment holds essential facts: ",
" ",
" - start and end times for the simulation.",
" - historical daily returns for your benchmark.",
" - historical treasury curves",
" - an assumed capital base for your portfolio",
" - a calendar of trading days based on your benchmark",
"",
"zipline ships with a compressed archives of the S&P daily returns, and US treasury curves to facilitate standalone development and testing. In the next cell we instantiate the environment using these defaults. You can see more of this in zipline/test/test_perf_tracking.py"
]
},
{
"cell_type": "code",
"collapsed": false,
"input": [
"benchmark_returns, treasury_curves = factory.load_market_data()",
" ",
"trading_environment = risk.TradingEnvironment(benchmark_returns, treasury_curves)"
],
"language": "python",
"outputs": [],
"prompt_number": 39
},
{
"cell_type": "code",
"collapsed": false,
"input": [
"trade_count = 100",
"sid = 133",
"price = 10.1 ",
"price_list = [price] * trade_count",
"volume = [100] * trade_count",
"start_date = datetime.datetime.strptime(\"01/01/2011\",\"%m/%d/%Y\")",
"start_date = start_date.replace(tzinfo=pytz.utc)",
"trade_time_increment = datetime.timedelta(days=1)",
"",
"trade_history = factory.create_trade_history( ",
" sid, ",
" price_list, ",
" volume, ",
" start_date, ",
" trade_time_increment, ",
" trading_environment ",
")",
"",
"sid2 = 134",
"price2 = 12.12",
"price2_list = [price2] * trade_count ",
"trade_history2 = factory.create_trade_history( ",
" sid2, ",
" price2_list, ",
" volume, ",
" start_date, ",
" trade_time_increment, ",
" trading_environment ",
")",
" ",
"trade_history.extend(trade_history2) ",
"trade_history = sorted(trade_history, key=lambda x: x.dt)"
],
"language": "python",
"outputs": [],
"prompt_number": 40
},
{
"cell_type": "markdown",
"source": [
"Now that we have a simulated history of trades for two companies and a corresponding trading environment, we can create a dataframe of trades."
]
},
{
"cell_type": "code",
"collapsed": false,
"input": [
"df = pandas.DataFrame(index = ['price', 'volume', 'dt'])",
"for event in trade_history:",
" series = event.as_series()",
" #df.index = df.index.tolist().append(event.sid)",
" #series.name = event.sid",
" df[event.sid] = series"
],
"language": "python",
"outputs": [],
"prompt_number": 92
},
{
"cell_type": "code",
"collapsed": false,
"input": [
"df"
],
"language": "python",
"outputs": [
{
"output_type": "pyout",
"prompt_number": 93,
"text": [
" 133 134",
"price 10.1 12.12",
"volume 100 100",
"dt 2011-04-08 00:00:00+00:00 2011-04-08 00:00:00+00:00"
]
}
],
"prompt_number": 93
},
{
"cell_type": "code",
"collapsed": false,
"input": [
"df_t = df.transpose()",
"df_t"
],
"language": "python",
"outputs": [
{
"output_type": "pyout",
"prompt_number": 94,
"text": [
" price volume dt",
"133 10.1 100 2011-04-08 00:00:00+00:00",
"134 12.12 100 2011-04-08 00:00:00+00:00"
]
}
],
"prompt_number": 94
},
{
"cell_type": "code",
"collapsed": false,
"input": [
"df[133]"
],
"language": "python",
"outputs": [
{
"output_type": "pyout",
"prompt_number": 56,
"text": [
"sid 133",
"volume 100",
"dt 2011-04-08 00:00:00+00:00",
"price 10.1",
"changed NaN",
"Name: 133"
]
}
],
"prompt_number": 56
},
{
"cell_type": "code",
"collapsed": false,
"input": [
"df_t['price']"
],
"language": "python",
"outputs": [
{
"output_type": "pyout",
"prompt_number": 57,
"text": [
"133 10.1",
"134 12.12",
"Name: price"
]
}
],
"prompt_number": 57
},
{
"cell_type": "code",
"collapsed": false,
"input": [
"df_t['price'].max()"
],
"language": "python",
"outputs": [
{
"output_type": "pyout",
"prompt_number": 50,
"text": [
"12.12"
]
}
],
"prompt_number": 50
},
{
"cell_type": "code",
"collapsed": true,
"input": [
"last = trade_history[23].dt"
],
"language": "python",
"outputs": [],
"prompt_number": 51
},
{
"cell_type": "code",
"collapsed": false,
"input": [
"df_t['changed'] = df_t['dt'] > last"
],
"language": "python",
"outputs": [],
"prompt_number": 53
},
{
"cell_type": "code",
"collapsed": false,
"input": [
"df_t"
],
"language": "python",
"outputs": [
{
"output_type": "pyout",
"prompt_number": 54,
"text": [
" sid volume dt price changed",
"133 133 100 2011-04-08 00:00:00+00:00 10.1 True",
"134 134 100 2011-04-08 00:00:00+00:00 12.12 True"
]
}
],
"prompt_number": 54
},
{
"cell_type": "code",
"collapsed": false,
"input": [
"df_t.index"
],
"language": "python",
"outputs": [
{
"output_type": "pyout",
"prompt_number": 59,
"text": [
"Int64Index([133, 134])"
]
}
],
"prompt_number": 59
},
{
"cell_type": "code",
"collapsed": false,
"input": [
"df.index"
],
"language": "python",
"outputs": [
{
"output_type": "pyout",
"prompt_number": 60,
"text": [
"Index([sid, volume, dt, price, changed], dtype=object)"
]
}
],
"prompt_number": 60
},
{
"cell_type": "code",
"collapsed": false,
"input": [
"df.columns"
],
"language": "python",
"outputs": [
{
"output_type": "pyout",
"prompt_number": 61,
"text": [
"Int64Index([133, 134])"
]
}
],
"prompt_number": 61
},
{
"cell_type": "code",
"collapsed": false,
"input": [
"df_t.columns"
],
"language": "python",
"outputs": [
{
"output_type": "pyout",
"prompt_number": 62,
"text": [
"Index([sid, volume, dt, price, changed], dtype=object)"
]
}
],
"prompt_number": 62
},
{
"cell_type": "code",
"collapsed": true,
"input": [],
"language": "python",
"outputs": []
}
]
}
]
}
setup.cfg
-9
View File
@@ -2,15 +2,6 @@
verbosity=2
detailed-errors=1
with-xcoverage=1
cover-package=zipline
cover-erase=1
cover-html=1
cover-html-dir=docs/_build/html/cover
with-xunit=1
# Drop into debugger on failure
#pdb=0
#pdb-failures=0
zipline/component.py
+13 -2
View File
@@ -14,7 +14,8 @@ import humanhash
from datetime import datetime
import zipline.util as qutil
from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_STATE
from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_STATE, \
COMPONENT_FAILURE, BACKTEST_STATE
class Component(object):
"""
@@ -66,6 +67,7 @@ class Component(object):
self.controller = None
self.heartbeat_timeout = 2000
self.state_flag = COMPONENT_STATE.OK
self.error_state = COMPONENT_FAILURE.NOFAILURE
self.on_done = None
self._exception = None
@@ -254,8 +256,17 @@ class Component(object):
# Internal Maintenance
# ----------------------
def signal_exception(self, exc=None):
def signal_exception(self, exc=None, scope=None):
if scope == 'algo':
self.error_state = COMPONENT_FAILURE.ALGOEXCEPT
else:
self.error_state = COMPONENT_FAILURE.HOSTEXCEPT
self.state_flag = COMPONENT_STATE.EXCEPTION
# mark the time of failure so we can track the failure
# progogation through the system.
self.stop_tic = time.time()
self._exception = exc
zipline/finance/performance.py
+305 -257
View File
@@ -1,58 +1,15 @@
import datetime
import pytz
import math
import pandas
"""
from zmq.core.poll import select
Performance Tracking
====================
import zipline.messaging as qmsg
import zipline.util as qutil
import zipline.protocol as zp
import zipline.finance.risk as risk
class PerformanceTracker():
def __init__(self, period_start, period_end, capital_base, trading_environment):
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.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:
+-----------------+----------------------------------------------------+
| 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 |
| | 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 |
@@ -62,7 +19,7 @@ class PerformanceTracker():
| 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 |
| 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 |
@@ -97,168 +54,18 @@ class PerformanceTracker():
| | overkill. |
+-----------------+----------------------------------------------------+
| cumulative_risk | A dictionary representing the risk metrics |
| _metrics | calculated based on the positions aggregated |
| _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`|
+-----------------+----------------------------------------------------+
"""
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):
#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.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.calculate_performance()
self.todays_performance = PerformancePeriod(
self.todays_performance.positions,
self.todays_performance.ending_value,
self.todays_performance.ending_cash
)
| timestamp | System time evevent occurs in zipilne |
+-----------------+----------------------------------------------------+
def handle_simulation_end(self):
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():
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('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:
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_price
def __repr__(self):
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:
Position Tracking
=================
+-----------------+----------------------------------------------------+
| key | value |
+=================+====================================================+
@@ -272,97 +79,338 @@ class Position():
| 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
| timestamp | System time evevent occurs in zipilne |
+-----------------+----------------------------------------------------+
Performance Period
==================
+---------------+------------------------------------------------------+
| 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 |
+---------------+------------------------------------------------------+
| timestamp | System time evevent occurs in zipilne |
+---------------+------------------------------------------------------+
"""
import datetime
import msgpack
import pandas
import math
import zmq
import zipline.util as qutil
import zipline.protocol as zp
import zipline.finance.risk as risk
class PerformanceTracker():
"""
Tracks the performance of the zipstream as it is running in
the simulotr, relays this out to the Deluge broker and then
to the client.
+--------------------+ Result Stream +--------+
| PerformanceTracker | ----------------> | Deluge |
+--------------------+ +--------+
"""
def __init__(self, trading_environment):
self.trading_environment = trading_environment
self.trading_day = datetime.timedelta(hours = 6, minutes = 30)
self.calendar_day = datetime.timedelta(hours = 24)
self.period_start = self.trading_environment.period_start
self.period_end = self.trading_environment.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 = self.trading_environment.capital_base
self.returns = []
self.txn_count = 0
self.event_count = 0
self.result_stream = None
self.cumulative_performance = PerformancePeriod(
{},
self.capital_base,
starting_cash = self.capital_base
)
self.todays_performance = PerformancePeriod(
{},
self.capital_base,
starting_cash = self.capital_base
)
def publish_to(self, zmq_socket, context=None):
"""
Publish the performance results asynchronously to a
socket.
"""
ctx = context or zmq.Context.instance()
sock = ctx.socket(zmq.PUSH)
sock.connect(zmq_socket)
self.result_stream = sock
def to_dict(self):
"""
Creates a dictionary representing the state of this tracker.
Returns a dict object of the form:
"""
returns_list = [x.to_dict() for x in self.returns]
return {
'period_start' : self.period_start,
'period_end' : self.period_end,
'progress' : self.progress,
'cumulative_captial_used' : self.cumulative_capital_used,
'max_capital_used' : self.max_capital_used,
'last_close' : self.market_close,
'last_open' : self.market_open,
'capital_base' : self.capital_base,
'returns' : returns_list,
'cumulative_perf' : self.cumulative_performance.to_dict(),
'todays_perf' : self.todays_performance.to_dict(),
'cumulative_risk_metrics' : self.cumulative_risk_metrics.to_dict(),
'timestamp' : datetime.datetime.now(),
}
def process_event(self, 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):
#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.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
# Output Results
if self.result_stream:
# TODO: proper framing
self.result_stream.send_pyobj(self.to_dict())
#roll over positions to current day.
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):
assert False
self.risk_report = risk.RiskReport(
self.returns,
self.trading_environment
)
# Output Results
if self.result_stream:
# TODO: proper framing
self.result_stream.send_pyobj(self.risk_report.to_dict())
self.result_stream.send_pyobj(None)
def round_to_nearest(self, x, base=5):
return int(base * round(float(x)/base))
class Position():
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('updating position with txn for a different sid')
#we're covering a short or closing a position
if(self.amount + txn.amount == 0):
self.cost_basis = 0.0
self.amount = 0
else:
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_price
def __repr__(self):
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:
"""
return {
'sid' : self.sid,
'amount' : self.amount,
'cost_basis' : self.cost_basis,
'last_sale_price' : self.last_sale_price,
'last_sale_date' : self.last_sale_date,
'timestamp' : datetime.datetime.now(),
}
class PerformancePeriod():
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.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()
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(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):
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_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
return {
'ending_value' : self.ending_value,
'capital_used' : self.period_capital_used,
'starting_value' : self.starting_value,
'starting_cash' : self.starting_cash,
'ending_cash' : self.ending_cash,
'positions' : self.positions,
'timestamp' : datetime.datetime.now(),
}
position_list = []
for pos in self.positions:
position_list.append(pos.to_dict())
d['positions'] = positions_list
return d
zipline/finance/risk.py
+18 -6
View File
@@ -5,7 +5,6 @@ import numpy as np
import numpy.linalg as la
import zipline.util as qutil
import zipline.protocol as zp
from pymongo import ASCENDING, DESCENDING
class DailyReturn():
@@ -46,6 +45,7 @@ class RiskMetrics():
)
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)
@@ -90,10 +90,10 @@ class RiskMetrics():
| | and self.end_date. |
+-----------------+----------------------------------------------------+
"""
d = {
return {
'trading_days' : self.trading_days,
'benchmark_volatility' : self.benchmark_volatility,
'algo_volatility' : self.algo_volatility,
'algo_volatility' : self.algorithm_volatility,
'treasury_period_return': self.treasury_period_return,
'sharpe' : self.sharpe,
'beta' : self.beta,
@@ -101,7 +101,7 @@ class RiskMetrics():
'excess_return' : self.excess_return,
'max_drawdown' : self.max_drawdown
}
def __repr__(self):
statements = []
for metric in [
@@ -137,7 +137,6 @@ class RiskMetrics():
return period_returns, returns
def calculate_volatility(self, daily_returns):
#qutil.LOGGER.debug("trading days {td}".format(td=self.trading_days))
return np.std(daily_returns, ddof=1) * math.sqrt(self.trading_days)
def calculate_sharpe(self):
@@ -326,11 +325,24 @@ def advance_by_months(dt, jump_in_months):
class TradingEnvironment(object):
def __init__(self, benchmark_returns, treasury_curves):
def __init__(
self,
benchmark_returns,
treasury_curves,
period_start=None,
period_end=None,
capital_base=None,
frame_index=None
):
self.trading_days = []
self.trading_day_map = {}
self.treasury_curves = treasury_curves
self.benchmark_returns = benchmark_returns
self.frame_index = frame_index
self.period_start = period_start
self.period_end = period_end
self.capital_base = capital_base
for bm in benchmark_returns:
self.trading_days.append(bm.date)
self.trading_day_map[bm.date] = bm
zipline/finance/trading.py
+25 -16
View File
@@ -8,19 +8,27 @@ from zmq.core.poll import select
import zipline.messaging as qmsg
import zipline.util as qutil
import zipline.protocol as zp
import zipline.finance.performance as perf
class TradeSimulationClient(qmsg.Component):
def __init__(self, simulation_dt):
def __init__(self, trading_environment):
qmsg.Component.__init__(self)
self.received_count = 0
self.prev_dt = None
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
self.received_count = 0
self.prev_dt = None
self.event_queue = None
self.event_callbacks = []
self.txn_count = 0
self.trading_environment = trading_environment
self.current_dt = trading_environment.period_start
self.last_iteration_dur = datetime.timedelta(seconds=0)
assert self.trading_environment.frame_index != None
self.event_frame = pandas.DataFrame(
index=self.trading_environment.frame_index
)
self.perf = perf.PerformanceTracker(self.trading_environment)
@property
def get_id(self):
@@ -67,9 +75,9 @@ class TradeSimulationClient(qmsg.Component):
self.run_callbacks()
#update time based on receipt of the order
self.last_iteration_duration = datetime.datetime.utcnow() - event_start
self.last_iteration_dur = datetime.datetime.utcnow() - event_start
self.current_dt = self.current_dt + self.last_iteration_duration
self.current_dt = self.current_dt + self.last_iteration_dur
#signal done to order source.
self.order_socket.send(str(zp.ORDER_PROTOCOL.BREAK))
@@ -95,15 +103,16 @@ class TradeSimulationClient(qmsg.Component):
self.order_socket.send(str(zp.ORDER_PROTOCOL.DONE))
def queue_event(self, event):
self.perf.process_event(event)
if self.event_queue == None:
self.event_queue = {}
self.event_queue = []
series = event.as_series()
self.event_queue[event.dt] = series
self.event_queue.append(series)
def get_frame(self):
frame = pandas.DataFrame(self.event_queue)
self.event_queue = None
return frame
for event in self.event_queue:
self.event_frame[event['sid']] = event
return self.event_frame
class OrderDataSource(qmsg.DataSource):
"""DataSource that relays orders from the client"""
zipline/protocol.py
+23 -87
View File
@@ -119,98 +119,13 @@ import numbers
import datetime
import pytz
import copy
import pandas
from collections import namedtuple
import zipline.util as qutil
from protocol_utils import Enum, FrameExceptionFactory, namedict
#import ujson
#import ultrajson_numpy
from ctypes import Structure, c_ubyte
def Enum(*options):
"""
Fast enums are very important when we want really tight zmq
loops. These are probably going to evolve into pure C structs
anyways so might as well get going on that.
"""
class cstruct(Structure):
_fields_ = [(o, c_ubyte) for o in options]
return cstruct(*range(len(options)))
def FrameExceptionFactory(name):
"""
Exception factory with a closure around the frame class name.
"""
class InvalidFrame(Exception):
def __init__(self, got):
self.got = got
def __str__(self):
return "Invalid {framecls} Frame: {got}".format(
framecls = name,
got = self.got,
)
return InvalidFrame
class namedict(object):
"""
So that you can use::
foo.BAR
-- or --
foo['BAR']
For more complex structs use collections.namedtuple:
"""
def __init__(self, dct=None):
if dct:
self.__dict__.update(dct)
def __setitem__(self, key, value):
"""
Required for use by pymongo as_class parameter to find.
"""
if(key == '_id'):
self.__dict__['id'] = value
else:
self.__dict__[key] = value
def __getitem__(self, key):
return self.__dict__[key]
def keys(self):
return self.__dict__.keys()
def as_dict(self):
# shallow copy is O(n)
return copy.copy(self.__dict__)
def delete(self, key):
del(self.__dict__[key])
def merge(self, other_nd):
assert isinstance(other_nd, namedict)
self.__dict__.update(other_nd.__dict__)
def __repr__(self):
return "namedict: " + str(self.__dict__)
def __eq__(self, other):
# !!!!!!!!!!!!!!!!!!!!
# !!!! DANGEROUS !!!!!
# !!!!!!!!!!!!!!!!!!!!
return other != None and self.__dict__ == other.__dict__
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
# ================
@@ -279,6 +194,27 @@ COMPONENT_STATE = Enum(
'EXCEPTION' , # 2
)
# NOFAILURE - Component is either not running or has not failed
# ALGOEXCEPT - Exception thrown in the given algorithm
# HOSTEXCEPT - Exception thrown on our end.
# INTERRUPT - Manually interuptted by user
COMPONENT_FAILURE = Enum(
'NOFAILURE' ,
'ALGOEXCEPT' ,
'HOSTEXCEPT' ,
'INTERRUPT' ,
)
BACKTEST_STATE = Enum(
'IDLE' ,
'QUEUED' ,
'INPROGRESS' ,
'CANCELLED' , # cancelled ( before natural completion )
'EXCEPTION' , # failure ( due to unnatural causes )
'DONE' , # done ( naturally completed )
)
# ==================
# Datasource Protocol
# ==================
zipline/protocol_utils.py
+93
View File
@@ -0,0 +1,93 @@
import copy
import pandas
from ctypes import Structure, c_ubyte
def Enum(*options):
"""
Fast enums are very important when we want really tight zmq
loops. These are probably going to evolve into pure C structs
anyways so might as well get going on that.
"""
class cstruct(Structure):
_fields_ = [(o, c_ubyte) for o in options]
return cstruct(*range(len(options)))
def FrameExceptionFactory(name):
"""
Exception factory with a closure around the frame class name.
"""
class InvalidFrame(Exception):
def __init__(self, got):
self.got = got
def __str__(self):
return "Invalid {framecls} Frame: {got}".format(
framecls = name,
got = self.got,
)
return InvalidFrame
class namedict(object):
"""
Namedicts are dict like objects that have fields accessible by attribute lookup
as well as being indexable and iterable::
HEARTBEAT_PROTOCOL = namedict({
'REQ' : b'\x01',
'REP' : b'\x02',
})
HEARTBEAT_PROTOCOL.REQ # syntactic sugar
HEARTBEAT_PROTOCOL.REP # oh suga suga
For more complex structs use collections.namedtuple:
"""
def __init__(self, dct=None):
if(dct):
self.__dict__.update(dct)
def __setitem__(self, key, value):
"""
Required for use by pymongo as_class parameter to find.
"""
if(key == '_id'):
self.__dict__['id'] = value
else:
self.__dict__[key] = value
def __getitem__(self, key):
return self.__dict__[key]
def keys(self):
return self.__dict__.keys()
def as_dict(self):
# shallow copy is O(n)
return copy.copy(self.__dict__)
def delete(self, key):
del(self.__dict__[key])
def merge(self, other_nd):
assert isinstance(other_nd, namedict)
self.__dict__.update(other_nd.__dict__)
def __repr__(self):
return "namedict: " + str(self.__dict__)
def __eq__(self, other):
# !!!!!!!!!!!!!!!!!!!!
# !!!! DANGEROUS !!!!!
# !!!!!!!!!!!!!!!!!!!!
return other != None and self.__dict__ == other.__dict__
def has_attr(self, name):
return self.__dict__.has_key(name)
def as_series(self):
s = pandas.Series(self.__dict__)
s.name = self.sid
return s
zipline/test/client.py
+2 -3
View File
@@ -84,9 +84,8 @@ class TestAlgorithm():
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
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/test_finance.py
+21 -15
View File
@@ -207,7 +207,13 @@ class FinanceTestCase(TestCase):
set1 = SpecificEquityTrades("flat-133", trade_history)
trading_client = TradeSimulationClient(start_date)
self.trading_environment.period_start = trade_history[0].dt
self.trading_environment.period_end = trade_history[-1].dt
self.trading_environment.capital_base = 10000
self.trading_environment.frame_index = ['sid', 'volume', 'dt', \
'price', 'changed']
trading_client = TradeSimulationClient(self.trading_environment)
#client will send 10 orders for 100 shares of 133
test_algo = TestAlgorithm(133, 100, 10, trading_client)
@@ -280,25 +286,25 @@ class FinanceTestCase(TestCase):
volume,
start_date,
trade_time_increment,
self.trading_environment )
self.trading_environment
)
self.trading_environment.period_start = trade_history[0].dt
self.trading_environment.period_end = trade_history[-1].dt
self.trading_environment.capital_base = 10000
self.trading_environment.frame_index = ['sid', 'volume', 'dt', \
'price', 'changed']
set1 = SpecificEquityTrades("flat-133", trade_history)
#client sill send 10 orders for 100 shares of 133
trading_client = TradeSimulationClient(start_date)
trading_client = TradeSimulationClient(self.trading_environment)
test_algo = TestAlgorithm(133, 100, 10, trading_client)
order_source = OrderDataSource()
transaction_sim = TransactionSimulator()
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([
trading_client,
order_source,
@@ -339,19 +345,19 @@ class FinanceTestCase(TestCase):
self.assertEqual(
transaction_sim.txn_count,
perf_tracker.txn_count,
trading_client.perf.txn_count,
"The perf tracker should handle the same number of transactions \
as the simulator emits."
)
self.assertEqual(
len(perf_tracker.cumulative_performance.positions),
len(trading_client.perf.cumulative_performance.positions),
1,
"Portfolio should have one position."
)
self.assertEqual(
perf_tracker.cumulative_performance.positions[133].sid,
trading_client.perf.cumulative_performance.positions[133].sid,
133,
"Portfolio should have one position in 133."
)
zipline/test/test_perf_tracking.py
+35 -21
View File
@@ -506,34 +506,46 @@ shares in position"
trade_count = 100
sid = 133
price = [10.1] * trade_count
price = 10.1
price_list = [price] * trade_count
volume = [100] * trade_count
start_date = datetime.datetime.strptime("01/01/2011","%m/%d/%Y")
start_date = start_date.replace(tzinfo=pytz.utc)
trade_time_increment = datetime.timedelta(days=1)
trade_history = factory.create_trade_history(
sid,
price,
price_list,
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
sid2 = 134
price2 = 12.12
price2_list = [price2] * trade_count
trade_history2 = factory.create_trade_history(
sid2,
price2_list,
volume,
start_date,
trade_time_increment,
self.trading_environment
)
trade_history.extend(trade_history2)
self.trading_environment.period_start = trade_history[0].dt
self.trading_environment.period_end = trade_history[-1].dt
self.trading_environment.capital_base = 1000.0
self.trading_environment.frame_index = ['sid', 'volume', 'dt', \
'price', 'changed']
client = TradeSimulationClient(self.trading_environment)
for event in trade_history:
#create a transaction for all but
#one trade, to simulate None transaction
if(event.dt != start):
#first trade in each sid, to simulate None transaction
if(event.dt != self.trading_environment.period_start):
txn = zp.namedict({
'sid' : event.sid,
'amount' : -25,
@@ -543,17 +555,19 @@ shares in position"
})
else:
txn = None
event[zp.TRANSFORM_TYPE.TRANSACTION] = txn
trade_client.queue_event(event)
event[zp.TRANSFORM_TYPE.TRANSACTION] = txn
client.queue_event(event)
df = trade_client.get_frame()
tracker.update(df)
df = client.get_frame()
#we skip one trade, to test case of None transaction
txn_count = len(trade_history) - 1
self.assertEqual(tracker.txn_count, txn_count)
self.assertEqual(df[133]['price'], price)
self.assertEqual(df[134]['price'], price2)
cumulative_pos = tracker.cumulative_performance.positions[sid]
expected_size = txn_count * -25
#we skip two trades, to test case of None transaction
txn_count = len(trade_history) - 2
self.assertEqual(client.perf.txn_count, txn_count)
cumulative_pos = client.perf.cumulative_performance.positions[sid]
expected_size = txn_count / 2 * -25
self.assertEqual(cumulative_pos.amount, expected_size)
zipline/test/test_risk.py
+5 -2
View File
@@ -11,6 +11,9 @@ class Risk(unittest.TestCase):
def setUp(self):
qutil.configure_logging()
start_date = datetime.datetime(year=2006, month=1, day=1, tzinfo=pytz.utc)
end_date = datetime.datetime(year=2006, month=12, day=31, tzinfo=pytz.utc)
self.benchmark_returns, self.treasury_curves = \
factory.load_market_data()
@@ -23,9 +26,9 @@ class Risk(unittest.TestCase):
self.oneday = datetime.timedelta(days=1)
self.tradingday = datetime.timedelta(hours=6, minutes=30)
self.dt = datetime.datetime.utcnow()
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.trading_calendar)
def tearDown(self):
zipline/zmq_utils.py
+16
View File
@@ -0,0 +1,16 @@
import gevent
from gevent_zeromq import zmq
def ZmqConsole(sock_typ, socket_addr, sock_conn=None, context=None):
context = context or zmq.Context.instance()
socket = context.socket(zmq.PULL)
socket.connect('tcp://127.0.0.1:3141')
def console():
while True:
msg = socket.recv()
print msg
import pdb; pdb.set_trace()
return gevent.spawn(console)