Sumitx369/Paper_Trading_Backtesting
2
1"""Tests for the Quant-Core decision engine (S1..S9 -> Bayes -> gate -> sim)."""2from __future__ import annotations3 4from datetime import datetime, time, timedelta5 6import pandas as pd7import pytest8 9from app.services.engine import bayes, decision, ledger, learning, portfolio, simulator10from app.services.engine.contracts import (DecisionRecord, MarketRegime,11 ProbabilityBreakdown, SetupTrigger)12from app.services.engine.indicators import IntradayIndicators13from app.services.engine.setups import BarCtx, SetupConfig, s114 15 16def _frame(rows, day=2):17 idx = [datetime(2026, 6, day, 9, 15) + timedelta(minutes=5 * i) for i in range(len(rows))]18 return pd.DataFrame(rows, index=idx, columns=["open", "high", "low", "close", "volume"])19 20 21# ----------------------------- S1 golden file ------------------------------22def test_s1_fires_per_spec():23 # Rising green bars so EMAs sit below price by the decision candle (index 2).24 rows = [[100, 100.5, 99.8, 100.4, 1000],25 [100.4, 101.0, 100.3, 100.9, 1000],26 [101.0, 101.8, 100.95, 101.7, 1500]] # decision candle: green, new high, tight stop27 df = _frame(rows)28 ind = IntradayIndicators.compute(df)29 cfg = SetupConfig(first_candle_index=2)30 ctx = BarCtx(bars=df, ind=ind, pdh=100.6, pdl=99.0, prev_close=100.0, i=2, cfg=cfg)31 t = s1(ctx, "TEST.NS")32 assert t.fired and t.side == "LONG" and t.entry_type == "BUY_STOP"33 assert t.entry_price == pytest.approx(101.8 + 0.05, abs=1e-6) # day-high + tick34 assert t.stop_price == pytest.approx(100.95 - 0.05, abs=1e-6) # signal low - tick35 assert t.valid_until == time(10, 0)36 assert all(t.conditions.values())37 38 39def test_s1_sl_too_wide_invalidates():40 # Huge candle range -> stop >1% from entry -> SL_TOO_WIDE, not fired.41 rows = [[100, 100.5, 99.8, 100.4, 1000],42 [100.4, 101, 100.3, 100.9, 1000],43 [100.9, 105.0, 98.0, 104.0, 1500]]44 df = _frame(rows)45 ind = IntradayIndicators.compute(df)46 cfg = SetupConfig(first_candle_index=2)47 ctx = BarCtx(bars=df, ind=ind, pdh=100.6, pdl=99, prev_close=100, i=2, cfg=cfg)48 t = s1(ctx, "TEST.NS")49 assert not t.fired and t.invalidation == "SL_TOO_WIDE"50 51 52# ----------------------------- Bayesian math -------------------------------53def test_bayes_odds_math():54 model = bayes.BayesModel() # fresh seed, prior S1 = 55/4555 conds = {"above_pdh": True, "above_both_emas": True, "green": True,56 "rel_volume_gt_1_2": False}57 out = bayes.score("S1", conds, model=model)58 # manual odds: prior 56/102... compute from model59 prior = model.prior_winrate("S1")60 odds = prior / (1 - prior)61 for c in ("above_pdh", "above_both_emas", "green"):62 odds *= model.lr("S1", c)63 expect = odds / (1 + odds)64 assert out["p_bayes"] == pytest.approx(expect, abs=1e-4)65 assert out["p_bayes"] > prior # bullish conditions lift the probability66 67 68# ----------------------------- sizing --------------------------------------69def test_sizing_under_risk_cap():70 sd = portfolio.size(100_000, 10_000, entry=100.0, stop=98.0, size_band="FULL")71 assert sd.qty > 0 and sd.risk_rupees <= 10_000 + 1e-672 half = portfolio.size(100_000, 10_000, 100.0, 98.0, "HALF")73 assert half.qty <= sd.qty74 75 76# ----------------------------- simulator -----------------------------------77def test_simulator_long_fill_and_squareoff():78 trig = SetupTrigger(symbol="X", setup_id="S1", fired=True, side="LONG",79 entry_type="BUY_STOP", entry_price=101.0, stop_price=99.0,80 valid_until=time(15, 0))81 bars = _frame([[100, 100.5, 99.5, 100.2, 1000], # trigger bar context (after)82 [100.3, 101.5, 100.2, 101.2, 1000], # high>=101 -> fill83 [101.2, 102.0, 101.0, 101.8, 1000]]) # square-off at close 101.884 pos = simulator.simulate(trig, qty=100, bars_after=bars, fee_bps=0, slippage_bps=0)85 assert pos.exit_reason == "square_off"86 assert pos.entry_price == pytest.approx(101.0)87 assert pos.pnl == pytest.approx((101.8 - 101.0) * 100, abs=1e-6)88 89 90def test_simulator_no_fill_when_untriggered():91 trig = SetupTrigger(symbol="X", setup_id="S1", fired=True, side="LONG",92 entry_type="BUY_STOP", entry_price=200.0, stop_price=198.0,93 valid_until=time(15, 0))94 bars = _frame([[100, 101, 99, 100.5, 1000], [100.5, 101, 100, 100.8, 1000]])95 pos = simulator.simulate(trig, 100, bars)96 assert pos.exit_reason == "no_fill" and pos.qty == 100 and pos.pnl == 0.097 98 99# ------------------- risk floors: min-SL widen + trailing profit-lock -------100def _long_trig(entry=100.0, stop=99.5):101 return SetupTrigger(symbol="X", setup_id="S1", fired=True, side="LONG",102 entry_type="BUY_STOP", entry_price=entry, stop_price=stop,103 valid_until=time(15, 0))104 105 106def test_min_sl_floor_widens_a_tight_stop():107 # 0.5% setup stop, but the user/agent floor is 2% -> stop widens to 98.108 bars = _frame([[99.8, 100.2, 99.7, 100.0, 1000],109 [100.0, 101.0, 99.9, 100.5, 1000],110 [100.5, 101.0, 100.4, 100.8, 1000]])111 pos = simulator.simulate(_long_trig(), 100, bars, fee_bps=0, slippage_bps=0,112 min_sl_pct=2.0, min_profit_pct=3.0, trail=True)113 assert pos.stop_price == pytest.approx(98.0, abs=1e-6)114 115 116def test_trailing_stop_banks_at_least_min_profit():117 # Peak 106 then pulls back: trailing locks >= min_profit_pct (3%).118 bars = _frame([[99.8, 100.2, 99.7, 100.0, 1000],119 [100.0, 103.0, 100.0, 102.8, 1000], # below activation (105)120 [102.8, 106.0, 102.5, 105.5, 1000], # peak 106 -> stop trails to 104121 [105.5, 105.6, 103.0, 103.2, 1000]]) # low 103 <= 104 -> trailing exit122 pos = simulator.simulate(_long_trig(), 100, bars, fee_bps=0, slippage_bps=0,123 min_sl_pct=2.0, min_profit_pct=3.0, trail=True)124 assert pos.exit_reason == "trailing_stop"125 assert pos.return_pct >= 3.0 # the profit floor is honoured126 assert pos.pnl == pytest.approx(400.0, abs=1e-6)127 assert pos.stop_price > 98.0 # the RATCHETED stop is recorded, not the floor128 129 130def test_profit_floor_holds_net_of_fees_and_slippage():131 # With real fees+slippage, a trailed exit must STILL net >= min_profit_pct132 # (the lock target is padded by the round-trip cost).133 bars = _frame([[99.8, 100.2, 99.7, 100.0, 1000],134 [100.0, 103.0, 100.0, 102.8, 1000],135 [102.8, 107.0, 102.5, 106.5, 1000],136 [106.5, 106.6, 103.0, 103.2, 1000]])137 pos = simulator.simulate(_long_trig(), 100, bars, fee_bps=5, slippage_bps=5,138 min_sl_pct=2.0, min_profit_pct=2.0, trail=True)139 assert pos.exit_reason == "trailing_stop"140 assert pos.return_pct >= 2.0 # floor holds AFTER costs, not just gross141 142 143def test_min_sl_floor_keeps_an_already_wider_stop():144 # A stop already 5% from entry must NOT be tightened to the 2% floor145 # (the floor is a minimum distance, never a cap).146 pos = simulator.simulate(_long_trig(entry=100.0, stop=95.0), 100,147 _frame([[99.8, 100.2, 99.7, 100.0, 1000],148 [100.0, 101.0, 99.9, 100.5, 1000],149 [100.5, 101.0, 100.4, 100.8, 1000]]),150 fee_bps=0, slippage_bps=0, min_sl_pct=2.0,151 min_profit_pct=3.0, trail=True)152 assert pos.stop_price == pytest.approx(95.0)153 154 155def test_below_activation_keeps_floored_stop_and_squares_off():156 # Never clears activation (105) -> stop stays at the 2% floor, EOD square-off.157 bars = _frame([[99.8, 100.2, 99.7, 100.0, 1000],158 [100.0, 104.0, 100.0, 103.8, 1000],159 [103.8, 104.2, 103.5, 104.0, 1000]])160 pos = simulator.simulate(_long_trig(), 100, bars, fee_bps=0, slippage_bps=0,161 min_sl_pct=2.0, min_profit_pct=3.0, trail=True)162 assert pos.exit_reason == "square_off" and pos.stop_price == pytest.approx(98.0)163 164 165def test_risk_floors_default_off_is_legacy():166 # No risk params -> identical to before (static setup stop, square-off).167 bars = _frame([[99.8, 100.2, 99.7, 100.0, 1000],168 [100.0, 101.0, 99.9, 100.5, 1000],169 [100.5, 101.0, 100.4, 100.8, 1000]])170 pos = simulator.simulate(_long_trig(), 100, bars, fee_bps=0, slippage_bps=0)171 assert pos.exit_reason == "square_off" and pos.stop_price == pytest.approx(99.5)172 173 174# ----------------------------- decision gate -------------------------------175def _regime(active=("S1",)):176 return MarketRegime(state="TRENDING", active_setups=list(active),177 threshold_delta={s: 0.0 for s in active}, risk_scalar=1.0)178 179 180def _trigger():181 return SetupTrigger(symbol="X", setup_id="S1", fired=True, side="LONG",182 entry_type="BUY_STOP", entry_price=101.0, stop_price=100.0)183 184 185def test_gate_buys_when_prob_clears_threshold():186 prob = ProbabilityBreakdown(p_final=0.80, p_raw=0.80, p_bayes=0.80)187 d = decision.decide(_trigger(), prob, _regime())188 assert d.action == "BUY" and d.size_band in ("FULL", "HALF")189 190 191def test_gate_skips_when_prob_below_threshold():192 prob = ProbabilityBreakdown(p_final=0.40, p_raw=0.40, p_bayes=0.40)193 d = decision.decide(_trigger(), prob, _regime())194 assert d.action == "SKIP" and "BELOW_THRESHOLD" in d.reason_codes195 196 197def test_regime_no_longer_hard_blocks_setup():198 # Regime no longer disables setups — every fired signal is judged on its199 # probability; a high-probability signal trades regardless of active_setups.200 prob = ProbabilityBreakdown(p_final=0.90, p_raw=0.9, p_bayes=0.9)201 d = decision.decide(_trigger(), prob, _regime(active=("S2",)))202 assert d.action == "BUY"203 204 205# ----------------------------- ledger + learning ---------------------------206def test_ledger_hash_chain_and_tamper():207 from app.database import SessionLocal208 from app.models import DecisionLog209 key = "test_ledger_user"210 ledger.clear(key)211 for i in range(3):212 ledger.append(key, DecisionRecord(record_id=0, ts=datetime.now(), symbol="X",213 setup_id="S1"))214 assert ledger.verify(key)["ok"] is True215 # tamper: edit a committed row's payload directly in the DB216 with SessionLocal() as db:217 row = db.query(DecisionLog).filter_by(user_key=key, seq=2).one()218 p = dict(row.payload); p["symbol"] = "HACKED"; row.payload = p219 db.commit()220 assert ledger.verify(key)["ok"] is False221 ledger.clear(key)222 223 224def test_verify_ok_after_head_purge():225 """A retention purge that removes the GENESIS row must NOT look like tampering."""226 from app.database import SessionLocal227 from app.models import DecisionLog228 key = "test_purge_user"229 ledger.clear(key)230 for _ in range(4):231 ledger.append(key, DecisionRecord(record_id=0, ts=datetime.now(), symbol="X", setup_id="S1"))232 # delete the oldest row (seq=1) as a retention purge would233 with SessionLocal() as db:234 db.query(DecisionLog).filter_by(user_key=key, seq=1).delete()235 db.commit()236 assert ledger.verify(key)["ok"] is True # chain still valid from the new head237 ledger.clear(key)238 239 240def test_eod_training_updates_priors():241 key = "test_train_user"242 ledger.clear(key)243 bayes.save_model(bayes.BayesModel()) # reset to seed for an isolated assertion244 # two resolved wins on S3245 for _ in range(2):246 rec = DecisionRecord(record_id=0, ts=datetime.now(), symbol="X", setup_id="S3",247 probability={"conditions": {"above_pdh": True, "green": True}},248 fill={"is_open": False, "exit_reason": "square_off", "pnl": 500})249 ledger.append(key, rec)250 before = bayes.load_model().prior_winrate("S3")251 report = learning.train(key)252 after = bayes.load_model().prior_winrate("S3")253 assert report["samples"] == 2 and report["wins"] == 2254 assert after > before # wins pushed the posterior up (idempotent: seed + counts)255 ledger.clear(key)256 bayes.save_model(bayes.BayesModel()) # leave the model at seed257