import backtrader as bt
import datetime
from backtrader_plotting import Bokeh
from backtrader_plotting.schemes import Tradimo
# ===============载入自己的数据===========================================
#需要设定每个列的含义,比如开盘价在第4列,则open=3(从0开始算起)===
# data = bt.feeds.GenericCSVData(
# dataname='数据文件所在位置',
# datetime=2,
# open=3,
# high=4,
# low=5,
# close=6,
# volume=10,
# dtformat=('%Y%m%d'),
# fromdate=datetime(2010, 1, 1),
# todate=datetime(2020, 4, 12)
# )
# =============================================================================
"""
构建策略
"""
# ========================自己的策略===============================================
# class TestStrategy(bt.Strategy):
# """
# 继承并构建自己的bt策略
# """
#
# def log(self, txt, dt=None, doprint=False):
# ''' 日志函数,用于统一输出日志格式 '''
# if doprint:
# dt = dt or self.datas[0].datetime.date(0)
# print('%s, %s' % (dt.isoformat(), txt))
# =============================================================================
# ===============均线金叉死叉策略 ======================================================
class TestStrategy(bt.Strategy):
"""
继承并构建自己的bt策略
"""
def log(self, txt, dt=None, doprint=False):
''' 日志函数,用于统一输出日志格式 '''
if doprint:
dt = dt or self.datas[0].datetime.date(0)
print('%s, %s' % (dt.isoformat(), txt))
def __init__(self):
# 初始化相关数据
self.dataclose = self.datas[0].close
self.order = None
self.buyprice = None
self.buycomm = None
#添加指标
# 五日移动平均线
self.sma5 = bt.indicators.SimpleMovingAverage(
self.datas[0], period=5)
# 十日移动平均线
self.sma10 = bt.indicators.SimpleMovingAverage(
self.datas[0], period=10)
# =============================================================================
# self.dataclose[0] # 当日的收盘价
# self.dataclose[-1] # 昨天的收盘价
# self.dataclose[-2] # 前天的收盘价
# =============================================================================
def notify_order(self, order):
"""
订单状态处理
Arguments:
order {object} -- 订单状态
"""
if order.status in [order.Submitted, order.Accepted]:
# 如订单已被处理,则不用做任何事情
return
# 检查订单是否完成
if order.status in [order.Completed]:
if order.isbuy():
self.buyprice = order.executed.price
self.buycomm = order.executed.comm
self.bar_executed = len(self)
# 订单因为缺少资金之类的原因被拒绝执行
elif order.status in [order.Canceled, order.Margin, order.Rejected]:
self.log('Order Canceled/Margin/Rejected')
# 订单状态处理完成,设为空
self.order = None
def notify_trade(self, trade):
"""
交易成果
Arguments:
trade {object} -- 交易状态
"""
if not trade.isclosed:
return
# 显示交易的毛利率和净利润
self.log('OPERATION PROFIT, GROSS %.2f, NET %.2f' %
(trade.pnl, trade.pnlcomm), doprint=True)
def next(self):
''' 下一次执行 '''
# 记录收盘价
self.log('Close, %.2f' % self.dataclose[0])
# 是否正在下单,如果是的话不能提交第二次订单
if self.order:
return
# 是否已经买入
if not self.position:
# 还没买,如果 MA5 > MA10 说明涨势,买入
if self.sma5[0] > self.sma10[0]:
self.order = self.buy()
else:
# 已经买了,如果 MA5 < MA10 ,说明跌势,卖出
if self.sma5[0] < self.sma10[0]:
self.order = self.sell()
def stop(self):
self.log(u' Ending Value: %.2f' %
(self.broker.getvalue()), doprint=True)
# =============================================================================
''' ====================MACD策略-> 信号线交叉交易法======================'''
#MACD线(平滑异同平均):价格短期EMA-价格高长期EMA ,一般是EMA(12)-EMA(26)
#信号线=MACD的EMA(9)周期
#MACD利用收盘价的短期指数移动平均线(DIF)与长期指数移动平均线(DEA)之间的聚合与分离状况,
#对买进、卖出时机作出研判的技术指标。MACD可以作为中长线投资者的首选参考指标。
#但由于该指标具有滞后性,当行情迅速大幅涨跌时,不适用。
##参考用法:
# 1.DIF、DEA均为正(负),DIF向上突破(向下跌破) DEA,考虑买入(卖出);
# 2.当MACD以大角度变化,代表了一个市场大趋势的转 变;
# 3.分析MACD柱状线,由红变绿(正变负),应卖出;由绿变红,考虑买入;
# 4.MACD、DMA、TRIX三者构成一组指标群,须综合搭配使用。
#####策略:
#买入:
#当 MACD线在前一天的值 < 信号线前一天的值
#且 当天MACD线的值 > 当天信号线的值 时
#说明发生了金叉,此时看涨,第二天买入。
#卖出:若已盈利10%,则卖出;若已亏损10%,则卖出。
import backtrader as bt
from backtrader.indicators import EMA
class MACD_Strategy(bt.Strategy):
params = ('maperiod', 15)
def log(self, txt, dt=None):
''' Logging function fot this strategy'''
dt = dt or self.datas[0].datetime.date(0)
print('%s, %s' % (dt.isoformat(), txt))
def percent(today, yesterday):
return float(today - yesterday) / today
def __init__(self):
self.dataclose = self.datas[0].close
self.volume = self.datas[0].volume
self.order = None
self.buyprice = None
self.buycomm = None
me1 = EMA(self.data, period=12)
me2 = EMA(self.data, period=26)
self.macd = me1 - me2
self.signal = EMA(self.macd, period=9)
bt.indicators.MACDHisto(self.data)
def notify_order(self, order):
if order.status in [order.Submitted, order.Accepted]:
return
if order.status in [order.Completed]:
if order.isbuy():
self.log(
'BUY EXECUTED, Price: %.2f, Cost: %.2f, Comm %.2f' %
(order.executed.price,
order.executed.value,
order.executed.comm))
self.buyprice = order.executed.price
self.buycomm = order.executed.comm
self.bar_executed_close = self.dataclose[0]
else:
self.log('SELL EXECUTED, Price: %.2f, Cost: %.2f, Comm %.2f' %
(order.executed.price,
order.executed.value,
order.executed.comm))
self.bar_executed = len(self)
elif order.status in [order.Canceled, order.Margin, order.Rejected]:
self.log('Order Canceled/Margin/Rejected')
self.order = None
def notify_trade(self, trade):
if not trade.isclosed:
return
self.log('OPERATION PROFIT, GROSS %.2f, NET %.2f' %
(trade.pnl, trade.pnlcomm))
# Python 实用宝典
def next(self):
self.log('Close, %.2f' % self.dataclose[0])
if self.order:
return
if not self.position:
condition1 = self.macd[-1] - self.signal[-1]
condition2 = self.macd[0] - self.signal[0]
if condition1 < 0 and condition2 > 0:
self.log('BUY CREATE, %.2f' % self.dataclose[0])
self.order = self.buy()
else:
condition = (self.dataclose[0] - self.bar_executed_close) / self.dataclose[0]
if condition > 0.1 or condition < -0.1:
self.log('SELL CREATE, %.2f' % self.dataclose[0])
self.order = self.sell()
# =============================================================================
if __name__ == '__main__':
# 初始化模型
cerebro = bt.Cerebro()
# 设定初始资金
cerebro.broker.setcash(1000000.0)
# 策略执行前的资金
print('Starting Portfolio Value: %.2f' % cerebro.broker.getvalue())
#每次交易需要支付的佣金
cerebro.broker.setcommission(0.005)
#设定需要设定每次交易买入的股数
cerebro.addsizer(bt.sizers.FixedSize, stake=100)
#加载数据
data = bt.feeds.YahooFinanceCSVData(#使用雅虎
dataname='6098.HK.csv',
fromdate=datetime.datetime(2022, 3, 8),
todate=datetime.datetime(2023, 3, 6)
)
# 安装策略到backtrade
strats = cerebro.addstrategy(MACD_Strategy)
#加载数据
cerebro.adddata(data)
# 策略执行
cerebro.run()
#使用brac=====ktrader_plotting
b = Bokeh(style='bar', plot_mode='single', scheme=Tradimo())
cerebro.plot(b,style='bar',iplot=False)#终于成功!不