sridattapradeep/India_Equity_Scanner
0
1"""2Unit tests for paper_trading.py. Hermetic — in-memory sqlite via a monkeypatched3AuthSession, synthetic SwingSetup-like rows and OHLC DataFrames, no network.4 5Run: pytest -q (from backend/, with the venv active)6"""7from datetime import datetime, timedelta, timezone8from types import SimpleNamespace9 10import pandas as pd11import pytest12from sqlalchemy import create_engine13from sqlalchemy.orm import sessionmaker14 15from auth import AuthBase16import paper_trading17from paper_trading import (18 PaperPosition,19 check_and_close_positions,20 close_momentum_positions,21 close_reversal_positions,22 get_paper_trading_summary,23 init_paper_trading_db,24 open_momentum_positions,25 open_new_positions,26 open_reversal_positions,27)28 29 30@pytest.fixture31def test_db(monkeypatch):32 engine = create_engine("sqlite:///:memory:")33 AuthBase.metadata.create_all(bind=engine)34 TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)35 36 monkeypatch.setattr("paper_trading.AuthSession", TestingSessionLocal)37 38 db = TestingSessionLocal()39 yield db40 db.close()41 42 43def test_init_paper_trading_db_survives_sqlite_alter_column_unsupported(monkeypatch):44 """SQLite doesn't support ALTER COLUMN ... DROP NOT NULL at all -- the45 migration's per-column try/except must swallow that cleanly rather than46 crashing app startup. (The actual bug this migration fixes -- stop_loss47 etc. still NOT NULL on a table that already existed in prod -- can only48 be exercised against real Postgres; this just proves init stays49 resilient regardless of what the dialect supports.)"""50 engine = create_engine("sqlite:///:memory:")51 monkeypatch.setattr(paper_trading, "auth_engine", engine)52 init_paper_trading_db() # must not raise53 init_paper_trading_db() # idempotent: safe to call again (e.g. a second boot)54 55 56def make_swing(symbol="TEST", bars_since_signal=0, entry=100.0, stop=90.0, t1=115.0, t2=125.0):57 return SimpleNamespace(58 symbol=symbol, bars_since_signal=bars_since_signal,59 entry_price=entry, stop_loss=stop, target_1=t1, target_2=t2,60 )61 62 63def make_bars(rows: list[dict], start="2026-01-01") -> pd.DataFrame:64 """rows: list of {open,high,low,close} dicts, one business day apart."""65 idx = pd.date_range(start, periods=len(rows), freq="B")66 return pd.DataFrame(rows, index=idx)67 68 69# ── get_paper_trading_summary (pre-existing coverage, kept) ────────────────70 71def test_get_paper_trading_summary(test_db):72 now = datetime.now(timezone.utc)73 pos1 = PaperPosition(74 symbol="WINNER", entry_date=now, entry_price=100.0, stop_loss=90.0, target_1=120.0, target_2=150.0, shares=100,75 status="closed", exit_date=now, exit_price=120.0, exit_reason="target_1", pnl_pct=20.0, r_multiple=2.076 )77 pos2 = PaperPosition(78 symbol="LOSER", entry_date=now, entry_price=100.0, stop_loss=90.0, target_1=120.0, target_2=150.0, shares=100,79 status="closed", exit_date=now, exit_price=90.0, exit_reason="stop", pnl_pct=-10.0, r_multiple=-1.080 )81 pos3 = PaperPosition(82 symbol="OPEN", entry_date=now, entry_price=100.0, stop_loss=90.0, target_1=120.0, target_2=150.0, shares=100,83 status="open"84 )85 test_db.add_all([pos1, pos2, pos3])86 test_db.commit()87 88 summary = get_paper_trading_summary()89 assert len(summary["open"]) == 190 assert summary["open"][0]["symbol"] == "OPEN"91 assert len(summary["closed"]) == 292 93 stats = summary["stats"]94 assert stats["total_trades"] == 295 assert stats["win_rate"] == 50.096 assert stats["profit_factor"] == 2.097 assert stats["avg_r"] == 0.598 assert len(summary["equity_curve"]) == 299 100 101# ── open_new_positions ──────────────────────────────────────────────────────102 103def test_open_new_positions_opens_fresh_signal(test_db):104 opened = open_new_positions(test_db, [make_swing("FRESH", bars_since_signal=0)])105 assert opened == 1106 rows = test_db.query(PaperPosition).all()107 assert len(rows) == 1108 assert rows[0].symbol == "FRESH"109 assert rows[0].status == "open"110 assert rows[0].shares == int(10_000 / (100.0 - 90.0)) # RISK_PER_TRADE_RS / risk_per_share111 112 113def test_open_new_positions_skips_stale_signal(test_db):114 opened = open_new_positions(test_db, [make_swing("OLD", bars_since_signal=5)])115 assert opened == 0116 assert test_db.query(PaperPosition).count() == 0117 118 119def test_open_new_positions_skips_symbol_already_open(test_db):120 test_db.add(PaperPosition(121 symbol="DUP", entry_date=datetime.now(timezone.utc), entry_price=100, stop_loss=90,122 target_1=110, target_2=120, shares=100, status="open",123 ))124 test_db.commit()125 opened = open_new_positions(test_db, [make_swing("DUP", bars_since_signal=0)])126 assert opened == 0127 assert test_db.query(PaperPosition).filter_by(symbol="DUP").count() == 1128 129 130def test_open_new_positions_skips_invalid_risk(test_db):131 # stop above entry -> non-positive risk per share, must not open132 opened = open_new_positions(test_db, [make_swing("BAD", entry=100.0, stop=105.0)])133 assert opened == 0134 135 136# ── check_and_close_positions ───────────────────────────────────────────────137 138def test_close_on_stop_hit(test_db):139 entry_date = datetime.now(timezone.utc) - timedelta(days=5)140 test_db.add(PaperPosition(141 symbol="STOPOUT", entry_date=entry_date, entry_price=100.0, stop_loss=90.0,142 target_1=115.0, target_2=125.0, shares=100, status="open",143 ))144 test_db.commit()145 146 df = make_bars([147 {"open": 99, "high": 101, "low": 98, "close": 100},148 {"open": 98, "high": 99, "low": 85, "close": 88}, # stop hit intraday149 ], start=(entry_date.date()))150 closed = check_and_close_positions(test_db, {"STOPOUT.NS": df})151 assert closed == 1152 pos = test_db.query(PaperPosition).filter_by(symbol="STOPOUT").one()153 assert pos.status == "closed"154 assert pos.exit_reason == "stop"155 assert pos.exit_price == 90.0156 assert pos.r_multiple == pytest.approx(-1.0)157 158 159def test_stop_wins_over_target_on_same_bar(test_db):160 """Same-bar conservatism: if a bar touches both stop and target, stop fills first."""161 entry_date = datetime.now(timezone.utc) - timedelta(days=5)162 test_db.add(PaperPosition(163 symbol="WHIPSAW", entry_date=entry_date, entry_price=100.0, stop_loss=90.0,164 target_1=115.0, target_2=125.0, shares=100, status="open",165 ))166 test_db.commit()167 168 df = make_bars([169 {"open": 100, "high": 101, "low": 99, "close": 100},170 {"open": 100, "high": 130, "low": 85, "close": 120}, # touches BOTH stop and targets171 ], start=(entry_date.date()))172 check_and_close_positions(test_db, {"WHIPSAW.NS": df})173 pos = test_db.query(PaperPosition).filter_by(symbol="WHIPSAW").one()174 assert pos.exit_reason == "stop"175 176 177def test_close_on_target_1_hit(test_db):178 entry_date = datetime.now(timezone.utc) - timedelta(days=5)179 test_db.add(PaperPosition(180 symbol="WINNER", entry_date=entry_date, entry_price=100.0, stop_loss=90.0,181 target_1=115.0, target_2=125.0, shares=100, status="open",182 ))183 test_db.commit()184 185 df = make_bars([186 {"open": 100, "high": 105, "low": 99, "close": 104},187 {"open": 105, "high": 118, "low": 104, "close": 116}, # target_1 hit, not target_2188 ], start=(entry_date.date()))189 check_and_close_positions(test_db, {"WINNER.NS": df})190 pos = test_db.query(PaperPosition).filter_by(symbol="WINNER").one()191 assert pos.exit_reason == "target_1"192 assert pos.exit_price == 115.0193 assert pos.r_multiple == pytest.approx(1.5) # (115-100)/(100-90)194 195 196def test_close_on_timeout(test_db):197 from backtest import SWING_MAX_HOLD198 entry_date = datetime.now(timezone.utc) - timedelta(days=SWING_MAX_HOLD + 10)199 test_db.add(PaperPosition(200 symbol="STALE", entry_date=entry_date, entry_price=100.0, stop_loss=90.0,201 target_1=200.0, target_2=250.0, shares=100, status="open", # unreachable targets202 ))203 test_db.commit()204 205 # Enough bars after entry to exceed SWING_MAX_HOLD, price drifting sideways206 # (never touches stop or the deliberately-unreachable targets).207 rows = [{"open": 100, "high": 102, "low": 98, "close": 100} for _ in range(SWING_MAX_HOLD + 5)]208 df = make_bars(rows, start=entry_date.date())209 closed = check_and_close_positions(test_db, {"STALE.NS": df})210 assert closed == 1211 pos = test_db.query(PaperPosition).filter_by(symbol="STALE").one()212 assert pos.exit_reason == "timeout"213 214 215def test_still_open_position_untouched(test_db):216 entry_date = datetime.now(timezone.utc) - timedelta(days=2)217 test_db.add(PaperPosition(218 symbol="HOLDING", entry_date=entry_date, entry_price=100.0, stop_loss=90.0,219 target_1=115.0, target_2=125.0, shares=100, status="open",220 ))221 test_db.commit()222 223 df = make_bars([224 {"open": 100, "high": 103, "low": 99, "close": 102},225 ], start=entry_date.date())226 closed = check_and_close_positions(test_db, {"HOLDING.NS": df})227 assert closed == 0228 pos = test_db.query(PaperPosition).filter_by(symbol="HOLDING").one()229 assert pos.status == "open"230 231 232def test_missing_symbol_in_universe_dfs_skipped_gracefully(test_db):233 entry_date = datetime.now(timezone.utc) - timedelta(days=2)234 test_db.add(PaperPosition(235 symbol="NODATA", entry_date=entry_date, entry_price=100.0, stop_loss=90.0,236 target_1=115.0, target_2=125.0, shares=100, status="open",237 ))238 test_db.commit()239 closed = check_and_close_positions(test_db, {})240 assert closed == 0241 242 243# ── strategy tagging + bracket close skips membership positions ────────────244 245def test_open_new_positions_tags_strategy(test_db):246 open_new_positions(test_db, [make_swing("SMCSIG")], strategy="smc")247 row = test_db.query(PaperPosition).filter_by(symbol="SMCSIG").one()248 assert row.strategy == "smc"249 250 251def test_open_new_positions_swing_smc_independent_open_symbol_sets(test_db):252 # A stock can be open under one strategy while a signal for the SAME253 # symbol under a DIFFERENT strategy also opens -- they don't collide.254 open_new_positions(test_db, [make_swing("DUAL")], strategy="swing")255 opened = open_new_positions(test_db, [make_swing("DUAL")], strategy="smc")256 assert opened == 1257 assert test_db.query(PaperPosition).filter_by(symbol="DUAL").count() == 2258 259 260def test_bracket_close_skips_membership_positions(test_db):261 entry_date = datetime.now(timezone.utc) - timedelta(days=2)262 test_db.add(PaperPosition(263 symbol="MOMPOS", strategy="momentum", entry_date=entry_date,264 entry_price=100.0, shares=50, status="open", # stop_loss=None265 ))266 test_db.commit()267 df = make_bars([{"open": 100, "high": 200, "low": 1, "close": 150}], start=entry_date.date())268 closed = check_and_close_positions(test_db, {"MOMPOS.NS": df})269 assert closed == 0 # would have "stopped out" at low=1 if not skipped270 pos = test_db.query(PaperPosition).filter_by(symbol="MOMPOS").one()271 assert pos.status == "open"272 273 274# ── momentum: membership-based open/close ───────────────────────────────────275 276def make_momentum_result(symbol="MOMSTOCK", close_price=100.0):277 # `.close`, NOT `.close_price` -- these are scanner.MomentumData objects278 # (main.py's scan_momentum `momentum_results` list), not MomentumResult279 # ORM rows. A `.close_price` typo here is exactly what let a real bug280 # ship silently in production (open_momentum_positions crashed on every281 # single call, swallowed by main.py's non-fatal try/except).282 # Deliberately NOT importing the real scanner.MomentumData here: scanner283 # module-level code runs a live NSE fetch on import (NIFTY_500_UNIVERSE284 # = _load_universe()), which turned this file from a ~1s hermetic suite285 # into a 13-minute network-dependent one the one time it was tried. Keep286 # this field name in sync with scanner.py's MomentumData by hand instead.287 return SimpleNamespace(symbol=symbol, close=close_price)288 289 290def test_open_momentum_positions_only_on_new_entrant(test_db):291 opened = open_momentum_positions(test_db, [make_momentum_result("NEW")], prior_passing_symbols=set())292 assert opened == 1293 row = test_db.query(PaperPosition).filter_by(symbol="NEW").one()294 assert row.strategy == "momentum"295 assert row.stop_loss is None296 297 # already passing last scan -- not a new entry event298 opened2 = open_momentum_positions(299 test_db, [make_momentum_result("OLD")], prior_passing_symbols={"OLD"}300 )301 assert opened2 == 0302 303 304def test_close_momentum_positions_on_template_exit(test_db):305 open_momentum_positions(test_db, [make_momentum_result("DROPOUT", 100.0)], prior_passing_symbols=set())306 closed = close_momentum_positions(test_db, current_passing_symbols=set(), price_by_symbol={"DROPOUT": 110.0})307 assert closed == 1308 pos = test_db.query(PaperPosition).filter_by(symbol="DROPOUT").one()309 assert pos.exit_reason == "template_exit"310 assert pos.exit_price == 110.0311 assert pos.pnl_pct == pytest.approx(10.0)312 313 314def test_momentum_position_stays_open_while_still_passing(test_db):315 open_momentum_positions(test_db, [make_momentum_result("HOLDING")], prior_passing_symbols=set())316 closed = close_momentum_positions(test_db, current_passing_symbols={"HOLDING"}, price_by_symbol={})317 assert closed == 0318 319 320# ── reversal: fixed-horizon open/close ──────────────────────────────────────321 322def make_reversal_candidate(symbol="REVSTOCK", close=100.0, days_above_200=15):323 return SimpleNamespace(symbol=symbol, close=close, days_above_200=days_above_200)324 325 326def test_open_reversal_positions(test_db):327 opened = open_reversal_positions(test_db, [make_reversal_candidate("REV1")])328 assert opened == 1329 row = test_db.query(PaperPosition).filter_by(symbol="REV1").one()330 assert row.strategy == "reversal"331 assert row.stop_loss is None332 333 # already open -- second appearance on the watchlist doesn't re-open334 opened2 = open_reversal_positions(test_db, [make_reversal_candidate("REV1")])335 assert opened2 == 0336 337 338def test_close_reversal_positions_waits_for_full_horizon(test_db):339 entry_date = datetime.now(timezone.utc) - timedelta(days=30)340 test_db.add(PaperPosition(341 symbol="REVHOLD", strategy="reversal", entry_date=entry_date,342 entry_price=100.0, shares=100, status="open",343 ))344 test_db.commit()345 346 # Only 5 bars after entry -- short of the 20-day horizon.347 rows = [{"open": 100, "high": 102, "low": 98, "close": 100 + i} for i in range(5)]348 df = make_bars(rows, start=entry_date.date())349 closed = close_reversal_positions(test_db, {"REVHOLD.NS": df}, hold_days=20)350 assert closed == 0351 352 353def test_close_reversal_positions_at_horizon(test_db):354 entry_date = datetime.now(timezone.utc) - timedelta(days=30)355 test_db.add(PaperPosition(356 symbol="REVDONE", strategy="reversal", entry_date=entry_date,357 entry_price=100.0, shares=100, status="open",358 ))359 test_db.commit()360 361 rows = [{"open": 100, "high": 102, "low": 98, "close": 100 + i} for i in range(25)]362 df = make_bars(rows, start=entry_date.date())363 closed = close_reversal_positions(test_db, {"REVDONE.NS": df}, hold_days=20)364 assert closed == 1365 pos = test_db.query(PaperPosition).filter_by(symbol="REVDONE").one()366 assert pos.exit_reason == "horizon_20d"367 # entry_date carries a time-of-day component (datetime.now() - 30 days),368 # so the midnight bar on the entry date itself is chronologically BEFORE369 # entry and correctly excluded -- bars_after_entry starts at rows[1], so370 # the 20th bar after entry (iloc[19]) is rows[20], close = 100 + 20 = 120.371 assert pos.exit_price == 120.0372 373 374# ── reversal: batched commits + resilience to a bad row ────────────────────375 376def test_open_reversal_positions_batches_large_watchlist(test_db):377 # Confirmed live: one giant multi-hundred-row INSERT failed against the378 # pooler. This proves batching still gets everyone opened, just in379 # multiple commits instead of one.380 candidates = [make_reversal_candidate(f"REV{i}", close=100.0 + i) for i in range(60)]381 opened = open_reversal_positions(test_db, candidates)382 assert opened == 60383 assert test_db.query(PaperPosition).filter_by(strategy="reversal", status="open").count() == 60384 385 386def test_open_reversal_positions_nan_close_skipped_not_crashed(test_db):387 # A NaN close price compares False against `<= 0` in Python (NaN388 # comparisons are always False), so a naive guard would let it through389 # and crash `int()` downstream instead of skipping it cleanly.390 candidates = [391 make_reversal_candidate("GOOD", close=100.0),392 make_reversal_candidate("BADNAN", close=float("nan")),393 ]394 opened = open_reversal_positions(test_db, candidates)395 assert opened == 1396 assert test_db.query(PaperPosition).filter_by(symbol="GOOD").count() == 1397 assert test_db.query(PaperPosition).filter_by(symbol="BADNAN").count() == 0398 399 400def test_open_reversal_positions_one_bad_batch_does_not_block_others(test_db, monkeypatch):401 # Simulate the real failure mode: one batch's commit() raises (pooler402 # rejects it) -- the other batches must still go through, and the403 # session must still be usable afterward (rollback happened).404 candidates = [make_reversal_candidate(f"BATCH{i}", close=100.0 + i) for i in range(75)]405 406 real_commit = test_db.commit407 calls = {"n": 0}408 409 def flaky_commit():410 calls["n"] += 1411 if calls["n"] == 2: # fail exactly the second batch412 test_db.rollback()413 raise RuntimeError("simulated pooler rejection")414 real_commit()415 416 monkeypatch.setattr(test_db, "commit", flaky_commit)417 opened = open_reversal_positions(test_db, candidates)418 419 # 3 batches of 25; batch 2 failed -> 50 opened, not 75.420 assert opened == 50421 # Session must still be usable after the failed batch (no cascade).422 assert test_db.query(PaperPosition).filter_by(strategy="reversal", status="open").count() == 50423 424 425# ── summary: per-strategy breakdown ─────────────────────────────────────────426 427def test_summary_by_strategy_breakdown(test_db):428 now = datetime.now(timezone.utc)429 test_db.add_all([430 PaperPosition(symbol="A", strategy="swing", entry_date=now, entry_price=100, stop_loss=90,431 shares=10, status="closed", exit_date=now, exit_price=110, r_multiple=1.0, pnl_pct=10.0),432 PaperPosition(symbol="B", strategy="momentum", entry_date=now, entry_price=100,433 shares=10, status="closed", exit_date=now, exit_price=90, pnl_pct=-10.0),434 ])435 test_db.commit()436 437 summary = get_paper_trading_summary()438 assert "swing" in summary["by_strategy"]439 assert "momentum" in summary["by_strategy"]440 assert summary["by_strategy"]["swing"]["total_trades"] == 1441 assert summary["by_strategy"]["momentum"]["total_trades"] == 1442 # momentum's win/loss falls back to pnl_pct sign since r_multiple is None443 assert summary["by_strategy"]["momentum"]["win_rate"] == 0.0444 assert summary["by_strategy"]["momentum"]["avg_r"] is None445 # equity curve only includes bracket trades with a real r_multiple446 assert len(summary["equity_curve"]) == 1447 448 449def test_strategy_filter_query_param(test_db):450 now = datetime.now(timezone.utc)451 test_db.add_all([452 PaperPosition(symbol="A", strategy="swing", entry_date=now, entry_price=100, stop_loss=90,453 shares=10, status="open"),454 PaperPosition(symbol="B", strategy="momentum", entry_date=now, entry_price=100,455 shares=10, status="open"),456 ])457 test_db.commit()458 459 summary = get_paper_trading_summary(strategy="swing")460 assert len(summary["open"]) == 1461 assert summary["open"][0]["symbol"] == "A"462