chore: remove test-only helpers and the duplicate confirmation map (no behaviour change)

- telegram_bot._format_scan_summary: multi-pair "auto-scan summary" leftover;
  production sends per-report format_signal_message. Its only caller was a test.
- telegram_bot.scan_keyboard: thin wrapper only a test called; main_menu_keyboard
  already builds from _scan_button_rows directly.
- telegram_bot.AUTOSCAN_CONFIRMATIONS was byte-identical to TF_CONFIRMATIONS;
  the candle loop now reads the one map (app_config comment updated).
- SignalLogger.news_for: one-line wrapper over the module-level read_news that
  every real reader uses; the test now calls read_news like its neighbours.
- pair_signals.get_supported_pairs: PAIR_PRODUCERS is the registry; the test
  lists it directly.

Suite: 461 passed, 0 failed (two helper-only tests removed).

Claude-Session: https://claude.ai/code/session_0156YhogLytzj9P2iH1skGbL
This commit is contained in:
Harwinder Sekhon
2026-09-06 16:50:21 -04:00
parent 985ec83ef3
commit 1b40f2166e
5 changed files with 6 additions and 78 deletions
+1 -1
View File
@@ -82,7 +82,7 @@ load_dotenv()
_VALID_QUALITIES = ("A+", "A", "B", "C")
_VALID_LOG_LEVELS = ("DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL")
# Timeframes the bot's candle-boundary scheduler supports (must have
# AUTOSCAN_CONFIRMATIONS entries in telegram_bot.py).
# TF_CONFIRMATIONS entries in telegram_bot.py).
_SCANNABLE_TFS = ("M5", "M15", "H1", "H4")
_KNOWN_TOP = {"telegram", "data_providers", "scanning", "pairs", "log_level",
-4
View File
@@ -21,10 +21,6 @@ def register_pair(symbol: str, config_factory: Callable):
PAIR_PRODUCERS[symbol] = config_factory
def get_supported_pairs() -> List[str]:
return list(PAIR_PRODUCERS.keys())
def produce_signal(
symbol: str,
primary_tf: Timeframe,
-5
View File
@@ -978,11 +978,6 @@ class SignalLogger:
(now_iso, digest))
return digest
def news_for(self, rowid: int) -> Tuple[Optional[str], Optional[str]]:
"""(events_json, headlines_json) for a row, legacy or interned."""
with self._lock:
return read_news(self._conn, rowid)
# ── advisory-event seeding / dedup ────────────────────────────
def _seed_outcome(self, rowid: int, row: Dict[str, Any]) -> None:
+1 -43
View File
@@ -188,13 +188,6 @@ TF_CONFIRMATIONS = {
Timeframe.H4: [Timeframe.H1, Timeframe.D1],
}
AUTOSCAN_CONFIRMATIONS = {
Timeframe.M5: [Timeframe.M15, Timeframe.H1],
Timeframe.M15: [Timeframe.M5, Timeframe.H1],
Timeframe.H1: [Timeframe.M15, Timeframe.H4],
Timeframe.H4: [Timeframe.H1, Timeframe.D1],
}
_executor = ThreadPoolExecutor(max_workers=4)
# ── Markdown escaping ─────────────────────────────────────────────────
@@ -441,37 +434,6 @@ def format_signal_message(report: SignalReport) -> str:
return "\n".join(lines)
# ── Auto-scan summary format ─────────────────────────────────────────
def _format_scan_summary(reports: List[SignalReport], tf: Timeframe) -> str:
now = datetime.now(timezone.utc).strftime("%H:%M UTC")
lines = [
f"📊 *{tf.value} Auto-Scan* | {now}",
"━━━━━━━━━━━━━━━━━━━━",
]
for r in reports:
emoji = _SIGNAL_EMOJI.get(r.final_signal, "⬜")
sig = _signal_label(r.final_signal)
mtf = "✅" if r.mtf_agreement else "❌"
lines.append(
f"{emoji} *{r.display_name}* {sig} "
f"{_QUALITY_EMOJI.get(r.signal_quality, '')}{r.signal_quality} "
f"{r.final_score:+.1f} {r.final_confidence:.0f}% {mtf}MTF"
)
lines.append("━━━━━━━━━━━━━━━━━━━━")
actionable = [r for r in reports if r.signal_quality in ("A+", "A", "B", "C")]
if actionable:
names = ", ".join(r.display_name for r in actionable)
lines.append(f"🔔 Actionable: {names}")
else:
lines.append("No actionable signals this cycle.")
lines += ["", _DISCLAIMER]
return "\n".join(lines)
# ── Keyboard helper ───────────────────────────────────────────────────
def _scan_button_rows() -> List[List[InlineKeyboardButton]]:
@@ -490,10 +452,6 @@ def _scan_button_rows() -> List[List[InlineKeyboardButton]]:
return rows
def scan_keyboard() -> InlineKeyboardMarkup:
return InlineKeyboardMarkup(_scan_button_rows())
def main_menu_keyboard() -> InlineKeyboardMarkup:
"""Full interactive menu: scans + settings + status in one keyboard."""
autoscan_label = "🔔 Auto-scan: ON" if _autoscan_chats else "🔕 Auto-scan: OFF"
@@ -1147,7 +1105,7 @@ async def _candle_loop(bot, tf: Timeframe, interval_minutes: int) -> None:
logger.info(f"{label} auto-scan: firing")
try:
await _scan_timeframe(
bot, tf, AUTOSCAN_CONFIRMATIONS[tf]
bot, tf, TF_CONFIRMATIONS[tf]
)
except Exception:
# One failed cycle must not kill the loop for good —
+4 -25
View File
@@ -1612,9 +1612,9 @@ print("\n═══ Module 7: telegram_bot.py ═══")
from telegram_bot import (
_escape_md, _signal_label, _mtf_line, _indicator_summary,
format_signal_message, _format_scan_summary, _news_text,
format_signal_message, _news_text,
_seconds_until_boundary, _check_level_breach,
scan_keyboard, SCAN_BUTTONS, SCAN_LOOKUP,
SCAN_BUTTONS, SCAN_LOOKUP,
_update_level_cache, _last_prices, _level_cache, _level_cache_ts,
)
@@ -1757,21 +1757,6 @@ def _():
assert "[Bullish]" not in msg and "[Bearish]" not in msg and "[Neutral]" not in msg
@test("_format_scan_summary produces valid text")
def _():
config = PlatformConfig()
agg = SignalAggregator(config)
dfs = {
Timeframe.H1: make_ohlcv(300, seed=42),
Timeframe.M15: make_ohlcv(300, seed=43),
Timeframe.H4: make_ohlcv(300, seed=44),
}
reports = [agg.analyze_from_dataframes("XAU/USD", dfs, use_news=False)]
summary = _format_scan_summary(reports, Timeframe.H1)
assert "Auto-Scan" in summary
assert "XAU/USD" in summary
@test("_seconds_until_boundary returns positive value")
def _():
result = _seconds_until_boundary(15, buffer=10)
@@ -1878,12 +1863,6 @@ def _():
assert "label" in btn
@test("scan_keyboard returns InlineKeyboardMarkup")
def _():
kb = scan_keyboard()
assert kb is not None
@test("_update_level_cache populates cache from reports")
def _():
config = PlatformConfig()
@@ -2070,7 +2049,7 @@ print("\n═══ Module 9: pair_signals registry ═══")
@test("pair_signals registry includes XAU/USD only")
def _():
import pair_signals
pairs = pair_signals.get_supported_pairs()
pairs = list(pair_signals.PAIR_PRODUCERS)
assert XAUUSD in pairs, f"XAU missing from registry: {pairs}"
assert pairs == [XAUUSD], f"registry must be XAU-only: {pairs}"
@@ -8900,7 +8879,7 @@ def _():
legacy_id = lg._conn.execute("SELECT MAX(id) FROM signals").fetchone()[0]
ev, hl = slog.read_news(lg._conn, legacy_id)
assert json.loads(hl)[0]["headline"] == "legacy"
ev, hl = lg.news_for(r1)
ev, hl = slog.read_news(lg._conn, r1)
assert json.loads(hl)[0]["headline"] == "Fed signals rate hike"
view = dict(lg._conn.execute(
"SELECT id, headlines_resolved FROM signals_with_news").fetchall())