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

added CLOSE_POSITION as source type, added pt.close_position_event(), added process_close_postion(), added close processing to tradesimulation, added unittest for close_position_event

Browse Source
This commit is contained in:
warren-oneill
2015-06-04 14:23:46 +02:00
parent f9d3dbff47
commit 44fbdff4ac
5 changed files with 62 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_perf_tracking.py
+33 -1
View File
@@ -46,7 +46,8 @@ from zipline.finance.commission import PerShare, PerTrade, PerDollar
from zipline.finance import trading
from zipline.utils.factory import create_random_simulation_parameters
import zipline.protocol as zp
from zipline.protocol import Event
from zipline.protocol import Event, DATASOURCE_TYPE
from zipline.sources.data_frame_source import DataPanelSource
logger = logging.getLogger('Test Perf Tracking')
@@ -1937,6 +1938,37 @@ class TestPerformanceTracker(unittest.TestCase):
check_perf_tracker_serialization(tracker)
def test_close_position_event(self):
pt = perf.PositionTracker()
dt = pd.Timestamp("1984/03/06 3:00PM")
pos1 = perf.Position(1, amount=np.float64(120.0),
last_sale_date=dt, last_sale_price=3.4)
pos2 = perf.Position(2, amount=np.float64(-100.0),
last_sale_date=dt, last_sale_price=3.4)
pt.update_positions({1: pos1, 2: pos2})
event_type = DATASOURCE_TYPE.CLOSE_POSITION
index = [dt + timedelta(days=1)]
pan = pd.Panel({1: pd.DataFrame({'price': 1, 'volume': 0,
'type': event_type}, index=index),
2: pd.DataFrame({'price': 1, 'volume': 0,
'type': event_type}, index=index),
3: pd.DataFrame({'price': 1, 'volume': 0,
'type': event_type}, index=index)})
source = DataPanelSource(pan)
for i, event in enumerate(source):
txn = pt.create_close_position_transaction(event)
if event.sid == 1:
# Test owned long
self.assertEqual(-120, txn.amount)
elif event.sid == 2:
# Test owned short
self.assertEqual(100, txn.amount)
elif event.sid == 3:
# Test not-owned SID
self.assertIsNone(txn)
def test_serialization(self):
start_dt = datetime(year=2008,
month=10,