File size: 6,068 Bytes
ef20ebe | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 | """
UI-level integration tests (audit items 17 & 20): proves values entered
in the Backtest tab's Advanced panel reach WalkForwardBacktester/
RiskManagement completely unchanged, all the way from
app.run_backtest()'s own arguments through to what the engine reports
having actually used. Also proves the specific invariant the root-cause
audit demanded: changing SL/TP changes ONLY exit behavior (exit_reason,
net_pnl) and never the set of candidate signals -- same eligible
origins, same model outputs, same entry signals, same number of
candidate trades, checked as an actual equality across three real
app.run_backtest() calls rather than assumed.
Requires gradio (app.py's only import these tests need that the rest of
the suite doesn't) -- gracefully reports itself as trivially passed
rather than failing the whole suite in a sandbox without it, matching
this project's existing "IMPLEMENTED, not runtime-verified without X"
pattern elsewhere (see README). The engine-level logic these tests
exercise is ALSO covered directly in test_backtest_no_lookahead.py
without needing gradio at all -- this file only adds the "did the UI
plumbing forward it correctly" layer on top.
"""
import numpy as np
try:
import app
_GRADIO_AVAILABLE = True
except ImportError:
_GRADIO_AVAILABLE = False
from fixtures import make_synthetic_ohlcv
from moirai_model import QuantileForecast
class _PureFunctionStub:
"""Behavior is a pure function of the input context's last close
value, never of call count/order/instance state. A stateful stub
(e.g. an internal call counter) would silently answer differently
on a second run against identical data -- exactly the kind of
self-inflicted test bug that could manufacture a fake "SL/TP
changed my signals" result. This class structurally cannot do that:
two separate instances given the same data always agree, and a
single instance called twice on the same origin always agrees too.
"""
def __init__(self, signal_close_value):
self.signal_close_value = signal_close_value
def predict(self, context_df, target_col, past_covariate_cols, prediction_length, freq):
last_close = float(context_df[target_col].iloc[-1])
if abs(last_close - self.signal_close_value) < 1e-6:
prices = [last_close * f for f in (1.0198, 1.0199, 1.02, 1.0201, 1.0202)]
else:
prices = [last_close * f for f in (0.9997, 0.9999, 1.0000, 1.0001, 1.0003)]
return QuantileForecast(horizon=prediction_length, quantile_levels=[0.1, 0.25, 0.5, 0.75, 0.9],
quantiles=np.array([prices]), samples=None,
context_length_used=len(context_df), prediction_length=prediction_length)
def _run_three_sl_tp_settings():
df = make_synthetic_ohlcv(n=600, seed=61, freq="1min")
signal_value = float(df["close"].iloc[400])
def fake_get_ohlcv(symbol, requested_tf, history_label):
class Report:
def __init__(self):
self.__dict__ = dict(rows_in=len(df), rows_out=len(df), duplicates_removed=0,
invalid_ohlc_removed=0, nan_rows_removed=0,
timezone="UTC", monotonic=True, warnings=[])
return df, "native", requested_tf, Report()
app._get_ohlcv = fake_get_ohlcv
app._model_status["moirai"]["loaded"] = True
settings = [(0.0, 0.0), (0.05, 0.10), (0.5, 0.10)]
outcomes = []
for sl, tp in settings:
app._forecasters["moirai"] = _PureFunctionStub(signal_value) # fresh instance, no shared state across runs
md, trades_df, diag = app.run_backtest("EURUSD=X", "1m", 5, "4 days", "forex",
context_length=64, stop_loss_pct=sl, take_profit_pct=tp)
outcomes.append((sl, tp, md, trades_df, diag))
return outcomes
def test_sl_tp_zero_means_off_reaches_backend_as_none():
if not _GRADIO_AVAILABLE:
print("SKIP (gradio not installed in this environment) -- see README")
return
outcomes = _run_three_sl_tp_settings()
sl, tp, md, trades_df, diag = outcomes[0]
assert "backend received**: SL=off, TP=off" in md, md
def test_nonzero_sl_tp_reach_backend_unchanged():
if not _GRADIO_AVAILABLE:
print("SKIP (gradio not installed in this environment) -- see README")
return
outcomes = _run_three_sl_tp_settings()
_, _, md1, _, _ = outcomes[1]
_, _, md2, _, _ = outcomes[2]
assert "requested**: SL=0.05%, TP=0.10%" in md1 and "backend received**: SL=0.05%, TP=0.10%" in md1, md1
assert "requested**: SL=0.50%, TP=0.10%" in md2 and "backend received**: SL=0.50%, TP=0.10%" in md2, md2
def test_changing_sl_tp_changes_only_exit_behavior_not_candidate_signals():
"""The core A/B/A/B invariant the audit demanded: same eligible
origins, same model outputs, same entry signals, same number of
candidate trades across all three SL/TP settings -- checked as
actual cross-run equality, not assumed."""
if not _GRADIO_AVAILABLE:
print("SKIP (gradio not installed in this environment) -- see README")
return
outcomes = _run_three_sl_tp_settings()
diags = [o[4] for o in outcomes]
invariant_keys = ["num_origins", "signals_buy", "signals_sell", "hold_wide_interval",
"hold_no_edge", "hold_low_confidence_buy", "hold_low_confidence_sell",
"model_errors", "trades_opened_buy", "trades_opened_sell"]
for key in invariant_keys:
values = [d[key] for d in diags]
assert len(set(values)) == 1, (
f"{key} differs across SL/TP settings ({values}) -- changing "
f"SL/TP must never change which signals fire, only how they exit"
)
# Exit behavior IS allowed (expected) to differ:
exit_reasons = [(d["exit_horizon"], d["exit_stop_loss"], d["exit_take_profit"]) for d in diags]
assert len(set(exit_reasons)) > 1, "test setup should make at least one setting's exit reason differ"
|