""" Confidence Engine — the "brain" of the bot. Combines trend, price action, volume, indicators (all computed locally from OHLCV) with option-chain, news, and global-market scores (fed in externally, since those modules are later phases) into one 0-100 confidence score. Trade only if total >= MIN_CONFIDENCE_TO_TRADE, else "NO TRADE TODAY". This module is deliberately transparent: every sub-score comes back with a plain-English reason list, so you can see exactly *why* the bot did or didn't want to trade. Never trust a black-box score on real capital. """ from dataclasses import dataclass, field import pandas as pd from config import settings as cfg from core import indicators as ind from core import candlestick_patterns as candles from core import price_action as pa @dataclass class ScoreBreakdown: trend: float = 0.0 price_action: float = 0.0 volume: float = 0.0 option_chain: float = 0.0 indicators: float = 0.0 news: float = 0.0 global_market: float = 0.0 reasons: list = field(default_factory=list) @property def total(self) -> float: return ( self.trend + self.price_action + self.volume + self.option_chain + self.indicators + self.news + self.global_market ) @dataclass class Decision: symbol: str total_confidence: float should_trade: bool direction: str # "BUY", "SELL", or "NONE" breakdown: ScoreBreakdown entry: float = None stop_loss: float = None target: float = None def summary(self) -> str: if not self.should_trade: lines = [f"[{self.symbol}] NO TRADE TODAY (confidence {self.total_confidence:.1f}/100, " f"need >= {cfg.MIN_CONFIDENCE_TO_TRADE})"] else: lines = [ f"[{self.symbol}] TRADE: {self.direction} (confidence {self.total_confidence:.1f}/100)", f" Entry: {self.entry:.2f} SL: {self.stop_loss:.2f} Target: {self.target:.2f} " f"(R:R = 1:{cfg.RISK_REWARD_RATIO})", ] lines.append(" Reasons:") for r in self.breakdown.reasons: lines.append(f" - {r}") return "\n".join(lines) def _score_trend(df: pd.DataFrame, structure: dict, weight: int, reasons: list) -> float: close = df["close"] ema50 = ind.ema(close, 50).iloc[-1] ema200 = ind.ema(close, 200).iloc[-1] if len(df) >= 200 else None price = close.iloc[-1] score = 0.0 aligned_up = price > ema50 and structure["trend"] == "uptrend" aligned_down = price < ema50 and structure["trend"] == "downtrend" if aligned_up or aligned_down: score += weight * 0.6 reasons.append(f"Price aligned with {structure['trend']} vs EMA50") if ema200 is not None: if (price > ema50 > ema200) or (price < ema50 < ema200): score += weight * 0.4 reasons.append("EMA stack confirms trend (price/EMA50/EMA200 aligned)") return min(score, weight) def _score_price_action(df: pd.DataFrame, pa_summary: dict, weight: int, reasons: list) -> float: score = 0.0 if pa_summary["last_bos"] is not None: score += weight * 0.4 reasons.append(f"Break of Structure confirmed ({pa_summary['last_bos']['direction']})") if pa_summary["bias"] == "continuation": score += weight * 0.3 reasons.append("Structure shows trend continuation, not reversal") if pa_summary["liquidity_sweeps"]: score += weight * 0.15 reasons.append("Recent liquidity sweep detected (stop hunt before move)") if pa_summary["fvgs"]: score += weight * 0.15 reasons.append("Unfilled Fair Value Gap present nearby") return min(score, weight) def _score_volume(df: pd.DataFrame, weight: int, reasons: list) -> float: vol = df["volume"] avg_vol = vol.rolling(20).mean().iloc[-1] last_vol = vol.iloc[-1] score = 0.0 if pd.notna(avg_vol) and avg_vol > 0: ratio = last_vol / avg_vol if ratio >= 1.5: score += weight reasons.append(f"Volume {ratio:.1f}x above 20-bar average — real participation") elif ratio >= 1.0: score += weight * 0.5 reasons.append(f"Volume near average ({ratio:.1f}x) — acceptable but not strong") else: reasons.append(f"Volume below average ({ratio:.1f}x) — weak conviction, avoided") return min(score, weight) def _score_indicators(df: pd.DataFrame, weight: int, reasons: list) -> float: close = df["close"] votes = 0 total_checks = 0 macd_df = ind.macd(close) total_checks += 1 if macd_df["macd"].iloc[-1] > macd_df["signal"].iloc[-1]: votes += 1 reasons.append("MACD bullish crossover") elif macd_df["macd"].iloc[-1] < macd_df["signal"].iloc[-1]: votes -= 1 reasons.append("MACD bearish crossover") rsi_val = ind.rsi(close).iloc[-1] total_checks += 1 if 50 < rsi_val < 70: votes += 1 reasons.append(f"RSI {rsi_val:.1f} — bullish momentum without being overbought") elif 30 < rsi_val < 50: votes -= 1 reasons.append(f"RSI {rsi_val:.1f} — bearish momentum") elif rsi_val >= 70: reasons.append(f"RSI {rsi_val:.1f} — overbought, caution") elif rsi_val <= 30: reasons.append(f"RSI {rsi_val:.1f} — oversold, caution") adx_df = ind.adx(df) total_checks += 1 if adx_df["adx"].iloc[-1] >= 25: votes += 1 if adx_df["plus_di"].iloc[-1] > adx_df["minus_di"].iloc[-1] else -1 reasons.append(f"ADX {adx_df['adx'].iloc[-1]:.1f} — trending market (not sideways)") else: reasons.append(f"ADX {adx_df['adx'].iloc[-1]:.1f} — weak/sideways trend, penalized") st_df = ind.supertrend(df) total_checks += 1 votes += 1 if st_df["trend"].iloc[-1] == 1 else -1 agreement = abs(votes) / total_checks if total_checks else 0 return weight * agreement, votes def calculate_confidence( df: pd.DataFrame, symbol: str, option_chain_score: float = 0.0, news_score: float = 0.0, global_market_score: float = 0.0, ) -> Decision: """ df: OHLCV dataframe, most recent bar last. option_chain_score / news_score / global_market_score: 0-100 scores from their respective modules (Phase 4+). Pass 0 if not available yet — the bot will simply never reach 90 confidence without them, which is the correct conservative behavior (better to under-trade than fake it). """ weights = cfg.CONFIDENCE_WEIGHTS reasons = [] structure = pa.market_structure(df) pa_summary = pa.analyze(df) pattern_signal = candles.latest_signal(df) trend_score = _score_trend(df, structure, weights["trend"], reasons) price_action_score = _score_price_action(df, pa_summary, weights["price_action"], reasons) volume_score = _score_volume(df, weights["volume"], reasons) indicators_score, indicator_votes = _score_indicators(df, weights["indicators"], reasons) option_chain_pts = weights["option_chain"] * (option_chain_score / 100) news_pts = weights["news"] * (news_score / 100) global_pts = weights["global_market"] * (global_market_score / 100) if option_chain_score == 0: reasons.append("Option chain score not provided (module not wired yet) — contributes 0") if news_score == 0: reasons.append("News score not provided (module not wired yet) — contributes 0") if global_market_score == 0: reasons.append("Global market score not provided (module not wired yet) — contributes 0") if pattern_signal: reasons.append(f"Candlestick pattern(s) on latest bar: {pattern_signal}") breakdown = ScoreBreakdown( trend=trend_score, price_action=price_action_score, volume=volume_score, option_chain=option_chain_pts, indicators=indicators_score, news=news_pts, global_market=global_pts, reasons=reasons, ) total = breakdown.total should_trade = total >= cfg.MIN_CONFIDENCE_TO_TRADE direction = "NONE" entry = stop_loss = target = None if should_trade: direction = "BUY" if indicator_votes > 0 and structure["trend"] != "downtrend" else "SELL" entry = float(df["close"].iloc[-1]) atr_val = float(ind.atr(df).iloc[-1]) if direction == "BUY": stop_loss = entry - atr_val target = entry + atr_val * cfg.RISK_REWARD_RATIO else: stop_loss = entry + atr_val target = entry - atr_val * cfg.RISK_REWARD_RATIO return Decision( symbol=symbol, total_confidence=total, should_trade=should_trade, direction=direction, breakdown=breakdown, entry=entry, stop_loss=stop_loss, target=target, )