macdstrategies.py

from flask import Flask, request, jsonify
import yfinance as yf
import pandas as pd
import numpy as np
import talib
from collections import OrderedDict
import datetime

    
# Calculate MACD, Signal and Histogram
def calculate_macdvalue(data, fast=12, slow=26, signal=9):
    
    close_prices = data['close']

    # Calculate MACD
    macd_line, signal_line, histogram = talib.MACD(
        close_prices,
        fastperiod=fast,
        slowperiod=slow,
        signalperiod=signal
    )

    return macd_line, signal_line, histogram

# MACD Line Crossover - completed
def get_macd_line_crossover_signal(macd, signal):    
    
    for i in range(len(macd) - 1):
        older_macd = macd[i]
        newer_macd = macd[i + 1]
        older_signal = signal[i]
        newer_signal = signal[i + 1]
        
        # Bullish crossover (MACD crosses above Signal)
        if older_macd <= older_signal and newer_macd > newer_signal:
            return "Bullish"
        # Bearish crossover (MACD crosses below Signal)
        elif older_macd >= older_signal and newer_macd < newer_signal:
            return "Bearish"
    
    return "Neutral"

# Zero Line Crossover - completed
def get_macd_zero_line_crossover_signal(macd):   
    
    for i in range(len(macd) - 1):
        older = macd[i]
        newer = macd[i + 1]
        
        if older <= 0 and newer > 0:
            return "Bullish"
        elif older >= 0 and newer < 0:
            return "Bearish"
    
    return "Neutral"

# MACD Momentum Signal - completed
def get_macd_momentum_signal(macd, signal, hist):

    for i in range(len(macd) - 1):
        older_macd = macd[i]
        newer_macd = macd[i + 1]
        older_signal = signal[i]
        newer_signal = signal[i + 1]
        current_hist = hist[i + 1]  # Use the histogram of the newer point

        # Bullish crossover (MACD crosses above Signal) with positive histogram
        if older_macd <= older_signal and newer_macd > newer_signal and current_hist > 0:
            return "Bullish"

        # Bearish crossover (MACD crosses below Signal) with negative histogram
        elif older_macd >= older_signal and newer_macd < newer_signal and current_hist < 0:
            return "Bearish"

    return "Neutral"

# MACD Volume Signal - completed
def get_macd_volume_signal(data, macd, signal):
   
    avg_volume = data['volume'].rolling(window=10).mean()
    recent_volume = data['volume'].values[-1:]
    recent_avg_volume = avg_volume.values[-1:]
    volume_confirm = recent_volume > recent_avg_volume
   
   
    for i in range(len(macd) - 1):
        older_macd = macd[i]
        newer_macd = macd[i + 1]
        older_signal = signal[i]
        newer_signal = signal[i + 1]
        
        if (older_macd <= older_signal and 
            newer_macd > newer_signal and 
            volume_confirm):
            return "Bullish"
        elif (older_macd >= older_signal and 
              newer_macd < newer_signal and 
              volume_confirm):
            return "Bearish"
    
    return "Neutral"

# MACD Multi-Timeframe - completed
def get_macd_multi_timeframe_confirmation(macd, signal,macd_hr, signal_hr):
    for i in range(len(macd) - 1):
        older_macd = macd[i]
        newer_macd = macd[i + 1]
        older_signal = signal[i]
        newer_signal = signal[i + 1]
        older_macd_hr = macd_hr[i]
        newer_macd_hr = macd_hr[i + 1]
        older_signal_hr = signal_hr[i]
        newer_signal_hr = signal_hr[i + 1]
        
        # Bullish crossover (MACD crosses above Signal)
        if older_macd <= older_signal and newer_macd > newer_signal and older_macd_hr <= older_signal_hr and newer_macd_hr > newer_signal_hr:
            return "Bullish"
        # Bearish crossover (MACD crosses below Signal)
        elif older_macd >= older_signal and newer_macd < newer_signal and older_macd_hr >= older_signal_hr and newer_macd_hr < newer_signal_hr:
            return "Bearish"
    
    return "Neutral"

# Price and MACD Divergence - completed
def get_macd_divergence_signal(macd, price):    
    
    # Bullish Divergence: Price makes lower lows, but MACD makes higher lows
    bullish_divergence = None
    for i in range(10, len(price)):  # Look at the last 5 candles
        if price[i] < price[i-1] and macd[i] > macd[i-1]:
            bullish_divergence = "Bullish"
            break

    # Bearish Divergence: Price makes higher highs, but MACD makes lower highs
    bearish_divergence = None
    for i in range(10, len(price)):  # Look at the last 5 candles
        if price[i] > price[i-1] and macd[i] < macd[i-1]:
            bearish_divergence = "Bearish"
            break
    
    if bullish_divergence:
        return bullish_divergence
    elif bearish_divergence:
        return bearish_divergence
    else:
        return "Neutral"

# MACD Hidden Divergence - completed
def get_macd_hidden_divergence_signal(macd, price):    
    
    # Bullish Hidden Divergence: Price makes a higher low, but MACD makes a lower low
    bullish_hidden_divergence = None
    for i in range(10, len(price)):  # Look at the last 5 candles
        if price[i] > price[i-10] and macd[i] < macd[i-10]:
            bullish_hidden_divergence = "Bullish"
            break

    # Bearish Hidden Divergence: Price makes a lower high, but MACD makes a higher high
    bearish_hidden_divergence = None
    for i in range(10, len(price)):  # Look at the last 5 candles
        if price[i] < price[i-10] and macd[i] > macd[i-10]:
            bearish_hidden_divergence = "Bearish"
            break
    
    if bullish_hidden_divergence:
        return bullish_hidden_divergence
    elif bearish_hidden_divergence:
        return bearish_hidden_divergence
    else:
        return "Neutral"


# macd_strategies and get_macd_trade_signal functions
def macd_strategies(data):     

    macd, signal, hist = calculate_macdvalue(data)    
    
    latest_macd = macd[-1]   
    signals = { 
        "MACD": round(latest_macd,2),
        "MACD Line Crossover": get_macd_line_crossover_signal(macd[-5:],signal[-5:]),
        "MACD Zero-Line Crossover": get_macd_zero_line_crossover_signal(macd[-5:]),
        "MACD Divergence": get_macd_divergence_signal(macd[-10:], data['close'][-10:]),
        "Hidden Divergence": get_macd_hidden_divergence_signal(macd[-10:], data['close'][-10:]),
        "MACD Volume": get_macd_volume_signal(data, macd[-5:],signal[-5:]),
        "MACD Momentum": get_macd_momentum_signal(macd[-5:],signal[-5:],hist[-5:]),
        
    }

    macd_signal_weights = {
        "MACD Line Crossover": 25,
        "MACD Zero-Line Crossover": 15, 
        "MACD Divergence": 20,
        "Hidden Divergence": 10,
        "MACD Volume": 15,
        "MACD Momentum": 15,
        
    }
    
    total_score = 0
    for strategy, weight in macd_signal_weights.items():
        signal = signals[strategy]
        if signal == "Bullish":
            total_score += weight
        elif signal == "Neutral":
            total_score += weight * 0.5

    overall_percentage = round((total_score / sum(macd_signal_weights.values())) * 100, 2)

    if overall_percentage >= 60:
        final_signal = "Buy"
    elif overall_percentage <= 40:
        final_signal = "DBuy"
    else:
        final_signal = "Neutral"
    
    return signals, overall_percentage, final_signal


def get_macd_trade_signal(data):    
    macd_signals, overallscore, final_signal = macd_strategies(data)
    macd_line, signal_line, hist = calculate_macdvalue(data)
    # Format and convert MACD and Signal Line for last 100 days
    macd_series = pd.Series(macd_line, index=data.index).dropna().tail(100)
    signal_series = pd.Series(signal_line, index=data.index).dropna().tail(100)

    macd_series.index = macd_series.index.strftime('%Y-%m-%d')
    signal_series.index = signal_series.index.strftime('%Y-%m-%d')
    hist_series = pd.Series(hist, index=data.index).dropna().tail(100)
    hist_series.index = hist_series.index.strftime('%Y-%m-%d')

    return {
        "macd_signals": macd_signals,
        "macd_score": overallscore,
        "macd_final_signal": final_signal,
        "macd_line": macd_series.round(2).to_dict(),
        "macd_signal_line": signal_series.round(2).to_dict(),
        "macd_histogram": hist_series.round(2).to_dict()
    }