Spaces:
Running
Running
File size: 4,708 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 |
from flask import Flask, request, jsonify
import yfinance as yf
import pandas as pd
import numpy as np
import talib
import datetime
# Calculate ADX, +DI, and -DI values
def calculate_adx(data, period=14):
high_prices = data['high']
low_prices = data['low']
close_prices = data['close']
adx = talib.ADX(high_prices, low_prices, close_prices, timeperiod=period)
plus_di = talib.PLUS_DI(high_prices, low_prices, close_prices, timeperiod=period)
minus_di = talib.MINUS_DI(high_prices, low_prices, close_prices, timeperiod=period)
return adx, plus_di, minus_di
# Detect ADX crossover (directional indicators +DI and -DI)
def adx_di_crossover_strategy(plus_di, minus_di, adx, threshold=20, lookback_days=5):
# We will loop over the last `lookback_days` to check for crossovers
for i in range(-lookback_days, 0):
# Check if ADX is above the threshold
if adx[i] > threshold:
# Bullish condition: +DI crosses above -DI and ADX is above threshold
if plus_di[i] > minus_di[i] and plus_di[i - 1] <= minus_di[i - 1]:
return "Bullish"
# Bearish condition: -DI crosses above +DI and ADX is above threshold
elif minus_di[i] > plus_di[i] and minus_di[i - 1] <= plus_di[i - 1]:
return "Bearish"
return "Neutral"
#ADX Breakout strategy
def adx_breakout_strategy(data, adx, threshold=25):
# Detect breakout condition (ADX above 25, price breaking resistance/support)
if adx[-1] > threshold:
if data['close'][-1] > data['high'][-2]: # Bullish breakout
return "Bullish"
elif data['close'][-1] < data['low'][-2]: # Bearish breakout
return "Bearish"
return "Neutral"
# ADX Slope Strategy
def get_adx_slope_signal(adx, days=5, threshold=0.1):
total_slope = 0
# Calculate slope for each of the last `days`
for i in range(-days, -1):
slope = adx[i + 1] - adx[i]
total_slope += slope
# Average slope
avg_slope = total_slope / (days - 1)
if avg_slope > threshold:
return "Bullish"
elif avg_slope < -threshold:
return "Bearish"
else:
return "Neutral"
# ADX Divergence Strategy
def adx_divergence_strategy(data, adx, threshold=25):
"""
Detects divergence between price and ADX.
A divergence occurs when price makes a new high/low, but ADX does not follow the same direction.
"""
price_high = data['high']
price_low = data['low']
# Checking for divergence
price_divergence_bullish = price_high[-1] > price_high[-2] and adx[-1] < adx[-2]
price_divergence_bearish = price_low[-1] < price_low[-2] and adx[-1] > adx[-2]
if price_divergence_bullish:
return "Bullish"
elif price_divergence_bearish:
return "Bearish"
return "Neutral"
# Main ADX strategy function
def adx_strategies(data):
# Calculate ADX, +DI, and -DI
adx, plus_di, minus_di = calculate_adx(data)
signals = {
"ADX": round(adx.iloc[-1], 2),
"ADX + DI Crossover": adx_di_crossover_strategy(plus_di, minus_di, adx),
"ADX Breakout": adx_breakout_strategy(data, adx),
"ADX Slope": get_adx_slope_signal(adx[-5:]),
"ADX Divergence": adx_divergence_strategy(data, adx)
}
weights = {
"ADX + DI Crossover": 35,
"ADX Breakout": 30,
"ADX Slope": 20,
"ADX Divergence": 15
}
total_score = 0
for strategy, weight in 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(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,adx, plus_di, minus_di
def extract_series(data, series, days=100):
series = pd.Series(series).dropna().tail(days)
series.index = data.index[-len(series):]
series.index = series.index.strftime('%Y-%m-%d')
return series.round(2).to_dict()
# API-style function
def get_adx_trade_signal(data):
adx_signals, overallscore, final_signal,adx, plus_di, minus_di = adx_strategies(data)
return {
"adx_signals": adx_signals,
"adx_score": overallscore,
"adx_final_signal": final_signal,
"ADX_Indicator": extract_series(data, adx),
"PLUS_DI": extract_series(data, plus_di),
"MINUS_DI": extract_series(data, minus_di)
}
|