feat(backtest): live-parity backtest engine with zero-lookahead AI/regime lookups
Phase 1 of the prove-before-live alignment. The backtest engine now mirrors the live signal path instead of hardcoding HOLD/NEUTRAL: - predict_batch (signal_classifier): vectorized per-bar AI signals reusing the exact live predict() pipeline (class-index mapping, conf<0.5 -> HOLD) - backtester: per-bar time-aligned AI lookup (M15 close = signal-available time, searchsorted, no lookahead), exogenous H4 regime plan with dynamic confluence-key deltas, AttentionMarketScorer alignment, per-strategy entry cooldown, max-hold timeout exit, breakeven + trailing capped at TP, running- equity pnl_pct (was: static initial balance) - fix: store direction.value on entry (Enum vs str compare silently disabled every exit - no trade ever closed) - fix: _check_mode_sync only clears override on genuine mode transition (grid threshold injection was wiped on first read) - SignalAggregator(use_adaptive_weights=True): backtest uses pure config weights so garbage live adaptive weights can't contaminate gate scores Co-Authored-By: Claude Code <noreply@anthropic.com>
This commit is contained in:
Claude Code
@@ -166,3 +166,60 @@ class SignalClassifier:
|
||||
except Exception as e:
|
||||
logger.error(f"XGBoost prediction error: {e}")
|
||||
return {"direction": "HOLD", "confidence": 0.0}
|
||||
|
||||
def predict_batch(self, df: pd.DataFrame) -> pd.DataFrame:
|
||||
"""
|
||||
Vectorized per-bar BUY/SELL/HOLD predictions for backtesting.
|
||||
|
||||
Uses the exact same feature pipeline as predict() (create_ml_features,
|
||||
missing-column fill-0, confidence<0.5 → HOLD), so backtest AI signals
|
||||
match live inference one-to-one. Features are trailing-window only →
|
||||
no lookahead bias. Rows whose feature window is incomplete (NaN) are
|
||||
marked direction=HOLD / confidence=0.0 (no prediction).
|
||||
|
||||
Returns:
|
||||
DataFrame indexed like df with columns:
|
||||
direction, confidence, p_buy, p_hold, p_sell
|
||||
"""
|
||||
if self._model is None:
|
||||
if not self.load_model():
|
||||
raise RuntimeError("XGBoost model not loaded — run scripts/train_ai.py first")
|
||||
|
||||
ta = TechnicalAnalyzer()
|
||||
feature_df = ta.create_ml_features(df)
|
||||
|
||||
for col in self._feature_columns:
|
||||
if col not in feature_df.columns:
|
||||
feature_df[col] = 0
|
||||
|
||||
X = feature_df[self._feature_columns].astype(float)
|
||||
valid_mask = X.notna().all(axis=1).values
|
||||
n = len(X)
|
||||
|
||||
probs = np.zeros((n, 3))
|
||||
if valid_mask.any():
|
||||
proba_v = self._model.predict_proba(X.loc[valid_mask].values)
|
||||
# Map model class indices (self._model.classes_) onto slots 0/1/2
|
||||
for k, cls in enumerate(self._model.classes_):
|
||||
probs[valid_mask, cls] = proba_v[:, k]
|
||||
|
||||
predicted = probs.argmax(axis=1)
|
||||
confidence = probs.max(axis=1)
|
||||
direction_map = {0: "BUY", 1: "HOLD", 2: "SELL"}
|
||||
|
||||
out = pd.DataFrame(
|
||||
{
|
||||
"direction": [
|
||||
direction_map[p] if c >= 0.5 else "HOLD"
|
||||
for p, c in zip(predicted, confidence)
|
||||
],
|
||||
"confidence": confidence,
|
||||
"p_buy": probs[:, 0],
|
||||
"p_hold": probs[:, 1],
|
||||
"p_sell": probs[:, 2],
|
||||
},
|
||||
index=df.index,
|
||||
)
|
||||
# Rows with incomplete feature windows → no prediction (HOLD)
|
||||
out.loc[~valid_mask, ["direction", "confidence"]] = ["HOLD", 0.0]
|
||||
return out
|
||||
|
||||
+224
-47
@@ -1,13 +1,20 @@
|
||||
"""
|
||||
Backtesting Engine for XAUUSD AI Trading Bot
|
||||
Backtesting Engine for XAUUSD AI Trading Bot — Live-Parity Edition
|
||||
|
||||
Simulates realistic multi-strategy execution against historical price data:
|
||||
- Zero lookahead bias (evaluates bars sequentially)
|
||||
- Dynamic ATR-based Stop Loss & Take Profit
|
||||
- Trailing stop simulation
|
||||
- Zero lookahead bias (evaluates bars sequentially, trailing-only features)
|
||||
- Real XGBoost AI prediction per bar (predict_batch lookup, same path as live signal_classifier.predict)
|
||||
- Real confluence scoring incl. dynamic RANGING/HIGH-vol threshold deltas (mirrors live aggregator)
|
||||
- Temporal Self-Attention alignment (same AttentionMarketScorer as live)
|
||||
- Dynamic ATR-based Stop Loss & Take Profit + breakeven + trailing (capped at TP)
|
||||
- Entry cooldown + max-hold timeout exits
|
||||
- Realistic spread & slippage modeling
|
||||
- Confluence score filtering
|
||||
- Detailed quantitative performance reporting
|
||||
|
||||
Known deviations from live (documented): single concurrent position (live allows up to
|
||||
max_concurrent=4, so concurrency compression understates/overstates risk depending on
|
||||
correlation); trade_learner mistake-guard bypassed (live weights are garbage — see plan
|
||||
phase 3); sentiment fixed at NEUTRAL/0 unless overridden.
|
||||
"""
|
||||
|
||||
import math
|
||||
@@ -37,19 +44,43 @@ class Backtester:
|
||||
lot_size: float = 0.02,
|
||||
spread_points: float = 0.25, # ~$0.25 on gold
|
||||
slippage_points: float = 0.05,
|
||||
min_confluence_score: float = 65.0,
|
||||
min_confluence_score: float = 50.0,
|
||||
trailing_stop_enabled: bool = True,
|
||||
# ── Live-parity addons ───────────────────────────────────────────
|
||||
ai_lookup: Optional[pd.DataFrame] = None, # per-bar AI (direction/confidence), positional with df
|
||||
regime_plan: Optional[pd.DataFrame] = None, # per-bar regime (regime/volatility/adx), positional with df
|
||||
sentiment: Optional[dict] = None, # {"direction","score"} for aggregation
|
||||
attention_enabled: bool = True, # mirror live AttentionMarketScorer
|
||||
cooldown_bars: int = 0, # bars to wait after an exit before same-strategy re-entry
|
||||
max_hold_bars: int = 0, # 0 = disabled; else time-based exit at close
|
||||
breakeven_rr: float = 0.0, # 0 = disabled; else move SL to entry at this RR of original SL
|
||||
atr_trail_multiplier: float = 1.2, # trail distance = multiplier × bar ATR
|
||||
trail_trigger_atr: float = 1.2, # start trailing after this much unrealized gain (× ATR)
|
||||
tick: float = 0.01, # XAUUSD price tick
|
||||
):
|
||||
self.initial_balance = initial_balance
|
||||
self.lot_size = lot_size
|
||||
self.spread = spread_points
|
||||
self.slippage = slippage_points
|
||||
self.min_confluence = min_confluence_score
|
||||
self.min_confluence = min_confluence_score # fallback; live prefers per-strategy property
|
||||
self.trailing_enabled = trailing_stop_enabled
|
||||
self.ai_lookup = ai_lookup
|
||||
self.regime_plan = regime_plan
|
||||
self.sentiment = sentiment or {"direction": "NEUTRAL", "score": 0.0}
|
||||
self.attention_enabled = attention_enabled
|
||||
self.cooldown_bars = cooldown_bars
|
||||
self.max_hold_bars = max_hold_bars
|
||||
self.breakeven_rr = breakeven_rr
|
||||
self.atr_trail_multiplier = atr_trail_multiplier
|
||||
self.trail_trigger_atr = trail_trigger_atr
|
||||
self.tick = tick
|
||||
|
||||
self.ta = TechnicalAnalyzer()
|
||||
self.regime_detector = RegimeDetector()
|
||||
self.aggregator = SignalAggregator()
|
||||
# Pure config weights for the gate — never the corrupted live adaptive ones
|
||||
self.aggregator = SignalAggregator(use_adaptive_weights=False)
|
||||
from ai.attention_scorer import AttentionMarketScorer
|
||||
self.attention_scorer = AttentionMarketScorer(window=25)
|
||||
|
||||
self.strategies = {
|
||||
"scalping": ScalpingStrategy(),
|
||||
@@ -109,6 +140,9 @@ class Backtester:
|
||||
def run(self, df: pd.DataFrame, warmup_bars: int = 200) -> PerformanceMetrics:
|
||||
"""
|
||||
Run the backtest across historical bars.
|
||||
|
||||
`ai_lookup` and `regime_plan` (if set on the instance) must be positionally
|
||||
aligned to `df` (same number of rows, same order).
|
||||
"""
|
||||
if len(df) <= warmup_bars:
|
||||
raise ValueError(f"Dataframe length ({len(df)}) must exceed warmup_bars ({warmup_bars})")
|
||||
@@ -118,23 +152,84 @@ class Backtester:
|
||||
# Precalculate technical indicators on the full dataset
|
||||
df_indicators = self.ta.add_all_indicators(df.copy())
|
||||
|
||||
# ── AI lookup: time-based (M5 bar → last CLOSED M15 signal) ────────
|
||||
# Model features at M15 row j use trailing windows only, so the signal
|
||||
# is known at bar close = open + 15m. Live bot also evaluates on closed
|
||||
# M15 bars. If no "time" column exists (synthetic tests), fall back to
|
||||
# positional pairing with the M5 rows.
|
||||
def _ai_lookup_fn(src: Optional[pd.DataFrame]):
|
||||
if src is None or len(src) == 0:
|
||||
return None
|
||||
src = src.reset_index(drop=True)
|
||||
if "time" in src.columns:
|
||||
opens = pd.to_datetime(src["time"]).values
|
||||
closes = opens + np.timedelta64(15, "m") # M15 close = signal-available time
|
||||
order = np.argsort(closes)
|
||||
closes_sorted = closes[order]
|
||||
dirs = src["direction"].values[order]
|
||||
confs = src["confidence"].values.astype(float)[order]
|
||||
|
||||
def lookup(bar_time):
|
||||
pos = np.searchsorted(closes_sorted, np.datetime64(bar_time), side="right") - 1
|
||||
if pos < 0:
|
||||
return {"direction": "HOLD", "confidence": 0.0}
|
||||
return {"direction": str(dirs[pos]), "confidence": float(confs[pos])}
|
||||
|
||||
return lookup
|
||||
else:
|
||||
# positional fallback (tests): assume src rows align with M5 rows
|
||||
rows = list(zip(src["direction"].astype(str), src["confidence"].astype(float)))
|
||||
pos_idx = [0]
|
||||
|
||||
def lookup(bar_time):
|
||||
row = rows[min(pos_idx[0], len(rows) - 1)]
|
||||
pos_idx[0] += 1
|
||||
return {"direction": row[0], "confidence": row[1]}
|
||||
|
||||
return lookup
|
||||
ai_lookup_fn = _ai_lookup_fn(self.ai_lookup)
|
||||
|
||||
def _regime_iter():
|
||||
if self.regime_plan is not None and len(self.regime_plan) == len(df_indicators):
|
||||
plan = self.regime_plan.reset_index(drop=True)
|
||||
for _, row in plan.iterrows():
|
||||
yield {
|
||||
"regime": row["regime"],
|
||||
"volatility": row["volatility"],
|
||||
"adx": float(row["adx"]) if pd.notna(row["adx"]) else 15.0,
|
||||
"should_trade": bool(row["should_trade"]) if "should_trade" in row else True,
|
||||
}
|
||||
else:
|
||||
while True:
|
||||
yield None
|
||||
|
||||
regime_gen = _regime_iter()
|
||||
|
||||
closed_trades: List[BacktestTrade] = []
|
||||
open_position: Optional[dict] = None
|
||||
trade_counter = 0
|
||||
running_equity = self.initial_balance
|
||||
last_exit_bar: Dict[str, int] = {name: -10**9 for name in self.strategies}
|
||||
|
||||
# Point value for XAUUSD (1 standard lot = 100 oz, so 1 point move = $100 on 1.0 lot, $100 * lot_size)
|
||||
# Point value for XAUUSD (1 standard lot = 100 oz, 1 point move = $100 on 1.0 lot)
|
||||
dollar_per_point = 100.0 * self.lot_size
|
||||
|
||||
for i in range(warmup_bars, len(df_indicators)):
|
||||
current_bar = df_indicators.iloc[i]
|
||||
prev_bar = df_indicators.iloc[i - 1]
|
||||
sub_df = df_indicators.iloc[:i + 1]
|
||||
|
||||
current_high = current_bar["high"]
|
||||
current_low = current_bar["low"]
|
||||
current_close = current_bar["close"]
|
||||
current_high = float(current_bar["high"])
|
||||
current_low = float(current_bar["low"])
|
||||
current_close = float(current_bar["close"])
|
||||
current_time = current_bar.get("time", i)
|
||||
atr = float(current_bar.get("atr", 3.0))
|
||||
if not math.isfinite(atr) or atr <= 0:
|
||||
atr = 3.0
|
||||
|
||||
# ── Per-bar exogenous signals (time-aligned) ────────────────
|
||||
ai_prediction = {"direction": "HOLD", "confidence": 0.0}
|
||||
if ai_lookup_fn is not None:
|
||||
ai_prediction = ai_lookup_fn(current_time)
|
||||
|
||||
# ── 1. Manage existing open position ───────────────────────
|
||||
if open_position is not None:
|
||||
@@ -144,48 +239,76 @@ class Backtester:
|
||||
entry_price = pos["entry_price"]
|
||||
sl = pos["stop_loss"]
|
||||
tp = pos["take_profit"]
|
||||
sl_orig = pos["stop_loss_original"]
|
||||
|
||||
closed = False
|
||||
exit_price = 0.0
|
||||
exit_reason = ""
|
||||
|
||||
# Check SL and TP within bar
|
||||
if direction == "BUY":
|
||||
# Check Stop Loss first (conservative)
|
||||
# Timeout exit path (smart-exit culture: don't hold forever)
|
||||
if self.max_hold_bars > 0 and pos["bars_held"] >= self.max_hold_bars:
|
||||
exit_price = round(current_close, 2)
|
||||
exit_reason = "TIMEOUT"
|
||||
closed = True
|
||||
elif direction == "BUY":
|
||||
# Stop Loss first (conservative within-bar ordering)
|
||||
if current_low <= sl:
|
||||
exit_price = sl - self.slippage
|
||||
exit_price = round(sl - self.slippage, 2)
|
||||
exit_reason = "STOP_LOSS"
|
||||
closed = True
|
||||
elif current_high >= tp:
|
||||
exit_price = tp
|
||||
exit_price = round(tp, 2)
|
||||
exit_reason = "TAKE_PROFIT"
|
||||
closed = True
|
||||
elif self.trailing_enabled:
|
||||
# Update Trailing Stop if price moved in profit by 1.5 ATR
|
||||
gain = current_close - entry_price
|
||||
if gain > 1.5 * atr:
|
||||
new_sl = current_close - 1.2 * atr
|
||||
if new_sl > pos["stop_loss"]:
|
||||
pos["stop_loss"] = new_sl
|
||||
|
||||
elif direction == "SELL":
|
||||
if current_high >= sl:
|
||||
exit_price = sl + self.slippage
|
||||
exit_price = round(sl + self.slippage, 2)
|
||||
exit_reason = "STOP_LOSS"
|
||||
closed = True
|
||||
elif current_low <= tp:
|
||||
exit_price = tp
|
||||
exit_price = round(tp, 2)
|
||||
exit_reason = "TAKE_PROFIT"
|
||||
closed = True
|
||||
elif self.trailing_enabled:
|
||||
gain = entry_price - current_close
|
||||
if gain > 1.5 * atr:
|
||||
new_sl = current_close + 1.2 * atr
|
||||
if new_sl < pos["stop_loss"]:
|
||||
pos["stop_loss"] = new_sl
|
||||
|
||||
if not closed:
|
||||
# ── Breakeven: lock in entry once price reaches breakeven_rr × original SL ──
|
||||
if self.breakeven_rr > 0 and sl_orig is not None:
|
||||
sl_distance = abs(entry_price - sl_orig)
|
||||
if sl_distance > 0:
|
||||
if direction == "BUY":
|
||||
gain_now = current_close - entry_price
|
||||
if gain_now >= sl_distance * self.breakeven_rr:
|
||||
be = round(entry_price + self.tick, 2)
|
||||
if be > sl:
|
||||
sl = be
|
||||
else:
|
||||
gain_now = entry_price - current_close
|
||||
if gain_now >= sl_distance * self.breakeven_rr:
|
||||
be = round(entry_price - self.tick, 2)
|
||||
if be < sl:
|
||||
sl = be
|
||||
|
||||
# ── Trailing stop: activate after trail_trigger_atr × ATR profit ──
|
||||
if self.trailing_enabled:
|
||||
if direction == "BUY":
|
||||
gain = current_close - entry_price
|
||||
if gain > self.trail_trigger_atr * atr:
|
||||
new_sl = round(current_close - self.atr_trail_multiplier * atr, 2)
|
||||
new_sl = min(new_sl, round(tp - self.tick, 2)) # never trail past TP
|
||||
if new_sl > sl:
|
||||
sl = new_sl
|
||||
else:
|
||||
gain = entry_price - current_close
|
||||
if gain > self.trail_trigger_atr * atr:
|
||||
new_sl = round(current_close + self.atr_trail_multiplier * atr, 2)
|
||||
new_sl = max(new_sl, round(tp + self.tick, 2)) # never trail past TP
|
||||
if new_sl < sl:
|
||||
sl = new_sl
|
||||
|
||||
pos["stop_loss"] = sl
|
||||
|
||||
if closed:
|
||||
# Calculate P&L
|
||||
# ── Calculate P&L ──────────────────────────────────
|
||||
if direction == "BUY":
|
||||
points_diff = exit_price - entry_price
|
||||
else:
|
||||
@@ -193,7 +316,7 @@ class Backtester:
|
||||
|
||||
pnl_dollars = points_diff * dollar_per_point
|
||||
pips = points_diff * 10.0 # 1 pip = 0.1 on gold
|
||||
pnl_pct = (pnl_dollars / self.initial_balance) * 100.0
|
||||
pnl_pct = (pnl_dollars / running_equity) * 100.0 if running_equity > 0 else 0.0
|
||||
|
||||
trade = BacktestTrade(
|
||||
trade_id=pos["trade_id"],
|
||||
@@ -203,7 +326,7 @@ class Backtester:
|
||||
entry_time=pos["entry_time"],
|
||||
exit_time=current_time,
|
||||
entry_price=round(entry_price, 2),
|
||||
exit_price=round(exit_price, 2),
|
||||
exit_price=exit_price,
|
||||
lot_size=self.lot_size,
|
||||
pnl=round(pnl_dollars, 2),
|
||||
pnl_pips=round(pips, 1),
|
||||
@@ -213,35 +336,68 @@ class Backtester:
|
||||
bars_held=pos["bars_held"],
|
||||
)
|
||||
closed_trades.append(trade)
|
||||
running_equity += pnl_dollars
|
||||
last_exit_bar[pos["strategy"]] = i
|
||||
open_position = None
|
||||
|
||||
# ── 2. Evaluate new entries if no position is open ────────
|
||||
if open_position is None:
|
||||
regime = self.regime_detector.analyze(sub_df)
|
||||
if not regime.should_trade:
|
||||
regime = next(regime_gen)
|
||||
should_trade = True
|
||||
if regime is not None:
|
||||
should_trade = regime["should_trade"]
|
||||
else:
|
||||
regime_obj = self.regime_detector.analyze(sub_df)
|
||||
should_trade = regime_obj.should_trade
|
||||
regime = {
|
||||
"regime": regime_obj.regime.value,
|
||||
"volatility": regime_obj.volatility_label,
|
||||
"adx": regime_obj.adx,
|
||||
"should_trade": should_trade,
|
||||
"position_modifier": regime_obj.position_size_modifier,
|
||||
}
|
||||
|
||||
if not should_trade:
|
||||
continue
|
||||
|
||||
best_signal = None
|
||||
best_confluence = None
|
||||
|
||||
for strat_name, strat in self.strategies.items():
|
||||
if not self.regime_detector.is_strategy_recommended(strat_name, regime):
|
||||
# Entry cooldown: wait cooldown_bars after that strategy's last exit
|
||||
if self.cooldown_bars > 0 and i - last_exit_bar[strat_name] < self.cooldown_bars:
|
||||
continue
|
||||
|
||||
# Match live Step 6 regime recommendation gate
|
||||
if not self.regime_detector.is_strategy_recommended(strat_name, self._to_regime_obj(regime, sub_df)):
|
||||
continue
|
||||
|
||||
sig = strat.run(sub_df)
|
||||
if not sig.is_actionable:
|
||||
continue
|
||||
|
||||
# Confluence check
|
||||
# Mirror live Step 6: attention on the strategy's own frame
|
||||
attention_analysis = None
|
||||
if self.attention_enabled:
|
||||
attention_analysis = self.attention_scorer.analyze_alignment(
|
||||
df=sub_df,
|
||||
signal_direction=sig.direction.value,
|
||||
)
|
||||
|
||||
# Confluence — live passes is_recommended=True hardcoded + real regime keys
|
||||
confluence = self.aggregator.calculate_confluence(
|
||||
strategy_signal=sig,
|
||||
ai_prediction={"direction": "HOLD", "confidence": 0.0},
|
||||
sentiment={"direction": "NEUTRAL", "score": 0.0},
|
||||
ai_prediction=ai_prediction,
|
||||
sentiment=self.sentiment,
|
||||
regime_analysis={
|
||||
"is_recommended": True,
|
||||
"position_modifier": regime.position_size_modifier,
|
||||
"position_modifier": regime.get("position_modifier", 1.0),
|
||||
"regime": regime["regime"],
|
||||
"adx": regime.get("adx", 15.0),
|
||||
"volatility": regime["volatility"],
|
||||
},
|
||||
min_score=self.min_confluence,
|
||||
attention_analysis=attention_analysis,
|
||||
min_score=strat.min_confluence_score,
|
||||
)
|
||||
|
||||
if confluence.should_execute:
|
||||
@@ -250,7 +406,11 @@ class Backtester:
|
||||
best_confluence = confluence
|
||||
|
||||
if best_signal and best_confluence and best_confluence.should_execute:
|
||||
direction = best_signal.direction
|
||||
# Store string direction (live stores signal.direction.value) — the exit
|
||||
# block compares against "BUY"/"SELL" strings. Storing the raw Enum here
|
||||
# silently disabled every exit: "Enum == str" is always False, so no trade
|
||||
# ever closed and open_position blocked all subsequent entries.
|
||||
direction = best_signal.direction.value
|
||||
raw_price = current_close
|
||||
|
||||
# Apply realistic execution spread & slippage
|
||||
@@ -271,6 +431,7 @@ class Backtester:
|
||||
"entry_time": current_time,
|
||||
"entry_price": entry_price,
|
||||
"stop_loss": sl,
|
||||
"stop_loss_original": sl,
|
||||
"take_profit": tp,
|
||||
"confluence_score": best_confluence.total_score,
|
||||
"bars_held": 0,
|
||||
@@ -281,6 +442,22 @@ class Backtester:
|
||||
self._print_report(metrics)
|
||||
return metrics
|
||||
|
||||
def _to_regime_obj(self, regime: dict, sub_df: pd.DataFrame):
|
||||
"""Wrap an exogenous regime dict as a RegimeAnalysis for is_strategy_recommended."""
|
||||
try:
|
||||
from strategies.regime_detector import MarketRegime, RegimeAnalysis
|
||||
if not isinstance(regime.get("regime"), str):
|
||||
return self.regime_detector.analyze(sub_df)
|
||||
return RegimeAnalysis(
|
||||
regime=MarketRegime(regime["regime"]),
|
||||
adx_value=float(regime.get("adx", 15.0)),
|
||||
volatility_label=regime.get("volatility", "NORMAL"),
|
||||
position_size_modifier=regime.get("position_modifier", 1.0),
|
||||
should_trade=regime.get("should_trade", True),
|
||||
)
|
||||
except Exception:
|
||||
return self.regime_detector.analyze(sub_df)
|
||||
|
||||
def _print_report(self, m: PerformanceMetrics):
|
||||
"""Display a formatted quantitative performance report."""
|
||||
border = "=" * 62
|
||||
@@ -319,4 +496,4 @@ class Backtester:
|
||||
lines.append(" (No completed trades executed)")
|
||||
lines.append(border)
|
||||
lines.append("")
|
||||
print("\n".join(lines))
|
||||
print("\n".join(lines))
|
||||
@@ -73,16 +73,25 @@ class BaseStrategy(ABC):
|
||||
self.enabled = params.get("enabled", True)
|
||||
self._last_active_mode: Optional[str] = None
|
||||
self._min_confluence_score_override: Optional[float] = None
|
||||
self._min_technical_score_override: Optional[float] = None
|
||||
self._default_sessions = params.get("active_sessions", [])
|
||||
self._last_signal: Optional[TradingSignal] = None
|
||||
|
||||
def _check_mode_sync(self):
|
||||
"""Ensure overrides are reset if the trading mode changes."""
|
||||
"""Ensure overrides are reset if the trading mode changes.
|
||||
|
||||
Only clear on a genuine mode *transition*. `_last_active_mode` starts
|
||||
as None, so naively treating None != current_mode as a change wiped
|
||||
overrides set before the first property read — e.g. the backtester's
|
||||
grid injection of min_confluence_score silently fell back to the YAML
|
||||
default (every threshold produced identical results).
|
||||
"""
|
||||
try:
|
||||
from config.settings import get_settings
|
||||
current_mode = get_settings().active_mode
|
||||
if getattr(self, "_last_active_mode", None) != current_mode:
|
||||
self._last_active_mode = current_mode
|
||||
prev = getattr(self, "_last_active_mode", None)
|
||||
self._last_active_mode = current_mode
|
||||
if prev is not None and prev != current_mode:
|
||||
self._min_confluence_score_override = None
|
||||
except Exception:
|
||||
pass
|
||||
@@ -135,6 +144,8 @@ class BaseStrategy(ABC):
|
||||
@property
|
||||
def min_technical_score(self) -> float:
|
||||
"""Threshold for the strategy itself to emit a directional signal instead of HOLD."""
|
||||
if self._min_technical_score_override is not None:
|
||||
return self._min_technical_score_override
|
||||
try:
|
||||
from config.settings import get_settings
|
||||
settings = get_settings()
|
||||
@@ -148,6 +159,10 @@ class BaseStrategy(ABC):
|
||||
pass
|
||||
return 35.0
|
||||
|
||||
@min_technical_score.setter
|
||||
def min_technical_score(self, val: float):
|
||||
self._min_technical_score_override = val
|
||||
|
||||
@abstractmethod
|
||||
def analyze(self, df: pd.DataFrame) -> dict:
|
||||
"""
|
||||
|
||||
@@ -43,10 +43,11 @@ class SignalAggregator:
|
||||
The confluence score determines whether a trade should be executed.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
def __init__(self, use_adaptive_weights: bool = True):
|
||||
settings = get_settings()
|
||||
weights = settings.confluence_weights
|
||||
|
||||
self.use_adaptive_weights = use_adaptive_weights
|
||||
self.technical_weight = weights.get("technical_weight", 40)
|
||||
self.ai_weight = weights.get("ai_prediction_weight", 25)
|
||||
self.sentiment_weight = weights.get("sentiment_weight", 20)
|
||||
@@ -83,16 +84,20 @@ class SignalAggregator:
|
||||
Returns:
|
||||
ConfluenceResult with final score and execution decision.
|
||||
"""
|
||||
# Dynamically synchronize with self-learning adaptive weights
|
||||
try:
|
||||
from ai.trade_learner import trade_learner
|
||||
cw = trade_learner.get_confluence_weights()
|
||||
self.technical_weight = cw.get("technical", self.technical_weight)
|
||||
self.ai_weight = cw.get("ai_prediction", self.ai_weight)
|
||||
self.sentiment_weight = cw.get("sentiment", self.sentiment_weight)
|
||||
self.regime_weight = cw.get("regime", self.regime_weight)
|
||||
except Exception:
|
||||
pass
|
||||
# Dynamically synchronize with self-learning adaptive weights.
|
||||
# Backtesting uses use_adaptive_weights=False so the gate measures the
|
||||
# pure config weights — live garbage adaptive weights (leak #3) would
|
||||
# otherwise contaminate every backtest confluence score.
|
||||
if self.use_adaptive_weights:
|
||||
try:
|
||||
from ai.trade_learner import trade_learner
|
||||
cw = trade_learner.get_confluence_weights()
|
||||
self.technical_weight = cw.get("technical", self.technical_weight)
|
||||
self.ai_weight = cw.get("ai_prediction", self.ai_weight)
|
||||
self.sentiment_weight = cw.get("sentiment", self.sentiment_weight)
|
||||
self.regime_weight = cw.get("regime", self.regime_weight)
|
||||
except Exception:
|
||||
pass
|
||||
reasons = list(strategy_signal.reasons)
|
||||
|
||||
# ── 1. Technical Score (from strategy) ───────────────────────────
|
||||
|
||||
Reference in New Issue
Block a user