Harwinder
3e078cfe7d
--regime-filter previously affected only backtests. Live mode now also checks higher-TF ADX after analysis: signals are downgraded to NEUTRAL with a warning when ADX < threshold, and confirming when above. This brings live behavior into parity with what the backtester measures.
463 lines
19 KiB
Python
463 lines
19 KiB
Python
"""
|
|
ForexBot2 — Signal Aggregator
|
|
==============================
|
|
Combines multi-timeframe technical analysis with optional news context
|
|
into a graded actionable signal. Same code path is used for live mode
|
|
and walk-forward backtesting; the only difference is whether the input
|
|
OHLCV is fetched live or sliced from a pre-loaded historical buffer.
|
|
"""
|
|
|
|
import logging
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timezone
|
|
from typing import Dict, List, Optional
|
|
|
|
import pandas as pd
|
|
|
|
from config import PlatformConfig, SignalStrength, Timeframe
|
|
from data_provider import ForexDataProvider
|
|
from engine import TechnicalEngine, TechnicalAnalysis
|
|
from news_engine import NewsEngine, FundamentalContext
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
@dataclass
|
|
class TradeSetup:
|
|
pair: str
|
|
display_name: str
|
|
direction: str # "BUY" / "SELL"
|
|
entry_price: float
|
|
stop_loss: float
|
|
take_profit_1: float
|
|
take_profit_2: float
|
|
risk_reward: float
|
|
risk_pips: float
|
|
reward_pips: float
|
|
pip_value: float
|
|
|
|
|
|
@dataclass
|
|
class SignalReport:
|
|
symbol: str
|
|
display_name: str
|
|
timestamp: str
|
|
|
|
primary_signal: SignalStrength = SignalStrength.NEUTRAL
|
|
primary_score: float = 0.0
|
|
primary_confidence: float = 0.0
|
|
|
|
mtf_agreement: bool = False
|
|
mtf_signals: Dict[str, SignalStrength] = field(default_factory=dict)
|
|
mtf_scores: Dict[str, float] = field(default_factory=dict)
|
|
|
|
primary_analysis: Optional[TechnicalAnalysis] = None
|
|
confirmation_analyses: List[TechnicalAnalysis] = field(default_factory=list)
|
|
|
|
event_risk: str = "clear"
|
|
event_description: str = ""
|
|
news_sentiment: float = 0.0
|
|
news_description: str = ""
|
|
|
|
final_signal: SignalStrength = SignalStrength.NEUTRAL
|
|
final_score: float = 0.0
|
|
final_confidence: float = 0.0
|
|
action_recommendation: str = ""
|
|
signal_quality: str = "" # A+, A, B, C, D, F
|
|
|
|
trade_setup: Optional[TradeSetup] = None
|
|
|
|
bullish_reasons: List[str] = field(default_factory=list)
|
|
bearish_reasons: List[str] = field(default_factory=list)
|
|
warnings: List[str] = field(default_factory=list)
|
|
|
|
current_price: float = 0.0
|
|
spread_estimate: float = 0.0
|
|
market_regime: str = ""
|
|
volatility_regime: str = ""
|
|
|
|
|
|
class SignalAggregator:
|
|
"""
|
|
Master aggregator. Designed so the backtester can drive it bar-by-bar
|
|
via `analyze_from_dataframes` without re-fetching from the API.
|
|
"""
|
|
|
|
def __init__(self, config: PlatformConfig):
|
|
self.config = config
|
|
self.data_provider = ForexDataProvider(config)
|
|
self.tech_engine = TechnicalEngine(config.technical)
|
|
self._news_engine: Optional[NewsEngine] = None # lazy
|
|
|
|
def _get_news_engine(self) -> NewsEngine:
|
|
if self._news_engine is None:
|
|
self._news_engine = NewsEngine(self.config.news)
|
|
return self._news_engine
|
|
|
|
# ── Public: live mode entry point ──────────────────────────────
|
|
|
|
def analyze_pair(
|
|
self,
|
|
symbol: Optional[str] = None,
|
|
context: Optional[FundamentalContext] = None,
|
|
) -> SignalReport:
|
|
"""Live-mode entry: fetch all needed timeframes then delegate."""
|
|
if symbol is None:
|
|
symbol = self.config.symbol
|
|
|
|
timeframes = [self.config.primary_timeframe] + list(
|
|
self.config.confirmation_timeframes
|
|
)
|
|
|
|
bt = self.config.backtest
|
|
if bt.regime_filter and bt.regime_filter_timeframe not in timeframes:
|
|
timeframes.append(bt.regime_filter_timeframe)
|
|
|
|
dfs: Dict[Timeframe, pd.DataFrame] = {}
|
|
for tf in timeframes:
|
|
try:
|
|
dfs[tf] = self.data_provider.fetch_ohlcv(symbol, tf)
|
|
except Exception as e:
|
|
logger.warning(f"Fetch failed for {symbol} {tf.value}: {e}")
|
|
|
|
report = self.analyze_from_dataframes(
|
|
symbol=symbol,
|
|
dfs=dfs,
|
|
context=context,
|
|
use_news=True,
|
|
)
|
|
|
|
if bt.regime_filter:
|
|
self._apply_regime_filter_live(report, dfs)
|
|
|
|
return report
|
|
|
|
def _apply_regime_filter_live(
|
|
self, report: SignalReport, dfs: Dict[Timeframe, pd.DataFrame]
|
|
) -> None:
|
|
"""If higher-TF ADX is below threshold, downgrade signal to NEUTRAL."""
|
|
bt = self.config.backtest
|
|
regime_tf = bt.regime_filter_timeframe
|
|
df = dfs.get(regime_tf)
|
|
if df is None or df.empty:
|
|
report.warnings.append(
|
|
f"Regime filter requested but {regime_tf.value} data unavailable"
|
|
)
|
|
return
|
|
|
|
adx_value = self._latest_adx(df, self.config.technical.adx_period)
|
|
if adx_value is None:
|
|
report.warnings.append(
|
|
f"Regime filter: ADX could not be computed on {regime_tf.value}"
|
|
)
|
|
return
|
|
|
|
if adx_value < bt.regime_filter_min_adx:
|
|
report.warnings.append(
|
|
f"REGIME FILTER: {regime_tf.value} ADX={adx_value:.1f} < "
|
|
f"{bt.regime_filter_min_adx:.0f} — market choppy, skipping signal"
|
|
)
|
|
report.final_signal = SignalStrength.NEUTRAL
|
|
report.signal_quality = "F"
|
|
report.action_recommendation = (
|
|
f"NO TRADE — Regime filter (ADX {adx_value:.1f} on {regime_tf.value})"
|
|
)
|
|
report.trade_setup = None
|
|
else:
|
|
report.bullish_reasons.append(
|
|
f"Regime: {regime_tf.value} ADX={adx_value:.1f} confirms trending market"
|
|
) if report.final_score > 0 else report.bearish_reasons.append(
|
|
f"Regime: {regime_tf.value} ADX={adx_value:.1f} confirms trending market"
|
|
)
|
|
|
|
@staticmethod
|
|
def _latest_adx(df: pd.DataFrame, period: int = 14) -> Optional[float]:
|
|
"""Wilder ADX on a fresh df; returns the last finite value or None."""
|
|
if df is None or len(df) < period * 3:
|
|
return None
|
|
high, low, close = df["High"], df["Low"], df["Close"]
|
|
tr = pd.concat([
|
|
high - low,
|
|
(high - close.shift()).abs(),
|
|
(low - close.shift()).abs(),
|
|
], axis=1).max(axis=1)
|
|
plus_dm = high.diff()
|
|
minus_dm = -low.diff()
|
|
plus_dm = plus_dm.where((plus_dm > minus_dm) & (plus_dm > 0), 0.0)
|
|
minus_dm = minus_dm.where((minus_dm > plus_dm) & (minus_dm > 0), 0.0)
|
|
atr_s = tr.ewm(alpha=1/period, min_periods=period).mean()
|
|
plus_di = 100 * (plus_dm.ewm(alpha=1/period, min_periods=period).mean() / atr_s)
|
|
minus_di = 100 * (minus_dm.ewm(alpha=1/period, min_periods=period).mean() / atr_s)
|
|
denom = (plus_di + minus_di).replace(0, float("nan"))
|
|
dx = 100 * (plus_di - minus_di).abs() / denom
|
|
adx = dx.ewm(alpha=1/period, min_periods=period).mean().dropna()
|
|
if adx.empty:
|
|
return None
|
|
return float(adx.iloc[-1])
|
|
|
|
# ── Public: bar-replayable entry point ─────────────────────────
|
|
|
|
def analyze_from_dataframes(
|
|
self,
|
|
symbol: str,
|
|
dfs: Dict[Timeframe, pd.DataFrame],
|
|
context: Optional[FundamentalContext] = None,
|
|
use_news: bool = True,
|
|
) -> SignalReport:
|
|
"""
|
|
Run the full analysis pipeline on already-fetched dataframes.
|
|
|
|
`dfs` must include the primary timeframe and ideally the
|
|
confirmation timeframes. Each df is treated as the visible history
|
|
at decision time — the backtester is responsible for slicing so
|
|
no future bars leak in.
|
|
"""
|
|
pair_info = self.data_provider.get_pair_info(symbol)
|
|
display = pair_info["display"]
|
|
pip_value = pair_info["pip_value"]
|
|
|
|
report = SignalReport(
|
|
symbol=symbol,
|
|
display_name=display,
|
|
timestamp=str(datetime.now(timezone.utc)),
|
|
)
|
|
|
|
# ── 1. Primary timeframe ───────────────────────────────────
|
|
primary_df = dfs.get(self.config.primary_timeframe)
|
|
if primary_df is None or primary_df.empty:
|
|
report.warnings.append("Primary timeframe data unavailable")
|
|
report.signal_quality = "F"
|
|
return report
|
|
|
|
primary_analysis = self.tech_engine.analyze(
|
|
primary_df, symbol, self.config.primary_timeframe.value
|
|
)
|
|
report.primary_analysis = primary_analysis
|
|
report.primary_signal = primary_analysis.composite_signal
|
|
report.primary_score = primary_analysis.composite_score
|
|
report.primary_confidence = primary_analysis.confidence
|
|
report.market_regime = primary_analysis.market_regime
|
|
report.volatility_regime = primary_analysis.volatility_regime
|
|
report.current_price = float(primary_df["Close"].iloc[-1])
|
|
report.timestamp = str(primary_df.index[-1])
|
|
report.mtf_signals[self.config.primary_timeframe.value] = primary_analysis.composite_signal
|
|
report.mtf_scores[self.config.primary_timeframe.value] = primary_analysis.composite_score
|
|
|
|
# ── 2. Confirmation timeframes ─────────────────────────────
|
|
for tf in self.config.confirmation_timeframes:
|
|
cf_df = dfs.get(tf)
|
|
if cf_df is None or cf_df.empty:
|
|
continue
|
|
try:
|
|
cf_analysis = self.tech_engine.analyze(cf_df, symbol, tf.value)
|
|
report.confirmation_analyses.append(cf_analysis)
|
|
report.mtf_signals[tf.value] = cf_analysis.composite_signal
|
|
report.mtf_scores[tf.value] = cf_analysis.composite_score
|
|
except Exception as e:
|
|
logger.warning(f"Confirmation TF {tf.value} failed for {symbol}: {e}")
|
|
|
|
# ── 3. MTF agreement ───────────────────────────────────────
|
|
report.mtf_agreement = self._check_mtf_agreement(report)
|
|
|
|
# ── 4. Fundamental context (optional) ──────────────────────
|
|
if use_news:
|
|
news = self._get_news_engine()
|
|
if context is None:
|
|
context = news.get_fundamental_context()
|
|
event_risk, event_desc = news.get_event_risk_level(symbol, context)
|
|
report.event_risk = event_risk
|
|
report.event_description = event_desc
|
|
sentiment, sent_desc = news.get_pair_sentiment(symbol, context)
|
|
report.news_sentiment = sentiment
|
|
report.news_description = sent_desc
|
|
else:
|
|
report.event_risk = "clear"
|
|
report.event_description = ""
|
|
report.news_sentiment = 0.0
|
|
report.news_description = ""
|
|
|
|
# ── 5. Spread estimate (only meaningful in live mode) ──────
|
|
if use_news:
|
|
try:
|
|
report.spread_estimate = self.data_provider.get_spread_estimate(symbol)
|
|
except Exception:
|
|
report.spread_estimate = 0.0
|
|
else:
|
|
# Backtest sets spread externally via BacktestConfig
|
|
report.spread_estimate = 0.0
|
|
|
|
# ── 6. Final signal ────────────────────────────────────────
|
|
self._compute_final_signal(report)
|
|
|
|
# ── 7. Trade setup for actionable signals ──────────────────
|
|
if report.signal_quality in ("A+", "A", "B"):
|
|
report.trade_setup = self._generate_trade_setup(report, pip_value)
|
|
|
|
# ── 8. Reasons ─────────────────────────────────────────────
|
|
self._collect_reasons(report)
|
|
|
|
return report
|
|
|
|
# ── Internals ──────────────────────────────────────────────────
|
|
|
|
def _check_mtf_agreement(self, report: SignalReport) -> bool:
|
|
signals = list(report.mtf_signals.values())
|
|
if len(signals) < 2:
|
|
return False
|
|
|
|
bullish = sum(1 for s in signals if s in (SignalStrength.BUY, SignalStrength.STRONG_BUY))
|
|
bearish = sum(1 for s in signals if s in (SignalStrength.SELL, SignalStrength.STRONG_SELL))
|
|
|
|
threshold = self.config.signals.mtf_agreement_threshold
|
|
return bullish >= threshold or bearish >= threshold
|
|
|
|
def _compute_final_signal(self, report: SignalReport):
|
|
score = report.primary_score
|
|
confidence = report.primary_confidence
|
|
|
|
# MTF
|
|
if report.mtf_agreement:
|
|
confidence = min(confidence + 15, 95)
|
|
all_scores = list(report.mtf_scores.values())
|
|
if all_scores:
|
|
mtf_avg = sum(all_scores) / len(all_scores)
|
|
score = score * 0.6 + mtf_avg * 0.4
|
|
else:
|
|
confidence *= 0.7
|
|
if len(report.mtf_scores) >= 2:
|
|
scores = list(report.mtf_scores.values())
|
|
has_bull = any(s > 10 for s in scores)
|
|
has_bear = any(s < -10 for s in scores)
|
|
if has_bull and has_bear:
|
|
report.warnings.append("Timeframes CONFLICT — mixed signals across timeframes")
|
|
confidence *= 0.6
|
|
|
|
# News sentiment
|
|
if abs(report.news_sentiment) > 0.2:
|
|
if (report.news_sentiment > 0 and score > 0) or \
|
|
(report.news_sentiment < 0 and score < 0):
|
|
confidence = min(confidence + 5, 95)
|
|
elif (report.news_sentiment > 0 and score < -10) or \
|
|
(report.news_sentiment < 0 and score > 10):
|
|
report.warnings.append("News sentiment CONFLICTS with technical signal")
|
|
confidence *= 0.85
|
|
|
|
# Event risk
|
|
if report.event_risk == "danger":
|
|
report.warnings.append(f"EVENT RISK: {report.event_description}")
|
|
confidence *= 0.3
|
|
report.action_recommendation = f"WAIT — {report.event_description}"
|
|
elif report.event_risk == "caution":
|
|
report.warnings.append(f"Upcoming event: {report.event_description}")
|
|
confidence *= 0.7
|
|
|
|
# Volatility regime
|
|
if report.volatility_regime == "extreme":
|
|
report.warnings.append("EXTREME volatility — wider stops needed, reduced position size")
|
|
confidence *= 0.8
|
|
elif report.volatility_regime == "low":
|
|
report.warnings.append("Low volatility — potential for breakout, tight ranges")
|
|
|
|
# Final signal
|
|
if score >= 40 and confidence >= 65:
|
|
report.final_signal = SignalStrength.STRONG_BUY
|
|
elif score >= 15 and confidence >= 50:
|
|
report.final_signal = SignalStrength.BUY
|
|
elif score <= -40 and confidence >= 65:
|
|
report.final_signal = SignalStrength.STRONG_SELL
|
|
elif score <= -15 and confidence >= 50:
|
|
report.final_signal = SignalStrength.SELL
|
|
else:
|
|
report.final_signal = SignalStrength.NEUTRAL
|
|
|
|
report.final_score = score
|
|
report.final_confidence = confidence
|
|
|
|
# Quality grade
|
|
abs_score = abs(score)
|
|
if abs_score >= 50 and confidence >= 75 and report.mtf_agreement:
|
|
report.signal_quality = "A+"
|
|
elif abs_score >= 35 and confidence >= 65 and report.mtf_agreement:
|
|
report.signal_quality = "A"
|
|
elif abs_score >= 25 and confidence >= 55:
|
|
report.signal_quality = "B"
|
|
elif abs_score >= 15 and confidence >= 45:
|
|
report.signal_quality = "C"
|
|
elif abs_score >= 10:
|
|
report.signal_quality = "D"
|
|
else:
|
|
report.signal_quality = "F"
|
|
|
|
if not report.action_recommendation:
|
|
sig_name = report.final_signal.name.replace("_", " ")
|
|
if report.signal_quality in ("A+", "A"):
|
|
report.action_recommendation = f"{sig_name} — High-quality setup, consider entry"
|
|
elif report.signal_quality == "B":
|
|
report.action_recommendation = f"{sig_name} — Decent setup, wait for confirmation"
|
|
elif report.signal_quality in ("C", "D"):
|
|
report.action_recommendation = "WATCH — Signal present but low confidence"
|
|
else:
|
|
report.action_recommendation = "NO TRADE — No clear signal"
|
|
|
|
def _generate_trade_setup(self, report: SignalReport, pip_value: float) -> Optional[TradeSetup]:
|
|
if not report.primary_analysis:
|
|
return None
|
|
|
|
price = report.current_price
|
|
atr = report.primary_analysis.key_levels.get("atr", 0)
|
|
if atr == 0:
|
|
return None
|
|
|
|
is_buy = report.final_signal in (SignalStrength.BUY, SignalStrength.STRONG_BUY)
|
|
rr = self.config.signals.default_rr_ratio
|
|
sl_mult = self.config.signals.atr_sl_multiplier
|
|
|
|
sl_distance = atr * sl_mult
|
|
tp1_distance = sl_distance * rr
|
|
tp2_distance = sl_distance * rr * 1.5
|
|
|
|
if is_buy:
|
|
entry, sl = price, price - sl_distance
|
|
tp1, tp2 = price + tp1_distance, price + tp2_distance
|
|
direction = "BUY"
|
|
else:
|
|
entry, sl = price, price + sl_distance
|
|
tp1, tp2 = price - tp1_distance, price - tp2_distance
|
|
direction = "SELL"
|
|
|
|
risk_pips = sl_distance / pip_value
|
|
reward_pips = tp1_distance / pip_value
|
|
|
|
return TradeSetup(
|
|
pair=report.symbol,
|
|
display_name=report.display_name,
|
|
direction=direction,
|
|
entry_price=entry,
|
|
stop_loss=sl,
|
|
take_profit_1=tp1,
|
|
take_profit_2=tp2,
|
|
risk_reward=rr,
|
|
risk_pips=risk_pips,
|
|
reward_pips=reward_pips,
|
|
pip_value=pip_value,
|
|
)
|
|
|
|
def _collect_reasons(self, report: SignalReport):
|
|
if not report.primary_analysis:
|
|
return
|
|
|
|
for name, ind in report.primary_analysis.indicators.items():
|
|
if name.startswith("_") or ind.weight == 0:
|
|
continue
|
|
if ind.signal in (SignalStrength.BUY, SignalStrength.STRONG_BUY):
|
|
prefix = "STRONG: " if ind.signal == SignalStrength.STRONG_BUY else ""
|
|
report.bullish_reasons.append(f"{prefix}{ind.name}: {ind.description}")
|
|
elif ind.signal in (SignalStrength.SELL, SignalStrength.STRONG_SELL):
|
|
prefix = "STRONG: " if ind.signal == SignalStrength.STRONG_SELL else ""
|
|
report.bearish_reasons.append(f"{prefix}{ind.name}: {ind.description}")
|
|
|
|
if report.news_sentiment > 0.1:
|
|
report.bullish_reasons.append(f"News: {report.news_description}")
|
|
elif report.news_sentiment < -0.1:
|
|
report.bearish_reasons.append(f"News: {report.news_description}")
|