From 43319d9a98097dd7803ec61c2fbb649ff9f907a1 Mon Sep 17 00:00:00 2001 From: harry Date: Mon, 4 May 2026 04:10:17 +0000 Subject: [PATCH] Add main.py --- main.py | 503 ++++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 503 insertions(+) create mode 100644 main.py diff --git a/main.py b/main.py new file mode 100644 index 0000000..6a58e85 --- /dev/null +++ b/main.py @@ -0,0 +1,503 @@ +#!/usr/bin/env python3 +""" +Forex Signal Intelligence Platform +==================================== +Main entry point. Runs full analysis and outputs formatted signal reports. + +Usage: + python main.py # Analyze all major pairs + python main.py --pairs EURUSD GBPUSD # Specific pairs + python main.py --timeframe H4 # Different primary TF + python main.py --json # JSON output + python main.py --watch # Continuous monitoring mode +""" + +import argparse +import json +import logging +import sys +import time +from datetime import datetime, timezone +from typing import List + +from config import ( + PlatformConfig, Timeframe, SignalStrength, + MAJOR_PAIRS, CROSS_PAIRS, SYMBOL_DISPLAY, +) +from signal_aggregator import SignalAggregator, SignalReport + + +# ══════════════════════════════════════════════════════════════════ +# TERMINAL FORMATTING +# ══════════════════════════════════════════════════════════════════ + +class Colors: + """ANSI color codes for terminal output.""" + 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" + + BG_RED = "\033[41m" + BG_GREEN = "\033[42m" + BG_YELLOW = "\033[43m" + BG_BLUE = "\033[44m" + BG_MAGENTA = "\033[45m" + BG_CYAN = "\033[46m" + + +def signal_color(signal: SignalStrength) -> str: + """Get color for a signal strength.""" + 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(quality: str) -> str: + """Get color for signal quality grade.""" + return { + "A+": Colors.GREEN + Colors.BOLD, + "A": Colors.GREEN, + "B": Colors.CYAN, + "C": Colors.YELLOW, + "D": Colors.GRAY, + "F": Colors.DIM, + }.get(quality, Colors.RESET) + + +def score_bar(score: float, width: int = 30) -> str: + """Create a visual score bar from -100 to +100.""" + normalized = (score + 100) / 200 # 0 to 1 + filled = int(normalized * width) + center = width // 2 + + bar = "" + for i in range(width): + if i == center: + bar += "│" + elif (score >= 0 and center < i <= filled) or \ + (score < 0 and filled <= i < center): + bar += "█" + else: + bar += "░" + + if score >= 0: + color = Colors.GREEN if score > 20 else Colors.GRAY + else: + color = Colors.RED if score < -20 else Colors.GRAY + + return f"{color}{bar}{Colors.RESET}" + + +def print_header(): + """Print the platform header.""" + now = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC") + width = 80 + + print(f"\n{Colors.CYAN}{'═' * width}{Colors.RESET}") + print(f"{Colors.BOLD}{Colors.CYAN} ╔═══════════════════════════════════════════════════════╗{Colors.RESET}") + print(f"{Colors.BOLD}{Colors.CYAN} ║ FOREX SIGNAL INTELLIGENCE PLATFORM ║{Colors.RESET}") + print(f"{Colors.BOLD}{Colors.CYAN} ║ Multi-Layer Signal Analysis Engine ║{Colors.RESET}") + print(f"{Colors.BOLD}{Colors.CYAN} ╚═══════════════════════════════════════════════════════╝{Colors.RESET}") + print(f"{Colors.GRAY} Scan Time: {now}{Colors.RESET}") + print(f"{Colors.CYAN}{'═' * width}{Colors.RESET}\n") + + +def print_summary_table(reports: List[SignalReport]): + """Print a summary table of all pairs.""" + print(f"{Colors.BOLD}{Colors.WHITE} ┌─────────────┬───────────┬──────────────────────────────────┬────────┬────────────┐{Colors.RESET}") + print(f"{Colors.BOLD}{Colors.WHITE} │ Pair │ Signal │ Score │ Grade │ Confidence │{Colors.RESET}") + print(f"{Colors.BOLD}{Colors.WHITE} ├─────────────┼───────────┼──────────────────────────────────┼────────┼────────────┤{Colors.RESET}") + + for r in reports: + sig = r.final_signal.name.replace("_", " ") + sc = signal_color(r.final_signal) + qc = quality_color(r.signal_quality) + bar = score_bar(r.final_score, 30) + + pair = f"{r.display_name:<11}" + signal_str = f"{sc}{sig:<9}{Colors.RESET}" + grade_str = f"{qc}{r.signal_quality:<6}{Colors.RESET}" + conf_str = f"{r.final_confidence:5.1f}%" + + # Add warning indicator + warn = "" + if r.event_risk == "danger": + warn = f" {Colors.RED}⚠{Colors.RESET}" + elif r.event_risk == "caution": + warn = f" {Colors.YELLOW}△{Colors.RESET}" + + print(f" │ {pair} │ {signal_str} │ {bar} │ {grade_str} │ {conf_str:>8}{warn} │") + + print(f"{Colors.BOLD}{Colors.WHITE} └─────────────┴───────────┴──────────────────────────────────┴────────┴────────────┘{Colors.RESET}") + + # Legend + print(f"\n{Colors.GRAY} Legend: " + f"{Colors.GREEN}██ Bullish{Colors.GRAY} | " + f"{Colors.RED}██ Bearish{Colors.GRAY} | " + f"│ Center (0) | " + f"{Colors.RED}⚠ Event Risk{Colors.GRAY} | " + f"{Colors.YELLOW}△ Caution{Colors.RESET}") + + +def print_detailed_report(report: SignalReport): + """Print detailed analysis for a single pair.""" + sc = signal_color(report.final_signal) + qc = quality_color(report.signal_quality) + sig_name = report.final_signal.name.replace("_", " ") + + print(f"\n{Colors.CYAN}{'─' * 80}{Colors.RESET}") + print(f" {Colors.BOLD}{Colors.WHITE}{report.display_name}{Colors.RESET}" + f" │ Price: {Colors.BOLD}{report.current_price:.5f}{Colors.RESET}" + f" │ Regime: {Colors.CYAN}{report.market_regime}{Colors.RESET}" + f" │ Vol: {report.volatility_regime}") + + print(f"\n {Colors.BOLD}SIGNAL:{Colors.RESET} {sc}{sig_name}{Colors.RESET}" + f" │ Score: {sc}{report.final_score:+.1f}{Colors.RESET}" + f" │ Grade: {qc}{report.signal_quality}{Colors.RESET}" + f" │ Confidence: {report.final_confidence:.0f}%") + + print(f" {Colors.BOLD}Action:{Colors.RESET} {report.action_recommendation}") + + # Multi-timeframe breakdown + if report.mtf_signals: + print(f"\n {Colors.BOLD}Multi-Timeframe:{Colors.RESET}", end="") + mtf_agree = f" {Colors.GREEN}✓ ALIGNED{Colors.RESET}" if report.mtf_agreement else f" {Colors.YELLOW}✗ DIVERGENT{Colors.RESET}" + print(mtf_agree) + for tf, sig in report.mtf_signals.items(): + tf_sc = signal_color(sig) + tf_score = report.mtf_scores.get(tf, 0) + print(f" {tf:>5}: {tf_sc}{sig.name.replace('_', ' '):<12}{Colors.RESET} ({tf_score:+.1f})") + + # Indicators detail + if 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) + ind_name = f"{ind.name:<18}" + print(f" {ind_name} {isc}{ind.signal.name.replace('_', ' '):<12}{Colors.RESET}" + f" ({ind.confidence:.0f}%) │ {Colors.GRAY}{ind.description}{Colors.RESET}") + + # Bullish reasons + if report.bullish_reasons: + print(f"\n {Colors.GREEN}▲ Bullish Factors:{Colors.RESET}") + for reason in report.bullish_reasons[:5]: + print(f" {Colors.GREEN}+{Colors.RESET} {reason}") + + # Bearish reasons + if report.bearish_reasons: + print(f"\n {Colors.RED}▼ Bearish Factors:{Colors.RESET}") + for reason in report.bearish_reasons[:5]: + print(f" {Colors.RED}-{Colors.RESET} {reason}") + + # Warnings + if report.warnings: + print(f"\n {Colors.YELLOW}⚠ Warnings:{Colors.RESET}") + for warning in report.warnings: + print(f" {Colors.YELLOW}!{Colors.RESET} {warning}") + + # News context + if report.news_description and report.news_description != "Neutral news sentiment": + print(f"\n {Colors.MAGENTA}📰 News:{Colors.RESET} {report.news_description}") + if report.event_description and report.event_risk != "clear": + print(f" {Colors.YELLOW}📅 Event:{Colors.RESET} {report.event_description}") + + # Trade setup + if report.trade_setup: + ts = report.trade_setup + dir_color = Colors.GREEN if ts.direction == "BUY" else Colors.RED + print(f"\n {Colors.BOLD}Trade Setup:{Colors.RESET}") + print(f" Direction: {dir_color}{ts.direction}{Colors.RESET}") + print(f" Entry: {ts.entry_price:.5f}") + print(f" Stop Loss: {Colors.RED}{ts.stop_loss:.5f}{Colors.RESET} ({ts.risk_pips:.1f} pips)") + print(f" Take Profit 1: {Colors.GREEN}{ts.take_profit_1:.5f}{Colors.RESET} ({ts.reward_pips:.1f} pips)") + print(f" Take Profit 2: {Colors.GREEN}{ts.take_profit_2:.5f}{Colors.RESET}") + print(f" Risk:Reward: 1:{ts.risk_reward:.1f}") + + # Key levels + if report.primary_analysis and report.primary_analysis.key_levels: + levels = report.primary_analysis.key_levels + print(f"\n {Colors.BOLD}Key Levels:{Colors.RESET}") + level_names = ["r3", "r2", "r1", "pivot", "s1", "s2", "s3"] + for lvl in level_names: + if lvl in levels: + val = levels[lvl] + if val > report.current_price: + color = Colors.RED # resistance + else: + color = Colors.GREEN # support + label = lvl.upper() + print(f" {label:>6}: {color}{val:.5f}{Colors.RESET}") + + +def print_event_calendar(reports: List[SignalReport]): + """Print upcoming economic events affecting watched pairs.""" + events_shown = set() + has_events = False + + for report in reports: + if report.event_risk != "clear" and report.event_description not in events_shown: + if not has_events: + print(f"\n{Colors.YELLOW}{Colors.BOLD} 📅 ECONOMIC CALENDAR ALERTS{Colors.RESET}") + has_events = True + events_shown.add(report.event_description) + risk_color = Colors.RED if report.event_risk == "danger" else Colors.YELLOW + print(f" {risk_color}● {report.event_description}{Colors.RESET} → affects {report.display_name}") + + if not has_events: + print(f"\n{Colors.GRAY} 📅 No high-impact events in the near term{Colors.RESET}") + + +def report_to_dict(report: SignalReport) -> dict: + """Convert a SignalReport to a JSON-serializable 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"] = {} + for name, ind in report.primary_analysis.indicators.items(): + if not name.startswith("_"): + d["indicators"][name] = { + "signal": ind.signal.name, + "confidence": round(ind.confidence, 1), + "description": ind.description, + } + d["key_levels"] = { + k: round(v, 5) for k, v in report.primary_analysis.key_levels.items() + } + return d + + +# ══════════════════════════════════════════════════════════════════ +# MAIN +# ══════════════════════════════════════════════════════════════════ + +def parse_args(): + parser = argparse.ArgumentParser( + description="Forex Signal Intelligence Platform", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +Examples: + python main.py Analyze all major pairs (H1) + python main.py --pairs EURUSD GBPUSD Analyze specific pairs + python main.py --timeframe H4 Use 4-hour as primary TF + python main.py --all Include cross pairs + python main.py --json Output as JSON + python main.py --detail Show detailed analysis per pair + python main.py --watch Continuous monitoring (5min refresh) + """, + ) + parser.add_argument( + "--pairs", nargs="+", default=None, + help="Specific pairs to analyze (e.g., EURUSD GBPUSD)", + ) + parser.add_argument( + "--timeframe", "-tf", default="H1", + choices=["M5", "M15", "M30", "H1", "H4", "D1"], + help="Primary analysis timeframe (default: H1)", + ) + parser.add_argument( + "--all", action="store_true", + help="Include cross pairs in analysis", + ) + parser.add_argument( + "--json", action="store_true", + help="Output results as JSON", + ) + parser.add_argument( + "--detail", action="store_true", + help="Show detailed analysis for each pair", + ) + parser.add_argument( + "--top", type=int, default=None, + help="Show only top N signals by strength", + ) + parser.add_argument( + "--watch", action="store_true", + help="Continuous monitoring mode (refreshes every 5 min)", + ) + parser.add_argument( + "--min-grade", default="F", + choices=["A+", "A", "B", "C", "D", "F"], + help="Minimum signal quality to display", + ) + parser.add_argument( + "--verbose", "-v", action="store_true", + help="Enable debug logging", + ) + return parser.parse_args() + + +def symbol_from_name(name: str) -> str: + """Convert pair name like 'EURUSD' to Yahoo symbol 'EURUSD=X'.""" + name = name.upper().replace("/", "").replace("_", "") + if not name.endswith("=X"): + name += "=X" + return name + + +def run_analysis(args) -> List[SignalReport]: + """Run a single analysis cycle.""" + config = PlatformConfig() + + # Configure timeframe + 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] + + # Configure confirmation timeframes based on primary + 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] + ) + + # Configure pairs + if args.pairs: + config.watch_pairs = [symbol_from_name(p) for p in args.pairs] + elif args.all: + config.watch_pairs = MAJOR_PAIRS + CROSS_PAIRS + else: + config.watch_pairs = MAJOR_PAIRS + + # Run analysis + aggregator = SignalAggregator(config) + reports = aggregator.analyze_all_pairs() + + # Filter by grade + grade_order = ["A+", "A", "B", "C", "D", "F"] + min_idx = grade_order.index(args.min_grade) if args.min_grade in grade_order else len(grade_order) - 1 + reports = [r for r in reports if grade_order.index(r.signal_quality) <= min_idx] + + # Top N + if args.top: + reports = reports[:args.top] + + return reports + + +def main(): + args = parse_args() + + # Setup logging + 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", + ) + + if args.watch: + # Continuous monitoring mode + print(f"{Colors.BOLD}Starting continuous monitoring mode (Ctrl+C to stop)...{Colors.RESET}") + while True: + try: + reports = run_analysis(args) + if args.json: + output = [report_to_dict(r) for r in reports] + print(json.dumps(output, indent=2)) + else: + print("\033[2J\033[H") # clear screen + print_header() + print_summary_table(reports) + print_event_calendar(reports) + if args.detail: + for r in reports: + print_detailed_report(r) + print(f"\n{Colors.GRAY} Next refresh in 5 minutes...{Colors.RESET}") + + time.sleep(300) + except KeyboardInterrupt: + print(f"\n{Colors.YELLOW}Monitoring stopped.{Colors.RESET}") + break + else: + # Single run + print(f"{Colors.GRAY} Fetching market data and running analysis...{Colors.RESET}\n") + reports = run_analysis(args) + + if args.json: + output = [report_to_dict(r) for r in reports] + print(json.dumps(output, indent=2)) + else: + print_header() + if reports: + print_summary_table(reports) + print_event_calendar(reports) + + if args.detail: + for r in reports: + print_detailed_report(r) + else: + # Show detail for top signals only + actionable = [r for r in reports if r.signal_quality in ("A+", "A", "B")] + if actionable: + print(f"\n{Colors.BOLD}{Colors.WHITE} ═══ ACTIONABLE SIGNALS (Grade B or above) ═══{Colors.RESET}") + for r in actionable: + print_detailed_report(r) + else: + print(f"\n{Colors.GRAY} No high-grade signals detected. " + f"Use --detail for full breakdown or --min-grade C to see more.{Colors.RESET}") + else: + print(f"{Colors.YELLOW} No data available. Check your internet connection.{Colors.RESET}") + + print(f"\n{Colors.CYAN}{'═' * 80}{Colors.RESET}") + print(f"{Colors.GRAY} Disclaimer: This tool is for educational/research purposes only.{Colors.RESET}") + print(f"{Colors.GRAY} Past performance ≠ future results. Always use proper risk management.{Colors.RESET}") + print(f"{Colors.CYAN}{'═' * 80}{Colors.RESET}\n") + + +if __name__ == "__main__": + main() \ No newline at end of file