【炒股Zero To Hero】MACD金叉死叉到底是否有效，加上这个指标回报率增加197倍

移动平均收敛散度（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