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

Resolve merge conflicts. Fix: use hasattr instead of checking class dict. Turn SimulatedTradingLite into a generator.

Browse Source
This commit is contained in:
Thomas Wiecki
2012-09-07 10:28:36 -04:00
parent 5fcaed3644 d6ddc1390d
commit 81ea337ae6
51 changed files with 259 additions and 3463 deletions
Download Patch File Download Diff File Expand all files Collapse all files
etc/ordered_pip.sh
+5 -4
View File
@@ -1,11 +1,12 @@
#!/bin/bash
#!/bin/bash -e
echo $hash
a=0
while read line
do
if [[ $line != \#* ]] ; then
if [[ -n "$line" && "$line" != \#* ]] ; then
#echo $line
pip install $line
fi
((a = a + 1))
done < $1
echo "Final line count is: $a";
echo "$0: Final package count is $a";
etc/requirements.txt
-6
View File
@@ -1,16 +1,11 @@
msgpack-python==0.1.12
humanhash==0.0.1
ujson==1.18
iso8601==0.1.4
# ZeroMQ
pyzmq==2.1.11
gevent-zeromq==0.2.2
# Packaging
distribute==0.6.27
setuptools==0.6c11
# Unix
setproctitle==1.1.6
@@ -18,4 +13,3 @@ setproctitle==1.1.6
Logbook==0.3
blist==1.3.4
psutil==0.4.1
etc/requirements_sci.txt
+2 -2
View File
@@ -13,8 +13,8 @@ matplotlib==1.1.0
numexpr==2.0.1
Cython==0.15.1
patsy==0.1.0
statsmodels==0.5.0-tutorial-beta
statsmodels>=0.5.0
scikit-learn==0.11
# ZeroMQ
pyzmq==2.1.11
notebooks/Experimenting with Frames.ipynb
-352
View File
@@ -1,352 +0,0 @@
{
"metadata": {
"name": "Experimenting with Frames"
},
"nbformat": 3,
"worksheets": [
{
"cells": [
{
"cell_type": "heading",
"source": [
"Performance Tracking"
]
},
{
"cell_type": "code",
"collapsed": false,
"input": [
"import datetime",
"import pandas",
"import pytz",
"",
"import zipline.test.factory as factory",
"import zipline.finance.performance as perf",
"import zipline.protocol as zp",
"import zipline.finance.risk as risk",
"import zipline.finance.trading as trading"
],
"language": "python",
"outputs": [],
"prompt_number": 38
},
{
"cell_type": "heading",
"source": [
"Create a simulated trade history using the test factory"
]
},
{
"cell_type": "markdown",
"source": [
"For any backtesting, zipline relies on a TradingEnvironment object. Trading environment holds essential facts: ",
" ",
" - start and end times for the simulation.",
" - historical daily returns for your benchmark.",
" - historical treasury curves",
" - an assumed capital base for your portfolio",
" - a calendar of trading days based on your benchmark",
"",
"zipline ships with a compressed archives of the S&amp;P daily returns, and US treasury curves to facilitate standalone development and testing. In the next cell we instantiate the environment using these defaults. You can see more of this in zipline/test/test_perf_tracking.py"
]
},
{
"cell_type": "code",
"collapsed": false,
"input": [
"benchmark_returns, treasury_curves = factory.load_market_data()",
" ",
"trading_environment = risk.TradingEnvironment(benchmark_returns, treasury_curves)"
],
"language": "python",
"outputs": [],
"prompt_number": 39
},
{
"cell_type": "code",
"collapsed": false,
"input": [
"trade_count = 100",
"sid = 133",
"price = 10.1 ",
"price_list = [price] * trade_count",
"volume = [100] * trade_count",
"start_date = datetime.datetime.strptime(\"01/01/2011\",\"%m/%d/%Y\")",
"start_date = start_date.replace(tzinfo=pytz.utc)",
"trade_time_increment = datetime.timedelta(days=1)",
"",
"trade_history = factory.create_trade_history( ",
" sid, ",
" price_list, ",
" volume, ",
" start_date, ",
" trade_time_increment, ",
" trading_environment ",
")",
"",
"sid2 = 134",
"price2 = 12.12",
"price2_list = [price2] * trade_count ",
"trade_history2 = factory.create_trade_history( ",
" sid2, ",
" price2_list, ",
" volume, ",
" start_date, ",
" trade_time_increment, ",
" trading_environment ",
")",
" ",
"trade_history.extend(trade_history2) ",
"trade_history = sorted(trade_history, key=lambda x: x.dt)"
],
"language": "python",
"outputs": [],
"prompt_number": 40
},
{
"cell_type": "markdown",
"source": [
"Now that we have a simulated history of trades for two companies and a corresponding trading environment, we can create a dataframe of trades."
]
},
{
"cell_type": "code",
"collapsed": false,
"input": [
"df = pandas.DataFrame(index = ['price', 'volume', 'dt'])",
"for event in trade_history:",
" series = event.as_series()",
" #df.index = df.index.tolist().append(event.sid)",
" #series.name = event.sid",
" df[event.sid] = series"
],
"language": "python",
"outputs": [],
"prompt_number": 92
},
{
"cell_type": "code",
"collapsed": false,
"input": [
"df"
],
"language": "python",
"outputs": [
{
"output_type": "pyout",
"prompt_number": 93,
"text": [
" 133 134",
"price 10.1 12.12",
"volume 100 100",
"dt 2011-04-08 00:00:00+00:00 2011-04-08 00:00:00+00:00"
]
}
],
"prompt_number": 93
},
{
"cell_type": "code",
"collapsed": false,
"input": [
"df_t = df.transpose()",
"df_t"
],
"language": "python",
"outputs": [
{
"output_type": "pyout",
"prompt_number": 94,
"text": [
" price volume dt",
"133 10.1 100 2011-04-08 00:00:00+00:00",
"134 12.12 100 2011-04-08 00:00:00+00:00"
]
}
],
"prompt_number": 94
},
{
"cell_type": "code",
"collapsed": false,
"input": [
"df[133]"
],
"language": "python",
"outputs": [
{
"output_type": "pyout",
"prompt_number": 56,
"text": [
"sid 133",
"volume 100",
"dt 2011-04-08 00:00:00+00:00",
"price 10.1",
"changed NaN",
"Name: 133"
]
}
],
"prompt_number": 56
},
{
"cell_type": "code",
"collapsed": false,
"input": [
"df_t['price']"
],
"language": "python",
"outputs": [
{
"output_type": "pyout",
"prompt_number": 57,
"text": [
"133 10.1",
"134 12.12",
"Name: price"
]
}
],
"prompt_number": 57
},
{
"cell_type": "code",
"collapsed": false,
"input": [
"df_t['price'].max()"
],
"language": "python",
"outputs": [
{
"output_type": "pyout",
"prompt_number": 50,
"text": [
"12.12"
]
}
],
"prompt_number": 50
},
{
"cell_type": "code",
"collapsed": true,
"input": [
"last = trade_history[23].dt"
],
"language": "python",
"outputs": [],
"prompt_number": 51
},
{
"cell_type": "code",
"collapsed": false,
"input": [
"df_t['changed'] = df_t['dt'] > last"
],
"language": "python",
"outputs": [],
"prompt_number": 53
},
{
"cell_type": "code",
"collapsed": false,
"input": [
"df_t"
],
"language": "python",
"outputs": [
{
"output_type": "pyout",
"prompt_number": 54,
"text": [
" sid volume dt price changed",
"133 133 100 2011-04-08 00:00:00+00:00 10.1 True",
"134 134 100 2011-04-08 00:00:00+00:00 12.12 True"
]
}
],
"prompt_number": 54
},
{
"cell_type": "code",
"collapsed": false,
"input": [
"df_t.index"
],
"language": "python",
"outputs": [
{
"output_type": "pyout",
"prompt_number": 59,
"text": [
"Int64Index([133, 134])"
]
}
],
"prompt_number": 59
},
{
"cell_type": "code",
"collapsed": false,
"input": [
"df.index"
],
"language": "python",
"outputs": [
{
"output_type": "pyout",
"prompt_number": 60,
"text": [
"Index([sid, volume, dt, price, changed], dtype=object)"
]
}
],
"prompt_number": 60
},
{
"cell_type": "code",
"collapsed": false,
"input": [
"df.columns"
],
"language": "python",
"outputs": [
{
"output_type": "pyout",
"prompt_number": 61,
"text": [
"Int64Index([133, 134])"
]
}
],
"prompt_number": 61
},
{
"cell_type": "code",
"collapsed": false,
"input": [
"df_t.columns"
],
"language": "python",
"outputs": [
{
"output_type": "pyout",
"prompt_number": 62,
"text": [
"Index([sid, volume, dt, price, changed], dtype=object)"
]
}
],
"prompt_number": 62
},
{
"cell_type": "code",
"collapsed": true,
"input": [],
"language": "python",
"outputs": []
}
]
}
]
}
pavement.py
+1 -72
View File
@@ -1,36 +1,14 @@
import os
import re
import sys
import glob
import time
from distutils.dep_util import newer
#from distutils.extension import Extension
from setuptools.extension import Extension
from paver.easy import options, Bunch, task, needs, path, info
from paver.easy import options, Bunch, task, path
from paver.setuputils import install_distutils_tasks, \
find_packages, find_package_data
from subprocess import call
install_distutils_tasks()
# =========
# Compilers
# =========
try:
from Cython.Compiler.Main import compile
from Cython.Distutils import build_ext
have_cython = True
except ImportError:
have_cython = False
try:
import numpy as np
have_numpy = True
except:
have_numpy = False
# ===================
# Release Information
# ===================
@@ -69,11 +47,6 @@ def parse_requirements(file_name):
requirements.append(line)
return requirements
example = Extension(
"zipline/speedups/example", ["zipline/speedups/example.pyx"],
#include_dirs=[np.get_include()],
)
# ============
# Dependencies
# ============
@@ -88,11 +61,6 @@ tests_require = install_requires + parse_requirements('./etc/requirements_dev.tx
# seutp.py
# ========
if have_numpy and have_cython:
cext = [example]
else:
cext = []
options(
sphinx = Bunch(
builddir="_build",
@@ -127,10 +95,6 @@ options(
'Topic :: Scientific/Engineering :: Information Analysis',
'Topic :: System :: Distributed Computing',
],
ext_modules = cext,
cmdclass = {
'build_ext': build_ext
},
entry_points = {
'console_scripts': [
'zipline = zipline.core.interpreter:main',
@@ -139,41 +103,6 @@ options(
),
)
# ============
# C Extensions
# ============
@task
def clean_inplace():
"""
Remove shared objects and C files from the extension
directory.
"""
for fn in glob.glob(os.path.join(SRC_PATH, 'speedups', '*.c')):
p = path(fn)
p.remove()
for fn in glob.glob(os.path.join(SRC_PATH, 'speedups', '*.so')):
p = path(fn)
p.remove()
@task
def build_cython():
for fn in glob.glob(os.path.join(SRC_PATH, 'speedups', '*.pyx')):
p = path(fn)
modname = p.splitext()[0].basename()
dest = p.splitext()[0] + '.c'
if newer(p.abspath(), dest.abspath()):
info('cython %s -o %s'%(p, dest.basename()))
compile(p.abspath(), full_module_name=modname)
@task
@needs(['build_cython', 'setuptools.command.build_ext'])
def build_ext():
pass
# ======
# Tasks
# ======
tests/client.py
-81
View File
@@ -1,81 +0,0 @@
import logging
import zipline.protocol as zp
from zipline.core.component import Component
from zipline.protocol import CONTROL_PROTOCOL, COMPONENT_TYPE
LOGGER = logging.getLogger('ZiplineLogger')
class TestClient(Component):
def init(self):
self.received_count = 0
self.prev_dt = None
self.result_streams = []
# Maximum outgoing result streams, really shouldn't ever
# need more than 1.
self.max_outgoing = 5
@property
def get_id(self):
return "TEST_CLIENT"
@property
def get_type(self):
return COMPONENT_TYPE.SINK
def open(self):
self.data_feed = self.connect_result()
def result_stream(self, zmq_socket, context=None):
"""
Asynchronously grab a socket to stream results out on.
"""
ctx = context or zmq.Context.instance()
sock = ctx.socket(zmq.PULL)
sock.bind(zmq_socket)
# Add
self.result_streams.append( sock )
def do_work(self):
socks = dict(self.poll.poll(self.heartbeat_timeout))
if socks.get(self.control_in) == self.zmq.POLLIN:
msg = self.control_in.recv()
if socks.get(self.data_feed) == self.zmq.POLLIN:
msg = self.data_feed.recv()
#logger.info('msg:' + str(msg))
if msg == str(CONTROL_PROTOCOL.DONE):
LOGGER.info("Client is DONE!")
self.signal_done()
return
self.received_count += 1
try:
event = self.unframe(msg)
# deserialization error
except zp.INVALID_MERGE_FRAME as exc:
return self.signal_exception(exc)
if self.prev_dt != None:
if not event['dt'] >= self.prev_dt:
raise Exception(
"Message out of order: {date} after {prev}".format(
date = event['dt'], prev = self.prev_dt
)
)
else:
self.prev_dt = event.dt
if self.received_count % 100 == 0:
LOGGER.info("received {n} messages".format(n=self.received_count))
def unframe(self, msg):
return zp.MERGE_UNFRAME(msg)
tests/test_components.py
-384
View File
@@ -1,384 +0,0 @@
import zmq
import pytz
from pprint import pformat as pf
from datetime import datetime, timedelta
from unittest2 import TestCase, skip
from collections import defaultdict
from zipline.gens.composites import date_sorted_sources, merged_transforms
from zipline.core.devsimulator import AddressAllocator
from zipline.gens.transform import Passthrough, StatefulTransform
from zipline.gens.mavg import MovingAverage
from zipline.gens.tradesimulation import TradeSimulationClient as tsc
from zipline.utils.factory import create_trading_environment
from zipline.test_algorithms import TestAlgorithm
from zipline.utils.test_utils import (
setup_logger,
teardown_logger,
create_monitor,
launch_monitor
)
from zipline.core import Component
from zipline.protocol import (
DATASOURCE_FRAME,
DATASOURCE_UNFRAME,
FEED_FRAME,
FEED_UNFRAME,
MERGE_FRAME,
MERGE_UNFRAME,
SIMULATION_STYLE,
PERF_FRAME,
BT_UPDATE_UNFRAME
)
from zipline.gens.tradegens import SpecificEquityTrades
import logbook
log = logbook.Logger('ComponentTestCase')
allocator = AddressAllocator(1000)
class ComponentTestCase(TestCase):
leased_sockets = defaultdict(list)
def setUp(self):
self.zipline_test_config = {
'allocator' : allocator,
'sid' : 133,
'devel' : False,
'results_socket' : allocator.lease(1)[0],
'simulation_style' : SIMULATION_STYLE.FIXED_SLIPPAGE
}
self.ctx = zmq.Context()
setup_logger(self)
count = 250
filter = [2,3]
#Set up source a. One minute between events.
args_a = tuple()
kwargs_a = {
'count' : 2*count,
'sids' : [1,2,3],
'start' : datetime(2002,1,3,15, tzinfo = pytz.utc),
'delta' : timedelta(hours = 6),
'filter' : filter
}
self.source_a = SpecificEquityTrades(*args_a, **kwargs_a)
#Set up source b. Two minutes between events.
args_b = tuple()
kwargs_b = {
'count' : count,
'sids' : [2,3,4],
'start' : datetime(2002,1,3,14, tzinfo = pytz.utc),
'delta' : timedelta(minutes = 5),
'filter' : filter
}
self.source_b = SpecificEquityTrades(*args_b, **kwargs_b)
self.environment = create_trading_environment(year = 2002)
def tearDown(self):
teardown_logger(self)
@skip
def test_source(self):
monitor = create_monitor(allocator)
socket_uri = allocator.lease(1)[0]
count = 100
filter = [1,2,3,4]
#Set up source a. One minute between events.
args_a = tuple()
kwargs_a = {
'sids' : [1,2],
'start' : datetime(2012,6,6,0,tzinfo=pytz.utc),
'delta' : timedelta(minutes = 1),
'filter' : filter,
'count' : count
}
trade_gen = SpecificEquityTrades(*args_a, **kwargs_a)
comp_a = Component(
trade_gen,
monitor,
socket_uri,
DATASOURCE_FRAME,
DATASOURCE_UNFRAME,
"source_a"
)
mon_proc = launch_monitor(monitor)
for event in comp_a:
log.info(event)
# wait for the sending process to exit
comp_a.proc.join()
mon_proc.join()
@skip
def test_sort(self):
monitor = create_monitor(allocator)
socket_uris = allocator.lease(3)
count = 100
filter = [1,2,3,4]
#Set up source a. One minute between events.
args_a = tuple()
kwargs_a = {
'sids' : [1,2],
'start' : datetime(2012,6,6,0,tzinfo=pytz.utc),
'delta' : timedelta(minutes = 1),
'filter' : filter,
'count' : count
}
trade_gen_a = SpecificEquityTrades(*args_a, **kwargs_a)
#Set up source b. Two minutes between events.
args_b = tuple()
kwargs_b = {
'sids' : [2],
'start' : datetime(2012,1,3,15, tzinfo = pytz.utc),
'delta' : timedelta(minutes = 1),
'filter' : filter,
'count' : count
}
trade_gen_b = SpecificEquityTrades(*args_b, **kwargs_b)
#Set up source c. Three minutes between events.
args_c = tuple()
kwargs_c = {
'sids' : [3],
'start' : datetime(2012,1,3,15, tzinfo = pytz.utc),
'delta' : timedelta(minutes = 1),
'filter' : filter,
'count' : count
}
trade_gen_c = SpecificEquityTrades(*args_c, **kwargs_c)
comp_a = Component(
trade_gen_a,
monitor,
socket_uris[0],
DATASOURCE_FRAME,
DATASOURCE_UNFRAME,
trade_gen_a.get_hash()
)
comp_b = Component(
trade_gen_b,
monitor,
socket_uris[1],
DATASOURCE_FRAME,
DATASOURCE_UNFRAME,
trade_gen_b.get_hash()
)
comp_c = Component(
trade_gen_c,
monitor,
socket_uris[2],
DATASOURCE_FRAME,
DATASOURCE_UNFRAME,
trade_gen_c.get_hash()
)
sources = [comp_a, comp_b, comp_c]
sorted_out = date_sorted_sources(*sources)
mon_proc = launch_monitor(monitor)
prev = None
sort_count = 0
for msg in sorted_out:
if prev:
self.assertTrue(msg.dt >= prev.dt, \
"Messages should be in date ascending order")
prev = msg
sort_count += 1
self.assertEqual(count*3, sort_count)
# wait for processes to finish
comp_a.proc.join()
comp_b.proc.join()
comp_c.proc.join()
mon_proc.join()
@skip
def test_full(self):
monitor = create_monitor(allocator)
# ------------------------
# Run sources in dedicated processes
comp_a = Component(
self.source_a,
monitor,
allocator.lease(1)[0],
DATASOURCE_FRAME,
DATASOURCE_UNFRAME,
self.source_a.get_hash()
)
comp_b = Component(
self.source_b,
monitor,
allocator.lease(1)[0],
DATASOURCE_FRAME,
DATASOURCE_UNFRAME,
self.source_b.get_hash()
)
# Date sort the sources, and run the sort in a dedicated
# process
sources = [comp_a, comp_b]
sorted_out = date_sorted_sources(*sources)
sorted = Component(
sorted_out,
monitor,
allocator.lease(1)[0],
FEED_FRAME,
FEED_UNFRAME,
"sort"
)
passthrough = StatefulTransform(Passthrough)
mavg_price = StatefulTransform(
MovingAverage,
['price'],
market_aware = False,
delta=timedelta(minutes = 20)
)
merged_gen = merged_transforms(sorted, passthrough, mavg_price)
merged = Component(
merged_gen,
monitor,
allocator.lease(1)[0],
MERGE_FRAME,
MERGE_UNFRAME,
"merge"
)
algo = TestAlgorithm(2, 10, 100, sid_filter = [2,3])
style = SIMULATION_STYLE.FIXED_SLIPPAGE
trading_client = tsc(algo, self.environment, style)
tsc_gen = trading_client.simulate(merged)
tsc_comp = Component(
tsc_gen,
monitor,
allocator.lease(1)[0],
PERF_FRAME,
BT_UPDATE_UNFRAME,
"tsc"
)
mon_proc = launch_monitor(monitor)
for message in tsc_comp:
log.info(pf(message))
# wait for processes to finish
comp_a.proc.join()
comp_b.proc.join()
sorted.proc.join()
merged.proc.join()
tsc_comp.proc.join()
mon_proc.join()
return
def test_single_thread(self):
#Set up source c. Three minutes between events.
sorted = date_sorted_sources(self.source_a, self.source_b)
passthrough = StatefulTransform(Passthrough)
mavg_price = StatefulTransform(
MovingAverage,
['price'],
market_aware=False,
delta=timedelta(minutes = 20),
)
merged = merged_transforms(sorted, passthrough, mavg_price)
algo = TestAlgorithm(2, 10, 100, sid_filter = [2,3])
style = SIMULATION_STYLE.FIXED_SLIPPAGE
trading_client = tsc(algo, self.environment, style)
for message in trading_client.simulate(merged):
log.info(pf(message))
@skip
def test_compound(self):
monitor = create_monitor(allocator)
sorted_out = date_sorted_sources(self.source_a, self.source_b)
sorted = Component(
sorted_out,
monitor,
allocator.lease(1)[0],
FEED_FRAME,
FEED_UNFRAME,
"feed"
)
passthrough = StatefulTransform(Passthrough)
mavg_price = StatefulTransform(
MovingAverage,
['price'],
market_aware = False,
delta=timedelta(minutes = 20)
)
merged_gen = merged_transforms(sorted, passthrough, mavg_price)
merged = Component(
merged_gen,
monitor,
allocator.lease(1)[0],
MERGE_FRAME,
MERGE_UNFRAME,
"merge"
)
algo = TestAlgorithm(2, 10, 100, sid_filter = [2,3])
style = SIMULATION_STYLE.FIXED_SLIPPAGE
trading_client = tsc(algo, self.environment, style)
tsc_gen = trading_client.simulate(merged)
mon_proc = launch_monitor(monitor)
for message in tsc_gen:
log.info(pf(message))
# wait for processes to finish
sorted.proc.join()
merged.proc.join()
mon_proc.join()
return
tests/test_exception_handling.py
+8 -9
View File
@@ -9,16 +9,15 @@ from zipline.finance.trading import SIMULATION_STYLE
from zipline.core.devsimulator import AddressAllocator
from zipline.lines import SimulatedTrading
from zipline.gens.transform import StatefulTransform
from zipline.gens.tradesimulation import HEARTBEAT_INTERVAL, \
MAX_HEARTBEAT_INTERVALS
from zipline.gens.tradesimulation import MAX_HEARTBEAT_INTERVALS
from zipline.utils.test_utils import \
drain_zipline, \
check, \
setup_logger, \
teardown_logger, \
ExceptionSource, \
ExceptionTransform
from zipline.utils.test_utils import (
drain_zipline,
setup_logger,
teardown_logger,
ExceptionSource,
ExceptionTransform
)
DEFAULT_TIMEOUT = 15 # seconds
EXTENDED_TIMEOUT = 90
tests/test_finance.py
-1
View File
@@ -12,7 +12,6 @@ from nose.tools import timed
import zipline.utils.factory as factory
from zipline.test_algorithms import TestAlgorithm
from zipline.finance.trading import TradingEnvironment
from zipline.core.devsimulator import AddressAllocator
from zipline.lines import SimulatedTrading
tests/test_logger.py
-62
View File
@@ -1,62 +0,0 @@
import logging
import logbook
import uuid
import zmq
from zipline import ndict
from zipline.utils.logger import configure_logging, tail
from zipline.utils.log_utils import ZeroMQLogHandler
from zipline.utils.test_utils import create_receiver, drain_receiver
from unittest2 import TestCase
class LoggerTestCase(TestCase):
def setUp(self):
configure_logging()
self.LOG = logging.getLogger("ZiplineLogger")
def test_log(self):
test_msg = uuid.uuid1().hex
self.LOG.info(test_msg)
logfile = open('/var/log/zipline/zipline.log','r')
with logfile:
last_line = tail(logfile, window=1)
logged_msg = last_line.split(" - ")[1]
self.assertEqual(test_msg, logged_msg)
def test_zmq_handler(self):
socket_addr = 'tcp://127.0.0.1:10000'
ctx = zmq.Context()
socket_push = ctx.socket(zmq.PUSH)
socket_push.connect(socket_addr)
recv = create_receiver(socket_addr, ctx)
zmq_out = ZeroMQLogHandler(
socket = socket_push,
filter = lambda r, h: r.channel in ['test zmq logger'],
context=ctx,
#bubble=False
)
log = logbook.Logger('test zmq logger')
x = ndict({})
x.a = 1
ex = example(133)
with zmq_out.threadbound():
log.info(ex.num)
output, _ = drain_receiver(recv, count=1)
self.assertEqual(output[-1]['prefix'], 'LOG')
self.assertTrue(isinstance(output[-1]['payload']['msg'], basestring))
class example(object):
def __init__(self, num):
self.num = num
tests/test_monitor.py
-28
View File
@@ -1,28 +0,0 @@
from zipline.utils.test_utils import setup_logger, teardown_logger
from unittest2 import TestCase, skip
from zipline.core.monitor import Monitor
class TestMonitor(TestCase):
def setUp(self):
setup_logger(self, '/var/log/qexec/qexec.log')
def tearDown(self):
teardown_logger(self)
def test_init(self):
pub_socket = 'tcp://127.0.0.1:5000'
route_socket = 'tcp://127.0.0.1:5001'
exception_socket = 'tcp://127.0.0.1:5002'
mon = Monitor(pub_socket, route_socket, exception_socket)
mon.manage([])
def test_init_topology(self):
pub_socket = 'tcp://127.0.0.1:5000'
route_socket = 'tcp://127.0.0.1:5001'
exception_socket = 'tcp://127.0.0.1:5002'
mon = Monitor(pub_socket, route_socket, exception_socket)
mon.manage([ 'a', 'b', 'c', 'd' ])
tests/test_sanity.py
-7
View File
@@ -1,7 +0,0 @@
from unittest2 import TestCase
class TestEnviroment(TestCase):
def test_universe(self):
# first order logic is working today. Yay!
self.assertTrue(True != False)
tests/test_sorting.py
+56 -55
View File
@@ -1,22 +1,22 @@
import pytz
from unittest2 import TestCase
from itertools import cycle, chain, izip, izip_longest
from itertools import chain, izip_longest
from datetime import datetime, timedelta
from collections import deque
from zipline import ndict
from zipline.gens.sort import \
date_sort, \
ready, \
done, \
queue_is_ready,\
from zipline.gens.sort import (
date_sort,
ready,
done,
queue_is_ready,
queue_is_done
from zipline.gens.utils import hash_args, alternate, done_message
from zipline.gens.tradegens import date_gen, SpecificEquityTrades
)
from zipline.gens.utils import alternate, done_message
from zipline.gens.tradegens import SpecificEquityTrades
from zipline.gens.composites import date_sorted_sources
import zipline.protocol as zp
class HelperTestCase(TestCase):
@@ -36,7 +36,6 @@ class HelperTestCase(TestCase):
assert queue_is_ready(queue)
assert not queue_is_done(queue)
queue.appendleft(to_dt('DONE'))
assert queue_is_ready(queue)
@@ -81,6 +80,7 @@ class HelperTestCase(TestCase):
assert ready(sources)
assert done(sources)
class DateSortTestCase(TestCase):
def setUp(self):
@@ -99,7 +99,7 @@ class DateSortTestCase(TestCase):
assert m1 == m2
def test_single_source(self):
# Just using the built-in defaults. See
# zipline/gens/tradegens.py
source = SpecificEquityTrades()
@@ -111,28 +111,28 @@ class DateSortTestCase(TestCase):
self.run_date_sort(with_done, expected, [source.get_hash()])
def test_multi_source(self):
filter = [2,3]
filter = [2, 3]
args_a = tuple()
kwargs_a = {
'count' : 100,
'sids' : [1,2,3],
'start' : datetime(2012,1,3,15, tzinfo = pytz.utc),
'delta' : timedelta(minutes = 6),
'filter' : filter
'count': 100,
'sids': [1, 2, 3],
'start': datetime(2012, 1, 3, 15, tzinfo=pytz.utc),
'delta': timedelta(minutes=6),
'filter': filter
}
source_a = SpecificEquityTrades(*args_a, **kwargs_a)
args_b = tuple()
kwargs_b = {
'count' : 100,
'sids' : [2,3,4],
'start' : datetime(2012,1,3,15, tzinfo = pytz.utc),
'delta' : timedelta(minutes = 5),
'filter' : filter
'count': 100,
'sids': [2, 3, 4],
'start': datetime(2012, 1, 3, 15, tzinfo=pytz.utc),
'delta': timedelta(minutes=5),
'filter': filter
}
source_b = SpecificEquityTrades(*args_b, **kwargs_b)
all_events = list(chain(source_a, source_b))
# The expected output is all events, sorted by dt with
@@ -150,76 +150,75 @@ class DateSortTestCase(TestCase):
with_done_b = chain(source_b, [done_message(source_b.get_hash())])
interleaved = alternate(with_done_a, with_done_b)
# Test sort with alternating messages from source_a and
# source_b.
self.run_date_sort(interleaved, expected, source_ids)
source_a.rewind()
source_b.rewind()
with_done_a = chain(source_a, [done_message(source_a.get_hash())])
with_done_b = chain(source_b, [done_message(source_b.get_hash())])
sequential = chain(with_done_a, with_done_b)
# Test sort with all messages from a, followed by all messages
# from b.
self.run_date_sort(sequential, expected, source_ids)
self.run_date_sort(sequential, expected, source_ids)
def test_sort_composite(self):
filter = [1,2]
filter = [1, 2]
#Set up source a. One hour between events.
args_a = tuple()
kwargs_a = {
'count' : 100,
'sids' : [1],
'start' : datetime(2012,6,6,0),
'delta' : timedelta(hours = 1),
'filter' : filter
'count': 100,
'sids': [1],
'start': datetime(2012, 6, 6, 0),
'delta': timedelta(hours=1),
'filter': filter
}
source_a = SpecificEquityTrades(*args_a, **kwargs_a)
#Set up source b. One day between events.
args_b = tuple()
kwargs_b = {
'count' : 50,
'sids' : [2],
'start' : datetime(2012,6,6,0),
'delta' : timedelta(days = 1),
'filter' : filter
'count': 50,
'sids': [2],
'start': datetime(2012, 6, 6, 0),
'delta': timedelta(days=1),
'filter': filter
}
source_b = SpecificEquityTrades(*args_b, **kwargs_b)
#Set up source c. One minute between events.
args_c = tuple()
kwargs_c = {
'count' : 150,
'sids' : [1,2],
'start' : datetime(2012,6,6,0),
'delta' : timedelta(minutes = 1),
'filter' : filter
'count': 150,
'sids': [1, 2],
'start': datetime(2012, 6, 6, 0),
'delta': timedelta(minutes=1),
'filter': filter
}
source_c = SpecificEquityTrades(*args_c, **kwargs_c)
# Set up source d. This should produce no events because the
# internal sids don't match the filter.
args_d = tuple()
kwargs_d = {
'count' : 50,
'sids' : [3],
'start' : datetime(2012,6,6,0),
'delta' : timedelta(minutes = 1),
'filter' : filter
'count': 50,
'sids': [3],
'start': datetime(2012, 6, 6, 0),
'delta': timedelta(minutes=1),
'filter': filter
}
source_d = SpecificEquityTrades(*args_d, **kwargs_d)
sources = [source_a, source_b, source_c, source_d]
hashes = [source.get_hash() for source in sources]
sort_out = date_sorted_sources(*sources)
# Read all the values from sort and assert that they arrive in
# the correct sorting with the expected hash values.
to_list = list(sort_out)
@@ -239,7 +238,8 @@ class DateSortTestCase(TestCase):
assert to_list == expected
def compare_by_dt_source_id(x,y):
def compare_by_dt_source_id(x, y):
if x.dt < y.dt:
return -1
elif x.dt > y.dt:
@@ -252,8 +252,9 @@ def compare_by_dt_source_id(x,y):
else:
return 0
#Alias for ease of use
#Alias for ease of use
comp = compare_by_dt_source_id
def to_dt(msg):
return ndict({'dt': msg})
tests/test_transforms.py
+5 -11
View File
@@ -2,14 +2,11 @@ import pytz
import numpy
from datetime import timedelta, datetime
from collections import defaultdict
from unittest2 import TestCase
from zipline import ndict
from zipline.lines import SimulatedTrading
from zipline.utils.test_utils import setup_logger, teardown_logger
from zipline.utils.test_utils import setup_logger
from zipline.utils.date_utils import utcnow
from zipline.gens.tradegens import SpecificEquityTrades
@@ -156,8 +153,7 @@ class FinanceTransformsTestCase(TestCase):
def test_vwap(self):
vwap = StatefulTransform(
VWAP,
vwap = VWAP(
market_aware = False,
delta = timedelta(days = 2)
)
@@ -180,7 +176,7 @@ class FinanceTransformsTestCase(TestCase):
def test_returns(self):
# Daily returns.
returns = StatefulTransform(Returns, 1)
returns = Returns(1)
transformed = list(returns.transform(self.source))
tnfm_vals = [message.tnfm_value for message in transformed]
@@ -221,8 +217,7 @@ class FinanceTransformsTestCase(TestCase):
def test_moving_average(self):
mavg = StatefulTransform(
MovingAverage,
mavg = MovingAverage(
market_aware = False,
fields = ['price', 'volume'],
delta = timedelta(days = 2),
@@ -263,8 +258,7 @@ class FinanceTransformsTestCase(TestCase):
self.trading_environment
)
stddev = StatefulTransform(
MovingStandardDev,
stddev = MovingStandardDev(
market_aware = False,
delta = timedelta(minutes = 150),
)
vb_suite/run_suite.py
-12
View File
@@ -1,12 +0,0 @@
from vbench.api import BenchmarkRunner
from suite import *
def run_process():
runner = BenchmarkRunner(benchmarks, REPO_PATH, REPO_URL,
BUILD, DB_PATH, TMP_DIR, PREPARE,
run_option='all', start_date=START_DATE,
module_dependencies=dependencies)
runner.run()
if __name__ == '__main__':
run_process()
vb_suite/suite.py
-111
View File
@@ -1,111 +0,0 @@
from vbench.api import Benchmark, GitRepo
from datetime import datetime
import os
modules = ['ziplines']
by_module = {}
benchmarks = []
for modname in modules:
ref = __import__(modname)
by_module[modname] = [v for v in ref.__dict__.values()
if isinstance(v, Benchmark)]
benchmarks.extend(by_module[modname])
for bm in benchmarks:
assert(bm.name is not None)
import getpass
import sys
USERNAME = getpass.getuser()
if sys.platform == 'darwin':
HOME = '/Users/%s' % USERNAME
else:
HOME = '/home/%s' % USERNAME
REPO_PATH = os.path.join(HOME, 'projects/qexec/zipline_repo')
REPO_URL = 'git@github.com:quantopian/zipline.git'
DB_PATH = os.path.join(REPO_PATH, 'vb_suite/benchmarks.db')
TMP_DIR = os.path.join(HOME, 'tmp/vb_zipline')
PREPARE = """
"""
BUILD = """
"""
dependencies = ['zipline_bench_functions.py']
START_DATE = datetime(2011, 6, 1)
repo = GitRepo(REPO_PATH)
RST_BASE = 'source'
# HACK!
#timespan = [datetime(2011, 1, 1), datetime(2012, 1, 1)]
def generate_rst_files(benchmarks):
import matplotlib as mpl
mpl.use('Agg')
import matplotlib.pyplot as plt
vb_path = os.path.join(RST_BASE, 'vbench')
fig_base_path = os.path.join(vb_path, 'figures')
if not os.path.exists(vb_path):
print 'creating %s' % vb_path
os.makedirs(vb_path)
if not os.path.exists(fig_base_path):
print 'creating %s' % fig_base_path
os.makedirs(fig_base_path)
for bmk in benchmarks:
print 'Generating rst file for %s' % bmk.name
rst_path = os.path.join(RST_BASE, 'vbench/%s.txt' % bmk.name)
fig_full_path = os.path.join(fig_base_path, '%s.png' % bmk.name)
# make the figure
plt.figure(figsize=(10, 6))
ax = plt.gca()
bmk.plot(DB_PATH, ax=ax)
start, end = ax.get_xlim()
plt.xlim([start - 30, end + 30])
plt.savefig(fig_full_path, bbox_inches='tight')
plt.close('all')
fig_rel_path = 'vbench/figures/%s.png' % bmk.name
rst_text = bmk.to_rst(image_path=fig_rel_path)
with open(rst_path, 'w') as f:
f.write(rst_text)
with open(os.path.join(RST_BASE, 'index.rst'), 'w') as f:
print >> f, """
Performance Benchmarks
======================
These historical benchmark graphs were produced with `vbench
<http://github.com/pydata/vbench>`__.
.. toctree::
:hidden:
:maxdepth: 3
"""
for modname, mod_bmks in sorted(by_module.items()):
print >> f, ' vb_%s' % modname
modpath = os.path.join(RST_BASE, 'vb_%s.rst' % modname)
with open(modpath, 'w') as mh:
header = '%s\n%s\n\n' % (modname, '=' * len(modname))
print >> mh, header
for bmk in mod_bmks:
print >> mh, bmk.name
print >> mh, '-' * len(bmk.name)
print >> mh, '.. include:: vbench/%s.txt\n' % bmk.name
vb_suite/zipline_bench_functions.py
-43
View File
@@ -1,43 +0,0 @@
try:
from zipline.simulator import AddressAllocator
pass
except Exception, e:
from zipline.core.devsimulator import AddressAllocator
from zipline.lines import SimulatedTrading
allocator = AddressAllocator(1001)
def get_zipline():
zipline_test_config = {
'allocator':allocator,
'sid':133
}
zipline = SimulatedTrading.create_test_zipline(
**zipline_test_config
)
return zipline
def run_basic_zipline():
zipline = get_zipline()
zipline.simulate(blocking=True)
def load_ndict():
from zipline import ndict
nd = ndict({})
keyname = 'a %i'
for i in xrange(1000000):
nd[keyname % i] = i
for i in xrange(1000000):
nd[keyname % i]
def mass_create_ndict():
from zipline import ndict
data = dict(('a %d' % a,a) for a in xrange(1000))
for i in xrange(10000):
ndict(data)
vb_suite/ziplines.py
-27
View File
@@ -1,27 +0,0 @@
from vbench.api import Benchmark
from datetime import datetime
setup = """
from zipline_bench_functions import *
"""
basic_zipline = Benchmark(
'run_basic_zipline()',
setup=setup,
start_date=datetime(2012,5,15),
name='basic_zipline_test'
)
load_ndict = Benchmark(
'load_ndict()',
setup=setup,
start_date=datetime(2012,5,15),
name='load_ndict'
)
mass_create_ndict = Benchmark(
'mass_create_ndict()',
setup=setup,
start_date=datetime(2012,5,1),
name='create_ndict'
)
zipline/core/__init__.py
-6
View File
@@ -1,7 +1 @@
from component import Component
from monitor import Monitor
__all__ = [
Component,
Monitor,
]
zipline/core/component.py
-651
View File
@@ -1,651 +0,0 @@
"""
Contains the base class for all components.
"""
import os
import sys
import uuid
import time
import socket
import logbook
import humanhash
import multiprocessing
from setproctitle import setproctitle
from collections import namedtuple
# pyzmq
import zmq
from zipline.core.monitor import PARAMETERS
from zipline.protocol import (
CONTROL_PROTOCOL,
COMPONENT_STATE,
CONTROL_FRAME,
CONTROL_UNFRAME,
EXCEPTION_FRAME
)
log = logbook.Logger('Component')
class KillSignal(Exception):
def __init__(self):
pass
class ShutdownSignal(Exception):
def __init__(self):
pass
ComponentSocketArgs = namedtuple('ComponentSocketArgs',['uri','style','bind'])
class Component(object):
# ------------
# Construction
# ------------
def __init__(self,
generator,
monitor,
socket_uri,
frame,
unframe,
component_id
):
# -----------------
# Generator
# -----------------
self.generator = generator
self.frame = frame
self.component_id = component_id
# lock for waiting on monitor "GO"
self.waiting = None
# -----------------
# ZMQ properties
# -----------------
self.in_socket_args = ComponentSocketArgs(
uri = socket_uri,
style = zmq.PULL,
bind = False
)
self.out_socket_args = ComponentSocketArgs(
uri = socket_uri,
style = zmq.PUSH,
bind = True
)
self.zmq = None
self.context = None
self.out_socket = None
self.in_socket = None
self.monitor = monitor
self.unframe = unframe
self.prefix = ""
# TODO: state_flag is deprecated, remove
self.state_flag = COMPONENT_STATE.OK
# track time of last ping we received from monitor
self.last_ping = time.time()
# Humanhashes make this way easier to debug because they stick
# in your mind unlike a 32 byte string of random hex.
self.guid = uuid.uuid4()
self.huid = humanhash.humanize(self.guid.hex)
# first, start the generator in its own process. Once
# Monitor says "go", Events from the generator will be
# FRAME'd and PUSH'd to self.socket_uri.
monitor.add_to_topology(self.component_id)
self.proc = multiprocessing.Process(
target=self.loop_send
)
self.proc.start()
# Placeholder for receive generator, which will be
# created in __iter__
self.recv_gen = None
# ------------
# Core Methods
# ------------
def loop_send(self):
"""
The main component loop. This is wrapped inside a
exception reporting context inside of run.
The core logic of the all components is run here.
"""
try:
# The process title so you can watch it in top, ps.
self.prefix = "FORK-"
setproctitle(self.get_id)
log.info("Start %r" % self)
log.info("Pid %s" % os.getpid())
log.info("Group %s" % os.getpgrp())
self.open()
self.signal_ready()
self.lock_ready()
msg = None
for event in self.generator:
if hasattr(event, 'dt') and event.dt == 'DONE':
continue
self.wait_ready()
self.heartbeat()
msg = self.frame(event)
self.out_socket.send(msg)
self.signal_done()
# keep heartbeating until we receive the shutdown
# message from the Monitor (raises a
# ShutdownSignal), or we don't hear from the Monitor
# for MAX_COMPONENT_WAIT.
while True:
self.heartbeat(timeout=1000)
except Exception as exc:
self.handle_exception(exc)
finally:
log.info("Exiting %r" % self)
def create_recv_gen(self):
try:
# return the generator
return self.loop_recv()
except Exception as exc:
self.handle_exception(exc)
finally:
log.info("Created Recv Gen for %r" % self)
def loop_recv(self):
try:
self.open(send=False)
self.signal_ready()
self.lock_ready()
# we block on ready here until monitor sends the GO
# self.wait_ready()
for event in self.gen_from_poller(self.poll, self.in_socket, self.unframe):
yield event
self.signal_done()
except Exception as exc:
self.handle_exception(exc)
finally:
log.info("Exiting %r" % self)
def gen_from_poller(self, poller, in_socket, unframe):
while True:
# Since we will yield None to avoid blocking, we need
# to have a small delay to give the poller a chance
# to receive a message from upstream.
socks = dict(poller.poll(100))
self.heartbeat()
if socks.get(in_socket) == zmq.POLLIN:
message = in_socket.recv()
if message == str(CONTROL_PROTOCOL.DONE):
break
else:
event = unframe(message)
yield event
else:
yield
def handle_exception(self, exc, re_raise=False):
if isinstance(exc, KillSignal):
# if we get a kill signal, forcibly close all the
# sockets.
self.teardown_sockets()
elif isinstance(exc, ShutdownSignal):
# signal from monitor of an orderly shutdown,
# do nothing.
pass
else:
self.signal_exception(exc)
def __iter__(self):
return self
def next(self):
if not self.recv_gen:
self.recv_gen = self.create_recv_gen()
return self.recv_gen.next()
# ----------------------------
# Cleanup & Modes of Failure
# ----------------------------
def teardown_sockets(self):
"""
Close all zmq sockets safely. This is universal, no matter where
this is running it will need the sockets closed.
"""
log.warn("{id} closing all sockets".format(id=self.get_id))
#close all the sockets
for sock in self.sockets:
sock.close()
def shutdown(self):
"""
Clean shutdown.
"""
raise ShutdownSignal()
def kill(self):
"""
Unclean shutdown.
Tear down ( fast ) as a mode of failure in the simulation or on
service halt.
"""
raise KillSignal()
def signal_exception(self, exc=None, scope=None):
"""
All exceptions inside any component should boil back to
this handler.
Will inform the system that the component has failed and how it
has failed.
"""
self.state_flag = COMPONENT_STATE.EXCEPTION
exc_type, exc_value, exc_traceback = sys.exc_info()
# if a downstream component fails, this component may try
# sending when there are zero connections to the socket,
# which will raise ZMQError(EAGAIN). So, it doesn't make
# sense to relay this exception to Monitor and the rest
# of the zipline.
if isinstance(exc, zmq.ZMQError) and exc.errno == zmq.EAGAIN:
log.warn("{id} raised a ZMQError(EAGAIN) not relaying"\
.format(id=self.get_id))
return
# sys.stdout.write(trace)
log.exception("Unexpected error in run for {id}.".format(id=self.get_id))
try:
log.info('{id} sending exception to monitor'\
.format(id=self.get_id))
msg = EXCEPTION_FRAME(
exc_traceback,
exc_type.__name__,
exc_value.message
)
exception_frame = CONTROL_FRAME(
CONTROL_PROTOCOL.EXCEPTION,
msg
)
self.control_out.send(exception_frame, self.zmq.NOBLOCK)
# The monitor should relay the exception back
# to all zipline components. Wait here until the
# notice arrives, and we can assume other zipline
# components have broken out of their message
# loops.
for i in xrange(PARAMETERS.MAX_COMPONENT_WAIT):
self.heartbeat(timeout=1000)
log.warn("{id} never heard back from monitor."\
.format(id=self.get_id))
except KillSignal:
log.info("{id} received confirmation from monitor"\
.format(id=self.get_id))
except:
log.exception("Exception waiting for monitor reply")
# ----------------------
# Internal Maintenance
# ----------------------
def lock_ready(self):
"""
Unlock the component, topology is now ready to run.
"""
self.waiting = True
def unlock_ready(self):
"""
Unlock the component, topology is still pending.
"""
self.waiting = False
def wait_ready(self):
# Implicit side-effect of unlocking the component iff
# the GO message is received from the monitor level.
# This then unlocks the barrier and proceeds to the
# do_work state.
# Poll on a subset of the control protocol while we exist
# in the locked quasimode. Respond to HEARTBEAT and GO
# messages.
start_wait = time.time()
while self.waiting:
socks = dict(self.poll.poll(0))
assert self.control_in, \
'Component does not have a control_in socket'
if socks.get(self.control_in) == zmq.POLLIN:
msg = self.control_in.recv()
event, payload = CONTROL_UNFRAME(msg)
# ====
# Go
# ====
# A distributed lock from the monitor to ensure
# synchronized start.
if event == CONTROL_PROTOCOL.HEARTBEAT:
heartbeat_frame = CONTROL_FRAME(
CONTROL_PROTOCOL.OK,
payload
)
self.control_out.send(heartbeat_frame)
log.info('Prestart Heartbeat ' + self.get_id)
elif event == CONTROL_PROTOCOL.GO:
# Side effectful call from the monitor to unlock
# and begin doing work only when the entire topology
# of the system beings to come online
log.info('Unlocking ' + self.get_id)
self.unlock_ready()
# =========
# Soft Kill
# =========
# Try and clean up properly and send out any reports or
# data that are done during a clean shutdown. Inform the
# monitor that we're done.
elif event == CONTROL_PROTOCOL.SHUTDOWN:
self.shutdown()
break
# =========
# Hard Kill
# =========
# Just exit.
elif event == CONTROL_PROTOCOL.KILL:
self.kill()
break
elif time.time() - start_wait > PARAMETERS.MAX_COMPONENT_WAIT:
log.info('No go signal from monitor, %s exiting' \
% self.get_id)
self.kill()
break
def heartbeat(self, timeout=0):
# wait for synchronization reply from the host
socks = dict(self.poll.poll(timeout))
# ----------------
# Control Dispatch
# ----------------
assert self.control_in, 'Component does not have a control_in socket'
if socks.get(self.control_in) == zmq.POLLIN:
msg = self.control_in.recv()
event, payload = CONTROL_UNFRAME(msg)
# ===========
# Heartbeat
# ===========
# The monitor will send out a single number packed in
# a CONTROL_FRAME with ``heartbeat`` event every
# (n)-seconds. The component then has n seconds to
# respond to it. If not then it will be considered as
# malfunctioning or maybe CPU bound.
if event == CONTROL_PROTOCOL.HEARTBEAT:
# Heart outgoing
heartbeat_frame = CONTROL_FRAME(
CONTROL_PROTOCOL.OK,
payload
)
self.last_ping = float(payload)
# Echo back the heartbeat identifier to tell the
# monitor that this component is still alive and
# doing work
self.control_out.send(heartbeat_frame)
# =========
# Soft Kill
# =========
# Try and clean up properly and send out any reports or
# data that are done during a clean shutdown. Inform the
# monitor that we're done.
elif event == CONTROL_PROTOCOL.SHUTDOWN:
self.shutdown()
# =========
# Hard Kill
# =========
# Just exit.
elif event == CONTROL_PROTOCOL.KILL:
self.kill()
# In case we didn't receive a ping, send a pre-emptive
# pong to the monitor.
elif time.time() - self.last_ping > 2:
# send a ping ahead of schedule
pre_pong = time.time()
heartbeat_frame = CONTROL_FRAME(
CONTROL_PROTOCOL.OK,
str(pre_pong)
)
# Echo back the heartbeat identifier to tell the
# monitor that this component is still alive and
# doing work
self.control_out.send(heartbeat_frame, self.zmq.NOBLOCK)
self.last_ping = pre_pong
elif time.time() - self.last_ping > PARAMETERS.MAX_COMPONENT_WAIT:
# monitor is gone without sending the shutdown
# signal, do a hard exit.
self.kill()
def signal_ready(self):
log.info(self.get_id + ' is ready')
frame = CONTROL_FRAME(
CONTROL_PROTOCOL.READY,
''
)
self.control_out.send(frame)
def signal_done(self):
"""
Notify down stream components that we're done.
"""
self.state_flag = COMPONENT_STATE.DONE
# notify internal work loop that we're done
self.done = True # TODO: use state flag
if self.out_socket:
msg = zmq.Message(str(CONTROL_PROTOCOL.DONE))
self.out_socket.send(msg)
# notify monitor we're done
done_frame = CONTROL_FRAME(
CONTROL_PROTOCOL.DONE,
''
)
self.control_out.send(done_frame)
log.info("[%s] sent control done" % self.get_id)
# -----------
# Messaging
# -----------
def open(self, send=True):
"""
Open the connections needed to start doing work.
Perform any setup that must be done within process.
"""
self.sockets = []
self.zmq = zmq
self.context = self.zmq.Context()
self.poll = self.zmq.Poller()
self.setup_control()
if send:
self.out_socket = self.open_socket(self.out_socket_args)
self.sockets.extend([self.out_socket])
else:
self.in_socket = self.open_socket(self.in_socket_args)
self.sockets.extend([self.in_socket])
def open_socket(self, sock_args):
if sock_args.bind:
return self.bind_socket(sock_args)
else:
return self.connect_socket(sock_args)
def bind_socket(self, sock_args):
if sock_args.style == zmq.PULL:
return self.bind_pull_socket(sock_args.uri)
if sock_args.style == zmq.PUSH:
return self.bind_push_socket(sock_args.uri)
if sock_args.style == zmq.PUB:
return self.bind_pub_socket(sock_args.uri)
raise Exception("Invalid socket arguments")
def connect_socket(self, sock_args):
if sock_args.style == zmq.PULL:
return self.connect_pull_socket(sock_args.uri)
if sock_args.style == zmq.PUSH:
return self.connect_push_socket(sock_args.uri)
if sock_args.style == zmq.SUB:
return self.connect_sub_socket(sock_args.uri)
raise Exception("Invalid socket arguments")
def bind_push_socket(self, addr):
push_socket = self.context.socket(self.zmq.PUSH)
push_socket.bind(addr)
self.sockets.append(push_socket)
return push_socket
def connect_pull_socket(self, addr):
pull_socket = self.context.socket(self.zmq.PULL)
pull_socket.connect(addr)
self.sockets.append(pull_socket)
self.poll.register(pull_socket, self.zmq.POLLIN)
return pull_socket
def bind_pull_socket(self, addr):
pull_socket = self.context.socket(self.zmq.PULL)
pull_socket.bind(addr)
self.poll.register(pull_socket, self.zmq.POLLIN)
self.sockets.append(pull_socket)
return pull_socket
def connect_push_socket(self, addr):
push_socket = self.context.socket(self.zmq.PUSH)
push_socket.connect(addr)
self.sockets.append(push_socket)
return push_socket
def setup_control(self):
"""
Set up the control socket. Used to monitor the overall status
of the simulation and to forcefully tear down the simulation in
case of a failure.
"""
self.control_out = self.monitor.message_sender(
identity = self.get_id,
context = self.context,
)
self.control_in = self.monitor.message_listener(
context = self.context
)
self.poll.register(self.control_in, self.zmq.POLLIN)
self.sockets.extend([self.control_in, self.control_out])
# ---------------------
# Description and Debug
# ---------------------
@property
def get_id(self):
"""
The time invariant name for this component.
Must be unique within this zipline.
"""
return self.prefix + self.component_id
def get_hash(self):
return self.component_id
def debug(self):
"""
Debug information about the component.
"""
return {
'id' : self.get_id ,
'huid' : self.huid ,
'host' : socket.gethostname() ,
'pid' : os.getpid() ,
'memaddress' : hex(id(self)) ,
'ready' : self.successful() ,
'successful' : self.ready() ,
}
def __repr__(self):
"""
Return a useful string representation of the component to
indicate its type, unique identifier, and computational context
identifier name.
"""
return "<{name} {uuid} at {host} {pid} {pointer}>".format(
name = self.get_id ,
uuid = self.guid ,
host = socket.gethostname() ,
pid = os.getpid() ,
pointer = hex(id(self)) ,
)
zipline/core/devsimulator.py
+3 -3
View File
@@ -4,7 +4,6 @@ See :py:method""
"""
import logbook
import threading
log = logbook.Logger('Dev Simulator')
@@ -14,6 +13,7 @@ THE DEVSIMULATOR IS DEPRECATED, IT WILL NOT BEHAVE LIKE ANY OTHER
SYSTEM USED IN TESTS OR IN PRODUCTION
"""
class AddressAllocator(object):
"""
Produces a iterator of 10000 sockets to allocate as needed.
@@ -28,9 +28,9 @@ class AddressAllocator(object):
]
def lease(self, n):
sockets = self.sockets[self.idx:self.idx+n]
sockets = self.sockets[self.idx: self.idx + n]
self.idx += n
return sockets
def reaquire(self, *conn):
pass
pass
zipline/core/monitor.py
-654
View File
@@ -1,654 +0,0 @@
import os
import zmq
import sys
import time
import itertools
import logbook
from setproctitle import setproctitle
from signal import SIGHUP, SIGINT
from collections import Counter
from zipline.protocol import (
CONTROL_PROTOCOL,
CONTROL_FRAME,
CONTROL_UNFRAME,
CONTROL_STATES,
INVALID_CONTROL_FRAME
)
from zipline.utils.protocol_utils import ndict
INIT, SOURCES_READY, RUNNING, TERMINATE = CONTROL_STATES
CONTROLLER_TRANSITIONS = frozenset([
(-1 , INIT),
(INIT , SOURCES_READY),
(SOURCES_READY , RUNNING),
(INIT , TERMINATE), # pseudo failure mode
(SOURCES_READY , TERMINATE), # pseudo failure mode
(RUNNING , TERMINATE),
])
class UnknownChatter(Exception):
def __init__(self, name):
self.named = name
def __str__(self):
return """Component calling itself "%s" talking on unexpected channel""" % self.named
log = logbook.Logger('Monitor')
# The scalars determining the timing of the monitor behavior for
# the system.
PARAMETERS = ndict(dict(
# time Monitor will wait for a heartbeat, in seconds
GENERATIONAL_PERIOD = 20,
# time Component will wait for GO and for a heartbeat before
# timing out.
MAX_COMPONENT_WAIT = 25,
ALLOWED_SKIPPED_HEARTBEATS = 10,
ALLOWED_INVALID_HEARTBEATS = 3,
PRESTART_HEARBEATS = 3,
SOURCES_START_HEARTBEATS = 3,
SYSTEM_TIMEOUT = 50,
))
class Monitor(object):
"""
A N to M messaging system for inter component communication.
:param pub_socket: Socket to publish messages, the starting
point of :func message_listener: .
:param route_socket: Socket to listen for status updates for
the individual components.
:func message_sender: .
"""
# Turn on debug for verbose logging of the system.
debug = True
period = PARAMETERS.GENERATIONAL_PERIOD
def __init__(
self,
pub_socket,
route_socket,
exception_socket,
send_sighup=False):
self.nosignals = False
self.context = None
self.zmq = None
self.zmq_poller = None
self.running = False
self.alive = False
self.tracked = set()
self.finished = set()
self.responses = set()
self.ctime = 0
self.tic = time.time()
self.freeform = False
self._state = -1
self.associated = []
self.pub_socket = pub_socket
self.route_socket = route_socket
self.exception_socket = exception_socket
self.missed_beats = Counter()
# start with an empty topology
self.topology = set([])
self.send_sighup = send_sighup
if self.send_sighup:
log.info("Request to send sighup/sigint")
def init_zmq(self):
self.zmq = zmq
self.context = self.zmq.Context()
self.zmq_poller = self.zmq.Poller
return
def add_to_topology(self, component_id):
add = set([component_id, "FORK-" + component_id])
self.topology.update(add)
def freeze_topology(self):
if isinstance(self.topology, frozenset):
return
# we've been incrementally adding components.
# time to freeze.
self.manage(self.topology)
def manage(self, topology):
"""
Give the controller a set set of components to manage and
a set of state transitions for the entire system.
"""
# A freeform topology is where we heartbeat with anything
# that shows up.
if topology == 'freeform':
self.freeform = True
self.topology = frozenset([])
else:
self.freeform = False
self.topology = frozenset(topology)
self.alive = True
@property
def state(self):
#log.info('returned %s' % self._state)
return self._state
@state.setter
def state(self, new):
old = self._state
if (old, new) in CONTROLLER_TRANSITIONS:
self._state = new
log.info("State Transition : %s -> %s" % (old, self._state))
else:
raise RuntimeError("Invalid State Transition : %s -> %s" %(old, new))
def run(self):
self.freeze_topology()
self.running = True
self.init_zmq()
setproctitle('Monitor')
self.state = CONTROL_STATES.INIT
# TODO: keep the exitfunc? the corresponding override on clean
# exit is commented out currently.
#
# Interpreter SIDE EFFECT
# -----------------------
# The last breathe of the interpreter will assume that we've
# failed unless we specify otherwise.
log.info('registering exit function')
sys.exitfunc = self.signal_interrupt
# We overload this if ( and only if ) the topology exits
# cleanly. This prevents failure modes where the monitor
# dies.
try:
return self._poll() # use a python loop
except KeyboardInterrupt:
log.info('Shutdown event loop')
def log_status(self):
"""
Snapshot of the tracked components at every period.
"""
#log.info("Tracking component : %s" % ([c for c in self.tracked],))
pass
def replay_errors(self):
"""
Replay the errors in the order they were reported to the
controller.
"""
return [ a for a in sorted(self.replay_errors.keys())]
# -------------
# Publications
# -------------
def send_go(self):
go_frame = CONTROL_FRAME(
CONTROL_PROTOCOL.GO,
''
)
self.pub.send(go_frame)
def send_heart(self):
if not self.running:
return
heartbeat_frame = CONTROL_FRAME(
CONTROL_PROTOCOL.HEARTBEAT,
str(self.ctime)
)
self.pub.send(heartbeat_frame)
def send_hardkill(self):
if not self.running:
return
kill_frame = CONTROL_FRAME(
CONTROL_PROTOCOL.KILL,
''
)
self.pub.send(kill_frame)
def send_softkill(self):
if not self.running:
return
soft_frame = CONTROL_FRAME(
CONTROL_PROTOCOL.SHUTDOWN,
''
)
self.pub.send(soft_frame)
# -----------
# Event Loops
# -----------
def _poll(self):
assert self.route_socket
assert self.pub_socket
assert self.topology,\
""""Must define topology to monitor, call setup_controller() on
your Zipline. """
# -- Publish --
# =============
self.pub = self.context.socket(self.zmq.PUB)
self.pub.bind(self.pub_socket)
self.pub.setsockopt(zmq.LINGER, 0)
# -- Router --
# =============
self.router = self.context.socket(self.zmq.ROUTER)
self.router.bind(self.route_socket)
self.router.setsockopt(zmq.LINGER, 0)
# -- Exception Out --
# ===================
self.ex_out = self.context.socket(self.zmq.PUSH)
self.ex_out.connect(self.exception_socket)
poller = self.zmq.Poller()
poller.register(self.router, self.zmq.POLLIN)
#poller.register(self.cancel, self.zmq.POLLIN)
self.associated += [self.pub, self.router]
# TODO: actually do this
self.state = CONTROL_STATES.SOURCES_READY
self.state = CONTROL_STATES.RUNNING
buffer = []
# ===================
# Heartbeat Iteration
# ===================
for i in itertools.count(0):
self.log_status()
# Reset the responses for this cycle
self.responses = set()
# broadcast the heartbeat packet
self.ctime = time.time()
self.send_heart()
# ==============
# Hearbeat Cycle
# ==============
initializing = len(self.tracked) == 0 and len(self.finished) == 0
# Wait the responses
while self.alive:
socks = dict(poller.poll(0))
tic = time.time()
if socks.get(self.router) == self.zmq.POLLIN:
rawmessage = self.router.recv()
if rawmessage:
buffer.append(rawmessage)
try:
if not self.router.getsockopt(self.zmq.RCVMORE):
self.handle_recv(buffer[:])
buffer = []
except INVALID_CONTROL_FRAME:
log.error('Invalid frame', rawmessage)
pass
# We break out of this loop if the time between
# sending and receiving the heartbeat is more
# than our poll period.
if tic - self.ctime > self.period:
log.info("heartbeat loop timedout: %s" % (tic - self.ctime))
log.info(repr(self.responses))
break
# if this is the first time heartbeating, break
# out early if we get everything tracked no need
# to hold out for the full heartbeat.
if initializing and not self.freeform:
if len(self.responses) == len(self.topology):
log.info("breaking out of initial heartbeat")
break
# Break out if the entire topology told us its DONE
if len(self.finished) == len(self.topology):
break
# ================
# Heartbeat Stats
# ================
complete = self.beat()
# ================
# Topology Status
# ================
# Has the entire topology told us its DONE
done = len(self.finished) == len(self.topology)
# Has the entire topology shown up to the party
complete = len(self.tracked) == len(self.topology)
if complete:
self.send_go()
log.info('Heartbeat (%s, %s)' % (done, complete))
# ================
# Exit Strategies
# ================
# Will also fall out of loop when done, if using
# non-freeform topology
if done:
log.info('Entire topology exited cleanly')
self.shutdown()
# Noop exit func
#sys.exitfunc = lambda: None
# Send SIGHUP to buritto
self.signal_hangup()
if not self.alive:
log.info('Breaking out of Monitor Loop')
break
def signal_hangup(self):
"""
A clean exit, inform the burrito ( and arbiter ) that
we're good. The topology exited cleanly and we can prove
it.
"""
if not self.send_sighup:
log.warning("Skipping SIGHUP")
return
ppid = os.getppid()
log.warning("Sending SIGHUP")
os.kill(ppid, SIGHUP)
def signal_interrupt(self):
"""
Send a SIGINT in the error mode that the monitor's
interpreter exits. If the monitor dies the system is
considered a failure.
"""
if not self.send_sighup:
log.warning("Skipping SIGINT")
return
ppid = os.getpid()
log.warning("Sending SIGINT")
os.kill(ppid, SIGINT)
def beat(self):
"""
The tracking logic of the system. It's the "stethoscope"
that inspects to the heartbeats in a generation and
infers the state of the system from the responses.
"""
# These the set overloaded operations
# A & B ~ set.intersection
# A - B ~ set.difference
# * good - Components we are currently tracking and who just sent
# us back the right response.
# * bad - Components we are currently tracking but who did not
# send us back a response.
# * new - Components we haven't heard from yet, but sent back the
# right response.
# * finished - Components we were tracking but have now
# finished, when this set goes to zero this
# triggers the end of the topology.
good = self.tracked & self.responses
bad = self.tracked - good - self.finished
new = self.responses - good - self.finished
for component in new:
self.new(component)
for component in bad:
self.timed_out(component)
missing = self.topology - self.tracked - self.finished
for component in missing:
if self.debug:
log.info('Missing component %r' % component)
for component in self.tracked:
if component not in self.topology:
log.info('Uninvited component %r' % component)
# --------------
# Init Handlers
# --------------
def new_universal(self):
pass
# The various "states of being that a component can inform us
# of
def new(self, component):
if self.state is CONTROL_STATES.TERMINATE:
return
if component in self.finished:
#log.info("Got heartbeat from supposedly finished component")
return
log.info('Now Tracking "%s" ' % component)
universal = self.new_universal
init_handlers = {}
if component in (self.topology - self.finished) or self.freeform:
init_handlers.get(component, universal)()
self.tracked.add(component)
else:
# Some sort of socket collision has occurred, this is
# a very bad failure mode.
raise UnknownChatter(component)
# ------------------
# Epic Fail Handling
# ------------------
def timed_out(self, component):
if self.state is CONTROL_STATES.TERMINATE:
return
if component in (self.topology - self.finished) or self.freeform:
log.warning('Component "%s" missed heartbeat' % component)
# we treat a time out as a severe failure, and
# conduct a rapid shutdown
self.kill()
# -------------------
# Completion Handling
# -------------------
def done(self, component):
self.finished.add(component)
self.tracked.discard(component)
log.info('Component "%s" finished.' % component)
# --------------
# Error Handling
# --------------
def exception(self, component, exception_data):
log.error('Component in exception state: %s. Shutting down system and sending exception data to listeners.'\
% component)
# Send the exception message out to listeners.
self.ex_out.send(exception_data)
# An exception in one component is treated as a hard
# failure, and we conduct a rapid shutdown.
self.kill()
# -----------------
# Protocol Handling
# -----------------
def handle_recv(self, msg):
"""
Check for proper framing at the transport layer.
Seperates the proper frames from anything else that might
be coming over the wire.
"""
identity = msg[0] # identity of the socket
id, status = CONTROL_UNFRAME(msg[1])
# I'm alive, condemned to be a free process in the cold
# cold dark absurd Zipline universe.
if id is CONTROL_PROTOCOL.READY:
self.responses.add(identity)
return
# The heartbeat love song between a component and the
# controller
if id is CONTROL_PROTOCOL.OK:
if status == str(self.ctime):
# Go to your bosom; knock there, and ask your heart what
# it doth know...
self.responses.add(identity)
elif float(status) < self.ctime:
# False face must hide what the false heart doth know.
log.warning('Delayed heartbeat received: %s' % msg)
elif float(status) > self.ctime:
# Pre-emptive heartbeat from the component
# log.info("pre-emptive pong: %s" % msg)
self.responses.add(identity)
else:
# Otherwise its something weird and we don't know
# what to do so just say so, probably line noise
# from ZeroMQ
# What's in a name? that which we call a rose...
log.error("Weird heartbeat packet happened: %s" % msg)
return
# A component is telling us it failed, and how
if id is CONTROL_PROTOCOL.EXCEPTION:
# status should be a msgpack emitted from
# EXCEPTION_FRAME
try:
exception_data = status
self.exception(identity, exception_data)
except:
# if an exception occurs when we try to handle
# the exception, signal the parent that we need
# to go down
# TODO: should we attempt to call self.exception?
log.exception("Unexpected exception sending exception data")
self.kill()
return
# A component is telling us its done with work and won't
# be talking to us anymore
if id is CONTROL_PROTOCOL.DONE:
self.done(identity)
return
# -------------------
# Hooks for Endpoints
# -------------------
# These are all connects so no complex allocation logic is
# needed. Dealers and Subscribers can all come and go as a
# function of time without impacting flow of the whole
# system.
def message_sender(self, identity, context = None):
"""
Spin off a socket used for sending messages to this
controller.
"""
if not context:
context = self.zmq.Context.instance()
s = context.socket(zmq.DEALER)
s.setsockopt(zmq.IDENTITY, identity)
s.connect(self.route_socket)
self.associated.append(s)
return s
def message_listener(self, context = None):
"""
Spin off a socket used for receiving messages from this
controller.
"""
if not context:
context = self.zmq.Context.instance()
s = context.socket(zmq.SUB)
s.connect(self.pub_socket)
s.setsockopt(zmq.SUBSCRIBE, '')
self.associated.append(s)
return s
def kill(self):
"""Aggressively exit the whole zipline.
"""
if self.state is CONTROL_STATES.TERMINATE:
return
log.info('Hard Shutdown')
self.send_hardkill()
self.state = CONTROL_STATES.TERMINATE
self.alive = False
# send burrito an interrupt, instructing it to kill all
# child processes assocated with this zipline.
time.sleep(3)
self.signal_interrupt()
def shutdown(self):
if self.state is CONTROL_STATES.TERMINATE:
return
log.info('Soft Shutdown')
self.send_softkill()
self.state = CONTROL_STATES.TERMINATE
self.alive = False
zipline/exceptions.py
-5
View File
@@ -1,5 +0,0 @@
from utils.exception_utils import CustomException
class ComponentNoInit(CustomException):
argmap = ('classname',)
message = """Class {classname} does not define an init method."""
zipline/finance/performance.py
-2
View File
@@ -125,8 +125,6 @@ import datetime
import pytz
import math
import zmq
import zipline.protocol as zp
import zipline.finance.risk as risk
zipline/gens/composites.py
+7 -28
View File
@@ -1,13 +1,11 @@
import datetime
from itertools import tee, starmap, chain
from collections import namedtuple
from itertools import tee, chain
from zipline.gens.tradegens import SpecificEquityTrades
from zipline.gens.utils import roundrobin, hash_args, done_message
from zipline.gens.utils import roundrobin, done_message
from zipline.gens.sort import date_sort
from zipline.gens.merge import merge
from zipline.gens.transform import StatefulTransform
def date_sorted_sources(*sources):
"""
Takes an iterable of sources, generating namestrings and
@@ -31,6 +29,7 @@ def date_sorted_sources(*sources):
return date_sort(stream_in, names)
def merged_transforms(sorted_stream, *transforms):
"""
A generator that takes the expected output of a date_sort, pipes
@@ -45,7 +44,7 @@ def merged_transforms(sorted_stream, *transforms):
assert isinstance(transform, StatefulTransform)
transform.merged = True
transform.sequential = False
# Generate expected hashes for each transform
namestrings = [tnfm.get_hash() for tnfm in transforms]
@@ -69,6 +68,7 @@ def merged_transforms(sorted_stream, *transforms):
# Return the merged events.
return add_done(dt_aliased)
def sequential_transforms(stream_in, *transforms):
"""
Apply each transform in transforms sequentially to each event in stream_in.
@@ -81,28 +81,6 @@ def sequential_transforms(stream_in, *transforms):
for tnfm in transforms:
tnfm.sequential = True
tnfm.merged = False
# Recursively apply all transforms to the stream.
stream_out = reduce(lambda stream, tnfm: tnfm.transform(stream),
transforms,
stream_in)
dt_aliased = alias_dt(stream_out)
return add_done(dt_aliased)
def sequential_transforms_dict(stream_in, transforms):
"""
Apply each transform in transforms sequentially to each event in stream_in.
Each transform application will add a new entry indexed to the transform's
hash string.
"""
assert isinstance(transforms, dict)
for tnfm in transforms.itervalues():
tnfm.forward_all = False
tnfm.update_in_place = False
tnfm.append_value = True
# Recursively apply all transforms to the stream.
stream_out = reduce(lambda stream, tnfm: tnfm.transform(stream),
@@ -121,6 +99,7 @@ def alias_dt(stream_in):
message['datetime'] = message['dt']
yield message
# Add a done message to a stream.
def add_done(stream_in):
return chain(stream_in, [done_message('Composite')])
zipline/gens/examples.py
-100
View File
@@ -1,100 +0,0 @@
import pytz
import time
from time import sleep
from pprint import pprint as pp
from datetime import datetime, timedelta
from itertools import izip
from zipline.utils.factory import create_trading_environment
from zipline.test_algorithms import TestAlgorithm
from zipline.gens.composites import date_sorted_sources, merged_transforms, sequential_transforms
from zipline.gens.tradegens import SpecificEquityTrades
from zipline.gens.mavg import MovingAverage
from zipline.gens.transform import Passthrough, StatefulTransform
from zipline.gens.tradesimulation import TradeSimulationClient as tsc
import zipline.protocol as zp
if __name__ == "__main__":
filter = [2,3]
#Set up source a. Six minutes between events.
args_a = tuple()
kwargs_a = {
'count' : 1000,
'sids' : [1,2,3],
'start' : datetime(2012,1,3,15, tzinfo = pytz.utc),
'delta' : timedelta(minutes = 6),
'filter' : filter
}
source_a = SpecificEquityTrades(*args_a, **kwargs_a)
source_a_prime = SpecificEquityTrades(*args_a, **kwargs_a)
#Set up source b. Five minutes between events.
args_b = tuple()
kwargs_b = {
'count' : 1000,
'sids' : [2,3,4],
'start' : datetime(2012,1,3,14, tzinfo = pytz.utc),
'delta' : timedelta(minutes = 5),
'filter' : filter
}
source_b = SpecificEquityTrades(*args_b, **kwargs_b)
source_b_prime = SpecificEquityTrades(*args_b, **kwargs_b)
sorted = date_sorted_sources(source_a, source_b)
sorted_prime = date_sorted_sources(
source_a_prime,
source_b_prime
)
passthrough = StatefulTransform(Passthrough)
mavg_price = StatefulTransform(
MovingAverage,
timedelta(minutes = 20),
['price']
)
passthrough_prime = StatefulTransform(Passthrough)
mavg_price_prime = StatefulTransform(
MovingAverage,
timedelta(minutes = 20),
['price']
)
merged = merged_transforms(sorted, passthrough, mavg_price)
start = time.time()
for message in merged:
assert 1 + 1 == 2
stop = time.time()
merge_time = stop - start
print "Merge time: %s" % str(merge_time)
sequential = sequential_transforms(
sorted_prime,
passthrough_prime,
mavg_price_prime
)
start = time.time()
for message in sequential:
assert 1 + 1 == 2
stop = time.time()
seq_time = stop - start
print "Sequential time: %s" % str(seq_time)
print "Merge/Seq: %s" % (str(merge_time/seq_time))
# merged = merged_transforms(sorted, passthrough, mavg_price)
# algo = TestAlgorithm(2, 10, 100, sid_filter = [2,3])
# environment = create_trading_environment(year = 2012)
# style = zp.SIMULATION_STYLE.FIXED_SLIPPAGE
# trading_client = tsc(algo, environment, style)
# for message in trading_client.simulate(merged):
# pp(message)
zipline/gens/mavg.py
+14 -11
View File
@@ -1,9 +1,9 @@
from numbers import Number
from datetime import datetime, timedelta
from collections import defaultdict
from zipline import ndict
from zipline.gens.transform import EventWindow
from zipline.gens.transform import EventWindow, TransformMeta
class MovingAverage(object):
"""
@@ -12,8 +12,9 @@ class MovingAverage(object):
averages over any number of distinct fields (For example, we can
maintain a sid's average volume as well as its average price.)
"""
__metaclass__ = TransformMeta
def __init__(self, fields, market_aware, days = None, delta = None):
def __init__(self, fields, market_aware, days=None, delta=None):
self.fields = fields
self.market_aware = market_aware
@@ -30,7 +31,7 @@ class MovingAverage(object):
else:
assert self.delta and not self.days, \
"Non-market-aware mode requires a timedelta."
# No way to pass arguments to the defaultdict factory, so we
# need to define a method to generate the correct EventWindows.
self.sid_windows = defaultdict(self.create_window)
@@ -40,12 +41,12 @@ class MovingAverage(object):
Factory method for self.sid_windows.
"""
return MovingAverageEventWindow(
self.fields,
self.market_aware,
self.days,
self.fields,
self.market_aware,
self.days,
self.delta
)
def update(self, event):
"""
Update the event window for this event's sid. Return an ndict
@@ -57,6 +58,7 @@ class MovingAverage(object):
window.update(event)
return window.get_averages()
class MovingAverageEventWindow(EventWindow):
"""
Iteratively calculates moving averages for a particular sid over a
@@ -102,7 +104,7 @@ class MovingAverageEventWindow(EventWindow):
# Averages are None by convention if we have no ticks.
if len(self.ticks) == 0:
return 0.0
# Calculate and return the average. len(self.ticks) is O(1).
else:
return self.totals[field] / len(self.ticks)
@@ -111,7 +113,7 @@ class MovingAverageEventWindow(EventWindow):
"""
Return an ndict of all our tracked averages.
"""
out = ndict()
out = ndict()
for field in self.fields:
out[field] = self.average(field)
return out
@@ -124,4 +126,5 @@ class MovingAverageEventWindow(EventWindow):
assert event.has_key(field), \
"Event missing [%s] in MovingAverageEventWindow" % field
assert isinstance(event[field], Number), \
"Got %s for %s in MovingAverageEventWindow" % (event[field], field)
"Got %s for %s in MovingAverageEventWindow" % (event[field],
field)
zipline/gens/returns.py
+7 -3
View File
@@ -1,10 +1,14 @@
from zipline.gens.transform import TransformMeta
from collections import defaultdict, deque
class Returns(object):
"""
Class that maintains a dictionary from sids to the sid's
closing price N trading days ago.
"""
__metaclass__ = TransformMeta
def __init__(self, days):
self.days = days
self.mapping = defaultdict(self._create)
@@ -13,8 +17,8 @@ class Returns(object):
"""
Update and return the calculated returns for this event's sid.
"""
assert event.has_key('dt')
assert event.has_key('price')
assert 'dt' in event
assert 'price' in event
tracker = self.mapping[event.sid]
tracker.update(event)
@@ -23,6 +27,7 @@ class Returns(object):
def _create(self):
return ReturnsFromPriorClose(self.days)
class ReturnsFromPriorClose(object):
"""
Records the last N closing events for a given security as well as the
@@ -69,6 +74,5 @@ class ReturnsFromPriorClose(object):
change = event.price - last_close
self.returns = change / last_close
# the current event is now the last_event
self.last_event = event
zipline/gens/stddev.py
+2 -1
View File
@@ -4,7 +4,7 @@ from collections import defaultdict
from math import sqrt
from zipline import ndict
from zipline.gens.transform import EventWindow
from zipline.gens.transform import EventWindow, TransformMeta
class MovingStandardDev(object):
"""
@@ -13,6 +13,7 @@ class MovingStandardDev(object):
standard deviation of all events falling within the specified
window.
"""
__metaclass__ = TransformMeta
def __init__(self, market_aware, days = None, delta = None):
zipline/gens/tradegens.py
+1 -33
View File
@@ -214,36 +214,4 @@ class DataFrameSource(SpecificEquityTrades):
# Return the filtered event stream.
return _generator()
# !!!!!!! Deprecated for now !!!!!!!!!
def RandomEquityTrades(object):
def __init__(self):
# We shouldn't get any positional args.
assert args == ()
self.count = config.get('count', 500)
self.sids = config.get('sids', [1,2])
self.filter = config.get('filter')
dates = fuzzy_dates(count)
prices = mock_prices(count, rand = True)
volumes = mock_volumes(count, rand = True)
sids = cycle(sids)
arg_gen = izip(sids, prices, volumes, dates)
unfiltered = (create_trade(*args) for args in arg_gen)
if filter:
filtered = ifilter(lambda event: event.sid in filter, unfiltered)
else:
filtered = unfiltered
return filtered
# if __name__ == "__main__":
# import nose.tools; nose.tools.set_trace()
# trades = SpecificEquityTrades(filter = [1])
return _generator()
zipline/gens/tradesimulation.py
+1 -4
View File
@@ -1,14 +1,11 @@
import signal
from logbook import Logger, Processor
from datetime import datetime, timedelta
from numbers import Integral
from datetime import datetime
from itertools import groupby
from zipline import ndict
from zipline.utils.timeout import Heartbeat, Timeout
from zipline.gens.transform import StatefulTransform
from zipline.finance.trading import TransactionSimulator
from zipline.finance.performance import PerformanceTracker
from zipline.utils.log_utils import stdout_only_pipe
zipline/gens/transform.py
+33 -9
View File
@@ -2,19 +2,16 @@
Generator versions of transforms.
"""
import types
import pytz
import logbook
from copy import deepcopy
from datetime import datetime, timedelta
from collections import deque, defaultdict
from numbers import Number
from datetime import datetime
from collections import deque
from abc import ABCMeta, abstractmethod
from zipline import ndict
from zipline.utils.tradingcalendar import non_trading_days
from zipline.gens.utils import assert_sort_unframe_protocol, \
assert_transform_protocol, hash_args
from zipline.gens.utils import assert_sort_unframe_protocol, hash_args
log = logbook.Logger('Transform')
@@ -29,6 +26,21 @@ class Passthrough(object):
def update(self, event):
pass
class TransformMeta(type):
"""
Metaclass that automatically packages a class inside of
StatefulTransform on initialization. Specifically, if Foo is a
class with its __metaclass__ attribute set to TransformMeta, then
calling Foo(*args, **kwargs) will return StatefulTransform(Foo,
*args, **kwargs) instead of an instance of Foo. (Note that you can
still recover an instance of a "raw" Foo by introspecting the
resulting StatefulTransform's 'state' field.
"""
def __call__(cls, *args, **kwargs):
return StatefulTransform(cls, *args, **kwargs)
class StatefulTransform(object):
"""
Generic transform generator that takes each message from an
@@ -55,11 +67,23 @@ class StatefulTransform(object):
self.merged = True
# Create an instance of our transform class.
self.state = tnfm_class(*args, **kwargs)
if isinstance(tnfm_class, TransformMeta):
# Classes derived TransformMeta have their __call__
# attribute overridden. Since this is what is usually
# used to create an instance, we have to delegate the
# responsibility of creating an instance to
# TransformMeta's parent class, which is 'type'. This is
# what is implicitly done behind the scenes by the python
# interpreter for most classes anyway, but here we have to
# be explicit because we've overridden the method that
# usually resolves to our super call.
self.state = super(TransformMeta, tnfm_class).__call__(*args, **kwargs)
# Normal object instantiation.
else:
self.state = tnfm_class(*args, **kwargs)
# Create the string associated with this generator's output.
self.namestring = tnfm_class.__name__ + hash_args(*args, **kwargs)
log.info('StatefulTransform [%s] initialized' % self.namestring)
def get_hash(self):
return self.namestring
@@ -126,7 +150,7 @@ class StatefulTransform(object):
log.info('Finished StatefulTransform [%s]' % self.get_hash())
class EventWindow:
class EventWindow(object):
"""
Abstract base class for transform classes that calculate iterative
metrics on events within a given timedelta. Maintains a list of
zipline/gens/utils.py
+39 -24
View File
@@ -8,26 +8,29 @@ from itertools import izip_longest
from zipline import ndict
from zipline.protocol import DATASOURCE_TYPE
def mock_raw_event(sid, dt):
event = {
'sid' : sid,
'dt' : dt,
'price' : 1.0,
'volume' : 1
'sid': sid,
'dt': dt,
'price': 1.0,
'volume': 1
}
return event
def mock_done(id):
return ndict({
'dt' : "DONE",
"source_id" : id,
'tnfm_id' : id,
'dt': "DONE",
"source_id": id,
'tnfm_id': id,
'tnfm_value': None,
'type' : DATASOURCE_TYPE.DONE
'type': DATASOURCE_TYPE.DONE
})
done_message = mock_done
def alternate(g1, g2):
"""Specialized version of roundrobin for just 2 generators."""
for e1, e2 in izip_longest(g1, g2):
@@ -36,6 +39,7 @@ def alternate(g1, g2):
if e2 != None:
yield e2
def roundrobin(sources, namestrings):
"""
Takes N generators, pulling one element off each until all inputs
@@ -56,32 +60,36 @@ def roundrobin(sources, namestrings):
yield done_message(namestring)
del mapping[namestring]
def hash_args(*args, **kwargs):
"""Define a unique string for any set of representable args."""
arg_string = '_'.join([str(arg) for arg in args])
kwarg_string = '_'.join([str(key) + '=' + str(value) for key, value in kwargs.iteritems()])
kwarg_string = '_'.join([str(key) + '=' + str(value)
for key, value in kwargs.iteritems()])
combined = ':'.join([arg_string, kwarg_string])
hasher = md5()
hasher.update(combined)
return hasher.hexdigest()
def create_trade(sid, price, amount, datetime, source_id = "test_factory"):
def create_trade(sid, price, amount, datetime, source_id="test_factory"):
row = ndict({
'source_id' : source_id,
'type' : DATASOURCE_TYPE.TRADE,
'sid' : sid,
'dt' : datetime,
'price' : price,
'close' : price,
'open' : price,
'low' : price * .95,
'high' : price * 1.05,
'volume' : amount
'source_id': source_id,
'type': DATASOURCE_TYPE.TRADE,
'sid': sid,
'dt': datetime,
'price': price,
'close': price,
'open': price,
'low': price * .95,
'high': price * 1.05,
'volume': amount
})
return row
def sum_true(bool_iterable):
"""
Takes an iterable of boolean values and returns the number of
@@ -102,6 +110,7 @@ def assert_datasource_protocol(event):
assert isinstance(event.dt, datetime)
assert event.dt.tzinfo == pytz.utc
def assert_trade_protocol(event):
"""Assert that an event meets the protocol for datasource TRADE outputs."""
assert_datasource_protocol(event)
@@ -113,32 +122,38 @@ def assert_trade_protocol(event):
assert isinstance(event.volume, numbers.Integral)
assert isinstance(event.dt, datetime)
def assert_datasource_unframe_protocol(event):
"""Assert that an event is valid output of zp.DATASOURCE_UNFRAME."""
assert isinstance(event, ndict)
assert isinstance(event.source_id, basestring)
assert event.type in DATASOURCE_TYPE
assert event.has_key('dt')
assert 'dt' in event
def assert_sort_protocol(event):
"""Assert that an event is valid input to zp.FEED_FRAME."""
assert isinstance(event, ndict)
assert isinstance(event.source_id, basestring)
assert event.type in DATASOURCE_TYPE
assert event.has_key('dt')
assert 'dt' in event
def assert_sort_unframe_protocol(event):
"""Same as above."""
assert isinstance(event, ndict)
assert isinstance(event.source_id, basestring)
assert event.type in DATASOURCE_TYPE
assert event.has_key('dt')
assert 'dt' in event
def assert_transform_protocol(event):
"""Transforms should return an ndict to be merged by merge."""
assert isinstance(event, ndict)
def assert_merge_protocol(tnfm_ids, message):
"""Merge should output an ndict with a field for each id in its transform set."""
"""Merge should output an ndict with a field for each id
in its transform set."""
assert isinstance(message, ndict)
assert set(tnfm_ids) == set(message.keys())
zipline/gens/vwap.py
+11 -9
View File
@@ -1,20 +1,21 @@
from numbers import Number
from datetime import datetime, timedelta
from collections import defaultdict
from zipline import ndict
from zipline.gens.transform import EventWindow
from zipline.gens.transform import EventWindow, TransformMeta
class VWAP(object):
"""
Class that maintains a dictionary from sids to VWAPEventWindows.
"""
__metaclass__ = TransformMeta
def __init__(self, market_aware, delta=None, days=None):
self.market_aware = market_aware
self.delta = delta
self.days = days
# Market-aware mode only works with full-day windows.
if self.market_aware:
assert self.days and not self.delta,\
@@ -28,13 +29,13 @@ class VWAP(object):
# No way to pass arguments to the defaultdict factory, so we
# need to define a method to generate the correct EventWindows.
self.sid_windows = defaultdict(self.create_window)
def create_window(self):
"""Factory method for self.sid_windows."""
return VWAPEventWindow(
self.market_aware,
days = self.days,
delta = self.delta
self.market_aware,
days=self.days,
delta=self.delta
)
def update(self, event):
@@ -48,6 +49,7 @@ class VWAP(object):
window.update(event)
return window.get_vwap()
class VWAPEventWindow(EventWindow):
"""
Iteratively maintains a vwap for a single sid over a given
@@ -69,7 +71,7 @@ class VWAPEventWindow(EventWindow):
def handle_remove(self, event):
self.flux -= event.volume * event.price
self.totalvolume -= event.volume
def get_vwap(self):
"""
Return the calculated vwap for this sid.
zipline/gens/zmq_gens.py
-18
View File
@@ -1,18 +0,0 @@
import zmq
import zipline.protocol as zp
def gen_from_zmq(poller, unframe, namestring):
"""
A generator that takes an initialized zmq poller and yields
messages from the poller until it gets a zp.CONTROL_PROTOCOL.DONE.
"""
while True:
message = poller.recv()
# Done protocol should now be a message type so that
# done messages can also have source_ids.
if message.type == zp.CONTROL_PROTOCOL.DONE:
yield done_message(message.source_id)
break
else:
yield unframe(message)
zipline/gens/zmqgen.py
-19
View File
@@ -1,19 +0,0 @@
import zmq
import zipline.protocol as zp
def gen_from_pull_socket(socket_uri, context, unframe):
"""
A generator that takes a socket_uri, and yields
messages from the poller until it gets a zp.CONTROL_PROTOCOL.DONE.
"""
pull_socket = context.socket(zmq.PULL)
pull_socket.connect(socket_uri)
poller = zmq.Poller()
poller.register(pull_socket, zmq.POLLIN)
return gen_from_poller(poller, pull_socket, unframe)
# this generator needs to know about the source_ids coming in via
# the poller, and need to yield DONE messages for each
# source_id.
zipline/lines.py
+50 -44
View File
@@ -68,26 +68,21 @@ from setproctitle import setproctitle
from zipline.test_algorithms import TestAlgorithm
from zipline.finance.trading import SIMULATION_STYLE
from zipline.utils.log_utils import ZeroMQLogHandler, stdout_only_pipe
from zipline.utils.log_utils import ZeroMQLogHandler
from zipline.utils import factory
from zipline.test_algorithms import TestAlgorithm
from zipline.gens.composites import \
date_sorted_sources, merged_transforms, sequential_transforms
from zipline.gens.transform import Passthrough, StatefulTransform
from zipline.gens.composites import (
date_sorted_sources,
sequential_transforms
)
from zipline.gens.tradesimulation import TradeSimulationClient as tsc
from logbook import Logger, NestedSetup, Processor
from logbook import Logger
import zipline.protocol as zp
log = Logger('Lines')
class CancelSignal(Exception):
def __init__(self):
pass
class SimulatedTrading(object):
def __init__(self,
@@ -103,7 +98,8 @@ class SimulatedTrading(object):
self.date_sorted = date_sorted_sources(*sources)
self.transforms = transforms
# Formerly merged_transforms.
self.with_tnfms = sequential_transforms(self.date_sorted, *self.transforms)
self.with_tnfms = sequential_transforms(self.date_sorted,
*self.transforms)
self.trading_client = tsc(algorithm, environment, style)
self.gen = self.trading_client.simulate(self.with_tnfms)
self.results_uri = results_socket_uri
@@ -150,7 +146,7 @@ class SimulatedTrading(object):
"Results socket must exist to stream results"
try:
for event in self.gen:
if event.has_key('daily_perf'):
if 'daily_perf' in event:
msg = zp.PERF_FRAME(event)
else:
msg = zp.RISK_FRAME(event)
@@ -171,6 +167,8 @@ class SimulatedTrading(object):
def close(self):
log.info("Closing Simulation: {id}".format(id=self.sim_id))
if self.results_socket:
self.results_socket.close()
if self.proc and self.send_sighup:
ppid = os.getppid()
if self.success:
@@ -181,12 +179,7 @@ class SimulatedTrading(object):
os.kill(ppid, SIGINT)
def handle_exception(self, exc):
if isinstance(exc, CancelSignal):
# signal from monitor of an orderly shutdown,
# do nothing.
pass
else:
self.signal_exception(exc)
self.signal_exception(exc)
def signal_exception(self, exc=None):
"""
@@ -227,8 +220,8 @@ class SimulatedTrading(object):
# bubbled. Since we do not want user logs in our system
# logs, we set bubble to False.
self.zmq_out = ZeroMQLogHandler(
socket = self.results_socket,
filter = lambda r, h: r.channel in ['Print', 'AlgoLog'],
socket=self.results_socket,
filter=lambda r, h: r.channel in ['Print', 'AlgoLog'],
bubble=False
)
@@ -242,12 +235,6 @@ class SimulatedTrading(object):
else:
return []
def __iter__(self):
return self
def next(self):
return self.gen.next()
@staticmethod
def create_test_zipline(**config):
"""
@@ -268,7 +255,8 @@ class SimulatedTrading(object):
is the source, with daily frequency in trades.
- simulation_style: optional parameter that configures the
:py:class:`zipline.finance.trading.TransactionSimulator`. Expects
a SIMULATION_STYLE as defined in :py:mod:`zipline.finance.trading`
a SIMULATION_STYLE as defined in
:py:mod:`zipline.finance.trading`
- transforms: optional parameter that provides a list
of StatefulTransform objects.
"""
@@ -283,22 +271,22 @@ class SimulatedTrading(object):
#--------------------
# Trading Environment
#--------------------
if config.has_key('environment'):
if 'environment' in config:
trading_environment = config['environment']
else:
trading_environment = factory.create_trading_environment()
if config.has_key('order_count'):
if 'order_count' in config:
order_count = config['order_count']
else:
order_count = 100
if config.has_key('order_amount'):
if 'order_amount' in config:
order_amount = config['order_amount']
else:
order_amount = 100
if config.has_key('trade_count'):
if 'trade_count' in config:
trade_count = config['trade_count']
else:
# to ensure all orders are filled, we provide one more
@@ -309,22 +297,21 @@ class SimulatedTrading(object):
if not simulation_style:
simulation_style = SIMULATION_STYLE.FIXED_SLIPPAGE
zmq_context = config.get('zmq_context', None)
simulation_id = config.get('simulation_id', 'test_simulation')
results_socket_uri = config.get('results_socket_uri', None)
zmq_context = config.get('zmq_context', None)
simulation_id = config.get('simulation_id', 'test_simulation')
results_socket_uri = config.get('results_socket_uri', None)
#-------------------
# Trade Source
#-------------------
sids = [sid]
#-------------------
if config.has_key('trade_source'):
if 'trade_source' in config:
trade_source = config['trade_source']
else:
trade_source = factory.create_daily_trade_source(
sids,
sid_list,
trade_count,
trading_environment
trading_environment,
concurrent=concurrent_trades
)
#-------------------
@@ -335,7 +322,7 @@ class SimulatedTrading(object):
#-------------------
# Create the Algo
#-------------------
if config.has_key('algorithm'):
if 'algorithm' in config:
test_algo = config['algorithm']
else:
test_algo = TestAlgorithm(
@@ -361,6 +348,7 @@ class SimulatedTrading(object):
return sim
class SimulatedTradingLite(object):
"""
SimulatedTrading without multiprocess and without zmq.
@@ -373,15 +361,33 @@ class SimulatedTradingLite(object):
algorithm,
environment,
style):
"""
@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
@style - protocol.SIMULATION_STYLE
"""
self.date_sorted = date_sorted_sources(*sources)
self.transforms = transforms
# Formerly merged_transforms.
self.with_tnfms = sequential_transforms(self.date_sorted, *self.transforms)
self.with_tnfms = sequential_transforms(self.date_sorted,
*self.transforms)
self.trading_client = tsc(algorithm, environment, style)
self.gen = self.trading_client.simulate(self.with_tnfms)
def get_results(self):
return self.gen
@@ -389,4 +395,4 @@ class SimulatedTradingLite(object):
return self
def next(self):
return self.gen.next()
return self.gen.next()
zipline/profile/__init__.py
View File
zipline/profile/prof.py
-104
View File
@@ -1,104 +0,0 @@
"""
Viscosity - Tools for benchmarking ZeroMQ data flow.
"""
import time as timer
import logging
import pycounters
from contextlib import contextmanager, nested
from pycounters import base
from pycounters.shortcuts import frequency, time
from pycounters import shortcuts, reporters, start_auto_reporting, register_reporter
from pycounters import shortcuts,reporters,report_value, output_report, \
counters, register_counter, _reporting_decorator_context_manager
JSONFile = "counters.json"
logger = logging.getLogger('simple_example')
logger.setLevel(logging.DEBUG)
ch = logging.StreamHandler()
ch.setLevel(logging.DEBUG)
logger.addHandler(ch)
reporter = reporters.JSONFileReporter(output_file=JSONFile)
logreport = reporters.LogReporter(logger)
register_reporter(logreport)
register_reporter(reporter)
class timecontext:
def __init__(self, name):
self.name = name
def __enter__(self):
cntr = base.GLOBAL_REGISTRY.get_counter(self.name, throw=False)
if not cntr:
counter = counters.AverageTimeCounter(self.name)
register_counter(counter)
self.tic = timer.time()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if not exc_type:
shortcuts.value(self.name, timer.time() - self.tic)
class ttimecontext:
def __init__(self, name):
self.name = name
def __enter__(self):
counter = base.GLOBAL_REGISTRY.get_counter(self.name, throw=False)
if not counter:
counter = counters.EventCounter(self.name)
counter.value = 0
register_counter(counter)
self.counter = counter
self.tic = timer.time()
return self
def __exit__(self, exc_type, exc_val, exc_tb):
if not exc_type:
val = (timer.time() - self.tic)
if not self.counter.value:
self.counter.value = long(0.0)
self.counter.value += val
class occurancecontext:
def __init__(self, name):
self.name = name
def __enter__(self):
cntr = base.GLOBAL_REGISTRY.get_counter(self.name, throw=False)
if not cntr:
cntr = counters.TotalCounter(self.name)
counter = counters.TotalCounter(self.name)
register_counter(counter)
return self
def __exit__(self, exc_type, exc_val, exc_tb):
shortcuts.value(self.name, 1)
if __name__ == '__main__':
with timecontext('average time'):
for i in xrange(5):
x = [2] * 1000
timer.sleep(0.01)
with occurancecontext('totalcount'):
for i in xrange(5):
x = [2] * 1000
with ttimecontext('total time'):
for i in xrange(5):
x = [2] * 1000
timer.sleep(1)
pycounters.output_report()
zipline/profile/prof_yappi.py
-93
View File
@@ -1,93 +0,0 @@
from __future__ import division
import logging
from zipline.core.devsimulator import AddressAllocator
import zipline.finance
from zipline.optimize.factory import create_predictable_zipline
import pandas as pd
import numpy as np
import os.path
def convert_ystats(ystats):
"""Convert yappi.get_stats().func_stats object to pandas
DataFrame.
"""
func_names = [os.path.split(item[0])[-1] for item in ystats]
ncall = [float(item[1]) for item in ystats]
ttot = [float(item[2]) for item in ystats]
tsub = [float(item[3]) for item in ystats]
tavg = [float(item[4]) for item in ystats]
stats = pd.DataFrame({'ncall': ncall, 'ttot': ttot, 'tsub': tsub, 'tavg': tavg}, index=func_names)
return stats
allocator = AddressAllocator(1000)
config = { 'allocator' :allocator,
'sid' :133,
'trade_count' :5000,
'amplitude' :30,
'base_price' :50
}
LOGGER = logging.getLogger('ZiplineLogger')
import yappi
def gen_single_stats(func, *args, **kwargs):
"""Profile func(*args, **kwargs) with yappi.
Returns DataFrame of statistics.
"""
yappi.start()
func(*args, **kwargs)
yappi.stop()
return convert_ystats(yappi.get_stats().func_stats)
def gen_avg_stats(func, runs=1, *args, **kwargs):
"""Profile func(*args, **kwargs) with yappi. Runs multiple times at computes the average.
Returns DataFrame of average statistics.
"""
avg_stats = pd.concat([gen_single_stats() for i in range(runs)], keys=range(runs))
grouped = avg_stats.groupby(level=1)
return grouped.aggregate(np.mean)
def run_updown(fname='before_stats.csv'):
"""Profile a zipline with the UpDown tradesource (does not require
DB access) and the buy/sell algorithm (requires no
computation).
Saves output statics under fname.
Returns Dataframe of statistics.
"""
zp, _ = create_predictable_zipline(config, simulate=False)
stats = gen_single_stats(zp.simulate, blocking=True)
stats.to_csv(fname)
return stats
def calc_speedup(before='before_stats.csv', after='after_stats.csv'):
"""Calculate speed-up between two previously run and saved
statistics under filename before and after.
Prints DataFrame of top 30 speed-ups and top 30 slow-downs.
"""
old = pd.DataFrame.from_csv(before)
new = pd.DataFrame.from_csv(after)
speed_up = old / new
speed_up = speed_up.fillna(1)
speed_up = speed_up.sort(column='ttot', ascending=False)
slow_down = speed_up.sort(column='ttot', ascending=True)
print speed_up[:30]
print slow_down[:30]
if __name__ == '__main__':
run_updown()
yappi.print_stats(sort_type=yappi.SORTTYPE_TTOT)
zipline/speedups/__init__.py
View File
zipline/speedups/example.pyx
-3
View File
@@ -1,3 +0,0 @@
from libc.stdio cimport printf
printf("Hello World!")
zipline/toys/__init__.py
View File
zipline/transitions.py
-69
View File
@@ -1,69 +0,0 @@
import types
from collections import Container, Hashable, Callable
class Any(object): pass
class Workflow(Container, Callable):
def __init__(self, states, transitions, initial_state):
self.simple = set()
self.complx = []
if isinstance(states[0], tuple):
self.groups = {b for _,b in states}
else:
self.groups = set()
matcher = lambda b: lambda f,t : t == b
for (a, b) in transitions.itervalues():
if a is Any:
self.complx.append(matcher(b))
if isinstance(a, Hashable) and isinstance(b, Hashable):
self.simple.add((a,b))
def __call__(self, **kwargs):
if 'group' in kwargs:
return self.groups
def __contains__(self, state):
if state in self.simple:
return True
for match in self.complx:
if match(*state):
return True
else:
return False
class WorkflowMeta(type):
"""
Base metaclass component workflows.
"""
def __new__(cls, name, mro, attrs):
base = 'Component'
state = attrs.get('states', None)
transitions = attrs.get('transitions', None)
initial_state = attrs.get('initial_state', None)
if not 'abstract' in attrs:
if attrs.get('workflow'):
raise RuntimeError('`workflow` is a reserved attribute.')
if not state:
raise RuntimeError('Must specify states')
if not transitions:
raise RuntimeError('Must specify transitions')
if not transitions:
raise RuntimeError('Must specify initial_state')
new_class = super(WorkflowMeta, cls).__new__(cls, name, mro, attrs)
if not 'abstract' in attrs:
new_class.workflow = Workflow(state, transitions, initial_state)
return new_class
zipline/utils/exception_utils.py
-16
View File
@@ -1,16 +0,0 @@
from textwrap import dedent
class CustomException(Exception):
argmap = {0: 'classname'}
def __init__(self, *args):
self.args = args
def format(self):
assert len(self.args) == len(self.argmap), \
"""Wrong number of arguments passed to custom exception %s.""" \
% self.__class__
return self.message.format(**dict(zip(self.argmap, self.args)))
def __str__(self):
return dedent(self.format()).strip('\n')
zipline/utils/factory.py
-1
View File
@@ -12,7 +12,6 @@ from datetime import datetime, timedelta
import zipline.finance.risk as risk
import zipline.protocol as zp
from zipline.gens.tradegens import RandomEquityTrades
from zipline.gens.tradegens import SpecificEquityTrades
from zipline.gens.utils import create_trade
from zipline.finance.trading import TradingEnvironment
zipline/utils/logger.py
-52
View File
@@ -1,52 +0,0 @@
"""
Small classes to assist with timezone calculations, LOGGER configuration,
and other common operations.
"""
# DEPRECATED DO NOT USE
import logging
import logging.config
from os.path import join, abspath, dirname
def configure_logging():
logging.config.fileConfig(
logger_path(),
disable_existing_loggers = False
)
def logger_path():
import zipline
log_path = dirname(abspath(zipline.__file__))
return join(log_path, 'logging.cfg')
# utility for tailing a log file.
def tail( f, window=20 ):
"""
from
http://stackoverflow.com/questions/136168/get-last-n-lines-of-a-file- \
with-python-similar-to-tail
"""
BUFSIZ = 1024
f.seek(0, 2)
bytes = f.tell()
size = window
block = -1
data = []
while size > 0 and bytes > 0:
if (bytes - BUFSIZ > 0):
# Seek back one whole BUFSIZ
f.seek(block*BUFSIZ, 2)
# read BUFFER
data.append(f.read(BUFSIZ))
else:
# file too small, start from begining
f.seek(0,0)
# only read what was not read
data.append(f.read(bytes))
linesFound = data[-1].count('\n')
size -= linesFound
bytes -= BUFSIZ
block -= 1
return '\n'.join(''.join(data).splitlines()[-window:])
zipline/utils/protocol_utils.py
-1
View File
@@ -2,7 +2,6 @@ import copy
import pandas
from ctypes import Structure, c_ubyte
from collections import MutableMapping
from itertools import izip
def Enum(*options):
"""
zipline/utils/serial.py
-56
View File
@@ -1,56 +0,0 @@
"""
Format serializer for Zipline.
Because I'm opinionated about how you should send things over
ZeroMQ. :)
"""
import zlib
import hmac
import base64
#import blosc
import cPickle as pickle
# Pickle does the equivelant of builtin ``eval``. Be afraid, be
# very afraid.
def send_zipped_pickle(socket, obj, flags=0, protocol=-1):
"""
Pickle an object, and zip the pickle before sending it.
"""
p = pickle.dumps(obj, protocol)
z = zlib.compress(p)
return socket.send(z, flags=flags)
def recv_zipped_pickle(socket, flags=0, protocol=-1):
"""
Unpickle and uncompress a received object.
"""
z = socket.recv(flags)
p = zlib.uncompress(z)
return pickle.loads(p, protocol=protocol)
# Cryptographically secure wire protocol for ZeroMQ Using HMAC.
# Compare byte strings, backported from Python 3.
def byte_eq(a, b):
return not sum(0 if x==y else 1 for x, y in zip(a, b)) and len(a) == len(b)
def send_secure(socket, data, key, flags=0):
msg = base64.b64encode(data)
sig = base64.b64encode(hmac.new(key, msg).digest())
return socket.send(bytes('!') + sig + bytes('?') + msg, flags=flags)
def recv_secure(socket, data, key, flags):
data = socket.recv(flags=flags)
try:
sig, msg = data.split(bytes('?'), 1)
except ValueError:
raise Exception('Invalid signature/message pair.')
if byte_eq(sig[1:], base64.b64encode(hmac.new(key, msg).digest())):
return base64.b64decode(msg)
else:
raise Exception('Cryptographically invalid message received')
zipline/utils/test_utils.py
+14 -23
View File
@@ -7,16 +7,18 @@ import blist
from zipline.utils.date_utils import EPOCH
from itertools import izip
from logbook import FileHandler
from zipline.core.monitor import Monitor
def setup_logger(test, path='/var/log/zipline/zipline.log'):
test.log_handler = FileHandler(path)
test.log_handler.push_application()
def teardown_logger(test):
test.log_handler.pop_application()
test.log_handler.close()
def check_list(test, a, b, label):
test.assertTrue(isinstance(a, (list, blist.blist)))
test.assertTrue(isinstance(b, (list, blist.blist)))
@@ -33,8 +35,8 @@ def check_dict(test, a, b, label):
if key in ['progress']:
continue
test.assertTrue(a.has_key(key), "missing key at: " + label + "." + key)
test.assertTrue(b.has_key(key), "missing key at: " + label + "." + key)
test.assertTrue(key in a, "missing key at: " + label + "." + key)
test.assertTrue(key in b, "missing key at: " + label + "." + key)
a_val = a[key]
b_val = b[key]
check(test, a_val, b_val, label + "." + key)
@@ -62,6 +64,7 @@ def check(test, a, b, label=None):
else:
test.assertEqual(a, b, "mismatch on path: " + label)
def drain_zipline(test, zipline, p_blocking=False):
assert test.ctx, "method expects a valid zmq context"
assert test.zipline_test_config, "method expects a valid test config"
@@ -86,15 +89,17 @@ def drain_zipline(test, zipline, p_blocking=False):
return output, transaction_count
def create_receiver(socket_addr, ctx):
receiver = ctx.socket(zmq.PULL)
receiver.bind(socket_addr)
return receiver
def drain_receiver(receiver, count=None):
output = []
transaction_count = 0
transaction_count = 0
msg_counter = 0
while True:
msg = receiver.recv()
@@ -119,7 +124,9 @@ def drain_receiver(receiver, count=None):
def assert_single_position(test, zipline, blocking=False):
output, transaction_count = drain_zipline(test, zipline, p_blocking=blocking)
output, transaction_count = drain_zipline(test,
zipline,
p_blocking=blocking)
test.assertEqual(output[-1]['prefix'], 'DONE')
test.assertEqual(
@@ -152,30 +159,13 @@ def launch_component(component):
proc.start()
return proc
def launch_monitor(monitor):
proc = multiprocessing.Process(target=monitor.run)
proc.start()
return proc
def create_monitor(allocator):
sockets = allocator.lease(3)
mon = Monitor(
# pub socket
sockets[0],
# route socket
sockets[1],
# exception socket to match tradesimclient's result
# socket, because we want to relay exceptions to the
# same listener
sockets[2],
# this controller is expected to run in a test, so no
# need to signal the parent process on success or error.
send_sighup=False
)
return mon
class ExceptionSource(object):
def __init__(self):
@@ -190,6 +180,7 @@ class ExceptionSource(object):
def next(self):
5 / 0
class ExceptionTransform(object):
def __init__(self):
zipline/utils/zmq_utils.py
-154
View File
@@ -1,154 +0,0 @@
"""
Misc ZeroMQ experimental tools
"""
import gevent
import msgpack
import numpy
from numpy import dtype
from pandas import DataFrame
from gevent_zeromq import zmq
from contextlib import closing
class ZmqDone(Exception):
def __init__(self, socket, frame):
self.ident = socket.identity
self.frame = str(frame)
def __str__(self):
return 'Socket ( %s ) finished with frame ( %s )' % \
( self.ident, self.frame )
class zs(object):
"""
A wrapper for the *very* common pattern of reading from a
upstream socket until you get a DONE or EXCEPTION frame.
# Eliminates all the boilerplate serialization logic
# and error handling cases into 3 lines.
halts = (ERROR_FRAME, CLOSE_FRAME)
stream = zs(socket, halts)
stream.on_error(YouFailAtFailing)
for msg in stream:
print msg
"""
def __init__(self, socket, halts, srl=msgpack):
self._socket = socket
self.exc_case = halts[0]
self.done_case = halts[1]
self.loads = srl.loads
self.halt_method = 'exception'
self.exception = ZmqDone
self.function = None
def __iter__(self):
self.last = msg = self.loads(self._socket.recv())
if msg == self.exc_case:
return self.halt()
if msg == self.done_case:
raise StopIteration
yield msg
def last(self):
return self.last
def halt(self):
if self.halt_method == 'exception':
raise self.exception
elif self.halt_method == 'function':
return self.function()
def on_error(self, callee):
if isinstance(callee, Exception):
self.halt_method = 'exception'
self.exception = callee
else:
self.halt_method = 'function'
self.function = callee
def ZmqConsole(sock_typ, socket_addr, sock_conn=None, context=None):
"""
A utility to drop into a ZeroMQ pdb console and inspect
messages as they come through. If you just want to pipe to
stdout, don't use this.
"""
context = context or zmq.Context.instance()
socket = context.socket(zmq.PULL)
socket.bind(socket_addr)
def console():
while True:
msg = socket.recv_pyobj()
print msg
import pdb; pdb.set_trace()
return gevent.spawn(console)
class NumpyChannel(zmq.Socket):
def recv_pandas(self, flags=0, copy=True, track=False):
# Pandas Metadata
index, columns, dtype_name, shape = msgpack.loads(self.recv(flags=flags))
# Pandas ndarray
ndbuffer = self.recv(flags=flags, copy=copy, track=track)
buf = buffer(ndbuffer)
ndarray = numpy.frombuffer(buf, dtype=dtype(dtype_name)).reshape(shape)
return DataFrame(data=ndarray, index=index,
columns=columns, dtype=dtype_name)
def send_pandas(self, df, flags=0, copy=True, track=False):
# Pandas Metadata
index = df.index.tolist()
columns = df.columns.tolist()
dtype_name = df.values.dtype.name
shape = df.values.shape
# Pandas ndarray
ndarray = df.values
metadata = msgpack.dumps((index, columns, dtype_name, shape))
self.send(metadata, flags|zmq.SNDMORE)
return self.send(ndarray, flags, copy=copy, track=track)
if __name__ == '__main__':
from numpy.random import randn
df = DataFrame(randn(5,5))
ctx = zmq.Context.instance()
def send():
pub = NumpyChannel(ctx, zmq.PUSH)
pub.bind('inproc://a')
for i in xrange(100):
pub.send_pandas(df, copy=False)
def recv():
sub = NumpyChannel(ctx, zmq.PULL)
sub.connect('inproc://a')
for i in xrange(100):
sub.recv_pandas(copy=False)
gevent.joinall([
gevent.spawn(send),
gevent.spawn(recv)
])