计时器
原文： https://www.backtrader.com/blog/posts/2017-05-03-timers/timers/
发行版1.9.44.116将计时器添加到backtrader中可用的工具库中。此功能允许在给定时间点通过细粒度的最终用户控制回拨notify_timer（可在Cerebro和Strategy中使用）。
笔记
在1.9.46.116中进行了一些更正
选择权
-
基于绝对时间输入或会话开始/结束时间的计时器
-
时间规范的时区规范，可以是直接或通过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实例。
有关参数的说明，请参见下文。
添加计时器-通过大脑
使用相同的方法完成，只需添加参数strats。如果设置为True，计时器不仅会通知大脑，还会通知系统中运行的所有策略。
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的良好指示。
与每日酒吧一起跑步
样本scheduled.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为 1st时间调用计时器，如前一示例中所示。但是，当系统获得下一批价格时，它们将在第二天发生。显然，计时器只能调用一次。需要更低的分辨率。
但是在转到较低的分辨率之前，让我们在会话结束之前调用计时器来作弊。
$ ./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
...
该策略添加了一个 2和计时器，并带有cheat=True。这增加了 2nd，因此将接收到 2ndtid（定时器 id），即1（参见上述示例中分配的tid为0）
而1在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 分钟的棒跑
样本scheduled-min.py默认使用backtrader发行版中可用的标准 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处按要求启动。日志显示了它在 1st两天内是如何做到这一点的。
在混合物中加入15 minutes的repeat
$ ./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
...
正如所料，1st呼叫在15:30触发，然后开始每 15 分钟重复一次，直到会话在17:30结束。当新会话开始时，计时器已再次重置为15: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
...
订单创建为 t09:05:00并在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）来指定必须在哪几天执行计时器，如中所示
-
weekdays=[5]这将要求计时器仅在星期五有效如果周五是非交易日，计时器应在下一个交易日启动，则可以添加
weekcarry=True
与此类似，您可以决定在每月的 15日采取行动，包括：
-
monthdays=[15]如果第 15次恰巧是非交易日，且计时器应在下一个交易日启动，则可以添加
monthcarry=True
类似于：3 月、6 月、9 月和 12 月的第三条周五（期货/期权到期日）等事项没有实施，但有可能通过以下方式实施规则：
-
allow=callable其中可调用对象接受datetime.date实例。请注意，这不是一个datetime.datetime实例，因为允许可调用仅用于决定给定日期是否适合计时器。要实施类似于上述规则的内容：
```py 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```
我们可以通过
allow=FutOpeExp()来创建计时器这将允许计时器在这几个月的第 3日周五开始计时，并可能在期货到期前平仓。
add_timer的参数
- `when`: can be
- `datetime.time` instance (see below `tzdata`)
- `bt.timer.SESSION_START` to reference a session start
- `bt.timer.SESSION_END` to reference a session end
-
offset必须是datetime.timedelta实例用于偏移值
when。它与SESSION_START和SESSION_END结合使用，有意义地表示在会话开始后调用15 minutes计时器之类的事情。repeat必须是datetime.timedelta实例
指示在 1st呼叫后，是否会在同一会话中以预定的
repeat增量安排更多呼叫一旦计时器在会话结束后被重置为
when的原始值weekdays：一个排序的iterable，带有整数，表示可以实际调用计时器的日期（iso 代码，周一是 1，周日是 7）
如果未指定，计时器将在所有日期都处于活动状态
-
weekcarry（默认为False）。如果True和工作日不可见（例如：交易假日），计时器将在第二天执行（即使在新的一周内） -
monthdays：一个排序的iterable，整数表示计时器必须在月份的哪几天执行。例如，始终在当月的第15天
如果未指定，计时器将在所有日期都处于活动状态
-
monthcarry（默认为True）。如果看不到该日（周末、交易日），计时器将在下一个可用日执行。 -
allow（默认为None）。接收 datetime.date实例并返回True的回调，如果允许计时器使用该日期，则返回False` -
tzdata可以是None（默认）、一个pytz实例或一个data feed实例。
None：when按面值解释（这意味着即使不是 UTC，也要像对待 UTC 一样处理）pytz实例：when将被解释为在时区实例指定的本地时间内指定。data feed实例：when将被解释为数据馈送实例的tz参数指定的本地时间。注：若
when为SESSION_START或py `SESSION_END` and `tzdata` is `None`, the 1st *data feed* in the system (aka `self.data0`) will be used as the reference to find out the session times.-
strats（默认值：False）同时调用策略的notify_timer -
cheat（默认False）如果True计时器将在经纪人有机会评估订单之前调用。这样就有机会根据开盘价发出订单，例如在会议开始之前 -
\*args：任何额外参数都将传递给notify_timer -
\*\*kwargs：任何额外的 KWARG 将传递给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=datetime.timedelta(),
repeat=datetime.timedelta(),
weekdays=[],
weekcarry=False,
monthdays=[],
monthcarry=True,
)
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,
weekcarry=self.p.weekcarry,
monthdays=self.p.monthdays,
monthcarry=self.p.monthcarry,
# tzdata=self.data0,
)
if self.p.cheat:
self.add_timer(
when=self.p.when,
offset=self.p.offset,
repeat=self.p.repeat,
weekdays=self.p.weekdays,
weekcarry=self.p.weekcarry,
monthdays=self.p.monthdays,
monthcarry=self.p.monthcarry,
tzdata=self.data0,
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.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')
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()
