Spaces:
Running
Running
File size: 8,284 Bytes
66dc1bf |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 |
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()
} |