harwinder
541ef7ff4a
- Remove websockets import, _ws_monitor, _handle_breach, _breach_tasks - Remove _WS_TO_PAIR, _WS_SYMBOLS, _WS_URL, _WS_COOLDOWN constants - Remove _last_prices, _scan_cooldowns, _cleanup_cooldowns - Remove ws_cooldown_seconds from PlatformConfig - Remove websockets from requirements.txt - Simplify _check_level_breach to proximity-based detection - Update dashboard to use report prices instead of WS prices - Clean up status/help messages, _start/_stop_scheduler - Update tests to match new level breach behavior Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2751 lines
100 KiB
Python
2751 lines
100 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
ForexBot2 — Comprehensive QA Test Suite
|
||
=========================================
|
||
Tests every function in the codebase without requiring live API connections.
|
||
"""
|
||
|
||
import sys
|
||
import os
|
||
import time
|
||
import traceback
|
||
from datetime import datetime, timedelta, timezone
|
||
from unittest.mock import patch, MagicMock
|
||
from io import StringIO
|
||
|
||
import numpy as np
|
||
import pandas as pd
|
||
|
||
sys.path.insert(0, os.path.dirname(__file__))
|
||
|
||
PASS = 0
|
||
FAIL = 0
|
||
ERRORS = []
|
||
|
||
|
||
def test(name):
|
||
def decorator(func):
|
||
global PASS, FAIL
|
||
try:
|
||
func()
|
||
PASS += 1
|
||
print(f" ✓ {name}")
|
||
except Exception as e:
|
||
FAIL += 1
|
||
tb = traceback.format_exc()
|
||
ERRORS.append((name, str(e), tb))
|
||
print(f" ✗ {name}: {e}")
|
||
return func
|
||
return decorator
|
||
|
||
|
||
def make_ohlcv(n=300, start_price=1.1000, trend=0.0001, seed=42):
|
||
"""Generate realistic OHLCV data for testing."""
|
||
np.random.seed(seed)
|
||
dates = pd.date_range(
|
||
end=pd.Timestamp.now(tz="UTC").floor("h"),
|
||
periods=n, freq="h"
|
||
)
|
||
close = start_price + np.cumsum(np.random.randn(n) * 0.001 + trend)
|
||
high = close + np.abs(np.random.randn(n) * 0.0005)
|
||
low = close - np.abs(np.random.randn(n) * 0.0005)
|
||
open_ = close + np.random.randn(n) * 0.0003
|
||
volume = np.random.randint(100, 10000, n).astype(float)
|
||
df = pd.DataFrame({
|
||
"Open": open_, "High": high, "Low": low,
|
||
"Close": close, "Volume": volume,
|
||
}, index=dates)
|
||
df.index.name = "Datetime"
|
||
return df
|
||
|
||
|
||
def make_xau_ohlcv(n=300, start_price=2400.0, seed=42):
|
||
"""Generate XAU/USD OHLCV data."""
|
||
np.random.seed(seed)
|
||
dates = pd.date_range(
|
||
end=pd.Timestamp.now(tz="UTC").floor("h"),
|
||
periods=n, freq="h"
|
||
)
|
||
close = start_price + np.cumsum(np.random.randn(n) * 2.0 + 0.05)
|
||
high = close + np.abs(np.random.randn(n) * 1.0)
|
||
low = close - np.abs(np.random.randn(n) * 1.0)
|
||
open_ = close + np.random.randn(n) * 0.5
|
||
volume = np.random.randint(100, 10000, n).astype(float)
|
||
df = pd.DataFrame({
|
||
"Open": open_, "High": high, "Low": low,
|
||
"Close": close, "Volume": volume,
|
||
}, index=dates)
|
||
df.index.name = "Datetime"
|
||
return df
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
# MODULE 1: config.py
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
|
||
print("\n═══ Module 1: config.py ═══")
|
||
|
||
from config import (
|
||
Timeframe, SignalStrength, MarketSession,
|
||
PlatformConfig, TechnicalConfig, SignalConfig, NewsConfig,
|
||
FilterConfig, XauStrategyConfig, EurUsdStrategyConfig,
|
||
XAUUSD, EURUSD, PAIR_INFO,
|
||
)
|
||
|
||
|
||
@test("Timeframe enum values")
|
||
def _():
|
||
assert Timeframe.M1.value == "1min"
|
||
assert Timeframe.M5.value == "5min"
|
||
assert Timeframe.M15.value == "15min"
|
||
assert Timeframe.M30.value == "30min"
|
||
assert Timeframe.H1.value == "1h"
|
||
assert Timeframe.H4.value == "4h"
|
||
assert Timeframe.D1.value == "1d"
|
||
assert Timeframe.W1.value == "1wk"
|
||
|
||
|
||
@test("SignalStrength enum values and ordering")
|
||
def _():
|
||
assert SignalStrength.STRONG_BUY.value == 2
|
||
assert SignalStrength.BUY.value == 1
|
||
assert SignalStrength.NEUTRAL.value == 0
|
||
assert SignalStrength.SELL.value == -1
|
||
assert SignalStrength.STRONG_SELL.value == -2
|
||
assert SignalStrength.STRONG_BUY.value > SignalStrength.BUY.value
|
||
assert SignalStrength.SELL.value > SignalStrength.STRONG_SELL.value
|
||
|
||
|
||
@test("MarketSession enum values")
|
||
def _():
|
||
assert MarketSession.SYDNEY.value == "sydney"
|
||
assert MarketSession.TOKYO.value == "tokyo"
|
||
assert MarketSession.LONDON.value == "london"
|
||
assert MarketSession.NEW_YORK.value == "new_york"
|
||
assert MarketSession.OVERLAP_LONDON_NY.value == "london_ny_overlap"
|
||
|
||
|
||
@test("PAIR_INFO completeness and correctness")
|
||
def _():
|
||
assert XAUUSD in PAIR_INFO
|
||
for pair, info in PAIR_INFO.items():
|
||
assert "display" in info, f"Missing 'display' for {pair}"
|
||
assert "pip_value" in info, f"Missing 'pip_value' for {pair}"
|
||
assert "is_jpy_pair" in info, f"Missing 'is_jpy_pair' for {pair}"
|
||
assert info["pip_value"] > 0, f"pip_value must be > 0 for {pair}"
|
||
assert PAIR_INFO[XAUUSD]["is_jpy_pair"] is False
|
||
assert PAIR_INFO[XAUUSD]["pip_value"] == 0.10
|
||
|
||
|
||
@test("TechnicalConfig defaults")
|
||
def _():
|
||
tc = TechnicalConfig()
|
||
assert tc.ema_fast == 9
|
||
assert tc.ema_medium == 21
|
||
assert tc.ema_slow == 50
|
||
assert tc.ema_trend == 200
|
||
assert tc.rsi_period == 14
|
||
assert tc.rsi_overbought == 70.0
|
||
assert tc.rsi_oversold == 30.0
|
||
assert tc.macd_fast == 12
|
||
assert tc.macd_slow == 26
|
||
assert tc.macd_signal == 9
|
||
assert tc.bb_period == 20
|
||
assert tc.bb_std == 2.0
|
||
assert tc.atr_period == 14
|
||
assert tc.adx_period == 14
|
||
assert tc.ema_mode == "4ema"
|
||
assert tc.rsi_mode == "classic"
|
||
assert tc.macd_as_pillar is True
|
||
assert tc.disabled_indicators == []
|
||
assert tc.roc_period == 10
|
||
assert tc.kc_period == 20
|
||
assert tc.adx_slope_lookback == 5
|
||
|
||
|
||
@test("SignalConfig defaults")
|
||
def _():
|
||
sc = SignalConfig()
|
||
assert sc.min_confidence == 55.0
|
||
assert sc.min_rr_ratio == 0.5
|
||
assert sc.atr_sl_multiplier == 1.5
|
||
assert sc.default_rr_ratio == 2.0
|
||
assert sc.cooldown_bars == 5
|
||
assert "ema_trend" in sc.weights
|
||
assert "rsi" in sc.weights
|
||
|
||
|
||
@test("PlatformConfig defaults")
|
||
def _():
|
||
pc = PlatformConfig()
|
||
assert pc.symbol == XAUUSD
|
||
assert pc.primary_timeframe == Timeframe.H1
|
||
assert Timeframe.M15 in pc.confirmation_timeframes
|
||
assert Timeframe.H4 in pc.confirmation_timeframes
|
||
assert pc.data_lookback_days == 60
|
||
assert pc.refresh_interval == 300
|
||
assert XAUUSD in pc.scan_pairs
|
||
|
||
|
||
@test("XauStrategyConfig defaults")
|
||
def _():
|
||
xc = XauStrategyConfig()
|
||
assert xc.enabled is True
|
||
assert xc.rsi_entry_threshold == 45.0
|
||
assert xc.min_ema200_tfs == 2
|
||
assert xc.allow_generic_trades is True
|
||
assert xc.tp_atr_multiplier == 1.0
|
||
assert xc.sl_atr_multiplier == 2.0
|
||
assert isinstance(xc.blocked_hours_utc, list)
|
||
|
||
|
||
@test("FilterConfig defaults")
|
||
def _():
|
||
fc = FilterConfig()
|
||
assert fc.session_filter is True
|
||
assert fc.regime_filter is False
|
||
|
||
|
||
@test("PlatformConfig dataclass mutable default independence")
|
||
def _():
|
||
c1 = PlatformConfig()
|
||
c2 = PlatformConfig()
|
||
c1.scan_pairs.append("TEST")
|
||
assert "TEST" not in c2.scan_pairs, "Mutable defaults should be independent"
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
# MODULE 2: data_provider.py
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
|
||
print("\n═══ Module 2: data_provider.py ═══")
|
||
|
||
from data_provider import ForexDataProvider, DataIntegrityError
|
||
|
||
|
||
@test("ForexDataProvider._to_td_symbol conversion")
|
||
def _():
|
||
assert ForexDataProvider._to_td_symbol("XAU/USD") == "XAU/USD"
|
||
|
||
|
||
@test("ForexDataProvider TD_INTERVALS complete mapping")
|
||
def _():
|
||
dp = ForexDataProvider(PlatformConfig())
|
||
for tf in Timeframe:
|
||
assert tf in dp.TD_INTERVALS, f"Missing TD_INTERVALS mapping for {tf}"
|
||
|
||
|
||
@test("ForexDataProvider BARS_PER_DAY mapping")
|
||
def _():
|
||
dp = ForexDataProvider(PlatformConfig())
|
||
for tf in Timeframe:
|
||
assert tf in dp.BARS_PER_DAY, f"Missing BARS_PER_DAY for {tf}"
|
||
assert dp.BARS_PER_DAY[Timeframe.H1] == 24
|
||
assert dp.BARS_PER_DAY[Timeframe.M15] == 96
|
||
assert dp.BARS_PER_DAY[Timeframe.D1] == 1
|
||
|
||
|
||
@test("ForexDataProvider CANDLE_SECONDS mapping")
|
||
def _():
|
||
dp = ForexDataProvider(PlatformConfig())
|
||
for tf in Timeframe:
|
||
assert tf in dp.CANDLE_SECONDS, f"Missing CANDLE_SECONDS for {tf}"
|
||
assert dp.CANDLE_SECONDS[Timeframe.M1] == 60
|
||
assert dp.CANDLE_SECONDS[Timeframe.H1] == 3600
|
||
assert dp.CANDLE_SECONDS[Timeframe.D1] == 86400
|
||
|
||
|
||
@test("ForexDataProvider._validate_and_clean basic valid data")
|
||
def _():
|
||
dp = ForexDataProvider(PlatformConfig())
|
||
df = make_ohlcv(300)
|
||
result = dp._validate_and_clean(df, "XAU/USD", Timeframe.H1)
|
||
assert len(result) > 0
|
||
assert result.index.is_monotonic_increasing
|
||
assert not result.index.has_duplicates
|
||
|
||
|
||
@test("ForexDataProvider._validate_and_clean empty df raises error")
|
||
def _():
|
||
dp = ForexDataProvider(PlatformConfig())
|
||
try:
|
||
dp._validate_and_clean(pd.DataFrame(), "XAU/USD", Timeframe.H1)
|
||
assert False, "Should have raised DataIntegrityError"
|
||
except DataIntegrityError:
|
||
pass
|
||
|
||
|
||
@test("ForexDataProvider._validate_and_clean None df raises error")
|
||
def _():
|
||
dp = ForexDataProvider(PlatformConfig())
|
||
try:
|
||
dp._validate_and_clean(None, "XAU/USD", Timeframe.H1)
|
||
assert False, "Should have raised DataIntegrityError"
|
||
except DataIntegrityError:
|
||
pass
|
||
|
||
|
||
@test("ForexDataProvider._validate_and_clean fixes OHLC inconsistencies")
|
||
def _():
|
||
dp = ForexDataProvider(PlatformConfig())
|
||
df = make_ohlcv(300)
|
||
df.iloc[10, df.columns.get_loc("High")] = df.iloc[10]["Low"] - 1
|
||
result = dp._validate_and_clean(df, "XAU/USD", Timeframe.H1)
|
||
assert (result["High"] >= result["Low"]).all()
|
||
|
||
|
||
@test("ForexDataProvider._validate_and_clean drops future timestamps")
|
||
def _():
|
||
dp = ForexDataProvider(PlatformConfig())
|
||
df = make_ohlcv(300)
|
||
future = pd.Timestamp.now(tz="UTC") + pd.Timedelta(hours=10)
|
||
df.loc[future] = df.iloc[-1]
|
||
result = dp._validate_and_clean(df, "XAU/USD", Timeframe.H1)
|
||
assert (result.index <= pd.Timestamp.utcnow() + pd.Timedelta(minutes=5)).all()
|
||
|
||
|
||
@test("ForexDataProvider._validate_and_clean removes duplicates")
|
||
def _():
|
||
dp = ForexDataProvider(PlatformConfig())
|
||
df = make_ohlcv(300)
|
||
dup = pd.concat([df, df.iloc[-5:]])
|
||
assert dup.index.has_duplicates, "Setup: should have duplicates"
|
||
result = dp._validate_and_clean(dup, "XAU/USD", Timeframe.H1)
|
||
assert not result.index.has_duplicates, "Should remove duplicate indices"
|
||
|
||
|
||
@test("ForexDataProvider._validate_and_clean forward fills NaN (limit 3)")
|
||
def _():
|
||
dp = ForexDataProvider(PlatformConfig())
|
||
df = make_ohlcv(300)
|
||
df.iloc[100, df.columns.get_loc("Close")] = np.nan
|
||
df.iloc[101, df.columns.get_loc("Close")] = np.nan
|
||
result = dp._validate_and_clean(df, "XAU/USD", Timeframe.H1)
|
||
assert not result["Close"].isna().any()
|
||
|
||
|
||
@test("ForexDataProvider._is_data_fresh weekend override")
|
||
def _():
|
||
dp = ForexDataProvider(PlatformConfig())
|
||
# Create data that ends on a Friday, call _is_data_fresh on a Saturday
|
||
friday = pd.Timestamp("2025-05-09 16:00:00", tz="UTC")
|
||
old_dates = pd.date_range(end=friday, periods=300, freq="h")
|
||
df = make_ohlcv(300)
|
||
df.index = old_dates
|
||
# Simulate calling on Saturday by patching pd.Timestamp.now
|
||
saturday = pd.Timestamp("2025-05-10 12:00:00", tz="UTC")
|
||
original_now = pd.Timestamp.now
|
||
with patch.object(pd.Timestamp, "now", return_value=saturday):
|
||
result = dp._is_data_fresh(df, "XAU/USD", Timeframe.H1)
|
||
assert result is True, "Weekend data should always be considered fresh"
|
||
|
||
|
||
@test("ForexDataProvider._is_data_fresh returns False for empty df")
|
||
def _():
|
||
dp = ForexDataProvider(PlatformConfig())
|
||
assert dp._is_data_fresh(pd.DataFrame(), "XAU/USD", Timeframe.H1) is False
|
||
assert dp._is_data_fresh(None, "XAU/USD", Timeframe.H1) is False
|
||
|
||
|
||
@test("ForexDataProvider.clear_cache clears all entries")
|
||
def _():
|
||
dp = ForexDataProvider(PlatformConfig())
|
||
dp._cache["test_key"] = make_ohlcv(10)
|
||
dp._cache_timestamps["test_key"] = time.time()
|
||
count = dp.clear_cache()
|
||
assert count == 1
|
||
assert len(dp._cache) == 0
|
||
assert len(dp._cache_timestamps) == 0
|
||
|
||
|
||
@test("ForexDataProvider.get_pair_info returns correct info")
|
||
def _():
|
||
dp = ForexDataProvider(PlatformConfig())
|
||
info = dp.get_pair_info(XAUUSD)
|
||
assert info["symbol"] == XAUUSD
|
||
assert info["display"] == "XAU/USD"
|
||
assert info["pip_value"] == 0.10
|
||
|
||
info_unknown = dp.get_pair_info("UNKNOWN")
|
||
assert info_unknown["display"] == "XAU/USD"
|
||
|
||
|
||
@test("ForexDataProvider._throttle rate limiting logic")
|
||
def _():
|
||
dp = ForexDataProvider(PlatformConfig())
|
||
dp._request_times.clear()
|
||
for _ in range(10):
|
||
dp._throttle(credits=1)
|
||
assert len(dp._request_times) == 10
|
||
|
||
|
||
@test("ForexDataProvider caching - in-memory cache returns same data")
|
||
def _():
|
||
dp = ForexDataProvider(PlatformConfig())
|
||
df = make_ohlcv(300)
|
||
dp._cache["XAU/USD_1h"] = df
|
||
dp._cache_timestamps["XAU/USD_1h"] = time.time()
|
||
cached = dp._cache.get("XAU/USD_1h")
|
||
assert cached is df
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
# MODULE 3: engine.py
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
|
||
print("\n═══ Module 3: engine.py ═══")
|
||
|
||
from engine import TechnicalEngine, TechnicalAnalysis, IndicatorSignal
|
||
|
||
|
||
@test("TechnicalEngine.analyze returns TechnicalAnalysis with insufficient data")
|
||
def _():
|
||
engine = TechnicalEngine(TechnicalConfig())
|
||
df = make_ohlcv(10)
|
||
result = engine.analyze(df, "XAU/USD", "1h")
|
||
assert isinstance(result, TechnicalAnalysis)
|
||
assert result.confidence == 0.0
|
||
|
||
|
||
@test("TechnicalEngine.analyze returns TechnicalAnalysis with None df")
|
||
def _():
|
||
engine = TechnicalEngine(TechnicalConfig())
|
||
result = engine.analyze(None, "XAU/USD", "1h")
|
||
assert isinstance(result, TechnicalAnalysis)
|
||
assert result.confidence == 0.0
|
||
|
||
|
||
@test("TechnicalEngine.analyze returns valid result with good data")
|
||
def _():
|
||
engine = TechnicalEngine(TechnicalConfig())
|
||
df = make_ohlcv(300)
|
||
result = engine.analyze(df, "XAU/USD", "1h")
|
||
assert isinstance(result, TechnicalAnalysis)
|
||
assert result.symbol == "XAU/USD"
|
||
assert result.timeframe == "1h"
|
||
assert result.composite_signal in list(SignalStrength)
|
||
assert -100 <= result.composite_score <= 100
|
||
assert 0 <= result.confidence <= 100
|
||
|
||
|
||
@test("TechnicalEngine EMA-4 mode computes all 4 EMAs")
|
||
def _():
|
||
config = TechnicalConfig(ema_mode="4ema")
|
||
engine = TechnicalEngine(config)
|
||
df = make_ohlcv(300)
|
||
result = engine.analyze(df, "XAU/USD", "1h")
|
||
assert "ema_trend" in result.indicators
|
||
ema_ind = result.indicators["ema_trend"]
|
||
assert isinstance(ema_ind, IndicatorSignal)
|
||
assert ema_ind.name == "EMA Trend"
|
||
assert ema_ind.signal in list(SignalStrength)
|
||
assert "ema_fast" in result.key_levels
|
||
assert "ema_medium" in result.key_levels
|
||
assert "ema_slow" in result.key_levels
|
||
|
||
|
||
@test("TechnicalEngine EMA-2 mode computes EMA50/200")
|
||
def _():
|
||
config = TechnicalConfig(ema_mode="2ema")
|
||
engine = TechnicalEngine(config)
|
||
df = make_ohlcv(300)
|
||
result = engine.analyze(df, "XAU/USD", "1h")
|
||
assert "ema_trend" in result.indicators
|
||
assert "ema_trend" in result.key_levels or "ema_slow" in result.key_levels
|
||
|
||
|
||
@test("TechnicalEngine RSI computation")
|
||
def _():
|
||
engine = TechnicalEngine(TechnicalConfig())
|
||
df = make_ohlcv(300)
|
||
result = engine.analyze(df, "XAU/USD", "1h")
|
||
assert "rsi" in result.indicators
|
||
rsi = result.indicators["rsi"]
|
||
assert 0 <= rsi.value <= 100, f"RSI value {rsi.value} out of range"
|
||
assert rsi.signal in list(SignalStrength)
|
||
assert rsi.weight == 1.5
|
||
|
||
|
||
@test("TechnicalEngine RSI with NaN handling")
|
||
def _():
|
||
engine = TechnicalEngine(TechnicalConfig())
|
||
df = make_ohlcv(300)
|
||
df["Close"] = df["Close"].iloc[0] # flat price = 0 delta
|
||
result = engine.analyze(df, "XAU/USD", "1h")
|
||
rsi = result.indicators.get("rsi")
|
||
if rsi:
|
||
assert not pd.isna(rsi.value), "RSI should handle NaN gracefully"
|
||
|
||
|
||
@test("TechnicalEngine Bollinger Bands computation")
|
||
def _():
|
||
engine = TechnicalEngine(TechnicalConfig())
|
||
df = make_ohlcv(300)
|
||
result = engine.analyze(df, "XAU/USD", "1h")
|
||
assert "bollinger" in result.indicators
|
||
bb = result.indicators["bollinger"]
|
||
assert isinstance(bb.value, float)
|
||
assert "bb_upper" in result.key_levels
|
||
assert "bb_lower" in result.key_levels
|
||
assert "bb_mid" in result.key_levels
|
||
assert result.key_levels["bb_upper"] > result.key_levels["bb_lower"]
|
||
|
||
|
||
@test("TechnicalEngine ADX computation")
|
||
def _():
|
||
engine = TechnicalEngine(TechnicalConfig())
|
||
df = make_ohlcv(300)
|
||
result = engine.analyze(df, "XAU/USD", "1h")
|
||
assert "adx" in result.indicators
|
||
adx = result.indicators["adx"]
|
||
assert adx.value >= 0, f"ADX value {adx.value} should be >= 0"
|
||
assert "+DI" in adx.description
|
||
assert "-DI" in adx.description
|
||
|
||
|
||
@test("TechnicalEngine ATR computation and volatility regime")
|
||
def _():
|
||
engine = TechnicalEngine(TechnicalConfig())
|
||
df = make_ohlcv(300)
|
||
result = engine.analyze(df, "XAU/USD", "1h")
|
||
assert "atr" in result.indicators
|
||
atr = result.indicators["atr"]
|
||
assert atr.value > 0, "ATR should be positive"
|
||
assert atr.weight == 0.0, "ATR weight should be 0 (not directional)"
|
||
assert result.volatility_regime in ("low", "normal", "high", "extreme")
|
||
assert "atr" in result.key_levels
|
||
assert "atr_pct" in result.key_levels
|
||
|
||
|
||
@test("TechnicalEngine MACD as pillar (enabled by default)")
|
||
def _():
|
||
engine = TechnicalEngine(TechnicalConfig())
|
||
df = make_ohlcv(300)
|
||
result = engine.analyze(df, "XAU/USD", "1h")
|
||
assert "macd" in result.indicators
|
||
macd = result.indicators["macd"]
|
||
assert macd.weight == 1.5
|
||
|
||
@test("TechnicalEngine MACD not as pillar when disabled")
|
||
def _():
|
||
config = TechnicalConfig(macd_as_pillar=False)
|
||
engine = TechnicalEngine(config)
|
||
df = make_ohlcv(300)
|
||
result = engine.analyze(df, "XAU/USD", "1h")
|
||
assert "macd" not in result.indicators
|
||
assert hasattr(result, "_macd_histogram")
|
||
|
||
|
||
@test("TechnicalEngine disabled indicators are skipped")
|
||
def _():
|
||
config = TechnicalConfig(disabled_indicators=["rsi", "bollinger", "divergence"])
|
||
engine = TechnicalEngine(config)
|
||
df = make_ohlcv(300)
|
||
result = engine.analyze(df, "XAU/USD", "1h")
|
||
assert "rsi" not in result.indicators
|
||
assert "bollinger" not in result.indicators
|
||
|
||
|
||
@test("TechnicalEngine candlestick pattern detection")
|
||
def _():
|
||
engine = TechnicalEngine(TechnicalConfig())
|
||
df = make_ohlcv(300)
|
||
# Create a hammer pattern: long lower wick, small body, no upper wick
|
||
idx = -1
|
||
close_val = df["Close"].iloc[idx]
|
||
df.iloc[idx, df.columns.get_loc("Open")] = close_val - 0.0001
|
||
df.iloc[idx, df.columns.get_loc("Close")] = close_val
|
||
df.iloc[idx, df.columns.get_loc("High")] = close_val + 0.00005
|
||
df.iloc[idx, df.columns.get_loc("Low")] = close_val - 0.001
|
||
result = engine.analyze(df, "XAU/USD", "1h")
|
||
# Pattern may or may not be detected depending on threshold, just ensure no crash
|
||
|
||
|
||
@test("TechnicalEngine swing point detection")
|
||
def _():
|
||
engine = TechnicalEngine(TechnicalConfig())
|
||
series = pd.Series([10, 8, 6, 4, 6, 8, 10, 8, 6, 4, 6, 8, 10])
|
||
lows = engine._find_swing_points(series, mode="low", window=2)
|
||
highs = engine._find_swing_points(series, mode="high", window=2)
|
||
assert len(lows) > 0, "Should find swing lows in V-pattern data"
|
||
assert any(v == 4 for _, v in lows), "Should find value 4 as swing low"
|
||
assert len(highs) > 0, "Should find swing highs"
|
||
|
||
|
||
@test("TechnicalEngine swing points with NaN values")
|
||
def _():
|
||
engine = TechnicalEngine(TechnicalConfig())
|
||
series = pd.Series([10, 8, np.nan, 4, 6, 8, 10, 8, 6, 4, 6, 8, 10])
|
||
lows = engine._find_swing_points(series, mode="low", window=2)
|
||
# Should not crash
|
||
|
||
|
||
@test("TechnicalEngine market regime classification")
|
||
def _():
|
||
engine = TechnicalEngine(TechnicalConfig())
|
||
df = make_ohlcv(300, trend=0.001) # strong uptrend
|
||
result = engine.analyze(df, "XAU/USD", "1h")
|
||
assert result.market_regime in ("trending_up", "trending_down", "trending", "ranging")
|
||
|
||
|
||
@test("TechnicalEngine composite score within bounds")
|
||
def _():
|
||
engine = TechnicalEngine(TechnicalConfig())
|
||
df = make_ohlcv(300)
|
||
result = engine.analyze(df, "XAU/USD", "1h")
|
||
assert -100 <= result.composite_score <= 100
|
||
assert result.composite_signal in list(SignalStrength)
|
||
|
||
|
||
@test("TechnicalEngine regime-aware RSI reinterpretation")
|
||
def _():
|
||
config = TechnicalConfig(rsi_mode="regime_aware")
|
||
engine = TechnicalEngine(config)
|
||
df = make_ohlcv(300, trend=0.001)
|
||
result = engine.analyze(df, "XAU/USD", "1h")
|
||
rsi = result.indicators.get("rsi")
|
||
assert rsi is not None
|
||
|
||
|
||
@test("TechnicalEngine structure S/R levels detection")
|
||
def _():
|
||
engine = TechnicalEngine(TechnicalConfig())
|
||
df = make_ohlcv(300)
|
||
result = engine.analyze(df, "XAU/USD", "1h")
|
||
swing_keys = [k for k in result.key_levels if k.startswith("swing_")]
|
||
assert len(swing_keys) >= 0 # may or may not have swings
|
||
|
||
|
||
@test("TechnicalEngine divergence detection (enabled by default)")
|
||
def _():
|
||
engine = TechnicalEngine(TechnicalConfig())
|
||
df = make_ohlcv(300)
|
||
result = engine.analyze(df, "XAU/USD", "1h")
|
||
# divergence may or may not be detected depending on data, just ensure no crash
|
||
|
||
@test("TechnicalEngine divergence detection (disabled)")
|
||
def _():
|
||
config = TechnicalConfig(disabled_indicators=["divergence"])
|
||
engine = TechnicalEngine(config)
|
||
df = make_ohlcv(300)
|
||
result = engine.analyze(df, "XAU/USD", "1h")
|
||
assert "divergence" not in result.indicators
|
||
|
||
|
||
@test("TechnicalEngine EMA-2 golden cross detection")
|
||
def _():
|
||
config = TechnicalConfig(ema_mode="2ema")
|
||
engine = TechnicalEngine(config)
|
||
df = make_ohlcv(300, trend=0.002)
|
||
result = engine.analyze(df, "XAU/USD", "1h")
|
||
assert "ema_trend" in result.indicators
|
||
|
||
|
||
@test("TechnicalEngine with very short data")
|
||
def _():
|
||
engine = TechnicalEngine(TechnicalConfig())
|
||
df = make_ohlcv(31)
|
||
result = engine.analyze(df, "XAU/USD", "1h")
|
||
assert isinstance(result, TechnicalAnalysis)
|
||
|
||
|
||
@test("TechnicalEngine ROC computation")
|
||
def _():
|
||
engine = TechnicalEngine(TechnicalConfig())
|
||
df = make_ohlcv(300, trend=0.002)
|
||
result = engine.analyze(df, "XAU/USD", "1h")
|
||
assert "roc" in result.indicators
|
||
roc = result.indicators["roc"]
|
||
assert roc.weight == 1.5
|
||
assert roc.signal in list(SignalStrength)
|
||
|
||
|
||
@test("TechnicalEngine Keltner Channels computation")
|
||
def _():
|
||
engine = TechnicalEngine(TechnicalConfig())
|
||
df = make_ohlcv(300)
|
||
result = engine.analyze(df, "XAU/USD", "1h")
|
||
assert "keltner" in result.indicators
|
||
kc = result.indicators["keltner"]
|
||
assert kc.weight == 1.5
|
||
assert "kc_upper" in result.key_levels
|
||
assert "kc_lower" in result.key_levels
|
||
assert "kc_mid" in result.key_levels
|
||
assert result.key_levels["kc_upper"] > result.key_levels["kc_lower"]
|
||
|
||
|
||
@test("TechnicalEngine ADX slope computation")
|
||
def _():
|
||
engine = TechnicalEngine(TechnicalConfig())
|
||
df = make_ohlcv(300, trend=0.002)
|
||
result = engine.analyze(df, "XAU/USD", "1h")
|
||
assert "adx" in result.indicators
|
||
assert hasattr(result, "_adx_slope")
|
||
|
||
|
||
@test("TechnicalEngine transitioning regime detection")
|
||
def _():
|
||
engine = TechnicalEngine(TechnicalConfig())
|
||
df = make_ohlcv(300)
|
||
result = engine.analyze(df, "XAU/USD", "1h")
|
||
assert result.market_regime in (
|
||
"trending_up", "trending_down", "trending", "ranging", "transitioning"
|
||
)
|
||
|
||
|
||
@test("TechnicalEngine gold-specific composite weights")
|
||
def _():
|
||
engine = TechnicalEngine(TechnicalConfig())
|
||
df = make_xau_ohlcv(300)
|
||
result = engine.analyze(df, "XAU/USD", "1h")
|
||
assert -100 <= result.composite_score <= 100
|
||
assert result.composite_signal in list(SignalStrength)
|
||
|
||
|
||
@test("TechnicalEngine EMA ribbon compression detection")
|
||
def _():
|
||
engine = TechnicalEngine(TechnicalConfig())
|
||
df = make_ohlcv(300)
|
||
result = engine.analyze(df, "XAU/USD", "1h")
|
||
assert "ema_trend" in result.indicators
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
# MODULE 4: news_engine.py
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
|
||
print("\n═══ Module 4: news_engine.py ═══")
|
||
|
||
from news_engine import NewsEngine, FundamentalContext, EconomicEvent, NewsSentiment
|
||
|
||
|
||
@test("NewsEngine._parse_pair with slash")
|
||
def _():
|
||
base, quote = NewsEngine._parse_pair("XAU/USD")
|
||
assert base == "XAU"
|
||
assert quote == "USD"
|
||
|
||
|
||
@test("NewsEngine._parse_pair with =X suffix")
|
||
def _():
|
||
base, quote = NewsEngine._parse_pair("EURUSD=X")
|
||
assert base == "EUR"
|
||
assert quote == "USD"
|
||
|
||
|
||
@test("NewsEngine._parse_pair for XAU pair")
|
||
def _():
|
||
base, quote = NewsEngine._parse_pair("XAU/USD")
|
||
assert base == "XAU"
|
||
assert quote == "USD"
|
||
|
||
|
||
@test("NewsEngine._analyze_headline_sentiment bullish")
|
||
def _():
|
||
ne = NewsEngine(NewsConfig())
|
||
score = ne._analyze_headline_sentiment("Fed hawkish rate hike boosts dollar rally")
|
||
assert score > 0, f"Expected positive sentiment, got {score}"
|
||
|
||
|
||
@test("NewsEngine._analyze_headline_sentiment bearish")
|
||
def _():
|
||
ne = NewsEngine(NewsConfig())
|
||
score = ne._analyze_headline_sentiment("Recession fears drive dovish rate cut expectations")
|
||
assert score < 0, f"Expected negative sentiment, got {score}"
|
||
|
||
|
||
@test("NewsEngine._analyze_headline_sentiment neutral")
|
||
def _():
|
||
ne = NewsEngine(NewsConfig())
|
||
score = ne._analyze_headline_sentiment("Markets remain quiet ahead of data release")
|
||
assert score == 0.0
|
||
|
||
|
||
@test("NewsEngine._extract_currencies USD")
|
||
def _():
|
||
ne = NewsEngine(NewsConfig())
|
||
currencies = ne._extract_currencies("Federal Reserve raises interest rates, dollar soars")
|
||
assert "USD" in currencies
|
||
|
||
|
||
@test("NewsEngine._extract_currencies EUR")
|
||
def _():
|
||
ne = NewsEngine(NewsConfig())
|
||
currencies = ne._extract_currencies("ECB meeting pushes euro higher")
|
||
assert "EUR" in currencies
|
||
|
||
|
||
@test("NewsEngine._extract_currencies GBP")
|
||
def _():
|
||
ne = NewsEngine(NewsConfig())
|
||
currencies = ne._extract_currencies("Bank of England holds rates, pound steady")
|
||
assert "GBP" in currencies
|
||
|
||
|
||
@test("NewsEngine._extract_currencies XAU")
|
||
def _():
|
||
ne = NewsEngine(NewsConfig())
|
||
currencies = ne._extract_currencies("Gold prices surge as safe haven demand rises")
|
||
assert "XAU" in currencies
|
||
|
||
|
||
@test("NewsEngine._extract_currencies multiple")
|
||
def _():
|
||
ne = NewsEngine(NewsConfig())
|
||
currencies = ne._extract_currencies("Gold drops as Fed hawkish, dollar surges")
|
||
assert "USD" in currencies
|
||
assert "XAU" in currencies
|
||
|
||
|
||
@test("NewsEngine._extract_keywords")
|
||
def _():
|
||
ne = NewsEngine(NewsConfig())
|
||
kws = ne._extract_keywords("Hawkish Fed rate hike surprises with positive surprise")
|
||
assert "rate hike" in kws
|
||
assert "hawkish" in kws
|
||
assert "positive surprise" in kws
|
||
|
||
|
||
@test("NewsEngine._parse_rss_date RFC822 format")
|
||
def _():
|
||
dt = NewsEngine._parse_rss_date("Mon, 12 May 2025 14:30:00 +0000")
|
||
assert dt is not None
|
||
assert dt.year == 2025
|
||
assert dt.month == 5
|
||
assert dt.day == 12
|
||
assert dt.tzinfo is not None
|
||
|
||
|
||
@test("NewsEngine._parse_rss_date ISO format")
|
||
def _():
|
||
dt = NewsEngine._parse_rss_date("2025-05-12T14:30:00+00:00")
|
||
assert dt is not None
|
||
assert dt.year == 2025
|
||
|
||
|
||
@test("NewsEngine._parse_rss_date naive format assumes UTC")
|
||
def _():
|
||
dt = NewsEngine._parse_rss_date("2025-05-12 14:30:00")
|
||
assert dt is not None
|
||
assert dt.tzinfo == timezone.utc
|
||
|
||
|
||
@test("NewsEngine._parse_rss_date None input")
|
||
def _():
|
||
assert NewsEngine._parse_rss_date(None) is None
|
||
|
||
|
||
@test("NewsEngine._parse_rss_date garbage input")
|
||
def _():
|
||
assert NewsEngine._parse_rss_date("not-a-date") is None
|
||
|
||
|
||
@test("NewsEngine._get_affected_pairs")
|
||
def _():
|
||
pairs = NewsEngine._get_affected_pairs("USD")
|
||
assert "XAU/USD" in pairs
|
||
|
||
pairs_xau = NewsEngine._get_affected_pairs("XAU")
|
||
assert "XAU/USD" in pairs_xau
|
||
|
||
pairs_unknown = NewsEngine._get_affected_pairs("ARS")
|
||
assert pairs_unknown == []
|
||
|
||
|
||
@test("EconomicEvent properties")
|
||
def _():
|
||
future_time = datetime.now(timezone.utc) + timedelta(minutes=30)
|
||
event = EconomicEvent(
|
||
time=future_time, currency="USD",
|
||
impact="high", event="NFP",
|
||
)
|
||
assert event.is_upcoming is True
|
||
assert 25 < event.minutes_until < 35
|
||
assert event.is_high_impact is True
|
||
|
||
past_time = datetime.now(timezone.utc) - timedelta(minutes=10)
|
||
past_event = EconomicEvent(
|
||
time=past_time, currency="EUR",
|
||
impact="low", event="CPI",
|
||
)
|
||
assert past_event.is_upcoming is False
|
||
assert past_event.is_high_impact is False
|
||
|
||
|
||
@test("NewsEngine.get_event_risk_level danger zone")
|
||
def _():
|
||
ne = NewsEngine(NewsConfig())
|
||
context = FundamentalContext()
|
||
context.events_upcoming_1h = [
|
||
EconomicEvent(
|
||
time=datetime.now(timezone.utc) + timedelta(minutes=15),
|
||
currency="USD", impact="high", event="NFP",
|
||
)
|
||
]
|
||
risk, desc = ne.get_event_risk_level("XAU/USD", context)
|
||
assert risk == "danger"
|
||
assert "NFP" in desc
|
||
|
||
|
||
@test("NewsEngine.get_event_risk_level caution zone")
|
||
def _():
|
||
ne = NewsEngine(NewsConfig())
|
||
context = FundamentalContext()
|
||
context.events_upcoming_1h = [
|
||
EconomicEvent(
|
||
time=datetime.now(timezone.utc) + timedelta(minutes=45),
|
||
currency="USD", impact="high", event="GDP",
|
||
)
|
||
]
|
||
risk, desc = ne.get_event_risk_level("XAU/USD", context)
|
||
assert risk == "caution"
|
||
|
||
|
||
@test("NewsEngine.get_event_risk_level clear when no events")
|
||
def _():
|
||
ne = NewsEngine(NewsConfig())
|
||
context = FundamentalContext()
|
||
risk, desc = ne.get_event_risk_level("XAU/USD", context)
|
||
assert risk == "clear"
|
||
|
||
|
||
@test("NewsEngine.get_event_risk_level ignores unrelated currency events")
|
||
def _():
|
||
ne = NewsEngine(NewsConfig())
|
||
context = FundamentalContext()
|
||
context.events_upcoming_1h = [
|
||
EconomicEvent(
|
||
time=datetime.now(timezone.utc) + timedelta(minutes=15),
|
||
currency="AUD", impact="high", event="RBA Rate",
|
||
)
|
||
]
|
||
risk, _ = ne.get_event_risk_level("XAU/USD", context)
|
||
assert risk == "clear"
|
||
|
||
|
||
@test("NewsEngine.get_pair_sentiment with positive base sentiment")
|
||
def _():
|
||
ne = NewsEngine(NewsConfig())
|
||
context = FundamentalContext()
|
||
context.news_sentiment = {"EUR": 0.5, "USD": -0.1}
|
||
score, desc = ne.get_pair_sentiment("XAU/USD", context)
|
||
assert score > 0, "EUR bullish + USD bearish should be net bullish for EUR/USD"
|
||
assert "bullish" in desc.lower()
|
||
|
||
|
||
@test("NewsEngine.get_pair_sentiment with negative pair sentiment")
|
||
def _():
|
||
ne = NewsEngine(NewsConfig())
|
||
context = FundamentalContext()
|
||
context.news_sentiment = {"EUR": -0.5, "USD": 0.3}
|
||
score, desc = ne.get_pair_sentiment("XAU/USD", context)
|
||
assert score < 0
|
||
|
||
|
||
@test("NewsEngine.get_pair_sentiment neutral")
|
||
def _():
|
||
ne = NewsEngine(NewsConfig())
|
||
context = FundamentalContext()
|
||
context.news_sentiment = {"EUR": 0.0, "USD": 0.0}
|
||
score, desc = ne.get_pair_sentiment("XAU/USD", context)
|
||
assert score == 0.0
|
||
assert "neutral" in desc.lower()
|
||
|
||
|
||
@test("FundamentalContext defaults")
|
||
def _():
|
||
ctx = FundamentalContext()
|
||
assert ctx.events_today == []
|
||
assert ctx.events_upcoming_1h == []
|
||
assert ctx.high_impact_warning is False
|
||
assert ctx.calendar_available is False
|
||
assert ctx.news_sentiment == {}
|
||
assert ctx.news_headlines == []
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
# MODULE 5: signal_aggregator.py
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
|
||
print("\n═══ Module 5: signal_aggregator.py ═══")
|
||
|
||
from signal_aggregator import SignalAggregator, SignalReport, TradeSetup
|
||
|
||
|
||
@test("SignalReport defaults")
|
||
def _():
|
||
r = SignalReport(symbol="XAU/USD", display_name="XAU/USD", timestamp="now")
|
||
assert r.final_signal == SignalStrength.NEUTRAL
|
||
assert r.final_score == 0.0
|
||
assert r.primary_signal == SignalStrength.NEUTRAL
|
||
assert r.trade_setup is None
|
||
assert r.mtf_agreement is False
|
||
assert r.signal_quality == ""
|
||
assert r.bullish_reasons == []
|
||
assert r.bearish_reasons == []
|
||
assert r.warnings == []
|
||
|
||
|
||
@test("TradeSetup dataclass creation")
|
||
def _():
|
||
ts = TradeSetup(
|
||
pair="XAU/USD", display_name="XAU/USD", direction="BUY",
|
||
entry_price=1.1000, stop_loss=1.0950, take_profit_1=1.1100,
|
||
take_profit_2=1.1200, risk_reward=2.0, risk_pips=50, reward_pips=100,
|
||
pip_value=0.0001,
|
||
)
|
||
assert ts.direction == "BUY"
|
||
assert ts.risk_reward == 2.0
|
||
assert ts.entry_price > ts.stop_loss
|
||
|
||
|
||
@test("SignalAggregator._check_mtf_agreement with 2 bullish")
|
||
def _():
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
report = SignalReport(symbol="XAU/USD", display_name="XAU/USD", timestamp="now")
|
||
report.mtf_signals = {
|
||
"15min": SignalStrength.BUY,
|
||
"1h": SignalStrength.STRONG_BUY,
|
||
"4h": SignalStrength.NEUTRAL,
|
||
}
|
||
assert agg._check_mtf_agreement(report) is True
|
||
|
||
|
||
@test("SignalAggregator._check_mtf_agreement with mixed signals")
|
||
def _():
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
report = SignalReport(symbol="XAU/USD", display_name="XAU/USD", timestamp="now")
|
||
report.mtf_signals = {
|
||
"15min": SignalStrength.BUY,
|
||
"1h": SignalStrength.SELL,
|
||
"4h": SignalStrength.NEUTRAL,
|
||
}
|
||
assert agg._check_mtf_agreement(report) is False
|
||
|
||
|
||
@test("SignalAggregator._check_mtf_agreement with single signal")
|
||
def _():
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
report = SignalReport(symbol="XAU/USD", display_name="XAU/USD", timestamp="now")
|
||
report.mtf_signals = {"1h": SignalStrength.BUY}
|
||
assert agg._check_mtf_agreement(report) is False
|
||
|
||
|
||
@test("SignalAggregator._latest_adx computes valid value")
|
||
def _():
|
||
df = make_ohlcv(300)
|
||
adx = SignalAggregator._latest_adx(df)
|
||
assert adx is not None
|
||
assert adx >= 0
|
||
|
||
|
||
@test("SignalAggregator._latest_adx with insufficient data")
|
||
def _():
|
||
df = make_ohlcv(10)
|
||
adx = SignalAggregator._latest_adx(df)
|
||
assert adx is None
|
||
|
||
|
||
@test("SignalAggregator.analyze_from_dataframes produces valid report")
|
||
def _():
|
||
config = PlatformConfig()
|
||
config.primary_timeframe = Timeframe.H1
|
||
config.confirmation_timeframes = [Timeframe.M15, Timeframe.H4]
|
||
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),
|
||
}
|
||
report = agg.analyze_from_dataframes("XAU/USD", dfs, use_news=False)
|
||
assert isinstance(report, SignalReport)
|
||
assert report.symbol == "XAU/USD"
|
||
assert report.display_name == "XAU/USD"
|
||
assert report.final_signal in list(SignalStrength)
|
||
assert report.signal_quality in ("A+", "A", "B", "C", "D", "F")
|
||
assert report.current_price > 0
|
||
|
||
|
||
@test("SignalAggregator._compute_final_signal quality grades")
|
||
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),
|
||
}
|
||
report = agg.analyze_from_dataframes("XAU/USD", dfs, use_news=False)
|
||
assert report.signal_quality in ("A+", "A", "B", "C", "D", "F")
|
||
assert report.action_recommendation != ""
|
||
|
||
|
||
@test("SignalAggregator._generate_trade_setup BUY direction")
|
||
def _():
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
dfs = {
|
||
Timeframe.H1: make_ohlcv(300, trend=0.003, seed=42),
|
||
Timeframe.M15: make_ohlcv(300, trend=0.003, seed=43),
|
||
Timeframe.H4: make_ohlcv(300, trend=0.003, seed=44),
|
||
}
|
||
report = agg.analyze_from_dataframes("XAU/USD", dfs, use_news=False)
|
||
if report.trade_setup:
|
||
ts = report.trade_setup
|
||
if ts.direction == "BUY":
|
||
assert ts.stop_loss < ts.entry_price, "BUY SL should be below entry"
|
||
assert ts.take_profit_1 > ts.entry_price, "BUY TP should be above entry"
|
||
assert ts.risk_reward >= config.signals.min_rr_ratio
|
||
|
||
|
||
@test("SignalAggregator._collect_reasons populates reasons")
|
||
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),
|
||
}
|
||
report = agg.analyze_from_dataframes("XAU/USD", dfs, use_news=False)
|
||
total_reasons = len(report.bullish_reasons) + len(report.bearish_reasons)
|
||
assert total_reasons > 0, "Should have at least one reason"
|
||
|
||
|
||
@test("SignalAggregator session filter blocks outside hours")
|
||
def _():
|
||
config = PlatformConfig()
|
||
config.backtest.session_filter = True
|
||
config.xau_strategy.blocked_hours_utc = list(range(24))
|
||
config.xau_strategy.enabled = False
|
||
agg = SignalAggregator(config)
|
||
dfs = {
|
||
Timeframe.H1: make_xau_ohlcv(300, seed=42),
|
||
Timeframe.M15: make_xau_ohlcv(300, seed=43),
|
||
Timeframe.H4: make_xau_ohlcv(300, seed=44),
|
||
}
|
||
report = agg.analyze_from_dataframes("XAU/USD", dfs, use_news=False)
|
||
assert report.signal_quality == "F"
|
||
|
||
|
||
@test("SignalAggregator with missing primary timeframe")
|
||
def _():
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
dfs = {
|
||
Timeframe.M15: make_ohlcv(300, seed=43),
|
||
}
|
||
report = agg.analyze_from_dataframes("XAU/USD", dfs, use_news=False)
|
||
assert report.signal_quality == "F"
|
||
assert any("unavailable" in w.lower() for w in report.warnings)
|
||
|
||
|
||
@test("SignalAggregator XAU strategy check (conditions not met)")
|
||
def _():
|
||
config = PlatformConfig()
|
||
config.xau_strategy.enabled = True
|
||
agg = SignalAggregator(config)
|
||
dfs = {
|
||
Timeframe.H1: make_xau_ohlcv(300, seed=42),
|
||
Timeframe.M15: make_xau_ohlcv(300, seed=43),
|
||
Timeframe.H4: make_xau_ohlcv(300, seed=44),
|
||
}
|
||
report = agg.analyze_from_dataframes("XAU/USD", dfs, use_news=False)
|
||
assert isinstance(report, SignalReport)
|
||
|
||
|
||
@test("SignalAggregator event risk affects confidence")
|
||
def _():
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
context = FundamentalContext()
|
||
context.events_upcoming_1h = [
|
||
EconomicEvent(
|
||
time=datetime.now(timezone.utc) + timedelta(minutes=10),
|
||
currency="USD", impact="high", event="NFP",
|
||
)
|
||
]
|
||
dfs = {
|
||
Timeframe.H1: make_ohlcv(300, seed=42),
|
||
Timeframe.M15: make_ohlcv(300, seed=43),
|
||
Timeframe.H4: make_ohlcv(300, seed=44),
|
||
}
|
||
report = agg.analyze_from_dataframes("XAU/USD", dfs, context=context, use_news=True)
|
||
assert any("EVENT RISK" in w or "event" in w.lower() for w in report.warnings)
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
# MODULE 6: main.py
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
|
||
print("\n═══ Module 6: main.py ═══")
|
||
|
||
from main import (
|
||
signal_color, quality_color, _compact_reason,
|
||
report_to_dict, build_config, Colors,
|
||
)
|
||
|
||
|
||
@test("signal_color maps all SignalStrength values")
|
||
def _():
|
||
for sig in SignalStrength:
|
||
color = signal_color(sig)
|
||
assert isinstance(color, str)
|
||
|
||
|
||
@test("quality_color maps all quality grades")
|
||
def _():
|
||
for q in ("A+", "A", "B", "C", "D", "F"):
|
||
color = quality_color(q)
|
||
assert isinstance(color, str)
|
||
|
||
|
||
@test("_compact_reason strips STRONG prefix")
|
||
def _():
|
||
result = _compact_reason("STRONG: EMA Trend: Full bullish alignment")
|
||
assert not result.startswith("STRONG: ")
|
||
|
||
|
||
@test("_compact_reason strips indicator prefix")
|
||
def _():
|
||
result = _compact_reason("RSI: RSI 45.3 — Neutral")
|
||
assert not result.startswith("RSI: ")
|
||
|
||
|
||
@test("_compact_reason strips detail after pipe")
|
||
def _():
|
||
result = _compact_reason("EMA alignment: 3/4 | Golden cross (fast/medium)")
|
||
assert "|" not in result
|
||
|
||
|
||
@test("_compact_reason handles plain text")
|
||
def _():
|
||
result = _compact_reason("Simple reason text")
|
||
assert result == "Simple reason text"
|
||
|
||
|
||
@test("report_to_dict basic conversion")
|
||
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),
|
||
}
|
||
report = agg.analyze_from_dataframes("XAU/USD", dfs, use_news=False)
|
||
d = report_to_dict(report)
|
||
assert "symbol" in d
|
||
assert "signal" in d
|
||
assert "score" in d
|
||
assert "confidence" in d
|
||
assert "quality" in d
|
||
assert "action" in d
|
||
assert "market_regime" in d
|
||
assert "volatility_regime" in d
|
||
assert "mtf_agreement" in d
|
||
assert "mtf_signals" in d
|
||
assert d["symbol"] == "XAU/USD"
|
||
assert d["display_name"] == "XAU/USD"
|
||
|
||
|
||
@test("report_to_dict includes trade_setup when present")
|
||
def _():
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
dfs = {
|
||
Timeframe.H1: make_ohlcv(300, trend=0.003, seed=42),
|
||
Timeframe.M15: make_ohlcv(300, trend=0.003, seed=43),
|
||
Timeframe.H4: make_ohlcv(300, trend=0.003, seed=44),
|
||
}
|
||
report = agg.analyze_from_dataframes("XAU/USD", dfs, use_news=False)
|
||
d = report_to_dict(report)
|
||
if report.trade_setup:
|
||
assert "trade_setup" in d
|
||
ts = d["trade_setup"]
|
||
assert "direction" in ts
|
||
assert "entry" in ts
|
||
assert "stop_loss" in ts
|
||
assert "take_profit_1" in ts
|
||
assert "risk_reward" in ts
|
||
|
||
|
||
@test("report_to_dict includes indicators")
|
||
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),
|
||
}
|
||
report = agg.analyze_from_dataframes("XAU/USD", dfs, use_news=False)
|
||
d = report_to_dict(report)
|
||
assert "indicators" in d
|
||
assert "key_levels" in d
|
||
for name, ind_dict in d["indicators"].items():
|
||
assert "signal" in ind_dict
|
||
assert "confidence" in ind_dict
|
||
assert "description" in ind_dict
|
||
|
||
|
||
@test("build_config sets correct timeframe and confirmations")
|
||
def _():
|
||
class FakeArgs:
|
||
timeframe = "H4"
|
||
config = build_config(FakeArgs())
|
||
assert config.primary_timeframe == Timeframe.H4
|
||
assert Timeframe.H1 in config.confirmation_timeframes
|
||
assert Timeframe.D1 in config.confirmation_timeframes
|
||
|
||
|
||
@test("build_config all valid timeframes")
|
||
def _():
|
||
for tf_str in ["M5", "M15", "M30", "H1", "H4", "D1"]:
|
||
class FakeArgs:
|
||
timeframe = tf_str
|
||
config = build_config(FakeArgs())
|
||
assert config.primary_timeframe is not None
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
# MODULE 7: testing_telegram_bot.py
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
|
||
print("\n═══ Module 7: testing_telegram_bot.py ═══")
|
||
|
||
from testing_telegram_bot import (
|
||
_escape_md, _signal_label, _mtf_line,
|
||
format_signal_message, _format_scan_summary,
|
||
_seconds_until_boundary, _check_level_breach,
|
||
_format_stats_dashboard,
|
||
scan_keyboard, SCAN_BUTTONS, SCAN_LOOKUP,
|
||
_update_level_cache, _level_cache, _level_cache_ts,
|
||
)
|
||
|
||
|
||
@test("_escape_md escapes markdown characters")
|
||
def _():
|
||
assert _escape_md("hello *world*") == "hello \\*world\\*"
|
||
assert _escape_md("test_value") == "test\\_value"
|
||
assert _escape_md("`code`") == "\\`code\\`"
|
||
assert _escape_md("[link]") == "\\[link\\]"
|
||
|
||
|
||
@test("_signal_label formatting")
|
||
def _():
|
||
assert _signal_label(SignalStrength.STRONG_BUY) == "STRONG BUY"
|
||
assert _signal_label(SignalStrength.STRONG_SELL) == "STRONG SELL"
|
||
assert _signal_label(SignalStrength.NEUTRAL) == "NEUTRAL"
|
||
assert _signal_label(SignalStrength.BUY) == "BUY"
|
||
assert _signal_label(SignalStrength.SELL) == "SELL"
|
||
|
||
|
||
@test("_mtf_line formatting with signals")
|
||
def _():
|
||
report = SignalReport(symbol="XAU/USD", display_name="XAU/USD", timestamp="now")
|
||
report.mtf_signals = {
|
||
"15min": SignalStrength.BUY,
|
||
"1h": SignalStrength.NEUTRAL,
|
||
}
|
||
report.mtf_scores = {"15min": 25.0, "1h": 5.0}
|
||
line = _mtf_line(report)
|
||
assert "15min" in line
|
||
assert "1h" in line
|
||
assert "BUY" in line
|
||
|
||
|
||
@test("_mtf_line empty when no signals")
|
||
def _():
|
||
report = SignalReport(symbol="XAU/USD", display_name="XAU/USD", timestamp="now")
|
||
assert _mtf_line(report) == ""
|
||
|
||
|
||
@test("format_signal_message 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),
|
||
}
|
||
report = agg.analyze_from_dataframes("XAU/USD", dfs, use_news=False)
|
||
msg = format_signal_message(report)
|
||
assert isinstance(msg, str)
|
||
assert len(msg) > 0
|
||
assert "XAU/USD" in msg
|
||
assert "Signal Analysis" in msg
|
||
assert "Price" in msg
|
||
assert "Indicators" 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)
|
||
assert result > 0
|
||
assert result <= 15 * 60 + 10 + 1
|
||
|
||
|
||
@test("_seconds_until_boundary for M5 interval")
|
||
def _():
|
||
result = _seconds_until_boundary(5, buffer=10)
|
||
assert result > 0
|
||
assert result <= 5 * 60 + 10 + 1
|
||
|
||
|
||
@test("_seconds_until_boundary for H1 interval")
|
||
def _():
|
||
result = _seconds_until_boundary(60, buffer=10)
|
||
assert result > 0
|
||
assert result <= 60 * 60 + 10 + 1
|
||
|
||
|
||
@test("_seconds_until_boundary for H4 interval")
|
||
def _():
|
||
result = _seconds_until_boundary(240, buffer=10)
|
||
assert result > 0
|
||
assert result <= 240 * 60 + 10 + 1
|
||
|
||
|
||
@test("_check_level_breach detects price near level")
|
||
def _():
|
||
_level_cache["TEST_PAIR"] = {"swing_high_1": 1.1000}
|
||
_level_cache_ts["TEST_PAIR"] = time.time()
|
||
result = _check_level_breach("TEST_PAIR", 1.10005)
|
||
assert result == "swing_high_1"
|
||
del _level_cache["TEST_PAIR"]
|
||
del _level_cache_ts["TEST_PAIR"]
|
||
|
||
|
||
@test("_check_level_breach detects price at exact level")
|
||
def _():
|
||
_level_cache["TEST_PAIR"] = {"swing_low_1": 1.0900}
|
||
_level_cache_ts["TEST_PAIR"] = time.time()
|
||
result = _check_level_breach("TEST_PAIR", 1.0900)
|
||
assert result == "swing_low_1"
|
||
del _level_cache["TEST_PAIR"]
|
||
del _level_cache_ts["TEST_PAIR"]
|
||
|
||
|
||
@test("_check_level_breach no breach when price is far from level")
|
||
def _():
|
||
_level_cache["TEST_PAIR"] = {"swing_high_1": 1.1000}
|
||
_level_cache_ts["TEST_PAIR"] = time.time()
|
||
result = _check_level_breach("TEST_PAIR", 1.0950)
|
||
assert result is None
|
||
del _level_cache["TEST_PAIR"]
|
||
del _level_cache_ts["TEST_PAIR"]
|
||
|
||
|
||
@test("_check_level_breach returns None for expired cache")
|
||
def _():
|
||
_level_cache["TEST_PAIR"] = {"swing_high_1": 1.1000}
|
||
_level_cache_ts["TEST_PAIR"] = time.time() - 7200 # 2h old
|
||
result = _check_level_breach("TEST_PAIR", 1.1000)
|
||
assert result is None
|
||
del _level_cache["TEST_PAIR"]
|
||
del _level_cache_ts["TEST_PAIR"]
|
||
|
||
|
||
@test("_check_level_breach returns None for empty cache")
|
||
def _():
|
||
result = _check_level_breach("NONEXISTENT_PAIR", 1.1000)
|
||
assert result is None
|
||
|
||
|
||
@test("SCAN_BUTTONS and SCAN_LOOKUP consistency")
|
||
def _():
|
||
assert len(SCAN_BUTTONS) == 10 # 5 per pair (4 TFs + All) × 2 pairs
|
||
for btn in SCAN_BUTTONS:
|
||
assert btn["cb"] in SCAN_LOOKUP
|
||
assert "symbol" in btn
|
||
assert "tf" in btn
|
||
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()
|
||
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),
|
||
}
|
||
report = agg.analyze_from_dataframes("XAU/USD", dfs, use_news=False)
|
||
_level_cache.clear()
|
||
_level_cache_ts.clear()
|
||
_update_level_cache([report])
|
||
if report.primary_analysis and report.primary_analysis.key_levels:
|
||
assert "XAU/USD" in _level_cache or len(report.primary_analysis.key_levels) == 0
|
||
_level_cache.clear()
|
||
_level_cache_ts.clear()
|
||
|
||
|
||
@test("_format_stats_dashboard returns waiting message when empty")
|
||
def _():
|
||
import testing_telegram_bot
|
||
old_stats = testing_telegram_bot._signal_stats.copy()
|
||
testing_telegram_bot._signal_stats.clear()
|
||
text = _format_stats_dashboard()
|
||
assert "Waiting" in text or "waiting" in text.lower()
|
||
testing_telegram_bot._signal_stats.update(old_stats)
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
# MODULE 8: Integration / Cross-module tests
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
|
||
print("\n═══ Module 8: Integration Tests ═══")
|
||
|
||
|
||
@test("Full pipeline: config → engine → aggregator → report → dict")
|
||
def _():
|
||
config = PlatformConfig()
|
||
engine = TechnicalEngine(config.technical)
|
||
df = make_ohlcv(300, seed=99)
|
||
analysis = engine.analyze(df, "XAU/USD", "1h")
|
||
assert isinstance(analysis, TechnicalAnalysis)
|
||
|
||
agg = SignalAggregator(config)
|
||
dfs = {
|
||
Timeframe.H1: make_ohlcv(300, seed=99),
|
||
Timeframe.M15: make_ohlcv(300, seed=100),
|
||
Timeframe.H4: make_ohlcv(300, seed=101),
|
||
}
|
||
report = agg.analyze_from_dataframes("XAU/USD", dfs, use_news=False)
|
||
d = report_to_dict(report)
|
||
assert isinstance(d, dict)
|
||
assert d["symbol"] == "XAU/USD"
|
||
|
||
|
||
@test("Full pipeline: XAU/USD analysis")
|
||
def _():
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
dfs = {
|
||
Timeframe.H1: make_xau_ohlcv(300, seed=42),
|
||
Timeframe.M15: make_xau_ohlcv(300, seed=43),
|
||
Timeframe.H4: make_xau_ohlcv(300, seed=44),
|
||
}
|
||
report = agg.analyze_from_dataframes("XAU/USD", dfs, use_news=False)
|
||
assert report.display_name == "XAU/USD"
|
||
assert report.current_price > 1000 # XAU should be in the thousands
|
||
|
||
|
||
@test("Full pipeline: format Telegram message from report")
|
||
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),
|
||
}
|
||
report = agg.analyze_from_dataframes("XAU/USD", dfs, use_news=False)
|
||
msg = format_signal_message(report)
|
||
assert isinstance(msg, str)
|
||
assert len(msg) > 100
|
||
|
||
|
||
@test("XAU/USD analysis with different seeds")
|
||
def _():
|
||
config = PlatformConfig()
|
||
for seed_base in [10, 20, 30]:
|
||
agg = SignalAggregator(config)
|
||
dfs = {
|
||
Timeframe.H1: make_xau_ohlcv(300, seed=seed_base),
|
||
Timeframe.M15: make_xau_ohlcv(300, seed=seed_base+1),
|
||
Timeframe.H4: make_xau_ohlcv(300, seed=seed_base+2),
|
||
}
|
||
report = agg.analyze_from_dataframes(XAUUSD, dfs, use_news=False)
|
||
assert report.symbol == XAUUSD
|
||
assert report.current_price > 0
|
||
assert report.signal_quality in ("A+", "A", "B", "C", "D", "F")
|
||
|
||
|
||
@test("Edge case: all EMAs at same price (flat market)")
|
||
def _():
|
||
engine = TechnicalEngine(TechnicalConfig())
|
||
dates = pd.date_range(end=pd.Timestamp.now(tz="UTC").floor("h"), periods=300, freq="h")
|
||
df = pd.DataFrame({
|
||
"Open": 1.1, "High": 1.1001, "Low": 1.0999, "Close": 1.1, "Volume": 1000.0,
|
||
}, index=dates)
|
||
df.index.name = "Datetime"
|
||
result = engine.analyze(df, "XAU/USD", "1h")
|
||
assert result.composite_signal == SignalStrength.NEUTRAL
|
||
|
||
|
||
@test("Edge case: extreme volatility data")
|
||
def _():
|
||
engine = TechnicalEngine(TechnicalConfig())
|
||
np.random.seed(77)
|
||
dates = pd.date_range(end=pd.Timestamp.now(tz="UTC").floor("h"), periods=300, freq="h")
|
||
close = 1.1 + np.cumsum(np.random.randn(300) * 0.01)
|
||
high = close + np.abs(np.random.randn(300) * 0.005)
|
||
low = close - np.abs(np.random.randn(300) * 0.005)
|
||
df = pd.DataFrame({
|
||
"Open": close + np.random.randn(300) * 0.003,
|
||
"High": high, "Low": low, "Close": close,
|
||
"Volume": np.random.randint(100, 50000, 300).astype(float),
|
||
}, index=dates)
|
||
df.index.name = "Datetime"
|
||
result = engine.analyze(df, "XAU/USD", "1h")
|
||
assert isinstance(result, TechnicalAnalysis)
|
||
assert result.volatility_regime in ("low", "normal", "high", "extreme")
|
||
|
||
|
||
@test("Determinism: same data produces same results")
|
||
def _():
|
||
engine = TechnicalEngine(TechnicalConfig())
|
||
df = make_ohlcv(300, seed=42)
|
||
r1 = engine.analyze(df.copy(), "XAU/USD", "1h")
|
||
r2 = engine.analyze(df.copy(), "XAU/USD", "1h")
|
||
assert r1.composite_score == r2.composite_score
|
||
assert r1.composite_signal == r2.composite_signal
|
||
assert r1.market_regime == r2.market_regime
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
# MODULE 9: EUR/USD pair support
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
|
||
print("\n═══ Module 9: EUR/USD pair support ═══")
|
||
|
||
|
||
@test("PAIR_INFO includes EUR/USD with correct pip_value")
|
||
def _():
|
||
assert EURUSD in PAIR_INFO
|
||
assert PAIR_INFO[EURUSD]["pip_value"] == 0.0001
|
||
assert PAIR_INFO[EURUSD]["is_jpy_pair"] is False
|
||
assert PAIR_INFO[EURUSD]["display"] == "EUR/USD"
|
||
|
||
|
||
@test("EurUsdStrategyConfig defaults are sensible")
|
||
def _():
|
||
cfg = EurUsdStrategyConfig()
|
||
assert cfg.enabled is True
|
||
assert 0 in cfg.blocked_hours_utc
|
||
assert 5 not in cfg.allowed_weekdays # No Saturday
|
||
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.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 "A" in cfg.min_quality_grades
|
||
assert "A" in cfg.min_quality_grades
|
||
assert "B" not in cfg.min_quality_grades
|
||
|
||
|
||
@test("PlatformConfig has eur_strategy field")
|
||
def _():
|
||
config = PlatformConfig()
|
||
assert hasattr(config, "eur_strategy")
|
||
assert isinstance(config.eur_strategy, EurUsdStrategyConfig)
|
||
|
||
|
||
@test("EUR/USD config factory produces correct config")
|
||
def _():
|
||
from pair_signals.eurusd import make_config
|
||
config = make_config(Timeframe.H1)
|
||
assert config.symbol == EURUSD
|
||
assert config.display_name == "EUR/USD"
|
||
assert config.pip_value == 0.0001
|
||
assert config.xau_strategy.enabled is False
|
||
assert config.technical.regime_weights_override
|
||
assert "trending" in config.technical.regime_weights_override
|
||
assert "ranging" in config.technical.regime_weights_override
|
||
|
||
|
||
@test("XAU/USD config factory unchanged")
|
||
def _():
|
||
from pair_signals.xauusd import make_config
|
||
config = make_config(Timeframe.H1)
|
||
assert config.symbol == XAUUSD
|
||
assert config.display_name == "XAU/USD"
|
||
assert config.pip_value == 0.10
|
||
assert config.xau_strategy.enabled is True
|
||
|
||
|
||
@test("pair_signals registry includes both pairs")
|
||
def _():
|
||
from pair_signals import get_supported_pairs
|
||
pairs = get_supported_pairs()
|
||
assert XAUUSD in pairs
|
||
assert EURUSD in pairs
|
||
|
||
|
||
@test("EUR/USD signal analysis produces valid report (mocked data)")
|
||
def _():
|
||
from pair_signals.eurusd import make_config
|
||
from signal_aggregator import SignalAggregator
|
||
config = make_config(Timeframe.H1)
|
||
config.backtest.session_filter = False
|
||
agg = SignalAggregator(config)
|
||
df = make_ohlcv(n=300, start_price=1.0800, trend=0.00005)
|
||
dfs = {
|
||
Timeframe.H1: df,
|
||
Timeframe.M15: df,
|
||
Timeframe.H4: df,
|
||
}
|
||
report = agg.analyze_from_dataframes(symbol=EURUSD, dfs=dfs, use_news=False)
|
||
assert report.display_name == "EUR/USD"
|
||
assert report.signal_quality in ("A+", "A", "B", "C", "D", "F")
|
||
assert not any("SELL FILTER" in w for w in report.warnings)
|
||
|
||
|
||
@test("XAU sell gate does not apply to EUR/USD")
|
||
def _():
|
||
from pair_signals.eurusd import make_config
|
||
from signal_aggregator import SignalAggregator
|
||
config = make_config(Timeframe.H1)
|
||
config.backtest.session_filter = False
|
||
agg = SignalAggregator(config)
|
||
df = make_ohlcv(n=300, start_price=1.0800, trend=-0.0005, seed=99)
|
||
dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||
report = agg.analyze_from_dataframes(symbol=EURUSD, dfs=dfs, use_news=False)
|
||
sell_filter_warnings = [w for w in report.warnings if "SELL FILTER" in w]
|
||
assert len(sell_filter_warnings) == 0
|
||
|
||
|
||
@test("NewsEngine._get_affected_pairs includes EUR/USD for EUR and USD events")
|
||
def _():
|
||
from news_engine import NewsEngine
|
||
assert "EUR/USD" in NewsEngine._get_affected_pairs("EUR")
|
||
assert "EUR/USD" in NewsEngine._get_affected_pairs("USD")
|
||
assert "XAU/USD" in NewsEngine._get_affected_pairs("USD")
|
||
assert "EUR/USD" not in NewsEngine._get_affected_pairs("GBP")
|
||
|
||
|
||
@test("EUR/USD trade setup uses tighter RR and SL")
|
||
def _():
|
||
from pair_signals.eurusd import make_config
|
||
from signal_aggregator import SignalAggregator
|
||
config = make_config(Timeframe.H1)
|
||
config.backtest.session_filter = False
|
||
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
|
||
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}
|
||
report = agg.analyze_from_dataframes(symbol=EURUSD, dfs=dfs, use_news=False)
|
||
if report.trade_setup:
|
||
assert report.trade_setup.risk_reward >= 0.5 - 0.01
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
# MODULE 10: Bug Fix Regressions & Comprehensive Edge Cases
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
|
||
print("\n═══ Module 10: Bug Fix Regressions & Edge Cases ═══")
|
||
|
||
# ── 10a. Swing Point Bug Fix (engine.py) ─────────────────────────────
|
||
|
||
@test("Swing points: all-NaN neighbors produces no false positives")
|
||
def _():
|
||
series = pd.Series([np.nan, np.nan, np.nan, 5.0, np.nan, np.nan, np.nan])
|
||
lows = TechnicalEngine._find_swing_points(series, mode="low", window=3)
|
||
highs = TechnicalEngine._find_swing_points(series, mode="high", window=3)
|
||
assert lows == [], f"Expected no lows with all-NaN neighbors, got {lows}"
|
||
assert highs == [], f"Expected no highs with all-NaN neighbors, got {highs}"
|
||
|
||
|
||
@test("Swing points: partial NaN neighbors still detect valid swings")
|
||
def _():
|
||
series = pd.Series([10.0, 8.0, np.nan, 4.0, np.nan, 8.0, 10.0])
|
||
lows = TechnicalEngine._find_swing_points(series, mode="low", window=2)
|
||
low_vals = [v for _, v in lows]
|
||
assert 4.0 in low_vals, f"Expected 4.0 as swing low, got {low_vals}"
|
||
|
||
|
||
@test("Swing points: empty series returns empty list")
|
||
def _():
|
||
series = pd.Series([], dtype=float)
|
||
assert TechnicalEngine._find_swing_points(series, mode="low") == []
|
||
assert TechnicalEngine._find_swing_points(series, mode="high") == []
|
||
|
||
|
||
@test("Swing points: series shorter than 2*window returns empty")
|
||
def _():
|
||
series = pd.Series([1.0, 2.0, 3.0])
|
||
assert TechnicalEngine._find_swing_points(series, mode="low", window=3) == []
|
||
|
||
|
||
@test("Swing points: monotonic increasing has no interior lows")
|
||
def _():
|
||
series = pd.Series([float(i) for i in range(20)])
|
||
lows = TechnicalEngine._find_swing_points(series, mode="low", window=2)
|
||
assert len(lows) == 0, f"Monotonic increase should have no interior lows, got {lows}"
|
||
|
||
|
||
@test("Swing points: monotonic decreasing has no interior highs")
|
||
def _():
|
||
series = pd.Series([float(20 - i) for i in range(20)])
|
||
highs = TechnicalEngine._find_swing_points(series, mode="high", window=2)
|
||
assert len(highs) == 0, f"Monotonic decrease should have no interior highs, got {highs}"
|
||
|
||
|
||
@test("Swing points: V-shape detects low at bottom")
|
||
def _():
|
||
series = pd.Series([10.0, 8.0, 6.0, 4.0, 6.0, 8.0, 10.0])
|
||
lows = TechnicalEngine._find_swing_points(series, mode="low", window=2)
|
||
low_vals = [v for _, v in lows]
|
||
assert 4.0 in low_vals, f"V-shape should detect 4.0 as low, got {low_vals}"
|
||
|
||
|
||
@test("Swing points: inverted V-shape detects high at top")
|
||
def _():
|
||
series = pd.Series([4.0, 6.0, 8.0, 10.0, 8.0, 6.0, 4.0])
|
||
highs = TechnicalEngine._find_swing_points(series, mode="high", window=2)
|
||
high_vals = [v for _, v in highs]
|
||
assert 10.0 in high_vals, f"Inverted V should detect 10.0 as high, got {high_vals}"
|
||
|
||
|
||
# ── 10b. ROC Bug Fix (engine.py) ─────────────────────────────────────
|
||
|
||
@test("ROC: zero close at lookback position does not produce inf")
|
||
def _():
|
||
df = make_ohlcv(n=100, seed=42)
|
||
df.iloc[-11, df.columns.get_loc("Close")] = 0.0
|
||
config = PlatformConfig()
|
||
engine = TechnicalEngine(config.technical)
|
||
result = engine.analyze(df, "EUR/USD", Timeframe.H1)
|
||
if "roc" in result.indicators:
|
||
assert np.isfinite(result.indicators["roc"].value), "ROC should not be inf"
|
||
|
||
|
||
@test("ROC: normal computation produces finite values")
|
||
def _():
|
||
df = make_ohlcv(n=100, seed=55)
|
||
config = PlatformConfig()
|
||
engine = TechnicalEngine(config.technical)
|
||
result = engine.analyze(df, "EUR/USD", Timeframe.H1)
|
||
if "roc" in result.indicators:
|
||
assert np.isfinite(result.indicators["roc"].value)
|
||
assert 0 <= result.indicators["roc"].confidence <= 100
|
||
|
||
|
||
# ── 10c. ADX Epsilon Fix (engine.py) ────────────────────────────────
|
||
|
||
@test("ADX: flat data produces finite ADX value")
|
||
def _():
|
||
n = 100
|
||
dates = pd.date_range(end=pd.Timestamp.now(tz="UTC"), periods=n, freq="h")
|
||
df = pd.DataFrame({
|
||
"Open": [100.0]*n, "High": [100.01]*n, "Low": [99.99]*n,
|
||
"Close": [100.0]*n, "Volume": [1000.0]*n,
|
||
}, index=dates)
|
||
config = PlatformConfig()
|
||
engine = TechnicalEngine(config.technical)
|
||
result = engine.analyze(df, "EUR/USD", Timeframe.H1)
|
||
if "adx" in result.indicators:
|
||
assert np.isfinite(result.indicators["adx"].value), "ADX should be finite on flat data"
|
||
|
||
|
||
@test("ADX: strong uptrend shows high ADX")
|
||
def _():
|
||
df = make_ohlcv(n=300, trend=0.001, seed=10)
|
||
config = PlatformConfig()
|
||
engine = TechnicalEngine(config.technical)
|
||
result = engine.analyze(df, "EUR/USD", Timeframe.H1)
|
||
assert "adx" in result.indicators
|
||
assert result.indicators["adx"].value > 0
|
||
|
||
|
||
# ── 10d. RSI Boundaries ─────────────────────────────────────────────
|
||
|
||
@test("RSI: monotonic up approaches 100")
|
||
def _():
|
||
n = 100
|
||
dates = pd.date_range(end=pd.Timestamp.now(tz="UTC"), periods=n, freq="h")
|
||
close = np.linspace(100, 150, n)
|
||
df = pd.DataFrame({
|
||
"Open": close - 0.1, "High": close + 0.5, "Low": close - 0.5,
|
||
"Close": close, "Volume": [1000.0]*n,
|
||
}, index=dates)
|
||
config = PlatformConfig()
|
||
engine = TechnicalEngine(config.technical)
|
||
result = engine.analyze(df, "EUR/USD", Timeframe.H1)
|
||
assert "rsi" in result.indicators
|
||
assert result.indicators["rsi"].value > 80, f"RSI should be high on uptrend, got {result.indicators['rsi'].value}"
|
||
|
||
|
||
@test("RSI: monotonic down approaches 0")
|
||
def _():
|
||
n = 100
|
||
dates = pd.date_range(end=pd.Timestamp.now(tz="UTC"), periods=n, freq="h")
|
||
close = np.linspace(150, 100, n)
|
||
df = pd.DataFrame({
|
||
"Open": close + 0.1, "High": close + 0.5, "Low": close - 0.5,
|
||
"Close": close, "Volume": [1000.0]*n,
|
||
}, index=dates)
|
||
config = PlatformConfig()
|
||
engine = TechnicalEngine(config.technical)
|
||
result = engine.analyze(df, "EUR/USD", Timeframe.H1)
|
||
assert "rsi" in result.indicators
|
||
assert result.indicators["rsi"].value < 20, f"RSI should be low on downtrend, got {result.indicators['rsi'].value}"
|
||
|
||
|
||
@test("RSI: flat price produces RSI near 50")
|
||
def _():
|
||
n = 100
|
||
dates = pd.date_range(end=pd.Timestamp.now(tz="UTC"), periods=n, freq="h")
|
||
np.random.seed(99)
|
||
close = 100.0 + np.random.randn(n) * 0.001
|
||
df = pd.DataFrame({
|
||
"Open": close, "High": close + 0.01, "Low": close - 0.01,
|
||
"Close": close, "Volume": [1000.0]*n,
|
||
}, index=dates)
|
||
config = PlatformConfig()
|
||
engine = TechnicalEngine(config.technical)
|
||
result = engine.analyze(df, "EUR/USD", Timeframe.H1)
|
||
if "rsi" in result.indicators:
|
||
rsi = result.indicators["rsi"].value
|
||
assert 30 <= rsi <= 70, f"RSI on flat should be near 50, got {rsi}"
|
||
|
||
|
||
# ── 10e. MACD Edge Cases ────────────────────────────────────────────
|
||
|
||
@test("MACD: very short data (< 35 bars) skips cleanly")
|
||
def _():
|
||
df = make_ohlcv(n=30, seed=42)
|
||
config = PlatformConfig()
|
||
engine = TechnicalEngine(config.technical)
|
||
result = engine.analyze(df, "EUR/USD", Timeframe.H1)
|
||
# Should not crash; macd may or may not be present
|
||
|
||
|
||
@test("MACD: strong uptrend produces non-SELL signal")
|
||
def _():
|
||
df = make_ohlcv(n=300, trend=0.001, seed=20)
|
||
config = PlatformConfig()
|
||
engine = TechnicalEngine(config.technical)
|
||
result = engine.analyze(df, "EUR/USD", Timeframe.H1)
|
||
if "macd" in result.indicators:
|
||
assert result.indicators["macd"].signal not in (SignalStrength.SELL, SignalStrength.STRONG_SELL), \
|
||
f"MACD on uptrend should not be bearish, got {result.indicators['macd'].signal}"
|
||
|
||
|
||
# ── 10f. Bollinger Bands Edge Cases ─────────────────────────────────
|
||
|
||
@test("Bollinger: flat price collapses bands, position near 0.5")
|
||
def _():
|
||
n = 100
|
||
dates = pd.date_range(end=pd.Timestamp.now(tz="UTC"), periods=n, freq="h")
|
||
df = pd.DataFrame({
|
||
"Open": [100.0]*n, "High": [100.01]*n, "Low": [99.99]*n,
|
||
"Close": [100.0]*n, "Volume": [1000.0]*n,
|
||
}, index=dates)
|
||
config = PlatformConfig()
|
||
engine = TechnicalEngine(config.technical)
|
||
result = engine.analyze(df, "EUR/USD", Timeframe.H1)
|
||
if "bollinger" in result.indicators:
|
||
pos = result.indicators["bollinger"].value
|
||
assert 0.0 <= pos <= 1.0, f"BB position should be in [0,1], got {pos}"
|
||
|
||
|
||
# ── 10g. Keltner Channel Edge Cases ────────────────────────────────
|
||
|
||
@test("Keltner: BB disabled does not crash squeeze detection")
|
||
def _():
|
||
df = make_ohlcv(n=200, seed=42)
|
||
config = PlatformConfig()
|
||
config.technical.disabled_indicators = ["bollinger"]
|
||
engine = TechnicalEngine(config.technical)
|
||
result = engine.analyze(df, "EUR/USD", Timeframe.H1)
|
||
assert "keltner" in result.indicators, "Keltner should still work with BB disabled"
|
||
|
||
|
||
# ── 10h. Candlestick Pattern Edge Cases ─────────────────────────────
|
||
|
||
@test("Candlestick: zero-range bar (H==L) does not crash")
|
||
def _():
|
||
df = make_ohlcv(n=100, seed=42)
|
||
df.iloc[-1, df.columns.get_loc("High")] = df.iloc[-1]["Close"]
|
||
df.iloc[-1, df.columns.get_loc("Low")] = df.iloc[-1]["Close"]
|
||
df.iloc[-1, df.columns.get_loc("Open")] = df.iloc[-1]["Close"]
|
||
config = PlatformConfig()
|
||
engine = TechnicalEngine(config.technical)
|
||
result = engine.analyze(df, "EUR/USD", Timeframe.H1)
|
||
assert result is not None
|
||
|
||
|
||
@test("Candlestick: very short data (< 5 bars) does not crash")
|
||
def _():
|
||
df = make_ohlcv(n=4, seed=42)
|
||
config = PlatformConfig()
|
||
engine = TechnicalEngine(config.technical)
|
||
result = engine.analyze(df, "EUR/USD", Timeframe.H1)
|
||
assert result is not None
|
||
|
||
|
||
# ── 10i. Regime Classification ──────────────────────────────────────
|
||
|
||
@test("Regime: strong uptrend produces trending_up or trending")
|
||
def _():
|
||
df = make_ohlcv(n=300, trend=0.002, seed=15)
|
||
config = PlatformConfig()
|
||
engine = TechnicalEngine(config.technical)
|
||
result = engine.analyze(df, "EUR/USD", Timeframe.H1)
|
||
assert result.market_regime in ("trending_up", "trending", "transitioning"), \
|
||
f"Strong uptrend should be trending, got {result.market_regime}"
|
||
|
||
|
||
@test("Regime: strong downtrend produces trending_down or trending")
|
||
def _():
|
||
df = make_ohlcv(n=300, trend=-0.002, seed=16)
|
||
config = PlatformConfig()
|
||
engine = TechnicalEngine(config.technical)
|
||
result = engine.analyze(df, "EUR/USD", Timeframe.H1)
|
||
assert result.market_regime in ("trending_down", "trending", "transitioning"), \
|
||
f"Strong downtrend should be trending, got {result.market_regime}"
|
||
|
||
|
||
@test("Regime: flat/choppy data produces ranging or transitioning")
|
||
def _():
|
||
n = 300
|
||
dates = pd.date_range(end=pd.Timestamp.now(tz="UTC"), periods=n, freq="h")
|
||
np.random.seed(17)
|
||
close = 100.0 + np.random.randn(n) * 0.001
|
||
df = pd.DataFrame({
|
||
"Open": close, "High": close + 0.002, "Low": close - 0.002,
|
||
"Close": close, "Volume": [1000.0]*n,
|
||
}, index=dates)
|
||
config = PlatformConfig()
|
||
engine = TechnicalEngine(config.technical)
|
||
result = engine.analyze(df, "EUR/USD", Timeframe.H1)
|
||
assert result.market_regime in ("ranging", "transitioning"), \
|
||
f"Flat data should be ranging, got {result.market_regime}"
|
||
|
||
|
||
# ── 10j. Composite Score ────────────────────────────────────────────
|
||
|
||
@test("Composite score always clamped to [-100, 100]")
|
||
def _():
|
||
for seed in [1, 2, 3, 10, 20, 30]:
|
||
df = make_ohlcv(n=300, trend=0.005, seed=seed)
|
||
config = PlatformConfig()
|
||
engine = TechnicalEngine(config.technical)
|
||
result = engine.analyze(df, "EUR/USD", Timeframe.H1)
|
||
assert -100 <= result.composite_score <= 100, \
|
||
f"Score {result.composite_score} out of bounds (seed={seed})"
|
||
|
||
|
||
@test("Composite: regime weights affect score magnitude")
|
||
def _():
|
||
df = make_ohlcv(n=300, trend=0.001, seed=42)
|
||
config1 = PlatformConfig()
|
||
config2 = PlatformConfig()
|
||
config2.technical.regime_weights_override = {
|
||
"trending_up": {"ema_trend": 0.1, "rsi": 0.1, "macd": 0.1,
|
||
"bollinger": 0.1, "keltner": 0.1, "roc": 0.1,
|
||
"candlestick": 0.1, "divergence": 0.1, "adx": 0.1},
|
||
"trending_down": {"ema_trend": 0.1, "rsi": 0.1, "macd": 0.1,
|
||
"bollinger": 0.1, "keltner": 0.1, "roc": 0.1,
|
||
"candlestick": 0.1, "divergence": 0.1, "adx": 0.1},
|
||
"ranging": {"ema_trend": 0.1, "rsi": 0.1, "macd": 0.1,
|
||
"bollinger": 0.1, "keltner": 0.1, "roc": 0.1,
|
||
"candlestick": 0.1, "divergence": 0.1, "adx": 0.1},
|
||
"transitioning": {"ema_trend": 0.1, "rsi": 0.1, "macd": 0.1,
|
||
"bollinger": 0.1, "keltner": 0.1, "roc": 0.1,
|
||
"candlestick": 0.1, "divergence": 0.1, "adx": 0.1},
|
||
}
|
||
engine1 = TechnicalEngine(config1.technical)
|
||
engine2 = TechnicalEngine(config2.technical)
|
||
r1 = engine1.analyze(df, "EUR/USD", Timeframe.H1)
|
||
r2 = engine2.analyze(df, "EUR/USD", Timeframe.H1)
|
||
assert r1.composite_score != r2.composite_score or r1.market_regime != r2.market_regime, \
|
||
"Different regime weights should produce different results"
|
||
|
||
|
||
# ── 10k. Divergence Edge Cases ──────────────────────────────────────
|
||
|
||
@test("Divergence: short lookback finds no divergence")
|
||
def _():
|
||
df = make_ohlcv(n=100, seed=42)
|
||
config = PlatformConfig()
|
||
config.technical.divergence_lookback = 5
|
||
engine = TechnicalEngine(config.technical)
|
||
result = engine.analyze(df, "EUR/USD", Timeframe.H1)
|
||
# Should not crash, divergence may or may not appear
|
||
|
||
|
||
@test("Divergence: disabled divergence not in indicators")
|
||
def _():
|
||
df = make_ohlcv(n=300, seed=42)
|
||
config = PlatformConfig()
|
||
config.technical.disabled_indicators = ["divergence"]
|
||
engine = TechnicalEngine(config.technical)
|
||
result = engine.analyze(df, "EUR/USD", Timeframe.H1)
|
||
assert "divergence" not in result.indicators
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
# MODULE 11: News Engine Edge Cases
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
|
||
print("\n═══ Module 11: News Engine Edge Cases ═══")
|
||
|
||
from news_engine import NewsEngine
|
||
|
||
@test("Currency extraction: USDA does not false-match USD")
|
||
def _():
|
||
ne = NewsEngine(PlatformConfig().news)
|
||
result = ne._extract_currencies("USDA releases crop report")
|
||
assert "USD" not in result, f"USDA should not match USD, got {result}"
|
||
|
||
|
||
@test("Currency extraction: custodian does not false-match USD")
|
||
def _():
|
||
ne = NewsEngine(PlatformConfig().news)
|
||
result = ne._extract_currencies("custodian reports quarterly earnings")
|
||
assert "USD" not in result, f"custodian should not match USD, got {result}"
|
||
|
||
|
||
@test("Currency extraction: euro matches EUR correctly")
|
||
def _():
|
||
ne = NewsEngine(PlatformConfig().news)
|
||
result = ne._extract_currencies("euro strengthens against yen")
|
||
assert "EUR" in result, f"euro should match EUR, got {result}"
|
||
|
||
|
||
@test("Currency extraction: Eurozone matches EUR")
|
||
def _():
|
||
ne = NewsEngine(PlatformConfig().news)
|
||
result = ne._extract_currencies("Eurozone GDP beats forecast")
|
||
assert "EUR" in result, f"Eurozone should match EUR, got {result}"
|
||
|
||
|
||
@test("Currency extraction: empty headline returns empty list")
|
||
def _():
|
||
ne = NewsEngine(PlatformConfig().news)
|
||
result = ne._extract_currencies("")
|
||
assert result == [], f"Empty headline should return empty list, got {result}"
|
||
|
||
|
||
@test("Currency extraction: headline with multiple currencies")
|
||
def _():
|
||
ne = NewsEngine(PlatformConfig().news)
|
||
result = ne._extract_currencies("EUR/USD pair rises as dollar weakens")
|
||
assert "EUR" in result, f"Should match EUR, got {result}"
|
||
assert "USD" in result, f"Should match USD, got {result}"
|
||
|
||
|
||
@test("Sentiment: both bullish and bearish keywords gives bounded score")
|
||
def _():
|
||
ne = NewsEngine(PlatformConfig().news)
|
||
score = ne._analyze_headline_sentiment("rate hike fears amid bullish recovery expectations")
|
||
assert -1.0 <= score <= 1.0, f"Score should be bounded, got {score}"
|
||
|
||
|
||
@test("Sentiment: empty headline returns 0.0")
|
||
def _():
|
||
ne = NewsEngine(PlatformConfig().news)
|
||
assert ne._analyze_headline_sentiment("") == 0.0
|
||
|
||
|
||
@test("Sentiment: very long headline does not crash")
|
||
def _():
|
||
ne = NewsEngine(PlatformConfig().news)
|
||
long_headline = "market " * 500 + "bullish recovery"
|
||
score = ne._analyze_headline_sentiment(long_headline)
|
||
assert isinstance(score, float)
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
# MODULE 12: Signal Aggregator Edge Cases
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
|
||
print("\n═══ Module 12: Signal Aggregator Edge Cases ═══")
|
||
|
||
# ── 12a. Quality Grade Boundaries ────────────────────────────────────
|
||
|
||
@test("Quality grade: A+ requires score>=40, confidence>=65, MTF agreement")
|
||
def _():
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
df = make_ohlcv(n=300, trend=0.003, seed=100)
|
||
dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||
report = agg.analyze_from_dataframes(symbol=XAUUSD, dfs=dfs, use_news=False)
|
||
if report.signal_quality == "A+":
|
||
assert abs(report.final_score) >= 40
|
||
assert report.final_confidence >= 65
|
||
assert report.mtf_agreement is True
|
||
|
||
|
||
@test("Quality grade: A requires score>=33, confidence>=55, MTF")
|
||
def _():
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
df = make_ohlcv(n=300, trend=0.002, seed=200)
|
||
dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||
report = agg.analyze_from_dataframes(symbol=XAUUSD, dfs=dfs, use_news=False)
|
||
if report.signal_quality == "A":
|
||
assert abs(report.final_score) >= 33
|
||
assert report.final_confidence >= 55
|
||
assert report.mtf_agreement is True
|
||
|
||
|
||
@test("Quality grade: B requires score>=28, confidence>=50")
|
||
def _():
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
df = make_ohlcv(n=300, seed=300)
|
||
dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||
report = agg.analyze_from_dataframes(symbol=XAUUSD, dfs=dfs, use_news=False)
|
||
if report.signal_quality == "B":
|
||
assert abs(report.final_score) >= 28
|
||
assert report.final_confidence >= 50
|
||
|
||
|
||
@test("Quality grade: C requires score>=25, confidence>=45")
|
||
def _():
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
df = make_ohlcv(n=300, seed=400)
|
||
dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||
report = agg.analyze_from_dataframes(symbol=XAUUSD, dfs=dfs, use_news=False)
|
||
if report.signal_quality == "C":
|
||
assert abs(report.final_score) >= 25
|
||
assert report.final_confidence >= 45
|
||
|
||
|
||
@test("Quality grade: D requires score>=15")
|
||
def _():
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
df = make_ohlcv(n=300, seed=500)
|
||
dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||
report = agg.analyze_from_dataframes(symbol=XAUUSD, dfs=dfs, use_news=False)
|
||
if report.signal_quality == "D":
|
||
assert abs(report.final_score) >= 15
|
||
|
||
|
||
@test("Quality grade: F has score<15 or zero confidence")
|
||
def _():
|
||
n = 300
|
||
dates = pd.date_range(end=pd.Timestamp.now(tz="UTC"), periods=n, freq="h")
|
||
np.random.seed(600)
|
||
close = 100.0 + np.random.randn(n) * 0.0001
|
||
df = pd.DataFrame({
|
||
"Open": close, "High": close + 0.001, "Low": close - 0.001,
|
||
"Close": close, "Volume": [1000.0]*n,
|
||
}, index=dates)
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||
report = agg.analyze_from_dataframes(symbol=XAUUSD, dfs=dfs, use_news=False)
|
||
if report.signal_quality == "F":
|
||
assert abs(report.final_score) < 15 or report.final_confidence < 45
|
||
|
||
|
||
@test("Quality grade: all grades belong to valid set")
|
||
def _():
|
||
valid_grades = {"A+", "A", "B", "C", "D", "F"}
|
||
for seed in range(42, 62):
|
||
df = make_ohlcv(n=300, seed=seed, trend=0.001 * (seed % 5 - 2))
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||
report = agg.analyze_from_dataframes(symbol=XAUUSD, dfs=dfs, use_news=False)
|
||
assert report.signal_quality in valid_grades, \
|
||
f"Unexpected grade '{report.signal_quality}' (seed={seed})"
|
||
|
||
|
||
# ── 12b. EUR Filter Pipeline ────────────────────────────────────────
|
||
|
||
@test("EUR filter: blocked regime sets quality F")
|
||
def _():
|
||
from pair_signals.eurusd import make_config as eur_make_config
|
||
config = eur_make_config(Timeframe.H1)
|
||
config.eur_strategy.blocked_regimes = ["ranging"]
|
||
agg = SignalAggregator(config)
|
||
df = make_ohlcv(n=300, seed=42)
|
||
dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||
report = agg.analyze_from_dataframes(symbol=EURUSD, dfs=dfs, use_news=False)
|
||
if report.market_regime == "ranging":
|
||
assert report.signal_quality == "F", \
|
||
f"Ranging should be blocked, got {report.signal_quality}"
|
||
|
||
|
||
@test("EUR filter: quality not in min_quality_grades sets F")
|
||
def _():
|
||
from pair_signals.eurusd import make_config as eur_make_config
|
||
config = eur_make_config(Timeframe.H1)
|
||
config.eur_strategy.min_quality_grades = ["A+"]
|
||
agg = SignalAggregator(config)
|
||
df = make_ohlcv(n=300, seed=42, trend=0.001)
|
||
dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||
report = agg.analyze_from_dataframes(symbol=EURUSD, dfs=dfs, use_news=False)
|
||
if report.signal_quality not in ("A+", "F"):
|
||
assert False, f"Non-A+ grade should be blocked, got {report.signal_quality}"
|
||
|
||
|
||
@test("EUR filter: score below min sets F")
|
||
def _():
|
||
from pair_signals.eurusd import make_config as eur_make_config
|
||
config = eur_make_config(Timeframe.H1)
|
||
config.eur_strategy.min_entry_score = 99.0
|
||
agg = SignalAggregator(config)
|
||
df = make_ohlcv(n=300, seed=42, trend=0.001)
|
||
dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||
report = agg.analyze_from_dataframes(symbol=EURUSD, dfs=dfs, use_news=False)
|
||
assert report.signal_quality == "F", \
|
||
f"Score 99 threshold should block all, got {report.signal_quality}"
|
||
|
||
|
||
@test("EUR filter: confidence below min sets F")
|
||
def _():
|
||
from pair_signals.eurusd import make_config as eur_make_config
|
||
config = eur_make_config(Timeframe.H1)
|
||
config.eur_strategy.min_entry_confidence = 99.0
|
||
agg = SignalAggregator(config)
|
||
df = make_ohlcv(n=300, seed=42, trend=0.001)
|
||
dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||
report = agg.analyze_from_dataframes(symbol=EURUSD, dfs=dfs, use_news=False)
|
||
assert report.signal_quality == "F", \
|
||
f"Confidence 99 threshold should block all, got {report.signal_quality}"
|
||
|
||
|
||
@test("EUR filter: RSI extended buy (RSI>70) blocked")
|
||
def _():
|
||
from pair_signals.eurusd import make_config as eur_make_config
|
||
config = eur_make_config(Timeframe.H1)
|
||
config.eur_strategy.block_rsi_extended = True
|
||
config.eur_strategy.min_entry_score = 0.0
|
||
config.eur_strategy.min_entry_confidence = 0.0
|
||
config.eur_strategy.min_quality_grades = ["A+", "A", "B", "C", "D"]
|
||
agg = SignalAggregator(config)
|
||
df = make_ohlcv(n=300, trend=0.005, seed=42)
|
||
dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||
report = agg.analyze_from_dataframes(symbol=EURUSD, dfs=dfs, use_news=False)
|
||
if report.primary_analysis and "rsi" in report.primary_analysis.indicators:
|
||
rsi_v = report.primary_analysis.indicators["rsi"].value
|
||
if report.final_score > 0 and rsi_v > 70:
|
||
assert report.signal_quality == "F", \
|
||
f"Buy with RSI {rsi_v:.0f} should be blocked"
|
||
|
||
|
||
# ── 12c. Trade Setup Edge Cases ─────────────────────────────────────
|
||
|
||
@test("Trade setup: ATR zero returns None")
|
||
def _():
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
df = make_ohlcv(n=300, trend=0.002, seed=42)
|
||
dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||
report = agg.analyze_from_dataframes(symbol=XAUUSD, dfs=dfs, use_news=False)
|
||
if report.primary_analysis:
|
||
report.primary_analysis.key_levels["atr"] = 0.0
|
||
setup = agg._generate_trade_setup(report, pip_value=config.pip_value)
|
||
assert setup is None, "ATR=0 should produce no trade setup"
|
||
|
||
|
||
@test("Trade setup: pip_value zero returns None")
|
||
def _():
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
df = make_ohlcv(n=300, trend=0.002, seed=42)
|
||
dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||
report = agg.analyze_from_dataframes(symbol=XAUUSD, dfs=dfs, use_news=False)
|
||
if report.primary_analysis and report.final_signal != SignalStrength.NEUTRAL:
|
||
setup = agg._generate_trade_setup(report, pip_value=0.0)
|
||
assert setup is None, "pip_value=0 should produce no trade setup"
|
||
|
||
|
||
@test("Trade setup: BUY SL always below entry")
|
||
def _():
|
||
for seed in [10, 20, 30, 40, 50]:
|
||
df = make_ohlcv(n=300, trend=0.002, seed=seed)
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||
report = agg.analyze_from_dataframes(symbol=XAUUSD, dfs=dfs, use_news=False)
|
||
if report.trade_setup and report.trade_setup.direction == "BUY":
|
||
assert report.trade_setup.stop_loss < report.trade_setup.entry_price, \
|
||
f"BUY SL should be below entry (seed={seed})"
|
||
|
||
|
||
@test("Trade setup: SELL SL always above entry")
|
||
def _():
|
||
for seed in [10, 20, 30, 40, 50]:
|
||
df = make_ohlcv(n=300, trend=-0.002, seed=seed)
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||
report = agg.analyze_from_dataframes(symbol=XAUUSD, dfs=dfs, use_news=False)
|
||
if report.trade_setup and report.trade_setup.direction == "SELL":
|
||
assert report.trade_setup.stop_loss > report.trade_setup.entry_price, \
|
||
f"SELL SL should be above entry (seed={seed})"
|
||
|
||
|
||
@test("Trade setup: risk_reward is always positive when present")
|
||
def _():
|
||
for seed in range(10, 30):
|
||
df = make_ohlcv(n=300, trend=0.001 * (seed % 5 - 2), seed=seed)
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||
report = agg.analyze_from_dataframes(symbol=XAUUSD, dfs=dfs, use_news=False)
|
||
if report.trade_setup:
|
||
assert report.trade_setup.risk_reward > 0, \
|
||
f"RR should be positive, got {report.trade_setup.risk_reward} (seed={seed})"
|
||
|
||
|
||
@test("Trade setup: EUR uses EUR-specific parameters")
|
||
def _():
|
||
from pair_signals.eurusd import make_config as eur_make_config
|
||
config = eur_make_config(Timeframe.H1)
|
||
agg = SignalAggregator(config)
|
||
df = make_ohlcv(n=300, trend=0.002, start_price=1.08, seed=77)
|
||
dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||
report = agg.analyze_from_dataframes(symbol=EURUSD, dfs=dfs, use_news=False)
|
||
if report.trade_setup:
|
||
assert report.trade_setup.risk_reward >= config.eur_strategy.min_rr_ratio - 0.01
|
||
|
||
|
||
# ── 12d. MTF Agreement ──────────────────────────────────────────────
|
||
|
||
@test("MTF: all NEUTRAL returns False")
|
||
def _():
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
report = SignalReport(symbol="XAU/USD", display_name="XAU/USD", timestamp="now")
|
||
report.mtf_signals = {
|
||
"15min": SignalStrength.NEUTRAL,
|
||
"1h": SignalStrength.NEUTRAL,
|
||
"4h": SignalStrength.NEUTRAL,
|
||
}
|
||
assert agg._check_mtf_agreement(report) is False
|
||
|
||
|
||
@test("MTF: 2 bearish + 1 bullish returns True (bearish agreement)")
|
||
def _():
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
report = SignalReport(symbol="XAU/USD", display_name="XAU/USD", timestamp="now")
|
||
report.mtf_signals = {
|
||
"15min": SignalStrength.SELL,
|
||
"1h": SignalStrength.STRONG_SELL,
|
||
"4h": SignalStrength.BUY,
|
||
}
|
||
assert agg._check_mtf_agreement(report) is True
|
||
|
||
|
||
@test("MTF: all same direction returns True")
|
||
def _():
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
report = SignalReport(symbol="XAU/USD", display_name="XAU/USD", timestamp="now")
|
||
report.mtf_signals = {
|
||
"15min": SignalStrength.BUY,
|
||
"1h": SignalStrength.BUY,
|
||
"4h": SignalStrength.STRONG_BUY,
|
||
}
|
||
assert agg._check_mtf_agreement(report) is True
|
||
|
||
|
||
@test("MTF: empty signals returns False")
|
||
def _():
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
report = SignalReport(symbol="XAU/USD", display_name="XAU/USD", timestamp="now")
|
||
report.mtf_signals = {}
|
||
assert agg._check_mtf_agreement(report) is False
|
||
|
||
|
||
# ── 12e. Final Signal Computation ───────────────────────────────────
|
||
|
||
@test("Final signal: STRONG_BUY requires score>=35, confidence>=55")
|
||
def _():
|
||
for seed in range(42, 62):
|
||
df = make_ohlcv(n=300, trend=0.001 * (seed % 4), seed=seed)
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||
report = agg.analyze_from_dataframes(symbol=XAUUSD, dfs=dfs, use_news=False)
|
||
if report.final_signal == SignalStrength.STRONG_BUY:
|
||
assert report.final_score >= 35, f"STRONG_BUY needs score>=35, got {report.final_score}"
|
||
assert report.final_confidence >= 55, f"STRONG_BUY needs conf>=55, got {report.final_confidence}"
|
||
|
||
|
||
@test("Final signal: STRONG_SELL requires score<=-35, confidence>=55")
|
||
def _():
|
||
for seed in range(42, 62):
|
||
df = make_ohlcv(n=300, trend=-0.001 * (seed % 4), seed=seed)
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||
report = agg.analyze_from_dataframes(symbol=XAUUSD, dfs=dfs, use_news=False)
|
||
if report.final_signal == SignalStrength.STRONG_SELL:
|
||
assert report.final_score <= -35, f"STRONG_SELL needs score<=-35, got {report.final_score}"
|
||
assert report.final_confidence >= 55, f"STRONG_SELL needs conf>=55, got {report.final_confidence}"
|
||
|
||
|
||
@test("Final signal: NEUTRAL when score is low regardless of confidence")
|
||
def _():
|
||
n = 300
|
||
dates = pd.date_range(end=pd.Timestamp.now(tz="UTC"), periods=n, freq="h")
|
||
np.random.seed(42)
|
||
close = 100.0 + np.random.randn(n) * 0.0001
|
||
df = pd.DataFrame({
|
||
"Open": close, "High": close + 0.001, "Low": close - 0.001,
|
||
"Close": close, "Volume": [1000.0]*n,
|
||
}, index=dates)
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||
report = agg.analyze_from_dataframes(symbol=XAUUSD, dfs=dfs, use_news=False)
|
||
if abs(report.final_score) < 25:
|
||
assert report.final_signal == SignalStrength.NEUTRAL
|
||
|
||
|
||
# ── 12f. Penalty System ─────────────────────────────────────────────
|
||
|
||
@test("Penalty: confidence never exceeds 95 or goes below 0")
|
||
def _():
|
||
for seed in range(42, 72):
|
||
df = make_ohlcv(n=300, trend=0.001 * (seed % 6 - 3), seed=seed)
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||
report = agg.analyze_from_dataframes(symbol=XAUUSD, dfs=dfs, use_news=False)
|
||
assert 0.0 <= report.final_confidence <= 95.0, \
|
||
f"Confidence out of [0,95] range: {report.final_confidence} (seed={seed})"
|
||
|
||
|
||
# ── 12g. Weekend Filter ─────────────────────────────────────────────
|
||
|
||
@test("Weekend filter: Friday 22:01 UTC is weekend")
|
||
def _():
|
||
from signal_aggregator import _is_forex_weekend
|
||
dt = datetime(2026, 5, 15, 22, 1, tzinfo=timezone.utc) # Friday
|
||
assert dt.weekday() == 4
|
||
assert _is_forex_weekend(dt) is True
|
||
|
||
|
||
@test("Weekend filter: Friday 21:59 UTC is NOT weekend")
|
||
def _():
|
||
from signal_aggregator import _is_forex_weekend
|
||
dt = datetime(2026, 5, 15, 21, 59, tzinfo=timezone.utc) # Friday
|
||
assert dt.weekday() == 4
|
||
assert _is_forex_weekend(dt) is False
|
||
|
||
|
||
@test("Weekend filter: Saturday noon is weekend")
|
||
def _():
|
||
from signal_aggregator import _is_forex_weekend
|
||
dt = datetime(2026, 5, 16, 12, 0, tzinfo=timezone.utc) # Saturday
|
||
assert dt.weekday() == 5
|
||
assert _is_forex_weekend(dt) is True
|
||
|
||
|
||
@test("Weekend filter: Sunday 21:59 UTC is weekend")
|
||
def _():
|
||
from signal_aggregator import _is_forex_weekend
|
||
dt = datetime(2026, 5, 17, 21, 59, tzinfo=timezone.utc) # Sunday
|
||
assert dt.weekday() == 6
|
||
assert _is_forex_weekend(dt) is True
|
||
|
||
|
||
@test("Weekend filter: Sunday 22:00 UTC is NOT weekend")
|
||
def _():
|
||
from signal_aggregator import _is_forex_weekend
|
||
dt = datetime(2026, 5, 17, 22, 0, tzinfo=timezone.utc) # Sunday
|
||
assert dt.weekday() == 6
|
||
assert _is_forex_weekend(dt) is False
|
||
|
||
|
||
@test("Weekend filter: Monday is not weekend")
|
||
def _():
|
||
from signal_aggregator import _is_forex_weekend
|
||
dt = datetime(2026, 5, 18, 10, 0, tzinfo=timezone.utc) # Monday
|
||
assert dt.weekday() == 0
|
||
assert _is_forex_weekend(dt) is False
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
# MODULE 13: Integration Edge Cases
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
|
||
print("\n═══ Module 13: Integration Edge Cases ═══")
|
||
|
||
|
||
@test("Integration: all indicators disabled produces valid report")
|
||
def _():
|
||
config = PlatformConfig()
|
||
config.technical.disabled_indicators = [
|
||
"ema_trend", "rsi", "bollinger", "macd", "adx",
|
||
"keltner", "roc", "candlestick", "divergence",
|
||
]
|
||
agg = SignalAggregator(config)
|
||
df = make_ohlcv(n=300, seed=42)
|
||
dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||
report = agg.analyze_from_dataframes(symbol=XAUUSD, dfs=dfs, use_news=False)
|
||
assert report is not None
|
||
assert report.signal_quality in ("A+", "A", "B", "C", "D", "F", "")
|
||
|
||
|
||
@test("Integration: data with 50% NaN still produces valid report")
|
||
def _():
|
||
df = make_ohlcv(n=300, seed=42)
|
||
mask = np.random.RandomState(99).choice([True, False], size=len(df), p=[0.5, 0.5])
|
||
df.loc[mask, "Close"] = np.nan
|
||
df.loc[mask, "High"] = np.nan
|
||
df.loc[mask, "Low"] = np.nan
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
dfs = {Timeframe.H1: df}
|
||
report = agg.analyze_from_dataframes(symbol=XAUUSD, dfs=dfs, use_news=False)
|
||
assert report is not None
|
||
assert report.signal_quality in ("A+", "A", "B", "C", "D", "F", "")
|
||
|
||
|
||
@test("Integration: XAU and EUR produce independent results")
|
||
def _():
|
||
from pair_signals.eurusd import make_config as eur_make_config
|
||
df = make_ohlcv(n=300, seed=42, trend=0.001)
|
||
|
||
xau_config = PlatformConfig()
|
||
xau_agg = SignalAggregator(xau_config)
|
||
xau_dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||
xau_report = xau_agg.analyze_from_dataframes(symbol=XAUUSD, dfs=xau_dfs, use_news=False)
|
||
|
||
eur_config = eur_make_config(Timeframe.H1)
|
||
eur_agg = SignalAggregator(eur_config)
|
||
eur_dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||
eur_report = eur_agg.analyze_from_dataframes(symbol=EURUSD, dfs=eur_dfs, use_news=False)
|
||
|
||
assert xau_report.display_name == "XAU/USD"
|
||
assert eur_report.display_name == "EUR/USD"
|
||
|
||
|
||
@test("Integration: missing confirmation TFs still produces valid report")
|
||
def _():
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
df = make_ohlcv(n=300, seed=42)
|
||
dfs = {Timeframe.H1: df}
|
||
report = agg.analyze_from_dataframes(symbol=XAUUSD, dfs=dfs, use_news=False)
|
||
assert report is not None
|
||
assert report.mtf_agreement is False
|
||
|
||
|
||
@test("Integration: extreme volatility data does not crash")
|
||
def _():
|
||
n = 300
|
||
dates = pd.date_range(end=pd.Timestamp.now(tz="UTC"), periods=n, freq="h")
|
||
np.random.seed(42)
|
||
close = 100.0 + np.cumsum(np.random.randn(n) * 10.0)
|
||
close = np.maximum(close, 1.0)
|
||
df = pd.DataFrame({
|
||
"Open": close * 0.99, "High": close * 1.05,
|
||
"Low": close * 0.95, "Close": close,
|
||
"Volume": [5000.0]*n,
|
||
}, index=dates)
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||
report = agg.analyze_from_dataframes(symbol=XAUUSD, dfs=dfs, use_news=False)
|
||
assert report is not None
|
||
assert -100 <= report.final_score <= 100
|
||
|
||
|
||
@test("Integration: very short data (50 bars) produces valid report")
|
||
def _():
|
||
df = make_ohlcv(n=50, seed=42)
|
||
config = PlatformConfig()
|
||
agg = SignalAggregator(config)
|
||
dfs = {Timeframe.H1: df}
|
||
report = agg.analyze_from_dataframes(symbol=XAUUSD, dfs=dfs, use_news=False)
|
||
assert report is not None
|
||
|
||
|
||
@test("Integration: multiple seeds produce deterministic results")
|
||
def _():
|
||
df = make_ohlcv(n=300, seed=42)
|
||
config = PlatformConfig()
|
||
agg1 = SignalAggregator(config)
|
||
agg2 = SignalAggregator(config)
|
||
dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||
r1 = agg1.analyze_from_dataframes(symbol=XAUUSD, dfs=dfs, use_news=False)
|
||
r2 = agg2.analyze_from_dataframes(symbol=XAUUSD, dfs=dfs, use_news=False)
|
||
assert r1.final_score == r2.final_score
|
||
assert r1.final_signal == r2.final_signal
|
||
assert r1.signal_quality == r2.signal_quality
|
||
|
||
|
||
@test("Integration: session filter blocks outside hours correctly")
|
||
def _():
|
||
from pair_signals.eurusd import make_config as eur_make_config
|
||
config = eur_make_config(Timeframe.H1)
|
||
config.eur_strategy.session_hours = [8, 9, 10, 11, 12, 13, 14, 15, 16]
|
||
agg = SignalAggregator(config)
|
||
n = 300
|
||
end_time = pd.Timestamp("2026-05-18 03:00:00", tz="UTC")
|
||
dates = pd.date_range(end=end_time, periods=n, freq="h")
|
||
np.random.seed(42)
|
||
close = 1.08 + np.cumsum(np.random.randn(n) * 0.001 + 0.0002)
|
||
df = pd.DataFrame({
|
||
"Open": close - 0.0001, "High": close + 0.0005,
|
||
"Low": close - 0.0005, "Close": close, "Volume": [1000.0]*n,
|
||
}, index=dates)
|
||
dfs = {Timeframe.H1: df, Timeframe.M15: df, Timeframe.H4: df}
|
||
report = agg.analyze_from_dataframes(symbol=EURUSD, dfs=dfs, use_news=False)
|
||
bar_hour = dates[-1].hour
|
||
if bar_hour not in config.eur_strategy.session_hours and report.signal_quality != "":
|
||
assert report.signal_quality == "F", \
|
||
f"Hour {bar_hour} should be blocked, got {report.signal_quality}"
|
||
|
||
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
# SUMMARY
|
||
# ══════════════════════════════════════════════════════════════════════
|
||
|
||
print(f"\n{'═' * 60}")
|
||
print(f" RESULTS: {PASS} passed, {FAIL} failed, {PASS + FAIL} total")
|
||
print(f"{'═' * 60}")
|
||
|
||
if ERRORS:
|
||
print("\n FAILURES:")
|
||
for name, err, tb in ERRORS:
|
||
print(f"\n ✗ {name}")
|
||
print(f" Error: {err}")
|
||
for line in tb.strip().split("\n")[-3:]:
|
||
print(f" {line}")
|
||
|
||
print()
|
||
sys.exit(1 if FAIL > 0 else 0)
|