Harwinder Sekhon
1b40f2166e
- 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
48 lines
1.3 KiB
Python
48 lines
1.3 KiB
Python
"""
|
|
Per-pair signal producers.
|
|
|
|
Each pair module exposes:
|
|
make_config(primary_tf, confirmation_tfs) -> PlatformConfig
|
|
|
|
To add a new pair: create pair_signals/<pair>.py with make_config(),
|
|
then register it in this file.
|
|
"""
|
|
|
|
from typing import Optional, List, Dict, Callable
|
|
|
|
from config import Timeframe
|
|
from signal_aggregator import SignalAggregator, SignalReport
|
|
from providers import BaseDataProvider
|
|
|
|
PAIR_PRODUCERS: Dict[str, Callable] = {}
|
|
|
|
|
|
def register_pair(symbol: str, config_factory: Callable):
|
|
PAIR_PRODUCERS[symbol] = config_factory
|
|
|
|
|
|
def produce_signal(
|
|
symbol: str,
|
|
primary_tf: Timeframe,
|
|
data_provider: BaseDataProvider,
|
|
confirmation_tfs: Optional[List[Timeframe]] = None,
|
|
log_source: str = "dashboard",
|
|
) -> SignalReport:
|
|
config_factory = PAIR_PRODUCERS.get(symbol)
|
|
if config_factory is None:
|
|
raise ValueError(
|
|
f"No signal producer registered for {symbol}. "
|
|
f"Available: {list(PAIR_PRODUCERS.keys())}"
|
|
)
|
|
|
|
config = config_factory(primary_tf, confirmation_tfs)
|
|
aggregator = SignalAggregator(config)
|
|
aggregator.data_provider = data_provider
|
|
return aggregator.analyze_pair(symbol=symbol, log_source=log_source)
|
|
|
|
|
|
from pair_signals import xauusd as _xau_mod
|
|
from config import XAUUSD
|
|
|
|
register_pair(XAUUSD, _xau_mod.make_config)
|