Harwinder Sekhon
80a2b89769
Resolves the Twelve Data API key through app_config.provider_api_key() (env → config.json → no default) instead of a hardcoded field on PlatformConfig. Adds data_provider.PROVIDER_REGISTRY + make_data_provider() so signal_aggregator and the backtester build the active provider through one seam; telegram_bot.py gets the same one-line WS-key fix now (per Task 4's already-planned edit for that line) so the suite doesn't crash on import between this commit and Task 4's full wiring pass. Claude-Session: https://claude.ai/code/session_01VPNF37MP9Kuv62B1VBrZ1b
1468 lines
65 KiB
Python
Executable File
1468 lines
65 KiB
Python
Executable File
"""
|
|
ForexBot2 — Signal Aggregator
|
|
==============================
|
|
Combines multi-timeframe technical analysis with optional news context
|
|
into a graded actionable signal.
|
|
"""
|
|
|
|
import logging
|
|
import os
|
|
from dataclasses import dataclass, field
|
|
from datetime import datetime, timezone
|
|
from typing import Dict, List, Optional
|
|
|
|
import pandas as pd
|
|
|
|
from chart_patterns import detect_chart_pattern
|
|
from config import PlatformConfig, SignalStrength, Timeframe
|
|
from data_provider import ForexDataProvider, make_data_provider
|
|
from engine import TechnicalEngine, TechnicalAnalysis
|
|
from news_engine import NewsEngine, FundamentalContext
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def _is_forex_weekend(now_utc: datetime) -> bool:
|
|
"""Forex market closed: Friday 22:00 UTC → Sunday 22:00 UTC (Sydney open)."""
|
|
wd = now_utc.weekday()
|
|
hour = now_utc.hour
|
|
if wd == 4 and hour >= 22: # Friday after 22:00
|
|
return True
|
|
if wd == 5: # All Saturday
|
|
return True
|
|
if wd == 6 and hour < 22: # Sunday before 22:00
|
|
return True
|
|
return False
|
|
|
|
|
|
@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
|
|
# Entry archetype ("GENERIC", "XAU_PB") — selects trail params.
|
|
strategy: str = "GENERIC"
|
|
|
|
|
|
@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 = ""
|
|
exhaustion_count: int = 0
|
|
|
|
|
|
class SignalAggregator:
|
|
"""Master aggregator for live signal analysis."""
|
|
|
|
# Diagnostic funnel for the M5 pullback route (shared across
|
|
# instances; read by backtest tooling, harmless in live use).
|
|
pb_funnel: Dict[str, int] = {}
|
|
# Same, for the chart-pattern route (XAU_CP).
|
|
cp_funnel: Dict[str, int] = {}
|
|
|
|
def __init__(self, config: PlatformConfig):
|
|
self.config = config
|
|
self.data_provider = make_data_provider(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
|
|
|
|
pair_info = self.data_provider.get_pair_info(symbol)
|
|
now_utc = datetime.now(timezone.utc)
|
|
|
|
if pair_info["display"] == "XAU/USD" and self.config.xau_strategy.enabled:
|
|
xau_cfg = self.config.xau_strategy
|
|
|
|
if xau_cfg.weekend_filter and _is_forex_weekend(now_utc):
|
|
report = SignalReport(
|
|
symbol=symbol,
|
|
display_name=pair_info["display"],
|
|
timestamp=str(now_utc),
|
|
)
|
|
report.final_signal = SignalStrength.NEUTRAL
|
|
report.signal_quality = "F"
|
|
report.trade_setup = None
|
|
report.warnings.append("WEEKEND FILTER: XAU/USD market closed — no liquidity")
|
|
report.action_recommendation = "NO TRADE — Weekend (market closed)"
|
|
return report
|
|
|
|
if xau_cfg.allowed_weekdays and now_utc.weekday() not in xau_cfg.allowed_weekdays:
|
|
day_names = {0: "Mon", 1: "Tue", 2: "Wed", 3: "Thu", 4: "Fri", 5: "Sat", 6: "Sun"}
|
|
allowed = "/".join(day_names.get(d, str(d)) for d in xau_cfg.allowed_weekdays)
|
|
report = SignalReport(
|
|
symbol=symbol,
|
|
display_name=pair_info["display"],
|
|
timestamp=str(now_utc),
|
|
)
|
|
report.final_signal = SignalStrength.NEUTRAL
|
|
report.signal_quality = "F"
|
|
report.trade_setup = None
|
|
report.warnings.append(
|
|
f"DAY FILTER: {day_names.get(now_utc.weekday(), '?')} not in allowed days ({allowed})"
|
|
)
|
|
report.action_recommendation = f"NO TRADE — Only trading {allowed}"
|
|
return report
|
|
|
|
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] = {}
|
|
stale_tfs = []
|
|
for tf in timeframes:
|
|
try:
|
|
df = self.data_provider.fetch_ohlcv(symbol, tf)
|
|
dfs[tf] = df
|
|
if not self.data_provider._is_data_fresh(df, symbol, tf):
|
|
stale_tfs.append(tf.value)
|
|
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 stale_tfs:
|
|
report.warnings.append(f"STALE DATA on {', '.join(stale_tfs)} — signal may be unreliable")
|
|
if self.config.primary_timeframe.value in stale_tfs:
|
|
report.final_signal = SignalStrength.NEUTRAL
|
|
report.signal_quality = "F"
|
|
report.trade_setup = None
|
|
report.warnings.append("Primary timeframe stale — signal suppressed")
|
|
|
|
if bt.regime_filter:
|
|
self._apply_regime_filter_live(report, dfs)
|
|
|
|
# Session window + blocked hours on the signal bar's CLOSE time —
|
|
# the exact predicate the backtester's replay loop applies before
|
|
# each bar, so live and backtest cannot diverge. Bar-close time
|
|
# (not wall clock) makes manual mid-candle scans behave like
|
|
# boundary auto-scans. This replaces an older wall-clock filter
|
|
# that was guarded by a marker string no code wrote.
|
|
if bt.session_filter and report.primary_analysis is not None:
|
|
try:
|
|
bar_close = (
|
|
pd.Timestamp(report.timestamp)
|
|
+ pd.Timedelta(
|
|
seconds=ForexDataProvider.CANDLE_SECONDS[
|
|
self.config.primary_timeframe
|
|
]
|
|
)
|
|
)
|
|
close_h = bar_close.hour
|
|
_is_xau = pair_info["display"] == "XAU/USD"
|
|
_blocked = (self.config.xau_strategy.blocked_hours_utc
|
|
if _is_xau else bt.blocked_hours_utc)
|
|
outside = not (bt.session_start_hour <= close_h < bt.session_end_hour)
|
|
if outside or close_h in _blocked:
|
|
report.warnings.append(
|
|
f"SESSION FILTER: bar closes {close_h:02}:00 UTC — "
|
|
+ ("outside window" if outside else "blocked hour")
|
|
)
|
|
report.final_signal = SignalStrength.NEUTRAL
|
|
report.signal_quality = "F"
|
|
report.action_recommendation = "NO TRADE — Outside tradeable session"
|
|
report.trade_setup = None
|
|
except Exception:
|
|
pass
|
|
|
|
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_raw = high.diff()
|
|
minus_dm_raw = -low.diff()
|
|
plus_dm = plus_dm_raw.where((plus_dm_raw > minus_dm_raw) & (plus_dm_raw > 0), 0.0)
|
|
minus_dm = minus_dm_raw.where((minus_dm_raw > plus_dm_raw) & (minus_dm_raw > 0), 0.0)
|
|
atr_s = tr.ewm(alpha=1/period, min_periods=period, adjust=False).mean()
|
|
plus_di = 100 * (plus_dm.ewm(alpha=1/period, min_periods=period, adjust=False).mean() / atr_s)
|
|
minus_di = 100 * (minus_dm.ewm(alpha=1/period, min_periods=period, adjust=False).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, adjust=False).mean().dropna()
|
|
if adx.empty:
|
|
return None
|
|
return float(adx.iloc[-1])
|
|
|
|
def _apply_m5_filters(self, report: SignalReport) -> None:
|
|
"""Extra selectivity for M5 — noisy TF needs stronger confirmation."""
|
|
if report.signal_quality not in ("A+", "A", "B", "C"):
|
|
return
|
|
|
|
xau_cfg = self.config.xau_strategy
|
|
abs_score = abs(report.final_score)
|
|
|
|
# Gate 1: minimum absolute score (M5 composite scores near the
|
|
# threshold are noise — require a wider margin).
|
|
if abs_score < xau_cfg.m5_min_abs_score:
|
|
report.warnings.append(
|
|
f"M5 FILTER: |score| {abs_score:.1f} < {xau_cfg.m5_min_abs_score:.0f} minimum"
|
|
)
|
|
report.signal_quality = "D"
|
|
report.trade_setup = None
|
|
return
|
|
|
|
# Gate 2: require MTF agreement (M5 alone is too noisy).
|
|
if xau_cfg.m5_require_mtf and not report.mtf_agreement:
|
|
report.warnings.append(
|
|
"M5 FILTER: no MTF agreement — M5 signals require confirmation"
|
|
)
|
|
report.signal_quality = "D"
|
|
report.trade_setup = None
|
|
return
|
|
|
|
# Gate 3: at least one confirmation TF must have a meaningful
|
|
# score in the same direction — NEUTRAL confirmations don't
|
|
# count for M5 since they indicate the move isn't visible on
|
|
# higher timeframes yet.
|
|
min_conf_score = xau_cfg.m5_min_confirmation_score
|
|
is_buy = report.final_score > 0
|
|
confirmed = False
|
|
primary_tf = self.config.primary_timeframe.value
|
|
for tf_val, tf_score in report.mtf_scores.items():
|
|
if tf_val == primary_tf:
|
|
continue
|
|
if is_buy and tf_score >= min_conf_score:
|
|
confirmed = True
|
|
break
|
|
if not is_buy and tf_score <= -min_conf_score:
|
|
confirmed = True
|
|
break
|
|
|
|
if not confirmed:
|
|
report.warnings.append(
|
|
"M5 FILTER: no confirmation TF shows directional conviction"
|
|
)
|
|
report.signal_quality = "D"
|
|
report.trade_setup = None
|
|
return
|
|
|
|
# Gate 4: cap SELL quality on M5 — A-grade SELLs enter at the
|
|
# extreme of temporary dips in gold's macro uptrend and lose
|
|
# money. B-grade SELLs (weaker conviction, shallower entries)
|
|
# are marginally profitable.
|
|
is_sell = report.final_score < 0
|
|
if is_sell and report.signal_quality not in xau_cfg.m5_sell_max_quality:
|
|
report.warnings.append(
|
|
f"M5 FILTER: SELL quality {report.signal_quality} capped — "
|
|
f"high-conviction M5 shorts enter at extremes"
|
|
)
|
|
report.signal_quality = "D"
|
|
report.trade_setup = None
|
|
return
|
|
|
|
# Gate 5: block chop-transition volatility. 'high' vol on M5 is
|
|
# elevated noise without the directional expansion of 'extreme'
|
|
# — stops get swept before targets (61% WR, negative total in
|
|
# every backtest window; low/normal/extreme all positive).
|
|
if report.volatility_regime in xau_cfg.m5_blocked_vol_regimes:
|
|
report.warnings.append(
|
|
f"M5 FILTER: '{report.volatility_regime}' volatility — "
|
|
f"chop-transition regime, stops get swept"
|
|
)
|
|
report.signal_quality = "D"
|
|
report.trade_setup = None
|
|
return
|
|
|
|
# Gate 6: SELL RSI floor — shorting with RSI already < 25 means
|
|
# entering at the extreme of the dip (same pathology as Gate 4,
|
|
# visible through RSI instead of quality grade).
|
|
if is_sell:
|
|
rsi_ind = report.primary_analysis.indicators.get("rsi")
|
|
if rsi_ind is not None and rsi_ind.value < xau_cfg.m5_sell_min_rsi:
|
|
report.warnings.append(
|
|
f"M5 FILTER: SELL blocked — RSI {rsi_ind.value:.0f} < "
|
|
f"{xau_cfg.m5_sell_min_rsi:.0f}, dip already extended"
|
|
)
|
|
report.signal_quality = "D"
|
|
report.trade_setup = None
|
|
return
|
|
|
|
# ── M5 pullback entry (XAU_PB) ──────────────────────────────────
|
|
|
|
def _detect_m5_pullback(
|
|
self, report: SignalReport, primary_df: pd.DataFrame
|
|
) -> Optional[Dict]:
|
|
"""Detect a pullback-continuation candidate on the primary TF.
|
|
|
|
Trend intact (EMA alignment >= 3/4, ADX floor) + price retraced
|
|
to EMA21/KC-mid within the last few bars + resumption bar closed
|
|
back in trend direction + RSI in the pullback zone (not the
|
|
extended zone breakout entries fire in).
|
|
"""
|
|
cfg = self.config.xau_strategy
|
|
pa = report.primary_analysis
|
|
if pa is None or len(primary_df) < 60:
|
|
return None
|
|
levels = pa.key_levels
|
|
atr = levels.get("atr", 0.0)
|
|
ema_f = levels.get("ema_fast")
|
|
ema_m = levels.get("ema_medium")
|
|
ema_s = levels.get("ema_slow")
|
|
ema_t = levels.get("ema_trend")
|
|
if not atr or None in (ema_f, ema_m, ema_s, ema_t):
|
|
return None
|
|
|
|
adx_ind = pa.indicators.get("adx")
|
|
rsi_ind = pa.indicators.get("rsi")
|
|
if adx_ind is None or rsi_ind is None:
|
|
return None
|
|
if adx_ind.value < cfg.m5_pullback_adx_min:
|
|
return None
|
|
|
|
close = float(primary_df["Close"].iloc[-1])
|
|
open_ = float(primary_df["Open"].iloc[-1])
|
|
n_touch = cfg.m5_pullback_touch_bars
|
|
recent_low = float(primary_df["Low"].iloc[-n_touch:].min())
|
|
recent_high = float(primary_df["High"].iloc[-n_touch:].max())
|
|
# Pullback swing anchor for SL: extreme of the last 8 bars.
|
|
pb_low = float(primary_df["Low"].iloc[-8:].min())
|
|
pb_high = float(primary_df["High"].iloc[-8:].max())
|
|
kc_mid = levels.get("kc_mid", ema_m)
|
|
lo_atr, hi_atr = cfg.m5_pullback_retrace_atr
|
|
rsi = rsi_ind.value
|
|
|
|
# BUY: uptrend structure, dip to the band, bullish resumption.
|
|
up_align = sum([ema_f > ema_m, ema_m > ema_s, ema_s > ema_t,
|
|
close > ema_t])
|
|
if up_align >= 3:
|
|
touched = recent_low <= max(ema_m, kc_mid)
|
|
swing_high = float(primary_df["High"].iloc[-40:].max())
|
|
depth = (swing_high - pb_low) / atr
|
|
resumed = close > open_ and close > ema_f
|
|
lo_z, hi_z = cfg.m5_pullback_rsi_buy
|
|
if (touched and resumed and lo_atr <= depth <= hi_atr
|
|
and lo_z <= rsi <= hi_z):
|
|
return {"direction": "BUY", "anchor": pb_low,
|
|
"retrace_atr": depth}
|
|
|
|
# SELL: mirrored.
|
|
dn_align = sum([ema_f < ema_m, ema_m < ema_s, ema_s < ema_t,
|
|
close < ema_t])
|
|
if dn_align >= 3:
|
|
touched = recent_high >= min(ema_m, kc_mid)
|
|
swing_low = float(primary_df["Low"].iloc[-40:].min())
|
|
depth = (pb_high - swing_low) / atr
|
|
resumed = close < open_ and close < ema_f
|
|
lo_z, hi_z = cfg.m5_pullback_rsi_sell
|
|
if (touched and resumed and lo_atr <= depth <= hi_atr
|
|
and lo_z <= rsi <= hi_z):
|
|
return {"direction": "SELL", "anchor": pb_high,
|
|
"retrace_atr": depth}
|
|
return None
|
|
|
|
def _try_m5_pullback(
|
|
self, report: SignalReport, primary_df: pd.DataFrame, pip_value: float
|
|
) -> None:
|
|
"""Build an XAU_PB trade setup if a pullback candidate passes gates."""
|
|
cfg = self.config.xau_strategy
|
|
# Rejection funnel counters (class-level diagnostic, cheap).
|
|
funnel = SignalAggregator.pb_funnel
|
|
|
|
def _reject(reason: str) -> None:
|
|
funnel[reason] = funnel.get(reason, 0) + 1
|
|
|
|
pb = self._detect_m5_pullback(report, primary_df)
|
|
if pb is None:
|
|
return
|
|
_reject("detected")
|
|
direction = pb["direction"]
|
|
is_buy = direction == "BUY"
|
|
|
|
# Directional composite floor (lower than Gate 1 by design —
|
|
# price mid-pullback scores weaker on the momentum indicators).
|
|
signed = report.final_score if is_buy else -report.final_score
|
|
if signed < cfg.m5_pullback_min_score:
|
|
_reject("score")
|
|
return
|
|
|
|
# Confirmation TF must agree (reuse Gate-3 threshold).
|
|
min_conf = cfg.m5_min_confirmation_score
|
|
confirmed = False
|
|
primary_tf = self.config.primary_timeframe.value
|
|
for tf_val, tf_score in report.mtf_scores.items():
|
|
if tf_val == primary_tf:
|
|
continue
|
|
if (is_buy and tf_score >= min_conf) or \
|
|
(not is_buy and tf_score <= -min_conf):
|
|
confirmed = True
|
|
break
|
|
if not confirmed:
|
|
_reject("confirmation_tf")
|
|
return
|
|
|
|
# Never against multi-TF consensus (mirror of gate 6f).
|
|
sigs = list(report.mtf_signals.values())
|
|
bull_n = sum(1 for s in sigs if s in (SignalStrength.BUY, SignalStrength.STRONG_BUY))
|
|
bear_n = sum(1 for s in sigs if s in (SignalStrength.SELL, SignalStrength.STRONG_SELL))
|
|
thr = self.config.signals.mtf_agreement_threshold
|
|
if (is_buy and bear_n >= thr) or (not is_buy and bull_n >= thr):
|
|
_reject("mtf_opposition")
|
|
return
|
|
|
|
# Volatility gates match the breakout path.
|
|
if report.volatility_regime in cfg.m5_blocked_vol_regimes:
|
|
_reject("vol_regime")
|
|
return
|
|
if (not is_buy and report.volatility_regime
|
|
and report.volatility_regime not in cfg.sell_vol_regimes):
|
|
_reject("sell_vol")
|
|
return
|
|
|
|
atr = report.primary_analysis.key_levels.get("atr", 0.0)
|
|
entry = report.current_price
|
|
min_sl_price = self.config.signals.min_sl_pips * pip_value
|
|
if is_buy:
|
|
sl_dist = max(entry - (pb["anchor"] - 0.5 * atr), min_sl_price)
|
|
sl = entry - sl_dist
|
|
tp1 = entry + 1.5 * sl_dist
|
|
tp2 = entry + 2.5 * sl_dist
|
|
else:
|
|
sl_dist = max((pb["anchor"] + 0.5 * atr) - entry, min_sl_price)
|
|
sl = entry + sl_dist
|
|
tp1 = entry - 1.5 * sl_dist
|
|
tp2 = entry - 2.5 * sl_dist
|
|
|
|
# Spread gate — same economics as the breakout generators.
|
|
min_tp = (self.config.signals.round_trip_cost
|
|
* self.config.signals.min_tp_cost_multiple)
|
|
if 1.5 * sl_dist < min_tp:
|
|
return
|
|
|
|
report.trade_setup = 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=1.5,
|
|
risk_pips=sl_dist / pip_value,
|
|
reward_pips=1.5 * sl_dist / pip_value,
|
|
pip_value=pip_value,
|
|
strategy="XAU_PB",
|
|
)
|
|
if report.signal_quality not in ("A+", "A", "B"):
|
|
report.signal_quality = "B"
|
|
report.final_signal = (SignalStrength.BUY if is_buy
|
|
else SignalStrength.SELL)
|
|
report.action_recommendation = (
|
|
f"{direction} — pullback continuation entry"
|
|
)
|
|
marker = (f"XAU PULLBACK STRATEGY: retrace {pb['retrace_atr']:.1f} "
|
|
f"ATR to EMA21/KC-mid, trend resumed")
|
|
target = report.bullish_reasons if is_buy else report.bearish_reasons
|
|
target.insert(0, marker)
|
|
|
|
# ── Chart-pattern entry (XAU_CP) ────────────────────────────────
|
|
|
|
def _try_chart_pattern(
|
|
self, report: SignalReport, primary_df: pd.DataFrame, pip_value: float
|
|
) -> None:
|
|
"""Build an XAU_CP trade setup if a completed chart pattern
|
|
passes the pair-level protections.
|
|
|
|
Unlike the composite paths there is no minimum-score gate: a
|
|
reversal pattern completes exactly where the composite is still
|
|
pointing the old way. Protection comes from the pattern's own
|
|
structural gates plus the validated directional filters below.
|
|
"""
|
|
cfg = self.config.xau_strategy
|
|
funnel = SignalAggregator.cp_funnel
|
|
|
|
def _reject(reason: str) -> None:
|
|
funnel[reason] = funnel.get(reason, 0) + 1
|
|
|
|
pa = report.primary_analysis
|
|
if pa is None:
|
|
return
|
|
atr = pa.key_levels.get("atr", 0.0)
|
|
# Same $-ATR floor the swing generator applies (dead tape).
|
|
if atr < cfg.min_atr_threshold:
|
|
return
|
|
|
|
pat = detect_chart_pattern(primary_df, atr, cfg)
|
|
if pat is None:
|
|
return
|
|
_reject("detected")
|
|
direction = pat["direction"]
|
|
is_buy = direction == "BUY"
|
|
|
|
# Never against multi-TF consensus (mirror of gate 6f).
|
|
sigs = list(report.mtf_signals.values())
|
|
bull_n = sum(1 for s in sigs if s in (SignalStrength.BUY, SignalStrength.STRONG_BUY))
|
|
bear_n = sum(1 for s in sigs if s in (SignalStrength.SELL, SignalStrength.STRONG_SELL))
|
|
thr = self.config.signals.mtf_agreement_threshold
|
|
if (is_buy and bear_n >= thr) or (not is_buy and bull_n >= thr):
|
|
_reject("mtf_opposition")
|
|
return
|
|
|
|
# Directional protections — gold's macro-uptrend asymmetry is
|
|
# structural, so the validated SELL gates apply to pattern
|
|
# shorts too (quiet-vol shorts fade, blocked SELL hours lose).
|
|
if not is_buy:
|
|
if (report.volatility_regime
|
|
and report.volatility_regime not in cfg.sell_vol_regimes):
|
|
_reject("sell_vol")
|
|
return
|
|
if cfg.sell_blocked_hours_utc:
|
|
try:
|
|
close_h = (
|
|
pd.Timestamp(report.timestamp)
|
|
+ pd.Timedelta(
|
|
seconds=ForexDataProvider.CANDLE_SECONDS[
|
|
self.config.primary_timeframe
|
|
]
|
|
)
|
|
).hour
|
|
except Exception:
|
|
close_h = None
|
|
if close_h in cfg.sell_blocked_hours_utc:
|
|
_reject("sell_hour")
|
|
return
|
|
if self.config.primary_timeframe == Timeframe.M5:
|
|
rsi_ind = pa.indicators.get("rsi")
|
|
if rsi_ind is not None and rsi_ind.value < cfg.m5_sell_min_rsi:
|
|
_reject("m5_sell_rsi")
|
|
return
|
|
|
|
# M5 chop-transition volatility block (validated, both ways).
|
|
if (self.config.primary_timeframe == Timeframe.M5
|
|
and report.volatility_regime in cfg.m5_blocked_vol_regimes):
|
|
_reject("vol_regime")
|
|
return
|
|
|
|
entry = report.current_price
|
|
min_sl_price = self.config.signals.min_sl_pips * pip_value
|
|
height = pat["height"]
|
|
if is_buy:
|
|
sl_dist = max(entry - (pat["anchor"] - 0.5 * atr), min_sl_price)
|
|
sl = entry - sl_dist
|
|
tp1 = entry + height
|
|
tp2 = entry + 1.5 * height
|
|
else:
|
|
sl_dist = max((pat["anchor"] + 0.5 * atr) - entry, min_sl_price)
|
|
sl = entry + sl_dist
|
|
tp1 = entry - height
|
|
tp2 = entry - 1.5 * height
|
|
|
|
# The measured move IS the pattern target — if it can't clear
|
|
# the RR floor or the spread gate, the pattern is untradeable.
|
|
rr = height / sl_dist if sl_dist > 0 else 0.0
|
|
if rr < self.config.signals.min_rr_ratio - 1e-9:
|
|
_reject("rr")
|
|
return
|
|
min_tp = (self.config.signals.round_trip_cost
|
|
* self.config.signals.min_tp_cost_multiple)
|
|
if height < min_tp:
|
|
_reject("spread_gate")
|
|
return
|
|
|
|
report.trade_setup = 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=round(rr, 2),
|
|
risk_pips=sl_dist / pip_value,
|
|
reward_pips=height / pip_value,
|
|
pip_value=pip_value,
|
|
strategy="XAU_CP",
|
|
)
|
|
if report.signal_quality not in ("A+", "A", "B"):
|
|
report.signal_quality = "B"
|
|
report.final_signal = (SignalStrength.BUY if is_buy
|
|
else SignalStrength.SELL)
|
|
report.action_recommendation = (
|
|
f"{direction} — {pat['pattern']} pattern completion"
|
|
)
|
|
marker = (f"XAU CHART PATTERN: {pat['pattern']} "
|
|
f"({direction}) break of {pat['trigger']:.2f}, "
|
|
f"height {height / atr:.1f} ATR")
|
|
target = report.bullish_reasons if is_buy else report.bearish_reasons
|
|
target.insert(0, marker)
|
|
_reject("accepted")
|
|
|
|
# ── 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 from primary data (avoids extra API call) ─
|
|
try:
|
|
report.spread_estimate = float(
|
|
(primary_df["High"] - primary_df["Low"]).median()
|
|
)
|
|
except Exception:
|
|
report.spread_estimate = 0.0
|
|
|
|
# ── 6. Final signal ────────────────────────────────────────
|
|
self._compute_final_signal(report)
|
|
|
|
# ── 6b. Session filter (works for both live and backtest) ─
|
|
if self.config.backtest.session_filter:
|
|
try:
|
|
bar_ts = pd.Timestamp(report.timestamp)
|
|
bar_hour = bar_ts.hour
|
|
bar_wd = bar_ts.weekday()
|
|
_is_xau_pair = report.display_name == "XAU/USD"
|
|
_blocked_hours = (self.config.xau_strategy.blocked_hours_utc
|
|
if _is_xau_pair else self.config.backtest.blocked_hours_utc)
|
|
if bar_hour in _blocked_hours:
|
|
report.final_signal = SignalStrength.NEUTRAL
|
|
report.signal_quality = "F"
|
|
report.warnings.append(f"SESSION FILTER: {bar_hour:02}:00 UTC blocked")
|
|
_allowed_wd = (self.config.xau_strategy.allowed_weekdays
|
|
if _is_xau_pair else self.config.backtest.allowed_weekdays)
|
|
if _allowed_wd and bar_wd not in _allowed_wd:
|
|
day_names = {0: "Mon", 1: "Tue", 2: "Wed", 3: "Thu",
|
|
4: "Fri", 5: "Sat", 6: "Sun"}
|
|
report.final_signal = SignalStrength.NEUTRAL
|
|
report.signal_quality = "F"
|
|
report.warnings.append(
|
|
f"DAY FILTER: {day_names.get(bar_wd, '?')} not in allowed days"
|
|
)
|
|
except Exception:
|
|
pass
|
|
|
|
# ── 6c. Transitioning regime quality boost ──────────────────
|
|
if report.market_regime == "transitioning" and report.signal_quality in ("B", "C", "D"):
|
|
abs_score = abs(report.final_score)
|
|
if abs_score >= 30 and report.final_confidence >= 50 and report.mtf_agreement:
|
|
report.signal_quality = "A"
|
|
report.action_recommendation = f"{report.final_signal.name.replace('_', ' ')} — Trend emerging, early entry"
|
|
elif abs_score >= 25 and report.final_confidence >= 45:
|
|
report.signal_quality = "B"
|
|
report.action_recommendation = f"{report.final_signal.name.replace('_', ' ')} — Trend forming, watch closely"
|
|
|
|
# ── 6d. Exhausted trend quality cap ───────────────────────
|
|
if (report.market_regime.startswith("trending")
|
|
and report.exhaustion_count >= 2
|
|
and report.signal_quality in ("C", "D")):
|
|
report.warnings.append(
|
|
"EXHAUSTED TREND FILTER: C-grade blocked in exhausted trend"
|
|
)
|
|
report.signal_quality = "D"
|
|
report.trade_setup = None
|
|
|
|
# ── 6e. SELL quality gate ─────────────────────────────────
|
|
xau_is_sell_dir = (
|
|
report.final_signal in (SignalStrength.SELL, SignalStrength.STRONG_SELL)
|
|
or (report.final_signal == SignalStrength.NEUTRAL and report.final_score < 0)
|
|
)
|
|
# Actionable grades only: without this guard the gate would "upgrade"
|
|
# a session-blocked F report to D, which re-opens the M5 pullback
|
|
# path (its entry check is quality != "F").
|
|
if (self.config.xau_strategy.enabled
|
|
and report.display_name == "XAU/USD"
|
|
and xau_is_sell_dir
|
|
and report.signal_quality in ("A+", "A", "B", "C")
|
|
and report.signal_quality not in self.config.xau_strategy.sell_min_quality):
|
|
report.warnings.append(
|
|
f"SELL FILTER: quality {report.signal_quality} below minimum for SELL"
|
|
)
|
|
report.signal_quality = "D"
|
|
report.trade_setup = None
|
|
|
|
# ── 6f. MTF opposition gate ───────────────────────────────
|
|
# Never trade against a multi-timeframe consensus: if the
|
|
# threshold number of TFs point opposite to the trade direction,
|
|
# block the setup (counter-trend entries in gold bleed).
|
|
if report.signal_quality in ("A+", "A", "B", "C") and report.mtf_signals:
|
|
sigs = list(report.mtf_signals.values())
|
|
bull_n = sum(1 for s in sigs if s in (SignalStrength.BUY, SignalStrength.STRONG_BUY))
|
|
bear_n = sum(1 for s in sigs if s in (SignalStrength.SELL, SignalStrength.STRONG_SELL))
|
|
thr = self.config.signals.mtf_agreement_threshold
|
|
opposed = (bear_n >= thr and report.final_score > 0) or \
|
|
(bull_n >= thr and report.final_score < 0)
|
|
if opposed:
|
|
report.warnings.append(
|
|
"MTF OPPOSITION: majority of timeframes point against signal — no trade"
|
|
)
|
|
report.signal_quality = "D"
|
|
report.trade_setup = None
|
|
|
|
# ── 6g. SELL volatility gate ──────────────────────────────
|
|
# Shorts in quiet markets fade against gold's macro uptrend;
|
|
# only take sells when volatility regime confirms real selling.
|
|
if (self.config.xau_strategy.enabled
|
|
and report.display_name == "XAU/USD"
|
|
and xau_is_sell_dir
|
|
and report.signal_quality in ("A+", "A", "B", "C")
|
|
and report.volatility_regime
|
|
and report.volatility_regime not in self.config.xau_strategy.sell_vol_regimes):
|
|
report.warnings.append(
|
|
f"SELL FILTER: volatility '{report.volatility_regime}' too quiet for shorts"
|
|
)
|
|
report.signal_quality = "D"
|
|
report.trade_setup = None
|
|
|
|
# ── 6g2. XAU regime gates (opt-in via per-TF factory overrides) ──
|
|
if (self.config.xau_strategy.enabled
|
|
and report.display_name == "XAU/USD"
|
|
and report.signal_quality in ("A+", "A", "B", "C")):
|
|
if report.market_regime in self.config.xau_strategy.blocked_regimes:
|
|
report.warnings.append(
|
|
f"REGIME FILTER: '{report.market_regime}' blocked for XAU"
|
|
)
|
|
report.signal_quality = "D"
|
|
report.trade_setup = None
|
|
elif (xau_is_sell_dir
|
|
and report.market_regime
|
|
in self.config.xau_strategy.sell_blocked_regimes):
|
|
report.warnings.append(
|
|
f"SELL FILTER: regime '{report.market_regime}' blocked for shorts"
|
|
)
|
|
report.signal_quality = "D"
|
|
report.trade_setup = None
|
|
elif xau_is_sell_dir and self.config.xau_strategy.sell_blocked_hours_utc:
|
|
# ── 6g3. XAU SELL hour gate (opt-in via per-TF factories) ──
|
|
# Evaluated on the bar's CLOSE time (= entry time under
|
|
# next-open fills), the same clock the session filter and
|
|
# backtest replay use — NOT the bar label hour.
|
|
try:
|
|
close_h = (
|
|
pd.Timestamp(report.timestamp)
|
|
+ pd.Timedelta(
|
|
seconds=ForexDataProvider.CANDLE_SECONDS[
|
|
self.config.primary_timeframe
|
|
]
|
|
)
|
|
).hour
|
|
except Exception:
|
|
close_h = None
|
|
if close_h in self.config.xau_strategy.sell_blocked_hours_utc:
|
|
report.warnings.append(
|
|
f"SELL FILTER: {close_h:02}:00 UTC entry blocked for shorts"
|
|
)
|
|
report.signal_quality = "D"
|
|
report.trade_setup = None
|
|
|
|
# ── 6i. M5 noise gate (gold-tuned thresholds — XAU only) ──
|
|
if (self.config.primary_timeframe == Timeframe.M5
|
|
and report.display_name == "XAU/USD"):
|
|
self._apply_m5_filters(report)
|
|
|
|
# ── 7. Trade setup for actionable signals ──────────────────
|
|
# ATR generator is XAU-only: it hardcodes gold pip value and
|
|
# 2-decimal price rounding.
|
|
if (report.trade_setup is None and report.signal_quality in ("A+", "A")
|
|
and report.display_name == "XAU/USD"):
|
|
atr = (report.primary_analysis.key_levels.get("atr", 0)
|
|
if report.primary_analysis else 0)
|
|
if atr > 0:
|
|
report.trade_setup = self._generate_xau_atr_trade_setup(report, atr)
|
|
if report.trade_setup is None and report.signal_quality in ("A+", "A", "B", "C"):
|
|
report.trade_setup = self._generate_trade_setup(report, pip_value)
|
|
|
|
# ── 8. Reasons ─────────────────────────────────────────────
|
|
self._collect_reasons(report)
|
|
|
|
# ── 9. M5 pullback entry (XAU_PB) ──────────────────────────
|
|
# Second archetype: when no breakout setup fired, look for a
|
|
# trend-continuation pullback (retrace to EMA21/KC-mid). Runs
|
|
# after reasons so its marker lands at reasons[0] for tagging.
|
|
if (self.config.primary_timeframe == Timeframe.M5
|
|
and report.display_name == "XAU/USD"
|
|
and self.config.xau_strategy.m5_pullback_enabled
|
|
and report.trade_setup is None
|
|
and report.signal_quality != "F"):
|
|
self._try_m5_pullback(report, primary_df, pip_value)
|
|
|
|
# ── 10. Chart-pattern entry (XAU_CP) ────────────────────────
|
|
# Third archetype: completion of a classical chart pattern
|
|
# (double top/bottom, H&S, flag — see chart_patterns.py). Only
|
|
# fills empty slots (core + pullback setups take precedence).
|
|
# Unlike XAU_PB this must NOT require quality != "F": a weak
|
|
# composite (grade F on |score| < 15) is the normal state at a
|
|
# reversal-pattern completion. Session/day blocks still bind —
|
|
# they are detected by their warnings (and the live wrapper +
|
|
# backtest replay apply the bar-close session predicate again
|
|
# on top, so a pattern setup can never trade a blocked hour).
|
|
if (report.display_name == "XAU/USD"
|
|
and self.config.xau_strategy.cp_enabled
|
|
and report.trade_setup is None
|
|
and not any(w.startswith(("SESSION FILTER", "DAY FILTER",
|
|
"WEEKEND FILTER"))
|
|
for w in report.warnings)):
|
|
self._try_chart_pattern(report, primary_df, pip_value)
|
|
|
|
return report
|
|
|
|
# ── Internals ──────────────────────────────────────────────────
|
|
|
|
_TP_QUALITY_SCALE = {"A+": 2.0, "A": 1.5, "B": 1.0, "C": 1.0}
|
|
|
|
def _generate_xau_atr_trade_setup(
|
|
self, report: SignalReport, atr: float
|
|
) -> Optional[TradeSetup]:
|
|
"""XAU ATR-based trade setup with quality-tiered TP scaling."""
|
|
cfg = self.config.xau_strategy
|
|
price = report.current_price
|
|
pip_value = 0.10
|
|
|
|
if report.final_signal in (SignalStrength.BUY, SignalStrength.STRONG_BUY):
|
|
is_buy = True
|
|
elif report.final_signal in (SignalStrength.SELL, SignalStrength.STRONG_SELL):
|
|
is_buy = False
|
|
else:
|
|
is_buy = report.final_score >= 0
|
|
|
|
quality_scale = self._TP_QUALITY_SCALE.get(report.signal_quality, 0.75)
|
|
if report.signal_quality == "A" and report.exhaustion_count >= 2:
|
|
quality_scale = 1.0
|
|
tp_mult = cfg.tp_atr_multiplier * quality_scale
|
|
|
|
if is_buy:
|
|
sl = price - cfg.sl_atr_multiplier * atr
|
|
tp1 = price + tp_mult * atr
|
|
tp2 = price + tp_mult * 1.5 * atr
|
|
direction = "BUY"
|
|
else:
|
|
sl = price + cfg.sl_atr_multiplier * atr
|
|
tp1 = price - tp_mult * atr
|
|
tp2 = price - tp_mult * 1.5 * atr
|
|
direction = "SELL"
|
|
|
|
tf = self.config.signals.tp_tightening_factor
|
|
tp1 = price + (tp1 - price) * tf
|
|
tp2 = price + (tp2 - price) * tf
|
|
|
|
sl_distance = abs(price - sl)
|
|
tp_distance = abs(tp1 - price)
|
|
rr = tp_distance / sl_distance if sl_distance > 0 else 0
|
|
|
|
min_rr = self.config.signals.min_rr_ratio
|
|
if rr < min_rr - 1e-9:
|
|
return None
|
|
|
|
# Spread gate (same rule as _generate_trade_setup).
|
|
min_tp = self.config.signals.round_trip_cost * self.config.signals.min_tp_cost_multiple
|
|
if tp_distance < min_tp:
|
|
report.warnings.append(
|
|
f"SPREAD GATE: target ${tp_distance:.2f} < ${min_tp:.2f} minimum — "
|
|
f"costs would eat the edge"
|
|
)
|
|
return None
|
|
|
|
return TradeSetup(
|
|
pair=report.symbol,
|
|
display_name=report.display_name,
|
|
direction=direction,
|
|
entry_price=price,
|
|
stop_loss=round(sl, 2),
|
|
take_profit_1=round(tp1, 2),
|
|
take_profit_2=round(tp2, 2),
|
|
risk_reward=round(rr, 2),
|
|
risk_pips=round(sl_distance / pip_value, 1),
|
|
reward_pips=round(tp_distance / pip_value, 1),
|
|
pip_value=pip_value,
|
|
)
|
|
|
|
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
|
|
# Agreement must point the same way as the primary timeframe —
|
|
# two bullish confirmation TFs must not "confirm" a bearish primary.
|
|
if report.primary_score > 0:
|
|
return bullish >= threshold
|
|
if report.primary_score < 0:
|
|
return bearish >= threshold
|
|
return bullish >= threshold or bearish >= threshold
|
|
|
|
def _compute_final_signal(self, report: SignalReport):
|
|
score = report.primary_score
|
|
confidence = report.primary_confidence
|
|
|
|
# ── Bonuses (additive, applied first) ─────────────────────
|
|
if report.mtf_agreement:
|
|
confidence += 10
|
|
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
|
|
|
|
if self.config.news.sentiment_affects_signals and abs(report.news_sentiment) > 0.2:
|
|
if (report.news_sentiment > 0 and score > 0) or \
|
|
(report.news_sentiment < 0 and score < 0):
|
|
confidence += 5
|
|
|
|
primary = report.primary_analysis
|
|
if primary and report.market_regime.startswith("trending"):
|
|
rsi_ind = primary.indicators.get("rsi")
|
|
bb_ind = primary.indicators.get("bollinger")
|
|
# Pullback bonuses require a KNOWN trend direction — the mixed
|
|
# "trending" regime (EMAs unaligned) gets no directional bonus.
|
|
if rsi_ind:
|
|
rsi_val = rsi_ind.value
|
|
if report.market_regime == "trending_up" and score > 0 and 38 <= rsi_val <= 55:
|
|
confidence += 5
|
|
elif report.market_regime == "trending_down" and score < 0 and 45 <= rsi_val <= 62:
|
|
confidence += 5
|
|
if bb_ind:
|
|
bb_pos = bb_ind.value
|
|
if report.market_regime == "trending_up" and score > 0 and bb_pos < 0.4:
|
|
confidence += 5
|
|
elif report.market_regime == "trending_down" and score < 0 and bb_pos > 0.6:
|
|
confidence += 5
|
|
|
|
# ── Event risk (hard cap, separate from penalty system) ───
|
|
# Warning always shown; the confidence cap is config-gated
|
|
# (news.event_risk_caps_signals) — see NewsConfig for the audit
|
|
# showing the cap costs R on XAU.
|
|
if report.event_risk == "danger":
|
|
report.warnings.append(f"EVENT RISK: {report.event_description}")
|
|
if self.config.news.event_risk_caps_signals:
|
|
confidence = min(confidence, 30)
|
|
report.action_recommendation = f"WAIT — {report.event_description}"
|
|
|
|
# ── Penalties (diminishing cap system) ────────────────────
|
|
# Collect all applicable penalties, then apply with exponential
|
|
# decay so stacking never guts a signal beyond 30%.
|
|
penalties = []
|
|
|
|
if not report.mtf_agreement:
|
|
penalties.append(("mtf_disagreement", 15))
|
|
if len(report.mtf_scores) >= 2:
|
|
scores_list = list(report.mtf_scores.values())
|
|
if any(s > 10 for s in scores_list) and any(s < -10 for s in scores_list):
|
|
report.warnings.append("Timeframes CONFLICT — mixed signals across timeframes")
|
|
|
|
if abs(report.news_sentiment) > 0.2:
|
|
if (report.news_sentiment > 0 and score < -10) or \
|
|
(report.news_sentiment < 0 and score > 10):
|
|
report.warnings.append("News sentiment CONFLICTS with technical signal")
|
|
if self.config.news.sentiment_affects_signals:
|
|
penalties.append(("news_conflict", 10))
|
|
|
|
if report.event_risk == "caution":
|
|
report.warnings.append(f"Upcoming event: {report.event_description}")
|
|
if self.config.news.event_risk_caps_signals:
|
|
penalties.append(("event_caution", 15))
|
|
|
|
is_counter_trend = (
|
|
(report.market_regime == "trending_up" and score < 0) or
|
|
(report.market_regime == "trending_down" and score > 0)
|
|
)
|
|
if is_counter_trend:
|
|
report.warnings.append("Counter-trend signal — reduced confidence")
|
|
penalties.append(("counter_trend", 15))
|
|
|
|
if report.volatility_regime == "extreme":
|
|
report.warnings.append("EXTREME volatility — wider stops needed")
|
|
penalties.append(("extreme_vol", 10))
|
|
elif report.volatility_regime == "low":
|
|
report.warnings.append("Low volatility — potential for breakout")
|
|
|
|
if primary is not None:
|
|
ema200 = primary.key_levels.get("ema_trend")
|
|
if ema200 and ema200 > 0 and report.current_price > 0:
|
|
ema_dist_pct = abs(report.current_price - ema200) / ema200 * 100.0
|
|
_lo = float(os.environ.get("FB2_EMA_LO", "0.5"))
|
|
if ema_dist_pct < _lo:
|
|
report.warnings.append(f"NEAR EMA200 ({ema_dist_pct:.2f}%) — chop risk")
|
|
penalties.append(("ema200_near", 5))
|
|
elif ema_dist_pct > 2.5:
|
|
report.warnings.append(f"FAR FROM EMA200 ({ema_dist_pct:.2f}%) — over-extended")
|
|
penalties.append(("ema200_far", 5))
|
|
|
|
atr = primary.key_levels.get("atr", 0)
|
|
if atr > 0 and score != 0:
|
|
if score > 0:
|
|
swing_lows = sorted(
|
|
v for k, v in primary.key_levels.items() if k.startswith("swing_low_")
|
|
)
|
|
if swing_lows and report.current_price - swing_lows[0] > 3.5 * atr:
|
|
report.warnings.append("TREND EXHAUSTION — price extended >3.5 ATR past swing low")
|
|
penalties.append(("exhaustion", 8))
|
|
else:
|
|
swing_highs = sorted(
|
|
(v for k, v in primary.key_levels.items() if k.startswith("swing_high_")),
|
|
reverse=True,
|
|
)
|
|
if swing_highs and swing_highs[0] - report.current_price > 3.5 * atr:
|
|
report.warnings.append("TREND EXHAUSTION — price extended >3.5 ATR past swing high")
|
|
penalties.append(("exhaustion", 8))
|
|
|
|
ema_medium = primary.key_levels.get("ema_medium")
|
|
if atr > 0 and ema_medium and report.market_regime.startswith("trending"):
|
|
pullback_dist = abs(report.current_price - ema_medium) / atr
|
|
if pullback_dist > 2.5:
|
|
report.warnings.append(f"NO PULLBACK — price {pullback_dist:.1f} ATR from EMA21")
|
|
penalties.append(("no_pullback", 8))
|
|
elif pullback_dist > 2.0:
|
|
report.warnings.append(f"Weak pullback — price {pullback_dist:.1f} ATR from EMA21")
|
|
penalties.append(("weak_pullback", 5))
|
|
|
|
if report.market_regime.startswith("trending"):
|
|
rsi_ind = primary.indicators.get("rsi")
|
|
bb_ind = primary.indicators.get("bollinger")
|
|
# Extension penalties apply in ANY strong-ADX regime; the
|
|
# mixed "trending" regime penalizes BOTH directions (it
|
|
# previously penalized only BUYs — the SELL side of the
|
|
# same bullish-bias bug fixed in the RSI reinterpretation).
|
|
if rsi_ind:
|
|
rsi_val = rsi_ind.value
|
|
if report.market_regime in ("trending_up", "trending") and score > 0 and rsi_val > 65:
|
|
report.warnings.append(f"Extended — RSI {rsi_val:.0f} in uptrend")
|
|
penalties.append(("rsi_extended", 5))
|
|
elif report.market_regime in ("trending_down", "trending") and score < 0 and rsi_val < 35:
|
|
report.warnings.append(f"Extended — RSI {rsi_val:.0f} in downtrend")
|
|
penalties.append(("rsi_extended", 5))
|
|
if bb_ind:
|
|
bb_pos = bb_ind.value
|
|
if report.market_regime in ("trending_up", "trending") and score > 0 and bb_pos > 0.85:
|
|
penalties.append(("bb_overextended", 8))
|
|
elif report.market_regime in ("trending_down", "trending") and score < 0 and bb_pos < 0.15:
|
|
penalties.append(("bb_overextended", 8))
|
|
|
|
# Hard block: static exhaustion + fading momentum → kill the trade
|
|
# Static signals alone are normal in gold trends. Only block when
|
|
# price is extended AND momentum indicators confirm deceleration.
|
|
_EXHAUSTION_TAGS = {"exhaustion", "no_pullback", "rsi_extended", "bb_overextended"}
|
|
exhaustion_count = sum(1 for name, _ in penalties if name in _EXHAUSTION_TAGS)
|
|
report.exhaustion_count = exhaustion_count
|
|
momentum_fading = False
|
|
if primary is not None and exhaustion_count >= 2:
|
|
roc_ind = primary.indicators.get("roc")
|
|
macd_ind = primary.indicators.get("macd")
|
|
roc_fading = False
|
|
macd_fading = False
|
|
if roc_ind:
|
|
if score > 0 and roc_ind.value < 0:
|
|
roc_fading = True
|
|
elif score < 0 and roc_ind.value > 0:
|
|
roc_fading = True
|
|
if macd_ind:
|
|
if score > 0 and macd_ind.signal in (SignalStrength.SELL, SignalStrength.STRONG_SELL):
|
|
macd_fading = True
|
|
elif score < 0 and macd_ind.signal in (SignalStrength.BUY, SignalStrength.STRONG_BUY):
|
|
macd_fading = True
|
|
momentum_fading = roc_fading or macd_fading
|
|
exhaustion_blocked = exhaustion_count >= 2 and momentum_fading
|
|
|
|
MAX_PENALTY = 30.0
|
|
penalties.sort(key=lambda x: x[1], reverse=True)
|
|
total_penalty = 0.0
|
|
decay = 1.0
|
|
for _name, amount in penalties:
|
|
applied = amount * decay
|
|
if total_penalty + applied > MAX_PENALTY:
|
|
applied = MAX_PENALTY - total_penalty
|
|
total_penalty += applied
|
|
decay *= 0.5
|
|
if total_penalty >= MAX_PENALTY:
|
|
break
|
|
|
|
confidence -= total_penalty
|
|
confidence = max(0.0, min(95.0, confidence))
|
|
|
|
if exhaustion_blocked:
|
|
report.warnings.append(
|
|
f"EXHAUSTION BLOCK — {exhaustion_count} exhaustion signals fired, trade rejected"
|
|
)
|
|
confidence = 0.0
|
|
|
|
# Final signal
|
|
if score >= 35 and confidence >= 55:
|
|
report.final_signal = SignalStrength.STRONG_BUY
|
|
elif score >= 25 and confidence >= 50:
|
|
report.final_signal = SignalStrength.BUY
|
|
elif score <= -35 and confidence >= 55:
|
|
report.final_signal = SignalStrength.STRONG_SELL
|
|
elif score <= -25 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 >= 40 and confidence >= 65 and report.mtf_agreement:
|
|
report.signal_quality = "A+"
|
|
elif abs_score >= 33 and confidence >= 55 and report.mtf_agreement:
|
|
report.signal_quality = "A"
|
|
elif abs_score >= 28 and confidence >= 50:
|
|
report.signal_quality = "B"
|
|
elif abs_score >= 25 and confidence >= 45:
|
|
report.signal_quality = "C"
|
|
elif abs_score >= 15:
|
|
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
|
|
if pip_value <= 0:
|
|
return None
|
|
|
|
# Fix 2: Minimum ATR threshold for XAU/USD
|
|
if report.display_name == "XAU/USD" and atr < self.config.xau_strategy.min_atr_threshold:
|
|
report.warnings.append(
|
|
f"LOW ATR FILTER: ATR ${atr:.2f} < ${self.config.xau_strategy.min_atr_threshold:.2f} minimum"
|
|
)
|
|
return None
|
|
|
|
if report.final_signal in (SignalStrength.BUY, SignalStrength.STRONG_BUY):
|
|
is_buy = True
|
|
elif report.final_signal in (SignalStrength.SELL, SignalStrength.STRONG_SELL):
|
|
is_buy = False
|
|
else:
|
|
is_buy = report.final_score >= 0
|
|
levels = report.primary_analysis.key_levels
|
|
|
|
min_sl_price = self.config.signals.min_sl_pips * pip_value
|
|
min_rr = self.config.signals.min_rr_ratio
|
|
tp_floor = max(min_rr, self.config.signals.tp_floor_rr)
|
|
tf = self.config.signals.tp_tightening_factor
|
|
pre_tight_min_rr = tp_floor / tf if tf > 0 else tp_floor
|
|
|
|
tp_mult = 1.5
|
|
tp_swing_cap = 2.5
|
|
|
|
sl_atr_floor = (self.config.xau_strategy.v2_sl_atr_multiplier
|
|
if report.display_name == "XAU/USD" else 1.5)
|
|
sl_swing_buffer = 1.0
|
|
|
|
if is_buy:
|
|
swing_lows = sorted(
|
|
[v for k, v in levels.items() if k.startswith("swing_low_") and v < price],
|
|
reverse=True,
|
|
)
|
|
if swing_lows:
|
|
sl_price = swing_lows[0] - atr * sl_swing_buffer
|
|
else:
|
|
sl_price = price - atr * (1.0 + sl_swing_buffer)
|
|
min_sl = max(atr * sl_atr_floor, min_sl_price)
|
|
if price - sl_price < min_sl:
|
|
sl_price = price - min_sl
|
|
|
|
sl_distance = price - sl_price
|
|
|
|
swing_highs = sorted(
|
|
v for k, v in levels.items() if k.startswith("swing_high_") and v > price + atr * 0.3
|
|
)
|
|
if (swing_highs
|
|
and swing_highs[0] - price <= atr * tp_swing_cap
|
|
and swing_highs[0] - price >= sl_distance * pre_tight_min_rr):
|
|
tp1 = swing_highs[0]
|
|
else:
|
|
tp1 = price + max(atr * tp_mult, sl_distance * pre_tight_min_rr)
|
|
if len(swing_highs) >= 2 and swing_highs[1] - price <= atr * 5.0:
|
|
tp2 = swing_highs[1]
|
|
else:
|
|
tp2 = price + max(atr * (tp_mult + 1.5), sl_distance * pre_tight_min_rr * 1.5)
|
|
|
|
direction = "BUY"
|
|
entry, sl = price, sl_price
|
|
else:
|
|
swing_highs = sorted(
|
|
v for k, v in levels.items() if k.startswith("swing_high_") and v > price
|
|
)
|
|
if swing_highs:
|
|
sl_price = swing_highs[0] + atr * sl_swing_buffer
|
|
else:
|
|
sl_price = price + atr * (1.0 + sl_swing_buffer)
|
|
min_sl = max(atr * sl_atr_floor, min_sl_price)
|
|
if sl_price - price < min_sl:
|
|
sl_price = price + min_sl
|
|
|
|
sl_distance = sl_price - price
|
|
|
|
swing_lows = sorted(
|
|
[v for k, v in levels.items() if k.startswith("swing_low_") and v < price - atr * 0.3],
|
|
reverse=True,
|
|
)
|
|
if (swing_lows
|
|
and price - swing_lows[0] <= atr * tp_swing_cap
|
|
and price - swing_lows[0] >= sl_distance * pre_tight_min_rr):
|
|
tp1 = swing_lows[0]
|
|
else:
|
|
tp1 = price - max(atr * tp_mult, sl_distance * pre_tight_min_rr)
|
|
if len(swing_lows) >= 2 and price - swing_lows[1] <= atr * 5.0:
|
|
tp2 = swing_lows[1]
|
|
else:
|
|
tp2 = price - max(atr * (tp_mult + 1.5), sl_distance * pre_tight_min_rr * 1.5)
|
|
|
|
direction = "SELL"
|
|
entry, sl = price, sl_price
|
|
|
|
tf = self.config.signals.tp_tightening_factor
|
|
tp1 = entry + (tp1 - entry) * tf
|
|
tp2 = entry + (tp2 - entry) * tf
|
|
|
|
# TP2 must sit beyond TP1 — swing picks can invert when TP1 falls
|
|
# back to the ATR target while TP2 takes a nearer second swing.
|
|
if is_buy:
|
|
tp2 = max(tp2, tp1 + atr)
|
|
else:
|
|
tp2 = min(tp2, tp1 - atr)
|
|
|
|
tp1_distance = abs(tp1 - entry)
|
|
rr = tp1_distance / sl_distance if sl_distance > 0 else 0
|
|
|
|
if rr < min_rr - 1e-9:
|
|
return None
|
|
|
|
# Spread gate: a target too close to entry hands most of the win
|
|
# to transaction costs (validated on M5: sub-$10 targets lose).
|
|
min_tp = self.config.signals.round_trip_cost * self.config.signals.min_tp_cost_multiple
|
|
if tp1_distance < min_tp:
|
|
report.warnings.append(
|
|
f"SPREAD GATE: target ${tp1_distance:.2f} < ${min_tp:.2f} minimum — "
|
|
f"costs would eat the edge"
|
|
)
|
|
return None
|
|
|
|
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=round(rr, 2),
|
|
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}")
|