signal-trader/tests/check_reference.py

# SPDX-License-Identifier: LicenseRef-PolyForm-Noncommercial-1.0.0 (see LICENSE)
"""Self-test of tests/reference.py on synthetic bars (no broker data needed).

    python3 tests/check_reference.py [seeds]        default 20 seeds, 3 years of H4 bars each

For every random walk and each source (UT Bot, SuperTrend, Range Filter) it checks: the last bar never carries a signal (it
is still forming); SuperTrend signals alternate buy, sell, buy; every trade enters at the open of the bar after a signal bar;
the stop distance is stop x ATR(14) of the signal bar; trades never overlap; a reversal exits and re-enters at the same open on
the other side; a stop exit is on the bar where the extreme touches the stop."""
import sys
from pathlib import Path

import numpy as np
import pandas as pd

sys.path.insert(0, str(Path(__file__).parent))
from reference import signals, run, atr14  # noqa: E402

SOURCES = {"ut": dict(key=2.0, atr=10, ema=0), "st": dict(atr=10, mult=2.0, ema=0), "rf": dict(period=100, mult=3.0)}


def synth(seed, years=3):
    rng = np.random.default_rng(seed)
    idx = pd.date_range("2021-01-01", periods=years * 365 * 6, freq="4h")
    idx = idx[idx.dayofweek < 5]
    drift = 0.15 * np.sin(np.arange(len(idx)) / 400.0)
    close = 1800 + np.cumsum(rng.normal(drift, 6.0))
    open_ = np.concatenate(([1800.0], close[:-1]))
    span = np.abs(rng.normal(0, 3.0, len(idx)))
    return pd.DataFrame({"open": open_, "high": np.maximum(open_, close) + span, "low": np.minimum(open_, close) - span,
                         "close": close, "spread": rng.choice([0.2, 0.3, 0.5], len(idx))}, index=idx)


def main(seeds):
    problems, n_sig, n_tr = [], 0, 0
    for seed in range(seeds):
        b = synth(seed)
        o, h, l, c = (b[k].to_numpy(float) for k in ("open", "high", "low", "close"))
        atr = atr14(h, l, c)
        for src, p in SOURCES.items():
            sig = signals(src, p, o, h, l, c)
            n_sig += int((sig != 0).sum())
            if sig[-1] != 0:
                problems.append(("signal on the last (forming) bar", src, seed))
            if src == "st":
                s = sig[sig != 0]
                if len(s) > 1 and (s[1:] == s[:-1]).any():
                    problems.append(("SuperTrend signals do not alternate", src, seed))
            tr = run(b, sig, 3.0)
            n_tr += len(tr)
            pos_idx = {t: k for k, t in enumerate(b.index)}
            last_exit = None
            for i, r in enumerate(tr.itertuples()):
                k = pos_idx[r.entry_bar]
                if sig[k - 1] != r.side:
                    problems.append(("entry not on the bar after a same-side signal", src, seed, str(r.entry_bar)))
                if abs(r.dist - 3.0 * atr[k - 1]) > 1e-9:
                    problems.append(("stop distance is not 3 x ATR(14) of the signal bar", src, seed, str(r.entry_bar)))
                if last_exit is not None and r.entry_bar < last_exit:
                    problems.append(("overlap", src, seed, str(r.entry_bar)))
                if i > 0 and tr.iloc[i - 1].why == "reverse":
                    q = tr.iloc[i - 1]
                    if not (q.exit_bar == r.entry_bar and q.side == -r.side):
                        problems.append(("reversal not at the same open", src, seed, str(r.entry_bar)))
                if r.why == "stop":
                    j = pos_idx[r.exit_bar]
                    touched = l[j] <= r.exit if r.side > 0 else h[j] + b.spread.iloc[j] >= r.exit
                    if not touched:
                        problems.append(("stop exit without a touch", src, seed, str(r.exit_bar)))
                last_exit = r.exit_bar
    print(f"seeds: {seeds}, signals: {n_sig}, trades: {n_tr}")
    print(f"violations: {len(problems)}")
    for p in problems[:10]:
        print("  ", p)
    print("RESULT:", "PASS" if not problems else "FAIL")
    return 0 if not problems else 1


if __name__ == "__main__":
    sys.exit(main(int(sys.argv[1]) if len(sys.argv) > 1 else 20))