Overview
The script fetches OHLCV data for crypto pairs from Kraken, computes the 9-day EMA, and compares it to the latest closing price. It then calculates the percentage difference between the price and EMA, determines the market trend (bullish, bearish, or neutral) using both short and long EMAs, and assigns a bounce likelihood based on how closely the price is approaching the 9EMA.
Key points:
- Data Fetching: Uses
ccxtto retrieve market data. - EMA Calculation: Uses pandas’ exponential weighted moving average to compute the 9EMA.
- Trend Analysis: Compares short (9-day) and long (21-day) EMAs to assess trend direction.
- Bounce Prediction: Evaluates if the current price is near the 9EMA and predicts the likelihood of a bounce.
import ccxt
import pandas as pd
import json
from concurrent.futures import ThreadPoolExecutor, as_completed
def calculate_ema(series, period=9):
return series.ewm(span=period, adjust=False).mean()
def get_ohlcv_df(exchange, symbol, timeframe='1d', limit=100):
data = exchange.fetch_ohlcv(symbol, timeframe=timeframe, limit=limit)
if not data:
return pd.DataFrame()
df = pd.DataFrame(data, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('timestamp', inplace=True)
return df
def get_trend_strength_and_direction(df, short_ema=9, long_ema=21):
df['short_ema'] = df['close'].ewm(span=short_ema, adjust=False).mean()
df['long_ema'] = df['close'].ewm(span=long_ema, adjust=False).mean()
latest_short = df.iloc[-1]['short_ema']
latest_long = df.iloc[-1]['long_ema']
if latest_short > latest_long:
return 'bullish'
elif abs(latest_short - latest_long) < 0.01 * latest_long:
return 'neutral'
else:
return 'bearish'
def assign_bounce_likelihood(diff_percent, trend_direction, approach_direction):
if trend_direction == 'bullish':
return "High" if approach_direction == 'above' else "Medium"
elif trend_direction == 'bearish':
return "High" if approach_direction == 'below' else "Medium"
return "Medium" if approach_direction == 'above' else "Low"
def fetch_symbol_data(exchange, symbol, timeframe, limit, threshold_percent):
try:
df = get_ohlcv_df(exchange, symbol, timeframe=timeframe, limit=limit)
if df.empty:
return None
df['9_EMA'] = calculate_ema(df['close'], period=9)
trend = get_trend_strength_and_direction(df, short_ema=9, long_ema=21)
latest = df.iloc[-1]
diff = latest['close'] - latest['9_EMA']
diff_percent = (diff / latest['9_EMA']) * 100 if latest['9_EMA'] != 0 else None
approach_direction = "above" if diff > 0 else "below"
touched_9ema = (latest['low'] <= latest['9_EMA'] <= latest['high'])
if diff_percent is not None and (abs(diff_percent) <= threshold_percent) and (not touched_9ema):
bounce_prediction = assign_bounce_likelihood(diff_percent, trend, approach_direction)
return {
'symbol': symbol,
'last_close': round(latest['close'], 4),
'ema_9': round(latest['9_EMA'], 4),
'diff': round(diff, 4),
'diff_percent': round(diff_percent, 2),
'trend': trend,
'approach': approach_direction,
'bounce_likelihood': bounce_prediction
}
except Exception as e:
print(f"Error fetching {symbol}: {e}")
return None
def scan_cryptos_close_to_ema_with_prediction(exchange, symbols, ema_period=9, threshold_percent=1.0, timeframe='1d', limit=100):
results = []
with ThreadPoolExecutor(max_workers=10) as executor:
future_to_symbol = {executor.submit(fetch_symbol_data, exchange, symbol, timeframe, limit, threshold_percent): symbol for symbol in symbols}
for future in as_completed(future_to_symbol):
result = future.result()
if result:
results.append(result)
if results:
out_df = pd.DataFrame(results)
out_df.sort_values(by='diff_percent', key=abs, inplace=True)
out_df.reset_index(drop=True, inplace=True)
return out_df
return pd.DataFrame(columns=['symbol', 'last_close', 'ema_9', 'diff', 'diff_percent', 'trend', 'approach', 'bounce_likelihood'])
if __name__ == "__main__":
kraken = ccxt.kraken()
kraken.load_markets()
tickers = kraken.fetch_tickers()
usd_pairs = [(sym, float(info.get('quoteVolume', info.get('volume', 0.0)))) for sym, info in tickers.items() if sym.endswith("/USD")]
usd_pairs.sort(key=lambda x: x[1], reverse=True)
top_50 = [x[0] for x in usd_pairs[:50]]
results_df = scan_cryptos_close_to_ema_with_prediction(kraken, top_50, ema_period=9, threshold_percent=1.0, timeframe='1d', limit=100)
output = {'status': 'success', 'data': results_df.to_dict(orient='records')} if not results_df.empty else {'status': 'success', 'data': [], 'message': 'No cryptos found within threshold.'}
print(json.dumps(output))