Harwinder Sekhon
11ff49befb
- Delete test_ws_dual.py (standalone probe, unused) and a stale tracked log - config.py: drop MarketSession enum, is_jpy_pair, candle_buffer_seconds, and the never-read legacy SignalConfig/XauStrategyConfig fields - data_provider.py: drop uncalled get_latest_price/get_spread_estimate - Remove unused imports (engine, telegram_bot, main, news_engine) - test_all.py: drop/trim the corresponding defaults assertions (199 tests) - CLAUDE.md updated to match Zero behavior change: no production code read any removed field (verified reference-by-reference); suite green at 199/199. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01VPNF37MP9Kuv62B1VBrZ1b
347 lines
14 KiB
Python
Executable File
347 lines
14 KiB
Python
Executable File
#!/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
|
|
|
|
# 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 pair_signals import PAIR_PRODUCERS
|
|
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 "<Indicator>: " 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 _fmt_price(price: float) -> str:
|
|
"""2dp for gold-scale prices, 5dp for sub-10 FX-scale prices."""
|
|
return f"{price:.2f}" if price >= 10 else f"{price:.5f}"
|
|
|
|
|
|
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}{_fmt_price(ts.entry_price)}{Colors.RESET} "
|
|
f"→ SL {Colors.RED}{_fmt_price(ts.stop_loss)}{Colors.RESET} "
|
|
f"({ts.risk_pips:.1f}p) "
|
|
f"→ TP {Colors.GREEN}{_fmt_price(ts.take_profit_1)}{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}{_fmt_price(report.current_price)}{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),
|
|
"news_description": report.news_description,
|
|
"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("--pair", "-p", default="XAUUSD",
|
|
help="Pair to analyze: XAUUSD (default: XAUUSD)")
|
|
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 _resolve_pair(s: str) -> str:
|
|
aliases = {
|
|
"XAUUSD": XAUUSD, "XAU/USD": XAUUSD, "GOLD": XAUUSD,
|
|
}
|
|
key = s.upper().strip()
|
|
if key in aliases:
|
|
return aliases[key]
|
|
raise SystemExit(f"Unknown pair '{s}'. Try one of: {', '.join(sorted(aliases))}")
|
|
|
|
|
|
def build_config(args) -> PlatformConfig:
|
|
pair = _resolve_pair(args.pair)
|
|
|
|
tf_map = {
|
|
"M5": Timeframe.M5, "M15": Timeframe.M15, "M30": Timeframe.M30,
|
|
"H1": Timeframe.H1, "H4": Timeframe.H4, "D1": Timeframe.D1,
|
|
}
|
|
primary_tf = 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],
|
|
}
|
|
confirmation_tfs = tf_confirmations.get(
|
|
args.timeframe, [Timeframe.M15, Timeframe.H4]
|
|
)
|
|
|
|
# Per-pair tuned configs from the pair_signals factories.
|
|
factory = PAIR_PRODUCERS.get(pair)
|
|
if factory is not None:
|
|
return factory(primary_tf, confirmation_tfs)
|
|
|
|
config = PlatformConfig()
|
|
config.primary_timeframe = primary_tf
|
|
config.confirmation_timeframes = confirmation_tfs
|
|
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(symbol=config.symbol)
|
|
|
|
if args.json:
|
|
print(json.dumps(report_to_dict(report), indent=2))
|
|
return
|
|
|
|
print_header(f"{config.display_name} 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()
|