509 lines
21 KiB
Python
509 lines
21 KiB
Python
#!/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
|
|
|
|
# Force UTF-8 output on Windows so box-drawing / Unicode chars render correctly
|
|
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,
|
|
MAJOR_PAIRS, CROSS_PAIRS,
|
|
)
|
|
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() |