harry/ForexBot2
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refocus on USD/JPY and add walk-forward backtester with regime filter

Browse Source 
- Strip multi-pair plumbing; drop synthetic-data fallback
- Add BacktestConfig + walk-forward replay through the live signal pipeline
- Add metrics module (PF, Sharpe, Sortino, max DD, trade journal)
- Add parallel grid optimizer with disk parquet cache
- Add H4 ADX regime gate (--regime-filter, --regime-min-adx) — cuts chop
  losses ~84% in 2023 H1 while preserving trend edge
This commit is contained in:
Harwinder
2026-05-07 17:15:41 -04:00
parent c04f451c0c
commit 045f4248f3
9 changed files with 1605 additions and 1203 deletions
Download Patch File Download Diff File Expand all files Collapse all files
signal_aggregator.py
+123 -142
View File
@@ -1,16 +1,10 @@
"""
Forex Signal Intelligence Platform - Signal Aggregator
========================================================
Combines technical analysis across multiple timeframes with
fundamental/news context to produce final actionable signals.
This is the brain — it decides what's actually worth paying attention to.
Key principles:
1. Multi-timeframe agreement = higher confidence
2. High-impact news overrides technicals
3. Divergences between timeframes = caution
4. No signal is better than a bad signal
ForexBot2 — Signal Aggregator
==============================
Combines multi-timeframe technical analysis with optional news context
into a graded actionable signal. Same code path is used for live mode
and walk-forward backtesting; the only difference is whether the input
OHLCV is fetched live or sliced from a pre-loaded historical buffer.
"""
import logging
@@ -20,10 +14,7 @@ from typing import Dict, List, Optional
import pandas as pd
from config import (
PlatformConfig, SignalStrength, Timeframe,
SYMBOL_DISPLAY,
)
from config import PlatformConfig, SignalStrength, Timeframe
from data_provider import ForexDataProvider
from engine import TechnicalEngine, TechnicalAnalysis
from news_engine import NewsEngine, FundamentalContext
@@ -33,10 +24,9 @@ logger = logging.getLogger(__name__)
@dataclass
class TradeSetup:
"""A complete trade setup recommendation."""
pair: str
display_name: str
direction: str # "BUY" or "SELL"
direction: str # "BUY" / "SELL"
entry_price: float
stop_loss: float
take_profit_1: float
@@ -49,47 +39,38 @@ class TradeSetup:
@dataclass
class SignalReport:
"""Complete signal report for a single pair."""
symbol: str
display_name: str
timestamp: str
# Composite signals
primary_signal: SignalStrength = SignalStrength.NEUTRAL
primary_score: float = 0.0
primary_confidence: float = 0.0
# Multi-timeframe agreement
mtf_agreement: bool = False
mtf_signals: Dict[str, SignalStrength] = field(default_factory=dict)
mtf_scores: Dict[str, float] = field(default_factory=dict)
# Technical details
primary_analysis: Optional[TechnicalAnalysis] = None
confirmation_analyses: List[TechnicalAnalysis] = field(default_factory=list)
# Fundamental context
event_risk: str = "clear"
event_description: str = ""
news_sentiment: float = 0.0
news_description: str = ""
# Final verdict
final_signal: SignalStrength = SignalStrength.NEUTRAL
final_score: float = 0.0
final_confidence: float = 0.0
action_recommendation: str = ""
signal_quality: str = "" # "A+", "A", "B", "C", "D", "F"
signal_quality: str = "" # A+, A, B, C, D, F
# Trade setup (only for actionable signals)
trade_setup: Optional[TradeSetup] = None
# Key reasons
bullish_reasons: List[str] = field(default_factory=list)
bearish_reasons: List[str] = field(default_factory=list)
warnings: List[str] = field(default_factory=list)
# Current price info
current_price: float = 0.0
spread_estimate: float = 0.0
market_regime: str = ""
@@ -98,53 +79,67 @@ class SignalReport:
class SignalAggregator:
"""
Master signal aggregator.
Orchestrates data fetching, technical analysis across timeframes,
fundamental analysis, and produces final graded signals.
Master aggregator. Designed so the backtester can drive it bar-by-bar
via `analyze_from_dataframes` without re-fetching from the API.
"""
def __init__(self, config: PlatformConfig):
self.config = config
self.data_provider = ForexDataProvider(config)
self.tech_engine = TechnicalEngine(config.technical)
self.news_engine = NewsEngine(config.news)
self._news_engine: Optional[NewsEngine] = None # lazy
def analyze_all_pairs(self) -> List[SignalReport]:
"""
Analyze all watched pairs and return sorted signal reports.
Returns reports sorted by signal quality (best first).
"""
reports = []
context = self.news_engine.get_fundamental_context()
def _get_news_engine(self) -> NewsEngine:
if self._news_engine is None:
self._news_engine = NewsEngine(self.config.news)
return self._news_engine
# Batch-prefetch all pairs × all needed timeframes (3 API calls instead of 21)
all_timeframes = (
[self.config.primary_timeframe] + self.config.confirmation_timeframes
)
self.data_provider.prefetch_batch(self.config.watch_pairs, all_timeframes)
for symbol in self.config.watch_pairs:
try:
report = self.analyze_pair(symbol, context)
reports.append(report)
except Exception as e:
logger.error(f"Failed to analyze {symbol}: {e}")
continue
# Sort by absolute score (strongest signals first)
reports.sort(key=lambda r: abs(r.final_score), reverse=True)
return reports
# ── Public: live mode entry point ──────────────────────────────
def analyze_pair(
self,
symbol: str,
symbol: Optional[str] = None,
context: Optional[FundamentalContext] = None,
) -> SignalReport:
"""Complete analysis of a single pair across all layers."""
if context is None:
context = self.news_engine.get_fundamental_context()
"""Live-mode entry: fetch all needed timeframes then delegate."""
if symbol is None:
symbol = self.config.symbol
timeframes = [self.config.primary_timeframe] + list(
self.config.confirmation_timeframes
)
dfs: Dict[Timeframe, pd.DataFrame] = {}
for tf in timeframes:
try:
dfs[tf] = self.data_provider.fetch_ohlcv(symbol, tf)
except Exception as e:
logger.warning(f"Fetch failed for {symbol} {tf.value}: {e}")
return self.analyze_from_dataframes(
symbol=symbol,
dfs=dfs,
context=context,
use_news=True,
)
# ── Public: bar-replayable entry point ─────────────────────────
def analyze_from_dataframes(
self,
symbol: str,
dfs: Dict[Timeframe, pd.DataFrame],
context: Optional[FundamentalContext] = None,
use_news: bool = True,
) -> SignalReport:
"""
Run the full analysis pipeline on already-fetched dataframes.
`dfs` must include the primary timeframe and ideally the
confirmation timeframes. Each df is treated as the visible history
at decision time — the backtester is responsible for slicing so
no future bars leak in.
"""
pair_info = self.data_provider.get_pair_info(symbol)
display = pair_info["display"]
pip_value = pair_info["pip_value"]
@@ -155,33 +150,33 @@ class SignalAggregator:
timestamp=str(datetime.now(timezone.utc)),
)
# ── 1. Primary Timeframe Analysis ──────────────────────────
try:
primary_df = self.data_provider.fetch_ohlcv(
symbol, self.config.primary_timeframe
)
primary_analysis = self.tech_engine.analyze(
primary_df, symbol, self.config.primary_timeframe.value
)
report.primary_analysis = primary_analysis
report.primary_signal = primary_analysis.composite_signal
report.primary_score = primary_analysis.composite_score
report.primary_confidence = primary_analysis.confidence
report.market_regime = primary_analysis.market_regime
report.volatility_regime = primary_analysis.volatility_regime
report.current_price = float(primary_df["Close"].iloc[-1])
report.mtf_signals[self.config.primary_timeframe.value] = primary_analysis.composite_signal
report.mtf_scores[self.config.primary_timeframe.value] = primary_analysis.composite_score
except Exception as e:
logger.error(f"Primary analysis failed for {symbol}: {e}")
report.warnings.append(f"Primary timeframe data unavailable: {e}")
# ── 1. Primary timeframe ───────────────────────────────────
primary_df = dfs.get(self.config.primary_timeframe)
if primary_df is None or primary_df.empty:
report.warnings.append("Primary timeframe data unavailable")
report.signal_quality = "F"
return report
# ── 2. Confirmation Timeframe Analyses ─────────────────────
primary_analysis = self.tech_engine.analyze(
primary_df, symbol, self.config.primary_timeframe.value
)
report.primary_analysis = primary_analysis
report.primary_signal = primary_analysis.composite_signal
report.primary_score = primary_analysis.composite_score
report.primary_confidence = primary_analysis.confidence
report.market_regime = primary_analysis.market_regime
report.volatility_regime = primary_analysis.volatility_regime
report.current_price = float(primary_df["Close"].iloc[-1])
report.timestamp = str(primary_df.index[-1])
report.mtf_signals[self.config.primary_timeframe.value] = primary_analysis.composite_signal
report.mtf_scores[self.config.primary_timeframe.value] = primary_analysis.composite_score
# ── 2. Confirmation timeframes ─────────────────────────────
for tf in self.config.confirmation_timeframes:
cf_df = dfs.get(tf)
if cf_df is None or cf_df.empty:
continue
try:
cf_df = self.data_provider.fetch_ohlcv(symbol, tf)
cf_analysis = self.tech_engine.analyze(cf_df, symbol, tf.value)
report.confirmation_analyses.append(cf_analysis)
report.mtf_signals[tf.value] = cf_analysis.composite_signal
@@ -189,35 +184,51 @@ class SignalAggregator:
except Exception as e:
logger.warning(f"Confirmation TF {tf.value} failed for {symbol}: {e}")
# ── 3. Multi-Timeframe Agreement Check ─────────────────────
# ── 3. MTF agreement ───────────────────────────────────────
report.mtf_agreement = self._check_mtf_agreement(report)
# ── 4. Fundamental Context ─────────────────────────────────
event_risk, event_desc = self.news_engine.get_event_risk_level(symbol, context)
report.event_risk = event_risk
report.event_description = event_desc
# ── 4. Fundamental context (optional) ──────────────────────
if use_news:
news = self._get_news_engine()
if context is None:
context = news.get_fundamental_context()
event_risk, event_desc = news.get_event_risk_level(symbol, context)
report.event_risk = event_risk
report.event_description = event_desc
sentiment, sent_desc = news.get_pair_sentiment(symbol, context)
report.news_sentiment = sentiment
report.news_description = sent_desc
else:
report.event_risk = "clear"
report.event_description = ""
report.news_sentiment = 0.0
report.news_description = ""
sentiment, sent_desc = self.news_engine.get_pair_sentiment(symbol, context)
report.news_sentiment = sentiment
report.news_description = sent_desc
# ── 5. Spread estimate (only meaningful in live mode) ──────
if use_news:
try:
report.spread_estimate = self.data_provider.get_spread_estimate(symbol)
except Exception:
report.spread_estimate = 0.0
else:
# Backtest sets spread externally via BacktestConfig
report.spread_estimate = 0.0
# ── 5. Spread Estimate ─────────────────────────────────────
report.spread_estimate = self.data_provider.get_spread_estimate(symbol)
# ── 6. Compute Final Signal ────────────────────────────────
# ── 6. Final signal ────────────────────────────────────────
self._compute_final_signal(report)
# ── 7. Generate Trade Setup (if signal is actionable) ──────
# ── 7. Trade setup for actionable signals ──────────────────
if report.signal_quality in ("A+", "A", "B"):
report.trade_setup = self._generate_trade_setup(report, pip_value)
# ── 8. Collect Reasons ─────────────────────────────────────
# ── 8. Reasons ─────────────────────────────────────────────
self._collect_reasons(report)
return report
# ── Internals ──────────────────────────────────────────────────
def _check_mtf_agreement(self, report: SignalReport) -> bool:
"""Check if multiple timeframes agree on direction."""
signals = list(report.mtf_signals.values())
if len(signals) < 2:
return False
@@ -229,28 +240,17 @@ class SignalAggregator:
return bullish >= threshold or bearish >= threshold
def _compute_final_signal(self, report: SignalReport):
"""
Compute the final signal by combining all layers.
Priority:
1. Event risk can downgrade or block signals
2. MTF agreement boosts confidence
3. News sentiment adds a small bias
4. Quality grade combines everything
"""
score = report.primary_score
confidence = report.primary_confidence
# ── MTF adjustment ─────────────────────────────────────────
# MTF
if report.mtf_agreement:
# All timeframes agree — boost confidence and blend scores
confidence = min(confidence + 15, 95)
all_scores = list(report.mtf_scores.values())
if all_scores:
mtf_avg = sum(all_scores) / len(all_scores)
score = score * 0.6 + mtf_avg * 0.4
else:
# Timeframes disagree — reduce confidence
confidence *= 0.7
if len(report.mtf_scores) >= 2:
scores = list(report.mtf_scores.values())
@@ -260,35 +260,33 @@ class SignalAggregator:
report.warnings.append("Timeframes CONFLICT — mixed signals across timeframes")
confidence *= 0.6
# ── News sentiment adjustment ──────────────────────────────
# News sentiment
if abs(report.news_sentiment) > 0.2:
# Sentiment aligns with technical
if (report.news_sentiment > 0 and score > 0) or \
(report.news_sentiment < 0 and score < 0):
confidence = min(confidence + 5, 95)
# Sentiment conflicts with technical
elif (report.news_sentiment > 0 and score < -10) or \
(report.news_sentiment < 0 and score > 10):
report.warnings.append("News sentiment CONFLICTS with technical signal")
confidence *= 0.85
# ── Event risk adjustment ──────────────────────────────────
# Event risk
if report.event_risk == "danger":
report.warnings.append(f"EVENT RISK: {report.event_description}")
confidence *= 0.3 # Heavily penalize — event will disrupt price action
confidence *= 0.3
report.action_recommendation = f"WAIT — {report.event_description}"
elif report.event_risk == "caution":
report.warnings.append(f"Upcoming event: {report.event_description}")
confidence *= 0.7
# ── Volatility regime adjustment ───────────────────────────
# Volatility regime
if report.volatility_regime == "extreme":
report.warnings.append("EXTREME volatility — wider stops needed, reduced position size")
confidence *= 0.8
elif report.volatility_regime == "low":
report.warnings.append("Low volatility — potential for breakout, tight ranges")
# ── Final signal determination ─────────────────────────────
# Final signal
if score >= 40 and confidence >= 65:
report.final_signal = SignalStrength.STRONG_BUY
elif score >= 15 and confidence >= 50:
@@ -303,7 +301,7 @@ class SignalAggregator:
report.final_score = score
report.final_confidence = confidence
# ── Quality grade ──────────────────────────────────────────
# Quality grade
abs_score = abs(score)
if abs_score >= 50 and confidence >= 75 and report.mtf_agreement:
report.signal_quality = "A+"
@@ -318,8 +316,7 @@ class SignalAggregator:
else:
report.signal_quality = "F"
# ── Action recommendation ──────────────────────────────────
if not report.action_recommendation: # not already set by event risk
if not report.action_recommendation:
sig_name = report.final_signal.name.replace("_", " ")
if report.signal_quality in ("A+", "A"):
report.action_recommendation = f"{sig_name} — High-quality setup, consider entry"
@@ -331,35 +328,29 @@ class SignalAggregator:
report.action_recommendation = "NO TRADE — No clear signal"
def _generate_trade_setup(self, report: SignalReport, pip_value: float) -> Optional[TradeSetup]:
"""Generate entry, SL, TP levels for actionable signals."""
if not report.primary_analysis:
return None
price = report.current_price
atr = report.primary_analysis.key_levels.get("atr", 0)
if atr == 0:
return None
is_buy = report.final_signal in (SignalStrength.BUY, SignalStrength.STRONG_BUY)
rr = self.config.signals.default_rr_ratio
sl_mult = self.config.signals.atr_sl_multiplier
# SL = 1.5x ATR from entry (adaptive to volatility)
sl_distance = atr * 1.5
sl_distance = atr * sl_mult
tp1_distance = sl_distance * rr
tp2_distance = sl_distance * rr * 1.5
if is_buy:
entry = price
sl = price - sl_distance
tp1 = price + tp1_distance
tp2 = price + tp2_distance
entry, sl = price, price - sl_distance
tp1, tp2 = price + tp1_distance, price + tp2_distance
direction = "BUY"
else:
entry = price
sl = price + sl_distance
tp1 = price - tp1_distance
tp2 = price - tp2_distance
entry, sl = price, price + sl_distance
tp1, tp2 = price - tp1_distance, price - tp2_distance
direction = "SELL"
risk_pips = sl_distance / pip_value
@@ -380,21 +371,12 @@ class SignalAggregator:
)
def _collect_reasons(self, report: SignalReport):
"""
Collect bullish/bearish reasons from all indicators.
Each indicator's description is captured once via the indicators
dict. Pattern signals (candlestick) are already included in the
'candlestick' indicator description, so we do NOT re-process
pattern_signals to avoid duplicate entries.
"""
if not report.primary_analysis:
return
for name, ind in report.primary_analysis.indicators.items():
if name.startswith("_") or ind.weight == 0:
continue
if ind.signal in (SignalStrength.BUY, SignalStrength.STRONG_BUY):
prefix = "STRONG: " if ind.signal == SignalStrength.STRONG_BUY else ""
report.bullish_reasons.append(f"{prefix}{ind.name}: {ind.description}")
@@ -402,7 +384,6 @@ class SignalAggregator:
prefix = "STRONG: " if ind.signal == SignalStrength.STRONG_SELL else ""
report.bearish_reasons.append(f"{prefix}{ind.name}: {ind.description}")
# Add news context
if report.news_sentiment > 0.1:
report.bullish_reasons.append(f"News: {report.news_description}")
elif report.news_sentiment < -0.1: