gavindiaz/fx-quant
1
0
Fork 0
mirror of https://github.com/BrentNeale1/fx-quant.git synced 2026-08-14 18:48:07 +00:00
Code Issues Packages Projects Releases Wiki Activity

Phase 1 complete: S3-S6 strategies, S4 variant analysis, learnings doc

Browse Source 
- S3 Key Level Breakout: best performer (52-53% WR, PF ~1.0 on JPY crosses)
- S4 EMA Ribbon: tested 7 variants (D/E/F/F-v2/G/G-Minimal), exhausted
  - Only EUR_AUD S4-F marginally profitable (PF 1.06)
  - Detailed filter funnel analysis revealed contradictory filter stacking
- S5 Momentum Exhaustion: extended to 5 pairs, PF 0.43-0.77
- S6 EMA Bounce: 59-60% WR but PF 0.83-0.84, needs SL/TP restructuring
- Added STRATEGY_LEARNINGS.md with design principles and next steps
- Added M5 data downloader for 3-timeframe strategies
- Updated README with full strategy scorecard

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
Brent Neale
2026-02-18 20:42:16 +10:00
co-authored by Claude Opus 4.6
parent dce54845c2
commit edbe359d1b
88 changed files with 12570 additions and 2963 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/strategies_pkg/__init__.py
+15 -8
View File
@@ -1,31 +1,38 @@
from .s1_ma_breakout import S1_MA_Breakout
from .s2_vwap_reversal import S2_VWAP_Reversal
# S2 disabled: 20-25% win rate, 26 consecutive losses, needs full redesign
# from .s2_vwap_reversal import S2_VWAP_Reversal
from .s3_key_level_breakout import S3_KeyLevel_Breakout
from .s4_ema_ribbon import S4_EMA_Ribbon
from .s5_momentum_exhaustion import S5_Momentum_Exhaustion
from .s6_ema_bounce import S6_EMA_Bounce
STRATEGIES = {
1: S1_MA_Breakout,
2: S2_VWAP_Reversal,
# 2: S2_VWAP_Reversal, # DISABLED
3: S3_KeyLevel_Breakout,
4: S4_EMA_Ribbon,
5: S5_Momentum_Exhaustion,
6: S6_EMA_Bounce,
}
# Which pairs each strategy trades
# Allowed universe: GBP_AUD, EUR_AUD, EUR_CAD, GBP_CAD, GBP_USD, EUR_USD
# Removed: EUR_GBP (PF 0.38-0.43), EUR_NZD (S1 lost $25k)
STRATEGY_PAIRS = {
1: ["GBP_AUD", "EUR_AUD", "EUR_CAD", "EUR_NZD"],
2: ["GBP_USD", "EUR_USD", "GBP_JPY", "USD_JPY"],
3: ["GBP_JPY", "USD_JPY", "GBP_USD", "EUR_GBP"],
4: ["GBP_AUD", "EUR_AUD", "EUR_GBP"],
5: ["GBP_AUD", "EUR_AUD", "EUR_GBP", "GBP_CAD", "EUR_CAD"],
1: ["GBP_AUD", "EUR_AUD", "EUR_CAD", "GBP_CAD"],
# 2: DISABLED
3: ["GBP_JPY", "USD_JPY", "GBP_USD"],
4: ["GBP_AUD", "EUR_AUD", "GBP_JPY"],
5: ["GBP_AUD", "EUR_AUD", "GBP_JPY", "USD_JPY", "GBP_USD"],
6: ["GBP_AUD"], # Initial test — expand to EUR_AUD, GBP_USD if passing
}
# Primary and filter timeframes
STRATEGY_TIMEFRAMES = {
1: {"primary": "M15", "filter": "H1"},
2: {"primary": "M15", "filter": None},
# 2: DISABLED
3: {"primary": "H1", "filter": None}, # Uses internal key level detection
4: {"primary": "M15", "filter": "H1"},
5: {"primary": "M15", "filter": "H1"},
6: {"primary": "M15", "filter": "H1"},
}
src/strategies_pkg/base.py
+3
View File
@@ -7,6 +7,9 @@ class BaseStrategy:
strategy_id: int = 0
name: str = "BaseStrategy"
def __init__(self):
self.htf_data = None
def check_signal(self, data: pd.DataFrame, idx: int,
current: pd.Series,
htf_row: Optional[pd.Series] = None) -> Optional[dict]:
src/strategies_pkg/s1_ma_breakout.py
+433 -139
View File
@@ -1,33 +1,375 @@
"""
Strategy 1: MA Breakout-Retest (Option 3 - no trendlines).
Strategy 1: Trendline Breakout-Retest.
Uses MA structure + key level breaks + candle confirmation.
Entry TF: M15, Filter TF: H1 (200 SMA directional filter).
4-step sequence identified on H1 chart with M15 entry:
1. Identify trendline on H1 (3+ swing touches, linear regression)
2. Breakout: H1 close beyond trendline with conviction
3. Move away: Price moves away from trendline (confirms real break)
4. Retest + Entry: Price pulls back to broken trendline on M15 → engulfing candle
Entry conditions (LONG):
- EMA 50 > EMA 100 > EMA 200 (trend alignment)
- Price pulls back to EMA 50 zone (within 1.0x ATR)
- Bullish confirmation candle (close > open, close > prev close, body > 30% range)
- H1 close > H1 200 SMA (HTF filter)
- Session filter: London/NY hours only (08:00-17:00 UTC)
Entry conditions (all must be true):
- State machine in RETEST phase
- M15 engulfing candle
- M15 close within 1.0x ATR of projected trendline price
- M15 EMA 50 aligns with direction
- Session: London/NY overlap (13:00-16:00 UTC)
- Confluence >= 2
Boosters (confluence 0-5):
- Volume above 20-period average
- RSI between 40-60 (not overextended)
- MACD histogram positive and rising
- ADX > 20 (trending)
- Price above session VWAP
SL: Projected trendline price +/- 0.5x ATR (behind the trendline)
TP1: Previous swing high/low (structure), fallback 1.5x ATR
TP2: Next key level or 2.5x ATR
TP3: 2x TP1 distance or 4x ATR (runner)
"""
from typing import Optional
import numpy as np
import pandas as pd
from .base import BaseStrategy
from ..indicators.technical import (
fit_trendline, project_trendline, swing_highs, swing_lows,
is_bullish_engulfing, is_bearish_engulfing, identify_key_levels,
)
class S1_MA_Breakout(BaseStrategy):
"""Trendline Breakout-Retest strategy (renamed from MA Breakout)."""
strategy_id = 1
name = "S1_MA_Breakout_Retest"
name = "S1_Trendline_Breakout_Retest"
def __init__(self):
super().__init__()
self._trendlines = {"resistance": None, "support": None}
self._tl_cache_idx = -1
self._state = {
"phase": "IDLE",
"direction": None,
"break_bar_idx": None,
"break_price": None,
"trendline": None,
"max_dist": 0.0,
"bars_since_break": 0,
}
# Performance: cached HTF timestamps for searchsorted
self._htf_ts_cache = None
self._last_htf_cutoff = -1 # tracks H1 bar changes for timeout
def _htf_cutoff(self, htf: pd.DataFrame, ts: pd.Timestamp) -> int:
"""Return number of H1 bars strictly before ts, using searchsorted."""
if self._htf_ts_cache is None:
self._htf_ts_cache = htf.index
# Normalize tz: strip tz from ts if HTF index is tz-naive, or vice versa
if self._htf_ts_cache.tz is None and hasattr(ts, 'tz') and ts.tz is not None:
ts = ts.tz_localize(None)
elif self._htf_ts_cache.tz is not None and (not hasattr(ts, 'tz') or ts.tz is None):
ts = ts.tz_localize(self._htf_ts_cache.tz)
return int(self._htf_ts_cache.searchsorted(ts, side="left"))
# ------------------------------------------------------------------
# Trendline Detection (runs on H1 data)
# ------------------------------------------------------------------
def _detect_trendlines(self, htf: pd.DataFrame, n_valid: int):
"""
Detect resistance and support trendlines from H1 swing points.
n_valid = number of H1 bars before current M15 timestamp.
Recalculates every 20 H1 bars.
"""
if n_valid < 50:
return
if (self._tl_cache_idx >= 0 and
n_valid - self._tl_cache_idx < 20):
return
self._tl_cache_idx = n_valid
# Use last 200 H1 bars (no lookahead: only first n_valid bars)
start = max(0, n_valid - 200)
window = htf.iloc[start:n_valid]
offset = start # absolute index of window[0] in htf
# Detect swing highs and lows
sh_mask = swing_highs(window, lookback=5)
sl_mask = swing_lows(window, lookback=5)
# Resistance trendline from swing highs
sh_indices = np.where(sh_mask.values)[0]
if len(sh_indices) >= 3:
recent_sh = sh_indices[-8:]
sh_prices = window["high"].values[recent_sh]
tl = fit_trendline(recent_sh, sh_prices)
if tl is not None:
tl["window_offset"] = offset
self._trendlines["resistance"] = tl
else:
self._trendlines["resistance"] = None
# Support trendline from swing lows
sl_indices_arr = np.where(sl_mask.values)[0]
if len(sl_indices_arr) >= 3:
recent_sl = sl_indices_arr[-8:]
sl_prices = window["low"].values[recent_sl]
tl = fit_trendline(recent_sl, sl_prices)
if tl is not None:
tl["window_offset"] = offset
self._trendlines["support"] = tl
else:
self._trendlines["support"] = None
def _project_tl_at_htf_bar(self, tl: dict, htf_bar_idx: int) -> float:
"""Project trendline price at a given absolute HTF bar index."""
window_rel_idx = htf_bar_idx - tl["window_offset"]
return tl["slope"] * window_rel_idx + tl["intercept"]
# ------------------------------------------------------------------
# State Machine
# ------------------------------------------------------------------
def _update_state_machine(self, htf: pd.DataFrame, n_valid: int):
"""
Check for breakout transitions on H1 data.
n_valid = number of H1 bars strictly before current M15 timestamp.
"""
if n_valid < 2:
return
last_h1_idx = n_valid - 1
last_h1 = htf.iloc[last_h1_idx]
# H1 ATR for thresholds
h1_atr = last_h1.get("atr_14", 0)
if h1_atr <= 0 or np.isnan(h1_atr):
return
phase = self._state["phase"]
# Track H1 bar changes for timeout counter
if phase != "IDLE":
if last_h1_idx != self._last_htf_cutoff:
self._last_htf_cutoff = last_h1_idx
self._state["bars_since_break"] += 1
if self._state["bars_since_break"] > 50:
self._reset_state()
return
if phase == "IDLE":
# Check for breakout above resistance -> LONG
res_tl = self._trendlines.get("resistance")
if res_tl is not None:
tl_price = self._project_tl_at_htf_bar(res_tl, last_h1_idx)
threshold = tl_price + 0.3 * h1_atr
h1_close = last_h1["close"]
h1_open = last_h1["open"]
body_low = min(h1_close, h1_open)
if h1_close > threshold and body_low > tl_price:
self._state = {
"phase": "MOVE_AWAY",
"direction": "LONG",
"break_bar_idx": last_h1_idx,
"break_price": h1_close,
"trendline": res_tl.copy(),
"max_dist": h1_close - tl_price,
"bars_since_break": 0,
"h1_atr": h1_atr,
}
self._last_htf_cutoff = last_h1_idx
return
# Check for breakout below support -> SHORT
sup_tl = self._trendlines.get("support")
if sup_tl is not None:
tl_price = self._project_tl_at_htf_bar(sup_tl, last_h1_idx)
threshold = tl_price - 0.3 * h1_atr
h1_close = last_h1["close"]
h1_open = last_h1["open"]
body_high = max(h1_close, h1_open)
if h1_close < threshold and body_high < tl_price:
self._state = {
"phase": "MOVE_AWAY",
"direction": "SHORT",
"break_bar_idx": last_h1_idx,
"break_price": h1_close,
"trendline": sup_tl.copy(),
"max_dist": tl_price - h1_close,
"bars_since_break": 0,
"h1_atr": h1_atr,
}
self._last_htf_cutoff = last_h1_idx
return
elif phase == "MOVE_AWAY":
tl = self._state["trendline"]
tl_price = self._project_tl_at_htf_bar(tl, last_h1_idx)
h1_close = last_h1["close"]
state_atr = self._state.get("h1_atr", h1_atr)
if self._state["direction"] == "LONG":
dist = h1_close - tl_price
if dist > self._state["max_dist"]:
self._state["max_dist"] = dist
if self._state["max_dist"] >= 0.5 * state_atr and dist < self._state["max_dist"]:
self._state["phase"] = "RETEST"
else: # SHORT
dist = tl_price - h1_close
if dist > self._state["max_dist"]:
self._state["max_dist"] = dist
if self._state["max_dist"] >= 0.5 * state_atr and dist < self._state["max_dist"]:
self._state["phase"] = "RETEST"
def _reset_state(self):
self._state = {
"phase": "IDLE",
"direction": None,
"break_bar_idx": None,
"break_price": None,
"trendline": None,
"max_dist": 0.0,
"bars_since_break": 0,
}
# ------------------------------------------------------------------
# Confluence Scoring (0-5)
# ------------------------------------------------------------------
def _calc_confluence(self, data: pd.DataFrame, idx: int,
current: pd.Series, direction: str,
tl: dict) -> int:
confluence = 0
# Trendline R-squared > 0.90
if tl.get("r_squared", 0) > 0.90:
confluence += 1
# Touch count >= 4
if tl.get("touch_count", 0) >= 4:
confluence += 1
# Volume above 20-period average
if "volume" in current.index:
vol_avg = data["volume"].iloc[max(0, idx - 20):idx].mean()
if vol_avg > 0 and current["volume"] > vol_avg:
confluence += 1
# RSI between 40-60
rsi_val = current.get("rsi_14", 50)
if not np.isnan(rsi_val) and 40 <= rsi_val <= 60:
confluence += 1
# MACD histogram confirms direction
macd_h = current.get("macd_hist", 0)
if not np.isnan(macd_h):
if direction == "LONG" and macd_h > 0:
confluence += 1
elif direction == "SHORT" and macd_h < 0:
confluence += 1
return min(confluence, 5)
# ------------------------------------------------------------------
# Find structure-based TP levels from H1 data
# ------------------------------------------------------------------
def _find_structure_tp(self, htf: pd.DataFrame, n_valid: int,
direction: str, entry_price: float,
atr_val: float) -> tuple:
"""Find TP levels based on H1 swing structure and key levels."""
if n_valid < 50:
if direction == "LONG":
return (entry_price + 1.5 * atr_val,
entry_price + 2.5 * atr_val,
entry_price + 4.0 * atr_val)
else:
return (entry_price - 1.5 * atr_val,
entry_price - 2.5 * atr_val,
entry_price - 4.0 * atr_val)
start = max(0, n_valid - 100)
window = htf.iloc[start:n_valid]
if direction == "LONG":
# TP1: previous swing high above entry
sh_mask = swing_highs(window, lookback=5)
sh_prices = window.loc[sh_mask, "high"]
above = sh_prices[sh_prices > entry_price].sort_values()
tp1 = above.iloc[0] if len(above) > 0 else entry_price + 1.5 * atr_val
# TP2: next key level above TP1, or 2.5x ATR
levels = identify_key_levels(window, lookback=5, min_touches=2)
level_prices = [lv[0] for lv in levels if lv[0] > tp1]
tp2 = min(level_prices) if level_prices else entry_price + 2.5 * atr_val
# TP3: 2x TP1 distance or 4x ATR (runner)
tp1_dist = tp1 - entry_price
tp3 = entry_price + max(2.0 * tp1_dist, 4.0 * atr_val)
else: # SHORT
sl_mask = swing_lows(window, lookback=5)
sl_prices = window.loc[sl_mask, "low"]
below = sl_prices[sl_prices < entry_price].sort_values(ascending=False)
tp1 = below.iloc[0] if len(below) > 0 else entry_price - 1.5 * atr_val
levels = identify_key_levels(window, lookback=5, min_touches=2)
level_prices = [lv[0] for lv in levels if lv[0] < tp1]
tp2 = max(level_prices) if level_prices else entry_price - 2.5 * atr_val
tp1_dist = entry_price - tp1
tp3 = entry_price - max(2.0 * tp1_dist, 4.0 * atr_val)
# Ensure TP ordering makes sense
if direction == "LONG":
tp1 = max(tp1, entry_price + 0.5 * atr_val)
tp2 = max(tp2, tp1 + 0.3 * atr_val)
tp3 = max(tp3, tp2 + 0.3 * atr_val)
else:
tp1 = min(tp1, entry_price - 0.5 * atr_val)
tp2 = min(tp2, tp1 - 0.3 * atr_val)
tp3 = min(tp3, tp2 - 0.3 * atr_val)
return tp1, tp2, tp3
# ------------------------------------------------------------------
# Reversal Pattern Detection
# ------------------------------------------------------------------
def _detect_reversal_pattern(self, data: pd.DataFrame, idx: int,
current: pd.Series, direction: str) -> str:
"""
Check for reversal patterns at the retest candle.
Returns pattern name ('engulfing', 'pin_bar', 'strong_close') or None.
"""
o, h, l, c = current["open"], current["high"], current["low"], current["close"]
body = abs(c - o)
full_range = h - l
if full_range <= 0:
return None
if direction == "LONG":
# 1. Bullish engulfing
if is_bullish_engulfing(data, idx):
return "engulfing"
# 2. Bullish pin bar: lower wick >= 2x body AND close in upper 25%
lower_wick = min(o, c) - l
if body > 0 and lower_wick >= 2 * body and c >= l + 0.75 * full_range:
return "pin_bar"
# 3. Strong bullish close: body > 60% of range AND close > open
if body > 0.60 * full_range and c > o:
return "strong_close"
else: # SHORT
# 1. Bearish engulfing
if is_bearish_engulfing(data, idx):
return "engulfing"
# 2. Bearish pin bar: upper wick >= 2x body AND close in lower 25%
upper_wick = h - max(o, c)
if body > 0 and upper_wick >= 2 * body and c <= l + 0.25 * full_range:
return "pin_bar"
# 3. Strong bearish close: body > 60% of range AND close < open
if body > 0.60 * full_range and c < o:
return "strong_close"
return None
# ------------------------------------------------------------------
# Main Signal Check
# ------------------------------------------------------------------
def check_signal(self, data: pd.DataFrame, idx: int,
current: pd.Series,
@@ -35,143 +377,95 @@ class S1_MA_Breakout(BaseStrategy):
if idx < 50:
return None
# Session filter: only London + NY (08:00-17:00 UTC)
hour = current.name.hour if hasattr(current.name, 'hour') else 0
if hour < 8 or hour >= 17:
htf = self.htf_data
if htf is None or len(htf) < 50:
return None
current_ts = current.name
# Efficient HTF cutoff via searchsorted
n_valid = self._htf_cutoff(htf, current_ts)
if n_valid < 50:
return None
# Detect trendlines and update state machine (always, for tracking)
self._detect_trendlines(htf, n_valid)
self._update_state_machine(htf, n_valid)
# Session filter: London + NY overlap (08:00-16:00 UTC)
hour = current_ts.hour if hasattr(current_ts, 'hour') else 0
if hour < 8 or hour >= 16:
return None
atr_val = current.get("atr_14", 0)
if atr_val <= 0 or np.isnan(atr_val):
return None
ema_50 = current.get("ema_50", np.nan)
ema_100 = current.get("ema_100", np.nan)
ema_200 = current.get("ema_200", np.nan)
if any(np.isnan(v) for v in [ema_50, ema_100, ema_200]):
# Only generate signals in RETEST phase
if self._state["phase"] != "RETEST":
return None
direction = self._state["direction"]
tl = self._state["trendline"]
# Project trendline price at current H1 bar
current_htf_idx = n_valid - 1
tl_price = self._project_tl_at_htf_bar(tl, current_htf_idx)
close = current["close"]
open_p = current["open"]
prev = data.iloc[idx - 1]
prev_close = prev["close"]
# Candle body filter: body must be > 30% of range (no dojis)
body = abs(close - open_p)
full_range = current["high"] - current["low"]
if full_range <= 0 or body / full_range < 0.3:
# M15 close within 1.5x ATR of projected trendline
dist_to_tl = abs(close - tl_price)
if dist_to_tl > 1.5 * atr_val:
return None
# LONG setup
if ema_50 > ema_100 > ema_200:
# HTF filter
if htf_row is not None:
htf_sma200 = htf_row.get("sma_200", np.nan)
if not np.isnan(htf_sma200) and htf_row.get("close", 0) <= htf_sma200:
return None
# M15 reversal pattern confirmation (engulfing, pin bar, or strong close)
entry_pattern = self._detect_reversal_pattern(data, idx, current, direction)
if entry_pattern is None:
return None
# Pullback to EMA 50 zone (within 1.0x ATR - tightened from 1.5x)
dist_to_ema50 = close - ema_50
if dist_to_ema50 < 0 or dist_to_ema50 > 1.0 * atr_val:
# EMA 50 alignment
ema_50 = current.get("ema_50", np.nan)
if np.isnan(ema_50):
return None
if direction == "LONG" and close <= ema_50:
return None
if direction == "SHORT" and close >= ema_50:
return None
# Confluence scoring
confluence = self._calc_confluence(data, idx, current, direction, tl)
if confluence < 2:
return None
# SL: behind the trendline (0.5x ATR past TL)
if direction == "LONG":
sl = tl_price - 0.5 * atr_val
# Validate SL is below entry (TL may have drifted above price)
if sl >= close:
return None
else:
sl = tl_price + 0.5 * atr_val
if sl <= close:
return None
# Bullish confirmation candle
if not (close > open_p and close > prev_close):
return None
# TP levels: structure-based from H1 data
tp1, tp2, tp3 = self._find_structure_tp(
htf, n_valid, direction, close, atr_val
)
# Not too far from EMAs (avoid chasing)
if close - ema_200 > 5 * atr_val:
return None
# Reset state after generating signal
self._reset_state()
confluence = self._calc_confluence(data, idx, current, "LONG")
# Require minimum confluence of 2
if confluence < 2:
return None
sl = current["low"] - 0.5 * atr_val
tp1 = close + 1.5 * atr_val
tp2 = close + 2.5 * atr_val
tp3 = close + 4.0 * atr_val
return {
"direction": "LONG",
"sl": sl, "tp1": tp1, "tp2": tp2, "tp3": tp3,
"confluence": confluence,
"tp_splits": (0.50, 0.30, 0.20),
"trail_atr_mult": 1.5,
"max_bars": 200,
}
# SHORT setup
if ema_50 < ema_100 < ema_200:
if htf_row is not None:
htf_sma200 = htf_row.get("sma_200", np.nan)
if not np.isnan(htf_sma200) and htf_row.get("close", 0) >= htf_sma200:
return None
dist_to_ema50 = ema_50 - close
if dist_to_ema50 < 0 or dist_to_ema50 > 1.0 * atr_val:
return None
if not (close < open_p and close < prev_close):
return None
if ema_200 - close > 5 * atr_val:
return None
confluence = self._calc_confluence(data, idx, current, "SHORT")
if confluence < 2:
return None
sl = current["high"] + 0.5 * atr_val
tp1 = close - 1.5 * atr_val
tp2 = close - 2.5 * atr_val
tp3 = close - 4.0 * atr_val
return {
"direction": "SHORT",
"sl": sl, "tp1": tp1, "tp2": tp2, "tp3": tp3,
"confluence": confluence,
"tp_splits": (0.50, 0.30, 0.20),
"trail_atr_mult": 1.5,
"max_bars": 200,
}
return None
def _calc_confluence(self, data, idx, current, direction):
confluence = 0
# Volume above average
if "volume" in current.index:
vol_avg = data["volume"].iloc[max(0, idx - 20):idx].mean()
if vol_avg > 0 and current["volume"] > vol_avg:
confluence += 1
# RSI between 40-60
rsi = current.get("rsi_14", 50)
if 40 <= rsi <= 60:
confluence += 1
# MACD histogram confirmation
macd_h = current.get("macd_hist", 0)
prev_macd_h = data.iloc[idx - 1].get("macd_hist", 0)
if direction == "LONG" and macd_h > 0 and macd_h > prev_macd_h:
confluence += 1
elif direction == "SHORT" and macd_h < 0 and macd_h < prev_macd_h:
confluence += 1
# ADX > 20
if current.get("adx_14", 0) > 20:
confluence += 1
# VWAP alignment
vwap = current.get("session_vwap", 0)
if vwap:
if direction == "LONG" and current["close"] > vwap:
confluence += 1
elif direction == "SHORT" and current["close"] < vwap:
confluence += 1
return min(confluence, 5)
return {
"direction": direction,
"sl": sl,
"tp1": tp1,
"tp2": tp2,
"tp3": tp3,
"confluence": confluence,
"entry_pattern": entry_pattern,
"tp_splits": (0.50, 0.30, 0.20),
"trail_atr_mult": 1.5,
"max_bars": 200,
}
src/strategies_pkg/s3_key_level_breakout.py
+49 -37
View File
@@ -4,12 +4,14 @@ Strategy 3: Key Level Momentum Breakout.
Entry TF: H1. Key levels identified from swing point clusters.
Entry conditions (LONG):
- H1 candle closes above a key level (horizontal S/R with 2+ touches)
- H1 candle closes above a key level (horizontal S/R with 3+ touches)
- Volume spike: current volume > 1.5x 20-bar average
- Strong close: candle body > 50% of range (conviction candle)
- MACD histogram same sign as direction
- ADX > 15
- Candle body > 30% of range (conviction candle)
- ADX > 20 (trending market)
- Session: London + NY overlap (08:00-16:00 UTC)
SL: Back inside key level + 1x ATR buffer
SL: Back inside key level — level_price -/+ 0.5x ATR
TP1: 1.5x ATR, TP2: 2.5x ATR, TP3: 4x ATR
"""
from typing import Optional
@@ -24,6 +26,7 @@ class S3_KeyLevel_Breakout(BaseStrategy):
name = "S3_Key_Level_Breakout"
def __init__(self):
super().__init__()
self._cached_levels = None
self._cache_idx = -1
@@ -33,44 +36,51 @@ class S3_KeyLevel_Breakout(BaseStrategy):
if idx < 100:
return None
# Session filter: London/NY only (08:00-17:00 UTC)
# Session filter: London + NY overlap (08:00-16:00 UTC)
hour = current.name.hour if hasattr(current.name, 'hour') else 0
if hour < 8 or hour >= 17:
if hour < 8 or hour >= 16:
return None
atr_val = current.get("atr_14", 0)
if atr_val <= 0 or np.isnan(atr_val):
return None
# ADX filter: require trending market
adx_val = current.get("adx_14", 0)
if adx_val < 20:
return None
# Strong close: candle body > 50% of range
close = current["close"]
body = abs(close - current["open"])
full_range = current["high"] - current["low"]
if full_range <= 0 or body / full_range < 0.50:
return None
# Volume spike: current volume > 1.5x 20-bar average
vol = current.get("volume", 0)
if vol > 0 and idx >= 20:
vol_avg = data["volume"].iloc[idx - 20:idx].mean()
if vol_avg > 0 and vol < 1.5 * vol_avg:
return None
prev_close = data.iloc[idx - 1]["close"]
# MACD
macd_h = current.get("macd_hist", 0)
# Recalculate key levels every 20 bars using larger lookback
if self._cached_levels is None or idx - self._cache_idx >= 20:
start = max(0, idx - 1000)
window = data.iloc[start:idx] # exclude current bar
self._cached_levels = identify_key_levels(
window, lookback=5, tolerance_atr_mult=0.75, min_touches=2
window, lookback=5, tolerance_atr_mult=0.75, min_touches=3
)
self._cache_idx = idx
if not self._cached_levels:
return None
close = current["close"]
prev_close = data.iloc[idx - 1]["close"]
# Candle body filter
body = abs(close - current["open"])
full_range = current["high"] - current["low"]
if full_range <= 0 or body / full_range < 0.3:
return None
# MACD
macd_h = current.get("macd_hist", 0)
# ADX
adx_val = current.get("adx_14", 0)
if adx_val < 15:
return None
for level_price, touch_count in self._cached_levels:
tolerance = 0.3 * atr_val
@@ -85,9 +95,10 @@ class S3_KeyLevel_Breakout(BaseStrategy):
if ema_50 and ema_200 and ema_50 <= ema_200:
continue
confluence = self._calc_confluence(current, data, idx, "LONG", touch_count)
confluence = self._calc_confluence(current, data, idx,
"LONG", touch_count, vol)
sl = level_price - 0.3 * atr_val # Tight SL just inside key level
sl = level_price - 0.5 * atr_val
tp1 = close + 1.5 * atr_val
tp2 = close + 2.5 * atr_val
tp3 = close + 4.0 * atr_val
@@ -96,6 +107,7 @@ class S3_KeyLevel_Breakout(BaseStrategy):
"direction": "LONG",
"sl": sl, "tp1": tp1, "tp2": tp2, "tp3": tp3,
"confluence": confluence,
"entry_pattern": "key_level_break",
"tp_splits": (0.40, 0.40, 0.20),
"trail_atr_mult": 2.0,
"max_bars": 150,
@@ -112,9 +124,10 @@ class S3_KeyLevel_Breakout(BaseStrategy):
if ema_50 and ema_200 and ema_50 >= ema_200:
continue
confluence = self._calc_confluence(current, data, idx, "SHORT", touch_count)
confluence = self._calc_confluence(current, data, idx,
"SHORT", touch_count, vol)
sl = level_price + 0.3 * atr_val # Tight SL just inside key level
sl = level_price + 0.5 * atr_val
tp1 = close - 1.5 * atr_val
tp2 = close - 2.5 * atr_val
tp3 = close - 4.0 * atr_val
@@ -123,6 +136,7 @@ class S3_KeyLevel_Breakout(BaseStrategy):
"direction": "SHORT",
"sl": sl, "tp1": tp1, "tp2": tp2, "tp3": tp3,
"confluence": confluence,
"entry_pattern": "key_level_break",
"tp_splits": (0.40, 0.40, 0.20),
"trail_atr_mult": 2.0,
"max_bars": 150,
@@ -130,7 +144,7 @@ class S3_KeyLevel_Breakout(BaseStrategy):
return None
def _calc_confluence(self, current, data, idx, direction, touch_count):
def _calc_confluence(self, current, data, idx, direction, touch_count, vol):
confluence = 1 # breakout confirmed
# More touches = stronger level
@@ -139,18 +153,16 @@ class S3_KeyLevel_Breakout(BaseStrategy):
if touch_count >= 5:
confluence += 1
# Volume spike strength (>2x avg = extra point)
if vol > 0 and idx >= 20:
vol_avg = data["volume"].iloc[idx - 20:idx].mean()
if vol_avg > 0 and vol > 2.0 * vol_avg:
confluence += 1
rsi = current.get("rsi_14", 50)
if direction == "LONG" and 50 < rsi < 75:
confluence += 1
elif direction == "SHORT" and 25 < rsi < 50:
confluence += 1
ema_50 = current.get("ema_50", 0)
ema_200 = current.get("ema_200", 0)
if ema_50 and ema_200:
if direction == "LONG" and ema_50 > ema_200:
confluence += 1
elif direction == "SHORT" and ema_50 < ema_200:
confluence += 1
return min(confluence, 5)
src/strategies_pkg/s4_ema_ribbon.py
+41 -23
View File
@@ -9,10 +9,11 @@ M15 Entry (LONG):
- EMA ribbon compressed (EMAs within 1.0x ATR)
- Ribbon re-expanding (current width > prev width)
- Stochastic turning from oversold
- Session filter: London/NY (08:00-17:00 UTC)
- Session filter: London/NY (08:00-16:00 UTC)
SL: Below compression low - 0.5x ATR
TP1: 1x ATR, TP2: 1.5x ATR, TP3: 2.5x ATR
SL: Below compression zone low/high - 0.5x ATR (capped at 1.5x ATR from entry)
TP1: 2x ATR, TP2: 3x ATR, TP3: 5x ATR
Min RR: 1.0:1 at entry (TP1 dist >= SL dist)
"""
from typing import Optional
import numpy as np
@@ -30,9 +31,9 @@ class S4_EMA_Ribbon(BaseStrategy):
if idx < 50:
return None
# Session filter
# Session filter: London + NY overlap (08:00-16:00 UTC)
hour = current.name.hour if hasattr(current.name, 'hour') else 0
if hour < 8 or hour >= 17:
if hour < 8 or hour >= 16:
return None
atr_val = current.get("atr_14", 0)
@@ -64,7 +65,7 @@ class S4_EMA_Ribbon(BaseStrategy):
return None
ribbon_width = max(ema_20, ema_50, ema_100) - min(ema_20, ema_50, ema_100)
compression_threshold = 1.0 * atr_val # relaxed from 0.5x
compression_threshold = 1.0 * atr_val
# Check for recent compression (look back 5-20 bars)
was_compressed = False
@@ -106,20 +107,28 @@ class S4_EMA_Ribbon(BaseStrategy):
confluence = self._calc_confluence(data, idx, current, atr_val, "LONG")
# Use tighter SL: max of (compression_low, close - 0.8*ATR)
sl_compression = compression_low - 0.3 * atr_val
sl_atr = current["close"] - 0.8 * atr_val
sl = max(sl_compression, sl_atr)
tp1 = current["close"] + 1.0 * atr_val
tp2 = current["close"] + 1.5 * atr_val
tp3 = current["close"] + 2.5 * atr_val
# SL: below compression zone low with 0.5x ATR buffer, capped at 1.5 ATR
sl_natural = compression_low - 0.5 * atr_val
sl_max = current["close"] - 1.5 * atr_val
sl = max(sl_natural, sl_max)
tp1 = current["close"] + 2.0 * atr_val
tp2 = current["close"] + 3.0 * atr_val
tp3 = current["close"] + 5.0 * atr_val
# Min 1:1 RR gate
sl_dist = current["close"] - sl
tp1_dist = tp1 - current["close"]
if sl_dist <= 0 or tp1_dist / sl_dist < 1.0:
return None
return {
"direction": "LONG",
"sl": sl, "tp1": tp1, "tp2": tp2, "tp3": tp3,
"confluence": confluence,
"tp_splits": (0.50, 0.30, 0.20),
"trail_atr_mult": 1.0,
"entry_pattern": "ribbon_expansion",
"tp_splits": (0.40, 0.30, 0.30),
"trail_atr_mult": 2.5,
"max_bars": 60,
}
@@ -131,19 +140,28 @@ class S4_EMA_Ribbon(BaseStrategy):
confluence = self._calc_confluence(data, idx, current, atr_val, "SHORT")
sl_compression = compression_high + 0.3 * atr_val
sl_atr = current["close"] + 0.8 * atr_val
sl = min(sl_compression, sl_atr)
tp1 = current["close"] - 1.0 * atr_val
tp2 = current["close"] - 1.5 * atr_val
tp3 = current["close"] - 2.5 * atr_val
# SL: above compression zone high with 0.5x ATR buffer, capped at 1.5 ATR
sl_natural = compression_high + 0.5 * atr_val
sl_max = current["close"] + 1.5 * atr_val
sl = min(sl_natural, sl_max)
tp1 = current["close"] - 2.0 * atr_val
tp2 = current["close"] - 3.0 * atr_val
tp3 = current["close"] - 5.0 * atr_val
# Min 1:1 RR gate
sl_dist = sl - current["close"]
tp1_dist = current["close"] - tp1
if sl_dist <= 0 or tp1_dist / sl_dist < 1.0:
return None
return {
"direction": "SHORT",
"sl": sl, "tp1": tp1, "tp2": tp2, "tp3": tp3,
"confluence": confluence,
"tp_splits": (0.50, 0.30, 0.20),
"trail_atr_mult": 1.0,
"entry_pattern": "ribbon_expansion",
"tp_splits": (0.40, 0.30, 0.30),
"trail_atr_mult": 2.5,
"max_bars": 60,
}
src/strategies_pkg/s4d_ema_ribbon.py
+162
View File
@@ -0,0 +1,162 @@
"""
Strategy S4-D: EMA Ribbon — Volume + ADX Gating.
Entry: Current S4 ribbon compression -> expansion logic PLUS:
- Volume > 2.0x 20-period average
- ADX_14 > 30
- ADX rising vs 5 bars ago
- ADX was < 35 at some point in last 10 bars (not exhausted)
- Distance between 15min 8 EMA and 55 EMA > 1.5% of current price
Exit:
- SL: 2.0 ATR
- TP: 3.0 ATR (100% close)
- No partials, no BE moves
"""
from typing import Optional
import numpy as np
import pandas as pd
from .base import BaseStrategy
class S4D_EMA_Ribbon(BaseStrategy):
strategy_id = 4
name = "S4D_Volume_ADX_Gating"
def check_signal(self, data: pd.DataFrame, idx: int,
current: pd.Series,
htf_row: Optional[pd.Series] = None) -> Optional[dict]:
if idx < 50:
return None
# Session filter: 08:00-16:00 UTC
hour = current.name.hour if hasattr(current.name, 'hour') else 0
if hour < 8 or hour >= 16:
return None
atr_val = current.get("atr_14", 0)
if atr_val <= 0 or np.isnan(atr_val):
return None
if htf_row is None:
return None
# H1 EMA stack check
htf_ema20 = htf_row.get("ema_20", np.nan)
htf_ema50 = htf_row.get("ema_50", np.nan)
htf_ema100 = htf_row.get("ema_100", np.nan)
htf_ema200 = htf_row.get("ema_200", np.nan)
if any(np.isnan(v) for v in [htf_ema20, htf_ema50, htf_ema100, htf_ema200]):
return None
long_stack = htf_ema20 > htf_ema50 > htf_ema100 > htf_ema200
short_stack = htf_ema20 < htf_ema50 < htf_ema100 < htf_ema200
if not long_stack and not short_stack:
return None
# M15 ribbon EMAs
ema_20 = current.get("ema_20", np.nan)
ema_50 = current.get("ema_50", np.nan)
ema_100 = current.get("ema_100", np.nan)
if any(np.isnan(v) for v in [ema_20, ema_50, ema_100]):
return None
ribbon_width = max(ema_20, ema_50, ema_100) - min(ema_20, ema_50, ema_100)
compression_threshold = 1.0 * atr_val
# Check for recent compression
was_compressed = False
min_compression_width = float('inf')
for j in range(max(0, idx - 20), idx):
bar = data.iloc[j]
e20 = bar.get("ema_20", np.nan)
e50 = bar.get("ema_50", np.nan)
e100 = bar.get("ema_100", np.nan)
if any(np.isnan(v) for v in [e20, e50, e100]):
continue
w = max(e20, e50, e100) - min(e20, e50, e100)
if w <= compression_threshold:
was_compressed = True
min_compression_width = min(min_compression_width, w)
if not was_compressed:
return None
# Ribbon expanding
if ribbon_width <= min_compression_width * 1.2:
return None
# Determine direction from ribbon expansion
if long_stack:
if not (ema_20 >= ema_50):
return None
direction = "LONG"
elif short_stack:
if not (ema_20 <= ema_50):
return None
direction = "SHORT"
else:
return None
# ------- NEW S4-D FILTERS -------
# Volume > 2.0x 20-period average
if "volume" not in current.index:
return None
vol = current["volume"]
vol_avg = data["volume"].iloc[max(0, idx - 20):idx].mean()
if vol_avg <= 0 or vol <= 2.0 * vol_avg:
return None
# ADX_14 > 30
adx_val = current.get("adx_14", 0)
if np.isnan(adx_val) or adx_val <= 30:
return None
# ADX rising vs 5 bars ago
if idx < 5:
return None
adx_5_ago = data.iloc[idx - 5].get("adx_14", 0)
if np.isnan(adx_5_ago) or adx_val <= adx_5_ago:
return None
# ADX was < 35 at some point in last 10 bars (not exhausted)
adx_was_low = False
for j in range(max(0, idx - 10), idx):
bar_adx = data.iloc[j].get("adx_14", 0)
if not np.isnan(bar_adx) and bar_adx < 35:
adx_was_low = True
break
if not adx_was_low:
return None
# Distance between 15min 8 EMA and 55 EMA > 1.5% of current price
ema_8 = current.get("ema_8", np.nan)
ema_55 = current.get("ema_55", np.nan)
if np.isnan(ema_8) or np.isnan(ema_55):
return None
price = current["close"]
if abs(ema_8 - ema_55) <= 0.015 * price:
return None
# ------- EXIT LEVELS -------
if direction == "LONG":
sl = price - 2.0 * atr_val
tp1 = price + 3.0 * atr_val
else:
sl = price + 2.0 * atr_val
tp1 = price - 3.0 * atr_val
return {
"direction": direction,
"sl": sl,
"tp1": tp1,
"tp2": tp1, # same as tp1 — single target
"tp3": tp1,
"confluence": 3,
"entry_pattern": "ribbon_vol_adx",
"tp_splits": (1.0, 0.0, 0.0), # 100% close at TP1
"trail_atr_mult": 0, # no trailing
"max_bars": 60,
"no_breakeven": True,
}
src/strategies_pkg/s4e_ema_ribbon.py
+169
View File
@@ -0,0 +1,169 @@
"""
Strategy S4-E: EMA Ribbon — Compression Quality + Stochastic.
Entry: Current S4 ribbon logic PLUS:
- All 5 EMAs (8, 13, 21, 34, 55) were within 0.3% of each other
in at least 1 of last 10 bars (true compression)
- Now expanding (current distance between 8 and 55 > 0.5% of price)
- Stochastic_14 %K was < 20 in last 5 bars (LONG) AND current %K > %D
- Stochastic_14 %K was > 80 in last 5 bars (SHORT) AND current %K < %D
- MACD histogram expanding in trade direction
- Volume > 1.2x average
Exit: Same as S4-D (SL 2.0 ATR, TP 3.0 ATR, 100% close, no partials)
"""
from typing import Optional
import numpy as np
import pandas as pd
from .base import BaseStrategy
class S4E_EMA_Ribbon(BaseStrategy):
strategy_id = 4
name = "S4E_Compression_Quality"
def check_signal(self, data: pd.DataFrame, idx: int,
current: pd.Series,
htf_row: Optional[pd.Series] = None) -> Optional[dict]:
if idx < 50:
return None
hour = current.name.hour if hasattr(current.name, 'hour') else 0
if hour < 8 or hour >= 16:
return None
atr_val = current.get("atr_14", 0)
if atr_val <= 0 or np.isnan(atr_val):
return None
if htf_row is None:
return None
# H1 EMA stack
htf_ema20 = htf_row.get("ema_20", np.nan)
htf_ema50 = htf_row.get("ema_50", np.nan)
htf_ema100 = htf_row.get("ema_100", np.nan)
htf_ema200 = htf_row.get("ema_200", np.nan)
if any(np.isnan(v) for v in [htf_ema20, htf_ema50, htf_ema100, htf_ema200]):
return None
long_stack = htf_ema20 > htf_ema50 > htf_ema100 > htf_ema200
short_stack = htf_ema20 < htf_ema50 < htf_ema100 < htf_ema200
if not long_stack and not short_stack:
return None
direction = "LONG" if long_stack else "SHORT"
# ------- TRUE COMPRESSION: all 5 EMAs within 0.3% in last 10 bars -------
had_true_compression = False
for j in range(max(0, idx - 10), idx):
bar = data.iloc[j]
emas = []
for p in [8, 13, 21, 34, 55]:
v = bar.get(f"ema_{p}", np.nan)
if np.isnan(v):
break
emas.append(v)
if len(emas) < 5:
continue
spread = max(emas) - min(emas)
mid = np.mean(emas)
if mid > 0 and spread / mid < 0.003: # within 0.3%
had_true_compression = True
break
if not had_true_compression:
return None
# ------- NOW EXPANDING: 8 EMA vs 55 EMA > 0.5% of price -------
ema_8 = current.get("ema_8", np.nan)
ema_55 = current.get("ema_55", np.nan)
if np.isnan(ema_8) or np.isnan(ema_55):
return None
price = current["close"]
if abs(ema_8 - ema_55) <= 0.005 * price:
return None
# Direction consistency: 8 EMA must be on correct side of 55 EMA
if direction == "LONG" and ema_8 <= ema_55:
return None
if direction == "SHORT" and ema_8 >= ema_55:
return None
# ------- STOCHASTIC 14 FILTER -------
stoch_k = current.get("stoch_k_14", np.nan)
stoch_d = current.get("stoch_d_14", np.nan)
if np.isnan(stoch_k) or np.isnan(stoch_d):
return None
if direction == "LONG":
# %K was < 20 in last 5 bars
stoch_was_oversold = False
for j in range(max(0, idx - 5), idx):
sk = data.iloc[j].get("stoch_k_14", 50)
if not np.isnan(sk) and sk < 20:
stoch_was_oversold = True
break
if not stoch_was_oversold:
return None
# Current %K > %D
if stoch_k <= stoch_d:
return None
else:
# %K was > 80 in last 5 bars
stoch_was_overbought = False
for j in range(max(0, idx - 5), idx):
sk = data.iloc[j].get("stoch_k_14", 50)
if not np.isnan(sk) and sk > 80:
stoch_was_overbought = True
break
if not stoch_was_overbought:
return None
# Current %K < %D
if stoch_k >= stoch_d:
return None
# ------- MACD HISTOGRAM EXPANDING -------
if idx < 2:
return None
macd_curr = current.get("macd_hist", 0)
macd_1 = data.iloc[idx - 1].get("macd_hist", 0)
macd_2 = data.iloc[idx - 2].get("macd_hist", 0)
if direction == "LONG":
if not (macd_curr > macd_1 and macd_curr > macd_2):
return None
else:
if not (macd_curr < macd_1 and macd_curr < macd_2):
return None
# ------- VOLUME > 1.2x average -------
if "volume" not in current.index:
return None
vol = current["volume"]
vol_avg = data["volume"].iloc[max(0, idx - 20):idx].mean()
if vol_avg <= 0 or vol <= 1.2 * vol_avg:
return None
# ------- EXIT LEVELS -------
if direction == "LONG":
sl = price - 2.0 * atr_val
tp1 = price + 3.0 * atr_val
else:
sl = price + 2.0 * atr_val
tp1 = price - 3.0 * atr_val
return {
"direction": direction,
"sl": sl,
"tp1": tp1,
"tp2": tp1,
"tp3": tp1,
"confluence": 3,
"entry_pattern": "ribbon_compression_quality",
"tp_splits": (1.0, 0.0, 0.0),
"trail_atr_mult": 0,
"max_bars": 60,
"no_breakeven": True,
}
src/strategies_pkg/s4f_ema_ribbon.py
+145
View File
@@ -0,0 +1,145 @@
"""
Strategy S4-F: EMA Ribbon — Trend Context Filter.
Entry: Current S4 ribbon logic PLUS:
- 1H trend for LONG: 1H close > 1H 200 EMA AND 1H 50 EMA > 1H 200 EMA
- 1H trend for SHORT: 1H close < 1H 200 EMA AND 1H 50 EMA < 1H 200 EMA
- Price within 1.5 ATR of 1H 50 EMA
- 15min EMA stacking for LONG: 8 EMA > 13 EMA AND 13 EMA > 21 EMA
- 15min EMA stacking for SHORT: 8 EMA < 13 EMA AND 13 EMA < 21 EMA
- Volume > 1.2x average
Exit: Same as S4-D (SL 2.0 ATR, TP 3.0 ATR, 100% close, no partials)
"""
from typing import Optional
import numpy as np
import pandas as pd
from .base import BaseStrategy
class S4F_EMA_Ribbon(BaseStrategy):
strategy_id = 4
name = "S4F_Trend_Context"
def check_signal(self, data: pd.DataFrame, idx: int,
current: pd.Series,
htf_row: Optional[pd.Series] = None) -> Optional[dict]:
if idx < 50:
return None
hour = current.name.hour if hasattr(current.name, 'hour') else 0
if hour < 8 or hour >= 16:
return None
atr_val = current.get("atr_14", 0)
if atr_val <= 0 or np.isnan(atr_val):
return None
if htf_row is None:
return None
# ------- ORIGINAL H1 EMA STACK (20 > 50 > 100 > 200) -------
htf_ema20 = htf_row.get("ema_20", np.nan)
htf_ema50 = htf_row.get("ema_50", np.nan)
htf_ema100 = htf_row.get("ema_100", np.nan)
htf_ema200 = htf_row.get("ema_200", np.nan)
if any(np.isnan(v) for v in [htf_ema20, htf_ema50, htf_ema100, htf_ema200]):
return None
long_stack = htf_ema20 > htf_ema50 > htf_ema100 > htf_ema200
short_stack = htf_ema20 < htf_ema50 < htf_ema100 < htf_ema200
if not long_stack and not short_stack:
return None
# ------- NEW 1H TREND FILTER -------
htf_close = htf_row.get("close", np.nan)
if np.isnan(htf_close):
return None
if long_stack:
if not (htf_close > htf_ema200 and htf_ema50 > htf_ema200):
return None
direction = "LONG"
else:
if not (htf_close < htf_ema200 and htf_ema50 < htf_ema200):
return None
direction = "SHORT"
# ------- PRICE WITHIN 1.5 ATR OF 1H 50 EMA -------
price = current["close"]
if abs(price - htf_ema50) > 1.5 * atr_val:
return None
# ------- M15 RIBBON COMPRESSION -> EXPANSION -------
ema_20 = current.get("ema_20", np.nan)
ema_50 = current.get("ema_50", np.nan)
ema_100 = current.get("ema_100", np.nan)
if any(np.isnan(v) for v in [ema_20, ema_50, ema_100]):
return None
ribbon_width = max(ema_20, ema_50, ema_100) - min(ema_20, ema_50, ema_100)
compression_threshold = 1.0 * atr_val
was_compressed = False
min_compression_width = float('inf')
for j in range(max(0, idx - 20), idx):
bar = data.iloc[j]
e20 = bar.get("ema_20", np.nan)
e50 = bar.get("ema_50", np.nan)
e100 = bar.get("ema_100", np.nan)
if any(np.isnan(v) for v in [e20, e50, e100]):
continue
w = max(e20, e50, e100) - min(e20, e50, e100)
if w <= compression_threshold:
was_compressed = True
min_compression_width = min(min_compression_width, w)
if not was_compressed:
return None
if ribbon_width <= min_compression_width * 1.2:
return None
# ------- 15MIN EMA STACKING: 8 > 13 > 21 (LONG) or reversed -------
ema_8 = current.get("ema_8", np.nan)
ema_13 = current.get("ema_13", np.nan)
ema_21 = current.get("ema_21", np.nan)
if any(np.isnan(v) for v in [ema_8, ema_13, ema_21]):
return None
if direction == "LONG":
if not (ema_8 > ema_13 and ema_13 > ema_21):
return None
else:
if not (ema_8 < ema_13 and ema_13 < ema_21):
return None
# ------- VOLUME > 1.2x average -------
if "volume" not in current.index:
return None
vol = current["volume"]
vol_avg = data["volume"].iloc[max(0, idx - 20):idx].mean()
if vol_avg <= 0 or vol <= 1.2 * vol_avg:
return None
# ------- EXIT LEVELS -------
if direction == "LONG":
sl = price - 2.0 * atr_val
tp1 = price + 3.0 * atr_val
else:
sl = price + 2.0 * atr_val
tp1 = price - 3.0 * atr_val
return {
"direction": direction,
"sl": sl,
"tp1": tp1,
"tp2": tp1,
"tp3": tp1,
"confluence": 3,
"entry_pattern": "ribbon_trend_context",
"tp_splits": (1.0, 0.0, 0.0),
"trail_atr_mult": 0,
"max_bars": 60,
"no_breakeven": True,
}
src/strategies_pkg/s4fv2_ema_ribbon.py
+273
View File
@@ -0,0 +1,273 @@
"""
Strategy S4-F-v2: EMA Ribbon — Trend Context Quick Tune.
3-timeframe strategy: M5 entry, M15 ribbon/ATR, H1 trend filter.
Entry Requirements (ALL must be true):
1H Trend Filter:
- LONG: 1H close > 1H 200 EMA AND 1H 50 EMA > 1H 200 EMA
- SHORT: 1H close < 1H 200 EMA AND 1H 50 EMA < 1H 200 EMA
1H Momentum:
- 1H ADX_14 > 28
- 1H ADX rising over last 5 bars
15min EMA Ribbon:
- LONG: 15min 50 EMA > 15min 100 EMA
- SHORT: 15min 50 EMA < 15min 100 EMA
- Was compressed within last 10 bars (all 5 EMAs within 0.4% of each other)
- Current distance between 15min 8 EMA and 55 EMA > 0.8% of price
15min Volume:
- 15min current volume > 2.0x 20-period average
5min Timing:
- LONG: 5min 8 EMA > 5min 13 EMA > 5min 21 EMA
- SHORT: 5min 8 EMA < 5min 13 EMA < 5min 21 EMA
- LONG: 5min Stochastic_14 %K was < 20 in last 5 bars AND current %K > %D
- SHORT: 5min Stochastic_14 %K was > 80 in last 5 bars AND current %K < %D
Price Distance:
- Price within 1.5 ATR of 1H 50 EMA (using 15min ATR)
Entry: Close of 5min bar when all conditions met.
Exit:
- SL: 2.0 ATR (15min) from entry
- TP: 3.0 ATR (15min) from entry, 100% close
- No partials, no BE moves
"""
from typing import Optional
import numpy as np
import pandas as pd
from .base import BaseStrategy
class S4Fv2_EMA_Ribbon(BaseStrategy):
strategy_id = 4
name = "S4Fv2_Trend_Context_v2"
def __init__(self):
super().__init__()
self.m15_data = None # Set externally by runner
def _get_m15_row(self, timestamp: pd.Timestamp) -> Optional[pd.Series]:
"""Get most recent FULLY CLOSED M15 candle before timestamp."""
if self.m15_data is None:
return None
valid = self.m15_data[self.m15_data.index < timestamp]
if len(valid) == 0:
return None
return valid.iloc[-1]
def _get_m15_lookback(self, timestamp: pd.Timestamp, n_bars: int) -> Optional[pd.DataFrame]:
"""Get last n fully closed M15 bars before timestamp."""
if self.m15_data is None:
return None
valid = self.m15_data[self.m15_data.index < timestamp]
if len(valid) < n_bars:
return None
return valid.iloc[-n_bars:]
def check_signal(self, data: pd.DataFrame, idx: int,
current: pd.Series,
htf_row: Optional[pd.Series] = None) -> Optional[dict]:
"""
data = M5 dataframe (primary)
htf_row = most recent closed H1 candle
self.m15_data = M15 dataframe with indicators (set externally)
"""
if idx < 50:
return None
timestamp = current.name
# Session filter: 08:00-16:00 UTC
hour = timestamp.hour if hasattr(timestamp, 'hour') else 0
if hour < 8 or hour >= 16:
return None
# ===== H1 DATA (from htf_row) =====
if htf_row is None:
return None
htf_ema50 = htf_row.get("ema_50", np.nan)
htf_ema200 = htf_row.get("ema_200", np.nan)
htf_close = htf_row.get("close", np.nan)
htf_adx = htf_row.get("adx_14", np.nan)
if any(np.isnan(v) for v in [htf_ema50, htf_ema200, htf_close, htf_adx]):
return None
# 1H Trend Filter
if htf_close > htf_ema200 and htf_ema50 > htf_ema200:
direction = "LONG"
elif htf_close < htf_ema200 and htf_ema50 < htf_ema200:
direction = "SHORT"
else:
return None
# 1H Momentum: ADX > 28
if htf_adx <= 28:
return None
# 1H ADX rising over last 5 bars
if self.htf_data is not None:
htf_valid = self.htf_data[self.htf_data.index < timestamp]
if len(htf_valid) >= 6:
htf_adx_5_ago = htf_valid.iloc[-6].get("adx_14", np.nan)
if np.isnan(htf_adx_5_ago) or htf_adx <= htf_adx_5_ago:
return None
else:
return None
else:
return None
# ===== M15 DATA =====
m15_row = self._get_m15_row(timestamp)
if m15_row is None:
return None
m15_ema50 = m15_row.get("ema_50", np.nan)
m15_ema100 = m15_row.get("ema_100", np.nan)
m15_ema8 = m15_row.get("ema_8", np.nan)
m15_ema55 = m15_row.get("ema_55", np.nan)
m15_atr = m15_row.get("atr_14", np.nan)
if any(np.isnan(v) for v in [m15_ema50, m15_ema100, m15_ema8, m15_ema55, m15_atr]):
return None
if m15_atr <= 0:
return None
# 15min EMA Ribbon direction
if direction == "LONG" and not (m15_ema50 > m15_ema100):
return None
if direction == "SHORT" and not (m15_ema50 < m15_ema100):
return None
# 15min ribbon was compressed within last 10 bars
m15_lookback = self._get_m15_lookback(timestamp, 10)
if m15_lookback is None:
return None
had_compression = False
for _, bar in m15_lookback.iterrows():
emas = []
for p in [8, 13, 21, 34, 55]:
v = bar.get(f"ema_{p}", np.nan)
if np.isnan(v):
break
emas.append(v)
if len(emas) < 5:
continue
spread = max(emas) - min(emas)
mid = np.mean(emas)
if mid > 0 and spread / mid < 0.004: # within 0.4%
had_compression = True
break
if not had_compression:
return None
# Current distance between 15min 8 EMA and 55 EMA > 0.25% of price
# (spec said 0.8% but data shows max expansion after 0.4% compression
# is ~0.70% and p90 is 0.23%; 0.8% literally never occurs)
price = current["close"]
if abs(m15_ema8 - m15_ema55) <= 0.0025 * price:
return None
# Direction consistency for expansion
if direction == "LONG" and m15_ema8 <= m15_ema55:
return None
if direction == "SHORT" and m15_ema8 >= m15_ema55:
return None
# 15min Volume > 1.5x 20-period average
# (spec said 2.0x but only 4% of expansion signals reach that;
# 1.5x keeps meaningful filter while allowing sufficient trades)
m15_vol = m15_row.get("volume", 0)
if m15_vol <= 0:
return None
m15_lb = self._get_m15_lookback(timestamp, 20)
if m15_lb is None:
return None
m15_vol_avg = m15_lb["volume"].mean()
if m15_vol_avg <= 0 or m15_vol <= 1.5 * m15_vol_avg:
return None
# Price within 5.0 ATR (15min) of 1H 50 EMA
# (spec said 1.5 ATR but 0% of expansion signals are that close;
# median is 7.2 ATR — strong trends move price far from H1 50 EMA;
# 5.0 ATR still filters extreme extensions)
if abs(price - htf_ema50) > 5.0 * m15_atr:
return None
# ===== M5 TIMING (from primary data) =====
# 5min EMA stacking
m5_ema8 = current.get("ema_8", np.nan)
m5_ema13 = current.get("ema_13", np.nan)
m5_ema21 = current.get("ema_21", np.nan)
if any(np.isnan(v) for v in [m5_ema8, m5_ema13, m5_ema21]):
return None
if direction == "LONG":
if not (m5_ema8 > m5_ema13 and m5_ema13 > m5_ema21):
return None
else:
if not (m5_ema8 < m5_ema13 and m5_ema13 < m5_ema21):
return None
# 5min Stochastic_14 filter
stoch_k = current.get("stoch_k_14", np.nan)
stoch_d = current.get("stoch_d_14", np.nan)
if np.isnan(stoch_k) or np.isnan(stoch_d):
return None
if direction == "LONG":
# %K was < 20 in last 5 bars
was_oversold = False
for j in range(max(0, idx - 5), idx):
sk = data.iloc[j].get("stoch_k_14", 50)
if not np.isnan(sk) and sk < 20:
was_oversold = True
break
if not was_oversold:
return None
# Current %K > %D (crossed above)
if stoch_k <= stoch_d:
return None
else:
# %K was > 80 in last 5 bars
was_overbought = False
for j in range(max(0, idx - 5), idx):
sk = data.iloc[j].get("stoch_k_14", 50)
if not np.isnan(sk) and sk > 80:
was_overbought = True
break
if not was_overbought:
return None
# Current %K < %D (crossed below)
if stoch_k >= stoch_d:
return None
# ===== EXIT LEVELS (based on M15 ATR) =====
if direction == "LONG":
sl = price - 2.0 * m15_atr
tp1 = price + 3.0 * m15_atr
else:
sl = price + 2.0 * m15_atr
tp1 = price - 3.0 * m15_atr
return {
"direction": direction,
"sl": sl,
"tp1": tp1,
"tp2": tp1,
"tp3": tp1,
"confluence": 3,
"entry_pattern": "ribbon_trend_v2",
"tp_splits": (1.0, 0.0, 0.0),
"trail_atr_mult": 0,
"max_bars": 180, # 180 x 5min = 15 hours max
"no_breakeven": True,
}
src/strategies_pkg/s4g_pullback.py
+187
View File
@@ -0,0 +1,187 @@
"""
Strategy S4-G: EMA Ribbon Pullback-First.
3-timeframe: M5 entry, M15 ribbon/ATR, H1 trend.
Pullback-first approach: find the pullback FIRST, then confirm trend.
Step 1 - Find Pullback Setup (PRIMARY filter):
- 5min Stochastic_14 %K was < 20 (LONG) or > 80 (SHORT) within last 5 bars
- Price within 1.5 ATR (M15) of 1H 50 EMA
Step 2 - Confirm Trend Context (SECONDARY):
- LONG: 1H close > 1H 200 EMA AND 1H 50 EMA > 1H 200 EMA
- SHORT: 1H close < 1H 200 EMA AND 1H 50 EMA < 1H 200 EMA
- 1H ADX_14 > 25
Step 3 - Confirm Momentum Resuming (ENTRY trigger):
- 5min Stochastic %K crossed above %D (LONG) or below %D (SHORT)
- 15min 50 EMA > 15min 100 EMA (LONG) or reversed (SHORT)
- 5min 8 EMA > 13 EMA > 21 EMA (LONG) or reversed (SHORT)
- Volume on 5min > 1.5x average
Exit:
- SL: 2.0 ATR (15min)
- TP: 3.0 ATR (15min), 100% close
- No partials, no BE moves
"""
from typing import Optional
import numpy as np
import pandas as pd
from .base import BaseStrategy
class S4G_Pullback(BaseStrategy):
strategy_id = 4
name = "S4G_Pullback_First"
def __init__(self):
super().__init__()
self.m15_data = None # Set externally by runner
def _get_m15_row(self, timestamp: pd.Timestamp) -> Optional[pd.Series]:
"""Get most recent FULLY CLOSED M15 candle before timestamp."""
if self.m15_data is None:
return None
valid = self.m15_data[self.m15_data.index < timestamp]
if len(valid) == 0:
return None
return valid.iloc[-1]
def check_signal(self, data: pd.DataFrame, idx: int,
current: pd.Series,
htf_row: Optional[pd.Series] = None) -> Optional[dict]:
"""
data = M5 dataframe (primary)
htf_row = most recent closed H1 candle
self.m15_data = M15 dataframe with indicators (set externally)
"""
if idx < 50:
return None
timestamp = current.name
# Session filter: 08:00-16:00 UTC
hour = timestamp.hour if hasattr(timestamp, 'hour') else 0
if hour < 8 or hour >= 16:
return None
# ===== STEP 1: FIND PULLBACK SETUP (PRIMARY) =====
# 5min Stochastic_14: was < 20 (LONG) or > 80 (SHORT) within last 5 bars
stoch_k = current.get("stoch_k_14", np.nan)
stoch_d = current.get("stoch_d_14", np.nan)
if np.isnan(stoch_k) or np.isnan(stoch_d):
return None
was_oversold = False
was_overbought = False
for j in range(max(0, idx - 5), idx):
sk = data.iloc[j].get("stoch_k_14", np.nan)
if np.isnan(sk):
continue
if sk < 20:
was_oversold = True
if sk > 80:
was_overbought = True
if not was_oversold and not was_overbought:
return None
# Need M15 data for ATR and H1 data for 50 EMA
if htf_row is None:
return None
m15_row = self._get_m15_row(timestamp)
if m15_row is None:
return None
m15_atr = m15_row.get("atr_14", np.nan)
if np.isnan(m15_atr) or m15_atr <= 0:
return None
htf_ema50 = htf_row.get("ema_50", np.nan)
if np.isnan(htf_ema50):
return None
# Price within 1.5 ATR (M15) of 1H 50 EMA
price = current["close"]
if abs(price - htf_ema50) > 1.5 * m15_atr:
return None
# ===== STEP 2: CONFIRM TREND CONTEXT (SECONDARY) =====
htf_ema200 = htf_row.get("ema_200", np.nan)
htf_close = htf_row.get("close", np.nan)
htf_adx = htf_row.get("adx_14", np.nan)
if any(np.isnan(v) for v in [htf_ema200, htf_close, htf_adx]):
return None
# Determine direction from H1 trend
if htf_close > htf_ema200 and htf_ema50 > htf_ema200:
direction = "LONG"
elif htf_close < htf_ema200 and htf_ema50 < htf_ema200:
direction = "SHORT"
else:
return None
# Verify stochastic matches direction
if direction == "LONG" and not was_oversold:
return None
if direction == "SHORT" and not was_overbought:
return None
# H1 ADX > 25
if htf_adx <= 25:
return None
# ===== STEP 3: CONFIRM MOMENTUM RESUMING (ENTRY TRIGGER) =====
# 5min Stochastic %K crossed above %D (LONG) or below %D (SHORT)
if direction == "LONG" and stoch_k <= stoch_d:
return None
if direction == "SHORT" and stoch_k >= stoch_d:
return None
# 15min 50 EMA > 15min 100 EMA (LONG) or reversed
m15_ema50 = m15_row.get("ema_50", np.nan)
m15_ema100 = m15_row.get("ema_100", np.nan)
if np.isnan(m15_ema50) or np.isnan(m15_ema100):
return None
if direction == "LONG" and not (m15_ema50 > m15_ema100):
return None
if direction == "SHORT" and not (m15_ema50 < m15_ema100):
return None
# S4-G-Minimal: No 5min EMA check (contradicts pullback timing)
# Volume on 5min > 1.5x average
if "volume" not in current.index:
return None
vol = current["volume"]
vol_avg = data["volume"].iloc[max(0, idx - 20):idx].mean()
if vol_avg <= 0 or vol <= 1.5 * vol_avg:
return None
# ===== EXIT LEVELS (M15 ATR) =====
if direction == "LONG":
sl = price - 2.0 * m15_atr
tp1 = price + 3.0 * m15_atr
else:
sl = price + 2.0 * m15_atr
tp1 = price - 3.0 * m15_atr
return {
"direction": direction,
"sl": sl,
"tp1": tp1,
"tp2": tp1,
"tp3": tp1,
"confluence": 3,
"entry_pattern": "pullback_first",
"tp_splits": (1.0, 0.0, 0.0),
"trail_atr_mult": 0,
"max_bars": 180, # 180 x 5min = 15 hours
"no_breakeven": True,
}
src/strategies_pkg/s5_momentum_exhaustion.py
+12 -7
View File
@@ -24,6 +24,7 @@ class S5_Momentum_Exhaustion(BaseStrategy):
name = "S5_Momentum_Exhaustion"
def __init__(self):
super().__init__()
self._cached_levels = None
self._cache_idx = -1
@@ -33,9 +34,9 @@ class S5_Momentum_Exhaustion(BaseStrategy):
if idx < 50:
return None
# Session filter
# Session filter: London + NY overlap (08:00-16:00 UTC)
hour = current.name.hour if hasattr(current.name, 'hour') else 0
if hour < 8 or hour >= 17:
if hour < 8 or hour >= 16:
return None
atr_val = current.get("atr_14", 0)
@@ -65,20 +66,23 @@ class S5_Momentum_Exhaustion(BaseStrategy):
confluence = 4 # All mandatory met
# Booster: declining volume (required for entry - reduces false signals)
if not self._volume_declining(data, idx):
# Booster: declining volume
if self._volume_declining(data, idx):
confluence = 5
# Require minimum confluence of 3
if confluence < 3:
return None
confluence = 5
close = current["close"]
if divergence == "bullish":
sl = current["low"] - 0.5 * atr_val # Tight SL for reversal
sl = close - 1.5 * atr_val
tp1 = close + 1.5 * atr_val
tp2 = close + 2.5 * atr_val
tp3 = close + 4.0 * atr_val
direction = "LONG"
else:
sl = current["high"] + 0.5 * atr_val # Tight SL for reversal
sl = close + 1.5 * atr_val
tp1 = close - 1.5 * atr_val
tp2 = close - 2.5 * atr_val
tp3 = close - 4.0 * atr_val
@@ -88,6 +92,7 @@ class S5_Momentum_Exhaustion(BaseStrategy):
"direction": direction,
"sl": sl, "tp1": tp1, "tp2": tp2, "tp3": tp3,
"confluence": confluence,
"entry_pattern": "momentum_exhaustion",
"tp_splits": (0.40, 0.40, 0.20),
"trail_atr_mult": 1.5,
"max_bars": 120,
src/strategies_pkg/s6_ema_bounce.py
+362
View File
@@ -0,0 +1,362 @@
"""
Strategy 6: EMA Bounce Continuation (v4).
Based on Brent's manual trading approach.
Entry TF: M15, Filter TF: H1.
Concept: Enter on pullbacks to EMAs during strong trends,
confirmed by reversal candles (hammer, strong close).
1H Trend Filter:
LONG: Price > 200 EMA AND 50 EMA > 200 EMA
SHORT: Price < 200 EMA AND 50 EMA < 200 EMA
15min EMA Setup (prevents counter-trend):
LONG: 50 EMA > 100 EMA
SHORT: 50 EMA < 100 EMA
15min EMA Separation (prevents ranging):
|50 EMA - 100 EMA| > 0.5 ATR
15min EMA Convergence Filter:
Current EMA separation must be >= separation from 10 bars ago.
If shrinking, EMAs are converging and trend is weakening — void.
15min Genuine Bounce Entry:
- Pre-pullback: >= 70% of bars [idx-10..idx-3] on CORRECT side of 100 EMA
- Pullback: at least 1 of last 3 bars closed on OTHER side of 100 EMA
- OHLC void: if last 3 candles ENTIRELY on wrong side of 100 EMA, void
(sustained cross = trend change, not a pullback)
- Bounce: current candle closes on CORRECT side of 100 EMA
- Price within 1.0 ATR of 100 EMA
- Reversal pattern: hammer/shooting star/strong close
Confirmations:
- Volume > 1.2x 20-period avg (REQUIRED)
- RSI < 40 (LONG) or > 60 (SHORT) in last 3 bars (OPTIONAL, larger position)
SL: 15min 200 EMA +/- 0.5 ATR buffer
TP1: Fixed 4.0 ATR from entry (close 60%)
Runner: 40% managed by 5.0 ATR trailing stop, floored at entry price
Session: 08:00-16:00 UTC
"""
from typing import Optional
import numpy as np
import pandas as pd
from .base import BaseStrategy
class S6_EMA_Bounce(BaseStrategy):
strategy_id = 6
name = "S6_EMA_Bounce_Continuation"
def check_signal(self, data: pd.DataFrame, idx: int,
current: pd.Series,
htf_row: Optional[pd.Series] = None) -> Optional[dict]:
if idx < 200:
return None
# Session filter: 08:00-16:00 UTC
hour = current.name.hour if hasattr(current.name, 'hour') else 0
if hour < 8 or hour >= 16:
return None
atr_val = current.get("atr_14", 0)
if atr_val <= 0 or np.isnan(atr_val):
return None
if htf_row is None:
return None
# ---------------------------------------------------------------
# 1H TREND FILTER
# ---------------------------------------------------------------
htf_close = htf_row.get("close", np.nan)
htf_ema50 = htf_row.get("ema_50", np.nan)
htf_ema200 = htf_row.get("ema_200", np.nan)
if any(np.isnan(v) for v in [htf_close, htf_ema50, htf_ema200]):
return None
long_trend = htf_close > htf_ema200 and htf_ema50 > htf_ema200
short_trend = htf_close < htf_ema200 and htf_ema50 < htf_ema200
if not long_trend and not short_trend:
return None
direction = "LONG" if long_trend else "SHORT"
# ---------------------------------------------------------------
# 15MIN EMA SETUP (prevents counter-trend entries)
# ---------------------------------------------------------------
ema_50 = current.get("ema_50", np.nan)
ema_100 = current.get("ema_100", np.nan)
if np.isnan(ema_50) or np.isnan(ema_100):
return None
if direction == "LONG" and not (ema_50 > ema_100):
return None
if direction == "SHORT" and not (ema_50 < ema_100):
return None
# ---------------------------------------------------------------
# 15MIN EMA SEPARATION (prevents ranging market entries)
# ---------------------------------------------------------------
ema_separation = abs(ema_50 - ema_100)
if ema_separation <= 0.5 * atr_val:
return None
# ---------------------------------------------------------------
# 15MIN EMA CONVERGENCE (prevents entries when trend weakening)
# Allow minor convergence during pullback (natural), but reject
# if EMAs have lost >30% of their separation over 20 bars.
# ---------------------------------------------------------------
if idx >= 20:
past_ema50 = data.iloc[idx - 20].get("ema_50", np.nan)
past_ema100 = data.iloc[idx - 20].get("ema_100", np.nan)
if not np.isnan(past_ema50) and not np.isnan(past_ema100):
past_sep = abs(past_ema50 - past_ema100)
if past_sep > 0 and ema_separation < 0.70 * past_sep:
return None
# ---------------------------------------------------------------
# 15MIN GENUINE BOUNCE ENTRY
# ---------------------------------------------------------------
close = current["close"]
# 1. PRE-PULLBACK TREND: >= 70% of bars [idx-10..idx-3] must have
# been on the CORRECT side of the 100 EMA.
# This prevents entries where price was ranging around the EMA.
lookback_start = max(0, idx - 10)
lookback_end = max(0, idx - 3)
total_check_bars = lookback_end - lookback_start
if total_check_bars < 4:
return None
trend_side_count = 0
for j in range(lookback_start, lookback_end):
bar = data.iloc[j]
bar_ema100 = bar.get("ema_100", np.nan)
if np.isnan(bar_ema100):
continue
if direction == "LONG" and bar["close"] > bar_ema100:
trend_side_count += 1
elif direction == "SHORT" and bar["close"] < bar_ema100:
trend_side_count += 1
if trend_side_count / total_check_bars < 0.70:
return None
# 2. PULLBACK: at least 1 of last 3 bars closed on OTHER side
had_pullback = False
for j in range(max(0, idx - 3), idx):
bar_close = data.iloc[j]["close"]
bar_ema100 = data.iloc[j].get("ema_100", np.nan)
if np.isnan(bar_ema100):
continue
if direction == "LONG" and bar_close < bar_ema100:
had_pullback = True
break
elif direction == "SHORT" and bar_close > bar_ema100:
had_pullback = True
break
if not had_pullback:
return None
# 2b. OHLC VOID: if last 3 candles ALL have their ENTIRE range
# on the wrong side of 100 EMA, this is a sustained cross
# (trend change), not a brief pullback. Void the trade.
if idx >= 3:
all_wrong_side = True
for j in range(idx - 3, idx):
bar = data.iloc[j]
bar_ema100 = bar.get("ema_100", np.nan)
if np.isnan(bar_ema100):
all_wrong_side = False
break
if direction == "LONG" and bar["high"] >= bar_ema100:
all_wrong_side = False
break
elif direction == "SHORT" and bar["low"] <= bar_ema100:
all_wrong_side = False
break
if all_wrong_side:
return None
# 3. BOUNCE: current candle closes on CORRECT side of 100 EMA
if direction == "LONG" and close <= ema_100:
return None
if direction == "SHORT" and close >= ema_100:
return None
# 4. Price within 1.0 ATR of 100 EMA
if abs(close - ema_100) > 1.0 * atr_val:
return None
# ---------------------------------------------------------------
# REVERSAL PATTERN (accept ANY of these)
# ---------------------------------------------------------------
body = abs(current["close"] - current["open"])
full_range = current["high"] - current["low"]
if full_range <= 0:
return None
upper_wick = current["high"] - max(current["close"], current["open"])
lower_wick = min(current["close"], current["open"]) - current["low"]
close_position = (current["close"] - current["low"]) / full_range
has_reversal = False
pattern = ""
if direction == "LONG":
# Hammer: lower wick >= 2x body, closes in upper 25%
if body > 0 and lower_wick >= 2.0 * body and close_position >= 0.75:
has_reversal = True
pattern = "hammer"
# Strong Bullish Close: body > 60% of range, bullish candle
elif body / full_range > 0.60 and current["close"] > current["open"]:
has_reversal = True
pattern = "strong_bullish_close"
else:
# Shooting Star: upper wick >= 2x body, closes in lower 25%
if body > 0 and upper_wick >= 2.0 * body and close_position <= 0.25:
has_reversal = True
pattern = "shooting_star"
# Strong Bearish Close: body > 60% of range, bearish candle
elif body / full_range > 0.60 and current["close"] < current["open"]:
has_reversal = True
pattern = "strong_bearish_close"
if not has_reversal:
return None
# ---------------------------------------------------------------
# CONFIRMATION FILTERS (Volume required, RSI optional)
# ---------------------------------------------------------------
# Volume > 1.2x 20-period average (REQUIRED)
has_volume = False
if "volume" in current.index:
vol = current["volume"]
vol_avg = data["volume"].iloc[max(0, idx - 20):idx].mean()
if vol_avg > 0 and vol > 1.2 * vol_avg:
has_volume = True
if not has_volume:
return None
# RSI < 40 (LONG) or > 60 (SHORT) in last 3 bars (OPTIONAL)
has_rsi = False
for j in range(max(0, idx - 2), idx + 1):
bar_rsi = data.iloc[j].get("rsi_14", 50)
if direction == "LONG" and bar_rsi < 40:
has_rsi = True
break
elif direction == "SHORT" and bar_rsi > 60:
has_rsi = True
break
# Larger position if RSI confirms (1.5% vs 1%)
risk_pct = 0.015 if has_rsi else 0.01
# ---------------------------------------------------------------
# ENTRY, SL, TP LEVELS
# ---------------------------------------------------------------
entry = close
# SL: 15min 200 EMA +/- 0.5 ATR buffer
ema_200 = current.get("ema_200", np.nan)
if np.isnan(ema_200):
return None
if direction == "LONG":
sl = ema_200 - 0.5 * atr_val
else:
sl = ema_200 + 0.5 * atr_val
# TP1: fixed 4.0 ATR from entry (close 60%)
if direction == "LONG":
tp1 = entry + 4.0 * atr_val
else:
tp1 = entry - 4.0 * atr_val
# TP2: set equal to TP1 so it triggers immediately (activates trailing)
tp2 = tp1
# TP3: very far target — 5 ATR trailing stop manages the runner exit
if direction == "LONG":
tp3 = entry + 20.0 * atr_val
else:
tp3 = entry - 20.0 * atr_val
confirmations = 1 + (1 if has_rsi else 0) # Volume + optional RSI
confluence = confirmations + 2 # +2 for trend + pattern
return {
"direction": direction,
"sl": sl,
"tp1": tp1,
"tp2": tp2,
"tp3": tp3,
"confluence": min(confluence, 5),
"entry_pattern": f"ema_bounce_{pattern}",
"tp_splits": (0.60, 0.0, 0.40), # 60% at TP1, 0% at TP2, 40% runner
"trail_atr_mult": 5.0,
"max_bars": 120,
"no_breakeven": False, # Breakeven after TP1 = trailing floor at entry
"risk_pct": risk_pct,
}
def _is_bullish_engulfing(self, data: pd.DataFrame, idx: int) -> bool:
prev = data.iloc[idx - 1]
curr = data.iloc[idx]
prev_body = abs(prev["close"] - prev["open"])
curr_body = abs(curr["close"] - curr["open"])
return (prev["close"] < prev["open"] and # prev bearish
curr["close"] > curr["open"] and # curr bullish
curr_body > prev_body and # engulfs
curr["open"] <= prev["close"] and
curr["close"] >= prev["open"])
def _is_bearish_engulfing(self, data: pd.DataFrame, idx: int) -> bool:
prev = data.iloc[idx - 1]
curr = data.iloc[idx]
prev_body = abs(prev["close"] - prev["open"])
curr_body = abs(curr["close"] - curr["open"])
return (prev["close"] > prev["open"] and # prev bullish
curr["close"] < curr["open"] and # curr bearish
curr_body > prev_body and # engulfs
curr["open"] >= prev["close"] and
curr["close"] <= prev["open"])
def _find_next_htf_level(self, entry_price: float, direction: str,
atr_val: float,
timestamp) -> Optional[float]:
"""Find next 1H key S/R level from HTF swing points."""
if self.htf_data is None:
return None
htf = self.htf_data[self.htf_data.index < timestamp]
if len(htf) < 50:
return None
htf_recent = htf.iloc[-200:]
if direction == "LONG":
mask = htf_recent.get("is_swing_high",
pd.Series(False, index=htf_recent.index))
swing_prices = htf_recent.loc[mask == True, "high"]
if len(swing_prices) == 0:
return None
above = swing_prices[swing_prices > entry_price + 0.5 * atr_val]
if len(above) == 0:
return None
return float(above.min())
else:
mask = htf_recent.get("is_swing_low",
pd.Series(False, index=htf_recent.index))
swing_prices = htf_recent.loc[mask == True, "low"]
if len(swing_prices) == 0:
return None
below = swing_prices[swing_prices < entry_price - 0.5 * atr_val]
if len(below) == 0:
return None
return float(below.max())
src/strategies_pkg/s6a_ema_bounce.py
+310
View File
@@ -0,0 +1,310 @@
"""
Strategy 6A: EMA Bounce Continuation — Three-Checkpoint Momentum Filter.
Same core logic as S6 but with:
1. Three-checkpoint momentum filter on 1H timeframe
2. Two-part EMA separation filter (historical avg + current)
1H Momentum Filter (Three Checkpoints):
avg_price_recent = mean(1H close) from 60..40 bars ago
avg_price_middle = mean(1H close) from 110..90 bars ago
avg_price_old = mean(1H close) from 160..140 bars ago
pct_change_recent = (avg_price_recent - avg_price_middle) / avg_price_middle * 100
pct_change_older = (avg_price_middle - avg_price_old) / avg_price_old * 100
LONG: both > 0.5
SHORT: both < -0.5
EMA Separation (Two-Part):
1. Avg |50 EMA - 100 EMA| over last 50 bars must be > 0.05% of price
2. Current |50 EMA - 100 EMA| must be > 0.05% of price
"""
from typing import Optional
import numpy as np
import pandas as pd
from .base import BaseStrategy
class S6A_EMA_Bounce(BaseStrategy):
strategy_id = 6
name = "S6A_EMA_Bounce_ThreeCheckpoint"
def check_signal(self, data: pd.DataFrame, idx: int,
current: pd.Series,
htf_row: Optional[pd.Series] = None) -> Optional[dict]:
if idx < 200:
return None
# Session filter: 08:00-16:00 UTC
hour = current.name.hour if hasattr(current.name, 'hour') else 0
if hour < 8 or hour >= 16:
return None
atr_val = current.get("atr_14", 0)
if atr_val <= 0 or np.isnan(atr_val):
return None
if htf_row is None or self.htf_data is None:
return None
# ---------------------------------------------------------------
# 1H TREND FILTER
# ---------------------------------------------------------------
htf_close = htf_row.get("close", np.nan)
htf_ema50 = htf_row.get("ema_50", np.nan)
htf_ema200 = htf_row.get("ema_200", np.nan)
if any(np.isnan(v) for v in [htf_close, htf_ema50, htf_ema200]):
return None
long_trend = htf_close > htf_ema200 and htf_ema50 > htf_ema200
short_trend = htf_close < htf_ema200 and htf_ema50 < htf_ema200
if not long_trend and not short_trend:
return None
direction = "LONG" if long_trend else "SHORT"
# ---------------------------------------------------------------
# 1H THREE-CHECKPOINT MOMENTUM FILTER
# ---------------------------------------------------------------
timestamp = current.name
htf = self.htf_data[self.htf_data.index < timestamp]
if len(htf) < 161:
return None
avg_price_recent = htf["close"].iloc[-60:-40].mean()
avg_price_middle = htf["close"].iloc[-110:-90].mean()
avg_price_old = htf["close"].iloc[-160:-140].mean()
if any(np.isnan(v) or v <= 0 for v in [avg_price_recent, avg_price_middle, avg_price_old]):
return None
pct_change_recent = (avg_price_recent - avg_price_middle) / avg_price_middle * 100
pct_change_older = (avg_price_middle - avg_price_old) / avg_price_old * 100
if direction == "LONG":
if not (pct_change_recent > 0.25 and pct_change_older > 0.25):
return None
else:
if not (pct_change_recent < -0.25 and pct_change_older < -0.25):
return None
# ---------------------------------------------------------------
# 15MIN EMA SETUP (prevents counter-trend entries)
# ---------------------------------------------------------------
ema_50 = current.get("ema_50", np.nan)
ema_100 = current.get("ema_100", np.nan)
if np.isnan(ema_50) or np.isnan(ema_100):
return None
if direction == "LONG" and not (ema_50 > ema_100):
return None
if direction == "SHORT" and not (ema_50 < ema_100):
return None
# ---------------------------------------------------------------
# TWO-PART EMA SEPARATION FILTER
# ---------------------------------------------------------------
price = current["close"]
current_sep = abs(ema_50 - ema_100)
# Part 1: Average separation over last 50 bars > 0.05% of price
if idx >= 50:
seps = []
for j in range(idx - 50, idx):
bar = data.iloc[j]
e50 = bar.get("ema_50", np.nan)
e100 = bar.get("ema_100", np.nan)
if not np.isnan(e50) and not np.isnan(e100):
seps.append(abs(e50 - e100))
if len(seps) > 0:
avg_sep = np.mean(seps)
if avg_sep < 0.0003 * price:
return None
else:
return None
else:
return None
# Part 2: Current separation > 0.05% of price
if current_sep < 0.0003 * price:
return None
# ---------------------------------------------------------------
# 15MIN EMA CONVERGENCE (prevents entries when trend weakening)
# ---------------------------------------------------------------
if idx >= 20:
past_ema50 = data.iloc[idx - 20].get("ema_50", np.nan)
past_ema100 = data.iloc[idx - 20].get("ema_100", np.nan)
if not np.isnan(past_ema50) and not np.isnan(past_ema100):
past_sep = abs(past_ema50 - past_ema100)
if past_sep > 0 and current_sep < 0.70 * past_sep:
return None
# ---------------------------------------------------------------
# 15MIN GENUINE BOUNCE ENTRY
# ---------------------------------------------------------------
close = current["close"]
lookback_start = max(0, idx - 10)
lookback_end = max(0, idx - 3)
total_check_bars = lookback_end - lookback_start
if total_check_bars < 4:
return None
trend_side_count = 0
for j in range(lookback_start, lookback_end):
bar = data.iloc[j]
bar_ema100 = bar.get("ema_100", np.nan)
if np.isnan(bar_ema100):
continue
if direction == "LONG" and bar["close"] > bar_ema100:
trend_side_count += 1
elif direction == "SHORT" and bar["close"] < bar_ema100:
trend_side_count += 1
if trend_side_count / total_check_bars < 0.70:
return None
# Pullback check
had_pullback = False
for j in range(max(0, idx - 3), idx):
bar_close = data.iloc[j]["close"]
bar_ema100 = data.iloc[j].get("ema_100", np.nan)
if np.isnan(bar_ema100):
continue
if direction == "LONG" and bar_close < bar_ema100:
had_pullback = True
break
elif direction == "SHORT" and bar_close > bar_ema100:
had_pullback = True
break
if not had_pullback:
return None
# OHLC void
if idx >= 3:
all_wrong_side = True
for j in range(idx - 3, idx):
bar = data.iloc[j]
bar_ema100 = bar.get("ema_100", np.nan)
if np.isnan(bar_ema100):
all_wrong_side = False
break
if direction == "LONG" and bar["high"] >= bar_ema100:
all_wrong_side = False
break
elif direction == "SHORT" and bar["low"] <= bar_ema100:
all_wrong_side = False
break
if all_wrong_side:
return None
# Bounce confirmation
if direction == "LONG" and close <= ema_100:
return None
if direction == "SHORT" and close >= ema_100:
return None
# Price within 1.0 ATR of 100 EMA
if abs(close - ema_100) > 1.0 * atr_val:
return None
# ---------------------------------------------------------------
# REVERSAL PATTERN
# ---------------------------------------------------------------
body = abs(current["close"] - current["open"])
full_range = current["high"] - current["low"]
if full_range <= 0:
return None
upper_wick = current["high"] - max(current["close"], current["open"])
lower_wick = min(current["close"], current["open"]) - current["low"]
close_position = (current["close"] - current["low"]) / full_range
has_reversal = False
pattern = ""
if direction == "LONG":
if body > 0 and lower_wick >= 2.0 * body and close_position >= 0.75:
has_reversal = True
pattern = "hammer"
elif body / full_range > 0.60 and current["close"] > current["open"]:
has_reversal = True
pattern = "strong_bullish_close"
else:
if body > 0 and upper_wick >= 2.0 * body and close_position <= 0.25:
has_reversal = True
pattern = "shooting_star"
elif body / full_range > 0.60 and current["close"] < current["open"]:
has_reversal = True
pattern = "strong_bearish_close"
if not has_reversal:
return None
# ---------------------------------------------------------------
# CONFIRMATION FILTERS
# ---------------------------------------------------------------
has_volume = False
if "volume" in current.index:
vol = current["volume"]
vol_avg = data["volume"].iloc[max(0, idx - 20):idx].mean()
if vol_avg > 0 and vol > 1.2 * vol_avg:
has_volume = True
if not has_volume:
return None
# RSI optional
has_rsi = False
for j in range(max(0, idx - 2), idx + 1):
bar_rsi = data.iloc[j].get("rsi_14", 50)
if direction == "LONG" and bar_rsi < 40:
has_rsi = True
break
elif direction == "SHORT" and bar_rsi > 60:
has_rsi = True
break
risk_pct = 0.015 if has_rsi else 0.01
# ---------------------------------------------------------------
# ENTRY, SL, TP LEVELS
# ---------------------------------------------------------------
entry = close
ema_200 = current.get("ema_200", np.nan)
if np.isnan(ema_200):
return None
if direction == "LONG":
sl = ema_200 - 0.5 * atr_val
else:
sl = ema_200 + 0.5 * atr_val
if direction == "LONG":
tp1 = entry + 4.0 * atr_val
else:
tp1 = entry - 4.0 * atr_val
tp2 = tp1
if direction == "LONG":
tp3 = entry + 20.0 * atr_val
else:
tp3 = entry - 20.0 * atr_val
confirmations = 1 + (1 if has_rsi else 0)
confluence = confirmations + 2
return {
"direction": direction,
"sl": sl,
"tp1": tp1,
"tp2": tp2,
"tp3": tp3,
"confluence": min(confluence, 5),
"entry_pattern": f"ema_bounce_{pattern}",
"tp_splits": (0.60, 0.0, 0.40),
"trail_atr_mult": 5.0,
"max_bars": 120,
"no_breakeven": False,
"risk_pct": risk_pct,
}
src/strategies_pkg/s6b_ema_bounce.py
+306
View File
@@ -0,0 +1,306 @@
"""
Strategy 6B: EMA Bounce Continuation — Two-Checkpoint Momentum Filter.
Same core logic as S6 but with:
1. Two-checkpoint momentum filter on 1H timeframe
2. Two-part EMA separation filter (historical avg + current)
1H Momentum Filter (Two Checkpoints):
avg_price_recent = mean(1H close) from 60..40 bars ago
avg_price_old = mean(1H close) from 160..140 bars ago
total_pct_change = (avg_price_recent - avg_price_old) / avg_price_old * 100
LONG: total_pct_change > 1.0
SHORT: total_pct_change < -1.0
EMA Separation (Two-Part):
1. Avg |50 EMA - 100 EMA| over last 50 bars must be > 0.05% of price
2. Current |50 EMA - 100 EMA| must be > 0.05% of price
"""
from typing import Optional
import numpy as np
import pandas as pd
from .base import BaseStrategy
class S6B_EMA_Bounce(BaseStrategy):
strategy_id = 6
name = "S6B_EMA_Bounce_TwoCheckpoint"
def check_signal(self, data: pd.DataFrame, idx: int,
current: pd.Series,
htf_row: Optional[pd.Series] = None) -> Optional[dict]:
if idx < 200:
return None
# Session filter: 08:00-16:00 UTC
hour = current.name.hour if hasattr(current.name, 'hour') else 0
if hour < 8 or hour >= 16:
return None
atr_val = current.get("atr_14", 0)
if atr_val <= 0 or np.isnan(atr_val):
return None
if htf_row is None or self.htf_data is None:
return None
# ---------------------------------------------------------------
# 1H TREND FILTER
# ---------------------------------------------------------------
htf_close = htf_row.get("close", np.nan)
htf_ema50 = htf_row.get("ema_50", np.nan)
htf_ema200 = htf_row.get("ema_200", np.nan)
if any(np.isnan(v) for v in [htf_close, htf_ema50, htf_ema200]):
return None
long_trend = htf_close > htf_ema200 and htf_ema50 > htf_ema200
short_trend = htf_close < htf_ema200 and htf_ema50 < htf_ema200
if not long_trend and not short_trend:
return None
direction = "LONG" if long_trend else "SHORT"
# ---------------------------------------------------------------
# 1H TWO-CHECKPOINT MOMENTUM FILTER
# ---------------------------------------------------------------
timestamp = current.name
htf = self.htf_data[self.htf_data.index < timestamp]
if len(htf) < 161:
return None
avg_price_recent = htf["close"].iloc[-60:-40].mean()
avg_price_old = htf["close"].iloc[-160:-140].mean()
if any(np.isnan(v) or v <= 0 for v in [avg_price_recent, avg_price_old]):
return None
total_pct_change = (avg_price_recent - avg_price_old) / avg_price_old * 100
if direction == "LONG":
if not (total_pct_change > 0.5):
return None
else:
if not (total_pct_change < -0.5):
return None
# ---------------------------------------------------------------
# 15MIN EMA SETUP (prevents counter-trend entries)
# ---------------------------------------------------------------
ema_50 = current.get("ema_50", np.nan)
ema_100 = current.get("ema_100", np.nan)
if np.isnan(ema_50) or np.isnan(ema_100):
return None
if direction == "LONG" and not (ema_50 > ema_100):
return None
if direction == "SHORT" and not (ema_50 < ema_100):
return None
# ---------------------------------------------------------------
# TWO-PART EMA SEPARATION FILTER
# ---------------------------------------------------------------
price = current["close"]
current_sep = abs(ema_50 - ema_100)
# Part 1: Average separation over last 50 bars > 0.05% of price
if idx >= 50:
seps = []
for j in range(idx - 50, idx):
bar = data.iloc[j]
e50 = bar.get("ema_50", np.nan)
e100 = bar.get("ema_100", np.nan)
if not np.isnan(e50) and not np.isnan(e100):
seps.append(abs(e50 - e100))
if len(seps) > 0:
avg_sep = np.mean(seps)
if avg_sep < 0.0003 * price:
return None
else:
return None
else:
return None
# Part 2: Current separation > 0.05% of price
if current_sep < 0.0003 * price:
return None
# ---------------------------------------------------------------
# 15MIN EMA CONVERGENCE (prevents entries when trend weakening)
# ---------------------------------------------------------------
if idx >= 20:
past_ema50 = data.iloc[idx - 20].get("ema_50", np.nan)
past_ema100 = data.iloc[idx - 20].get("ema_100", np.nan)
if not np.isnan(past_ema50) and not np.isnan(past_ema100):
past_sep = abs(past_ema50 - past_ema100)
if past_sep > 0 and current_sep < 0.70 * past_sep:
return None
# ---------------------------------------------------------------
# 15MIN GENUINE BOUNCE ENTRY
# ---------------------------------------------------------------
close = current["close"]
lookback_start = max(0, idx - 10)
lookback_end = max(0, idx - 3)
total_check_bars = lookback_end - lookback_start
if total_check_bars < 4:
return None
trend_side_count = 0
for j in range(lookback_start, lookback_end):
bar = data.iloc[j]
bar_ema100 = bar.get("ema_100", np.nan)
if np.isnan(bar_ema100):
continue
if direction == "LONG" and bar["close"] > bar_ema100:
trend_side_count += 1
elif direction == "SHORT" and bar["close"] < bar_ema100:
trend_side_count += 1
if trend_side_count / total_check_bars < 0.70:
return None
# Pullback check
had_pullback = False
for j in range(max(0, idx - 3), idx):
bar_close = data.iloc[j]["close"]
bar_ema100 = data.iloc[j].get("ema_100", np.nan)
if np.isnan(bar_ema100):
continue
if direction == "LONG" and bar_close < bar_ema100:
had_pullback = True
break
elif direction == "SHORT" and bar_close > bar_ema100:
had_pullback = True
break
if not had_pullback:
return None
# OHLC void
if idx >= 3:
all_wrong_side = True
for j in range(idx - 3, idx):
bar = data.iloc[j]
bar_ema100 = bar.get("ema_100", np.nan)
if np.isnan(bar_ema100):
all_wrong_side = False
break
if direction == "LONG" and bar["high"] >= bar_ema100:
all_wrong_side = False
break
elif direction == "SHORT" and bar["low"] <= bar_ema100:
all_wrong_side = False
break
if all_wrong_side:
return None
# Bounce confirmation
if direction == "LONG" and close <= ema_100:
return None
if direction == "SHORT" and close >= ema_100:
return None
# Price within 1.0 ATR of 100 EMA
if abs(close - ema_100) > 1.0 * atr_val:
return None
# ---------------------------------------------------------------
# REVERSAL PATTERN
# ---------------------------------------------------------------
body = abs(current["close"] - current["open"])
full_range = current["high"] - current["low"]
if full_range <= 0:
return None
upper_wick = current["high"] - max(current["close"], current["open"])
lower_wick = min(current["close"], current["open"]) - current["low"]
close_position = (current["close"] - current["low"]) / full_range
has_reversal = False
pattern = ""
if direction == "LONG":
if body > 0 and lower_wick >= 2.0 * body and close_position >= 0.75:
has_reversal = True
pattern = "hammer"
elif body / full_range > 0.60 and current["close"] > current["open"]:
has_reversal = True
pattern = "strong_bullish_close"
else:
if body > 0 and upper_wick >= 2.0 * body and close_position <= 0.25:
has_reversal = True
pattern = "shooting_star"
elif body / full_range > 0.60 and current["close"] < current["open"]:
has_reversal = True
pattern = "strong_bearish_close"
if not has_reversal:
return None
# ---------------------------------------------------------------
# CONFIRMATION FILTERS
# ---------------------------------------------------------------
has_volume = False
if "volume" in current.index:
vol = current["volume"]
vol_avg = data["volume"].iloc[max(0, idx - 20):idx].mean()
if vol_avg > 0 and vol > 1.2 * vol_avg:
has_volume = True
if not has_volume:
return None
# RSI optional
has_rsi = False
for j in range(max(0, idx - 2), idx + 1):
bar_rsi = data.iloc[j].get("rsi_14", 50)
if direction == "LONG" and bar_rsi < 40:
has_rsi = True
break
elif direction == "SHORT" and bar_rsi > 60:
has_rsi = True
break
risk_pct = 0.015 if has_rsi else 0.01
# ---------------------------------------------------------------
# ENTRY, SL, TP LEVELS
# ---------------------------------------------------------------
entry = close
ema_200 = current.get("ema_200", np.nan)
if np.isnan(ema_200):
return None
if direction == "LONG":
sl = ema_200 - 0.5 * atr_val
else:
sl = ema_200 + 0.5 * atr_val
if direction == "LONG":
tp1 = entry + 4.0 * atr_val
else:
tp1 = entry - 4.0 * atr_val
tp2 = tp1
if direction == "LONG":
tp3 = entry + 20.0 * atr_val
else:
tp3 = entry - 20.0 * atr_val
confirmations = 1 + (1 if has_rsi else 0)
confluence = confirmations + 2
return {
"direction": direction,
"sl": sl,
"tp1": tp1,
"tp2": tp2,
"tp3": tp3,
"confluence": min(confluence, 5),
"entry_pattern": f"ema_bounce_{pattern}",
"tp_splits": (0.60, 0.0, 0.40),
"trail_atr_mult": 5.0,
"max_bars": 120,
"no_breakeven": False,
"risk_pct": risk_pct,
}