计时器Timers – backtrader中文教程
选项
- 基于绝对时间输入或关于会话开始/结束时间的计时器
- 时间规范的时区规范,直接或通过 pytz兼容对象或通过数据馈送会话结束时间
- 关于指定时间的起始偏移量
- 重复间隔
- 工作日过滤器(带有结转选项)
- Monthdays 过滤器(带有结转选项)
- 自定义回调过滤器
使用模式
在Cerebro和Strategy子类中,定时器回调都将在以下方法中接收。
def notify_timer(self, timer, when, *args, **kwargs):
'''Receives a timer notification where ``timer`` is the timer which was
returned by ``add_timer``, and ``when`` is the calling time. ``args``
and ``kwargs`` are any additional arguments passed to ``add_timer``
The actual ``when`` time can be later, but the system may have not be
able to call the timer before. This value is the timer value and not the
system time.
'''
添加计时器 – 通过策略
def add_timer(self, when,
offset=datetime.timedelta(), repeat=datetime.timedelta(),
weekdays=[], weekcarry=False,
monthdays=[], monthcarry=True,
allow=None,
tzdata=None, cheat=False,
*args, **kwargs):
'''
它返回创建的Timer实例。
有关参数的说明,请参见下文。
添加计时器 – 通过 Cerebro
使用相同的方法完成,只需添加参数strats。如果设置为True定时器,不仅会通知到cerebro,还会通知到系统中运行的所有策略。
def add_timer(self, when,
offset=datetime.timedelta(), repeat=datetime.timedelta(),
weekdays=[], weekcarry=False,
monthdays=[], monthcarry=True,
allow=None,
tzdata=None, cheat=False, strats=False,
*args, **kwargs):
'''
它返回创建的Timer实例。
什么时候调用定时器
如果cheat=False
这是默认设置。在这种情况下,将调用一个计时器:
- 数据馈送加载当前柱的新值后
- 在经纪人评估订单并重新计算投资组合价值之后
- 在重新计算指标之前(因为这是由策略触发的)
- 在
next调用任何策略的任何方法之前
如果cheat=True
在这种情况下,将调用一个计时器:
- 数据馈送加载当前柱的新值后
- 在经纪人评估订单并重新计算投资组合价值之前
- 因此,在重新计算指标和
next调用任何策略的方法之前
例如,它允许以下带有每日柱的场景:
- 在经纪人评估新柱之前,将调用计时器
- 指标具有前一天收盘时的值,可用于生成进入/退出信号(或在最后一次评估期间可能已设置标志
next) - 因为有新的价格可用,所以可以使用开盘价计算本金。这假设一个人
open从观看开盘拍卖中得到了一个很好的指示。
与每日bar一起运行
示例 schedule.py 默认使用 backtrader 分发中可用的标准每日柱线运行。策略参数是:
class St(bt.Strategy):
params = dict(
when=bt.timer.SESSION_START,
timer=True,
cheat=False,
offset=datetime.timedelta(),
repeat=datetime.timedelta(),
weekdays=[],
)
并且数据具有以下会话时间:
- 开始: 09:00
- 结束时间:17:30
只运行一次
$ ./scheduled.py --strat when='datetime.time(15,30)' strategy notify_timer with tid 0, when 2005-01-03 15:30:00 cheat False 1, 2005-01-03 17:30:00, Week 1, Day 1, O 2952.29, H 2989.61, L 2946.8, C 2970.02 strategy notify_timer with tid 0, when 2005-01-04 15:30:00 cheat False 2, 2005-01-04 17:30:00, Week 1, Day 2, O 2969.78, H 2979.88, L 2961.14, C 2971.12 strategy notify_timer with tid 0, when 2005-01-05 15:30:00 cheat False 3, 2005-01-05 17:30:00, Week 1, Day 3, O 2969.0, H 2969.0, L 2942.69, C 2947.19 strategy notify_timer with tid 0, when 2005-01-06 15:30:00 cheat False ...
正如指定的那样,计时器正在滴答作响15:30。那里没有惊喜。让我们添加一个 30 分钟的偏移量。
$ ./scheduled.py --strat when='datetime.time(15,30)',offset='datetime.timedelta(minutes=30)' strategy notify_timer with tid 0, when 2005-01-03 16:00:00 cheat False 1, 2005-01-03 17:30:00, Week 1, Day 1, O 2952.29, H 2989.61, L 2946.8, C 2970.02 strategy notify_timer with tid 0, when 2005-01-04 16:00:00 cheat False 2, 2005-01-04 17:30:00, Week 1, Day 2, O 2969.78, H 2979.88, L 2961.14, C 2971.12 strategy notify_timer with tid 0, when 2005-01-05 16:00:00 cheat False ...
并且时间已经从15:30变为16:00为计时器。没有惊喜。让我们做同样的事情,但引用会话的开始。
$ ./scheduled.py --strat when='bt.timer.SESSION_START',offset='datetime.timedelta(minutes=30)' strategy notify_timer with tid 0, when 2005-01-03 09:30:00 cheat False 1, 2005-01-03 17:30:00, Week 1, Day 1, O 2952.29, H 2989.61, L 2946.8, C 2970.02 strategy notify_timer with tid 0, when 2005-01-04 09:30:00 cheat False 2, 2005-01-04 17:30:00, Week 1, Day 2, O 2969.78, H 2979.88, L 2961.14, C 2971.12 ...
瞧!调用回调的时间是09:30。会话开始,见上文,是09:00. 这使得能够简单地说一个人想要在会话开始后30 分钟执行一个动作。
让我们添加一个重复:
$ ./scheduled.py --strat when='bt.timer.SESSION_START',offset='datetime.timedelta(minutes=30)',repeat='datetime.timedelta(minutes=30)' strategy notify_timer with tid 0, when 2005-01-03 09:30:00 cheat False 1, 2005-01-03 17:30:00, Week 1, Day 1, O 2952.29, H 2989.61, L 2946.8, C 2970.02 strategy notify_timer with tid 0, when 2005-01-04 09:30:00 cheat False 2, 2005-01-04 17:30:00, Week 1, Day 2, O 2969.78, H 2979.88, L 2961.14, C 2971.12 strategy notify_timer with tid 0, when 2005-01-05 09:30:00 cheat False ...
没有重复。原因是价格的决议是每天的。与前面的示例一样,第一次调用计时器。09:30但是当系统获得下一批价格时,它们是在第二天发生的。显然,计时器只能被调用一次。需要较低的分辨率。
但在转向较低的分辨率之前,让我们通过在会话结束前调用计时器来作弊。
$ ./scheduled.py --strat when='bt.timer.SESSION_START',cheat=True strategy notify_timer with tid 1, when 2005-01-03 09:00:00 cheat True -- 2005-01-03 Create buy order strategy notify_timer with tid 0, when 2005-01-03 09:00:00 cheat False 1, 2005-01-03 17:30:00, Week 1, Day 1, O 2952.29, H 2989.61, L 2946.8, C 2970.02 strategy notify_timer with tid 1, when 2005-01-04 09:00:00 cheat True strategy notify_timer with tid 0, when 2005-01-04 09:00:00 cheat False -- 2005-01-04 Buy Exec @ 2969.78 2, 2005-01-04 17:30:00, Week 1, Day 2, O 2969.78, H 2979.88, L 2961.14, C 2971.12 strategy notify_timer with tid 1, when 2005-01-05 09:00:00 cheat True strategy notify_timer with tid 0, when 2005-01-05 09:00:00 cheat False ...
该策略添加了第二个计时器cheat=True。这是第 2 次添加的,因此将收到第 2 次( tid计时器id),即1(请参阅上面的示例中分配的tid是0)
And1被调用 before 0,因为该计时器作弊并且在系统中的许多事件发生之前被调用(参见上面的解释)
由于价格的每日分辨率,它并没有太大的区别,除了:
- 该策略还在开盘前发出订单……并在第二天与开盘价匹配
这即使在开盘前作弊,仍然是正常的行为,因为在经纪商中也没有激活开盘作弊。
相同,但coo=True对于经纪人
$ ./scheduled.py --strat when='bt.timer.SESSION_START',cheat=True --broker coo=True strategy notify_timer with tid 1, when 2005-01-03 09:00:00 cheat True -- 2005-01-03 Create buy order strategy notify_timer with tid 0, when 2005-01-03 09:00:00 cheat False -- 2005-01-03 Buy Exec @ 2952.29 1, 2005-01-03 17:30:00, Week 1, Day 1, O 2952.29, H 2989.61, L 2946.8, C 2970.02 strategy notify_timer with tid 1, when 2005-01-04 09:00:00 cheat True strategy notify_timer with tid 0, when 2005-01-04 09:00:00 cheat False 2, 2005-01-04 17:30:00, Week 1, Day 2, O 2969.78, H 2979.88, L 2961.14, C 2971.12 strategy notify_timer with tid 1, when 2005-01-05 09:00:00 cheat True strategy notify_timer with tid 0, when 2005-01-05 09:00:00 cheat False ...
有些事情发生了变化。
- 命令
2005-01-03在作弊计时器上发出 - 订单
2005-01-03以开盘价执行就像一个人在市场真正开盘前几秒钟对开盘价采取行动一样。
用 5 分钟的小节跑步
该示例默认使用backtrader分布scheduled-min.py中可用的标准 5 分钟柱线运行。策略的参数被扩展为包括进位选项monthdays
class St(bt.Strategy):
params = dict(
when=bt.timer.SESSION_START,
timer=True,
cheat=False,
offset=datetime.timedelta(),
repeat=datetime.timedelta(),
weekdays=[],
weekcarry=False,
monthdays=[],
monthcarry=True,
)
数据具有相同的会话时间:
- 开始: 09:00
- 结束时间:17:30
让我们做一些实验。首先是一个计时器。
$ ./scheduled-min.py --strat when='datetime.time(15, 30)' 1, 2006-01-02 09:05:00, Week 1, Day 1, O 3578.73, H 3587.88, L 3578.73, C 3582.99 2, 2006-01-02 09:10:00, Week 1, Day 1, O 3583.01, H 3588.4, L 3583.01, C 3588.03 ... 77, 2006-01-02 15:25:00, Week 1, Day 1, O 3599.07, H 3599.68, L 3598.47, C 3599.68 strategy notify_timer with tid 0, when 2006-01-02 15:30:00 cheat False 78, 2006-01-02 15:30:00, Week 1, Day 1, O 3599.64, H 3599.73, L 3599.0, C 3599.67 ... 179, 2006-01-03 15:25:00, Week 1, Day 2, O 3634.72, H 3635.0, L 3634.06, C 3634.87 strategy notify_timer with tid 0, when 2006-01-03 15:30:00 cheat False 180, 2006-01-03 15:30:00, Week 1, Day 2, O 3634.81, H 3634.89, L 3634.04, C 3634.23 ...
计时器按要求启动15:30。该日志显示了它在第 1天两天内是如何做到的。
在混合物repeat中添加一个15 minutes
$ ./scheduled-min.py --strat when='datetime.time(15, 30)',repeat='datetime.timedelta(minutes=15)' ... 74, 2006-01-02 15:10:00, Week 1, Day 1, O 3596.12, H 3596.63, L 3595.92, C 3596.63 75, 2006-01-02 15:15:00, Week 1, Day 1, O 3596.36, H 3596.65, L 3596.19, C 3596.65 76, 2006-01-02 15:20:00, Week 1, Day 1, O 3596.53, H 3599.13, L 3596.12, C 3598.9 77, 2006-01-02 15:25:00, Week 1, Day 1, O 3599.07, H 3599.68, L 3598.47, C 3599.68 strategy notify_timer with tid 0, when 2006-01-02 15:30:00 cheat False 78, 2006-01-02 15:30:00, Week 1, Day 1, O 3599.64, H 3599.73, L 3599.0, C 3599.67 79, 2006-01-02 15:35:00, Week 1, Day 1, O 3599.61, H 3600.29, L 3599.52, C 3599.92 80, 2006-01-02 15:40:00, Week 1, Day 1, O 3599.96, H 3602.06, L 3599.76, C 3602.05 strategy notify_timer with tid 0, when 2006-01-02 15:45:00 cheat False 81, 2006-01-02 15:45:00, Week 1, Day 1, O 3601.97, H 3602.07, L 3601.45, C 3601.83 82, 2006-01-02 15:50:00, Week 1, Day 1, O 3601.74, H 3602.8, L 3601.63, C 3602.8 83, 2006-01-02 15:55:00, Week 1, Day 1, O 3602.53, H 3602.74, L 3602.33, C 3602.61 strategy notify_timer with tid 0, when 2006-01-02 16:00:00 cheat False 84, 2006-01-02 16:00:00, Week 1, Day 1, O 3602.58, H 3602.75, L 3601.81, C 3602.14 85, 2006-01-02 16:05:00, Week 1, Day 1, O 3602.16, H 3602.16, L 3600.86, C 3600.96 86, 2006-01-02 16:10:00, Week 1, Day 1, O 3601.2, H 3601.49, L 3600.94, C 3601.27 ... strategy notify_timer with tid 0, when 2006-01-02 17:15:00 cheat False 99, 2006-01-02 17:15:00, Week 1, Day 1, O 3603.96, H 3603.96, L 3602.89, C 3603.79 100, 2006-01-02 17:20:00, Week 1, Day 1, O 3603.94, H 3605.95, L 3603.87, C 3603.91 101, 2006-01-02 17:25:00, Week 1, Day 1, O 3604.0, H 3604.76, L 3603.85, C 3604.64 strategy notify_timer with tid 0, when 2006-01-02 17:30:00 cheat False 102, 2006-01-02 17:30:00, Week 1, Day 1, O 3604.06, H 3604.41, L 3603.95, C 3604.33 103, 2006-01-03 09:05:00, Week 1, Day 2, O 3604.08, H 3609.6, L 3604.08, C 3609.6 104, 2006-01-03 09:10:00, Week 1, Day 2, O 3610.34, H 3617.31, L 3610.34, C 3617.31 105, 2006-01-03 09:15:00, Week 1, Day 2, O 3617.61, H 3617.87, L 3616.03, C 3617.51 106, 2006-01-03 09:20:00, Week 1, Day 2, O 3617.24, H 3618.86, L 3616.09, C 3618.42 ... 179, 2006-01-03 15:25:00, Week 1, Day 2, O 3634.72, H 3635.0, L 3634.06, C 3634.87 strategy notify_timer with tid 0, when 2006-01-03 15:30:00 cheat False 180, 2006-01-03 15:30:00, Week 1, Day 2, O 3634.81, H 3634.89, L 3634.04, C 3634.23 ...
正如预期的那样,第一个呼叫被触发,然后开始每 15 分钟重复一次,直到会话结束。当新会话开始时,计时器已重新设置为。15:3017:3015:30
现在在会议开始前作弊
$ ./scheduled-min.py --strat when='bt.timer.SESSION_START',cheat=True strategy notify_timer with tid 1, when 2006-01-02 09:00:00 cheat True -- 2006-01-02 09:05:00 Create buy order strategy notify_timer with tid 0, when 2006-01-02 09:00:00 cheat False 1, 2006-01-02 09:05:00, Week 1, Day 1, O 3578.73, H 3587.88, L 3578.73, C 3582.99 -- 2006-01-02 09:10:00 Buy Exec @ 3583.01 2, 2006-01-02 09:10:00, Week 1, Day 1, O 3583.01, H 3588.4, L 3583.01, C 3588.03 ...
09:05:00订单创建和执行时间为 t ,09:10:00因为经纪人未处于打开时作弊模式。让我们设置它…
$ ./scheduled-min.py --strat when='bt.timer.SESSION_START',cheat=True --broker coo=True strategy notify_timer with tid 1, when 2006-01-02 09:00:00 cheat True -- 2006-01-02 09:05:00 Create buy order strategy notify_timer with tid 0, when 2006-01-02 09:00:00 cheat False -- 2006-01-02 09:05:00 Buy Exec @ 3578.73 1, 2006-01-02 09:05:00, Week 1, Day 1, O 3578.73, H 3587.88, L 3578.73, C 3582.99 2, 2006-01-02 09:10:00, Week 1, Day 1, O 3583.01, H 3588.4, L 3583.01, C 3588.03 ...
发行时间和执行时间均以09:05:00执行价格为开盘价09:05:00。
其他场景
计时器允许通过传递天数列表(遵循 ISO 规范的整数,其中 Mon=1 和 Sun=7)来指定必须在哪些天执行,如
与此类似,可以决定在每个月的第 15 天采取行动:
monthdays=[15]如果第 15天恰好是非交易日,并且计时器应该在下一个交易日开始,可以添加
monthcarry=True
没有针对以下内容的实施: 3月、6 月、9 月和 12 月的第三个星期五(期货/期权到期),但有可能通过以下方式实施规则:
allow=callable可调用对象接受datetime.date实例的地方。请注意,这不是一个datetime.datetime实例,因为 allow callable 仅用于决定给定日期是否适合计时器。要实现类似于上面列出的规则:
class FutOpExp(object):
def __init__(self):
self.fridays = 0
self.curmonth = -1
def __call__(self, d):
_, _, isowkday = d.isocalendar()
if d.month != self.curmonth:
self.curmonth = d.month
self.fridays = 0
# Mon=1 ... Sun=7
if isowkday == 5 and self.curmonth in [3, 6, 9, 12]:
self.fridays += 1
if self.friday == 3: # 3rd Friday
return True # timer allowed
return False # timer disallowed
参数为add_timer
* `when`:可以是 * `datetime.time` 实例(见下文`tzdata`) * `bt.timer.SESSION_START` 引用会话开始 * `bt.timer.SESSION_END` 引用会话结束
offset这必须是一个datetime.timedelta实例
用于偏移值when。SESSION_START它与和结合使用具有有意义的用途,以指示在会话开始后SESSION_END调用计时器之类的事情。15 minutes
repeat这必须是一个datetime.timedelta实例
指示在第一次通话后,是否将在同一会话中以预定的repeat增量安排进一步的通话
一旦计时器超过会话结束,它将重置为原始值when
weekdays: 一个排序的迭代器,带有整数,指示在哪些天(iso 代码,星期一是 1,星期日是 7)可以实际调用计时器
如果未指定,计时器将在所有日子都处于活动状态
weekcarry(默认值:False)。如果True没有看到工作日(例如:交易假期),计时器将在第二天执行(即使在新的一周)monthdays:一个排序的迭代器,带有整数,指示必须在每月的哪几天执行计时器。例如总是在每月的第15天
如果未指定,计时器将在所有日子都处于活动状态
monthcarry(默认值:True)。如果没有看到这一天(周末、交易假日),则计时器将在下一个可用日期执行。allow(默认值:None)。一个回调,它接收一个 datetime.date` 实例,True如果计时器允许该日期,则返回,否则返回Falsetzdata可以是None(默认)、pytz实例或data feed实例。
None:when以面值解释(这意味着将其视为 UTC 处理,即使它不是)
pytz实例:when将被解释为在时区实例指定的本地时间中指定。
data feed实例:将被解释为在数据馈送实例when的参数指定的本地时间中指定。tz
Tips:如果 `when` 是 `SESSION_START` 或 `SESSION_END` 并且 `tzdata` 是`None`,则系统中的第一个 *data feed*(又名 `self.data0`)将用作查找会话的参考次。
strats(默认False)也调用notify_timer策略cheat(默认False)如果True在经纪人有机会评估订单之前调用计时器。这打开了基于开盘价发出订单的机会,例如在会话开始之前*args: 任何额外的参数都将传递给notify_timer**kwargs: 任何额外的 kwargs 将被传递给notify_timer
示例使用scheduled.py
$ ./scheduled.py --help
usage: scheduled.py [-h] [--data0 DATA0] [--fromdate FROMDATE]
[--todate TODATE] [--cerebro kwargs] [--broker kwargs]
[--sizer kwargs] [--strat kwargs] [--plot [kwargs]]
Sample Skeleton
optional arguments:
-h, --help show this help message and exit
--data0 DATA0 Data to read in (default:
../../datas/2005-2006-day-001.txt)
--fromdate FROMDATE Date[time] in YYYY-MM-DD[THH:MM:SS] format (default: )
--todate TODATE Date[time] in YYYY-MM-DD[THH:MM:SS] format (default: )
--cerebro kwargs kwargs in key=value format (default: )
--broker kwargs kwargs in key=value format (default: )
--sizer kwargs kwargs in key=value format (default: )
--strat kwargs kwargs in key=value format (default: )
--plot [kwargs] kwargs in key=value format (default: )
示例使用scheduled-min.py
$ ./scheduled-min.py --help
usage: scheduled-min.py [-h] [--data0 DATA0] [--fromdate FROMDATE]
[--todate TODATE] [--cerebro kwargs] [--broker kwargs]
[--sizer kwargs] [--strat kwargs] [--plot [kwargs]]
Timer Test Intraday
optional arguments:
-h, --help show this help message and exit
--data0 DATA0 Data to read in (default: ../../datas/2006-min-005.txt)
--fromdate FROMDATE Date[time] in YYYY-MM-DD[THH:MM:SS] format (default: )
--todate TODATE Date[time] in YYYY-MM-DD[THH:MM:SS] format (default: )
--cerebro kwargs kwargs in key=value format (default: )
--broker kwargs kwargs in key=value format (default: )
--sizer kwargs kwargs in key=value format (default: )
--strat kwargs kwargs in key=value format (default: )
--plot [kwargs] kwargs in key=value format (default: )
样品来源scheduled.py
from __future__ import (absolute_import, division, print_function,
unicode_literals)
import argparse
import datetime
import backtrader as bt
class St(bt.Strategy):
params = dict(
when=bt.timer.SESSION_START,
timer=True,
cheat=False,
offset=datetime.timedelta(),
repeat=datetime.timedelta(),
weekdays=[],
)
def __init__(self):
bt.ind.SMA()
if self.p.timer:
self.add_timer(
when=self.p.when,
offset=self.p.offset,
repeat=self.p.repeat,
weekdays=self.p.weekdays,
)
if self.p.cheat:
self.add_timer(
when=self.p.when,
offset=self.p.offset,
repeat=self.p.repeat,
cheat=True,
)
self.order = None
def prenext(self):
self.next()
def next(self):
_, isowk, isowkday = self.datetime.date().isocalendar()
txt = '{}, {}, Week {}, Day {}, O {}, H {}, L {}, C {}'.format(
len(self), self.datetime.datetime(),
isowk, isowkday,
self.data.open[0], self.data.high[0],
self.data.low[0], self.data.close[0])
print(txt)
def notify_timer(self, timer, when, *args, **kwargs):
print('strategy notify_timer with tid {}, when {} cheat {}'.
format(timer.p.tid, when, timer.p.cheat))
if self.order is None and timer.p.cheat:
print('-- {} Create buy order'.format(self.data.datetime.date()))
self.order = self.buy()
def notify_order(self, order):
if order.status == order.Completed:
print('-- {} Buy Exec @ {}'.format(
self.data.datetime.date(), order.executed.price))
def runstrat(args=None):
args = parse_args(args)
cerebro = bt.Cerebro()
# Data feed kwargs
kwargs = dict(
timeframe=bt.TimeFrame.Days,
compression=1,
sessionstart=datetime.time(9, 0),
sessionend=datetime.time(17, 30),
)
# Parse from/to-date
dtfmt, tmfmt = '%Y-%m-%d', 'T%H:%M:%S'
for a, d in ((getattr(args, x), x) for x in ['fromdate', 'todate']):
if a:
strpfmt = dtfmt + tmfmt * ('T' in a)
kwargs[d] = datetime.datetime.strptime(a, strpfmt)
# Data feed
data0 = bt.feeds.BacktraderCSVData(dataname=args.data0, **kwargs)
cerebro.adddata(data0)
# Broker
cerebro.broker = bt.brokers.BackBroker(**eval('dict(' + args.broker + ')'))
# Sizer
cerebro.addsizer(bt.sizers.FixedSize, **eval('dict(' + args.sizer + ')'))
# Strategy
cerebro.addstrategy(St, **eval('dict(' + args.strat + ')'))
# Execute
cerebro.run(**eval('dict(' + args.cerebro + ')'))
if args.plot: # Plot if requested to
cerebro.plot(**eval('dict(' + args.plot + ')'))
def parse_args(pargs=None):
parser = argparse.ArgumentParser(
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
description=(
'Sample Skeleton'
)
)
parser.add_argument('--data0', default='../../datas/2005-2006-day-001.txt',
required=False, help='Data to read in')
# Defaults for dates
parser.add_argument('--fromdate', required=False, default='',
help='Date[time] in YYYY-MM-DD[THH:MM:SS] format')
parser.add_argument('--todate', required=False, default='',
help='Date[time] in YYYY-MM-DD[THH:MM:SS] format')
parser.add_argument('--cerebro', required=False, default='',
metavar='kwargs', help='kwargs in key=value format')
parser.add_argument('--broker', required=False, default='',
metavar='kwargs', help='kwargs in key=value format')
parser.add_argument('--sizer', required=False, default='',
metavar='kwargs', help='kwargs in key=value format')
parser.add_argument('--strat', required=False, default='',
metavar='kwargs', help='kwargs in key=value format')
parser.add_argument('--plot', required=False, default='',
nargs='?', const='{}',
metavar='kwargs', help='kwargs in key=value format')
return parser.parse_args(pargs)
if __name__ == '__main__':
runstrat()
样品来源scheduled-min.py
from __future__ import ( absolute_import , division , print_function ,
unicode_literals )
import argparse
import datetime
import backtrader as bt
class St ( bt . Strategy ):
params = dict (
when = bt . timer . SESSION_START ,
timer = True ,
cheat = False ,
offset =日期时间。timedelta (),
重复=日期时间。timedelta (),
weekdays = [],
weekcarry = False ,
monthdays = [],
monthcarry = True ,
)
def __init__ ( self ):
bt 。印第安纳州 SMA ()
如果 自我。磷。计时器:
自我。add_timer (
when = self . p. 当,
偏移=自我。磷。偏移量,
重复=自我。磷。重复,
工作日=自我。磷。工作日,
weekcarry =自我。磷。weekcarry ,
monthdays = self 。磷。月份,monthcarry =自我。磷。monthcarry , # tzdata=self.data0,
)
如果 自我。磷。作弊:
自我。add_timer (
when = self . p . when ,
offset = self . p . offset ,
repeat = self . p . repeat ,
weekdays = self . p . weekdays ,
weekcarry = self . p . weekcarry ,
monthdays =自我。磷。月份,monthcarry =自我。磷。monthcarry , # tzdata=self.data0, cheat = True , ) self 。order =无def prenext ( self ): self 。next () def next ( self ): _ , isowk , isowkday = self 。日期时间。日期() 。
isocalendar()
txt = '{}, {}, Week {}, Day {}, O {}, H {}, L {}, C {}'.format(
len(self), self.datetime.datetime(),
isowk, isowkday,
self.data.open[0], self.data.high[0],
self.data.low[0], self.data.close[0])
print(txt)
def notify_timer(self, timer, when, *args, **kwargs):
print('strategy notify_timer with tid {}, when {} cheat {}'.
format(timer.p.tid, when, timer.p.cheat))
if self.order is None and timer.params.cheat:
print('-- {} Create buy order'.format(
self.data.datetime.datetime()))
self.order = self.buy()
def notify_order(self, order):
if order.status == order.Completed:
print('-- {} Buy Exec @ {}'.format(
self.data.datetime.datetime(), order.executed.price))
def runstrat(args=None):
args = parse_args(args)
cerebro = bt.Cerebro()
# Data feed kwargs
kwargs = dict(
timeframe=bt.TimeFrame.Minutes,
compression=5,
sessionstart=datetime.time(9, 0),
sessionend=datetime.time(17, 30),
)
# Parse from/to-date
dtfmt, tmfmt = '%Y-%m-%d', 'T%H:%M:%S'
for a, d in ((getattr(args, x), x) for x in ['fromdate', 'todate']):
if a:
strpfmt = dtfmt + tmfmt * ('T' in a)
kwargs[d] = datetime.datetime.strptime(a, strpfmt)
# Data feed
data0 = bt.feeds.BacktraderCSVData(dataname=args.data0, **kwargs)
cerebro.adddata(data0)
# Broker
cerebro.broker = bt.brokers.BackBroker(**eval('dict(' + args.broker + ')'))
# Sizer
cerebro.addsizer(bt.sizers.FixedSize, **eval('dict(' + args.sizer + ')'))
# Strategy
cerebro.addstrategy(St, **eval('dict(' + args.strat + ')'))
# Execute
cerebro.run(**eval('dict(' + args.cerebro + ')'))
if args.plot: # Plot if requested to
cerebro.plot(**eval('dict(' + args.plot + ')'))
def parse_args(pargs=None):
parser = argparse.ArgumentParser(
formatter_class=argparse.ArgumentDefaultsHelpFormatter,
description=(
'Timer Test Intraday'
)
)
parser.add_argument('--data0', default='../../datas/2006-min-005.txt',
required=False, help='Data to read in')
# Defaults for dates
parser.add_argument('--fromdate', required=False, default='',
help='Date[time] in YYYY-MM-DD[THH:MM:SS] format')
parser.add_argument('--todate', required=False, default='',
help='Date[time] in YYYY-MM-DD[THH:MM:SS] format')
parser.add_argument('--cerebro', required=False, default='',
metavar='kwargs', help='kwargs in key=value format')
parser.add_argument('--broker', required=False, default='',
metavar='kwargs', help='kwargs in key=value format')
parser.add_argument('--sizer', required=False, default='',
metavar='kwargs', help='kwargs in key=value format' )
解析器。add_argument ( '--strat' , required = False , default = '' ,
metavar = 'kwargs' , help = 'kwargs in key=value format' )
解析器。add_argument ( '--plot' , required = False , default = '' ,
nargs = '?' , const = '{}' ,
metavar ='kwargs' , help = 'kwargs in key=value format' )
返回 parser 。parse_args ( pargs )
if __name__ == '__main__' :
runstrat ()