prior-day-breakout/tests/reference.py

# SPDX-License-Identifier: LicenseRef-PolyForm-Noncommercial-1.0.0 (see LICENSE)
"""Python statement of the Prior Day Breakout rule, written from the rule text (README), on M1 bars.

    trades = run(m1, tf_minutes=60, target_r=2.0)

`m1` is a pandas DataFrame indexed by server time (naive timestamps, M1 open time) with columns open, high, low, close
(bid prices) and spread (in PRICE units, e.g. 0.35, not points). It returns one row per trade (gross, before swap):
entry/exit time, side, fill prices, stop distance, exit reason.

Rule: levels = high/low of the previous trading day (the previous day that has bars); on the chart timeframe, from the
bar that opens 01:00 on, the first CLOSE above the high (buy) or below the low (sell), once per side per day; the entry is
the next bar's open if it is before 20:00 of the same day; buys fill at open + spread, sells at the bid; the stop is the
other side of the range; skipped when the stop distance is under 8 x the entry spread; target = target_r x the stop
distance; the position is flat at 23:00 (the first M1 bar at or after it); one position at a time. Stops and targets are
tested on the M1 extremes (a buy on the bid low and high, a sell on the ask = bid + spread); the stop first when both fall
in one M1 bar; a gap fills the stop at the open."""
import numpy as np
import pandas as pd

FROM_SEC = 3600           # signal bars from 01:00
CUTOFF_SEC = 20 * 3600    # no entry from 20:00
FLAT_SEC = 23 * 3600      # flat at 23:00
MIN_SPREADS = 8.0         # skip when the stop is under 8 x the spread


def run(m1: pd.DataFrame, tf_minutes: int = 60, target_r: float = 2.0) -> pd.DataFrame:
    t = m1.index.values.astype("datetime64[s]").astype(np.int64)
    o, h, l, c, sp = (m1[k].to_numpy(float) for k in ("open", "high", "low", "close", "spread"))
    n = len(t)
    day = t // 86400
    starts = np.concatenate(([0], np.flatnonzero(np.diff(day)) + 1))
    dh, dl = np.maximum.reduceat(h, starts), np.minimum.reduceat(l, starts)
    pdh = np.concatenate(([np.nan], dh[:-1]))
    pdl = np.concatenate(([np.nan], dl[:-1]))
    day_of_start = day[starts]
    gidx = np.searchsorted(day_of_start, day)             # M1 index -> day group
    sec = tf_minutes * 60
    key = t // sec
    fm = np.concatenate(([0], np.flatnonzero(np.diff(key)) + 1))            # first M1 index of each chart bar
    last = np.concatenate((fm[1:] - 1, [n - 1]))
    bar_open, bar_close = key[fm] * sec, c[last]
    rows, free, cur_day, done = [], 0, -1, [False, False]
    for j in range(len(fm) - 1):
        bday, btod = bar_open[j] // 86400, bar_open[j] % 86400
        if bday != cur_day:
            cur_day, done = bday, [False, False]
        if btod < FROM_SEC:
            continue
        g = gidx[fm[j]]
        hi, lo = pdh[g], pdl[g]
        if np.isnan(hi) or np.isnan(lo):
            continue
        side = 0
        if bar_close[j] > hi and not done[0]:
            done[0], side = True, 1
        elif bar_close[j] < lo and not done[1]:
            done[1], side = True, -1
        if side == 0:
            continue
        e = fm[j + 1]
        if t[e] // 86400 != cur_day or t[e] % 86400 >= CUTOFF_SEC or e < free:
            continue
        fill = o[e] + sp[e] if side > 0 else o[e]
        sl = lo if side > 0 else hi
        sd = (fill - sl) * side
        if not (sd > 0) or sd < MIN_SPREADS * sp[e]:
            continue
        tp = fill + side * target_r * sd if target_r > 0 else np.nan
        fx = int(np.searchsorted(t, cur_day * 86400 + FLAT_SEC, side="left"))
        px, why, m_exit, nxt = (o[fx] if side > 0 else o[fx] + sp[fx]) if fx < n else c[-1], "time", min(fx, n - 1), min(fx, n - 1)
        for m in range(e, min(fx, n)):
            if side > 0:
                if l[m] <= sl:
                    px, why, m_exit, nxt = min(sl, o[m]), "stop", m, m + 1
                    break
                if tp == tp and h[m] >= tp:
                    px, why, m_exit, nxt = max(tp, o[m]), "target", m, m + 1
                    break
            else:
                if h[m] + sp[m] >= sl:
                    px, why, m_exit, nxt = max(sl, o[m] + sp[m]), "stop", m, m + 1
                    break
                if tp == tp and l[m] + sp[m] <= tp:
                    px, why, m_exit, nxt = min(tp, o[m] + sp[m]), "target", m, m + 1
                    break
        rows.append(dict(entry_t=t[e], exit_t=t[m_exit], side=side, entry=fill, exit=px, sd=sd, why=why,
                         gross_r=(px - fill) * side / sd))
        free = nxt
    return pd.DataFrame(rows)