Writing a custom strategy

Every strategy in edgekit is three methods on a subclass of BaseStrategy: prepare precomputes causal indicator arrays, entry decides when to open, and exit decides when to close. The engine does the rest — position bookkeeping, gap-aware fills, pessimistic stops, and cost charged in R. This guide builds a complete strategy from scratch and runs it through the gauntlet.

The interface#

BaseStrategy is an ABC satisfying the engine's Strategy protocol. You implement three abstract methods and inherit the .backtest() wiring for free:

prepare(self, bars) -> dict                    # precompute causal arrays (the "P" dict), once
entry(self, bars, P, i) -> EntryIntent | None  # on a FLAT bar i: open a position, or None
exit(self, bars, P, pos, i) -> float | None    # on an IN-POSITION bar i: exit price, or None

prepare runs once at the start of a backtest and returns the P dict of numpy arrays aligned to bars. This is where every indicator is computed — and lagged.
entry is polled on each bar i while flat. Return an EntryIntent(direction, level, stop_dist) to open, or None to stay out.
exit is polled on each bar while in a position (the current position is passed as pos). Return an exit price, or None to hold.

!Causality is your responsibility in prepare()

Indicators from edgekit.indicators are computed through bar i inclusive — they are NOT lagged. If your entry reads P["ema"][i] and that value already includes bar i's close, the backtest is reading the future. Always edgekit.core.lag(arr, 1) every indicator in prepare so that arr[i] reflects information only through bar i-1. Keeping the lag visible at the call site is the whole point of the design.

A complete worked example: EMA-cross breakout#

A fast/slow EMA trend-follower: go long on a golden cross, short on a death cross, risk a fixed multiple of ATR, and exit when the trend flips or the stop is hit. Every array is lagged in prepare; every decision reads only lagged values.

ma_cross.py

import numpy as np
from edgekit.strategy import BaseStrategy
from edgekit.engine import EntryIntent
from edgekit.core import lag
from edgekit import indicators as ind


class MACrossBreakout(BaseStrategy):
    name = "ma_cross"

    def __init__(self, fast: int = 20, slow: int = 100, atr_n: int = 20, stop_mult: float = 2.0):
        # tunables live in __init__ (never hard-coded in the loop) so the gauntlet can sweep them
        self.fast = fast
        self.slow = slow
        self.atr_n = atr_n
        self.stop_mult = stop_mult

    def prepare(self, bars) -> dict:
        o = bars["open"].to_numpy(float)
        h = bars["high"].to_numpy(float)
        l = bars["low"].to_numpy(float)
        c = bars["close"].to_numpy(float)
        # every indicator LAGGED by 1: arr[i] reflects info only through bar i-1
        return dict(
            o=o, h=h, l=l, c=c,
            fast=lag(ind.ema(c, self.fast), 1),
            slow=lag(ind.ema(c, self.slow), 1),
            atr=lag(ind.atr(h, l, c, self.atr_n), 1),
        )

    def entry(self, bars, P, i) -> EntryIntent | None:
        fast, slow, N = P["fast"], P["slow"], P["atr"]
        if not (np.isfinite(fast[i]) and np.isfinite(slow[i]) and np.isfinite(N[i]) and N[i] > 0):
            return None
        crossed_up = fast[i] > slow[i] and fast[i - 1] <= slow[i - 1]   # golden cross
        crossed_dn = fast[i] < slow[i] and fast[i - 1] >= slow[i - 1]   # death cross
        if crossed_up:
            d = 1
        elif crossed_dn:
            d = -1
        else:
            return None
        # level = this bar's open => a market entry (gap-aware fill collapses to the open);
        # risk one 'stop_mult * ATR' as the R denominator.
        return EntryIntent(direction=d, level=P["o"][i], stop_dist=self.stop_mult * N[i])

    def exit(self, bars, P, pos, i) -> float | None:
        o, h, l = P["o"], P["h"], P["l"]
        fast, slow = P["fast"], P["slow"]
        d = pos["d"]
        if d > 0:
            if l[i] <= pos["stop"]:            # hard stop touched intrabar
                return min(pos["stop"], o[i])  # pessimistic + gap-aware fill
            if fast[i] < slow[i]:              # trend flipped -> flatten at the open
                return o[i]
        else:
            if h[i] >= pos["stop"]:
                return max(pos["stop"], o[i])
            if fast[i] > slow[i]:
                return o[i]
        return None

The pos dict#

exit receives the live position the engine is tracking. The keys you can read:

Key	Meaning
`pos["d"]`	direction, +1 / -1
`pos["e"]`	the actual (gap-aware) entry fill price
`pos["stop"]`	the stop price, computed by the engine as `e - d * stop_dist`
`pos["rpx"]`	the stop distance in price units (1R)
`pos["ei"]`	the bar index the position opened on

You do not set the stop yourself — you supplied stop_dist in the EntryIntent and the engine turned it into a stop price. Your exit just checks whether the bar reached it.

How the engine applies fills and cost#

run_bar_loop is the workhorse behind .backtest(). Walking one bar at a time, one position at a time:

Entry fill is gap-aware. For a long, entry = max(level, open) — a market that gaps through your trigger fills at the (worse) open, never the stale level. Shorts mirror with min. Passing level = open (as above) makes it a clean market fill; passing a breakout price makes it a stop-entry.
Stops are pessimistic. When your exit returns a stop price, gap-through is filled at the worse of stop and open. When a bar could have hit both stop and target, edgekit takes the loss.
Cost is charged in R. Each trade's spread + swap (from the CostModel) is converted to R via its own stop distance and days held (bars_per_day converts bars → days), then subtracted. The r column is always net.

Run it#

import edgekit as ek

bars = ek.data.resample_ohlcv(ek.data.load_bars("US100_M1.csv"), "4h")

strat = MACrossBreakout(fast=20, slow=100, stop_mult=2.0)
trades = strat.backtest(bars, warmup=210, bars_per_day=6)   # H4 => 6 bars/day
stats = ek.trade_stats(trades.r.to_numpy(), dates=trades.date)
print(f"{stats['n']} trades | PF {stats['pf']:.2f} | EV {stats['ev_r']:+.3f}R | MAR {stats['mar']:.2f}")

warmup skips the leading bars where your longest indicator is still NaN (100-EMA needs a run of history); bars_per_day must match the timeframe (H4 = 6, daily = 1, a 5-min session = 48 per RTH day) so the swap cost is right.

Then prove it — the same gauntlet#

A custom strategy earns no trust until it survives the exact battery every candidate faces. Because your class satisfies the same interface, the null test is a two-line change from the proving-an-edge guide:

import numpy as np, pandas as pd
from edgekit.core import bootstrap_rng

o, h, l, c = (bars[x].to_numpy(float) for x in ("open", "high", "low", "close"))
idx = bars.index

def null_stat(rng):
    po, ph, pl, pc = ek.validation.permute_ohlc(o, h, l, c, rng)
    shuffled = pd.DataFrame({"open": po, "high": ph, "low": pl, "close": pc}, index=idx)
    t = strat.backtest(shuffled)          # YOUR strategy on permuted bars
    return float(t.r.sum()) if len(t) else 0.0

p = ek.validation.mcpt(float(trades.r.sum()), null_stat, n=100, rng=bootstrap_rng())
print(f"permutation p = {p:.4f}")         # p < 0.01 or it does not ship

✕Most custom strategies die here — that is the point

The permutation test is unforgiving, and it should be. If your EMA-cross returns p = 0.4, you have found noise, not an edge. Do not tune parameters until the number drops — that is the overfitting the gauntlet exists to catch. Kill it and move on.

Next#

API · strategy — BaseStrategy and the built-in strategy zoo.
API · engine — run_bar_loop, EntryIntent, fill_entry / fill_stop.
Causality — the lag discipline that keeps a backtest honest.
Proving an edge — the full gauntlet your strategy must pass.