Concepts
Three Types of Signals
Every strategy produces three distinct signal types. Mixing them up is the most common architecture mistake beginners make.
| Type | Purpose | Example |
|---|---|---|
| Entry Signal | When to open a new position | MACD crossover + RSI > 55 |
| Exit Signal | When to close an existing position | Opposite crossover OR stop hit |
| Filter Signal | Blocks trades in unfavourable conditions | Price below SMA200, VIX > 25, news window |
Design rule
A strategy should only produce an entry signal when all filters pass. Filters reject noise before the entry logic even runs — this alone cuts false signals by 40–60% in most strategies.
Architecture
State Machine Design
A strategy is always in one of three states. Transitions happen when signals fire. This mental model prevents the classic bug of entering a new trade while already in one.
FLAT
No position
No position
→entry LONG
LONG
Holding long
Holding long
→exit signal / SL
FLAT
FLAT
→entry SHORT
SHORT
Holding short
Holding short
→exit signal / SL
FLAT
PYTHONstrategy_base.py
from abc import ABC, abstractmethod from dataclasses import dataclass, field from typing import Literal import pandas as pd State = Literal["FLAT", "LONG", "SHORT"] @dataclass class Signal: direction : Literal["LONG", "SHORT", "EXIT", "NONE"] reason : str = "" strength : float = 1.0 # 0.0–1.0 confidence SIGNAL_NONE = Signal("NONE") SIGNAL_EXIT = Signal("EXIT") class StrategyBase(ABC): """ Base class for all strategies. Subclasses implement generate_signal() only. State tracking, trade logging handled here. """ def __init__(self, name: str): self.name = name self.state : State = "FLAT" self.trades : list[dict] = [] self._entry_price : float | None = None self._entry_time = None @abstractmethod def generate_signal(self, df: pd.DataFrame) -> Signal: """Given enriched OHLCV df, return the current signal.""" ... def on_candle(self, df: pd.DataFrame) -> Signal: """Call this each time a new candle closes.""" sig = self.generate_signal(df) row = df.iloc[-1] if self.state == "FLAT": if sig.direction in ("LONG", "SHORT"): self.state = sig.direction self._entry_price = row["close"] self._entry_time = row.name elif self.state in ("LONG", "SHORT"): if sig.direction == "EXIT" or ( sig.direction != "NONE" and sig.direction != self.state ): self.trades.append({ "entry_time" : self._entry_time, "exit_time" : row.name, "direction" : self.state, "entry_price" : self._entry_price, "exit_price" : row["close"], "pnl_pts" : ( row["close"] - self._entry_price if self.state == "LONG" else self._entry_price - row["close"] ), "reason": sig.reason, }) self.state = "FLAT" self._entry_price = None return sig def trade_log(self) -> pd.DataFrame: return pd.DataFrame(self.trades)
Implementation
EMA Crossover Strategy — Concrete Example
The cleanest way to learn the pattern is to implement a simple strategy. Here's a complete EMA-crossover strategy that inherits StrategyBase.
PYTHONstrategy_ema_cross.py
from strategy_base import StrategyBase, Signal, SIGNAL_NONE, SIGNAL_EXIT from indicators import ema, sma, rsi class EMACrossStrategy(StrategyBase): def __init__(self, fast=9, slow=21, rsi_period=14): super().__init__(name=f"EMA{fast}x{slow}") self.fast = fast self.slow = slow self.rsi_period = rsi_period def generate_signal(self, df): if len(df) < self.slow + 5: return SIGNAL_NONE c = df["close"] fast = ema(c, self.fast) slow = ema(c, self.slow) r = rsi(c, self.rsi_period) trend = sma(c, 200) # ── Filters ────────────────────────────────────────────── above_trend = c.iloc[-1] > trend.iloc[-1] below_trend = c.iloc[-1] < trend.iloc[-1] # ── Entry signals ──────────────────────────────────────── bull_cross = (fast.iloc[-2] < slow.iloc[-2]) and \ (fast.iloc[-1] > slow.iloc[-1]) bear_cross = (fast.iloc[-2] > slow.iloc[-2]) and \ (fast.iloc[-1] < slow.iloc[-1]) # ── Exit signals ───────────────────────────────────────── if self.state == "LONG" and bear_cross: return Signal("EXIT", "bear_cross") if self.state == "SHORT" and bull_cross: return Signal("EXIT", "bull_cross") # ── Entry (filter first) ────────────────────────────────── if bull_cross and above_trend and r.iloc[-1] > 50: return Signal("LONG", "bull_cross+trend", strength=r.iloc[-1]/100) if bear_cross and below_trend and r.iloc[-1] < 50: return Signal("SHORT", "bear_cross+trend", strength=(100-r.iloc[-1])/100) return SIGNAL_NONE
Key architectural benefits
Because StrategyBase handles all state transitions and trade logging, EMACrossStrategy contains zero boilerplate. Writing a new strategy only means implementing generate_signal() — typically 10–20 lines.
Patterns
Vectorised vs Event-Driven
There are two modes to run a strategy. You'll use both — vectorised for backtesting (fast), event-driven for live trading (real-time).
| Mode | How it works | When to use | Speed |
|---|---|---|---|
| Vectorised | Apply signals across entire DataFrame at once | Backtesting over years of data | Milliseconds |
| Event-driven | Feed candles one-by-one via on_candle() | Live trading / paper trading | Real-time |
PYTHONrun modes
# ── Event-driven (live) ─────────────────────────────────── strategy = EMACrossStrategy(fast=9, slow=21) def on_candle_close(candle): df = get_buffer(candle["token"]) # enriched DataFrame sig = strategy.on_candle(df) if sig.direction in ("LONG", "SHORT"): place_order(sig) # ── Vectorised (backtest) ───────────────────────────────── from backtest_engine import BacktestEngine # ← built in L19 engine = BacktestEngine(strategy=EMACrossStrategy()) result = engine.run(df_historical) print(result.summary())
Exercise
Build Your Own Strategy
Using StrategyBase, implement a SupertrendStrategy that:
- Enters LONG when Supertrend flips from -1 to +1 AND price > VWAP
- Enters SHORT when Supertrend flips from +1 to -1 AND price < VWAP
- Exits when Supertrend flips back to the opposite direction
You only need to implement one method: generate_signal(self, df). State management is handled by the base class.