Example of using CEOChange in pipeline¶
Trying to use CEOChange in pipeline
In [10]:
import pandas as pd
from pytz import utc
from quantopian.pipeline import Pipeline
from quantopian.pipeline.data.eventvestor import CEOChangeAnnouncements as cc
In [11]:
def make_pipeline(minmal_trade_million=10):
ceo_gender = cc.in_ceogender.latest
input_factor_pipeline = {
'ceo_gender': ceo_gender
}
return Pipeline(
columns=input_factor_pipeline,
# 1 day after earnings
screen=(
# ensure we are looking at Q500 + 95 quantile
#Q500US() & #& adv.percentile_between(95, 100)
ceo_gender.notnull()
)
)
pipe = make_pipeline()
pipe.show_graph(format='png')
Out[11]:
In [13]:
from quantopian.research import run_pipeline
#START = pd.Timestamp('2007-01-01', tz=utc)
START = pd.Timestamp('2013-01-01', tz=utc)
#END = pd.Timestamp('2014-10-01', tz=utc)
END = pd.Timestamp('2014-10-01', tz=utc)
pipeline_data = run_pipeline(
pipe,
start_date=START,
end_date=END
)
--------------------------------------------------------------------------- ValueError Traceback (most recent call last) <ipython-input-13-2b5604e49634> in <module>() 9 pipe, 10 start_date=START, ---> 11 end_date=END 12 ) /build/src/extensions/extensions/main.py in run_pipeline(pipeline, start_date, end_date) 405 start_date=adjust_date, 406 end_date=adjust_date, --> 407 ) 408 def run_pipeline(pipeline, start_date, end_date): 409 """ /build/src/extensions/extensions/main.py in run_pipeline(pipeline, start_date, end_date) 413 PipelineEngine.run_pipeline. 414 """ --> 415 return engine.run_pipeline(pipeline, start_date, end_date) 416 return run_pipeline 417 /build/src/qexec_repo/zipline_repo/zipline/pipeline/engine.pyc in run_pipeline(self, pipeline, start_date, end_date) 246 dates, 247 assets, --> 248 initial_workspace, 249 ) 250 /build/src/qexec_repo/zipline_repo/zipline/pipeline/engine.pyc in compute_chunk(self, graph, dates, assets, initial_workspace) 415 loader = get_loader(term) 416 loaded = loader.load_adjusted_array( --> 417 to_load, mask_dates, assets, mask, 418 ) 419 workspace.update(loaded) /build/src/qexec_repo/qexec/databazaar/pipeline/core.py in load_adjusted_array(self, columns, dates, assets, mask) 81 dates, 82 assets, ---> 83 mask, 84 ) 85 /build/src/qexec_repo/zipline_repo/zipline/pipeline/loaders/blaze/events.py in load_adjusted_array(self, columns, dates, assets, mask) 98 self._data_query_tz, 99 self._expr, --> 100 self._odo_kwargs) 101 102 return EventsLoader( /build/src/qexec_repo/zipline_repo/zipline/pipeline/loaders/blaze/utils.py in load_raw_data(assets, dates, data_query_time, data_query_tz, expr, odo_kwargs, checkpoints) 54 upper_dt, 55 checkpoints=checkpoints, ---> 56 odo_kwargs=odo_kwargs, 57 ) 58 sids = raw[SID_FIELD_NAME] /build/src/qexec_repo/zipline_repo/zipline/pipeline/loaders/blaze/core.pyc in ffill_query_in_range(expr, lower, upper, checkpoints, odo_kwargs, ts_field) 1250 expr[pred], 1251 pd.DataFrame, -> 1252 **odo_kwargs 1253 ), 1254 ), /build/src/qexec_repo/src/odo/odo/odo.pyc in odo(source, target, **kwargs) 89 odo.append.append - Add things onto existing things 90 """ ---> 91 return into(target, source, **kwargs) /usr/local/lib/python2.7/dist-packages/multipledispatch/dispatcher.pyc in __call__(self, *args, **kwargs) 162 self._cache[types] = func 163 try: --> 164 return func(*args, **kwargs) 165 166 except MDNotImplementedError: /build/src/qexec_repo/src/blaze/blaze/interactive.pyc in into(a, b, **kwargs) 402 @dispatch((object, type, str, unicode), Expr) 403 def into(a, b, **kwargs): --> 404 result = compute(b, return_type='native', **kwargs) 405 kwargs['dshape'] = b.dshape 406 return into(a, result, **kwargs) /usr/local/lib/python2.7/dist-packages/multipledispatch/dispatcher.pyc in __call__(self, *args, **kwargs) 162 self._cache[types] = func 163 try: --> 164 return func(*args, **kwargs) 165 166 except MDNotImplementedError: /build/src/qexec_repo/src/blaze/blaze/interactive.pyc in compute(expr, **kwargs) 203 raise ValueError("No data resources found") 204 else: --> 205 return compute(expr, resources, **kwargs) 206 207 /usr/local/lib/python2.7/dist-packages/multipledispatch/dispatcher.pyc in __call__(self, *args, **kwargs) 162 self._cache[types] = func 163 try: --> 164 return func(*args, **kwargs) 165 166 except MDNotImplementedError: /build/src/qexec_repo/src/blaze/blaze/compute/core.pyc in compute(expr, d, return_type, **kwargs) 410 d4 = d3 411 --> 412 result = top_then_bottom_then_top_again_etc(expr3, d4, **kwargs) 413 if post_compute_: 414 result = post_compute_(expr3, result, scope=d4) /build/src/qexec_repo/src/blaze/blaze/compute/core.pyc in top_then_bottom_then_top_again_etc(expr, scope, **kwargs) 151 # 1. See if we have a direct computation path with compute_down 152 try: --> 153 return compute_down(expr, *leaf_data, **kwargs) 154 except NotImplementedError: 155 pass /usr/local/lib/python2.7/dist-packages/multipledispatch/dispatcher.pyc in __call__(self, *args, **kwargs) 162 self._cache[types] = func 163 try: --> 164 return func(*args, **kwargs) 165 166 except MDNotImplementedError: /build/src/qexec_repo/src/blaze/blaze/server/client.pyc in compute_down(expr, ec, profiler_output, compute_kwargs, odo_kwargs, **kwargs) 240 241 if not ok(r): --> 242 raise ValueError("Bad response: %s" % reason(r)) 243 response = serial.loads(content(r)) 244 if profiler_output is not None: ValueError: Bad response: Computation failed with message: AssertionError: names = ['event_id', 'asof_date', 'trade_date', 'symbol', 'event_type', 'event_headline', 'change_status', 'change_scenario', 'change_type', 'change_source', 'change_reason', 'in_ceoname', 'in_ceogender', 'out_ceoname', 'out_ceogender', 'effective_date', 'event_rating', 'timestamp', 'sid'] expr._child.fields = ['event_id', 'trade_date', 'symbol', 'event_type', 'event_headline', 'change_status', 'change_scenario', 'change_type', 'change_source', 'change_reason', 'in_ceoname', 'in_ceogender', 'out_ceoname', 'out_ceogender', 'effective_date', 'event_rating', 'sid', 'asof_date', 'timestamp'] File "/home/databazaar/databazaar-clone/submodules/blaze/blaze/server/server.py", line 643, in compserver **compute_kwargs), File "/home/databazaar/.venv/lib/python3.4/site-packages/multipledispatch/dispatcher.py", line 164, in __call__ return func(*args, **kwargs) File "/home/databazaar/databazaar-clone/submodules/blaze/blaze/compute/core.py", line 412, in compute result = top_then_bottom_then_top_again_etc(expr3, d4, **kwargs) File "/home/databazaar/databazaar-clone/submodules/blaze/blaze/compute/core.py", line 189, in top_then_bottom_then_top_again_etc return top_then_bottom_then_top_again_etc(expr3, scope4, **kwargs) File "/home/databazaar/databazaar-clone/submodules/blaze/blaze/compute/core.py", line 189, in top_then_bottom_then_top_again_etc return top_then_bottom_then_top_again_etc(expr3, scope4, **kwargs) File "/home/databazaar/databazaar-clone/submodules/blaze/blaze/compute/core.py", line 153, in top_then_bottom_then_top_again_etc return compute_down(expr, *leaf_data, **kwargs) File "/home/databazaar/.venv/lib/python3.4/site-packages/multipledispatch/dispatcher.py", line 164, in __call__ return func(*args, **kwargs) File "/home/databazaar/databazaar-clone/databazaar/utils/throttler.py", line 95, in compute_throttler return_type='native', File "/home/databazaar/.venv/lib/python3.4/site-packages/multipledispatch/dispatcher.py", line 164, in __call__ return func(*args, **kwargs) File "/home/databazaar/databazaar-clone/submodules/blaze/blaze/compute/core.py", line 412, in compute result = top_then_bottom_then_top_again_etc(expr3, d4, **kwargs) File "/home/databazaar/databazaar-clone/submodules/blaze/blaze/compute/core.py", line 158, in top_then_bottom_then_top_again_etc expr2, scope2 = bottom_up_until_type_break(expr, scope, **kwargs) File "/home/databazaar/databazaar-clone/submodules/blaze/blaze/compute/core.py", line 301, in bottom_up_until_type_break for i in inputs]) File "/home/databazaar/databazaar-clone/submodules/blaze/blaze/compute/core.py", line 301, in <listcomp> for i in inputs]) File "/home/databazaar/databazaar-clone/submodules/blaze/blaze/compute/core.py", line 325, in bottom_up_until_type_break **kwargs)} File "/home/databazaar/.venv/lib/python3.4/site-packages/multipledispatch/dispatcher.py", line 164, in __call__ return func(*args, **kwargs) File "/home/databazaar/databazaar-clone/submodules/blaze/blaze/compute/sql.py", line 1040, in compute_up 'names = %r\nexpr._child.fields = %r' % (names, expr._child.fields)