#!/usr/bin/env python3 """ ForexBot2 — XAU/USD Signal Intelligence ======================================== Usage: python main.py # Live signal on H1 python main.py --timeframe H4 --detail # Detailed H4 signal python main.py --json # JSON output """ import argparse import json import logging import sys from datetime import datetime, timezone from typing import Dict, List # Force UTF-8 output on Windows so box-drawing chars render if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(encoding="utf-8") if hasattr(sys.stderr, "reconfigure"): sys.stderr.reconfigure(encoding="utf-8") from config import PlatformConfig, Timeframe, SignalStrength, XAUUSD from signal_aggregator import SignalAggregator, SignalReport # ══════════════════════════════════════════════════════════════════ # TERMINAL FORMATTING # ══════════════════════════════════════════════════════════════════ class Colors: RESET = "\033[0m" BOLD = "\033[1m" DIM = "\033[2m" UNDERLINE = "\033[4m" RED = "\033[91m" GREEN = "\033[92m" YELLOW = "\033[93m" BLUE = "\033[94m" MAGENTA = "\033[95m" CYAN = "\033[96m" WHITE = "\033[97m" GRAY = "\033[90m" def signal_color(signal: SignalStrength) -> str: return { SignalStrength.STRONG_BUY: Colors.GREEN + Colors.BOLD, SignalStrength.BUY: Colors.GREEN, SignalStrength.NEUTRAL: Colors.GRAY, SignalStrength.SELL: Colors.RED, SignalStrength.STRONG_SELL: Colors.RED + Colors.BOLD, }.get(signal, Colors.RESET) def quality_color(q: str) -> str: return { "A+": Colors.GREEN + Colors.BOLD, "A": Colors.GREEN, "B": Colors.CYAN, "C": Colors.YELLOW, "D": Colors.GRAY, "F": Colors.DIM, }.get(q, Colors.RESET) def print_header(title: str = "XAU/USD SIGNAL INTELLIGENCE"): now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC") print(f"\n{Colors.CYAN}{'═' * 80}{Colors.RESET}") print(f"{Colors.BOLD}{Colors.CYAN} {title}{Colors.RESET}") print(f"{Colors.GRAY} Time: {now}{Colors.RESET}") print(f"{Colors.CYAN}{'═' * 80}{Colors.RESET}\n") # ── Live signal renderer ─────────────────────────────────────────── _INDICATOR_PREFIXES = ( "EMA Trend", "RSI", "Bollinger Bands", "Divergence", "Candlestick", "ADX", "ATR", "News", "Regime", ) def _compact_reason(reason: str) -> str: """Shorten a verbose reason string for the compact view.""" s = reason if s.startswith("STRONG: "): s = s[8:] # Strip the leading ": " prefix — it's always an indicator name for prefix in _INDICATOR_PREFIXES: if s.startswith(prefix + ": "): s = s[len(prefix) + 2:] break # Drop trailing detail after first ' | ' if " | " in s: s = s.split(" | ", 1)[0] return s.strip() def print_signal_report(report: SignalReport, detail: bool = False): sc = signal_color(report.final_signal) qc = quality_color(report.signal_quality) sig_name = report.final_signal.name.replace("_", " ") # ── 1. DECISION ──────────────────────────────────────────────── mtf_tag = (f"{Colors.GREEN}MTF✓{Colors.RESET}" if report.mtf_agreement else f"{Colors.YELLOW}MTF✗{Colors.RESET}") print(f" {sc}▶ {sig_name}{Colors.RESET} " f"{qc}{report.signal_quality}{Colors.RESET} · " f"{sc}{report.final_score:+.1f}{Colors.RESET} · " f"{report.final_confidence:.0f}% · {mtf_tag}") print(f" {Colors.GRAY}{report.action_recommendation}{Colors.RESET}") # ── 2. TRADE SETUP (if signal is actionable) ─────────────────── if report.trade_setup: ts = report.trade_setup dc = Colors.GREEN if ts.direction == "BUY" else Colors.RED print(f"\n {Colors.BOLD}Setup{Colors.RESET} " f"{dc}{ts.direction}{Colors.RESET} @ {Colors.BOLD}{ts.entry_price:.3f}{Colors.RESET} " f"→ SL {Colors.RED}{ts.stop_loss:.3f}{Colors.RESET} " f"({ts.risk_pips:.1f}p) " f"→ TP {Colors.GREEN}{ts.take_profit_1:.3f}{Colors.RESET} " f"({ts.reward_pips:.1f}p) " f"RR 1:{ts.risk_reward:.1f}") # ── 3. CONTEXT (price · regime · vol · MTF breakdown) ────────── print(f"\n {Colors.WHITE}{report.display_name}{Colors.RESET} " f"{Colors.BOLD}{report.current_price:.3f}{Colors.RESET} · " f"{Colors.CYAN}{report.market_regime}{Colors.RESET} · " f"vol {report.volatility_regime}") if report.mtf_signals: parts = [] for tf, sig in report.mtf_signals.items(): tf_sc = signal_color(sig) sc_val = report.mtf_scores.get(tf, 0) parts.append(f"{tf} {tf_sc}{sig.name.replace('_',' ')}{Colors.RESET} {sc_val:+.1f}") print(f" {Colors.GRAY}MTF{Colors.RESET} " + " · ".join(parts)) # ── 4. WHY — top 3 each side, compact ────────────────────────── bulls = [_compact_reason(r) for r in report.bullish_reasons[:3]] bears = [_compact_reason(r) for r in report.bearish_reasons[:3]] if bulls or bears: print() rows = max(len(bulls), len(bears)) for i in range(rows): left = f"{Colors.GREEN}+{Colors.RESET} {bulls[i]}" if i < len(bulls) else "" right = f"{Colors.RED}-{Colors.RESET} {bears[i]}" if i < len(bears) else "" print(f" {left:<60} {right}") # ── 5. WARNINGS / NEWS / EVENTS ──────────────────────────────── if report.warnings: for w in report.warnings: print(f" {Colors.YELLOW}⚠{Colors.RESET} {w}") if report.event_description and report.event_risk != "clear": print(f" {Colors.YELLOW}📅{Colors.RESET} {report.event_description}") if report.news_description and report.news_description != "Neutral news sentiment": print(f" {Colors.MAGENTA}📰{Colors.RESET} {report.news_description}") # ── 6. DETAIL (only with --detail) ───────────────────────────── if detail and report.primary_analysis: print(f"\n {Colors.BOLD}Indicators ({report.primary_analysis.timeframe}){Colors.RESET}") for name, ind in report.primary_analysis.indicators.items(): if name.startswith("_") or ind.weight == 0: continue isc = signal_color(ind.signal) print(f" {ind.name:<16} {isc}{ind.signal.name.replace('_',' '):<11}{Colors.RESET} " f"{ind.confidence:>3.0f}% {Colors.GRAY}│ {ind.description}{Colors.RESET}") levels = report.primary_analysis.key_levels if levels: swing_keys = sorted( [k for k in levels if k.startswith("swing_")], key=lambda k: levels[k], reverse=True, ) if swing_keys: print(f"\n {Colors.BOLD}Structure Levels{Colors.RESET}") for k in swing_keys: val = levels[k] color = Colors.RED if val > report.current_price else Colors.GREEN label = k.replace("_", " ").title() print(f" {label:>20} {color}{val:.3f}{Colors.RESET}") def report_to_dict(report: SignalReport) -> dict: d = { "symbol": report.symbol, "display_name": report.display_name, "timestamp": report.timestamp, "current_price": report.current_price, "signal": report.final_signal.name, "score": round(report.final_score, 2), "confidence": round(report.final_confidence, 2), "quality": report.signal_quality, "action": report.action_recommendation, "market_regime": report.market_regime, "volatility_regime": report.volatility_regime, "mtf_agreement": report.mtf_agreement, "mtf_signals": {k: v.name for k, v in report.mtf_signals.items()}, "event_risk": report.event_risk, "event_description": report.event_description, "news_sentiment": round(report.news_sentiment, 3), "bullish_reasons": report.bullish_reasons, "bearish_reasons": report.bearish_reasons, "warnings": report.warnings, } if report.trade_setup: ts = report.trade_setup d["trade_setup"] = { "direction": ts.direction, "entry": ts.entry_price, "stop_loss": ts.stop_loss, "take_profit_1": ts.take_profit_1, "take_profit_2": ts.take_profit_2, "risk_reward": ts.risk_reward, "risk_pips": round(ts.risk_pips, 1), "reward_pips": round(ts.reward_pips, 1), } if report.primary_analysis: d["indicators"] = { name: { "signal": ind.signal.name, "confidence": round(ind.confidence, 1), "description": ind.description, } for name, ind in report.primary_analysis.indicators.items() if not name.startswith("_") } d["key_levels"] = { k: round(v, 5) for k, v in report.primary_analysis.key_levels.items() } return d # ══════════════════════════════════════════════════════════════════ # CLI # ══════════════════════════════════════════════════════════════════ def parse_args(): parser = argparse.ArgumentParser( description="ForexBot2 — XAU/USD Signal Intelligence", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=""" Examples: python main.py Live signal (H1, summary) python main.py --detail Live signal with full breakdown python main.py --timeframe H4 --detail H4 timeframe python main.py --json JSON output """, ) parser.add_argument("--timeframe", "-tf", default="H1", choices=["M5", "M15", "M30", "H1", "H4", "D1"], help="Primary analysis timeframe (default: H1)") parser.add_argument("--detail", action="store_true", help="Show full indicator breakdown") parser.add_argument("--json", action="store_true", help="Output as JSON") parser.add_argument("--verbose", "-v", action="store_true", help="Debug logging") return parser.parse_args() def build_config(args) -> PlatformConfig: config = PlatformConfig() tf_map = { "M5": Timeframe.M5, "M15": Timeframe.M15, "M30": Timeframe.M30, "H1": Timeframe.H1, "H4": Timeframe.H4, "D1": Timeframe.D1, } config.primary_timeframe = tf_map[args.timeframe] tf_confirmations = { "M5": [Timeframe.M1, Timeframe.M15], "M15": [Timeframe.M5, Timeframe.H1], "M30": [Timeframe.M15, Timeframe.H1], "H1": [Timeframe.M15, Timeframe.H4], "H4": [Timeframe.H1, Timeframe.D1], "D1": [Timeframe.H4, Timeframe.W1], } config.confirmation_timeframes = tf_confirmations.get( args.timeframe, [Timeframe.M15, Timeframe.H4] ) return config def run_live(args, config: PlatformConfig): print(f"{Colors.GRAY} Fetching market data and running analysis...{Colors.RESET}\n") aggregator = SignalAggregator(config) report = aggregator.analyze_pair() if args.json: print(json.dumps(report_to_dict(report), indent=2)) return print_header("XAU/USD SIGNAL INTELLIGENCE") print_signal_report(report, detail=args.detail) print(f"\n{Colors.CYAN}{'═' * 80}{Colors.RESET}") print(f"{Colors.GRAY} Disclaimer: For educational/research purposes only.{Colors.RESET}") print(f"{Colors.CYAN}{'═' * 80}{Colors.RESET}\n") def main(): args = parse_args() log_level = logging.DEBUG if args.verbose else logging.WARNING logging.basicConfig( level=log_level, format="%(asctime)s [%(levelname)s] %(name)s: %(message)s", datefmt="%H:%M:%S", ) config = build_config(args) run_live(args, config) if __name__ == "__main__": main()