Tune 5M signals, fix XAU regression, add signal dedup cooldown

EUR/USD 5M: raise min_entry_score 28→35, block hours 4/19/21
  → 83.4% WR, 145 trades, +9.68R, PF 1.42 (120-day)

XAU/USD 5M: fix B-grade exclusion regression in setup_grades,
  add breakeven trail (trigger 0.15R, buffer 0.05R)
  → 84.9% WR, 4810 trades, +164.54R, PF 1.23 (730-day)

EUR/USD M15: tighten defaults (SL 1.5x ATR, TP 1.0x, min_rr 0.5,
  block ranging+transitioning, BE trigger 0.2R)

Backtest: multiprocess fork parallelism, vectorized pre-filter,
  CUDA preload fix, per-pair BE trigger routing

Telegram bot: add signal deduplication — suppress repeated
  same-direction messages within 5-bar cooldown per pair+TF.
  Resend on direction flip, quality upgrade, or cooldown expiry.
  Manual scans always ungated.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
harwinder
2026-05-19 20:19:11 -04:00
parent 9e5410d617
commit b7ec5341cf
7 changed files with 280 additions and 151 deletions
+206 -129
View File
@@ -31,11 +31,13 @@ from __future__ import annotations
import argparse
import logging
import multiprocessing as mp
import os
import sys
import time
from dataclasses import dataclass, field
from datetime import datetime, timedelta, timezone
from itertools import chain
from pathlib import Path
from typing import Dict, List, Optional, Tuple
@@ -65,26 +67,42 @@ from pair_signals import PAIR_PRODUCERS # noqa: E402
# An importable CuPy can still fail at first kernel launch if CUDA libs
# are missing (containers, headless boxes), so verify with a tiny op.
# ─────────────────────────────────────────────────────────────────────
def _auto_detect_cuda_path():
"""Set CUDA_PATH from pip-installed nvidia packages if not already set."""
if os.environ.get("CUDA_PATH"):
return
for p in sys.path:
nvrtc_lib = os.path.join(p, "nvidia", "cuda_nvrtc", "lib")
if os.path.isdir(nvrtc_lib):
os.environ["CUDA_PATH"] = os.path.join(p, "nvidia")
ld = os.environ.get("LD_LIBRARY_PATH", "")
if nvrtc_lib not in ld:
os.environ["LD_LIBRARY_PATH"] = nvrtc_lib + (":" + ld if ld else "")
return
def _preload_nvidia_libs():
"""Pre-load versioned CUDA shared libraries so CuPy can find them.
_auto_detect_cuda_path()
pip-installed nvidia packages ship libnvrtc.so.12 (versioned) but CuPy's
dlopen looks for the unversioned libnvrtc.so. Loading the versioned .so
via ctypes with RTLD_GLOBAL makes the symbols available process-wide.
"""
import ctypes
search_paths = list(sys.path)
for venv_dir in (PROJECT_ROOT / "venv", PROJECT_ROOT / ".venv"):
if venv_dir.is_dir():
for sp in venv_dir.glob("lib/python*/site-packages"):
search_paths.append(str(sp))
for p in search_paths:
nvidia_base = os.path.join(p, "nvidia")
if not os.path.isdir(nvidia_base):
continue
for pkg in sorted(os.listdir(nvidia_base)):
lib_dir = os.path.join(nvidia_base, pkg, "lib")
if not os.path.isdir(lib_dir):
continue
for f in sorted(os.listdir(lib_dir)):
if f.endswith(".so") or (".so." in f and not f.endswith(".o")):
try:
ctypes.CDLL(os.path.join(lib_dir, f), mode=ctypes.RTLD_GLOBAL)
except OSError:
pass
return
_preload_nvidia_libs()
def _detect_gpu():
try:
import cupy as cp_mod # type: ignore
# Force runtime init: a kernel launch will surface missing libs.
_ = cp_mod.asarray(np.zeros(2, dtype=np.float64))
_ = cp_mod.cumsum(_).get()
return cp_mod, True
@@ -1707,127 +1725,81 @@ def _is_pair_blocked_by_calendar(ts: pd.Timestamp, cfg: PlatformConfig) -> Optio
return None
def run_backtest(
bt_cfg: BacktestConfig,
platform_cfg: Optional[PlatformConfig] = None,
) -> Tuple[List[Trade], Dict[Timeframe, pd.DataFrame]]:
"""Replay history and collect trades."""
if platform_cfg is None:
platform_cfg = PlatformConfig(
symbol=bt_cfg.pair,
display_name=PAIR_INFO[bt_cfg.pair]["display"],
pip_value=PAIR_INFO[bt_cfg.pair]["pip_value"],
primary_timeframe=bt_cfg.primary_tf,
confirmation_timeframes=list(bt_cfg.confirmation_tfs),
)
def _prefilter_bars(primary_df: pd.DataFrame, bt_cfg, platform_cfg,
first_pos: int, last_pos: int):
"""Vectorized calendar/weekend pre-filter. Returns (eligible, skipped_counts)."""
idx = primary_df.index[first_pos:last_pos]
wd = np.array(idx.weekday)
hour = np.array(idx.hour)
provider = ForexDataProvider(platform_cfg)
weekend = ((wd == 4) & (hour >= 22)) | (wd == 5) | ((wd == 6) & (hour < 22))
# ── Load all required timeframes ──
timeframes = [bt_cfg.primary_tf] + list(bt_cfg.confirmation_tfs)
# Pad lookback so earliest TF has enough warmup for EMA200, ADX, etc.
pad_days = max(60, bt_cfg.warmup_bars // 4)
fetch_start = bt_cfg.start - timedelta(days=pad_days)
dfs_full: Dict[Timeframe, pd.DataFrame] = {}
for tf in timeframes:
df = fetch_history(
provider, bt_cfg.pair, tf,
start=fetch_start, end=bt_cfg.end,
use_cache=bt_cfg.use_cache,
)
if df.empty:
raise RuntimeError(f"No data for {bt_cfg.pair} {tf.value}")
dfs_full[tf] = df
# ── Pre-compute indicators on each TF ──
t0 = time.perf_counter()
caches: Dict[str, IndicatorCache] = {
tf.value: precompute_indicators(dfs_full[tf], platform_cfg.technical, gpu_log=True)
for tf in timeframes
}
logger.info(f"Indicator pre-compute took {time.perf_counter() - t0:.2f}s")
# ── Inject GPU engine into a fresh aggregator ──
aggregator = SignalAggregator(platform_cfg)
aggregator.tech_engine = GPUTechnicalEngine(platform_cfg.technical, caches)
# ── Walk primary TF bar-by-bar ──
primary_df = dfs_full[bt_cfg.primary_tf]
# bt_cfg.start/end are already tz-aware UTC datetimes; specifying tz=
# again would raise. Just wrap.
bt_start_ts = pd.Timestamp(bt_cfg.start)
bt_end_ts = pd.Timestamp(bt_cfg.end)
# Start at warmup_bars OR first bar inside the bt window (whichever later).
first_eligible_pos = max(
bt_cfg.warmup_bars,
int(primary_df.index.searchsorted(bt_start_ts)),
is_xau = PAIR_INFO[bt_cfg.pair]["display"] == "XAU/USD"
allowed = (
platform_cfg.xau_strategy.allowed_weekdays if is_xau
else platform_cfg.backtest.allowed_weekdays
)
last_pos = int(primary_df.index.searchsorted(bt_end_ts, side="right"))
last_pos = min(last_pos, len(primary_df))
if allowed:
day_ok = np.isin(wd, allowed)
else:
day_ok = np.ones(len(idx), dtype=bool)
trades: List[Trade] = []
last_entry_pos = -10_000
skipped_reasons: Dict[str, int] = {}
eligible_mask = ~weekend & day_ok
eligible = np.where(eligible_mask)[0] + first_pos
logger.info(
f"Replaying {bt_cfg.pair} {bt_cfg.primary_tf.value}: "
f"bars {first_eligible_pos}–{last_pos} of {len(primary_df)}"
)
skipped: Dict[str, int] = {}
n_weekend = int(weekend.sum())
n_day = int((~weekend & ~day_ok).sum())
if n_weekend:
skipped["WEEKEND_FILTER"] = n_weekend
if n_day:
skipped["DAY_FILTER"] = n_day
primary_dur = pd.Timedelta(seconds=BAR_SECONDS[bt_cfg.primary_tf])
return eligible.tolist(), skipped
t_loop = time.perf_counter()
for pos in range(first_eligible_pos, last_pos):
ts = primary_df.index[pos]
# Decision time = close of the current primary bar.
decision_time = ts + primary_dur
# Calendar filter (matches live analyze_pair behavior).
if PAIR_INFO[bt_cfg.pair]["display"] == "XAU/USD":
block = _is_xau_blocked_by_calendar(ts, platform_cfg)
else:
block = _is_pair_blocked_by_calendar(ts, platform_cfg)
if block:
skipped_reasons[block] = skipped_reasons.get(block, 0) + 1
continue
# ── Parallel replay worker state (set before fork) ──────────────────
_mp_dfs_full: Optional[Dict] = None
_mp_caches: Optional[Dict] = None
_mp_platform_cfg: Optional[PlatformConfig] = None
_mp_bt_cfg: Optional["BacktestConfig"] = None
# Cooldown.
if pos - last_entry_pos < bt_cfg.cooldown_bars:
continue
# Build sliced dfs per TF — only include bars fully closed by
# decision_time. A bar timestamped T closes at T + bar_dur, so we
# keep bars where T + bar_dur <= decision_time → T <= decision_time - bar_dur.
# Without this, the H1 bar containing the current M15 close would be
# included with future high/low/close values: a look-ahead leak.
def _replay_worker(positions: List[int]) -> List[Tuple[int, Trade]]:
"""Process a chunk of bar positions and return candidate trades."""
aggregator = SignalAggregator(_mp_platform_cfg)
aggregator.tech_engine = GPUTechnicalEngine(_mp_platform_cfg.technical, _mp_caches)
timeframes = [_mp_bt_cfg.primary_tf] + list(_mp_bt_cfg.confirmation_tfs)
primary_df = _mp_dfs_full[_mp_bt_cfg.primary_tf]
primary_dur = pd.Timedelta(seconds=BAR_SECONDS[_mp_bt_cfg.primary_tf])
bar_durs = {tf: pd.Timedelta(seconds=BAR_SECONDS[tf]) for tf in timeframes}
n_primary = len(primary_df)
candidates: List[Tuple[int, Trade]] = []
for pos in positions:
decision_time = primary_df.index[pos] + primary_dur
slice_dfs: Dict[Timeframe, pd.DataFrame] = {}
for tf in timeframes:
full = dfs_full[tf]
bar_dur = pd.Timedelta(seconds=BAR_SECONDS[tf])
cutoff = decision_time - bar_dur
full = _mp_dfs_full[tf]
cutoff = decision_time - bar_durs[tf]
cut = int(full.index.searchsorted(cutoff, side="right"))
if cut == 0:
continue
slice_dfs[tf] = full.iloc[:cut]
if cut > 0:
slice_dfs[tf] = full.iloc[:cut]
if bt_cfg.primary_tf not in slice_dfs:
if _mp_bt_cfg.primary_tf not in slice_dfs:
continue
report: SignalReport = aggregator.analyze_from_dataframes(
symbol=bt_cfg.pair,
dfs=slice_dfs,
context=None,
use_news=False,
symbol=_mp_bt_cfg.pair, dfs=slice_dfs, context=None, use_news=False,
)
if report.trade_setup is None:
continue
if report.signal_quality not in bt_cfg.quality_filter:
if report.signal_quality not in _mp_bt_cfg.quality_filter:
continue
# Strategy tag (from bullish_reasons[0] which the aggregator pre-pends).
tag = "GENERIC"
if report.bullish_reasons:
first = report.bullish_reasons[0]
@@ -1838,28 +1810,23 @@ def run_backtest(
elif "GBP STRATEGY" in first:
tag = "GBP_PB"
if bt_cfg.strategies_only and tag not in bt_cfg.strategies_only:
if _mp_bt_cfg.strategies_only and tag not in _mp_bt_cfg.strategies_only:
continue
ts_setup: TradeSetup = report.trade_setup
# Entry on the NEXT bar's open price would be ideal, but live signals fire
# at bar close and we use that close as entry (matches setup.entry_price).
# For simulation realism we step entry to next bar so bar `pos` is not
# used to resolve the trade outcome (bar `pos` is the signal bar).
entry_pos = pos + 1
if entry_pos >= len(primary_df):
if entry_pos >= n_primary:
continue
entry_time = primary_df.index[entry_pos]
breakdown, agree, oppose, neutral, total = _build_indicator_breakdown(
report, ts_setup.direction
report, ts_setup.direction,
)
reasons = (report.bullish_reasons if ts_setup.direction == "BUY"
else report.bearish_reasons)
trades.append(Trade(
pair=bt_cfg.pair,
entry_time=entry_time,
candidates.append((pos, Trade(
pair=_mp_bt_cfg.pair,
entry_time=primary_df.index[entry_pos],
entry_price=ts_setup.entry_price,
sl=ts_setup.stop_loss,
tp1=ts_setup.take_profit_1,
@@ -1882,21 +1849,131 @@ def run_backtest(
indicators_opposing=oppose,
indicators_neutral=neutral,
indicators_total=total,
))
)))
return candidates
def run_backtest(
bt_cfg: BacktestConfig,
platform_cfg: Optional[PlatformConfig] = None,
) -> Tuple[List[Trade], Dict[Timeframe, pd.DataFrame]]:
"""Replay history and collect trades."""
if platform_cfg is None:
platform_cfg = PlatformConfig(
symbol=bt_cfg.pair,
display_name=PAIR_INFO[bt_cfg.pair]["display"],
pip_value=PAIR_INFO[bt_cfg.pair]["pip_value"],
primary_timeframe=bt_cfg.primary_tf,
confirmation_timeframes=list(bt_cfg.confirmation_tfs),
)
provider = ForexDataProvider(platform_cfg)
# ── Load all required timeframes ──
timeframes = [bt_cfg.primary_tf] + list(bt_cfg.confirmation_tfs)
pad_days = max(60, bt_cfg.warmup_bars // 4)
fetch_start = bt_cfg.start - timedelta(days=pad_days)
dfs_full: Dict[Timeframe, pd.DataFrame] = {}
for tf in timeframes:
df = fetch_history(
provider, bt_cfg.pair, tf,
start=fetch_start, end=bt_cfg.end,
use_cache=bt_cfg.use_cache,
)
if df.empty:
raise RuntimeError(f"No data for {bt_cfg.pair} {tf.value}")
dfs_full[tf] = df
# ── Pre-compute indicators on each TF ──
t0 = time.perf_counter()
caches: Dict[str, IndicatorCache] = {
tf.value: precompute_indicators(dfs_full[tf], platform_cfg.technical, gpu_log=True)
for tf in timeframes
}
logger.info(f"Indicator pre-compute took {time.perf_counter() - t0:.2f}s")
# ── Determine bar range ──
primary_df = dfs_full[bt_cfg.primary_tf]
bt_start_ts = pd.Timestamp(bt_cfg.start)
bt_end_ts = pd.Timestamp(bt_cfg.end)
first_eligible_pos = max(
bt_cfg.warmup_bars,
int(primary_df.index.searchsorted(bt_start_ts)),
)
last_pos = min(
int(primary_df.index.searchsorted(bt_end_ts, side="right")),
len(primary_df),
)
# ── Vectorized pre-filter (weekend + weekday) ──
eligible, skipped_reasons = _prefilter_bars(
primary_df, bt_cfg, platform_cfg, first_eligible_pos, last_pos,
)
total_bars = last_pos - first_eligible_pos
logger.info(
f"Replaying {bt_cfg.pair} {bt_cfg.primary_tf.value}: "
f"{len(eligible)} eligible bars of {total_bars} "
f"(filtered {total_bars - len(eligible)})"
)
# ── Parallel signal replay across CPU cores ──
n_workers = min(os.cpu_count() or 4, max(1, len(eligible) // 200))
t_loop = time.perf_counter()
if n_workers > 1:
global _mp_dfs_full, _mp_caches, _mp_platform_cfg, _mp_bt_cfg
_mp_dfs_full = dfs_full
_mp_caches = caches
_mp_platform_cfg = platform_cfg
_mp_bt_cfg = bt_cfg
chunks = [eligible[i::n_workers] for i in range(n_workers)]
ctx = mp.get_context("fork")
with ctx.Pool(n_workers) as pool:
chunk_results = pool.map(_replay_worker, chunks)
_mp_dfs_full = _mp_caches = _mp_platform_cfg = _mp_bt_cfg = None
all_candidates = sorted(chain.from_iterable(chunk_results), key=lambda x: x[0])
else:
_mp_dfs_full = dfs_full
_mp_caches = caches
_mp_platform_cfg = platform_cfg
_mp_bt_cfg = bt_cfg
all_candidates = _replay_worker(eligible)
_mp_dfs_full = _mp_caches = _mp_platform_cfg = _mp_bt_cfg = None
# ── Apply cooldown (sequential — depends on previous trade) ──
trades: List[Trade] = []
last_entry_pos = -10_000
for pos, trade in all_candidates:
if pos - last_entry_pos < bt_cfg.cooldown_bars:
continue
trades.append(trade)
last_entry_pos = pos
elapsed = time.perf_counter() - t_loop
logger.info(
f"Replay complete: {len(trades)} entries in "
f"{time.perf_counter() - t_loop:.1f}s "
f"({(last_pos - first_eligible_pos) / max(time.perf_counter() - t_loop, 1e-6):.0f} bars/s)"
f"Replay complete: {len(trades)} entries in {elapsed:.1f}s "
f"({len(eligible) / max(elapsed, 1e-6):.0f} bars/s, {n_workers} workers)"
)
if skipped_reasons:
logger.info(f"Skipped bars by reason: {skipped_reasons}")
# ── Resolve trade outcomes ──
t_sim = time.perf_counter()
be_trigger = getattr(platform_cfg.eur_strategy, 'be_trigger_r', 0.0) if platform_cfg.eur_strategy.enabled and bt_cfg.pair != XAUUSD else 0.0
be_buffer = getattr(platform_cfg.eur_strategy, 'be_buffer_r', 0.0) if be_trigger > 0 else 0.0
if bt_cfg.pair == XAUUSD:
be_trigger = platform_cfg.xau_strategy.be_trigger_r
be_buffer = platform_cfg.xau_strategy.be_buffer_r
elif platform_cfg.eur_strategy.enabled:
be_trigger = platform_cfg.eur_strategy.be_trigger_r
be_buffer = platform_cfg.eur_strategy.be_buffer_r
else:
be_trigger, be_buffer = 0.0, 0.0
simulate_trades(trades, primary_df, max_bars=bt_cfg.max_hold_bars,
be_trigger_r=be_trigger, be_buffer_r=be_buffer)
logger.info(
+10 -8
View File
@@ -159,6 +159,8 @@ class XauStrategyConfig:
rsi_zone: tuple = (40.0, 65.0)
allowed_weekdays: List[int] = field(default_factory=lambda: [0, 1, 2, 3, 4, 6]) # Mon-Fri + Sun (Sydney open)
sell_min_quality: List[str] = field(default_factory=lambda: ["A+", "A", "B"])
be_trigger_r: float = 0.0
be_buffer_r: float = 0.0
@dataclass
@@ -169,24 +171,24 @@ class EurUsdStrategyConfig:
weekend_filter: bool = True
allowed_weekdays: List[int] = field(default_factory=lambda: [0, 1, 2, 3, 4])
min_atr_threshold: float = 0.0003
sl_atr_multiplier: float = 0.5
tp_atr_multiplier: float = 1.5
sl_atr_multiplier: float = 1.5
tp_atr_multiplier: float = 1.0
sl_atr_floor: float = 1.2
min_rr_ratio: float = 1.0
min_rr_ratio: float = 0.5
mature_move_threshold: float = 35.0
tp_quality_scale: Dict[str, float] = field(default_factory=lambda: {
"A+": 1.0, "A": 1.0, "B": 1.0, "C": 1.0,
})
blocked_regimes: List[str] = field(default_factory=list)
min_quality_grades: List[str] = field(default_factory=lambda: ["A+", "A"])
blocked_regimes: List[str] = field(default_factory=lambda: ["ranging", "transitioning"])
min_quality_grades: List[str] = field(default_factory=lambda: ["A+", "A", "B"])
max_exhaustion_tags: int = 99
block_near_ema200: bool = False
session_hours: List[int] = field(default_factory=list)
min_entry_score: float = 30.0
min_entry_confidence: float = 50.0
min_entry_score: float = 38.0
min_entry_confidence: float = 60.0
block_rsi_extended: bool = True
be_trigger_r: float = 0.3
be_trigger_r: float = 0.2
be_buffer_r: float = 0.05
+2 -1
View File
@@ -80,8 +80,9 @@ def make_config(
config.eur_strategy.sl_atr_floor = 0.8
config.eur_strategy.sl_atr_multiplier = 0.3
config.eur_strategy.min_quality_grades = ["A+", "A", "B"]
config.eur_strategy.min_entry_score = 28.0
config.eur_strategy.min_entry_score = 35.0
config.eur_strategy.min_entry_confidence = 50.0
config.backtest.blocked_hours_utc = [22, 23, 0, 1, 2, 3, 4, 19, 21]
else:
config.technical.regime_weights_override = EUR_REGIME_WEIGHTS
+5
View File
@@ -20,4 +20,9 @@ def make_config(
config.confirmation_timeframes = (
confirmation_tfs or _CONFIRMATIONS.get(primary_tf, [Timeframe.M15, Timeframe.H4])
)
if primary_tf == Timeframe.M5:
config.xau_strategy.be_trigger_r = 0.15
config.xau_strategy.be_buffer_r = 0.05
return config
+2 -1
View File
@@ -498,7 +498,8 @@ class SignalAggregator:
if report.primary_analysis else 0)
if atr > 0:
report.trade_setup = self._generate_xau_atr_trade_setup(report, atr)
if report.trade_setup is None and report.signal_quality in ("A+", "A", "B", "C"):
setup_grades = ("A+", "A", "B", "C")
if report.trade_setup is None and report.signal_quality in setup_grades:
report.trade_setup = self._generate_trade_setup(report, pip_value)
# ── 8. Reasons ─────────────────────────────────────────────
+4 -5
View File
@@ -1656,15 +1656,14 @@ def _():
assert 6 not in cfg.allowed_weekdays # No Sunday
assert cfg.min_atr_threshold == 0.0003
assert cfg.sl_atr_floor == 1.2
assert cfg.min_rr_ratio == 1.0
assert cfg.min_rr_ratio == 0.5
assert cfg.mature_move_threshold == 35.0
assert cfg.tp_quality_scale["A+"] == 1.0
assert cfg.tp_quality_scale["A"] == 1.0
assert cfg.tp_quality_scale["C"] == 1.0
assert cfg.blocked_regimes == []
assert cfg.blocked_regimes == ["ranging", "transitioning"]
assert "A" in cfg.min_quality_grades
assert "A" in cfg.min_quality_grades
assert "B" not in cfg.min_quality_grades
assert "B" in cfg.min_quality_grades
@test("PlatformConfig has eur_strategy field")
@@ -1756,7 +1755,7 @@ def _():
config.eur_strategy.blocked_regimes = []
config.eur_strategy.min_quality_grades = []
assert config.eur_strategy.sl_atr_floor == 1.2
assert config.eur_strategy.min_rr_ratio == 1.0
assert config.eur_strategy.min_rr_ratio == 0.5
agg = SignalAggregator(config)
df = make_ohlcv(n=300, start_price=1.0800, trend=0.0003, seed=77)
dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
+51 -7
View File
@@ -86,6 +86,8 @@ _LEVEL_KEYS = (
_signal_stats: Dict[str, Dict[str, SignalReport]] = {}
_stats_updated: Optional[float] = None
_last_sent: Dict[str, Dict] = {}
SCAN_BUTTONS = [
{"cb": "scan_xauusd_m5", "symbol": XAUUSD, "tf": Timeframe.M5, "label": "XAU 5m"},
{"cb": "scan_xauusd_m15", "symbol": XAUUSD, "tf": Timeframe.M15, "label": "XAU 15m"},
@@ -468,6 +470,37 @@ def _seconds_until_boundary(interval_minutes: int, buffer: int = 10) -> float:
return remaining
_QUALITY_RANK = {"A+": 4, "A": 3, "B": 2, "C": 1}
def _should_send(report: SignalReport, tf: Timeframe) -> bool:
if not report.trade_setup:
return False
key = f"{report.symbol}:{tf.value}"
prev = _last_sent.get(key)
if prev is None:
return True
direction = report.trade_setup.direction
quality = report.signal_quality
if direction != prev["direction"]:
return True
if _QUALITY_RANK.get(quality, 0) > _QUALITY_RANK.get(prev["quality"], 0):
return True
cooldown = 5 * _TF_SECONDS.get(tf, 300)
if time.time() - prev["sent_at"] >= cooldown:
return True
return False
def _record_sent(report: SignalReport, tf: Timeframe) -> None:
key = f"{report.symbol}:{tf.value}"
_last_sent[key] = {
"direction": report.trade_setup.direction,
"quality": report.signal_quality,
"sent_at": time.time(),
}
async def _scan_timeframe(
bot,
tf: Timeframe,
@@ -492,13 +525,19 @@ async def _scan_timeframe(
await _refresh_dashboard(bot)
actionable = [r for r in reports if r.signal_quality in ("A+", "A", "B", "C")]
for chat_id in list(_autoscan_chats):
for report in actionable:
detail = format_signal_message(report)
try:
await bot.send_message(chat_id, detail, parse_mode=ParseMode.MARKDOWN)
except Exception as exc:
logger.warning(f"Failed to send signal for chat {chat_id}: {exc}")
sendable = [r for r in actionable if _should_send(r, tf)]
for report in sendable:
_record_sent(report, tf)
if sendable:
for chat_id in list(_autoscan_chats):
for report in sendable:
detail = format_signal_message(report)
try:
await bot.send_message(chat_id, detail, parse_mode=ParseMode.MARKDOWN)
except Exception as exc:
logger.warning(f"Failed to send signal for chat {chat_id}: {exc}")
elif actionable:
logger.info(f"Auto-scan {tf.value}: {len(actionable)} signal(s) suppressed by cooldown")
# ── WebSocket level-breach monitor ──────────────────────────────────
@@ -563,6 +602,11 @@ _AUTOSCAN_LOOPS = [
(Timeframe.H4, 240),
]
_TF_SECONDS = {
Timeframe.M5: 300, Timeframe.M15: 900,
Timeframe.H1: 3600, Timeframe.H4: 14400,
}
def _start_scheduler(bot) -> None:
global _dashboard_task