本文整理汇总了Python中zipline.algorithm.TradingAlgorithm类的典型用法代码示例。如果您正苦于以下问题:Python TradingAlgorithm类的具体用法?Python TradingAlgorithm怎么用?Python TradingAlgorithm使用的例子?那么恭喜您, 这里精选的类代码示例或许可以为您提供帮助。
在下文中一共展示了TradingAlgorithm类的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_pipeline_output_after_initialize
def test_pipeline_output_after_initialize(self):
"""
Assert that calling pipeline_output after initialize raises correctly.
"""
def initialize(context):
attach_pipeline(Pipeline(), "test")
pipeline_output("test")
raise AssertionError("Shouldn't make it past pipeline_output()")
def handle_data(context, data):
raise AssertionError("Shouldn't make it past initialize!")
def before_trading_start(context, data):
raise AssertionError("Shouldn't make it past initialize!")
algo = TradingAlgorithm(
initialize=initialize,
handle_data=handle_data,
before_trading_start=before_trading_start,
data_frequency="daily",
get_pipeline_loader=lambda column: self.pipeline_loader,
start=self.first_asset_start - self.trading_day,
end=self.last_asset_end + self.trading_day,
env=self.env,
)
with self.assertRaises(PipelineOutputDuringInitialize):
algo.run(self.data_portal)
开发者ID:RoyHsiao,项目名称:zipline,代码行数:29,代码来源:test_pipeline_algo.py
示例2: test_pipeline_output_after_initialize
def test_pipeline_output_after_initialize(self):
"""
Assert that calling pipeline_output after initialize raises correctly.
"""
def initialize(context):
attach_pipeline(Pipeline('test'))
pipeline_output('test')
raise AssertionError("Shouldn't make it past pipeline_output()")
def handle_data(context, data):
raise AssertionError("Shouldn't make it past initialize!")
def before_trading_start(context, data):
raise AssertionError("Shouldn't make it past initialize!")
algo = TradingAlgorithm(
initialize=initialize,
handle_data=handle_data,
before_trading_start=before_trading_start,
data_frequency='daily',
pipeline_loader=self.pipeline_loader,
start=self.first_asset_start - trading_day,
end=self.last_asset_end + trading_day,
env=self.env,
)
with self.assertRaises(PipelineOutputDuringInitialize):
algo.run(source=self.closes)
开发者ID:icecube11,项目名称:zipline,代码行数:28,代码来源:test_pipeline_algo.py
示例3: test_get_output_nonexistent_pipeline
def test_get_output_nonexistent_pipeline(self):
"""
Assert that calling add_pipeline after initialize raises appropriately.
"""
def initialize(context):
attach_pipeline(Pipeline(), "test")
def handle_data(context, data):
raise AssertionError("Shouldn't make it past before_trading_start")
def before_trading_start(context, data):
pipeline_output("not_test")
raise AssertionError("Shouldn't make it past pipeline_output!")
algo = TradingAlgorithm(
initialize=initialize,
handle_data=handle_data,
before_trading_start=before_trading_start,
data_frequency="daily",
get_pipeline_loader=lambda column: self.pipeline_loader,
start=self.first_asset_start - self.trading_day,
end=self.last_asset_end + self.trading_day,
env=self.env,
)
with self.assertRaises(NoSuchPipeline):
algo.run(self.data_portal)
开发者ID:RoyHsiao,项目名称:zipline,代码行数:28,代码来源:test_pipeline_algo.py
示例4: run_algorithm
def run_algorithm(
security='AAPL',
start_date='20100101',
end_date='20150101',
initial_cash=100000,
rsi_window=15,
low_RSI=30,
high_RSI=70):
logging.debug('run_algorithm begin')
# dates
start = dateutil.parser.parse(start_date)
end = dateutil.parser.parse(end_date)
# get data from yahoo
data = load_from_yahoo(stocks=[security], indexes={}, start=start, end=end)
logging.debug('done loading from yahoo. {} {} {}'.format(
security, start_date, end_date))
# create and run algorithm
algo = TradingAlgorithm(
initialize=initialize,
handle_data=handle_data,
capital_base=initial_cash)
algo.security = security
initialize.low_RSI = low_RSI
initialize.high_RSI = high_RSI
initialize.rsi_window = rsi_window
logging.debug('starting to run algo...')
results = algo.run(data).dropna()
logging.debug('done running algo')
return results
开发者ID:nikivasilev,项目名称:zipline,代码行数:31,代码来源:relative_strength_index.py
示例5: test_schedule_funtion_rule_creation
def test_schedule_funtion_rule_creation(self, mode):
nop = lambda *args, **kwargs: None
self.sim_params.data_frequency = mode
algo = TradingAlgorithm(
initialize=nop, handle_data=nop, sim_params=self.sim_params,
)
# Schedule something for NOT Always.
algo.schedule_function(nop, time_rule=zipline.utils.events.Never())
event_rule = algo.event_manager._events[1].rule
self.assertIsInstance(event_rule, zipline.utils.events.OncePerDay)
inner_rule = event_rule.rule
self.assertIsInstance(inner_rule, zipline.utils.events.ComposedRule)
first = inner_rule.first
second = inner_rule.second
composer = inner_rule.composer
self.assertIsInstance(first, zipline.utils.events.Always)
if mode == 'daily':
self.assertIsInstance(second, zipline.utils.events.Always)
else:
self.assertIsInstance(second, zipline.utils.events.Never)
self.assertIs(composer, zipline.utils.events.ComposedRule.lazy_and)
开发者ID:CallingWisdom,项目名称:zipline,代码行数:30,代码来源:test_algorithm.py
示例6: test_api_get_environment
def test_api_get_environment(self):
platform = 'zipline'
metadata = {0: {'symbol': 'TEST',
'asset_type': 'equity'}}
algo = TradingAlgorithm(script=api_get_environment_algo,
asset_metadata=metadata,
platform=platform)
algo.run(self.df)
self.assertEqual(algo.environment, platform)
开发者ID:kapil0187,项目名称:zipline,代码行数:9,代码来源:test_algorithm.py
示例7: wrapper
def wrapper(self, data_frequency, days=None):
sim_params, source = self.sim_and_source[data_frequency]
algo = TradingAlgorithm(
initialize=initialize_with(self, tfm_name, days),
handle_data=handle_data_wrapper(f),
sim_params=sim_params,
)
algo.run(source)
开发者ID:1TTT9,项目名称:zipline,代码行数:9,代码来源:test_transforms.py
示例8: test_assets_appear_on_correct_days
def test_assets_appear_on_correct_days(self, test_name, chunks):
"""
Assert that assets appear at correct times during a backtest, with
correctly-adjusted close price values.
"""
if chunks == 'all_but_one_day':
chunks = (
self.dates.get_loc(self.last_asset_end) -
self.dates.get_loc(self.first_asset_start)
) - 1
elif chunks == 'custom_iter':
chunks = []
st = np.random.RandomState(12345)
remaining = (
self.dates.get_loc(self.last_asset_end) -
self.dates.get_loc(self.first_asset_start)
)
while remaining > 0:
chunk = st.randint(3)
chunks.append(chunk)
remaining -= chunk
def initialize(context):
p = attach_pipeline(Pipeline(), 'test', chunks=chunks)
p.add(USEquityPricing.close.latest, 'close')
def handle_data(context, data):
results = pipeline_output('test')
date = get_datetime().normalize()
for asset in self.assets:
# Assets should appear iff they exist today and yesterday.
exists_today = self.exists(date, asset)
existed_yesterday = self.exists(date - self.trading_day, asset)
if exists_today and existed_yesterday:
latest = results.loc[asset, 'close']
self.assertEqual(latest, self.expected_close(date, asset))
else:
self.assertNotIn(asset, results.index)
before_trading_start = handle_data
algo = TradingAlgorithm(
initialize=initialize,
handle_data=handle_data,
before_trading_start=before_trading_start,
data_frequency='daily',
get_pipeline_loader=lambda column: self.pipeline_loader,
start=self.first_asset_start,
end=self.last_asset_end,
env=self.env,
)
# Run for a week in the middle of our data.
algo.run(self.data_portal)
开发者ID:FranSal,项目名称:zipline,代码行数:55,代码来源:test_pipeline_algo.py
示例9: test_algo_record_allow_mock
def test_algo_record_allow_mock(self):
"""
Test that values from "MagicMock"ed methods can be passed to record.
Relevant for our basic/validation and methods like history, which
will end up returning a MagicMock instead of a DataFrame.
"""
test_algo = TradingAlgorithm(script=record_variables, sim_params=self.sim_params)
set_algo_instance(test_algo)
test_algo.record(foo=MagicMock())
开发者ID:ChrisBg,项目名称:zipline,代码行数:11,代码来源:test_algorithm.py
示例10: test_order_in_init
def test_order_in_init(self):
"""
Test that calling order in initialize
will raise an error.
"""
with self.assertRaises(OrderDuringInitialize):
test_algo = TradingAlgorithm(
script=call_order_in_init,
sim_params=self.sim_params,
)
set_algo_instance(test_algo)
test_algo.run(self.source)
开发者ID:acycliq,项目名称:zipline,代码行数:12,代码来源:test_algorithm.py
示例11: algoRunNext
def algoRunNext(self):
if(self.hasNext()):
currentSeriesValue = self.timeDF.iloc[self.index];
self.index += 1;
start, end = self._getStartEndTimeFromCurrentSeries(currentSeriesValue);
algo = TradingAlgorithm(initialize=self.initialize,
handle_data=self.handle_data, identifiers=self.stockName)
data = loader.load_bars_from_yahoo(stocks=self.stockName,
start=start, end=end);
return algo.run(data);
else:
raise ValueError;
开发者ID:tomyitav,项目名称:zipline,代码行数:12,代码来源:AlgoRunnerIterator.py
示例12: test_add_history_in_handle_data
def test_add_history_in_handle_data(self):
def handle_data(algo, data):
algo.add_history(1, '1m', 'price')
algo = TradingAlgorithm(
initialize=lambda _: None,
handle_data=handle_data,
sim_params=self.sim_params,
)
algo.run(self.source)
self.assertIsNotNone(algo.history_container)
self.assertEqual(algo.history_container.buffer_panel.window_length, 1)
开发者ID:CallingWisdom,项目名称:zipline,代码行数:13,代码来源:test_algorithm.py
示例13: runMaster
def runMaster():
""" Loads backtest data, and runs the backtest."""
global TRAINING_STOCK, BACKTEST_STOCK
TRAINING_STOCK = 'SPY'
SELECT_STOCKS = ['AAPL', 'DIS', 'XOM', 'UNH', 'WMT']
algo_obj = TradingAlgorithm(initialize=initialize, handle_data=handle_data)
perf_manual = []
for stock in SELECT_STOCKS:
BACKTEST_STOCK = stock
backtestData = loadData(2002, 2002+BACKTEST_TIME, stock_list=[stock, 'SPY']) #, startM=1, endM=2,
print "Create algorithm..."
perf_manual.append(algo_obj.run(backtestData))
analyze(perf_manual)
开发者ID:nsbradford,项目名称:AI,代码行数:13,代码来源:runner.py
示例14: test_multiple_pipelines
def test_multiple_pipelines(self):
"""
Test that we can attach multiple pipelines and access the correct
output based on the pipeline name.
"""
def initialize(context):
pipeline_close = attach_pipeline(Pipeline(), 'test_close')
pipeline_volume = attach_pipeline(Pipeline(), 'test_volume')
pipeline_close.add(USEquityPricing.close.latest, 'close')
pipeline_volume.add(USEquityPricing.volume.latest, 'volume')
def handle_data(context, data):
closes = pipeline_output('test_close')
volumes = pipeline_output('test_volume')
date = get_datetime().normalize()
for asset in self.assets:
# Assets should appear iff they exist today and yesterday.
exists_today = self.exists(date, asset)
existed_yesterday = self.exists(date - self.trading_day, asset)
if exists_today and existed_yesterday:
self.assertEqual(
closes.loc[asset, 'close'],
self.expected_close(date, asset)
)
self.assertEqual(
volumes.loc[asset, 'volume'],
self.expected_volume(date, asset)
)
else:
self.assertNotIn(asset, closes.index)
self.assertNotIn(asset, volumes.index)
column_to_loader = {
USEquityPricing.close: self.pipeline_close_loader,
USEquityPricing.volume: self.pipeline_volume_loader,
}
algo = TradingAlgorithm(
initialize=initialize,
handle_data=handle_data,
data_frequency='daily',
get_pipeline_loader=lambda column: column_to_loader[column],
start=self.first_asset_start,
end=self.last_asset_end,
env=self.env,
)
algo.run(self.data_portal)
开发者ID:SJCosgrove,项目名称:quantopianresearch,代码行数:49,代码来源:test_pipeline_algo.py
示例15: test_get_open_orders
def test_get_open_orders(self):
def initialize(algo):
algo.minute = 0
def handle_data(algo, data):
if algo.minute == 0:
# Should be filled by the next minute
algo.order(1, 1)
# Won't be filled because the price is too low.
algo.order(2, 1, style=LimitOrder(0.01))
algo.order(2, 1, style=LimitOrder(0.01))
algo.order(2, 1, style=LimitOrder(0.01))
all_orders = algo.get_open_orders()
self.assertEqual(list(all_orders.keys()), [1, 2])
self.assertEqual(all_orders[1], algo.get_open_orders(1))
self.assertEqual(len(all_orders[1]), 1)
self.assertEqual(all_orders[2], algo.get_open_orders(2))
self.assertEqual(len(all_orders[2]), 3)
if algo.minute == 1:
# First order should have filled.
# Second order should still be open.
all_orders = algo.get_open_orders()
self.assertEqual(list(all_orders.keys()), [2])
self.assertEqual([], algo.get_open_orders(1))
orders_2 = algo.get_open_orders(2)
self.assertEqual(all_orders[2], orders_2)
self.assertEqual(len(all_orders[2]), 3)
for order in orders_2:
algo.cancel_order(order)
all_orders = algo.get_open_orders()
self.assertEqual(all_orders, {})
algo.minute += 1
algo = TradingAlgorithm(initialize=initialize,
handle_data=handle_data,
sim_params=self.sim_params)
algo.run(self.source)
开发者ID:erain,项目名称:zipline,代码行数:49,代码来源:test_algorithm.py
示例16: _create_generator
def _create_generator(self, sim_params):
# Call the simulation trading algorithm for side-effects:
# it creates the perf tracker
TradingAlgorithm._create_generator(self, sim_params)
self.trading_client = LiveAlgorithmExecutor(
self,
sim_params,
self.data_portal,
self._create_clock(),
self._create_benchmark_source(),
self.restrictions,
universe_func=self._calculate_universe
)
return self.trading_client.transform()
开发者ID:huangzhengyong,项目名称:zipline,代码行数:15,代码来源:algorithm_live.py
示例17: run_clusters
def run_clusters(strategy_class, clustering_tickers, cluster_num, epochs_num, training_start,
training_end, backtest_start, backtest_end, is_graph, is_elbow):
""" Run the test given command-line args.
Cluster.
For each cluster, train a strategy on that cluster.
For each stock in that cluster, run a backtest.
Graph results.
"""
print "\nGathering data..."
ticker_list, raw_stock_data_list = Manager.getRawStockDataList(clustering_tickers, training_start, training_end, 252)
normalized_stock_data_list = [Manager.preprocessData(x) for x in raw_stock_data_list]
print "\nClustering..."
tickers, clusters = createClusters(ticker_list, normalized_stock_data_list, cluster_num)
print "# of stocks: " + str(len(normalized_stock_data_list))
print "# of clusters: " + str(len(clusters))
print ""
for t, c in itertools.izip(tickers, clusters):
print "\tCluster: " + str(len(c)), "stocks: ",
for symbol in t:
print symbol,
print ""
if is_graph:
graphClusters(clusters)
if is_elbow:
graphElbowMethod(normalized_stock_data_list)
for t, cluster in itertools.izip(tickers, clusters):
settings.STRATEGY_OBJECT = trainStrategy(strategy_class, cluster, epochs_num)
for ticker in t:
print "Cluster:", t
print "Stock:", ticker
tmp_ticks, tmp_data = Manager.getRawStockDataList([ticker], training_start, training_end, 252)
settings.BACKTEST_STOCK = ticker
settings.PRE_BACKTEST_DATA = tmp_data[0]
print "Create Algorithm..."
algo_obj = TradingAlgorithm(initialize=initialize, handle_data=handle_data)
try:
backtest_data = load_bars_from_yahoo(stocks=[ticker, 'SPY'], start=backtest_start, end=backtest_end)
try:
perf = algo_obj.run(backtest_data)
analyze([ticker], [perf])
except ValueError, e:
print str(e)
except IOError, e:
print "Stock Error: could not load", ticker, "from Yahoo."
print "Only testing one cluster for now - Done!"
return
开发者ID:nsbradford,项目名称:quantraider,代码行数:48,代码来源:runner.py
示例18: test_pipeline_beyond_daily_bars
def test_pipeline_beyond_daily_bars(self):
"""
Ensure that we can run an algo with pipeline beyond the max date
of the daily bars.
"""
# For ensuring we call before_trading_start.
count = [0]
current_day = self.trading_calendar.next_session_label(
self.pipeline_loader.raw_price_loader.last_available_dt,
)
def initialize(context):
pipeline = attach_pipeline(Pipeline(), 'test')
vwap = VWAP(window_length=10)
pipeline.add(vwap, 'vwap')
# Nothing should have prices less than 0.
pipeline.set_screen(vwap < 0)
def handle_data(context, data):
pass
def before_trading_start(context, data):
context.results = pipeline_output('test')
self.assertTrue(context.results.empty)
count[0] += 1
algo = TradingAlgorithm(
initialize=initialize,
handle_data=handle_data,
before_trading_start=before_trading_start,
data_frequency='daily',
get_pipeline_loader=lambda column: self.pipeline_loader,
start=self.dates[0],
end=current_day,
env=self.env,
)
algo.run(
FakeDataPortal(),
overwrite_sim_params=False,
)
self.assertTrue(count[0] > 0)
开发者ID:AtwooTM,项目名称:zipline,代码行数:47,代码来源:test_pipeline_algo.py
示例19: test_history
def test_history(self):
history_algo = """
from zipline.api import history, add_history
def initialize(context):
add_history(10, '1d', 'price')
def handle_data(context, data):
df = history(10, '1d', 'price')
"""
algo = TradingAlgorithm(
script=history_algo,
sim_params=self.sim_params,
)
output = algo.run(self.source)
self.assertIsNot(output, None)
开发者ID:CallingWisdom,项目名称:zipline,代码行数:17,代码来源:test_algorithm.py
示例20: test_history
def test_history(self):
history_algo = """
from zipline.api import history, add_history
def initialize(context):
add_history(10, '1d', 'price')
def handle_data(context, data):
df = history(10, '1d', 'price')
"""
start = pd.Timestamp('1991-01-01', tz='UTC')
end = pd.Timestamp('1991-01-15', tz='UTC')
source = RandomWalkSource(start=start,
end=end)
algo = TradingAlgorithm(script=history_algo, data_frequency='minute')
output = algo.run(source)
self.assertIsNot(output, None)
开发者ID:aswizzl,项目名称:luckybomb,代码行数:17,代码来源:test_algorithm.py
注:本文中的zipline.algorithm.TradingAlgorithm类示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论