Add files via upload

This commit is contained in:
Richard
2026-05-16 10:12:02 +01:00
committed by GitHub
parent e43fea1fd9
commit 956e81bf54
+233
View File
@@ -10,6 +10,11 @@ Credentials are loaded exclusively from the .env file in the same directory.
Install: pip install MetaTrader5 pandas numpy colorama python-dotenv
Sound Alerts (Windows only):
- High-Hz triple beep for STRONG BUY signals (score >= 65)
- Low-Hz triple beep for STRONG SELL signals (score >= 65)
- Type 'm' + Enter at any time to mute/unmute sound alerts
Usage:
# Live scanner — all 5 timeframes (default)
python mt5_multitf_pattern_scanner_v6.py
@@ -35,6 +40,9 @@ Usage:
# Live scanner with custom account sizing
python mt5_multitf_pattern_scanner_v6.py --mode live --account-balance 25000 --risk-percent 0.5
# Test sound alerts (plays both BUY and SELL test beeps)
python mt5_multitf_pattern_scanner_v6.py --test-sound
"""
import MetaTrader5 as mt5
@@ -51,6 +59,20 @@ import re
import warnings
warnings.filterwarnings('ignore')
# ── Windows sound & keyboard support ────────────────────────────────
try:
import winsound
_WINSOUND = True
except ImportError:
_WINSOUND = False
import threading
# Global mute state for sound alerts (toggled by 'm' key)
# Starts MUTED — type 'm' + Enter to unmute and hear alerts
_sound_muted = True
_sound_listener_thread = None
# ── Load credentials from .env file (REQUIRED) ─────────────────────
try:
from dotenv import load_dotenv
@@ -217,6 +239,14 @@ CFG = {
# ── Position Sizing ────────────────────────────────────────────
'account_balance': 100000,
'risk_percent': 1.0, # % of account per trade
# ── Sound Alerts (Windows only) ─────────────────────────────────
'sound_enabled': True, # master switch for sound alerts
'sound_buy_hz': 1200, # Hz for strong buy triple beep
'sound_sell_hz': 400, # Hz for strong sell triple beep
'sound_beep_duration': 150, # ms per individual beep
'sound_beep_pause': 100, # ms pause between beeps
'sound_strong_threshold': 65.0, # signal score >= this = STRONG
}
# ── Derived Paths ───────────────────────────────────────────────────
@@ -244,6 +274,164 @@ PATTERN_PRIORITY = {
# UTILITY FUNCTIONS
# ============================================================
# ============================================================
# SOUND ALERT FUNCTIONS
# ============================================================
def play_signal_beep(direction, score, cfg=None):
"""Play a triple beep sound alert for strong signals.
High-Hz triple beep for STRONG BUY, Low-Hz triple beep for STRONG SELL.
Respects the global mute state and configuration thresholds.
Args:
direction: 'Bullish' or 'Bearish'
score: signal quality score (0-100)
cfg: configuration dict
"""
global _sound_muted
if cfg is None:
cfg = CFG
# Check master switch and mute state
if not cfg.get('sound_enabled', True) or _sound_muted:
return
if not _WINSOUND:
return
# Only beep for STRONG signals (score >= threshold)
threshold = cfg.get('sound_strong_threshold', 65.0)
if score is None or score < threshold:
return
# Determine frequency based on direction
if direction == 'Bullish':
freq = cfg.get('sound_buy_hz', 1200)
elif direction == 'Bearish':
freq = cfg.get('sound_sell_hz', 400)
else:
return # Neutral signals don't beep
duration = cfg.get('sound_beep_duration', 150)
pause = cfg.get('sound_beep_pause', 100)
# Play triple beep
try:
for i in range(3):
winsound.Beep(int(freq), int(duration))
if i < 2: # No pause after last beep
time.sleep(pause / 1000.0)
except Exception:
pass # Silently ignore sound errors
def _sound_key_listener():
"""Background daemon thread that listens for 'm' + Enter to toggle mute.
Uses input() which works in ALL terminals including PyCharm.
Runs as a daemon thread so it dies automatically when the main program exits.
"""
global _sound_muted
while True:
try:
cmd = input().strip().lower()
if cmd == 'm':
_sound_muted = not _sound_muted
status = C('red', 'MUTED') if _sound_muted else C('green', 'UNMUTED')
print(f"\n Sound {status} | Type 'm' + Enter to toggle")
except (EOFError, KeyboardInterrupt):
break
except Exception:
pass
def start_sound_key_listener():
"""Start the background keyboard listener thread (if not already running)."""
global _sound_listener_thread
if _sound_listener_thread is not None and _sound_listener_thread.is_alive():
return
_sound_listener_thread = threading.Thread(target=_sound_key_listener, daemon=True)
_sound_listener_thread.start()
def check_mute_key():
"""Legacy stub — keyboard is now handled by background thread.
Kept for API compatibility but does nothing. The background thread
started by start_sound_key_listener() handles all key detection.
"""
pass
def test_sound(cfg=None):
"""Play test beeps for STRONG BUY and STRONG SELL so the user can verify audio.
Temporarily forces sound enabled and unmuted for the test, then restores
the original state. Exits after playing both test beeps.
"""
global _sound_muted
if cfg is None:
cfg = CFG
if not _WINSOUND:
print(C('red', "ERROR: winsound not available — sound requires Windows"))
return
# Save original mute state, force unmute for the test
was_muted = _sound_muted
_sound_muted = False
buy_hz = cfg.get('sound_buy_hz', 1200)
sell_hz = cfg.get('sound_sell_hz', 400)
duration = cfg.get('sound_beep_duration', 150)
pause = cfg.get('sound_beep_pause', 100)
print("")
print(C('cyan', "=" * 50))
print(C('bold', " SOUND TEST"))
print(C('cyan', "=" * 50))
print(f" STRONG BUY beep: {buy_hz} Hz x 3")
print(f" STRONG SELL beep: {sell_hz} Hz x 3")
print(f" Beep duration: {duration} ms")
print(f" Pause between: {pause} ms")
print("")
# Test STRONG BUY
print(C('green', " >>> Playing STRONG BUY test beep..."))
try:
for i in range(3):
winsound.Beep(int(buy_hz), int(duration))
if i < 2:
time.sleep(pause / 1000.0)
except Exception as e:
print(C('red', f" ERROR: {e}"))
time.sleep(0.5)
# Test STRONG SELL
print(C('red', " >>> Playing STRONG SELL test beep..."))
try:
for i in range(3):
winsound.Beep(int(sell_hz), int(duration))
if i < 2:
time.sleep(pause / 1000.0)
except Exception as e:
print(C('red', f" ERROR: {e}"))
# Restore original mute state
_sound_muted = was_muted
print("")
print(C('cyan', "=" * 50))
if was_muted:
print(f" Sound is {C('red', 'MUTED')} (restored to original state)")
print(f" Type {C('yellow', '\"m\" + Enter')} to unmute during live scanner")
else:
print(f" Sound is {C('green', 'UNMUTED')} (restored to original state)")
print(C('cyan', "=" * 50))
print("")
def broker_now():
"""Return current broker server time. MT5 timestamps interpreted as UTC = broker time."""
return datetime.utcnow()
@@ -2341,6 +2529,20 @@ def run_scanner(cfg=None):
if cfg.get('volume_filter', False):
log_message(f"Volume Filter: {C('green', 'ENABLED')} ({cfg['volume_threshold']}x avg)", cfg)
# Sound alert status
if cfg.get('sound_enabled', True) and _WINSOUND:
buy_hz = cfg.get('sound_buy_hz', 1200)
sell_hz = cfg.get('sound_sell_hz', 400)
threshold = cfg.get('sound_strong_threshold', 65.0)
log_message(f"Sound Alerts: {C('green', 'ENABLED')} | Buy: {buy_hz}Hz | Sell: {sell_hz}Hz | Threshold: {threshold:.0f}", cfg)
log_message(f"Sound is {C('red', 'MUTED')} — Type {C('yellow', '\"m\" + Enter')} to unmute", cfg)
elif not _WINSOUND:
log_message(C('yellow', "Sound Alerts: DISABLED (winsound not available — Windows only)"), cfg)
# Start background keyboard listener for mute toggle
if cfg.get('sound_enabled', True) and _WINSOUND:
start_sound_key_listener()
# Load backtest stats for historical edge display
stats = load_latest_backtest_stats(cfg=cfg)
stats_last_refresh = datetime.now()
@@ -2410,6 +2612,13 @@ def run_scanner(cfg=None):
pats = scan_patterns(list(rates[:-1]), cfg, d1_rates_cache, tf_label, htf_atr_rates)
pats = apply_signal_score_filter(pats, stats, cfg)
log_message(format_pattern_output(rates[-2], pats, cfg, stats, tf_label), cfg)
# Play sound alert for strong signals on startup scan
for pat in pats:
if pat.get('direction') in ('Bullish', 'Bearish'):
score = pat.get('signal_score')
if score is None:
score = compute_signal_score(pat['name'], pat['session'], pat['direction'], stats, cfg)
play_signal_beep(pat['direction'], score, cfg)
continue
if bar_time != last_candle_time[tf_label]:
@@ -2424,8 +2633,17 @@ def run_scanner(cfg=None):
output = format_pattern_output(rates[-2], pats, cfg, stats, tf_label)
log_message(output, cfg)
# Play sound alert for strong directional signals
for pat in pats:
if pat.get('direction') in ('Bullish', 'Bearish'):
score = pat.get('signal_score')
if score is None:
score = compute_signal_score(pat['name'], pat['session'], pat['direction'], stats, cfg)
play_signal_beep(pat['direction'], score, cfg)
last_candle_time[tf_label] = bar_time
# Keyboard listener runs in background thread — no manual check needed
time.sleep(min(cfg.get('poll_interval_by_tf', {}).get(tf, 30) for tf in active_tfs))
except Exception as e:
@@ -2497,6 +2715,14 @@ def run_single_scan(cfg=None):
pats = apply_signal_score_filter(pats, stats, cfg)
log_message(format_pattern_output(closed, pats, cfg, stats, tf_label), cfg)
# Play sound alert for strong directional signals
for pat in pats:
if pat.get('direction') in ('Bullish', 'Bearish'):
score = pat.get('signal_score')
if score is None:
score = compute_signal_score(pat['name'], pat['session'], pat['direction'], stats, cfg)
play_signal_beep(pat['direction'], score, cfg)
try: mt5.shutdown()
except Exception: pass
log_message("Single scan complete.", cfg)
@@ -2895,6 +3121,8 @@ Examples:
# Misc
p.add_argument("--warmup-bars", type=int, default=cfg['warmup_bars'])
p.add_argument("--bars-to-fetch", type=int, default=cfg['bars_to_fetch'])
p.add_argument("--test-sound", action="store_true",
help="Play test beeps (STRONG BUY then STRONG SELL) and exit")
p.set_defaults(
deduplicate_signals=cfg['deduplicate_signals'],
@@ -2951,6 +3179,11 @@ def main():
# Override the per-TF mapping to use the specified TF for all
runtime_cfg['atr_tf_by_tf'] = {tf: override_tf for tf in TIMEFRAME_MAP.keys()}
# Handle --test-sound: play both test beeps and exit
if args.test_sound:
test_sound(runtime_cfg)
return
if args.mode == 'live':
run_scanner(runtime_cfg)
elif args.mode == 'scan':