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). Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0156YhogLytzj9P2iH1skGbL
This commit is contained in:
+1
-1
@@ -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",
|
||||
|
||||
@@ -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,
|
||||
|
||||
@@ -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
@@ -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
@@ -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())
|
||||
|
||||
Reference in New Issue
Block a user