Multi-Indicator Strategies

Single-indicator strategies are simple but fragile. They work in certain market conditions and fail in others. Combining multiple indicators — looking for confluence — produces more reliable signals because each indicator must independently agree before you take a trade.

Why Combine Indicators?

Each indicator has blind spots:

• Moving averages identify trends but lag at reversals
• RSI catches reversals but whipsaws in trends
• Volume confirms moves but does not give direction alone
• ATR measures volatility but says nothing about direction

When multiple indicators agree simultaneously, the probability of a good trade increases significantly. This is called confluence.

Note

The goal is not to pile on indicators until one of them always says “buy.” The goal is to find 2-3 indicators that measure different things (trend, momentum, volatility) and require all of them to agree.

Indicator Categories

Choose one indicator from each category:

Category	Measures	Indicators
Trend	Direction	EMA, SMA, Price vs. MA
Momentum	Speed of change	RSI, Stochastic, MACD
Volatility	Range/expansion	ATR, Bollinger Bands, Bar Range
Volume	Participation	Volume vs. Average, OBV

Rule: Never combine two indicators from the same category. Two trend indicators (e.g., EMA + SMA) give redundant information and a false sense of confirmation.

---

Building Blocks: Reusable Indicator Classes

Here are the indicator classes you will use. Include these at the top of your strategy:

class EMA:
    def __init__(self, period):
        self.period = period
        self.multiplier = 2 / (period + 1)
        self.value = None
        self._buffer = []
    def update(self, price):
        if self.value is None:
            self._buffer.append(price)
            if len(self._buffer) == self.period:
                self.value = sum(self._buffer) / self.period
                self._buffer = None
        else:
            self.value = (price - self.value) * self.multiplier + self.value
        return self.value

class RSI:
    def __init__(self, period=14):
        self.period = period
        self.value = None
        self._prev_close = None
        self._avg_gain = None
        self._avg_loss = None
        self._gains = []
        self._losses = []
    def update(self, close):
        if self._prev_close is not None:
            change = close - self._prev_close
            gain = max(change, 0)
            loss = abs(min(change, 0))
            if self._avg_gain is None:
                self._gains.append(gain)
                self._losses.append(loss)
                if len(self._gains) == self.period:
                    self._avg_gain = sum(self._gains) / self.period
                    self._avg_loss = sum(self._losses) / self.period
                    self._gains = None
                    self._losses = None
                    if self._avg_loss == 0:
                        self.value = 100.0
                    else:
                        self.value = 100 - (100 / (1 + self._avg_gain / self._avg_loss))
            else:
                self._avg_gain = (self._avg_gain * (self.period - 1) + gain) / self.period
                self._avg_loss = (self._avg_loss * (self.period - 1) + loss) / self.period
                if self._avg_loss == 0:
                    self.value = 100.0
                else:
                    self.value = 100 - (100 / (1 + self._avg_gain / self._avg_loss))
        self._prev_close = close
        return self.value

class ATR:
    def __init__(self, period=14):
        self.period = period
        self.value = None
        self._prev_close = None
        self._tr_buffer = []
    def update(self, high, low, close):
        if self._prev_close is not None:
            tr = max(high - low, abs(high - self._prev_close), abs(low - self._prev_close))
            if self.value is None:
                self._tr_buffer.append(tr)
                if len(self._tr_buffer) == self.period:
                    self.value = sum(self._tr_buffer) / self.period
                    self._tr_buffer = None
            else:
                self.value = (self.value * (self.period - 1) + tr) / self.period
        self._prev_close = close
        return self.value

---

Strategy: Trend + Momentum + Volume

This strategy requires three conditions to align for entry:

1. Trend (EMA 50): Price above EMA = bullish trend, price below = bearish
2. Momentum (RSI 14): RSI between 40-60 for entries (not overextended)
3. Volume: Current volume above 20-bar average (participation confirmation)

The idea: enter in the direction of the trend when RSI shows a pullback (not extreme) and volume confirms interest.

#!/usr/bin/env python3
"""
Multi-Indicator Strategy — TestMax Algo Playground
Trend (EMA) + Momentum (RSI) + Volume confirmation
"""
import os, time

# --- Setup ---
ACCOUNT_ID = int(os.environ.get("ACCOUNT_ID", "0"))
CONTRACT_ID = os.environ.get("CONTRACT_ID", "")
TOTAL_BARS = int(os.environ.get("TOTAL_BARS", "5000"))
STEP_DELAY = float(os.environ.get("STEP_DELAY", "0.02"))
SPEED_FILE = os.environ.get("SPEED_FILE", "")

# --- Parameters ---
EMA_PERIOD = 50          # Trend filter
RSI_PERIOD = 14          # Momentum
RSI_BUY_ZONE = (35, 50)  # Buy when RSI is in this range (pullback, not oversold)
RSI_SELL_ZONE = (50, 65)  # Sell when RSI is in this range (bounce, not overbought)
VOL_MULT = 1.2           # Volume must be 1.2x average
SL_TICKS = 20
TP_RATIO = 2.0

print(f"[INFO] Strategy: EMA({EMA_PERIOD}) + RSI({RSI_PERIOD}) + Volume Filter")
print(f"[INFO] Buy zone: RSI {RSI_BUY_ZONE} | Sell zone: RSI {RSI_SELL_ZONE}")
print("-" * 50)

# (Include EMA, RSI, ATR classes from above)

# --- State ---
ema = EMA(EMA_PERIOD)
rsi = RSI(RSI_PERIOD)
volumes = []
position = None
entry_price = 0.0
trade_count = 0
wins = 0
balance_before = 50000

# Daily risk
current_day = ""
daily_pnl = 0.0
daily_trades = 0
daily_locked = False
MAX_DAILY_LOSS = 1000
MAX_TRADES = 8

try:
    for i in range(TOTAL_BARS):
        bar = get_next_bar()
        if bar is None:
            break
        read_speed()
        if STEP_DELAY > 0:
            time.sleep(STEP_DELAY)

        price = bar["c"]
        volumes.append(bar["v"])

        # Update indicators
        ema_val = ema.update(price)
        rsi_val = rsi.update(price)

        if ema_val is None or rsi_val is None or len(volumes) < 20:
            continue

        # --- Daily Reset ---
        bar_day = bar["t"][:10]
        if bar_day != current_day:
            if current_day:
                print(f"[DAY]  {current_day} | P&L: ${daily_pnl:+,.2f} | Trades: {daily_trades}")
            current_day = bar_day
            daily_pnl = 0.0
            daily_trades = 0
            daily_locked = False

        # --- Check position closed by engine ---
        if position is not None and i % 5 == 0:
            positions = get_positions(ACCOUNT_ID)
            if len(positions) == 0:
                acct = get_account(ACCOUNT_ID)
                bal = acct["balance"] if acct else balance_before
                pnl = bal - balance_before
                daily_pnl += pnl
                trade_count += 1
                daily_trades += 1
                if pnl >= 0:
                    wins += 1
                print(f"[CLOSE] P&L: ${pnl:+,.0f} | Bal: ${bal:,.0f}")
                position = None

        # --- Risk checks ---
        if daily_pnl <= -MAX_DAILY_LOSS and not daily_locked:
            daily_locked = True
            if position:
                close_all_positions(ACCOUNT_ID, CONTRACT_ID)
                position = None
            print(f"[RISK] Daily limit hit: ${daily_pnl:,.2f}")

        if daily_trades >= MAX_TRADES:
            daily_locked = True

        if position is not None or daily_locked:
            continue

        # === MULTI-INDICATOR SIGNAL ===

        # 1. Trend: price vs EMA
        trend_up = price > ema_val
        trend_down = price < ema_val

        # 2. Momentum: RSI in the buy/sell zone
        rsi_buy = RSI_BUY_ZONE[0] <= rsi_val <= RSI_BUY_ZONE[1]
        rsi_sell = RSI_SELL_ZONE[0] <= rsi_val <= RSI_SELL_ZONE[1]

        # 3. Volume: above average
        avg_vol = sum(volumes[-20:]) / 20
        vol_confirm = bar["v"] > avg_vol * VOL_MULT

        # --- Entry ---
        tp_ticks = int(SL_TICKS * TP_RATIO)

        if trend_up and rsi_buy and vol_confirm:
            result = place_order(ACCOUNT_ID, CONTRACT_ID, 0, 1,
                                 sl_ticks=SL_TICKS, tp_ticks=tp_ticks)
            if result.get("orderId") or result.get("success"):
                position = "LONG"
                entry_price = price
                acct = get_account(ACCOUNT_ID)
                balance_before = acct["balance"] if acct else balance_before
                print(f"[BUY]  Bar {i} @ {price:.2f} | RSI: {rsi_val:.1f} | Vol: {bar['v']}/{avg_vol:.0f}")

        elif trend_down and rsi_sell and vol_confirm:
            result = place_order(ACCOUNT_ID, CONTRACT_ID, 1, 1,
                                 sl_ticks=SL_TICKS, tp_ticks=tp_ticks)
            if result.get("orderId") or result.get("success"):
                position = "SHORT"
                entry_price = price
                acct = get_account(ACCOUNT_ID)
                balance_before = acct["balance"] if acct else balance_before
                print(f"[SELL] Bar {i} @ {price:.2f} | RSI: {rsi_val:.1f} | Vol: {bar['v']}/{avg_vol:.0f}")

        # Log every 500 bars
        if i % 500 == 0:
            trend = "UP" if price > ema_val else "DOWN"
            print(f"[INFO] Bar {i} | {price:.2f} | EMA: {ema_val:.2f} ({trend}) | RSI: {rsi_val:.1f}")

finally:
    try:
        close_all_positions(ACCOUNT_ID, CONTRACT_ID)
    except:
        pass
    account = get_account(ACCOUNT_ID)
    balance = account["balance"] if account else 50000
    print("-" * 50)
    print(f"[DONE] Balance: ${balance:,.2f} | P&L: ${balance - 50000:+,.2f}")
    print(f"[DONE] Trades: {trade_count} | Win Rate: {wins/trade_count*100:.1f}%" if trade_count else "[DONE] No trades")

---

How the Confluence Logic Works

Let’s trace through a buy signal scenario:

Bar 1500:
  Price:  21,480.00
  EMA50:  21,460.00   → price > EMA → trend UP  ✓
  RSI14:  42.3        → in range [35, 50]        ✓
  Volume: 1,800       → avg is 1,400 × 1.2 = 1,680 → 1,800 > 1,680  ✓

  All three conditions met → BUY signal

If any one condition fails, no trade:

Bar 1600:
  Price:  21,490.00
  EMA50:  21,465.00   → price > EMA → trend UP  ✓
  RSI14:  68.2        → NOT in range [35, 50]    ✗  (too high, overextended)
  Volume: 2,100       → above average            ✓

  RSI condition fails → NO trade (momentum overextended)

---

Adding ATR as a Fourth Filter

You can add a volatility filter to avoid entering in low-volatility chop:

atr = ATR(14)
atr_values = []

# In your loop:
atr_val = atr.update(bar["h"], bar["l"], bar["c"])
if atr_val:
    atr_values.append(atr_val)

# Check if volatility is expanding (favorable for breakouts)
if len(atr_values) >= 50:
    avg_atr = sum(atr_values[-50:]) / 50
    vol_expanding = atr_val > avg_atr  # Current ATR above its own average
else:
    vol_expanding = True  # Not enough data, skip filter

# Add to your entry condition:
if trend_up and rsi_buy and vol_confirm and vol_expanding:
    # Even higher-confidence signal
    pass

Scoring-Based Approach

Instead of requiring all conditions to be true (AND logic), you can assign scores and require a minimum threshold:

score = 0
max_score = 0

# Trend: +2 for strong, +1 for moderate
max_score += 2
if price > ema_val:
    ema_distance = (price - ema_val) / ema_val * 100  # % above EMA
    if ema_distance > 0.5:
        score += 2  # Strong uptrend
    else:
        score += 1  # Weak uptrend

# RSI: +2 for ideal zone, +1 for acceptable
max_score += 2
if 35 <= rsi_val <= 45:
    score += 2  # Ideal pullback zone
elif 45 < rsi_val <= 55:
    score += 1  # Acceptable

# Volume: +1 for confirmation
max_score += 1
if bar["v"] > avg_vol * VOL_MULT:
    score += 1

# Require at least 4 out of 5 points
MIN_SCORE = 4
if score >= MIN_SCORE:
    print(f"[SIGNAL] Score: {score}/{max_score}")
    # Enter trade

Tip

The scoring approach gives you more flexibility than strict AND logic. It allows a very strong trend signal to compensate for a marginally weak RSI reading, for example.

Common Mistakes

1. Too many indicators — 6+ indicators will almost never agree, and you will never trade. Stick to 2-3.
2. Redundant indicators — Two MAs or RSI + Stochastic measure the same thing. Use indicators from different categories.
3. Ignoring the market regime — A strategy tuned for trending markets will fail in ranges. Consider adding a regime detection filter (e.g., ADX > 25 for trending).
4. Over-optimization — Finding the “perfect” RSI range on past data does not mean it will work going forward. Keep parameters simple and robust.

What’s Next

• ICT Smart Money Strategy — the most advanced multi-concept strategy, combining market structure, FVGs, liquidity sweeps, and kill zones
• Parameter Optimization — systematically test different parameter combinations