gavindiaz/mt5-candlestick-scanner-backtester
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add files via upload

Browse Source 
Next close time is calculated from wrong candle time reference
This commit is contained in:
Richard
2026-05-20 20:41:47 +01:00
committed by GitHub
parent a7f03c6620
commit 83ae839186
1 changed files with 162 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files
mt5_multitf_pattern_scanner.py
+162 -38
View File
@@ -11,8 +11,9 @@ 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)
- Tier-aware: Tier A (Elite) always alerts, Tier B (Tradeable) alerts if score >= 60, Tier C/D never alerts
- High-Hz triple beep for BUY signals, Low-Hz triple beep for SELL signals
- Configurable via sound_alert_tier ('A', 'B', 'C') and sound_alert_tier_b_min_score
- Type 'm' + Enter at any time to mute/unmute sound alerts
Usage:
@@ -69,7 +70,7 @@ Usage:
import MetaTrader5 as mt5
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from datetime import datetime, timedelta, timezone
import argparse
import time
import os
@@ -234,7 +235,8 @@ CFG = {
},
# ── Session Classifier ─────────────────────────────────────────
'broker_utc_offset': 2, # UTC+2 broker server time
'broker_utc_offset': 2, # UTC+2 broker server time (auto-detected at startup; set 0 to disable auto-detect)
# Typical: UTC+2 winter / UTC+3 summer (follows US DST for NY-close brokers)
# ── Signal Deduplication & Entry Verification ─────────────────
'deduplicate_signals': True,
@@ -273,7 +275,8 @@ CFG = {
'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
'sound_alert_tier': 'B', # Minimum tier to trigger sound: 'A' (elite only), 'B' (tradeable+), 'C' (all directional)
'sound_alert_tier_b_min_score': 60.0, # Tier B patterns must also have score >= this to alert
# ── Trade Management (Backtest) ────────────────────────────────
'trade_management_mode': 'fixed', # 'fixed', 'breakeven', 'trail', 'partial'
@@ -328,15 +331,21 @@ PATTERN_PRIORITY = {
# SOUND ALERT FUNCTIONS
# ============================================================
def play_signal_beep(direction, score, cfg=None):
"""Play a triple beep sound alert for strong signals.
def play_signal_beep(direction, score, tier='D', cfg=None):
"""Play a triple beep sound alert for tier-qualified signals.
High-Hz triple beep for STRONG BUY, Low-Hz triple beep for STRONG SELL.
Respects the global mute state and configuration thresholds.
Tier-aware alert logic (replaces flat score threshold):
- Tier A (ELITE): Always alerts — highest confidence signals
- Tier B (TRADEABLE): Alerts if score >= sound_alert_tier_b_min_score (default 60)
- Tier C (MARGINAL): Never alerts — use only with strong confluence
- Tier D (AVOID): Never alerts — negative edge
High-Hz triple beep for BUY signals, Low-Hz triple beep for SELL signals.
Args:
direction: 'Bullish' or 'Bearish'
score: signal quality score (0-100)
tier: pattern tier letter 'A', 'B', 'C', or 'D'
cfg: configuration dict
"""
global _sound_muted
@@ -349,9 +358,21 @@ def play_signal_beep(direction, score, cfg=None):
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:
# Tier-based alert gate
min_tier = cfg.get('sound_alert_tier', 'B')
tier_b_min = cfg.get('sound_alert_tier_b_min_score', 60.0)
# Tier rank: A=1, B=2, C=3, D=4 — only alert if pattern's tier is at least as good as min_tier
tier_rank = {'A': 1, 'B': 2, 'C': 3, 'D': 4}
if tier_rank.get(tier, 4) > tier_rank.get(min_tier, 2):
return # Pattern tier is worse than minimum allowed
# Tier B requires minimum score (good confluence/session needed)
if tier == 'B' and (score is None or score < tier_b_min):
return
# Tier A always passes; Tier C/D already blocked above
if score is None:
return
# Determine frequency based on direction
@@ -408,7 +429,7 @@ def start_sound_key_listener():
def test_sound(cfg=None):
"""Play test beeps for STRONG BUY and STRONG SELL so the user can verify audio.
"""Play test beeps for Tier A BUY and SELL signals 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.
@@ -429,19 +450,21 @@ def test_sound(cfg=None):
sell_hz = cfg.get('sound_sell_hz', 400)
duration = cfg.get('sound_beep_duration', 150)
pause = cfg.get('sound_beep_pause', 100)
alert_tier = cfg.get('sound_alert_tier', 'B')
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" Alert Tier: {alert_tier} (Tier A always, Tier B if score >= {cfg.get('sound_alert_tier_b_min_score', 60):.0f})")
print(f" BUY beep: {buy_hz} Hz x 3")
print(f" 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..."))
# Test Tier A BUY
print(C('green', " >>> Playing Tier A BUY test beep..."))
try:
for i in range(3):
winsound.Beep(int(buy_hz), int(duration))
@@ -452,8 +475,8 @@ def test_sound(cfg=None):
time.sleep(0.5)
# Test STRONG SELL
print(C('red', " >>> Playing STRONG SELL test beep..."))
# Test Tier A SELL
print(C('red', " >>> Playing Tier A SELL test beep..."))
try:
for i in range(3):
winsound.Beep(int(sell_hz), int(duration))
@@ -477,17 +500,69 @@ def test_sound(cfg=None):
def broker_now():
"""Return current broker server time. MT5 timestamps interpreted as UTC = broker time."""
return datetime.utcnow()
"""Return current local time for log timestamps.
Uses datetime.now() so log timestamps match the user's wall clock.
Candle display times are converted to local time separately via
to_local_time().
"""
return datetime.now()
def broker_time(ts):
"""Convert a Unix timestamp (from MT5) to broker server time.
"""Convert MT5 Unix timestamp to broker server clock time.
MT5 brokers encode their server time directly in the timestamps,
so fromtimestamp with UTC returns the broker's clock time.
MT5 encodes broker server time directly in Unix timestamps as if it were
UTC. For example, if the broker is at UTC+2 and it's 12:00 broker time,
the Unix timestamp represents 12:00 UTC (not the true 10:00 UTC).
Decoding as UTC therefore returns the broker's clock time directly.
"""
return datetime.fromtimestamp(int(ts), tz=None)
return datetime.fromtimestamp(int(ts), tz=timezone.utc).replace(tzinfo=None)
def to_local_time(broker_dt, cfg=None):
"""Convert a broker-time datetime to local machine time for display.
Formula: local_time = broker_time - broker_utc_offset + local_utc_offset
The broker UTC offset comes from config (auto-detected at startup).
The local UTC offset is computed from the system clock (handles DST
automatically).
"""
if cfg is None:
cfg = CFG
broker_offset = cfg.get('broker_utc_offset', 2)
local_offset_td = datetime.now() - datetime.utcnow()
return broker_dt - timedelta(hours=broker_offset) + local_offset_td
def auto_detect_broker_offset(cfg=None):
"""Auto-detect broker UTC offset by comparing the latest MT5 candle time
with the current UTC time.
MT5 timestamps encode broker time as UTC, so the difference between
the decoded broker time and true UTC gives the broker's offset.
Returns the detected offset (integer hours), or the configured default
if detection fails.
"""
if cfg is None:
cfg = CFG
configured = cfg.get('broker_utc_offset', 2)
try:
symbol = cfg.get('symbol', 'EURUSD')
rates = mt5.copy_rates_from_pos(symbol, mt5.TIMEFRAME_M1, 0, 1)
if rates is not None and len(rates) > 0:
broker_clock = datetime.fromtimestamp(
int(rates[-1]['time']), tz=timezone.utc
).replace(tzinfo=None)
utc_now = datetime.utcnow()
diff_hours = (broker_clock - utc_now).total_seconds() / 3600
detected = round(diff_hours)
if abs(detected - diff_hours) < 0.5:
return detected
except Exception:
pass
return configured
_log_lock = threading.Lock()
@@ -1872,7 +1947,16 @@ def format_pattern_output(candle, patterns, cfg=None, stats=None, tf_label='H4')
ct = candle['time']
if isinstance(ct, (int, float, np.integer, np.floating)):
ct = broker_time(int(ct))
time_str = ct.strftime("%Y-%m-%d %H:%M:%S") if hasattr(ct, 'strftime') else str(ct)
# MT5's 'time' field is the candle's OPEN time in broker time.
# 1. Add candle duration to get the CLOSE time
# 2. Convert broker time to local time for display
tf_minutes = TIMEFRAME_MAP.get(tf_label, {}).get('minutes', 0)
if tf_minutes and hasattr(ct, '__add__'):
close_ct_broker = ct + timedelta(minutes=tf_minutes)
else:
close_ct_broker = ct
close_ct = to_local_time(close_ct_broker, cfg)
time_str = close_ct.strftime("%Y-%m-%d %H:%M:%S") if hasattr(close_ct, 'strftime') else str(close_ct)
bt = max(candle['open'], candle['close'])
bb = min(candle['open'], candle['close'])
lines = []
@@ -2012,13 +2096,18 @@ def format_pattern_output(candle, patterns, cfg=None, stats=None, tf_label='H4')
if score is None:
score = compute_signal_score(pat['name'], pat['session'], pat['direction'], stats, cfg, tf_label=tf_label)
if score is not None:
if score >= 65:
if tier_letter == 'A':
score_label = "ELITE"; score_color = 'green'
elif tier_letter == 'B':
score_label = "TRADEABLE"; score_color = 'yellow'
elif score >= 65:
score_label = "STRONG"; score_color = 'green'
elif score >= 52:
score_label = "MODERATE"; score_color = 'yellow'
else:
score_label = "WEAK"; score_color = 'red'
lines.append(f" Signal Score: {C(score_color, C('bold', f'{score:.1f}/100 [{score_label}]'))}")
alert_tag = " [ALERT]" if (tier_letter in ('A', 'B')) else ""
lines.append(f" Signal Score: {C(score_color, C('bold', f'{score:.1f}/100 [{score_label}{alert_tag}]'))}")
elif not has_edge:
lines.append(C('dim', f" HISTORICAL EDGE: Insufficient data (<{min_sig} signals)"))
@@ -3423,6 +3512,7 @@ def run_scanner(cfg=None):
if len(cfg.get('watchlist', [])) > 1:
log_message(f" Watchlist: {watchlist_display} (live scanner: {symbol})", cfg)
log_message(f"Active timeframes: {C('yellow', ', '.join(active_tfs))}", cfg)
log_message(f"Timestamps: Local time ({datetime.now().strftime('%Z')}, auto-detected broker UTC+{cfg.get('broker_utc_offset', 2)})", cfg)
sl_str = f"{cfg['sl_multiplier']}x ATR"
log_message(f"SL: {C('red', sl_str)} | TP R:R = 1:{cfg['tp_multiplier']/cfg['sl_multiplier']:.1f}", cfg)
@@ -3442,8 +3532,13 @@ def run_scanner(cfg=None):
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)
alert_tier = cfg.get('sound_alert_tier', 'B')
tier_b_min = cfg.get('sound_alert_tier_b_min_score', 60.0)
tier_labels = {'A': 'Elite only', 'B': 'Tradeable+', 'C': 'All directional'}
tier_desc = tier_labels.get(alert_tier, alert_tier)
log_message(f"Sound Alerts: {C('green', 'ENABLED')} | Buy: {buy_hz}Hz | Sell: {sell_hz}Hz | Alert Tier: {alert_tier} ({tier_desc})", cfg)
if alert_tier == 'B':
log_message(f" Tier B requires score >= {tier_b_min:.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)
@@ -3472,6 +3567,16 @@ def run_scanner(cfg=None):
if not connect_mt5(cfg):
return
# Auto-detect broker UTC offset (handles DST changes automatically)
configured_offset = cfg.get('broker_utc_offset', 2)
detected_offset = auto_detect_broker_offset(cfg)
if detected_offset != configured_offset:
log_message(
C('yellow', f"Broker UTC offset: auto-detected {detected_offset} (config says {configured_offset}, using detected)"), cfg)
cfg['broker_utc_offset'] = detected_offset
else:
log_message(f"Broker UTC offset: {detected_offset} (confirmed)", cfg)
# Track last candle time per timeframe
last_candle_time = {tf: None for tf in active_tfs}
d1_rates_cache = None
@@ -3530,14 +3635,18 @@ def run_scanner(cfg=None):
score = pat.get('signal_score')
if score is None:
score = compute_signal_score(pat['name'], pat['session'], pat['direction'], stats, cfg, tf_label=tf_label)
play_signal_beep(pat['direction'], score, cfg)
tier_letter, _, _ = compute_pattern_tier(pat['name'], stats, cfg)
play_signal_beep(pat['direction'], score, tier_letter, cfg)
continue
if bar_time != last_candle_time[tf_label]:
next_close = bar_time + timedelta(minutes=tf_info['minutes'])
# bar_time is the NEW (forming) candle's open time (broker time).
# Its close time = bar_time + tf_minutes, converted to local time.
next_close_local = to_local_time(
bar_time + timedelta(minutes=tf_info['minutes']), cfg)
log_message(
C('bold', C('yellow',
f"\nNEW {tf_label} CANDLE CLOSED! | Next: {next_close.strftime('%Y-%m-%d %H:%M')}"
f"\nNEW {tf_label} CANDLE CLOSED! | Next {tf_label} close: {next_close_local.strftime('%Y-%m-%d %H:%M')}"
)), cfg
)
pats = scan_patterns(list(rates[:-1]), cfg, d1_rates_cache, tf_label, htf_atr_rates)
@@ -3551,7 +3660,8 @@ def run_scanner(cfg=None):
score = pat.get('signal_score')
if score is None:
score = compute_signal_score(pat['name'], pat['session'], pat['direction'], stats, cfg, tf_label=tf_label)
play_signal_beep(pat['direction'], score, cfg)
tier_letter, _, _ = compute_pattern_tier(pat['name'], stats, cfg)
play_signal_beep(pat['direction'], score, tier_letter, cfg)
last_candle_time[tf_label] = bar_time
@@ -3633,7 +3743,8 @@ def run_single_scan(cfg=None):
score = pat.get('signal_score')
if score is None:
score = compute_signal_score(pat['name'], pat['session'], pat['direction'], stats, cfg, tf_label=tf_label)
play_signal_beep(pat['direction'], score, cfg)
tier_letter, _, _ = compute_pattern_tier(pat['name'], stats, cfg)
play_signal_beep(pat['direction'], score, tier_letter, cfg)
try: mt5.shutdown()
except Exception: pass
@@ -3743,7 +3854,9 @@ def run_full_backtest(args, cfg=None):
atr_src = get_atr_tf(tf, cfg)
atr_map_display.append(f"{tf}→{atr_src}" if atr_src != tf else tf)
print(f" ATR Source : {', '.join(atr_map_display)}")
print(f" SL/TP : {cfg.get('sl_multiplier', 1.5)}x / {cfg.get('tp_multiplier', 1.5)}x ATR")
sl_m = cfg.get('sl_multiplier', 1.5)
tp_m = cfg.get('tp_multiplier', 1.5)
print(f" SL/TP : {sl_m}x / {tp_m}x ATR | R:R 1:{tp_m/sl_m:.1f}")
print(f" SL Mode : {cfg.get('sl_mode', 'atr')}")
print(f" Trade Mgmt : {cfg.get('trade_management_mode', 'fixed')}")
print(f" Output : {out_dir}")
@@ -4070,7 +4183,13 @@ Examples:
help="Override ATR source timeframe for ALL timeframes (e.g. H1, H4). "
"By default, atr_tf_by_tf config is used (M5→H1, M15→H1, etc.)")
p.add_argument("--sl", type=float, default=cfg['sl_multiplier'])
p.add_argument("--tp", type=float, default=cfg['tp_multiplier'])
p.add_argument("--tp", type=float, default=cfg['tp_multiplier'],
help="TP distance as ATR multiplier (e.g. 3.0). "
"Alternatively use --tp-rr for R:R-based setting")
p.add_argument("--tp-rr", type=float, default=None,
help="TP R:R ratio relative to SL (default: derived from --tp/--sl). "
"E.g. --tp-rr 2.0 sets TP at 2x the SL distance (1:2 R:R). "
"Overrides --tp if both are given")
p.add_argument("--forward",type=int, default=cfg['default_forward_candles'],
help="Default forward candles for evaluation (H4 default)")
p.add_argument("--output", default=DEFAULT_OUTPUT_DIR)
@@ -4126,7 +4245,7 @@ Examples:
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")
help="Play test beeps (Tier A BUY then SELL) and exit")
p.set_defaults(
deduplicate_signals=cfg['deduplicate_signals'],
@@ -4191,6 +4310,11 @@ def main():
if args.timeout_mode:
runtime_cfg['timeout_mode'] = args.timeout_mode
# Handle --tp-rr: override tp_multiplier from R:R ratio
# tp_multiplier = sl_multiplier × tp_rr (e.g. sl=1.5, tp-rr=2.0 → tp=3.0 → R:R 1:2)
if args.tp_rr is not None:
runtime_cfg['tp_multiplier'] = runtime_cfg['sl_multiplier'] * args.tp_rr
# Handle --test-sound: play both test beeps and exit
if args.test_sound:
test_sound(runtime_cfg)