移动平均收敛散度（MACD - Moving Average Convergence Divergence）是一种趋势跟踪动量指标，显示了证券价格的两个移动平均之间的关系。它用于识别趋势的方向和强度，属于技术分析中振荡器的一类。
MACD如何衡量股票及其趋势
有两条线和一个柱：
第一条叫DIF线(差离值)又叫快线，有时也称为MACD线：12周期指数移动平均线(EMA)，减去26周期指数移动平均线(EMA)，计算得出两个线的差距。
在持续的涨势中，12日EMA在26日EMA之上。其间的正差离值(+DIF)会愈来愈大。反之在下跌趋势中，差离值可能变负(-DIF)，此时是绝对值愈来愈大。至于行情开始回转，正或负差离值要缩小到一定的程度，才真正是行情反转的信号。
第二条叫DEA线(信号线)又叫慢线：根据DIF值计算其指数9日移动平均值(EMA)，即离差(DIF)的平均值。
除此之外，还有一个柱状图，即快线–慢线所得的值呈现出来的直方图。上图的绿色柱代表DIF在DEA上方运行，红色柱代表DIF在DEA下方运行。美股和A股可能颜色是相反的，但是不影响我们理解。
交易者可能会在DIF线穿越其DEA线以上时买入证券，并在DIF线穿越DEA线以下时卖出或做空证券。
为什么它可能有效
MACD之所以有效，是因为它将动量和趋势结合在一个指标中。MACD线和信号线之间的差异指示了价格运动的强度，使交易者能够识别趋势变化周围的潜在买入或卖出机会。其有效性基于动量往往会在价格之前改变方向的原理。因此，MACD可以提供趋势反转的早期信号。
举个例子，一个物体在加速运行的时候，它的动能是很大的，除非有很大的外力阻止它，否则它会受惯性影响一直运动下去，直到动能为0。
MACD背后的逻辑
MACD的逻辑在于其能够通过比较短期和长期价格趋势来监测动量变化。当短期价格趋势开始超过长期趋势时，它可以是价格运动动量增加的信号，可能标志着购买或出售的机会。相反，当短期趋势相对于长期趋势减弱时，它可以表明动量减少，可能标志着趋势的逆转或放缓。
使用MACD的有效交易模型
基于MACD指标，有几种有效的交易模型和策略：
- MACD交叉：当DIF线穿越DEA线以上时买入，当它穿越以下时卖出。
- MACD背离：寻找MACD趋势和价格趋势之间的差异作为可能的逆转信号。
- MACD超买/超卖条件：当MACD从零线延伸得太远时，标识潜在的买入或卖出点，指示超买或超卖条件。
以特斯拉股票为例，时间从2020.1.1 ~ 2024.1.1四年数据，初始资金为1万美元，以日线级别作为回测已经，假设条件触发就全仓买入或者全仓卖出，利用MACD金叉死叉策略回测结果。
a) 编写MACD金叉死叉策略：
当macd线穿越Singal线向上运行时，买入；相反，当macd线穿越Singal线往下运行时，卖出；
import os
import yfinance as yf
import backtrader as bt
from datetime import datetime
import matplotlib.pyplot as plt
import pandas as pd
%matplotlib inline
#Only MACD
class MACDStrategy(bt.Strategy):
params = (
('macd1', 12),
('macd2', 26),
('macdsignal', 9),
)
def log(self, txt, dt=None):
''' Logging function for this strategy'''
dt = dt or self.datas[0].datetime.date(0)
print(f'{dt.isoformat()}, {txt}')
def __init__(self):
self.macd = bt.indicators.MACD(self.data.close,
period_me1=self.params.macd1,
period_me2=self.params.macd2,
period_signal=self.params.macdsignal)
self.crossover = bt.indicators.CrossOver(self.macd.macd, self.macd.signal)
self.order = None
self.buy_signals = []
self.sell_signals = []
#self.macd_hist = self.macd.macd - self.macd.signal
def notify_order(self, order):
#print('notify_order')
if order.status == order.Submitted:
# Order has been submitted/accepted to/by broker
self.log('Order Submitted')
if order.status == order.Accepted:
# Order has been submitted/accepted to/by broker
self.log('Order Accepted')
if order.status in [order.Completed]:
if order.isbuy():
self.log(f'BUY EXECUTED, Price: {order.executed.price}, Cost: {order.executed.value}, Commission: {order.executed.comm}')
self.log(f'Cash after buying: {self.broker.getcash()}')
self.buy_signals.append(self.datas[0].datetime.date(0))
elif order.issell():
self.log(f'SELL EXECUTED, Price: {order.executed.price}, Cost: {order.executed.value}, Commission: {order.executed.comm}')
self.log(f'Cash after selling: {self.broker.getcash()}')
self.sell_signals.append(self.datas[0].datetime.date(0))
self.order = None
def next(self):
if self.order: # Check if an order is pending, if so, we cannot send a 2nd one
return
cash_available = self.broker.getcash() # Get the current cash level
current_price = self.data.close[0] # Current price of the stock
size = int(cash_available / current_price) # Number of shares you can buy
#print(f'Current Close: {self.data.close[0]}')
if not self.position: # Not in the market
if self.crossover > 0: # MACD crosses above signal line
#self.log('BUY CREATE, %.2f' % self.data.close[0])
#self.order = self.buy() # MACD crosses above signal line
if cash_available > current_price: # Check if you have enough cash to buy at least one share
self.log(f'BUY CREATE, %.2f' % current_price)
self.order = self.buy(size=size) # Adjust the size according to your strategy needs
else:
self.log(f'Insufficient cash to buy. Available cash: {cash_available}, Current price: {current_price}')
elif self.crossover < 0: # MACD crosses below signal line
self.log('SELL CREATE, %.2f' % self.data.close[0])
self.order = self.close()
#self.sell_signals.append(self.datas[0].datetime.date(0))
b) 开始回测：
#Back test only for one stock
cerebro = bt.Cerebro()
cerebro.addstrategy(MACDStrategy)
datapath = 'data.csv'
data_df = yf.download('TSLA', start="2020-01-01", end="2024-01-01")
#print(data_df)
data = bt.feeds.PandasData(dataname=data_df)
#data = bt.feeds.YahooFinanceData(dataname=, fromdate=datetime(2019, 1, 1), todate=datetime(2020, 12, 31))
cerebro.adddata(data)
cerebro.addsizer(bt.sizers.FixedSize, stake=100)
cerebro.broker.set_cash(10000)
cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe_ratio')
cerebro.addanalyzer(bt.analyzers.DrawDown, _name='draw_down')
cerebro.addanalyzer(bt.analyzers.Returns, _name='returns') # For Total and Annualized Returns
cerebro.addanalyzer(bt.analyzers.SQN, _name='sqn') # System Quality Number can hint at the win rate
cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name='trade_analyzer') # For Win Rate, Profit-Loss Ratio
print('Starting Portfolio Value: %.2f' % cerebro.broker.getvalue())
results = cerebro.run()
strat = results[0]
buy_signals = strat.buy_signals
sell_signals = strat.sell_signals
# Sharpe Ratio
sharpe_ratio = strat.analyzers.sharpe_ratio.get_analysis()['sharperatio']
print(f"Sharpe Ratio: {sharpe_ratio}")
# Maximum Drawdown
max_drawdown = strat.analyzers.draw_down.get_analysis()['max']['drawdown']
print(f"Maximum Drawdown: {max_drawdown}%")
# Total Return
total_return = strat.analyzers.returns.get_analysis()['rtot']
print(f"Total Return: {total_return * 100}%")
# Annualized Return
annual_return = strat.analyzers.returns.get_analysis()['rnorm100']
print(f"Annualized Return: {annual_return}%")
# Sortino Ratio
#sortino_ratio = strat.analyzers.sortino_ratio.get_analysis()['sortino']
#print(f"Sortino Ratio: {sortino_ratio}")
# Win Rate and Profit-Loss Ratio
trade_analysis = strat.analyzers.trade_analyzer.get_analysis()
total_closed = trade_analysis.get('total', {}).get('closed', 0)
if total_closed > 0:
win_rate = (trade_analysis.get('won', {}).get('total', 0) / total_closed) * 100
won_avg = trade_analysis.get('won', {}).get('pnl', {}).get('average', 0)
lost_avg = trade_analysis.get('lost', {}).get('pnl', {}).get('average', 0)
profit_loss_ratio = abs(won_avg / lost_avg) if lost_avg != 0 else "undefined"
print(f"Win Rate: {win_rate}%")
print(f"Profit-Loss Ratio: {profit_loss_ratio}")
else:
print("No trades closed.")
print('Ending Portfolio Value: %.2f' % cerebro.broker.getvalue())
#cerebro.plot()
回测结果为：
- 初始投资组合价值：10000.00元
- 夏普比率：0.08717907196315927
- 最大回撤：48.50573641490451%
- 总回报率：-4.509942626137697%
- 年化回报率：-1.1233697243742937%
- 胜率：35.0%
- 盈亏比：1.8030867851982686
- 结束投资组合价值：9559.02元
注：每次触发条件一会我们会下单，但是并非每次都一定买到，有可能因为下单价格和实际价格有出入导致无法达成交易，但是不影响我们的回测结果。
我们发现，我们的回报率为负，初始资金未必能得到保障。我们展示下买入点：
# Assuming 'data' is a DataFrame with your price data and 'dates' as its index
plt.figure(figsize=(14, 7))
plt.plot(data_df['Close'], label='Close Price', alpha=0.5)
# Convert buy and sell signal dates from datetime.date to pandas timestamps for indexing
buy_signals_dt = pd.to_datetime(buy_signals)
sell_signals_dt = pd.to_datetime(sell_signals)
plt.scatter(buy_signals_dt, data_df.loc[buy_signals_dt, 'Close'], label='Buy Signal', marker='^', color='green')
plt.scatter(sell_signals_dt, data_df.loc[sell_signals_dt, 'Close'], label='Sell Signal', marker='v', color='red')
plt.title("MACD Strategy Backtest")
plt.legend()
plt.show()
下面我们引入RSI（相对强弱指数）是技术分析中广泛使用的一个强大指标，特别是在股票交易领域。它主要用于帮助交易者识别股票或其他资产价格中的超买或超卖条件，这些条件暗示着潜在的反转点。我们结合MACD金叉死叉策略进行再次尝试:
import os
import yfinance as yf
import backtrader as bt
from datetime import datetime
import matplotlib.pyplot as plt
import pandas as pd
%matplotlib inline
class MACD_RSI_Strategy(bt.Strategy):
params = (
('macd1', 12),
('macd2', 26),
('macdsig', 9),
('rsi_period', 14),
('rsi_upper', 70), # RSI overbought threshold
('rsi_lower', 30), # RSI oversold threshold
)
def log(self, txt, dt=None):
''' Logging function for this strategy '''
dt = dt or self.datas[0].datetime.date(0)
print(f'{dt.isoformat()}, {txt}')
def __init__(self):
# Initialize MACD and RSI indicators
self.macd = bt.indicators.MACD(self.data.close,
period_me1=self.p.macd1,
period_me2=self.p.macd2,
period_signal=self.p.macdsig)
self.rsi = bt.indicators.RSI(self.data.close, period=self.p.rsi_period)
self.crossover = bt.indicators.CrossOver(self.macd.macd, self.macd.signal)
# Attributes to track buy and sell signal dates
self.buy_signals = []
self.sell_signals = []
# To keep track of pending orders
self.order = None
def notify_order(self, order):
if order.status == order.Submitted:
self.log('Order Submitted')
if order.status == order.Accepted:
self.log('Order Accepted')
if order.status == order.Completed:
if order.isbuy():
self.log(f'BUY EXECUTED, Price: {order.executed.price}, Cost: {order.executed.value}, Commission: {order.executed.comm}')
self.log(f'Cash after buying: {self.broker.getcash()}')
self.buy_signals.append(self.datas[0].datetime.date(0))
elif order.issell():
self.log(f'SELL EXECUTED, Price: {order.executed.price}, Cost: {order.executed.value}, Commission: {order.executed.comm}')
self.log(f'Cash after selling: {self.broker.getcash()}')
self.sell_signals.append(self.datas[0].datetime.date(0))
elif order.status in [order.Canceled, order.Margin, order.Rejected]:
self.log('Order Canceled/Margin/Rejected')
def next(self):
if self.order: # Check if an order is pending, if so, we cannot send a 2nd one
return
cash_available = self.broker.getcash()
current_price = self.data.close[0]
size = int(cash_available / current_price) # Simplistic size calculation
if not self.position:
if self.crossover > 0 and self.rsi < self.p.rsi_upper: # Buy condition
if cash_available > current_price: # Ensure sufficient cash
self.log(f'BUY CREATE, {current_price}')
self.order = self.buy(size=size)
else:
self.log(f'Insufficient cash to buy. Available cash: {cash_available}, Current price: {current_price}')
elif self.crossover < 0 or self.rsi > self.p.rsi_upper: # Sell condition
self.log(f'SELL CREATE, {current_price}')
self.order = self.sell() # Use self.close() if wanting to close the entire position再次回测结果为
#Back test only for one stock
cerebro = bt.Cerebro()
cerebro.addstrategy(MACD_RSI_Strategy)
datapath = 'data.csv'
data_df = yf.download('TSLA', start="2020-01-01", end="2024-01-01")
#print(data_df)
data = bt.feeds.PandasData(dataname=data_df)
#data = bt.feeds.YahooFinanceData(dataname=, fromdate=datetime(2019, 1, 1), todate=datetime(2020, 12, 31))
cerebro.adddata(data)
cerebro.addsizer(bt.sizers.FixedSize, stake=100)
cerebro.broker.set_cash(10000)
cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe_ratio')
cerebro.addanalyzer(bt.analyzers.DrawDown, _name='draw_down')
cerebro.addanalyzer(bt.analyzers.Returns, _name='returns') # For Total and Annualized Returns
cerebro.addanalyzer(bt.analyzers.SQN, _name='sqn') # System Quality Number can hint at the win rate
cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name='trade_analyzer') # For Win Rate, Profit-Loss Ratio
print('Starting Portfolio Value: %.2f' % cerebro.broker.getvalue())
results = cerebro.run()
strat = results[0]
buy_signals = strat.buy_signals
sell_signals = strat.sell_signals
# Sharpe Ratio
sharpe_ratio = strat.analyzers.sharpe_ratio.get_analysis()['sharperatio']
print(f"Sharpe Ratio: {sharpe_ratio}")
# Maximum Drawdown
max_drawdown = strat.analyzers.draw_down.get_analysis()['max']['drawdown']
print(f"Maximum Drawdown: {max_drawdown}%")
# Total Return
total_return = strat.analyzers.returns.get_analysis()['rtot']
print(f"Total Return: {total_return * 100}%")
# Annualized Return
annual_return = strat.analyzers.returns.get_analysis()['rnorm100']
print(f"Annualized Return: {annual_return}%")
# Sortino Ratio
#sortino_ratio = strat.analyzers.sortino_ratio.get_analysis()['sortino']
#print(f"Sortino Ratio: {sortino_ratio}")
# Win Rate and Profit-Loss Ratio
trade_analysis = strat.analyzers.trade_analyzer.get_analysis()
total_closed = trade_analysis.get('total', {}).get('closed', 0)
if total_closed > 0:
win_rate = (trade_analysis.get('won', {}).get('total', 0) / total_closed) * 100
won_avg = trade_analysis.get('won', {}).get('pnl', {}).get('average', 0)
lost_avg = trade_analysis.get('lost', {}).get('pnl', {}).get('average', 0)
profit_loss_ratio = abs(won_avg / lost_avg) if lost_avg != 0 else "undefined"
print(f"Win Rate: {win_rate}%")
print(f"Profit-Loss Ratio: {profit_loss_ratio}")
else:
print("No trades closed.")
print('Ending Portfolio Value: %.2f' % cerebro.broker.getvalue())
#cerebro.plot()起始投资组合价值：10000.00
夏普比率：0.6716775968520321
最大回撤：73.56996064486884%
总回报：197.974046126522%
年化回报：64.20066841327395%
胜率：未平仓交易。
结束时的投资组合价值：72408.63
初始和结束投资组合价值
- MACD的策略起始和结束投资组合价值从10000元降低到9559.02元，表明在策略执行期间投资组合价值下降了。
- MACD-RSI策略的投资组合价值从10000元增加到72408.63元，显示出显著的正回报。
夏普比率
- MACD的夏普比率为0.087，表明每承担一单位总风险获得的额外回报较低。
- MACD-RSI的夏普比率为0.672，显著高于MACD策略，意味着相对于风险而言，回报率更优。
最大回撤
- MACD策略的最大回撤为48.51%，指出最大的价值下降比例。
- MACD-RSI策略有更高的最大回撤，为73.57%，表明其面临更高的价格波动和潜在的下跌风险。
总回报率和年化回报率
- MACD策略的总回报率为-4.51%，年化回报率为-1.12%，表明投资组合在期间内亏损。
- MACD-RSI策略的总回报率为197.97%，年化回报率为64.20%，显示出极高的盈利性。
胜率和盈亏比
- MACD策略的胜率为35%，盈亏比为1.80，表明虽然盈利次数少，但盈利交易相对于亏损交易仍然具有较好的表现。
- MACD-RSI策略的胜率未提供，但结束时投资组合的显著增值可能意味着高胜率或几笔大额盈利交易。
结论
比较这两种策略，MACD-RSI在总回报率和年化回报率方面表现显著优于单独使用MACD的策略。尽管MACD-RSI策略承受了更高的最大回撤，表明更大的价格波动和下行风险，但其显著的高夏普比率和总回报表明了其在风险调整后的回报方面的优越性。然而，投资者在选择策略时应考虑自己的风险容忍度和投资目标。
