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

RFT: Remove Position management from PerformancePeriod. This cuts down

Browse Source 
on the number of per-tick update that occur since they were duplicated
per each PerformancePeriod. Also opens up the path to cythonizing the
entire object
This commit is contained in:
Dale Jung
2015-03-18 22:48:14 -04:00
committed by Eddie Hebert
parent 6cd2f4f1f2
commit 7892a6943f
14 changed files with 79955 additions and 503 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/generate_new_sample_saved_state.py
+2
View File
@@ -25,6 +25,7 @@ from zipline.finance.blotter import Blotter, Order
from zipline.finance.commission import PerShare, PerTrade, PerDollar
from zipline.finance.performance.period import PerformancePeriod
from zipline.finance.performance.position import Position
from zipline.finance.performance.position_tracker import PositionTracker
from zipline.finance.performance.tracker import PerformanceTracker
from zipline.finance.risk.cumulative import RiskMetricsCumulative
from zipline.finance.risk.period import RiskMetricsPeriod
@@ -75,6 +76,7 @@ argument_list = [
(PerDollar, ()),
(PerformancePeriod, (10000,)),
(Position, (8554,)),
(PositionTracker, ()),
(PerformanceTracker, (sim_params_minute,)),
(RiskMetricsCumulative, (sim_params_minute,)),
(RiskMetricsPeriod, (returns.index[0], returns.index[0], returns)),
tests/resources/saved_state_archive/zipline.finance.performance.period.PerformancePeriod/State_Version_2
+152
View File
@@ -0,0 +1,152 @@
(dp0
S'obj_state'
p1
(dp2
S'_account_store'
p3
ccopy_reg
_reconstructor
p4
(czipline.protocol
Account
p5
c__builtin__
object
p6
Ntp7
Rp8
(dp9
S'regt_margin'
p10
Finf
sS'maintenance_margin_requirement'
p11
F0.0
sS'day_trades_remaining'
p12
Finf
sS'buying_power'
p13
Finf
sS'net_leverage'
p14
F0.0
sS'settled_cash'
p15
F0.0
sS'cushion'
p16
F0.0
sS'_stateversion_'
p17
I1
sS'leverage'
p18
F0.0
sS'regt_equity'
p19
F0.0
sS'excess_liquidity'
p20
F0.0
sS'available_funds'
p21
F0.0
sS'equity_with_loan'
p22
F0.0
sS'initial_margin_requirement'
p23
F0.0
sS'net_liquidation'
p24
F0.0
sS'total_positions_value'
p25
F0.0
sS'accrued_interest'
p26
F0.0
sbsS'orders_by_modified'
p27
(dp28
sS'keep_transactions'
p29
I01
sS'ending_cash'
p30
I10000
sS'processed_transactions'
p31
(dp32
sS'ending_value'
p33
F0.0
sS'starting_cash'
p34
I10000
sg17
I2
sS'pnl'
p35
F0.0
sS'period_cash_flow'
p36
F0.0
sS'serialize_positions'
p37
I01
sS'keep_orders'
p38
I00
sS'_portfolio_store'
p39
g4
(czipline.protocol
Portfolio
p40
g6
Ntp41
Rp42
(dp43
g17
I1
sS'portfolio_value'
p44
F0.0
sS'cash'
p45
F0.0
sg34
F0.0
sS'returns'
p46
F0.0
sS'capital_used'
p47
F0.0
sg35
F0.0
sS'positions'
p48
(dp49
sS'positions_value'
p50
F0.0
sS'start_date'
p51
NsbsS'starting_value'
p52
F0.0
sS'period_open'
p53
NsS'period_close'
p54
NsS'orders_by_id'
p55
(dp56
ssS'initargs'
p57
NsS'newargs'
p58
Ns.
tests/resources/saved_state_archive/zipline.finance.performance.position_tracker.PositionTracker/State_Version_1
+145
View File
@@ -0,0 +1,145 @@
(dp0
S'obj_state'
p1
(dp2
S'positions'
p3
ccopy_reg
_reconstructor
p4
(czipline.finance.performance.position
positiondict
p5
c__builtin__
dict
p6
(dp7
tp8
Rp9
sS'unpaid_dividends'
p10
g4
(cpandas.core.frame
DataFrame
p11
c__builtin__
object
p12
Ntp13
Rp14
g4
(cpandas.core.internals
BlockManager
p15
g12
Ntp16
Rp17
((lp18
cnumpy.core.multiarray
_reconstruct
p19
(cpandas.core.index
Index
p20
(I0
tp21
S'b'
p22
tp23
Rp24
((I1
(I4
tp25
cnumpy
dtype
p26
(S'O8'
p27
I0
I1
tp28
Rp29
(I3
S'|'
p30
NNNI-1
I-1
I63
tp31
bI00
(lp32
S'id'
p33
aS'payment_sid'
p34
aS'cash_amount'
p35
aS'share_count'
p36
atp37
(Ntp38
tp39
bag19
(g20
(I0
tp40
g22
tp41
Rp42
((I1
(I0
tp43
g29
I00
(lp44
tp45
(Ntp46
tp47
ba(lp48
g19
(cnumpy
ndarray
p49
(I0
tp50
g22
tp51
Rp52
(I1
(I4
I0
tp53
g29
I00
(lp54
tp55
ba(lp56
g19
(g20
(I0
tp57
g22
tp58
Rp59
((I1
(I4
tp60
g29
I00
(lp61
g33
ag34
ag35
ag36
atp62
(Ntp63
tp64
batp65
bbsS'_stateversion_'
p66
I1
ssS'initargs'
p67
NsS'newargs'
p68
Ns.
tests/resources/saved_state_archive/zipline.finance.performance.tracker.PerformanceTracker/State_Version_2
+78877
View File
File diff suppressed because one or more lines are too long
tests/serialization_cases.py
+113
View File
@@ -0,0 +1,113 @@
import datetime
import pytz
import nose.tools as nt
import pandas.util.testing as tm
import pandas as pd
from zipline.finance.blotter import Blotter, Order
from zipline.finance.commission import PerShare, PerTrade, PerDollar
from zipline.finance.performance.period import PerformancePeriod
from zipline.finance.performance.position import Position
from zipline.finance.performance.tracker import PerformanceTracker
from zipline.finance.performance.position_tracker import PositionTracker
from zipline.finance.risk.cumulative import RiskMetricsCumulative
from zipline.finance.risk.period import RiskMetricsPeriod
from zipline.finance.risk.report import RiskReport
from zipline.finance.slippage import (
FixedSlippage,
Transaction,
VolumeShareSlippage
)
from zipline.protocol import Account
from zipline.protocol import Portfolio
from zipline.protocol import Position as ProtocolPosition
from zipline.finance.trading import SimulationParameters
from zipline.utils import factory
def stringify_cases(cases, func=None):
# get better test case names
results = []
if func is None:
func = lambda case: case[0].__name__
for case in cases:
new_case = list(case)
key = func(case)
new_case.insert(0, key)
results.append(new_case)
return results
sim_params_daily = SimulationParameters(
datetime.datetime(2013, 6, 19, tzinfo=pytz.UTC),
datetime.datetime(2013, 6, 19, tzinfo=pytz.UTC),
10000,
emission_rate='daily')
sim_params_minute = SimulationParameters(
datetime.datetime(2013, 6, 19, tzinfo=pytz.UTC),
datetime.datetime(2013, 6, 19, tzinfo=pytz.UTC),
10000,
emission_rate='minute')
returns = factory.create_returns_from_list(
[1.0], sim_params_daily)
def object_serialization_cases(skip_daily=False):
# Wrapped in a function to recreate DI objects.
cases = [
(Blotter, (), {}, 'repr'),
(Order, (datetime.datetime(2013, 6, 19), 8554, 100), {}, 'dict'),
(PerShare, (), {}, 'dict'),
(PerTrade, (), {}, 'dict'),
(PerDollar, (), {}, 'dict'),
(PerformancePeriod,
(10000,), {'position_tracker': PositionTracker()}, 'to_dict'),
(Position, (8554,), {}, 'dict'),
(PositionTracker, (), {}, 'dict'),
(PerformanceTracker, (sim_params_minute,), {}, 'to_dict'),
(RiskMetricsCumulative, (sim_params_minute,), {}, 'to_dict'),
(RiskMetricsPeriod,
(returns.index[0], returns.index[0], returns), {}, 'to_dict'),
(RiskReport, (returns, sim_params_minute), {}, 'to_dict'),
(FixedSlippage, (), {}, 'dict'),
(Transaction,
(8554, 10, datetime.datetime(2013, 6, 19), 100, "0000"), {},
'dict'),
(VolumeShareSlippage, (), {}, 'dict'),
(Account, (), {}, 'dict'),
(Portfolio, (), {}, 'dict'),
(ProtocolPosition, (8554,), {}, 'dict')
]
if not skip_daily:
cases.extend([
(PerformanceTracker, (sim_params_daily,), {}, 'to_dict'),
(RiskMetricsCumulative, (sim_params_daily,), {}, 'to_dict'),
(RiskReport, (returns, sim_params_daily), {}, 'to_dict'),
])
return stringify_cases(cases)
def assert_dict_equal(d1, d2):
# check keys
nt.assert_is_instance(d1, dict)
nt.assert_is_instance(d2, dict)
nt.assert_set_equal(set(d1.keys()), set(d2.keys()))
for k in d1:
v1 = d1[k]
v2 = d2[k]
asserter = nt.assert_equal
if isinstance(v1, pd.DataFrame):
asserter = tm.assert_frame_equal
if isinstance(v1, pd.Series):
asserter = tm.assert_series_equal
try:
asserter(v1, v2)
except AssertionError:
raise AssertionError('{k} is not equal'.format(k=k))
tests/test_perf_tracking.py
+214 -33
View File
@@ -15,6 +15,7 @@
from __future__ import division
import pickle
import collections
from datetime import (
datetime,
@@ -25,6 +26,7 @@ import operator
import unittest
from nose_parameterized import parameterized
import nose.tools as nt
import pytz
import itertools
@@ -52,6 +54,10 @@ onesec = timedelta(seconds=1)
oneday = timedelta(days=1)
tradingday = timedelta(hours=6, minutes=30)
# nose.tools changed name in python 3
if not hasattr(nt, 'assert_count_equal'):
nt.assert_count_equal = nt.assert_items_equal
def check_perf_period(pp,
gross_leverage,
@@ -215,6 +221,34 @@ def calculate_results(host,
return results
def check_perf_tracker_serialization(perf_tracker):
scalar_keys = [
'emission_rate',
'txn_count',
'market_open',
'last_close',
'_dividend_count',
'period_start',
'event_count',
'day_count',
'capital_base',
'market_close',
'saved_dt',
'period_end',
'total_days',
]
p_string = pickle.dumps(perf_tracker)
test = pickle.loads(p_string)
for k in scalar_keys:
nt.assert_equal(getattr(test, k), getattr(perf_tracker, k), k)
for period in test.perf_periods:
nt.assert_true(hasattr(period, '_position_tracker'))
class TestSplitPerformance(unittest.TestCase):
def setUp(self):
self.sim_params, self.dt, self.end_dt = \
@@ -879,12 +913,16 @@ class TestPositionPerformance(unittest.TestCase):
txn1 = create_txn(trades_1[1], 10.0, 100)
txn2 = create_txn(trades_2[1], 10.0, -100)
pt = perf.PositionTracker()
pp = perf.PerformancePeriod(1000.0)
pp.execute_transaction(txn1)
pp.execute_transaction(txn2)
pp.position_tracker = pt
pt.execute_transaction(txn1)
pp.handle_execution(txn1)
pt.execute_transaction(txn2)
pp.handle_execution(txn2)
for trade in itertools.chain(trades_1[:-2], trades_2[:-2]):
pp.update_last_sale(trade)
pt.update_last_sale(trade)
pp.calculate_performance()
@@ -911,9 +949,9 @@ class TestPositionPerformance(unittest.TestCase):
net_liquidation=1000.0)
# now simulate stock1 going to $9
pp.update_last_sale(trades_1[-1])
pt.update_last_sale(trades_1[-1])
# and stock2 going to $11
pp.update_last_sale(trades_2[-1])
pt.update_last_sale(trades_2[-1])
pp.calculate_performance()
@@ -956,12 +994,15 @@ class TestPositionPerformance(unittest.TestCase):
)
txn = create_txn(trades[1], 10.0, 1000)
pt = perf.PositionTracker()
pp = perf.PerformancePeriod(1000.0)
pp.position_tracker = pt
pp.execute_transaction(txn)
pt.execute_transaction(txn)
pp.handle_execution(txn)
for trade in trades[:-2]:
pp.update_last_sale(trade)
pt.update_last_sale(trade)
pp.calculate_performance()
@@ -989,7 +1030,7 @@ class TestPositionPerformance(unittest.TestCase):
net_liquidation=1000.0)
# now simulate a price jump to $11
pp.update_last_sale(trades[-1])
pt.update_last_sale(trades[-1])
pp.calculate_performance()
@@ -1032,9 +1073,12 @@ class TestPositionPerformance(unittest.TestCase):
)
txn = create_txn(trades[1], 10.0, 100)
pt = perf.PositionTracker()
pp = perf.PerformancePeriod(1000.0)
pp.position_tracker = pt
pp.execute_transaction(txn)
pt.execute_transaction(txn)
pp.handle_execution(txn)
# This verifies that the last sale price is being correctly
# set in the positions. If this is not the case then returns can
@@ -1044,7 +1088,7 @@ class TestPositionPerformance(unittest.TestCase):
self.assertEqual(pp.positions[1].last_sale_price, 10.0)
for trade in trades:
pp.update_last_sale(trade)
pt.update_last_sale(trade)
pp.calculate_performance()
@@ -1134,11 +1178,14 @@ single short-sale transaction"""
trades_1 = trades[:-2]
txn = create_txn(trades[1], 10.0, -100)
pt = perf.PositionTracker()
pp = perf.PerformancePeriod(1000.0)
pp.position_tracker = pt
pp.execute_transaction(txn)
pt.execute_transaction(txn)
pp.handle_execution(txn)
for trade in trades_1:
pp.update_last_sale(trade)
pt.update_last_sale(trade)
pp.calculate_performance()
@@ -1195,7 +1242,7 @@ single short-sale transaction"""
pp.rollover()
for trade in trades_2:
pp.update_last_sale(trade)
pt.update_last_sale(trade)
pp.calculate_performance()
@@ -1249,15 +1296,18 @@ single short-sale transaction"""
)
# now run a performance period encompassing the entire trade sample.
ptTotal = perf.PositionTracker()
ppTotal = perf.PerformancePeriod(1000.0)
ppTotal.position_tracker = pt
for trade in trades_1:
ppTotal.update_last_sale(trade)
ptTotal.update_last_sale(trade)
ppTotal.execute_transaction(txn)
ptTotal.execute_transaction(txn)
ppTotal.handle_execution(txn)
for trade in trades_2:
ppTotal.update_last_sale(trade)
ptTotal.update_last_sale(trade)
ppTotal.calculate_performance()
@@ -1351,13 +1401,17 @@ trade after cover"""
)
cover_txn = create_txn(trades[6], 7.0, 100)
pt = perf.PositionTracker()
pp = perf.PerformancePeriod(1000.0)
pp.position_tracker = pt
pp.execute_transaction(short_txn)
pp.execute_transaction(cover_txn)
pt.execute_transaction(short_txn)
pp.handle_execution(short_txn)
pt.execute_transaction(cover_txn)
pp.handle_execution(cover_txn)
for trade in trades:
pp.update_last_sale(trade)
pt.update_last_sale(trade)
pp.calculate_performance()
@@ -1446,16 +1500,19 @@ shares in position"
trades = factory.create_trade_history(*history_args)
transactions = factory.create_txn_history(*history_args)
pt = perf.PositionTracker()
pp = perf.PerformancePeriod(1000.0)
pp.position_tracker = pt
average_cost = 0
for i, txn in enumerate(transactions):
pp.execute_transaction(txn)
pt.execute_transaction(txn)
pp.handle_execution(txn)
average_cost = (average_cost * i + txn.price) / (i + 1)
self.assertEqual(pp.positions[1].cost_basis, average_cost)
for trade in trades:
pp.update_last_sale(trade)
pt.update_last_sale(trade)
pp.calculate_performance()
@@ -1490,8 +1547,9 @@ shares in position"
pp.rollover()
pp.execute_transaction(sale_txn)
pp.update_last_sale(down_tick)
pt.execute_transaction(sale_txn)
pp.handle_execution(sale_txn)
pt.update_last_sale(down_tick)
pp.calculate_performance()
self.assertEqual(
@@ -1509,19 +1567,23 @@ shares in position"
self.assertEqual(pp.pnl, -800, "this period goes from +400 to -400")
pt3 = perf.PositionTracker()
pp3 = perf.PerformancePeriod(1000.0)
pp3.position_tracker = pt3
average_cost = 0
for i, txn in enumerate(transactions):
pp3.execute_transaction(txn)
pt3.execute_transaction(txn)
pp3.handle_execution(txn)
average_cost = (average_cost * i + txn.price) / (i + 1)
self.assertEqual(pp3.positions[1].cost_basis, average_cost)
pp3.execute_transaction(sale_txn)
pt3.execute_transaction(sale_txn)
pp3.handle_execution(sale_txn)
trades.append(down_tick)
for trade in trades:
pp3.update_last_sale(trade)
pt3.update_last_sale(trade)
pp3.calculate_performance()
self.assertEqual(
@@ -1555,14 +1617,17 @@ shares in position"
trades = factory.create_trade_history(*history_args)
transactions = factory.create_txn_history(*history_args)
pt = perf.PositionTracker()
pp = perf.PerformancePeriod(1000.0)
pp.position_tracker = pt
for txn, cb in zip(transactions, cost_bases):
pp.execute_transaction(txn)
pt.execute_transaction(txn)
pp.handle_execution(txn)
self.assertEqual(pp.positions[1].cost_basis, cb)
for trade in trades:
pp.update_last_sale(trade)
pt.update_last_sale(trade)
pp.calculate_performance()
@@ -1713,12 +1778,18 @@ class TestPerformanceTracker(unittest.TestCase):
self.assertEqual(perf_tracker.txn_count, len(txns))
self.assertEqual(perf_tracker.txn_count, len(orders))
cumulative_pos = perf_tracker.cumulative_performance.positions[sid]
expected_size = len(txns) / 2 * -25
self.assertEqual(cumulative_pos.amount, expected_size)
positions = perf_tracker.cumulative_performance.positions
if len(txns) == 0:
self.assertNotIn(sid, positions)
else:
expected_size = len(txns) / 2 * -25
cumulative_pos = positions[sid]
self.assertEqual(cumulative_pos.amount, expected_size)
self.assertEqual(len(perf_messages),
sim_params.days_in_period)
self.assertEqual(len(perf_messages),
sim_params.days_in_period)
check_perf_tracker_serialization(perf_tracker)
def trades_with_txns(self, events, no_txn_dt):
for event in events:
@@ -1844,3 +1915,113 @@ class TestPerformanceTracker(unittest.TestCase):
# In the second bar we can start establishing a sharpe ratio.
self.assertIsNone(msg_1['cumulative_risk_metrics']['sharpe'])
self.assertIsNotNone(msg_2['cumulative_risk_metrics']['sharpe'])
check_perf_tracker_serialization(tracker)
def test_serialization(self):
start_dt = datetime(year=2008,
month=10,
day=9,
tzinfo=pytz.utc)
end_dt = datetime(year=2008,
month=10,
day=16,
tzinfo=pytz.utc)
sim_params = SimulationParameters(
period_start=start_dt,
period_end=end_dt
)
perf_tracker = perf.PerformanceTracker(
sim_params
)
check_perf_tracker_serialization(perf_tracker)
class TestPosition(unittest.TestCase):
def setUp(self):
pass
def test_serialization(self):
dt = pd.Timestamp("1984/03/06 3:00PM")
pos = perf.Position(10, amount=np.float64(120.0), last_sale_date=dt,
last_sale_price=3.4)
p_string = pickle.dumps(pos)
test = pickle.loads(p_string)
nt.assert_dict_equal(test.__dict__, pos.__dict__)
class TestPositionTracker(unittest.TestCase):
def setUp(self):
pass
def test_empty_positions(self):
"""
make sure all the empty position stats return a numeric 0
Originally this bug was due to np.dot([], []) returning
np.bool_(False)
"""
pt = perf.PositionTracker()
stats = [
'calculate_positions_value',
'_net_exposure',
'_gross_exposure',
'_short_exposure',
'_shorts_count',
'_long_exposure',
'_longs_count',
]
for name in stats:
meth = getattr(pt, name)
val = meth()
self.assertEquals(val, 0)
self.assertNotIsInstance(val, (bool, np.bool_))
def test_serializaition(self):
pt = perf.PositionTracker()
dt = pd.Timestamp("1984/03/06 3:00PM")
pos1 = perf.Position('AAPL', amount=np.float64(120.0),
last_sale_date=dt, last_sale_price=3.4)
pos2 = perf.Position('IBM', amount=np.float64(100.0),
last_sale_date=dt, last_sale_price=3.4)
pt.update_positions({'AAPL': pos1, 'IBM': pos2})
p_string = pickle.dumps(pt)
test = pickle.loads(p_string)
nt.assert_dict_equal(test._position_amounts, pt._position_amounts)
nt.assert_dict_equal(test._position_last_sale_prices,
pt._position_last_sale_prices)
nt.assert_count_equal(test.positions.keys(), pt.positions.keys())
for sid in pt.positions:
nt.assert_dict_equal(test.positions[sid].__dict__,
pt.positions[sid].__dict__)
class TestPerformancePeriod(unittest.TestCase):
def setUp(self):
pass
def test_serialization(self):
pt = perf.PositionTracker()
pp = perf.PerformancePeriod(100)
pp.position_tracker = pt
p_string = pickle.dumps(pp)
test = pickle.loads(p_string)
correct = pp.__dict__.copy()
del correct['_position_tracker']
nt.assert_count_equal(test.__dict__.keys(), correct.keys())
equal_keys = list(correct.keys())
equal_keys.remove('_account_store')
equal_keys.remove('_portfolio_store')
for k in equal_keys:
nt.assert_equal(test.__dict__[k], correct[k])
tests/test_pickle_serialization.py
+12 -61
View File
@@ -13,88 +13,39 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import datetime
import pickle
import pytz
from nose_parameterized import parameterized
from unittest import TestCase
from zipline.finance.blotter import Blotter, Order
from zipline.finance.commission import PerShare, PerTrade, PerDollar
from zipline.finance.performance.period import PerformancePeriod
from zipline.finance.performance.position import Position
from zipline.finance.performance.tracker import PerformanceTracker
from zipline.finance.risk.cumulative import RiskMetricsCumulative
from zipline.finance.risk.period import RiskMetricsPeriod
from zipline.finance.risk.report import RiskReport
from zipline.finance.slippage import (
FixedSlippage,
Transaction,
VolumeShareSlippage
from .serialization_cases import (
object_serialization_cases,
assert_dict_equal
)
from zipline.protocol import Account
from zipline.protocol import Portfolio
from zipline.protocol import Position as ProtocolPosition
from zipline.finance.trading import SimulationParameters
from zipline.utils import factory
sim_params_daily = SimulationParameters(
datetime.datetime(2013, 6, 19, tzinfo=pytz.UTC),
datetime.datetime(2013, 6, 19, tzinfo=pytz.UTC),
10000,
emission_rate='daily')
sim_params_minute = SimulationParameters(
datetime.datetime(2013, 6, 19, tzinfo=pytz.UTC),
datetime.datetime(2013, 6, 19, tzinfo=pytz.UTC),
10000,
emission_rate='minute')
returns = factory.create_returns_from_list(
[1.0], sim_params_daily)
class PickleSerializationTestCase(TestCase):
@parameterized.expand([
(Blotter, (), 'repr'),
(Order, (datetime.datetime(2013, 6, 19), 8554, 100), 'dict'),
(PerShare, (), 'dict'),
(PerTrade, (), 'dict'),
(PerDollar, (), 'dict'),
(PerformancePeriod, (10000,), 'to_dict'),
(Position, (8554,), 'dict'),
(PerformanceTracker, (sim_params_daily,), 'to_dict'),
(PerformanceTracker, (sim_params_minute,), 'to_dict'),
(RiskMetricsCumulative, (sim_params_daily,), 'to_dict'),
(RiskMetricsCumulative, (sim_params_minute,), 'to_dict'),
(RiskMetricsPeriod,
(returns.index[0], returns.index[0], returns), 'to_dict'),
(RiskReport, (returns, sim_params_daily), 'to_dict'),
(RiskReport, (returns, sim_params_minute), 'to_dict'),
(FixedSlippage, (), 'dict'),
(Transaction,
(8554, 10, datetime.datetime(2013, 6, 19), 100, "0000"), 'dict'),
(VolumeShareSlippage, (), 'dict'),
(Account, (), 'dict'),
(Portfolio, (), 'dict'),
(ProtocolPosition, (8554,), 'dict')
])
@parameterized.expand(object_serialization_cases())
def test_object_serialization(self,
_,
cls,
initargs,
di_vars,
comparison_method='dict'):
obj = cls(*initargs)
for k, v in di_vars.items():
setattr(obj, k, v)
state = pickle.dumps(obj)
obj2 = pickle.loads(state)
for k, v in di_vars.items():
setattr(obj2, k, v)
if comparison_method == 'repr':
self.assertEqual(obj.__repr__(), obj2.__repr__())
elif comparison_method == 'to_dict':
self.assertEqual(obj.to_dict(), obj2.to_dict())
assert_dict_equal(obj.to_dict(), obj2.to_dict())
else:
self.assertEqual(obj.__dict__, obj2.__dict__)
assert_dict_equal(obj.__dict__, obj2.__dict__)
tests/test_serialization.py
+13 -77
View File
@@ -13,49 +13,16 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import datetime
import pytz
from nose_parameterized import parameterized
from unittest import TestCase
from zipline.finance.blotter import Blotter, Order
from zipline.finance.commission import PerShare, PerTrade, PerDollar
from zipline.finance.performance.period import PerformancePeriod
from zipline.finance.performance.position import Position
from zipline.finance.performance.tracker import PerformanceTracker
from zipline.finance.risk.cumulative import RiskMetricsCumulative
from zipline.finance.risk.period import RiskMetricsPeriod
from zipline.finance.risk.report import RiskReport
from zipline.finance.slippage import (
FixedSlippage,
Transaction,
VolumeShareSlippage
from .serialization_cases import (
object_serialization_cases,
assert_dict_equal
)
from zipline.protocol import Account
from zipline.protocol import Portfolio
from zipline.protocol import Position as ProtocolPosition
from zipline.finance.trading import SimulationParameters
from zipline.utils import factory
from six import iteritems
sim_params_daily = SimulationParameters(
datetime.datetime(2013, 6, 19, tzinfo=pytz.UTC),
datetime.datetime(2013, 6, 19, tzinfo=pytz.UTC),
10000,
emission_rate='daily')
sim_params_minute = SimulationParameters(
datetime.datetime(2013, 6, 19, tzinfo=pytz.UTC),
datetime.datetime(2013, 6, 19, tzinfo=pytz.UTC),
10000,
emission_rate='minute')
returns = factory.create_returns_from_list(
[1.0], sim_params_daily)
def gather_bad_dicts(state):
bad = []
@@ -68,54 +35,20 @@ def gather_bad_dicts(state):
return bad
def stringify_cases(cases, func=None):
# get better test case names
results = []
if func is None:
func = lambda case: case[0].__name__
for case in cases:
new_case = list(case)
key = func(case)
new_case.insert(0, key)
results.append(new_case)
return results
class SerializationTestCase(TestCase):
object_serialization_cases = stringify_cases([
(Blotter, (), 'repr'),
(Order, (datetime.datetime(2013, 6, 19), 8554, 100), 'dict'),
(PerShare, (), 'dict'),
(PerTrade, (), 'dict'),
(PerDollar, (), 'dict'),
(PerformancePeriod, (10000,), 'to_dict'),
(Position, (8554,), 'dict'),
(PerformanceTracker, (sim_params_daily,), 'to_dict'),
(PerformanceTracker, (sim_params_minute,), 'to_dict'),
(RiskMetricsCumulative, (sim_params_daily,), 'to_dict'),
(RiskMetricsCumulative, (sim_params_minute,), 'to_dict'),
(RiskMetricsPeriod,
(returns.index[0], returns.index[0], returns), 'to_dict'),
(RiskReport, (returns, sim_params_daily), 'to_dict'),
(RiskReport, (returns, sim_params_minute), 'to_dict'),
(FixedSlippage, (), 'dict'),
(Transaction,
(8554, 10, datetime.datetime(2013, 6, 19), 100, "0000"), 'dict'),
(VolumeShareSlippage, (), 'dict'),
(Account, (), 'dict'),
(Portfolio, (), 'dict'),
(ProtocolPosition, (8554,), 'dict')
])
@parameterized.expand(object_serialization_cases)
@parameterized.expand(object_serialization_cases())
def test_object_serialization(self,
_,
cls,
initargs,
di_vars,
comparison_method='dict'):
obj = cls(*initargs)
for k, v in di_vars.items():
setattr(obj, k, v)
state = obj.__getstate__()
bad_dicts = gather_bad_dicts(state)
@@ -142,9 +75,12 @@ class SerializationTestCase(TestCase):
obj2.__init__(*initargs)
obj2.__setstate__(state)
for k, v in di_vars.items():
setattr(obj2, k, v)
if comparison_method == 'repr':
self.assertEqual(obj.__repr__(), obj2.__repr__())
elif comparison_method == 'to_dict':
self.assertEqual(obj.to_dict(), obj2.to_dict())
assert_dict_equal(obj.to_dict(), obj2.to_dict())
else:
self.assertEqual(obj.__dict__, obj2.__dict__)
assert_dict_equal(obj.__dict__, obj2.__dict__)
tests/test_versioning.py
+14 -58
View File
@@ -13,49 +13,19 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import datetime
import os
import pandas
import pickle
import pytz
from nose_parameterized import parameterized
from unittest import TestCase
from zipline.finance.blotter import Blotter, Order
from zipline.finance.commission import PerShare, PerTrade, PerDollar
from zipline.finance.performance.period import PerformancePeriod
from zipline.finance.performance.position import Position
from zipline.finance.performance.tracker import PerformanceTracker
from zipline.finance.risk.cumulative import RiskMetricsCumulative
from zipline.finance.risk.period import RiskMetricsPeriod
from zipline.finance.risk.report import RiskReport
from zipline.finance.slippage import (
FixedSlippage,
Transaction,
VolumeShareSlippage
from zipline.finance.blotter import Order
from .serialization_cases import (
object_serialization_cases,
assert_dict_equal
)
from zipline.protocol import Account
from zipline.protocol import Portfolio
from zipline.protocol import Position as ProtocolPosition
from zipline.finance.trading import SimulationParameters
from zipline.utils import factory
sim_params_daily = SimulationParameters(
datetime.datetime(2013, 6, 19, tzinfo=pytz.UTC),
datetime.datetime(2013, 6, 19, tzinfo=pytz.UTC),
10000,
emission_rate='daily')
sim_params_minute = SimulationParameters(
datetime.datetime(2013, 6, 19, tzinfo=pytz.UTC),
datetime.datetime(2013, 6, 19, tzinfo=pytz.UTC),
10000,
emission_rate='minute')
returns = factory.create_returns_from_list(
[1.0], sim_params_daily)
base_state_dir = 'tests/resources/saved_state_archive'
@@ -80,30 +50,12 @@ class VersioningTestCase(TestCase):
yield pickle.load(f)
# Only test versioning in minutely mode right now
@parameterized.expand([
(Blotter, (), 'repr'),
(Order, (datetime.datetime(2013, 6, 19), 8554, 100), 'dict'),
(PerShare, (), 'dict'),
(PerTrade, (), 'dict'),
(PerDollar, (), 'dict'),
(PerformancePeriod, (10000,), 'to_dict'),
(Position, (8554,), 'dict'),
(PerformanceTracker, (sim_params_minute,), 'to_dict'),
(RiskMetricsCumulative, (sim_params_minute,), 'to_dict'),
(RiskMetricsPeriod,
(returns.index[0], returns.index[0], returns), 'to_dict'),
(RiskReport, (returns, sim_params_minute), 'to_dict'),
(FixedSlippage, (), 'dict'),
(Transaction,
(8554, 10, datetime.datetime(2013, 6, 19), 100, "0000"), 'dict'),
(VolumeShareSlippage, (), 'dict'),
(Account, (), 'dict'),
(Portfolio, (), 'dict'),
(ProtocolPosition, (8554,), 'dict')
])
@parameterized.expand(object_serialization_cases(skip_daily=True))
def test_object_serialization(self,
_,
cls,
initargs,
di_vars,
comparison_method='dict'):
# The state generated under one version of pandas may not be
@@ -116,6 +68,8 @@ class VersioningTestCase(TestCase):
# Make reference object
obj = cls(*initargs)
for k, v in di_vars.items():
setattr(obj, k, v)
# Fetch state
state_versions = self.load_state_from_disk(cls)
@@ -136,6 +90,8 @@ class VersioningTestCase(TestCase):
if initargs is not None:
obj2.__init__(*initargs)
obj2.__setstate__(state)
for k, v in di_vars.items():
setattr(obj2, k, v)
# The ObjectId generated on instantiation of Order will
# not be the same as the one loaded from saved state.
@@ -145,6 +101,6 @@ class VersioningTestCase(TestCase):
if comparison_method == 'repr':
self.assertEqual(obj.__repr__(), obj2.__repr__())
elif comparison_method == 'to_dict':
self.assertEqual(obj.to_dict(), obj2.to_dict())
assert_dict_equal(obj.to_dict(), obj2.to_dict())
else:
self.assertEqual(obj.__dict__, obj2.__dict__)
assert_dict_equal(obj.__dict__, obj2.__dict__)
zipline/finance/performance/__init__.py
+2
View File
@@ -16,9 +16,11 @@
from . tracker import PerformanceTracker
from . period import PerformancePeriod
from . position import Position
from . position_tracker import PositionTracker
__all__ = [
'PerformanceTracker',
'PerformancePeriod',
'Position',
'PositionTracker',
]
zipline/finance/performance/period.py
+79 -242
View File
@@ -72,11 +72,8 @@ omitted).
from __future__ import division
import logbook
from operator import mul
import numpy as np
import pandas as pd
from pandas.lib import checknull
from collections import (
defaultdict,
)
@@ -87,19 +84,38 @@ try:
except ImportError:
from collections import OrderedDict
from six import iteritems, itervalues
from six import itervalues, iteritems
import zipline.protocol as zp
from . position import positiondict
from zipline.utils.serialization_utils import (
VERSION_LABEL
)
from .position_tracker import PositionTracker
log = logbook.Logger('Performance')
TRADE_TYPE = zp.DATASOURCE_TYPE.TRADE
def position_proxy(func):
def _proxied(self, *args, **kwargs):
meth_name = func.__name__
meth = getattr(self.position_tracker, meth_name)
return meth(*args, **kwargs)
return _proxied
class ProxyError(Exception):
def __init__(self):
import inspect
meth_name = inspect.stack()[1][3]
TEMPLATE = "{meth_name} should have been proxied to position_tracker."
msg = TEMPLATE.format(meth_name=meth_name)
super(ProxyError, self).__init__(msg)
class PerformancePeriod(object):
def __init__(
@@ -118,44 +134,32 @@ class PerformancePeriod(object):
self.period_cash_flow = 0.0
self.pnl = 0.0
# sid => position object
self.positions = positiondict()
self.ending_cash = starting_cash
# rollover initializes a number of self's attributes:
self.rollover()
self.keep_transactions = keep_transactions
self.keep_orders = keep_orders
self.initialize_position_calc_arrays()
self.calculate_performance()
# An object to recycle via assigning new values
# when returning portfolio information.
# So as not to avoid creating a new object for each event
self._portfolio_store = zp.Portfolio()
self._account_store = zp.Account()
self._positions_store = zp.Positions()
self.serialize_positions = serialize_positions
self._unpaid_dividends = pd.DataFrame(
columns=zp.DIVIDEND_PAYMENT_FIELDS,
)
_position_tracker = None
self.loc_map = {}
@property
def position_tracker(self):
return self._position_tracker
def initialize_position_calc_arrays(self):
# Arrays for quick calculations of positions value.
self._position_amounts = OrderedDict()
self._position_last_sale_prices = OrderedDict()
def set_positions(self, positions):
self.positions = positions
for sid, pos in positions.iteritems():
self._position_amounts[sid] = pos.amount
self._position_last_sale_prices[sid] = pos.last_sale_price
# Invalidate cache.
self._position_values = None # invalidate cache
@position_tracker.setter
def position_tracker(self, obj):
if obj is None:
raise ValueError("position_tracker can not be None")
self._position_tracker = obj
# we only calculate perf once we inject PositionTracker
self.calculate_performance()
def rollover(self):
self.starting_value = self.ending_value
@@ -166,113 +170,17 @@ class PerformancePeriod(object):
self.orders_by_modified = defaultdict(OrderedDict)
self.orders_by_id = OrderedDict()
def set_position_amount(self, sid, amount):
self._position_amounts[sid] = amount
self._position_values = None # invalidate cache
def set_position_last_sale_price(self, sid, last_sale_price):
self._position_last_sale_prices[sid] = last_sale_price
self._position_values = None # invalidate cache
def handle_split(self, split):
if split.sid in self.positions:
# Make the position object handle the split. It returns the
# leftover cash from a fractional share, if there is any.
position = self.positions[split.sid]
leftover_cash = position.handle_split(split)
self._position_amounts[split.sid] = position.amount
self._position_last_sale_prices[split.sid] = \
position.last_sale_price
self._position_values = None # invalidate cache
if leftover_cash > 0:
self.handle_cash_payment(leftover_cash)
def earn_dividends(self, dividend_frame):
"""
Given a frame of dividends whose ex_dates are all the next trading day,
calculate and store the cash and/or stock payments to be paid on each
dividend's pay date.
"""
earned = dividend_frame.apply(self._maybe_earn_dividend, axis=1)\
.dropna(how='all')
if len(earned) > 0:
# Store the earned dividends so that they can be paid on the
# dividends' pay_dates.
self._unpaid_dividends = pd.concat(
[self._unpaid_dividends, earned],
)
def _maybe_earn_dividend(self, dividend):
"""
Take a historical dividend record and return a Series with fields in
zipline.protocol.DIVIDEND_FIELDS (plus an 'id' field) representing
the cash/stock amount we are owed when the dividend is paid.
"""
if dividend['sid'] in self.positions:
return self.positions[dividend['sid']].earn_dividend(dividend)
else:
return zp.dividend_payment()
def pay_dividends(self, dividend_frame):
"""
Given a frame of dividends whose pay_dates are all the next trading
day, grant the cash and/or stock payments that were calculated on the
given dividends' ex dates.
"""
payments = dividend_frame.apply(self._maybe_pay_dividend, axis=1)\
.dropna(how='all')
# Mark these dividends as paid by dropping them from our unpaid
# table.
self._unpaid_dividends.drop(payments.index)
# Add cash equal to the net cash payed from all dividends. Note that
# "negative cash" is effectively paid if we're short a security,
# representing the fact that we're required to reimburse the owner of
# the stock for any dividends paid while borrowing.
net_cash_payment = payments['cash_amount'].fillna(0).sum()
def handle_dividends_paid(self, net_cash_payment):
if net_cash_payment:
self.handle_cash_payment(net_cash_payment)
# Add stock for any stock dividends paid. Again, the values here may
# be negative in the case of short positions.
stock_payments = payments[payments['payment_sid'].notnull()]
for _, row in stock_payments.iterrows():
stock = row['payment_sid']
share_count = row['share_count']
position = self.positions[stock]
position.amount += share_count
self._position_amounts[stock] = position.amount
self._position_last_sale_prices[stock] = position.last_sale_price
self._position_values = None # invalidate cache
# Recalculate performance after applying dividend benefits.
self.calculate_performance()
def _maybe_pay_dividend(self, dividend):
"""
Take a historical dividend record, look up any stored record of
cash/stock we are owed for that dividend, and return a Series
with fields drawn from zipline.protocol.DIVIDEND_PAYMENT_FIELDS.
"""
try:
unpaid_dividend = self._unpaid_dividends.loc[dividend['id']]
return unpaid_dividend
except KeyError:
return zp.dividend_payment()
def handle_cash_payment(self, payment_amount):
self.adjust_cash(payment_amount)
def handle_commission(self, commission):
# Deduct from our total cash pool.
self.adjust_cash(-commission.cost)
# Adjust the cost basis of the stock if we own it
if commission.sid in self.positions:
self.positions[commission.sid].\
adjust_commission_cost_basis(commission)
def adjust_cash(self, amount):
self.period_cash_flow += amount
@@ -306,79 +214,56 @@ class PerformancePeriod(object):
del self.orders_by_id[order.id]
self.orders_by_id[order.id] = order
def update_position(self, sid, amount=None, last_sale_price=None,
last_sale_date=None, cost_basis=None):
pos = self.positions[sid]
if amount is not None:
pos.amount = amount
self._position_amounts[sid] = amount
self._position_values = None # invalidate cache
if last_sale_price is not None:
pos.last_sale_price = last_sale_price
self._position_last_sale_prices[sid] = last_sale_price
self._position_values = None # invalidate cache
if last_sale_date is not None:
pos.last_sale_date = last_sale_date
if cost_basis is not None:
pos.cost_basis = cost_basis
def execute_transaction(self, txn):
# Update Position
# ----------------
# NOTE: self.positions has defaultdict semantics, so this will create
# an empty position if one does not already exist.
sid = txn.sid
position = self.positions[sid]
position.update(txn)
self._position_amounts[sid] = position.amount
self._position_last_sale_prices[sid] = position.last_sale_price
self._position_values = None # invalidate cache
def handle_execution(self, txn):
self.period_cash_flow -= txn.price * txn.amount
if self.keep_transactions:
self.processed_transactions[txn.dt].append(txn)
_position_values = None
# backwards compat. TODO: remove?
@property
def positions(self):
return self.position_tracker.positions
@property
def position_values(self):
"""
Invalidate any time self._position_amounts or
self._position_last_sale_prices is changed.
"""
if self._position_values is None:
vals = list(map(mul, self._position_amounts.values(),
self._position_last_sale_prices.values()))
self._position_values = vals
return self._position_values
def position_amounts(self):
return self.position_tracker.position_amounts
@property
def position_last_sale_prices(self):
return self.position_tracker.position_last_sale_prices
@position_proxy
def calculate_positions_value(self):
if len(self.position_values) == 0:
return np.float64(0)
raise ProxyError()
return sum(self.position_values)
@position_proxy
def set_positions(self):
raise ProxyError()
@position_proxy
def _longs_count(self):
return sum(map(lambda x: x > 0, self.position_values))
raise ProxyError()
@position_proxy
def _long_exposure(self):
return sum(filter(lambda x: x > 0, self.position_values))
raise ProxyError()
@position_proxy
def _shorts_count(self):
return sum(map(lambda x: x < 0, self.position_values))
raise ProxyError()
@position_proxy
def _short_exposure(self):
return sum(filter(lambda x: x < 0, self.position_values))
raise ProxyError()
@position_proxy
def _gross_exposure(self):
return self._long_exposure() + abs(self._short_exposure())
raise ProxyError()
@position_proxy
def _net_exposure(self):
return self.calculate_positions_value()
raise ProxyError()
@property
def _net_liquidation_value(self):
@@ -400,22 +285,6 @@ class PerformancePeriod(object):
return np.inf
def update_last_sale(self, event):
sid = event.sid
if sid not in self.positions:
return
if event.type != TRADE_TYPE:
return
price = event.price
if not checknull(price):
pos = self.positions[sid]
pos.last_sale_date = event.dt
pos.last_sale_price = price
self._position_last_sale_prices[sid] = price
self._position_values = None # invalidate cache
def __core_dict(self):
rval = {
'ending_value': self.ending_value,
@@ -547,63 +416,30 @@ class PerformancePeriod(object):
getattr(self, 'net_liquidation', self._net_liquidation_value)
return account
@position_proxy
def get_positions(self):
raise ProxyError()
positions = self._positions_store
for sid, pos in iteritems(self.positions):
if pos.amount == 0:
# Clear out the position if it has become empty since the last
# time get_positions was called. Catching the KeyError is
# faster than checking `if sid in positions`, and this can be
# potentially called in a tight inner loop.
try:
del positions[sid]
except KeyError:
pass
continue
# Note that this will create a position if we don't currently have
# an entry
position = positions[sid]
position.amount = pos.amount
position.cost_basis = pos.cost_basis
position.last_sale_price = pos.last_sale_price
return positions
@position_proxy
def get_positions_list(self):
positions = []
for sid, pos in iteritems(self.positions):
if pos.amount != 0:
positions.append(pos.to_dict())
return positions
raise ProxyError()
def __getstate__(self):
state_dict = \
{k: v for k, v in iteritems(self.__dict__)
if not k.startswith('_')}
state_dict = {k: v for k, v in iteritems(self.__dict__)
if not k.startswith('_')}
state_dict['_portfolio_store'] = self._portfolio_store
state_dict['_account_store'] = self._account_store
# We need to handle the defaultdict specially, otherwise
# msgpack will unpack it as a dict, causing KeyError
# nastiness.
state_dict['processed_transactions'] = \
dict(self.processed_transactions)
state_dict['orders_by_id'] = \
dict(self.orders_by_id)
state_dict['orders_by_modified'] = \
dict(self.orders_by_modified)
state_dict['positions'] = \
dict(self.positions)
state_dict['_positions_store'] = \
dict(self._positions_store)
STATE_VERSION = 1
STATE_VERSION = 2
state_dict[VERSION_LABEL] = STATE_VERSION
return state_dict
def __setstate__(self, state):
@@ -622,19 +458,20 @@ class PerformancePeriod(object):
orders_by_modified = defaultdict(OrderedDict)
orders_by_modified.update(state.pop('orders_by_modified'))
positions = positiondict()
positions.update(state.pop('positions'))
_positions_store = zp.Positions()
_positions_store.update(state.pop('_positions_store'))
self.processed_transactions = processed_transactions
self.orders_by_id = orders_by_id
self.orders_by_modified = orders_by_modified
self.positions = positions
self._positions_store = _positions_store
# pop positions to use for v1
positions = state.pop('positions', None)
self.__dict__.update(state)
self.initialize_position_calc_arrays()
if version == 1:
# version 1 had PositionTracker logic inside of Period
# we create the PositionTracker here.
# Note: that in V2 it is assumed that the position_tracker
# will be dependency injected and so is not reconstructed
assert positions is not None, "positions should exist in v1"
position_tracker = PositionTracker()
position_tracker.update_positions(positions)
self.position_tracker = position_tracker
zipline/finance/performance/position.py
-1
View File
@@ -214,7 +214,6 @@ last_sale_price: {last_sale_price}"
}
def __getstate__(self):
state_dict = copy(self.__dict__)
STATE_VERSION = 1
zipline/finance/performance/position_tracker.py
+280
View File
@@ -0,0 +1,280 @@
from __future__ import division
from operator import mul
import logbook
import numpy as np
import pandas as pd
from pandas.lib import checknull
try:
# optional cython based OrderedDict
from cyordereddict import OrderedDict
except ImportError:
from collections import OrderedDict
from six import iteritems
from six.moves import map, filter
from zipline.utils.serialization_utils import (
VERSION_LABEL
)
import zipline.protocol as zp
from . position import positiondict
log = logbook.Logger('Performance')
class PositionTracker(object):
def __init__(self):
# sid => position object
self.positions = positiondict()
# Arrays for quick calculations of positions value
self._position_amounts = OrderedDict()
self._position_last_sale_prices = OrderedDict()
self._unpaid_dividends = pd.DataFrame(
columns=zp.DIVIDEND_PAYMENT_FIELDS,
)
self._positions_store = zp.Positions()
def update_last_sale(self, event):
# NOTE, PerformanceTracker already vetted as TRADE type
sid = event.sid
if sid not in self.positions:
return
price = event.price
if not checknull(price):
pos = self.positions[sid]
pos.last_sale_date = event.dt
pos.last_sale_price = price
self._position_last_sale_prices[sid] = price
self._position_values = None # invalidate cache
sid = event.sid
price = event.price
def update_positions(self, positions):
# update positions in batch
self.positions.update(positions)
for sid, pos in iteritems(positions):
self._position_amounts[sid] = pos.amount
self._position_last_sale_prices[sid] = pos.last_sale_price
# Invalidate cache.
self._position_values = None # invalidate cache
def update_position(self, sid, amount=None, last_sale_price=None,
last_sale_date=None, cost_basis=None):
pos = self.positions[sid]
if amount is not None:
pos.amount = amount
self._position_amounts[sid] = amount
self._position_values = None # invalidate cache
if last_sale_price is not None:
pos.last_sale_price = last_sale_price
self._position_last_sale_prices[sid] = last_sale_price
self._position_values = None # invalidate cache
if last_sale_date is not None:
pos.last_sale_date = last_sale_date
if cost_basis is not None:
pos.cost_basis = cost_basis
def execute_transaction(self, txn):
# Update Position
# ----------------
sid = txn.sid
position = self.positions[sid]
position.update(txn)
self._position_amounts[sid] = position.amount
self._position_last_sale_prices[sid] = position.last_sale_price
self._position_values = None # invalidate cache
def handle_commission(self, commission):
# Adjust the cost basis of the stock if we own it
if commission.sid in self.positions:
self.positions[commission.sid].\
adjust_commission_cost_basis(commission)
_position_values = None
@property
def position_values(self):
"""
Invalidate any time self._position_amounts or
self._position_last_sale_prices is changed.
"""
if self._position_values is None:
vals = list(map(mul, self._position_amounts.values(),
self._position_last_sale_prices.values()))
self._position_values = vals
return self._position_values
def calculate_positions_value(self):
if len(self.position_values) == 0:
return np.float64(0)
return sum(self.position_values)
def _longs_count(self):
return sum(map(lambda x: x > 0, self.position_values))
def _long_exposure(self):
return sum(filter(lambda x: x > 0, self.position_values))
def _shorts_count(self):
return sum(map(lambda x: x < 0, self.position_values))
def _short_exposure(self):
return sum(filter(lambda x: x < 0, self.position_values))
def _gross_exposure(self):
return self._long_exposure() + abs(self._short_exposure())
def _net_exposure(self):
return self.calculate_positions_value()
def handle_split(self, split):
if split.sid in self.positions:
# Make the position object handle the split. It returns the
# leftover cash from a fractional share, if there is any.
position = self.positions[split.sid]
leftover_cash = position.handle_split(split)
self._position_amounts[split.sid] = position.amount
self._position_last_sale_prices[split.sid] = \
position.last_sale_price
self._position_values = None # invalidate cache
return leftover_cash
def _maybe_earn_dividend(self, dividend):
"""
Take a historical dividend record and return a Series with fields in
zipline.protocol.DIVIDEND_FIELDS (plus an 'id' field) representing
the cash/stock amount we are owed when the dividend is paid.
"""
if dividend['sid'] in self.positions:
return self.positions[dividend['sid']].earn_dividend(dividend)
else:
return zp.dividend_payment()
def earn_dividends(self, dividend_frame):
"""
Given a frame of dividends whose ex_dates are all the next trading day,
calculate and store the cash and/or stock payments to be paid on each
dividend's pay date.
"""
earned = dividend_frame.apply(self._maybe_earn_dividend, axis=1)\
.dropna(how='all')
if len(earned) > 0:
# Store the earned dividends so that they can be paid on the
# dividends' pay_dates.
self._unpaid_dividends = pd.concat(
[self._unpaid_dividends, earned],
)
def _maybe_pay_dividend(self, dividend):
"""
Take a historical dividend record, look up any stored record of
cash/stock we are owed for that dividend, and return a Series
with fields drawn from zipline.protocol.DIVIDEND_PAYMENT_FIELDS.
"""
try:
unpaid_dividend = self._unpaid_dividends.loc[dividend['id']]
return unpaid_dividend
except KeyError:
return zp.dividend_payment()
def pay_dividends(self, dividend_frame):
"""
Given a frame of dividends whose pay_dates are all the next trading
day, grant the cash and/or stock payments that were calculated on the
given dividends' ex dates.
"""
payments = dividend_frame.apply(self._maybe_pay_dividend, axis=1)\
.dropna(how='all')
# Mark these dividends as paid by dropping them from our unpaid
# table.
self._unpaid_dividends.drop(payments.index)
# Add stock for any stock dividends paid. Again, the values here may
# be negative in the case of short positions.
stock_payments = payments[payments['payment_sid'].notnull()]
for _, row in stock_payments.iterrows():
stock = row['payment_sid']
share_count = row['share_count']
# note we create a Position for stock dividend if we don't
# already own the security
position = self.positions[stock]
position.amount += share_count
self._position_amounts[stock] = position.amount
self._position_last_sale_prices[stock] = position.last_sale_price
# Add cash equal to the net cash payed from all dividends. Note that
# "negative cash" is effectively paid if we're short a security,
# representing the fact that we're required to reimburse the owner of
# the stock for any dividends paid while borrowing.
net_cash_payment = payments['cash_amount'].fillna(0).sum()
return net_cash_payment
def get_positions(self):
positions = self._positions_store
for sid, pos in iteritems(self.positions):
if pos.amount == 0:
# Clear out the position if it has become empty since the last
# time get_positions was called. Catching the KeyError is
# faster than checking `if sid in positions`, and this can be
# potentially called in a tight inner loop.
try:
del positions[sid]
except KeyError:
pass
continue
# Note that this will create a position if we don't currently have
# an entry
position = positions[sid]
position.amount = pos.amount
position.cost_basis = pos.cost_basis
position.last_sale_price = pos.last_sale_price
return positions
def get_positions_list(self):
positions = []
for sid, pos in iteritems(self.positions):
if pos.amount != 0:
positions.append(pos.to_dict())
return positions
def __getstate__(self):
state_dict = {}
state_dict['positions'] = dict(self.positions)
state_dict['unpaid_dividends'] = self._unpaid_dividends
STATE_VERSION = 1
state_dict[VERSION_LABEL] = STATE_VERSION
return state_dict
def __setstate__(self, state):
OLDEST_SUPPORTED_STATE = 1
version = state.pop(VERSION_LABEL)
if version < OLDEST_SUPPORTED_STATE:
raise BaseException("PositionTracker saved state is too old.")
self.positions = positiondict()
# note that positions_store is temporary and gets regened from
# .positions
self._positions_store = zp.Positions()
self._unpaid_dividends = state['unpaid_dividends']
# Arrays for quick calculations of positions value
self._position_amounts = OrderedDict()
self._position_last_sale_prices = OrderedDict()
self.update_positions(state['positions'])
zipline/finance/performance/tracker.py
+52 -31
View File
@@ -75,6 +75,7 @@ from zipline.finance.trading import with_environment
from zipline.utils.serialization_utils import (
VERSION_LABEL
)
from . position_tracker import PositionTracker
log = logbook.Logger('Performance')
@@ -108,6 +109,8 @@ class PerformanceTracker(object):
self.dividend_frame = pd.DataFrame()
self._dividend_count = 0
self.position_tracker = PositionTracker()
self.perf_periods = []
if self.emission_rate == 'daily':
@@ -118,12 +121,9 @@ class PerformanceTracker(object):
risk.RiskMetricsCumulative(self.sim_params)
elif self.emission_rate == 'minute':
self.all_benchmark_returns = pd.Series(
index=env.minutes_for_days_in_range(
self.sim_params.first_open,
self.sim_params.last_close
)
)
self.all_benchmark_returns = pd.Series(index=pd.date_range(
self.sim_params.first_open, self.sim_params.last_close,
freq='Min'))
self.intraday_risk_metrics = \
risk.RiskMetricsCumulative(self.sim_params)
@@ -146,6 +146,7 @@ class PerformanceTracker(object):
# don't serialize positions for cumualtive period
serialize_positions=False
)
self.minute_performance.position_tracker = self.position_tracker
self.perf_periods.append(self.minute_performance)
# this performance period will span the entire simulation from
@@ -161,8 +162,9 @@ class PerformanceTracker(object):
keep_transactions=False,
keep_orders=False,
# don't serialize positions for cumualtive period
serialize_positions=False
serialize_positions=False,
)
self.cumulative_performance.position_tracker = self.position_tracker
self.perf_periods.append(self.cumulative_performance)
# this performance period will span just the current market day
@@ -174,8 +176,10 @@ class PerformanceTracker(object):
self.market_close,
keep_transactions=True,
keep_orders=True,
serialize_positions=True
serialize_positions=True,
)
self.todays_performance.position_tracker = self.position_tracker
self.perf_periods.append(self.todays_performance)
self.saved_dt = self.period_start
@@ -280,29 +284,32 @@ class PerformanceTracker(object):
if event.type == zp.DATASOURCE_TYPE.TRADE:
# update last sale
for perf_period in self.perf_periods:
perf_period.update_last_sale(event)
self.position_tracker.update_last_sale(event)
elif event.type == zp.DATASOURCE_TYPE.TRANSACTION:
# Trade simulation always follows a transaction with the
# TRADE event that was used to simulate it, so we don't
# check for end of day rollover messages here.
self.txn_count += 1
self.position_tracker.execute_transaction(event)
for perf_period in self.perf_periods:
perf_period.execute_transaction(event)
perf_period.handle_execution(event)
elif event.type == zp.DATASOURCE_TYPE.DIVIDEND:
log.info("Ignoring DIVIDEND event.")
elif event.type == zp.DATASOURCE_TYPE.SPLIT:
for perf_period in self.perf_periods:
perf_period.handle_split(event)
leftover_cash = self.position_tracker.handle_split(event)
if leftover_cash > 0:
for perf_period in self.perf_periods:
perf_period.handle_cash_payment(leftover_cash)
elif event.type == zp.DATASOURCE_TYPE.ORDER:
for perf_period in self.perf_periods:
perf_period.record_order(event)
elif event.type == zp.DATASOURCE_TYPE.COMMISSION:
self.position_tracker.handle_commission(event)
for perf_period in self.perf_periods:
perf_period.handle_commission(event)
@@ -371,17 +378,18 @@ class PerformanceTracker(object):
pay_date_mask = (self.dividend_frame['pay_date'] == next_trading_day)
dividends_payable = self.dividend_frame[pay_date_mask]
for period in self.perf_periods:
# TODO SS: There's no reason we should have to duplicate this
# computation, but we do it currently because each perf
# period maintains its own separate positiondict. We
# should eventually remove this duplication and give each
# period a (preferably read-only) DataFrame of positions.
if len(dividends_earnable):
period.earn_dividends(dividends_earnable)
position_tracker = self.position_tracker
if len(dividends_earnable):
position_tracker.earn_dividends(dividends_earnable)
if len(dividends_payable):
period.pay_dividends(dividends_payable)
if not len(dividends_payable):
return
net_cash_payment = position_tracker.pay_dividends(dividends_payable)
for period in self.perf_periods:
# notify periods to update their stats
period.handle_dividends_paid(net_cash_payment)
def handle_minute_close(self, dt):
self.update_performance()
@@ -498,7 +506,10 @@ class PerformanceTracker(object):
state_dict['_dividend_count'] = self._dividend_count
STATE_VERSION = 1
# we already store perf periods as attributes
del state_dict['perf_periods']
STATE_VERSION = 2
state_dict[VERSION_LABEL] = STATE_VERSION
return state_dict
@@ -516,10 +527,20 @@ class PerformanceTracker(object):
# Handle the dividend frame specially
self.dividend_frame = pickle.loads(state['dividend_frame'])
# We have to restore the references to the objects,
# as the perf periods have been reconstructed as different objects
# with the same values.
self.perf_periods[0] = self.cumulative_performance
self.perf_periods[1] = self.todays_performance
if self.sim_params.emission_rate == 'minute':
self.perf_periods[2] = self.minute_performance
if version == 1:
# V1 had PositionTracker duties on Period.
# default to grabbing the position_tracker from cumulatve
assert 'position_tracker' not in state
position_tracker = self.cumulative_performance.position_tracker
self.position_tracker = position_tracker
# properly setup the perf periods
self.perf_periods = []
p_types = ['cumulative', 'todays', 'minute']
for p_type in p_types:
name = p_type + '_performance'
period = getattr(self, name, None)
if period is None:
continue
period._position_tracker = self.position_tracker
self.perf_periods.append(period)