Add pair_signals per-pair producer registry

Ported from feature/eurusd-support: PAIR_PRODUCERS registry with
make_config() factories per pair. xauusd.make_config is a pure
PlatformConfig-defaults wrapper (equality with the inline construction
is enforced by test); eurusd.make_config carries the donor branch's
regime weights and gates as a pre-realism baseline.

Adds 4 registry tests. 154/154 pass.
This commit is contained in:
Harwinder Sekhon
2026-07-10 20:52:06 -04:00
parent 9d3729ea73
commit 1bc48915bd
4 changed files with 235 additions and 1 deletions
+52
View File
@@ -0,0 +1,52 @@
"""
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 PlatformConfig, Timeframe
from signal_aggregator import SignalAggregator, SignalReport
from data_provider import ForexDataProvider
PAIR_PRODUCERS: Dict[str, Callable] = {}
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,
data_provider: ForexDataProvider,
confirmation_tfs: Optional[List[Timeframe]] = None,
) -> 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)
from pair_signals import xauusd as _xau_mod
from pair_signals import eurusd as _eur_mod
from config import XAUUSD, EURUSD
register_pair(XAUUSD, _xau_mod.make_config)
register_pair(EURUSD, _eur_mod.make_config)
+100
View File
@@ -0,0 +1,100 @@
"""EUR/USD signal producer config.
Ported from feature/eurusd-support. All tuned values here are a
pre-realism baseline (tuned before spread charging and next-open fills)
and are re-tuned against the realistic backtester at 1-pip round-trip
cost — see EurUsdStrategyConfig in config.py.
"""
from typing import Optional, List
from config import PlatformConfig, Timeframe, EURUSD
_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],
}
_TRENDING = {
"ema_trend": 2.0, "rsi": 1.0, "macd": 2.0,
"bollinger": 0.5, "keltner": 1.0, "roc": 1.5,
"candlestick": 0.5, "divergence": 2.0, "adx": 1.0,
}
EUR_REGIME_WEIGHTS = {
"trending_up": _TRENDING,
"trending_down": _TRENDING,
"trending": _TRENDING,
"ranging": {
"ema_trend": 0.3, "rsi": 2.0, "macd": 1.0,
"bollinger": 2.0, "keltner": 1.5, "roc": 0.5,
"candlestick": 0.8, "divergence": 2.5, "adx": 0.5,
},
"transitioning": {
"ema_trend": 1.5, "rsi": 1.5, "macd": 2.0,
"bollinger": 1.0, "keltner": 1.5, "roc": 2.0,
"candlestick": 0.8, "divergence": 2.0, "adx": 1.0,
},
}
_M5_TRENDING = {
"ema_trend": 1.5, "rsi": 1.0, "macd": 2.5,
"bollinger": 0.5, "keltner": 2.0, "roc": 2.0,
"candlestick": 0.3, "divergence": 1.0, "adx": 1.0,
}
M5_REGIME_WEIGHTS = {
"trending_up": _M5_TRENDING,
"trending_down": _M5_TRENDING,
"trending": _M5_TRENDING,
"ranging": {
"ema_trend": 0.3, "rsi": 1.5, "macd": 1.5,
"bollinger": 1.5, "keltner": 2.0, "roc": 1.0,
"candlestick": 0.3, "divergence": 1.5, "adx": 0.5,
},
"transitioning": {
"ema_trend": 1.0, "rsi": 1.0, "macd": 2.5,
"bollinger": 0.5, "keltner": 2.5, "roc": 2.0,
"candlestick": 0.3, "divergence": 1.0, "adx": 1.0,
},
}
def make_config(
primary_tf: Timeframe,
confirmation_tfs: Optional[List[Timeframe]] = None,
) -> PlatformConfig:
config = PlatformConfig()
config.symbol = EURUSD
config.display_name = "EUR/USD"
config.pip_value = 0.0001
config.primary_timeframe = primary_tf
config.confirmation_timeframes = (
confirmation_tfs or _CONFIRMATIONS.get(primary_tf, [Timeframe.M15, Timeframe.H4])
)
# XAU gates are display-gated in the aggregator; disabling here is
# belt-and-braces so no gold logic can ever fire on an EUR config.
config.xau_strategy.enabled = False
config.backtest.session_start_hour = 7
config.backtest.session_end_hour = 20
config.backtest.blocked_hours_utc = [22, 23, 0, 1, 2, 3]
config.backtest.allowed_weekdays = [0, 1, 2, 3, 4]
if primary_tf == Timeframe.M5:
config.technical.regime_weights_override = M5_REGIME_WEIGHTS
config.technical.swing_sr_lookback = 20
config.eur_strategy.sl_atr_floor = 0.7
config.eur_strategy.sl_atr_multiplier = 0.3
config.eur_strategy.min_quality_grades = ["A+", "A", "B"]
config.eur_strategy.min_entry_score = 35.0
config.eur_strategy.min_entry_confidence = 50.0
config.eur_strategy.blocked_vol_regimes = ["high"]
config.eur_strategy.min_indicator_agreement = 6
config.backtest.blocked_hours_utc = [22, 23, 0, 1, 2, 3, 4, 6, 9, 11, 19, 21]
else:
config.technical.regime_weights_override = EUR_REGIME_WEIGHTS
return config
+29
View File
@@ -0,0 +1,29 @@
"""XAU/USD signal producer config.
PlatformConfig defaults ARE the tuned XAU/USD configuration, so this
factory only sets the timeframes. It must stay equivalent to the inline
PlatformConfig(symbol=XAUUSD, ...) construction used by the backtester
and Telegram bot (enforced by a test in test_all.py).
"""
from typing import Optional, List
from config import PlatformConfig, Timeframe
_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],
}
def make_config(
primary_tf: Timeframe,
confirmation_tfs: Optional[List[Timeframe]] = None,
) -> PlatformConfig:
config = PlatformConfig()
config.primary_timeframe = primary_tf
config.confirmation_timeframes = (
confirmation_tfs or _CONFIRMATIONS.get(primary_tf, [Timeframe.M15, Timeframe.H4])
)
return config
+54 -1
View File
@@ -1761,7 +1761,60 @@ def _():
assert r1.market_regime == r2.market_regime assert r1.market_regime == r2.market_regime
print("\n═══ Module 9: Engine Bug Fix Regressions ═══") print("\n═══ Module 9: pair_signals registry ═══")
@test("pair_signals registry includes XAU/USD and EUR/USD")
def _():
import pair_signals
pairs = pair_signals.get_supported_pairs()
assert XAUUSD in pairs, f"XAU missing from registry: {pairs}"
assert "EUR/USD" in pairs, f"EUR missing from registry: {pairs}"
@test("xauusd.make_config equals inline PlatformConfig construction")
def _():
import dataclasses
from pair_signals import PAIR_PRODUCERS
for tf, ctfs in [
(Timeframe.M5, [Timeframe.M15, Timeframe.H1]),
(Timeframe.M15, [Timeframe.M5, Timeframe.H1]),
(Timeframe.H1, [Timeframe.M15, Timeframe.H4]),
]:
factory_cfg = PAIR_PRODUCERS[XAUUSD](tf, ctfs)
inline_cfg = PlatformConfig(
symbol=XAUUSD,
display_name="XAU/USD",
pip_value=0.10,
primary_timeframe=tf,
confirmation_timeframes=ctfs,
)
assert dataclasses.asdict(factory_cfg) == dataclasses.asdict(inline_cfg), \
f"XAU factory config differs from inline construction for {tf}"
@test("eurusd.make_config produces EUR config with XAU strategy disabled")
def _():
from pair_signals import PAIR_PRODUCERS
cfg = PAIR_PRODUCERS["EUR/USD"](Timeframe.M5, None)
assert cfg.symbol == "EUR/USD"
assert cfg.pip_value == 0.0001
assert cfg.xau_strategy.enabled is False
assert cfg.eur_strategy.enabled is True
assert cfg.technical.regime_weights_override, "EUR regime weights not applied"
assert cfg.confirmation_timeframes == [Timeframe.M15, Timeframe.H1]
@test("produce_signal raises for unregistered pair")
def _():
from pair_signals import produce_signal
try:
produce_signal("GBP/JPY", Timeframe.H1, None)
assert False, "Expected ValueError for unregistered pair"
except ValueError as e:
assert "GBP/JPY" in str(e)
print("\n═══ Module 9b: Engine Bug Fix Regressions ═══")
# ── 9a. Swing point NaN fix ────────────────────────────────────────── # ── 9a. Swing point NaN fix ──────────────────────────────────────────