mahmoud20138 Multi-Timeframe Confluence Scorer
Overview
Systematic top-down analysis across multiple timeframes. Scores directional agreement between timeframes and produces a confluence heat map. Higher confluence = higher probability.
1. Single Timeframe Bias Calculator
import pandas as pd
import numpy as np
def timeframe_bias(df: pd.DataFrame) -> dict:
"""
Calculate directional bias for a single timeframe.
Uses: MA alignment, RSI, MACD, price position relative to structure.
Returns score from -1 (strong bearish) to +1 (strong bullish).
"""
close = df["close"]
scores = []
# MA alignment
sma20 = close.rolling(20).mean().iloc[-1]
sma50 = close.rolling(50).mean().iloc[-1] if len(close) >= 50 else sma20
ema10 = close.ewm(span=10).mean().iloc[-1]
current = close.iloc[-1]
if current > sma20 > sma50: scores.append(1.0)
elif current < sma20 < sma50: scores.append(-1.0)
elif current > sma20: scores.append(0.3)
elif current < sma20: scores.append(-0.3)
else: scores.append(0)
# RSI
delta = close.diff()
gain = delta.where(delta > 0, 0).rolling(14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(14).mean()
rs = gain / loss.replace(0, np.nan)
rsi = (100 - (100 / (1 + rs))).iloc[-1]
if rsi > 60: scores.append(0.5)
elif rsi < 40: scores.append(-0.5)
else: scores.append(0)
# MACD
fast = close.ewm(span=12).mean()
slow = close.ewm(span=26).mean()
macd_line = (fast - slow).iloc[-1]
signal = (fast - slow).ewm(span=9).mean().iloc[-1]
if macd_line > signal and macd_line > 0: scores.append(0.7)
elif macd_line < signal and macd_line < 0: scores.append(-0.7)
elif macd_line > signal: scores.append(0.3)
elif macd_line < signal: scores.append(-0.3)
else: scores.append(0)
# Price position in recent range
high_20 = df["high"].tail(20).max()
low_20 = df["low"].tail(20).min()
position = (current - low_20) / (high_20 - low_20) if high_20 != low_20 else 0.5
scores.append(position - 0.5) # center at 0
bias_score = np.mean(scores)
return {
"score": round(bias_score, 3),
"direction": "BULLISH" if bias_score > 0.2 else "BEARISH" if bias_score < -0.2 else "NEUTRAL",
"strength": "strong" if abs(bias_score) > 0.5 else "moderate" if abs(bias_score) > 0.2 else "weak",
"rsi": round(rsi, 1),
"macd_signal": "bullish" if macd_line > signal else "bearish",
"price_vs_sma20": "above" if current > sma20 else "below",
}
2. Multi-Timeframe Confluence Engine
# TF weights: higher timeframes carry more weight
TF_WEIGHTS = {
"MN1": 1.5, "W1": 1.3, "D1": 1.2, "H4": 1.0,
"H1": 0.8, "M30": 0.6, "M15": 0.5, "M5": 0.4, "M1": 0.2,
}
def confluence_score(
data_by_tf: dict, # {"H1": df_h1, "H4": df_h4, "D1": df_d1, ...}
symbol: str = ""
) -> dict:
"""
Score multi-timeframe confluence.
Higher score = more timeframes agree on direction.
"""
tf_results = {}
weighted_sum = 0
total_weight = 0
for tf, df in data_by_tf.items():
if df.empty or len(df) < 50:
continue
bias = timeframe_bias(df)
weight = TF_WEIGHTS.get(tf, 0.5)
tf_results[tf] = {**bias, "weight": weight}
weighted_sum += bias["score"] * weight
total_weight += weight
overall_score = weighted_sum / max(total_weight, 1e-10)
# Count agreements
bullish_tfs = [tf for tf, r in tf_results.items() if r["direction"] == "BULLISH"]
bearish_tfs = [tf for tf, r in tf_results.items() if r["direction"] == "BEARISH"]
neutral_tfs = [tf for tf, r in tf_results.items() if r["direction"] == "NEUTRAL"]
agreement_pct = max(len(bullish_tfs), len(bearish_tfs)) / max(len(tf_results), 1) * 100
return {
"symbol": symbol,
"overall_score": round(overall_score, 3),
"direction": "BULLISH" if overall_score > 0.15 else "BEARISH" if overall_score < -0.15 else "NEUTRAL",
"confluence_pct": round(agreement_pct, 1),
"strength": "STRONG" if agreement_pct >= 75 else "MODERATE" if agreement_pct >= 50 else "WEAK",
"bullish_timeframes": bullish_tfs,
"bearish_timeframes": bearish_tfs,
"neutral_timeframes": neutral_tfs,
"timeframe_details": tf_results,
"trade_recommendation": _trade_recommendation(overall_score, agreement_pct),
}
def _trade_recommendation(score: float, agreement: float) -> str:
if agreement >= 75 and abs(score) > 0.3:
return f"STRONG {'BUY' if score > 0 else 'SELL'} — {agreement:.0f}% TF agreement, full position"
if agreement >= 60 and abs(score) > 0.2:
return f"MODERATE {'BUY' if score > 0 else 'SELL'} — {agreement:.0f}% agreement, reduced size"
if agreement < 50:
return "NO TRADE — timeframes disagree, wait for alignment"
return "WEAK signal — consider only with additional confirmation"
def multi_pair_confluence_scan(
pairs_data: dict, # {"EURUSD": {"H1": df, "H4": df, "D1": df}, "GBPUSD": ...}
) -> pd.DataFrame:
"""Scan multiple pairs for best confluence opportunities."""
results = []
for symbol, tf_data in pairs_data.items():
score = confluence_score(tf_data, symbol)
results.append({
"symbol": symbol,
"score": score["overall_score"],
"direction": score["direction"],
"confluence_pct": score["confluence_pct"],
"strength": score["strength"],
"recommendation": score["trade_recommendation"],
})
return pd.DataFrame(results).sort_values("confluence_pct", ascending=False)
3. Confluence Heat Map Data
def generate_heatmap_data(pairs_data: dict) -> dict:
"""Generate data for visual heat map rendering."""
matrix = {}
for symbol, tf_data in pairs_data.items():
matrix[symbol] = {}
for tf, df in tf_data.items():
if not df.empty and len(df) >= 20:
bias = timeframe_bias(df)
matrix[symbol][tf] = bias["score"]
return {
"matrix": matrix,
"color_scale": "green (+1) → white (0) → red (-1)",
"timeframes": list(TF_WEIGHTS.keys()),
"symbols": list(pairs_data.keys()),
}