| """ |
| 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) |
| 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_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" |
|
|
