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

Merge pull request #35 from quantopian/bt_persist

Browse Source 
Bt persist
This commit is contained in:
fawce
2012-04-18 09:45:29 -07:00
parent 89e6796062 1b5b92d75f
commit 12d3963431
6 changed files with 139 additions and 135 deletions
Download Patch File Download Diff File Expand all files Collapse all files
zipline/finance/performance.py
+27 -31
View File
@@ -38,10 +38,6 @@ Performance Tracking
+-----------------+----------------------------------------------------+
| capital_base | The initial capital assumed for this tracker. |
+-----------------+----------------------------------------------------+
| returns | List of dicts representing daily returns. See the |
| | comments for |
| | :py:meth:`zipline.finance.risk.DailyReturn.to_dict`|
+-----------------+----------------------------------------------------+
| cumulative_perf | A dictionary representing the cumulative |
| | performance through all the events delivered to |
| | this tracker. For details see the comments on |
@@ -61,8 +57,6 @@ Performance Tracking
| | For details look at the comments for |
| | :py:meth:`zipline.finance.risk.RiskMetrics.to_dict`|
+-----------------+----------------------------------------------------+
| timestamp | System time evevent occurs in zipilne |
+-----------------+----------------------------------------------------+
Position Tracking
@@ -78,14 +72,10 @@ Position Tracking
+-----------------+----------------------------------------------------+
| last_sale_price | price at last sale of the security on the exchange |
+-----------------+----------------------------------------------------+
| last_sale_date | datetime of the last trade of the position's |
| | security on the exchange |
+-----------------+----------------------------------------------------+
| transactions | all the transactions that were acrued into this |
| | position. |
+-----------------+----------------------------------------------------+
| timestamp | System time event occurs in zipilne |
+-----------------+----------------------------------------------------+
Performance Period
==================
@@ -116,8 +106,7 @@ Performance Period
| returns | percentage returns for the entire portfolio over the |
| | period |
+---------------+------------------------------------------------------+
| timestamp | System time evevent occurs in zipilne |
+---------------+------------------------------------------------------+
"""
import datetime
@@ -185,7 +174,9 @@ class PerformanceTracker():
# initial portfolio positions have zero value
0,
# initial cash is your capital base.
starting_cash = self.capital_base
starting_cash = self.capital_base,
# save the transactions for the daily periods
keep_transactions = True
)
def get_portfolio(self):
@@ -210,24 +201,19 @@ class PerformanceTracker():
Creates a dictionary representing the state of this tracker.
Returns a dict object of the form described in header comments.
"""
returns_list = [x.to_dict() for x in self.returns]
return {
'started_at' : self.started_at,
'period_start' : self.period_start,
'period_end' : self.period_end,
'progress' : self.progress,
'cumulative_captial_used' : self.cumulative_perf.cumulative_capital_used,
'max_capital_used' : self.cumulative_perf.max_capital_used,
'cumulative_capital_used' : self.cumulative_performance.cumulative_capital_used,
'max_capital_used' : self.cumulative_performance.max_capital_used,
'last_close' : self.market_close,
'last_open' : self.market_open,
'capital_base' : self.capital_base,
'returns' : returns_list,
'cumulative_perf' : self.cumulative_performance.to_dict(),
'todays_perf' : self.todays_performance.to_dict(),
'daily_perf' : self.todays_performance.to_dict(),
'cumulative_risk_metrics' : self.cumulative_risk_metrics.to_dict(),
'timestamp' : datetime.datetime.now(),
}
def log_order(self, order):
@@ -271,6 +257,7 @@ class PerformanceTracker():
trading_environment=self.trading_environment
)
# increment the day counter before we move markers forward.
self.day_count += 1.0
# calculate progress of test
@@ -295,7 +282,8 @@ class PerformanceTracker():
self.todays_performance = PerformancePeriod(
self.todays_performance.positions,
self.todays_performance.ending_value,
self.todays_performance.ending_cash
self.todays_performance.ending_cash,
keep_transactions = True
)
def handle_simulation_end(self):
@@ -375,15 +363,13 @@ class Position():
'sid' : self.sid,
'amount' : self.amount,
'cost_basis' : self.cost_basis,
'last_sale_price' : self.last_sale_price,
'last_sale_date' : self.last_sale_date,
'timestamp' : datetime.datetime.now()
'last_sale_price' : self.last_sale_price
}
class PerformancePeriod():
def __init__(self, initial_positions, starting_value, starting_cash):
def __init__(self, initial_positions, starting_value, starting_cash, keep_transactions=False):
self.ending_value = 0.0
self.period_capital_used = 0.0
self.pnl = 0.0
@@ -393,6 +379,7 @@ class PerformancePeriod():
#cash balance at start of period
self.starting_cash = starting_cash
self.ending_cash = starting_cash
self.keep_transactions = keep_transactions
self.processed_transactions = []
self.cumulative_capital_used = 0.0
self.max_capital_used = 0.0
@@ -443,7 +430,8 @@ class PerformancePeriod():
self.max_leverage = 1.1 * self.max_capital_used / self.starting_cash
# add transaction to the list of processed transactions
self.processed_transactions.append(txn)
if self.keep_transactions:
self.processed_transactions.append(txn)
def round_to_nearest(self, x, base=5):
return int(base * round(float(x)/base))
@@ -465,7 +453,8 @@ class PerformancePeriod():
Creates a dictionary representing the state of this performance
period. See header comments for a detailed description.
"""
positions = self.get_positions()
positions = self.get_positions_list()
transactions = [x.as_dict() for x in self.processed_transactions]
return {
'ending_value' : self.ending_value,
@@ -475,10 +464,9 @@ class PerformancePeriod():
'ending_cash' : self.ending_cash,
'portfolio_value': self.ending_cash + self.ending_value,
'positions' : positions,
'timestamp' : datetime.datetime.now(),
'pnl' : self.pnl,
'returns' : self.returns,
'transactions' : self.processed_transactions,
'transactions' : transactions,
}
def to_namedict(self):
@@ -512,6 +500,14 @@ class PerformancePeriod():
return positions
#
def get_positions_list(self):
positions = []
for sid, pos in self.positions.iteritems():
cur = pos.to_dict()
positions.append(cur)
return positions
zipline/finance/risk.py
+4 -4
View File
@@ -352,10 +352,10 @@ class RiskReport():
provided for each period.
"""
return {
'1_month' : [x.to_dict() for x in self.month_periods],
'3_month' : [x.to_dict() for x in self.three_month_periods],
'6_month' : [x.to_dict() for x in self.six_month_periods],
'12_month' : [x.to_dict() for x in self.year_periods]
'one_month' : [x.to_dict() for x in self.month_periods],
'three_month' : [x.to_dict() for x in self.three_month_periods],
'six_month' : [x.to_dict() for x in self.six_month_periods],
'twelve_month' : [x.to_dict() for x in self.year_periods]
}
def periodsInRange(self, months_per, start, end):
zipline/finance/trading.py
+20 -21
View File
@@ -2,6 +2,7 @@ import datetime
import pytz
import math
import pandas
import time
from collections import Counter
@@ -37,8 +38,8 @@ class TradeSimulationClient(qmsg.Component):
self.current_dt = trading_environment.period_start
self.last_iteration_dur = datetime.timedelta(seconds=0)
self.algorithm = None
self.attempts = 0
self.max_attempts = 1000
self.max_wait = datetime.timedelta(seconds=7)
self.last_msg_dt = datetime.datetime.utcnow()
assert self.trading_environment.frame_index != None
self.event_frame = pandas.DataFrame(
@@ -75,7 +76,7 @@ class TradeSimulationClient(qmsg.Component):
if self.result_feed in socks and \
socks[self.result_feed] == self.zmq.POLLIN:
self.attempts = 0
self.last_msg_dt = datetime.datetime.utcnow()
# get the next message from the result feed
msg = self.result_feed.recv()
@@ -105,10 +106,10 @@ class TradeSimulationClient(qmsg.Component):
# drained. Signal the order_source that we're done, and
# the done will cascade through the whole zipline.
# shutdown the feedback loop to the OrderDataSource
if self.attempts > self.max_attempts:
self.signal_order_done()
else:
self.attempts += 1
wait_time = datetime.datetime.utcnow() - self.last_msg_dt
if wait_time > self.max_wait:
self.signal_order_done()
def process_event(self, event):
# track the number of transactions, for testing purposes.
if(event.TRANSACTION != None):
@@ -420,20 +421,7 @@ class TransactionSimulator(qmsg.BaseTransform):
# we cap the volume share at 25% of a trade
if volume_share == .25:
break
if simulated_amount == 0:
warning = """
Calculated a zero volume transation on trade:
{event}
for order:
{order}
"""
warning = warning.format(
event=str(event),
order=str(order)
)
qutil.LOGGER.warn(warning)
orders = [ x for x in orders if abs(x.amount - x.filled) > 0 and x.dt.day >= event.dt.day]
self.open_orders[event.sid] = orders
@@ -448,6 +436,17 @@ for order:
direction
)
else:
warning = """
Calculated a zero volume transaction on trade:
{event}
for orders:
{orders}
"""
warning = warning.format(
event=str(event),
orders=str(orders)
)
qutil.LOGGER.warn(warning)
return None
zipline/protocol.py
+55 -70
View File
@@ -615,90 +615,69 @@ def PERF_FRAME(perf):
"""
Frame the performance update created at the end of each simulated trading
day. The msgpack is a tuple with the first element statically set to 'PERF'.
Frames prepared by this method are sent via the same socket as
Frames prepared by RISK_FRAME. So, both methods prefix the payload with
a shorthand for their type. That way, all messages received from the socket
can be PERF_UNFRAMED(), whether they are risk or perf.
Like RISK_FRAME, this method calls BT_UPDATE_FRAME internally, so that
clients can call BT_UPDATE_UNFRAME for all messages from the backtest.
:param perf: the dictionary created by zipline.trade_client.perf
:rvalue: a msgpack string
"""
#TODO: add asserts...
assert isinstance(perf['started_at'], datetime.datetime)
assert isinstance(perf['period_start'], datetime.datetime)
assert isinstance(perf['period_end'], datetime.datetime)
assert isinstance(perf['last_close'], datetime.datetime)
assert isinstance(perf['last_open'], datetime.datetime)
#pull some special fields from the perf for easy access
date = perf['last_close']
tp = perf['todays_perf']
assert isinstance(perf['daily_perf'], dict)
assert isinstance(perf['cumulative_perf'], dict)
tp = perf['daily_perf']
cp = perf['cumulative_perf']
risk = perf['cumulative_risk_metrics']
# aggregate the day's transactions, which are nested in their
# respsective positions.
transactions = []
for txn in tp['transactions']:
cur = {
'date':EPOCH(txn.dt),
'amount': txn.amount,
'price': txn.price,
'sid':txn.sid
}
transactions.append(cur)
positions = []
for sid, pos in tp['positions'].iteritems():
cur = {
'cost_basis':pos['cost_basis'],
'sid' :pos['sid'],
'last_sale' :pos['last_sale_price'],
'amount' :pos['amount']
}
positions.append(cur)
daily_perf = {
'date' : EPOCH(date),
'returns' : tp['returns'],
'pnl' : tp['pnl'],
'market_value' : tp['ending_value'],
'portfolio_value' : tp['portfolio_value'],
'starting_cash' : tp['starting_cash'],
'ending_cash' : tp['ending_cash'],
'capital_used' : tp['capital_used'],
'transactions' : transactions,
'positions' : positions
}
cumulative_perf = {
'alpha' : risk['alpha'],
'beta' : risk['beta'],
'sharpe' : risk['sharpe'],
'volatility' : risk['algo_volatility'],
'benchmark_volatility' : risk['benchmark_volatility'],
'benchmark_returns' : risk['benchmark_period_return'],
'max_drawdown' : risk['max_drawdown'],
'total_returns' : cp['returns'],
'pnl' : cp['pnl'],
'capital_used' : cp['capital_used']
}
assert isinstance(tp['transactions'], list)
assert isinstance(cp['transactions'], list)
assert isinstance(tp['positions'], list)
assert isinstance(cp['positions'], list)
perf['started_at'] = EPOCH(perf['started_at'])
perf['period_start'] = EPOCH(perf['period_start'])
perf['period_end'] = EPOCH(perf['period_end'])
perf['last_close'] = EPOCH(perf['last_close'])
perf['last_open'] = EPOCH(perf['last_open'])
# nest the cumulative performance data in the daily.
daily_perf['cumulative'] = cumulative_perf
result = {
'started_at' : EPOCH(perf['started_at']),
'daily' : [daily_perf],
'percent_complete' : perf['progress'],
}
return msgpack.dumps(tuple(['PERF', result]))
tp['transactions'] = convert_transactions(tp['transactions'])
cp['transactions'] = convert_transactions(cp['transactions'])
return BT_UPDATE_FRAME('PERF', perf)
def convert_transactions(transactions):
results = []
for txn in transactions:
txn['date'] = EPOCH(txn['dt'])
del(txn['dt'])
results.append(txn)
return results
def RISK_FRAME(risk):
return msgpack.dumps(tuple(['RISK', risk]))
return BT_UPDATE_FRAME('RISK', risk)
def PERF_UNFRAME(msg):
prefix, payload = msgpack.loads(msg)
def BT_UPDATE_FRAME(prefix, payload):
"""
Frames prepared by RISK_FRAME and PERF_FRAME methods are sent via the same
socket. This method provides a prefix to allow for muxing the messages
onto a single socket.
"""
return msgpack.dumps(tuple([prefix, payload]))
def BT_UPDATE_UNFRAME(msg):
"""
Risk and Perf framing methods prefix the payload with
a shorthand for their type. That way, all messages received from the socket
can be PERF_FRAMED(), whether they are risk or perf.
"""
prefix, payload = msgpack.loads(msg, use_list=True)
return dict(prefix=prefix, payload=payload)
# -----------------------
@@ -730,6 +709,12 @@ def EPOCH(utc_datetime):
ms = seconds * 1000
return ms
def UN_EPOCH(ms_since_epoch):
seconds_since_epoch = ms_since_epoch / 1000
delta = datetime.timedelta(seconds = seconds_since_epoch)
dt = UNIX_EPOCH + delta
return dt
def PACK_DATE(event):
"""
Packs the datetime property of event into msgpack'able longs.
zipline/test/algorithms.py
+33 -1
View File
@@ -76,7 +76,39 @@ class TestAlgorithm():
self.incr += 1
def get_sid_filter(self):
return [self.sid]
return [self.sid]
#
class HeavyBuyAlgorithm():
"""
This algorithm will send a specified number of orders, to allow unit tests
to verify the orders sent/received, transactions created, and positions
at the close of a simulation.
"""
def __init__(self, sid, amount):
self.sid = sid
self.amount = amount
self.incr = 0
self.done = False
self.order = None
self.frame_count = 0
self.portfolio = None
def set_order(self, order_callable):
self.order = order_callable
def set_portfolio(self, portfolio):
self.portfolio = portfolio
def handle_frame(self, frame):
self.frame_count += 1
#place an order for 100 shares of sid
self.order(self.sid, self.amount)
self.incr += 1
def get_sid_filter(self):
return [self.sid]
class NoopAlgorithm(object):
"""
zipline/test/test_finance.py
-8
View File
@@ -139,14 +139,6 @@ class FinanceTestCase(TestCase):
zipline.trading_client.order_count
)
# the number of transactions in the performance tracker's cumulative
# period should be the same as the number of orders place by the
# algorithm.
self.assertEqual(
zipline.trading_client.order_count,
len(zipline.trading_client.perf.cumulative_performance.processed_transactions)
)
@timed(EXTENDED_TIMEOUT)
def test_aggressive_buying(self):