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

Merge pull request #131 from quantopian/integrate-algo-class-and-remove-sids

Browse Source 
Integrate algo class and remove sids
This commit is contained in:
Eddie Hebert
2012-10-12 07:37:45 -07:00
parent 7c2bdb7b56 df729e01ed
commit 699c304321
19 changed files with 449 additions and 658 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/finance/__init__.py
View File
tests/finance/test_slippage.py
+77
View File
@@ -0,0 +1,77 @@
#
# Copyright 2012 Quantopian, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
Unit tests for finance.slippage
"""
import datetime
import pytz
from unittest2 import TestCase
from zipline.finance.slippage import VolumeShareSlippage
from zipline import ndict
class SlippageTestCase(TestCase):
def test_volume_share_slippage(self):
event = ndict(
{'volume': 200,
'TRANSACTION': None,
'type': 4,
'price': 3.0,
'datetime': datetime.datetime(
2006, 1, 5, 14, 31, tzinfo=pytz.utc),
'high': 3.15,
'low': 2.85,
'sid': 133,
'source_id': 'test_source',
'close': 3.0,
'dt':
datetime.datetime(2006, 1, 5, 14, 31, tzinfo=pytz.utc),
'open': 3.0}
)
slippage_model = VolumeShareSlippage()
open_orders = {133: [
ndict(
{'dt': datetime.datetime(2006, 1, 5, 14, 30, tzinfo=pytz.utc),
'amount': 100,
'filled': 0, 'sid': 133})
]
}
txn = slippage_model.simulate(
event,
open_orders
)
expected_txn = {
'price': float(3.01875),
'dt': datetime.datetime(
2006, 1, 5, 14, 31, tzinfo=pytz.utc),
'amount': int(50),
'sid': int(133)
}
self.assertIsNotNone(txn)
for key, value in expected_txn.items():
self.assertEquals(value, txn[key])
tests/test_algorithm.py
-1
View File
@@ -65,7 +65,6 @@ class TestTransformAlgorithm(TestCase):
def test_transform_registered(self):
algo = TestRegisterTransformAlgorithm(sids=[133])
algo.run(self.source)
assert algo.get_sid_filter() == algo.sids == [133]
assert 'mavg' in algo.registered_transforms
assert algo.registered_transforms['mavg']['args'] == (['price'],)
assert algo.registered_transforms['mavg']['kwargs'] == \
tests/test_exception_handling.py
+4 -56
View File
@@ -17,11 +17,12 @@ from unittest2 import TestCase
from collections import defaultdict
import zipline.utils.simfactory as simfactory
from zipline.test_algorithms import ExceptionAlgorithm, DivByZeroAlgorithm, \
InitializeTimeoutAlgorithm, TooMuchProcessingAlgorithm
from zipline.test_algorithms import (
ExceptionAlgorithm,
DivByZeroAlgorithm,
)
from zipline.finance.slippage import FixedSlippage
from zipline.gens.transform import StatefulTransform
from zipline.utils.timeout import TimeoutException
from zipline.utils.test_utils import (
@@ -78,25 +79,6 @@ class ExceptionTestCase(TestCase):
self.assertEqual(ctx.exception.message,
'An assertion message')
def test_exception_in_init(self):
# Simulation
# ----------
self.zipline_test_config['algorithm'] = \
ExceptionAlgorithm(
'initialize',
self.zipline_test_config['sid']
)
zipline = simfactory.create_test_zipline(
**self.zipline_test_config
)
with self.assertRaises(Exception) as ctx:
output, _ = drain_zipline(self, zipline)
self.assertEqual(ctx.exception.message,
'Algo exception in initialize')
def test_exception_in_handle_data(self):
# Simulation
# ----------
@@ -134,37 +116,3 @@ class ExceptionTestCase(TestCase):
self.assertEqual(ctx.exception.message,
'integer division or modulo by zero')
def test_initialize_timeout(self):
self.zipline_test_config['algorithm'] = \
InitializeTimeoutAlgorithm(
self.zipline_test_config['sid']
)
zipline = simfactory.create_test_zipline(
**self.zipline_test_config
)
with self.assertRaises(TimeoutException) as ctx:
output, _ = drain_zipline(self, zipline)
self.assertEqual(ctx.exception.message, 'Call to initialize timed out')
def test_heartbeat(self):
self.zipline_test_config['algorithm'] = \
TooMuchProcessingAlgorithm(
self.zipline_test_config['sid']
)
zipline = simfactory.create_test_zipline(
**self.zipline_test_config
)
with self.assertRaises(TimeoutException) as ctx:
output, _ = drain_zipline(self, zipline)
self.assertEqual(
ctx.exception.message,
'Too much time spent in handle_data call'
)
tests/test_finance.py
+10 -31
View File
@@ -197,40 +197,19 @@ class FinanceTestCase(TestCase):
}
self.transaction_sim(**params2)
@timed(DEFAULT_TIMEOUT)
def test_partial_expiration_orders(self):
# create a scenario where orders expire without being filled
# entirely
params1 = {
'trade_count': 100,
# Runs the collapsed trades over daily trade intervals.
# Ensuring that our delay works for daily intervals as well.
params3 = {
'trade_count': 6,
'trade_amount': 100,
'trade_delay': timedelta(minutes=5),
'trade_interval': timedelta(days=1),
'order_count': 3,
'order_amount': 1000,
'order_interval': timedelta(minutes=30),
# because we placed an orders totaling less than 25% of one trade
# the simulator should produce just one transaction.
'order_count': 24,
'order_amount': 1,
'order_interval': timedelta(minutes=1),
'expected_txn_count': 1,
'expected_txn_volume': 25
'expected_txn_volume': 24 * 1
}
self.transaction_sim(**params1)
# same scenario, but short sales.
params2 = {
'trade_count': 100,
'trade_amount': 100,
'trade_delay': timedelta(minutes=5),
'trade_interval': timedelta(days=1),
'order_count': 3,
'order_amount': -1000,
'order_interval': timedelta(minutes=30),
# because we placed an orders totaling less than 25% of one trade
# the simulator should produce just one transaction.
'expected_txn_count': 1,
'expected_txn_volume': -25
}
self.transaction_sim(**params2)
self.transaction_sim(**params3)
@timed(DEFAULT_TIMEOUT)
def test_alternating_long_short(self):
@@ -320,7 +299,7 @@ class FinanceTestCase(TestCase):
self.assertEqual(order.sid, sid)
self.assertEqual(order.amount, order_amount * alternator ** i)
tracker = PerformanceTracker(trading_environment, [sid])
tracker = PerformanceTracker(trading_environment)
# this approximates the loop inside TradingSimulationClient
transactions = []
tests/test_perf_tracking.py
+1 -2
View File
@@ -567,8 +567,7 @@ shares in position"
'price',
'changed']
perf_tracker = perf.PerformanceTracker(
self.trading_environment,
[sid, sid2]
self.trading_environment
)
for event in trade_history:
tests/test_transforms.py
+3 -2
View File
@@ -32,6 +32,7 @@ from zipline.gens.stddev import MovingStandardDev
from zipline.gens.returns import Returns
import zipline.utils.factory as factory
from zipline.test_algorithms import BatchTransformAlgorithm
@@ -315,7 +316,7 @@ class BatchTransformTestCase(TestCase):
self.source, self.df = factory.create_test_df_source()
def test_event_window(self):
algo = BatchTransformAlgorithm(sids=[0, 1])
algo = BatchTransformAlgorithm()
algo.run(self.source)
self.assertEqual(algo.history_return_price_class[:2],
@@ -344,7 +345,7 @@ class BatchTransformTestCase(TestCase):
)
def test_passing_of_args(self):
algo = BatchTransformAlgorithm([0, 1], 1, kwarg='str')
algo = BatchTransformAlgorithm(1, kwarg='str')
self.assertEqual(algo.args, (1,))
self.assertEqual(algo.kwargs, {'kwarg': 'str'})
zipline/MESSAGES.py
+38
View File
@@ -0,0 +1,38 @@
from zipline import ndict
# ---------------------
# Error Messages.
# User facing.
# ---------------------
ERRORS = ndict({
# Raised if a user script calls the override_slippage magic
# with a slipage object that isn't a VolumeShareSlippage or
# FixedSlipapge
'UNSUPPORTED_SLIPPAGE_MODEL':
"You attempted to override slippage with an unsupported class. \
Please use VolumeShareSlippage or FixedSlippage.",
# Raised if a users script calls override_slippage magic
# after the initialize method has returned.
'OVERRIDE_SLIPPAGE_POST_INIT':
"You attempted to override slippage after the simulation has \
started. You may only call override_slippage in your initialize \
method.",
# Raised if a user script calls the override_commission magic
# with a commission object that isn't a PerShare or
# PerTrade commission
'UNSUPPORTED_COMMISSION_MODEL':
"You attempted to override commission with an unsupported class. \
Please use PerShare or PerTrade.",
# Raised if a users script calls override_commission magic
# after the initialize method has returned.
'OVERRIDE_COMMISSION_POST_INIT':
"You attempted to override commission after the simulation has \
started. You may only call override_commission in your initialize \
method.",
})
zipline/algorithm.py
+96 -43
View File
@@ -19,9 +19,20 @@ import numpy as np
from zipline.gens.tradegens import DataFrameSource
from zipline.utils.factory import create_trading_environment
from zipline.gens.transform import StatefulTransform
from zipline.lines import SimulatedTrading
from zipline.finance.slippage import FixedSlippage, transact_partial
from zipline.finance.commission import PerShare
from zipline.finance.slippage import (
VolumeShareSlippage,
FixedSlippage,
transact_partial
)
from zipline.finance.commission import PerShare, PerTrade
from zipline.gens.composites import (
date_sorted_sources,
sequential_transforms
)
from zipline.gens.tradesimulation import TradeSimulationClient as tsc
from zipline import MESSAGES
class TradingAlgorithm(object):
@@ -44,55 +55,59 @@ class TradingAlgorithm(object):
>>> stats = my_algo.run(data)
"""
def __init__(self, sids, *args, **kwargs):
def __init__(self, *args, **kwargs):
"""
Initialize sids and other state variables.
Calls user-defined initialize() forwarding *args and **kwargs.
"""
self.sids = sids
self.done = False
self.order = None
self.frame_count = 0
self.portfolio = None
self.registered_transforms = {}
self.transforms = []
self.sources = []
# call to user-defined initialize method
self.logger = None
# default components for transact
self.slippage = VolumeShareSlippage()
self.commission = PerShare()
# an algorithm subclass needs to set initialized to True
# when it is fully initialized.
self.initialized = False
# call to user-defined constructor method
self.initialize(*args, **kwargs)
self.initialized = True
def _create_simulator(self, start, end):
def _create_generator(self, environment):
"""
Create trading environment, transforms and SimulatedTrading object.
Gets called by self.run().
Create a basic generator setup using the sources and
transforms attached to this algorithm.
"""
environment = create_trading_environment(start=start, end=end)
# Create transforms by wrapping them into StatefulTransforms
transforms = []
for namestring, trans_descr in self.registered_transforms.iteritems():
sf = StatefulTransform(
trans_descr['class'],
*trans_descr['args'],
**trans_descr['kwargs']
)
sf.namestring = namestring
self.date_sorted = date_sorted_sources(*self.sources)
self.with_tnfms = sequential_transforms(self.date_sorted,
*self.transforms)
self.trading_client = tsc(self, environment)
transforms.append(sf)
transact_method = transact_partial(self.slippage, self.commission)
self.set_transact(transact_method)
# SimulatedTrading is the main class handling data streaming,
# application of transforms and calling of the user algo.
return SimulatedTrading(
self.sources,
transforms,
self,
environment,
transact_partial(FixedSlippage(), PerShare(0.0))
)
return self.trading_client.simulate(self.with_tnfms)
def get_generator(self, environment):
"""
Override this method to add new logic to the construction
of the generator. Overrides can use the _create_generator
method to get a standard construction generator.
"""
return self._create_generator(environment)
# TODO: make a new subclass, e.g. BatchAlgorithm, and move
# the run method to the subclass, and refactor to put the
# generator creation logic into get_generator.
def run(self, source, start=None, end=None):
"""Run the algorithm.
@@ -121,7 +136,7 @@ start and end date have to be specified."""
elif isinstance(source, pd.DataFrame):
assert isinstance(source.index, pd.tseries.index.DatetimeIndex)
# if DataFrame provided, wrap in DataFrameSource
source = DataFrameSource(source, sids=self.sids)
source = DataFrameSource(source)
# If values not set, try to extract from source.
if start is None:
@@ -134,12 +149,25 @@ start and end date have to be specified."""
else:
self.sources = source
# Create transforms by wrapping them into StatefulTransforms
for namestring, trans_descr in self.registered_transforms.iteritems():
sf = StatefulTransform(
trans_descr['class'],
*trans_descr['args'],
**trans_descr['kwargs']
)
sf.namestring = namestring
self.transforms.append(sf)
environment = create_trading_environment(start=start, end=end)
# create transforms and zipline
self.simulated_trading = self._create_simulator(start=start, end=end)
self.gen = self._create_generator(environment)
# loop through simulated_trading, each iteration returns a
# perf ndict
perfs = list(self.simulated_trading)
perfs = list(self.gen)
# convert perf ndict to pandas dataframe
daily_stats = self._create_daily_stats(perfs)
@@ -186,14 +214,39 @@ start and end date have to be specified."""
def set_order(self, order_callable):
self.order = order_callable
def get_sid_filter(self):
return self.sids
def set_logger(self, logger):
self.logger = logger
def initialize(self, *args, **kwargs):
def init(self, *args, **kwargs):
"""Called from constructor."""
pass
def set_transact_setter(self, transact_setter):
pass
def set_transact(self, transact):
"""
Set the method that will be called to create a
transaction from open orders and trade events.
"""
self.trading_client.ordering_client.transact = transact
def set_slippage(self, slippage):
assert isinstance(slippage, (VolumeShareSlippage, FixedSlippage)), \
MESSAGES.ERRORS.UNSUPPORTED_SLIPPAGE_MODEL
if self.initialized:
raise Exception(MESSAGES.ERRORS.OVERRIDE_SLIPPAGE_POST_INIT)
self.slippage = slippage
def set_commission(self, commission):
assert isinstance(commission, (PerShare, PerTrade)), \
MESSAGES.ERRORS.UNSUPPORTED_COMMISSION_MODEL
if self.initialized:
raise Exception(MESSAGES.ERRORS.OVERRIDE_COMMISSION_POST_INIT)
self.commission = commission
def set_sources(self, sources):
assert isinstance(sources, list)
self.sources = sources
def set_transforms(self, transforms):
assert isinstance(transforms, list)
self.transforms = transforms
zipline/finance/performance.py
+29 -21
View File
@@ -154,7 +154,7 @@ class PerformanceTracker(object):
"""
def __init__(self, trading_environment, sid_list):
def __init__(self, trading_environment):
self.trading_environment = trading_environment
self.trading_day = datetime.timedelta(hours=6, minutes=30)
@@ -178,7 +178,7 @@ class PerformanceTracker(object):
# this performance period will span the entire simulation.
self.cumulative_performance = PerformancePeriod(
# initial positions are empty
{},
positiondict(),
# initial portfolio positions have zero value
0,
# initial cash is your capital base.
@@ -191,7 +191,7 @@ class PerformanceTracker(object):
# this performance period will span just the current market day
self.todays_performance = PerformancePeriod(
# initial positions are empty
{},
positiondict(),
# initial portfolio positions have zero value
0,
# initial cash is your capital base.
@@ -203,10 +203,6 @@ class PerformanceTracker(object):
keep_transactions=True
)
for sid in sid_list:
self.cumulative_performance.positions[sid] = Position(sid)
self.todays_performance.positions[sid] = Position(sid)
def transform(self, stream_in):
"""
Main generator work loop.
@@ -409,7 +405,11 @@ class PerformancePeriod(object):
self.period_capital_used = 0.0
self.pnl = 0.0
#sid => position object
self.positions = initial_positions
if not isinstance(initial_positions, positiondict):
self.positions = positiondict()
self.positions.update(initial_positions)
else:
self.positions = initial_positions
self.starting_value = starting_value
#cash balance at start of period
self.starting_cash = starting_cash
@@ -436,11 +436,8 @@ class PerformancePeriod(object):
self.returns = 0.0
def execute_transaction(self, txn):
# Update Position
# ----------------
if txn.sid not in self.positions:
self.positions[txn.sid] = Position(txn.sid)
self.positions[txn.sid].update(txn)
self.period_capital_used += -1 * txn.price * txn.amount
@@ -547,21 +544,16 @@ class PerformancePeriod(object):
del(portfolio['max_leverage'])
del(portfolio['max_capital_used'])
portfolio['positions'] = self.get_positions(ndicted=True)
portfolio['positions'] = self.get_positions()
return zp.ndict(portfolio)
def get_positions(self, ndicted=False):
if ndicted:
positions = zp.ndict({})
else:
positions = {}
def get_positions(self):
positions = zp.ndict(internal=position_ndict())
for sid, pos in self.positions.iteritems():
cur = pos.to_dict()
if ndicted:
positions[sid] = zp.ndict(cur)
else:
positions[sid] = cur
positions[sid] = zp.ndict(cur)
return positions
@@ -571,3 +563,19 @@ class PerformancePeriod(object):
cur = pos.to_dict()
positions.append(cur)
return positions
class positiondict(dict):
def __missing__(self, key):
pos = Position(key)
self[key] = pos
return pos
class position_ndict(dict):
def __missing__(self, key):
pos = Position(key)
self[key] = zp.ndict(pos.to_dict())
return pos
zipline/finance/slippage.py
+32 -32
View File
@@ -12,6 +12,7 @@
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
from datetime import timedelta
import pytz
import math
@@ -21,13 +22,12 @@ from functools import partial
import zipline.protocol as zp
def transact_stub(slippage, commission, open_orders, events):
def transact_stub(slippage, commission, event, open_orders):
"""
This is intended to be wrapped in a partial, so that the
slippage and commission models can be enclosed.
"""
transaction = slippage.simulate(open_orders, events)
transaction = slippage.simulate(event, open_orders)
if transaction and transaction.amount != 0:
direction = abs(transaction.amount) / transaction.amount
per_share, total_commission = commission.calculate(transaction)
@@ -55,11 +55,13 @@ def create_transaction(sid, amount, price, dt):
class VolumeShareSlippage(object):
def __init__(self,
volume_limit=.25,
price_impact=0.1):
volume_limit=.25,
price_impact=0.1,
delay=timedelta(minutes=1)):
self.volume_limit = volume_limit
self.price_impact = price_impact
self.delay = delay
def simulate(self, event, open_orders):
@@ -71,6 +73,10 @@ class VolumeShareSlippage(object):
if event.sid in open_orders:
orders = open_orders[event.sid]
orders = sorted(orders, key=lambda o: o.dt)
# Only use orders for the current day or before
current_orders = filter(
lambda o: o.dt + self.delay <= event.dt,
orders)
else:
return None
@@ -79,42 +85,36 @@ class VolumeShareSlippage(object):
simulated_amount = 0
simulated_impact = 0.0
direction = 1.0
for order in orders:
if(order.dt < event.dt):
for order in current_orders:
# orders are only good on the day they are issued
if order.dt.day < event.dt.day:
continue
open_amount = order.amount - order.filled
open_amount = order.amount - order.filled
if(open_amount != 0):
direction = open_amount / math.fabs(open_amount)
else:
direction = 1
if(open_amount != 0):
direction = open_amount / math.fabs(open_amount)
else:
direction = 1
desired_order = total_order + open_amount
desired_order = total_order + open_amount
volume_share = min(direction * (desired_order) / event.volume,
self.volume_limit)
simulated_amount = int(volume_share * event.volume * direction)
simulated_impact = (volume_share) ** 2 \
* self.price_impact * direction * event.price
volume_share = direction * (desired_order) / event.volume
if volume_share > self.volume_limit:
volume_share = self.volume_limit
simulated_amount = int(volume_share * event.volume * direction)
simulated_impact = (volume_share) ** 2 \
* self.price_impact * direction * event.price
order.filled += (simulated_amount - total_order)
total_order = simulated_amount
order.filled += (simulated_amount - total_order)
total_order = simulated_amount
# we cap the volume share at configured % of a trade
if volume_share == self.volume_limit:
break
# we cap the volume share at configured % of a trade
if volume_share == self.volume_limit:
break
filled_orders = [x for x in orders
if abs(x.amount - x.filled) > 0
and x.dt.day >= event.dt.day]
orders = [x for x in orders
if abs(x.amount - x.filled) > 0
and x.dt.day >= event.dt.day]
open_orders[event.sid] = orders
open_orders[event.sid] = filled_orders
if simulated_amount != 0:
return create_transaction(
zipline/finance/trading.py
+5 -13
View File
@@ -21,9 +21,9 @@ from collections import defaultdict
import zipline.protocol as zp
from zipline.finance.slippage import (
VolumeShareSlippage,
transact_partial
)
VolumeShareSlippage,
transact_partial
)
from zipline.finance.commission import PerShare
log = logbook.Logger('Transaction Simulator')
@@ -31,16 +31,8 @@ log = logbook.Logger('Transaction Simulator')
class TransactionSimulator(object):
def __init__(self, transact=None):
if transact is not None:
self.transact = transact
else:
self.transact = transact_partial(
VolumeShareSlippage(),
PerShare()
)
def __init__(self):
self.transact = transact_partial(VolumeShareSlippage(), PerShare())
self.open_orders = defaultdict(list)
def place_order(self, order):
zipline/gens/tradesimulation.py
+9 -56
View File
@@ -15,13 +15,13 @@
from logbook import Logger, Processor
from collections import defaultdict
from datetime import datetime
from itertools import groupby
from operator import attrgetter
from zipline import ndict
from zipline.utils.timeout import Heartbeat, Timeout
from zipline.finance.trading import TransactionSimulator
from zipline.finance.performance import PerformanceTracker
@@ -29,11 +29,6 @@ from zipline.gens.utils import hash_args
log = Logger('Trade Simulation')
# TODO: make these arguments rather than global constants
INIT_TIMEOUT = 5
HEARTBEAT_INTERVAL = 1 # seconds
MAX_HEARTBEAT_INTERVALS = 15 # count
class TradeSimulationClient(object):
"""
@@ -69,15 +64,13 @@ class TradeSimulationClient(object):
is sent to the algo.
"""
def __init__(self, algo, environment, transact):
def __init__(self, algo, environment):
self.algo = algo
self.sids = algo.get_sid_filter()
self.environment = environment
self.transact = transact
self.ordering_client = TransactionSimulator(self.transact)
self.perf_tracker = PerformanceTracker(self.environment, self.sids)
self.ordering_client = TransactionSimulator()
self.perf_tracker = PerformanceTracker(self.environment)
self.algo_start = self.environment.first_open
self.algo_sim = AlgorithmSimulator(
@@ -139,44 +132,20 @@ class AlgorithmSimulator(object):
self.order_book = order_book
self.algo = algo
self.sids = algo.get_sid_filter()
self.algo_start = algo_start
# Monkey patch the user algorithm to place orders in the
# TransactionSimulator's order book and use our logger.
self.algo.set_order(self.order)
self.algolog = Logger("AlgoLog")
self.algo.set_logger(self.algolog)
# Provide user algorithm with a setter for the transact
# method (method that constructs transactions based on
# open orders and trade events).
self.algo.set_transact_setter(self.set_transact)
# Handler for heartbeats during calls to handle_data.
def log_heartbeats(beat_count, stackframe):
t = beat_count * HEARTBEAT_INTERVAL
warning = "handle_data has been processing for %i seconds" % t
self.algolog.warn(warning)
# Context manager that calls log_heartbeats every HEARTBEAT_INTERVAL
# seconds, raising an exception after MAX_HEARTBEATS
self.heartbeat_monitor = Heartbeat(
HEARTBEAT_INTERVAL,
MAX_HEARTBEAT_INTERVALS,
frame_handler=log_heartbeats,
timeout_message="Too much time spent in handle_data call"
)
# ==============
# Snapshot Setup
# ==============
# The algorithm's universe as of our most recent event.
self.universe = ndict()
for sid in self.sids:
self.universe[sid] = ndict()
self.universe.portfolio = None
# We want an ndict that will have empty ndicts as default
# values on missing keys.
self.universe = ndict(internal=defaultdict(ndict))
# We don't have a datetime for the current snapshot until we
# receive a message.
@@ -193,19 +162,11 @@ class AlgorithmSimulator(object):
record.extra['algo_dt'] = self.snapshot_dt
self.processor = Processor(inject_algo_dt)
def set_transact(self, transact):
"""
Set the method that will be called to create a
transaction from open orders and trade events.
"""
self.order_book.transact = transact
def order(self, sid, amount):
"""
Closure to pass into the user's algo to allow placing orders
into the transaction simulator's dict of open orders.
"""
assert sid in self.sids, "Order on invalid sid: %i" % sid
order = ndict({
'dt': self.simulation_dt,
'sid': sid,
@@ -233,12 +194,6 @@ class AlgorithmSimulator(object):
"""
Main generator work loop.
"""
# Call user's initialize method with a timeout (only if
# initialize wasn't called already).
if not getattr(self.algo, 'initialized', False):
with Timeout(INIT_TIMEOUT, message="Call to initialize timed out"):
self.algo.initialize()
# inject the current algo
# snapshot time to any log record generated.
with self.processor.threadbound():
@@ -298,7 +253,7 @@ class AlgorithmSimulator(object):
Update the universe with new event information.
"""
# Update our portfolio.
self.universe.portfolio = event.portfolio
self.algo.set_portfolio(event.portfolio)
# Update our knowledge of this event's sid
for field in event.keys():
@@ -312,10 +267,8 @@ class AlgorithmSimulator(object):
# Needs to be set so that we inject the proper date into algo
# log/print lines.
self.snapshot_dt = date
start_tic = datetime.now()
with self.heartbeat_monitor:
self.algo.handle_data(self.universe)
self.algo.handle_data(self.universe)
stop_tic = datetime.now()
# How long did you take?
zipline/gens/transform.py
+24 -13
View File
@@ -348,16 +348,16 @@ class BatchTransform(EventWindow):
refresh_period=None,
market_aware=True,
delta=None,
days=None,
sids=None):
super(BatchTransform, self).__init__(
market_aware, days=days, delta=delta)
days=None):
super(BatchTransform, self).__init__(market_aware,
days=days, delta=delta)
if func is not None:
self.compute_transform_value = func
else:
self.compute_transform_value = self.get_value
self.sids = sids
self.refresh_period = refresh_period
self.days = days
@@ -373,15 +373,20 @@ class BatchTransform(EventWindow):
handle_data method.
"""
# extract dates
dts = [data[sid].datetime for sid in self.sids]
#dts = [data[sid].datetime for sid in self.sids]
dts = [event.datetime for event in data.itervalues()]
# we have to provide the event with a dt. This is only for
# checking if the event is outside the window or not so a
# couple of seconds shouldn't matter
data.dt = max(dts)
# couple of seconds shouldn't matter. We don't add it to
# the data parameter, because it would mix dt with the
# sid keys.
event = ndict()
event.dt = max(dts)
event.data = data
# append data frame to window. update() will call handle_add() and
# handle_remove() appropriately
self.update(data)
self.update(event)
# return newly computed or cached value
return self.get_transform_value(*args, **kwargs)
@@ -403,17 +408,23 @@ class BatchTransform(EventWindow):
#
# This Panel data structure ultimately gets passed to the
# user-overloaded get_value() method.
#
# self.ticks contains ndicts with data, dt keys.
# event parameter is an ndict with data, dt keys.
fields = {}
for field_name in ['price', 'volume']:
sids = self.ticks[0].data.keys()
# Skip non-existant fields
if field_name not in self.ticks[0][self.sids[0]]:
if field_name not in self.ticks[0].data[sids[0]]:
continue
values_per_sid = {}
for sid in self.sids:
for sid in sids:
values_per_sid[sid] = pd.Series(
{tick[sid].dt: tick[sid][field_name]
for tick in self.ticks})
{tick.data[sid].dt: tick.data[sid][field_name]
for tick in self.ticks}
)
# concatenate different sids into one df
fields[field_name] = pd.DataFrame.from_dict(values_per_sid)
zipline/lines.py
-114
View File
@@ -1,114 +0,0 @@
"""
Ziplines are composed of multiple components connected by asynchronous
messaging. All ziplines follow a general topology of parallel sources,
datetimestamp serialization, parallel transformations, and finally sinks.
Furthermore, many ziplines have common needs. For example, all trade
simulations require a
:py:class:`~zipline.finance.trading.TradeSimulationClient`.
To establish best practices and minimize code replication, the lines module
provides complete zipline topologies. You can extend any zipline without
the need to extend the class. Simply instantiate any additional components
that you would like included in the zipline, and add them to the zipline
before invoking simulate.
Here is a diagram of the SimulatedTrading zipline:
+----------------------+ +------------------------+
| Trade History | | (DataSource added |
| | | via add_source) |
| | | |
+--------------------+-+ +-+----------------------+
| |
| |
v v
+---------+
| Feed | (ensures events are serialized
+-+------++ in chronological order)
| |
| |
v v
+----------------------+ +----------------------+
| (Transforms added | | (Transforms added |
| via add_transform) | | via add_transform) |
+-------------------+--+ +-+--------------------+
| |
| |
v v
+------------+
| Merge | (combines original event and
+------+-----+ transforms into one vector)
|
|
V
+---------------+ +--------------------------------+
| Risk and Perf | | |
| Tracker | | TradingSimulationClient |
+---------------+ | tracks performance and |
^ Trades and | provides API to algorithm. |
| simulated | |
| transactions +--+------------------+----------+
| | ^ |
+---------------------+ | orders | frames
| |
| v
+---------------------------------+
| Algorithm added via |
| __init__. |
+---------------------------------+
"""
from zipline.gens.composites import (
date_sorted_sources,
sequential_transforms
)
from zipline.gens.tradesimulation import TradeSimulationClient as tsc
from logbook import Logger
log = Logger('Lines')
class SimulatedTrading(object):
def __init__(self,
sources,
transforms,
algorithm,
environment,
sim_method=None):
"""
@sources - an iterable of iterables
These iterables must yield ndicts that contain:
- type :: a ziplines.protocol.DATASOURCE_TYPE
- dt :: a milliseconds since epoch timestamp in UTC
@transforms - An iterable of instances of StatefulTransform.
@algorithm - An object that implements:
`def initialize(self)`
`def handle_data(self, data)`
`def get_sid_filter(self)`
`def set_logger(self, logger)`
`def set_order(self, order_callable)`
@environment - An instance of finance.trading.TradingEnvironment
@slippage - an object with a simulate method that takes a
trade event and returns a transaction
"""
self.date_sorted = date_sorted_sources(*sources)
self.transforms = transforms
# Formerly merged_transforms.
self.with_tnfms = sequential_transforms(self.date_sorted,
*self.transforms)
self.trading_client = tsc(algorithm, environment, sim_method)
self.gen = self.trading_client.simulate(self.with_tnfms)
def __iter__(self):
return self
def next(self):
return self.gen.next()
zipline/test_algorithms.py
+21 -220
View File
@@ -67,42 +67,28 @@ The algorithm must expose methods:
and trade events.
"""
from zipline.algorithm import TradingAlgorithm
from zipline.finance.slippage import FixedSlippage
class TestAlgorithm():
class TestAlgorithm(TradingAlgorithm):
"""
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, order_count, sid_filter=None):
def initialize(self, sid, amount, order_count, sid_filter=None):
self.count = order_count
self.sid = sid
self.amount = amount
self.incr = 0
self.done = False
self.order = None
self.frame_count = 0
self.portfolio = None
if sid_filter:
self.sid_filter = sid_filter
else:
self.sid_filter = [self.sid]
def initialize(self):
pass
def set_order(self, order_callable):
self.order = order_callable
def set_logger(self, logger):
pass
def set_portfolio(self, portfolio):
self.portfolio = portfolio
def handle_data(self, data):
self.frame_count += 1
#place an order for 100 shares of sid
@@ -110,40 +96,18 @@ class TestAlgorithm():
self.order(self.sid, self.amount)
self.incr += 1
def get_sid_filter(self):
return self.sid_filter
def set_transact_setter(self, txn_sim_callable):
pass
class HeavyBuyAlgorithm():
class HeavyBuyAlgorithm(TradingAlgorithm):
"""
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):
def initialize(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 initialize(self):
pass
def set_order(self, order_callable):
self.order = order_callable
def set_logger(self, logger):
pass
def set_portfolio(self, portfolio):
self.portfolio = portfolio
def handle_data(self, data):
self.frame_count += 1
@@ -151,33 +115,11 @@ class HeavyBuyAlgorithm():
self.order(self.sid, self.amount)
self.incr += 1
def get_sid_filter(self):
return [self.sid]
def set_transact_setter(self, txn_sim_callable):
pass
class NoopAlgorithm(object):
class NoopAlgorithm(TradingAlgorithm):
"""
Dolce fa niente.
"""
def initialize(self):
pass
def set_order(self, order_callable):
pass
def set_logger(self, logger):
pass
def set_portfolio(self, portfolio):
pass
def handle_data(self, data):
pass
def get_sid_filter(self):
return []
@@ -185,17 +127,17 @@ class NoopAlgorithm(object):
pass
class ExceptionAlgorithm(object):
class ExceptionAlgorithm(TradingAlgorithm):
"""
Throw an exception from the method name specified in the
constructor.
"""
def __init__(self, throw_from, sid):
def initialize(self, throw_from, sid):
self.throw_from = throw_from
self.sid = sid
def initialize(self):
if self.throw_from == "initialize":
raise Exception("Algo exception in initialize")
else:
@@ -207,9 +149,6 @@ class ExceptionAlgorithm(object):
else:
pass
def set_logger(self, logger):
pass
def set_portfolio(self, portfolio):
if self.throw_from == "set_portfolio":
raise Exception("Algo exception in set_portfolio")
@@ -232,81 +171,23 @@ class ExceptionAlgorithm(object):
pass
class DivByZeroAlgorithm():
class DivByZeroAlgorithm(TradingAlgorithm):
def __init__(self, sid):
def initialize(self, sid):
self.sid = sid
self.incr = 0
def initialize(self):
pass
def set_order(self, order_callable):
pass
def set_logger(self, logger):
pass
def set_portfolio(self, portfolio):
pass
def handle_data(self, data):
self.incr += 1
if self.incr > 4:
5 / 0
pass
def get_sid_filter(self):
return [self.sid]
def set_transact_setter(self, txn_sim_callable):
pass
class TooMuchProcessingAlgorithm(TradingAlgorithm):
class InitializeTimeoutAlgorithm():
def __init__(self, sid):
def initialize(self, sid):
self.sid = sid
self.incr = 0
def initialize(self):
import time
from zipline.gens.tradesimulation import INIT_TIMEOUT
time.sleep(INIT_TIMEOUT + 1000)
def set_order(self, order_callable):
pass
def set_logger(self, logger):
pass
def set_portfolio(self, portfolio):
pass
def handle_data(self, data):
pass
def get_sid_filter(self):
return [self.sid]
def set_transact_setter(self, txn_sim_callable):
pass
class TooMuchProcessingAlgorithm():
def __init__(self, sid):
self.sid = sid
def initialize(self):
pass
def set_order(self, order_callable):
pass
def set_logger(self, logger):
pass
def set_portfolio(self, portfolio):
pass
def handle_data(self, data):
# Unless we're running on some sort of
@@ -314,100 +195,19 @@ class TooMuchProcessingAlgorithm():
for i in xrange(1000000000):
self.foo = i
def get_sid_filter(self):
return [self.sid]
def set_transact_setter(self, txn_sim_callable):
pass
class TimeoutAlgorithm(TradingAlgorithm):
class TimeoutAlgorithm():
def __init__(self, sid):
def initialize(self, sid):
self.sid = sid
self.incr = 0
def initialize(self):
pass
def set_order(self, order_callable):
pass
def set_logger(self, logger):
pass
def set_portfolio(self, portfolio):
pass
def handle_data(self, data):
if self.incr > 4:
import time
time.sleep(100)
pass
def get_sid_filter(self):
return [self.sid]
def set_transact_setter(self, txn_sim_callable):
pass
class TestPrintAlgorithm():
def __init__(self, sid):
self.sid = sid
def initialize(self):
print "Initializing..."
def set_order(self, order_callable):
pass
def set_logger(self, logger):
pass
def set_portfolio(self, portfolio):
pass
def handle_data(self, data):
print "Handling Data..."
pass
def get_sid_filter(self):
return [self.sid]
def set_transact_setter(self, txn_sim_callable):
pass
class TestLoggingAlgorithm():
def __init__(self, sid):
self.log = None
self.sid = sid
def initialize(self):
self.log.info("Initializing...")
def set_order(self, order_callable):
pass
def set_logger(self, logger):
self.log = logger
def set_portfolio(self, portfolio):
pass
def handle_data(self, data):
self.log.info("Handling Data...")
def get_sid_filter(self):
return [self.sid]
def set_transact_setter(self, txn_sim_callable):
pass
from datetime import timedelta
from zipline.algorithm import TradingAlgorithm
from zipline.gens.transform import BatchTransform, batch_transform
@@ -415,11 +215,13 @@ from zipline.gens.mavg import MovingAverage
class TestRegisterTransformAlgorithm(TradingAlgorithm):
def initialize(self):
def initialize(self, *args, **kwargs):
self.add_transform(MovingAverage, 'mavg', ['price'],
market_aware=True,
days=2)
self.set_slippage(FixedSlippage())
def handle_data(self, data):
pass
@@ -454,26 +256,25 @@ class BatchTransformAlgorithm(TradingAlgorithm):
self.kwargs = kwargs
self.return_price_class = ReturnPriceBatchTransform(
sids=self.sids,
market_aware=False,
refresh_period=2,
delta=timedelta(days=self.days)
)
self.return_price_decorator = return_price_batch_decorator(
sids=self.sids,
market_aware=False,
refresh_period=2,
delta=timedelta(days=self.days)
)
self.return_args_batch = return_args_batch_decorator(
sids=self.sids,
market_aware=False,
refresh_period=2,
delta=timedelta(days=self.days)
)
self.set_slippage(FixedSlippage())
def handle_data(self, data):
self.history_return_price_class.append(
self.return_price_class.handle_data(data))
zipline/utils/date_utils.py
+47
View File
@@ -41,6 +41,53 @@ def parse_iso8061(date_string):
dt = dt.replace(tzinfo=pytz.utc)
return dt
# quarter utilities
# ---------------------
def get_quarter(dt):
"""
convert the given datetime to an integer representing
the number of calendar quarters since 0.
"""
quarters = dt.year * 4
month = dt.month
if month <= 3:
return quarters + 1
elif month <= 6:
return quarters + 2
elif month <= 9:
return quarters + 3
else:
return quarters + 4
def dates_of_quarter(quarter_num):
year = quarter_num / 4
quarter = quarter_num % 4
if quarter == 0:
quarter = 4
if quarter == 1:
start = datetime(year, 1, 1, 0, 0, tzinfo=pytz.utc)
end = datetime(year, 3, 31, 23, 59, tzinfo=pytz.utc)
return start, end
elif quarter == 2:
start = datetime(year, 4, 1, 0, 0, tzinfo=pytz.utc)
end = datetime(year, 6, 30, 23, 59, tzinfo=pytz.utc)
return start, end
elif quarter == 3:
start = datetime(year, 7, 1, 0, 0, tzinfo=pytz.utc)
end = datetime(year, 9, 30, 23, 59, tzinfo=pytz.utc)
return start, end
elif quarter == 4:
start = datetime(year, 10, 1, 0, 0, tzinfo=pytz.utc)
end = datetime(year, 12, 31, 23, 59, tzinfo=pytz.utc)
return start, end
# Epoch utilities
# ---------------------
UNIX_EPOCH = datetime(1970, 1, 1, 0, 0, tzinfo=pytz.utc)
zipline/utils/protocol_utils.py
+5 -2
View File
@@ -45,8 +45,11 @@ class ndict(MutableMapping):
cls = None
__slots__ = ['cls', '__internal']
def __init__(self, dct=None):
self.__internal = dict()
def __init__(self, dct=None, internal=None):
if internal is not None:
self.__internal = internal
else:
self.__internal = dict()
if not ndict.cls:
ndict.cls = frozenset(dir(self))
zipline/utils/simfactory.py
+48 -52
View File
@@ -1,36 +1,33 @@
import zipline.utils.factory as factory
from zipline.test_algorithms import TestAlgorithm
from zipline.lines import SimulatedTrading
from zipline.finance.slippage import FixedSlippage, transact_partial
from zipline.finance.commission import PerShare
def create_test_zipline(**config):
"""
:param config: A configuration object that is a dict with:
:param config: A configuration object that is a dict with:
- environment - a \
:py:class:`zipline.finance.trading.TradingEnvironment`
- sid - an integer, which will be used as the security ID.
- order_count - the number of orders the test algo will place,
defaults to 100
- order_amount - the number of shares per order, defaults to 100
- trade_count - the number of trades to simulate, defaults to 101
to ensure all orders are processed.
- algorithm - optional parameter providing an algorithm. defaults
to :py:class:`zipline.test.algorithms.TestAlgorithm`
- trade_source - optional parameter to specify trades, if present.
If not present :py:class:`zipline.sources.SpecificEquityTrades`
is the source, with daily frequency in trades.
- slippage: optional parameter that configures the
:py:class:`zipline.gens.tradingsimulation.TransactionSimulator`. Expects
an object with a simulate mehod, such as
:py:class:`zipline.gens.tradingsimulation.FixedSlippage`.
:py:mod:`zipline.finance.trading`
- transforms: optional parameter that provides a list
of StatefulTransform objects.
"""
- environment - a \
:py:class:`zipline.finance.trading.TradingEnvironment`
- sid - an integer, which will be used as the security ID.
- order_count - the number of orders the test algo will place,
defaults to 100
- order_amount - the number of shares per order, defaults to 100
- trade_count - the number of trades to simulate, defaults to 101
to ensure all orders are processed.
- algorithm - optional parameter providing an algorithm. defaults
to :py:class:`zipline.test.algorithms.TestAlgorithm`
- trade_source - optional parameter to specify trades, if present.
If not present :py:class:`zipline.sources.SpecificEquityTrades`
is the source, with daily frequency in trades.
- slippage: optional parameter that configures the
:py:class:`zipline.gens.tradingsimulation.TransactionSimulator`.
Expects an object with a simulate mehod, such as
:py:class:`zipline.gens.tradingsimulation.FixedSlippage`.
:py:mod:`zipline.finance.trading`
- transforms: optional parameter that provides a list
of StatefulTransform objects.
"""
assert isinstance(config, dict)
sid_list = config.get('sid_list')
if not sid_list:
@@ -64,12 +61,20 @@ def create_test_zipline(**config):
# trade than order
trade_count = 101
slippage = config.get('slippage', FixedSlippage())
commission = PerShare()
transact_method = transact_partial(slippage, commission)
#-------------------
# Create the Algo
#-------------------
if 'algorithm' in config:
test_algo = config['algorithm']
else:
test_algo = TestAlgorithm(
sid,
order_amount,
order_count
)
#-------------------
# Trade Source
# Trade Source
#-------------------
if 'trade_source' in config:
trade_source = config['trade_source']
@@ -81,34 +86,25 @@ def create_test_zipline(**config):
concurrent=concurrent_trades
)
test_algo.set_sources([trade_source])
#-------------------
# Transforms
#-------------------
transforms = config.get('transforms', [])
transforms = config.get('transforms', None)
if transforms is not None:
test_algo.set_transforms(transforms)
#-------------------
# Create the Algo
#-------------------
if 'algorithm' in config:
test_algo = config['algorithm']
else:
test_algo = TestAlgorithm(
sid,
order_amount,
order_count
)
# Slippage
# ------------------
slippage = config.get('slippage', None)
if slippage is not None:
test_algo.set_slippage(slippage)
#-------------------
# Simulation
#-------------------
sim = SimulatedTrading(
[trade_source],
transforms,
test_algo,
trading_environment,
transact_method
)
#-------------------
# ------------------
# generator/simulator
sim = test_algo.get_generator(trading_environment)
return sim