本文整理汇总了Python中zipline.utils.factory.create_trade_history函数的典型用法代码示例。如果您正苦于以下问题:Python create_trade_history函数的具体用法?Python create_trade_history怎么用?Python create_trade_history使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了create_trade_history函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: test_algo_with_rl_violation
def test_algo_with_rl_violation(self):
sim_params = factory.create_simulation_parameters(
start=list(LEVERAGED_ETFS.keys())[0], num_days=4)
trade_history = factory.create_trade_history(
'BZQ',
[10.0, 10.0, 11.0, 11.0],
[100, 100, 100, 300],
timedelta(days=1),
sim_params
)
self.source = SpecificEquityTrades(event_list=trade_history)
algo = RestrictedAlgoWithoutCheck(symbol='BZQ', sim_params=sim_params)
with self.assertRaises(TradingControlViolation) as ctx:
algo.run(self.source)
self.check_algo_exception(algo, ctx, 0)
# repeat with a symbol from a different lookup date
trade_history = factory.create_trade_history(
'JFT',
[10.0, 10.0, 11.0, 11.0],
[100, 100, 100, 300],
timedelta(days=1),
sim_params
)
self.source = SpecificEquityTrades(event_list=trade_history)
algo = RestrictedAlgoWithoutCheck(symbol='JFT', sim_params=sim_params)
with self.assertRaises(TradingControlViolation) as ctx:
algo.run(self.source)
self.check_algo_exception(algo, ctx, 0)
开发者ID:AshBT,项目名称:zipline,代码行数:35,代码来源:test_security_list.py
示例2: test_short_position_pays_dividend
def test_short_position_pays_dividend(self):
# post some trades in the market
events = factory.create_trade_history(
1, [10, 10, 10, 10, 10], [100, 100, 100, 100, 100], oneday, self.sim_params
)
dividend = factory.create_dividend(
1,
10.00,
# declare at open of test
events[0].dt,
# ex_date same as trade 2
events[2].dt,
events[3].dt,
)
txn = create_txn(1, 10.0, -100, events[1].dt)
events.insert(1, txn)
events.insert(0, dividend)
results = calculate_results(self, events)
self.assertEqual(len(results), 5)
cumulative_returns = [event["cumulative_perf"]["returns"] for event in results]
self.assertEqual(cumulative_returns, [0.0, 0.0, 0.0, -0.1, -0.1])
daily_returns = [event["daily_perf"]["returns"] for event in results]
self.assertEqual(daily_returns, [0.0, 0.0, 0.0, -0.1, 0.0])
cash_flows = [event["daily_perf"]["capital_used"] for event in results]
self.assertEqual(cash_flows, [0, 1000, 0, -1000, 0])
cumulative_cash_flows = [event["cumulative_perf"]["capital_used"] for event in results]
self.assertEqual(cumulative_cash_flows, [0, 1000, 1000, 0, 0])
开发者ID:bogdan-kulynych,项目名称:zipline,代码行数:30,代码来源:test_perf_tracking.py
示例3: test_commission_event
def test_commission_event(self):
with trading.TradingEnvironment():
events = factory.create_trade_history(
1,
[10, 10, 10, 10, 10],
[100, 100, 100, 100, 100],
oneday,
self.sim_params
)
cash_adj_dt = self.sim_params.period_start \
+ datetime.timedelta(hours=3)
cash_adjustment = factory.create_commission(1, 300.0,
cash_adj_dt)
# Insert a purchase order.
events.insert(0, create_txn(events[0], 20, 1))
events.insert(1, cash_adjustment)
results = calculate_results(self, events)
# Validate that we lost 320 dollars from our cash pool.
self.assertEqual(results[-1]['cumulative_perf']['ending_cash'],
9680)
# Validate that the cost basis of our position changed.
self.assertEqual(results[-1]['daily_perf']['positions']
[0]['cost_basis'], 320.0)
开发者ID:commonlisp,项目名称:zipline,代码行数:26,代码来源:test_perf_tracking.py
示例4: test_long_position_receives_dividend
def test_long_position_receives_dividend(self):
# post some trades in the market
events = factory.create_trade_history(
1, [10, 10, 10, 10, 10], [100, 100, 100, 100, 100], oneday, self.sim_params
)
dividend = factory.create_dividend(
1,
10.00,
# declared date, when the algorithm finds out about
# the dividend
events[1].dt,
# ex_date, when the algorithm is credited with the
# dividend
events[1].dt,
# pay date, when the algorithm receives the dividend.
events[2].dt,
)
txn = create_txn(1, 10.0, 100, events[0].dt)
events.insert(0, txn)
events.insert(1, dividend)
results = calculate_results(self, events)
self.assertEqual(len(results), 5)
cumulative_returns = [event["cumulative_perf"]["returns"] for event in results]
self.assertEqual(cumulative_returns, [0.0, 0.0, 0.1, 0.1, 0.1])
daily_returns = [event["daily_perf"]["returns"] for event in results]
self.assertEqual(daily_returns, [0.0, 0.0, 0.10, 0.0, 0.0])
cash_flows = [event["daily_perf"]["capital_used"] for event in results]
self.assertEqual(cash_flows, [-1000, 0, 1000, 0, 0])
cumulative_cash_flows = [event["cumulative_perf"]["capital_used"] for event in results]
self.assertEqual(cumulative_cash_flows, [-1000, -1000, 0, 0, 0])
cash_pos = [event["cumulative_perf"]["ending_cash"] for event in results]
self.assertEqual(cash_pos, [9000, 9000, 10000, 10000, 10000])
开发者ID:bogdan-kulynych,项目名称:zipline,代码行数:35,代码来源:test_perf_tracking.py
示例5: test_ending_before_pay_date
def test_ending_before_pay_date(self):
# post some trades in the market
events = factory.create_trade_history(
1, [10, 10, 10, 10, 10], [100, 100, 100, 100, 100], oneday, self.sim_params
)
pay_date = self.sim_params.first_open
# find pay date that is much later.
for i in xrange(30):
pay_date = factory.get_next_trading_dt(pay_date, oneday)
dividend = factory.create_dividend(1, 10.00, events[0].dt, events[1].dt, pay_date)
buy_txn = create_txn(1, 10.0, 100, events[1].dt)
events.insert(2, buy_txn)
events.insert(1, dividend)
results = calculate_results(self, events)
self.assertEqual(len(results), 5)
cumulative_returns = [event["cumulative_perf"]["returns"] for event in results]
self.assertEqual(cumulative_returns, [0, 0, 0, 0.0, 0.0])
daily_returns = [event["daily_perf"]["returns"] for event in results]
self.assertEqual(daily_returns, [0, 0, 0, 0, 0])
cash_flows = [event["daily_perf"]["capital_used"] for event in results]
self.assertEqual(cash_flows, [0, -1000, 0, 0, 0])
cumulative_cash_flows = [event["cumulative_perf"]["capital_used"] for event in results]
self.assertEqual(cumulative_cash_flows, [0, -1000, -1000, -1000, -1000])
开发者ID:bogdan-kulynych,项目名称:zipline,代码行数:26,代码来源:test_perf_tracking.py
示例6: test_set_max_order_count
def test_set_max_order_count(self):
# Override the default setUp to use six-hour intervals instead of full
# days so we can exercise trading-session rollover logic.
trade_history = factory.create_trade_history(
self.sid,
[10.0, 10.0, 11.0, 11.0],
[100, 100, 100, 300],
timedelta(hours=6),
self.sim_params
)
self.source = SpecificEquityTrades(event_list=trade_history)
def handle_data(algo, data):
for i in range(5):
algo.order(self.sid, 1)
algo.order_count += 1
algo = SetMaxOrderCountAlgorithm(3)
self.check_algo_fails(algo, handle_data, 3)
# Second call to handle_data is the same day as the first, so the last
# order of the second call should fail.
algo = SetMaxOrderCountAlgorithm(9)
self.check_algo_fails(algo, handle_data, 9)
# Only ten orders are placed per day, so this should pass even though
# in total more than 20 orders are placed.
algo = SetMaxOrderCountAlgorithm(10)
self.check_algo_succeeds(algo, handle_data, order_count=20)
开发者ID:erain,项目名称:zipline,代码行数:30,代码来源:test_algorithm.py
示例7: test_returns
def test_returns(self):
# Daily returns.
returns = Returns(1)
transformed = list(returns.transform(self.source))
tnfm_vals = [message[returns.get_hash()] for message in transformed]
# No returns for the first event because we don't have a
# previous close.
expected = [0.0, 0.0, 0.1, 0.0]
self.assertEquals(tnfm_vals, expected)
# Two-day returns. An extra kink here is that the
# factory will automatically skip a weekend for the
# last event. Results shouldn't notice this blip.
trade_history = factory.create_trade_history(
133, [10.0, 15.0, 13.0, 12.0, 13.0], [100, 100, 100, 300, 100], timedelta(days=1), self.trading_environment
)
self.source = SpecificEquityTrades(event_list=trade_history)
returns = StatefulTransform(Returns, 2)
transformed = list(returns.transform(self.source))
tnfm_vals = [message[returns.get_hash()] for message in transformed]
expected = [0.0, 0.0, (13.0 - 10.0) / 10.0, (12.0 - 15.0) / 15.0, (13.0 - 13.0) / 13.0]
self.assertEquals(tnfm_vals, expected)
开发者ID:gawecoti,项目名称:zipline,代码行数:30,代码来源:test_transforms.py
示例8: test_commission_zero_position
def test_commission_zero_position(self):
"""
Ensure no div-by-zero errors.
"""
with trading.TradingEnvironment():
events = factory.create_trade_history(
1,
[10, 10, 10, 10, 10],
[100, 100, 100, 100, 100],
oneday,
self.sim_params
)
cash_adj_dt = self.sim_params.first_open \
+ datetime.timedelta(hours=3)
cash_adjustment = factory.create_commission(1, 300.0,
cash_adj_dt)
# Insert a purchase order.
events.insert(0, create_txn(events[0], 20, 1))
# Sell that order.
events.insert(1, create_txn(events[1], 20, -1))
events.insert(2, cash_adjustment)
results = calculate_results(self, events)
# Validate that we lost 300 dollars from our cash pool.
self.assertEqual(results[-1]['cumulative_perf']['ending_cash'],
9700)
开发者ID:ChrisBg,项目名称:zipline,代码行数:29,代码来源:test_perf_tracking.py
示例9: test_algo_with_rl_violation_cumulative
def test_algo_with_rl_violation_cumulative(self):
"""
Add a new restriction, run a test long after both
knowledge dates, make sure stock from original restriction
set is still disallowed.
"""
sim_params = factory.create_simulation_parameters(
start=list(
LEVERAGED_ETFS.keys())[0] + timedelta(days=7), num_days=4)
with security_list_copy():
add_security_data(['AAPL'], [])
trade_history = factory.create_trade_history(
'BZQ',
[10.0, 10.0, 11.0, 11.0],
[100, 100, 100, 300],
timedelta(days=1),
sim_params,
env=self.env,
)
self.source = SpecificEquityTrades(event_list=trade_history,
env=self.env)
algo = RestrictedAlgoWithoutCheck(
symbol='BZQ', sim_params=sim_params, env=self.env)
with self.assertRaises(TradingControlViolation) as ctx:
algo.run(self.source)
self.check_algo_exception(algo, ctx, 0)
开发者ID:AlexanderAA,项目名称:zipline,代码行数:28,代码来源:test_security_list.py
示例10: test_cost_basis_calc_close_pos
def test_cost_basis_calc_close_pos(self):
history_args = (
1,
[10, 9, 11, 8, 9, 12, 13, 14],
[200, -100, -100, 100, -300, 100, 500, 400],
onesec,
self.sim_params
)
cost_bases = [10, 10, 0, 8, 9, 9, 13, 13.5]
trades = factory.create_trade_history(*history_args)
transactions = factory.create_txn_history(*history_args)
pp = perf.PerformancePeriod(1000.0)
for txn, cb in zip(transactions, cost_bases):
pp.execute_transaction(txn)
self.assertEqual(pp.positions[1].cost_basis, cb)
for trade in trades:
pp.update_last_sale(trade)
pp.calculate_performance()
self.assertEqual(pp.positions[1].cost_basis, cost_bases[-1])
开发者ID:MridulS,项目名称:zipline,代码行数:25,代码来源:test_perf_tracking.py
示例11: test_commission_zero_position
def test_commission_zero_position(self):
"""
Ensure no div-by-zero errors.
"""
events = factory.create_trade_history(
1,
[10, 10, 10, 10, 10],
[100, 100, 100, 100, 100],
oneday,
self.sim_params
)
# Buy and sell the same sid so that we have a zero position by the
# time of events[3].
txns = [
create_txn(events[0], 20, 1),
create_txn(events[1], 20, -1),
]
# Add a cash adjustment at the time of event[3].
cash_adj_dt = events[3].dt
cash_adjustment = factory.create_commission(1, 300.0, cash_adj_dt)
events.append(cash_adjustment)
results = calculate_results(self, events, txns=txns)
# Validate that we lost 300 dollars from our cash pool.
self.assertEqual(results[-1]['cumulative_perf']['ending_cash'],
9700)
开发者ID:MridulS,项目名称:zipline,代码行数:29,代码来源:test_perf_tracking.py
示例12: test_no_position_receives_no_dividend
def test_no_position_receives_no_dividend(self):
# post some trades in the market
events = factory.create_trade_history(
1,
[10, 10, 10, 10, 10],
[100, 100, 100, 100, 100],
oneday,
self.sim_params
)
dividend = factory.create_dividend(
1,
10.00,
events[0].dt,
events[1].dt,
events[2].dt
)
results = calculate_results(
self,
events,
dividend_events=[dividend],
)
self.assertEqual(len(results), 5)
cumulative_returns = \
[event['cumulative_perf']['returns'] for event in results]
self.assertEqual(cumulative_returns, [0.0, 0.0, 0.0, 0.0, 0.0])
daily_returns = [event['daily_perf']['returns'] for event in results]
self.assertEqual(daily_returns, [0.0, 0.0, 0.0, 0.0, 0.0])
cash_flows = [event['daily_perf']['capital_used'] for event in results]
self.assertEqual(cash_flows, [0, 0, 0, 0, 0])
cumulative_cash_flows = \
[event['cumulative_perf']['capital_used'] for event in results]
self.assertEqual(cumulative_cash_flows, [0, 0, 0, 0, 0])
开发者ID:MridulS,项目名称:zipline,代码行数:35,代码来源:test_perf_tracking.py
示例13: test_moving_stddev
def test_moving_stddev(self):
trade_history = factory.create_trade_history(
133,
[10.0, 15.0, 13.0, 12.0],
[100, 100, 100, 100],
timedelta(hours=1),
self.trading_environment
)
stddev = MovingStandardDev(
market_aware=False,
delta=timedelta(minutes=150),
)
self.source = SpecificEquityTrades(event_list=trade_history)
transformed = list(stddev.transform(self.source))
vals = [message.tnfm_value for message in transformed]
expected = [
None,
np.std([10.0, 15.0], ddof=1),
np.std([10.0, 15.0, 13.0], ddof=1),
np.std([15.0, 13.0, 12.0], ddof=1),
]
# np has odd rounding behavior, cf.
# http://docs.scipy.org/doc/np/reference/generated/np.std.html
for v1, v2 in zip(vals, expected):
if v1 is None:
assert v2 is None
continue
assert round(v1, 5) == round(v2, 5)
开发者ID:snth,项目名称:zipline,代码行数:34,代码来源:test_transforms.py
示例14: test_moving_stddev
def test_moving_stddev(self):
trade_history = factory.create_trade_history(
133,
[10.0, 15.0, 13.0, 12.0],
[100, 100, 100, 100],
timedelta(days=1),
self.sim_params
)
stddev = MovingStandardDev(
market_aware=True,
window_length=3,
)
self.source = SpecificEquityTrades(event_list=trade_history)
transformed = list(stddev.transform(self.source))
vals = [message[stddev.get_hash()] for message in transformed]
expected = [
None,
np.std([10.0, 15.0], ddof=1),
np.std([10.0, 15.0, 13.0], ddof=1),
np.std([15.0, 13.0, 12.0], ddof=1),
]
# np has odd rounding behavior, cf.
# http://docs.scipy.org/doc/np/reference/generated/np.std.html
for v1, v2 in zip(vals, expected):
if v1 is None:
self.assertIsNone(v2)
continue
self.assertEquals(round(v1, 5), round(v2, 5))
开发者ID:AmbroiseKritz,项目名称:zipline,代码行数:35,代码来源:test_transforms.py
示例15: setUp
def setUp(self):
self.trading_environment = factory.create_trading_environment()
setup_logger(self)
trade_history = factory.create_trade_history(
133, [10.0, 10.0, 11.0, 11.0], [100, 100, 100, 300], timedelta(days=1), self.trading_environment
)
self.source = SpecificEquityTrades(event_list=trade_history)
开发者ID:gawecoti,项目名称:zipline,代码行数:8,代码来源:test_transforms.py
示例16: setUp
def setUp(self):
self.sim_params = factory.create_simulation_parameters()
setup_logger(self)
trade_history = factory.create_trade_history(
133, [10.0, 10.0, 11.0, 11.0], [100, 100, 100, 300], timedelta(days=1), self.sim_params
)
self.source = SpecificEquityTrades(event_list=trade_history)
开发者ID:heber,项目名称:zipline,代码行数:8,代码来源:test_transforms.py
示例17: setUp
def setUp(self):
self.sim_params = factory.create_simulation_parameters(num_days=4)
trade_history = factory.create_trade_history(
133, [10.0, 10.0, 11.0, 11.0], [100, 100, 100, 300], timedelta(days=1), self.sim_params
)
self.source = SpecificEquityTrades(event_list=trade_history)
self.df_source, self.df = factory.create_test_df_source(self.sim_params)
开发者ID:hackliff,项目名称:zipline,代码行数:8,代码来源:test_algorithm.py
示例18: test_split_long_position
def test_split_long_position(self):
with trading.TradingEnvironment() as env:
events = factory.create_trade_history(
1,
[20, 20],
[100, 100],
oneday,
self.sim_params
)
# set up a long position in sid 1
# 100 shares at $20 apiece = $2000 position
events.insert(0, create_txn(events[0], 20, 100))
# set up a split with ratio 3
events.append(factory.create_split(1, 3,
env.next_trading_day(events[1].dt)))
results = calculate_results(self, events)
# should have 33 shares (at $60 apiece) and $20 in cash
self.assertEqual(2, len(results))
latest_positions = results[1]['daily_perf']['positions']
self.assertEqual(1, len(latest_positions))
# check the last position to make sure it's been updated
position = latest_positions[0]
self.assertEqual(1, position['sid'])
self.assertEqual(33, position['amount'])
self.assertEqual(60, position['cost_basis'])
self.assertEqual(60, position['last_sale_price'])
# since we started with $10000, and we spent $2000 on the
# position, but then got $20 back, we should have $8020
# (or close to it) in cash.
# we won't get exactly 8020 because sometimes a split is
# denoted as a ratio like 0.3333, and we lose some digits
# of precision. thus, make sure we're pretty close.
daily_perf = results[1]['daily_perf']
self.assertTrue(
zp_math.tolerant_equals(8020,
daily_perf['ending_cash'], 1))
for i, result in enumerate(results):
for perf_kind in ('daily_perf', 'cumulative_perf'):
perf_result = result[perf_kind]
# prices aren't changing, so pnl and returns should be 0.0
self.assertEqual(0.0, perf_result['pnl'],
"day %s %s pnl %s instead of 0.0" %
(i, perf_kind, perf_result['pnl']))
self.assertEqual(0.0, perf_result['returns'],
"day %s %s returns %s instead of 0.0" %
(i, perf_kind, perf_result['returns']))
开发者ID:ChrisBg,项目名称:zipline,代码行数:57,代码来源:test_perf_tracking.py
示例19: test_ending_before_pay_date
def test_ending_before_pay_date(self):
#post some trades in the market
events = factory.create_trade_history(
1,
[10, 10, 10, 10, 10],
[100, 100, 100, 100, 100],
oneday,
self.sim_params
)
dividend = factory.create_dividend(
1,
10.00,
events[0].dt,
events[1].dt,
events[-1].dt + 10 * oneday
)
buy_txn = create_txn(1, 10.0, 100, events[1].dt)
events.insert(2, buy_txn)
events.insert(1, dividend)
perf_tracker = perf.PerformanceTracker(self.sim_params)
all_events = (msg[1] for msg in heapq.merge(
((event.dt, event) for event in events),
((event.dt, event) for event in self.benchmark_events)))
transformed_events = list(perf_tracker.transform(
itertools.groupby(all_events, attrgetter('dt'))))
#flatten the list of events
results = []
for te in transformed_events:
for event in te[1]:
for message in event.perf_messages:
results.append(message)
perf_messages, risk = perf_tracker.handle_simulation_end()
results.append(perf_messages[0])
self.assertEqual(len(results), 5)
cumulative_returns = \
[event['cumulative_perf']['returns'] for event in results]
self.assertEqual(cumulative_returns, [0, 0, 0, 0.0, 0.0])
daily_returns = [event['daily_perf']['returns'] for event in results]
self.assertEqual(daily_returns, [0, 0, 0, 0, 0])
cash_flows = [event['daily_perf']['capital_used'] for event in results]
self.assertEqual(cash_flows, [0, -1000, 0, 0, 0])
cumulative_cash_flows = \
[event['cumulative_perf']['capital_used'] for event in results]
self.assertEqual(
cumulative_cash_flows,
[0, -1000, -1000, -1000, -1000]
)
开发者ID:paper-ghost,项目名称:zipline,代码行数:54,代码来源:test_perf_tracking.py
示例20: test_long_position_receives_dividend
def test_long_position_receives_dividend(self):
#post some trades in the market
events = factory.create_trade_history(
1,
[10, 10, 10, 10, 10],
[100, 100, 100, 100, 100],
oneday,
self.trading_environment
)
dividend = factory.create_dividend(
1,
10.00,
events[0].dt,
events[1].dt,
events[2].dt
)
events.insert(1, dividend)
txn = factory.create_txn(1, 10.0, 100, self.dt+oneday)
events[2].TRANSACTION = txn
perf_tracker = perf.PerformanceTracker(self.trading_environment)
transformed_events = list(perf_tracker.transform(
((event.dt, [event]) for event in events))
)
#flatten the list of events
results = []
for te in transformed_events:
for event in te[1]:
for message in event.perf_messages:
results.append(message)
perf_messages, risk = perf_tracker.handle_simulation_end()
results.append(perf_messages[0])
self.assertEqual(results[0]['daily_perf']['period_open'], events[0].dt)
self.assertEqual(
results[-1]['daily_perf']['period_open'],
events[-1].dt
)
self.assertEqual(len(results), 5)
cumulative_returns = \
[event['cumulative_perf']['returns'] for event in results]
self.assertEqual(cumulative_returns, [0.0, 0.0, 0.1, 0.1, 0.1])
daily_returns = [event['daily_perf']['returns'] for event in results]
self.assertEqual(daily_returns, [0.0, 0.0, 0.10, 0.0, 0.0])
cash_flows = [event['daily_perf']['capital_used'] for event in results]
self.assertEqual(cash_flows, [0, -1000, 1000, 0, 0])
cumulative_cash_flows = \
[event['cumulative_perf']['capital_used'] for event in results]
self.assertEqual(cumulative_cash_flows, [0, -1000, 0, 0, 0])
开发者ID:jtwaleson,项目名称:zipline,代码行数:53,代码来源:test_perf_tracking.py
注:本文中的zipline.utils.factory.create_trade_history函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论