"""His VWAP + pivot + Stoch RSI rules on hand-built sessions, his own Bajaj Finance days, then the whole history against the audit engine.""" import json import sys from dataclasses import replace from pathlib import Path import numpy as np import pandas as pd import pytest from backtest.data import Candles, Session, load_nse_stock, session_bars from backtest.indicators import session_vwap from backtest.strategies import vwap_pivot as vp from backtest.strategies.vwap_pivot import EXACT, LONG, NO_COSTS, SHORT, EquityCosts, Indicators, trade_day ROOT = Path(__file__).resolve().parents[1] RAW = ROOT / "projects/boomingbulls-audit/data_raw" RULES_JSON = ROOT / "projects/boomingbulls-audit/bb_rules.json" LEVELS = (900.0, 940.0, 960.0, 1000.0, 1040.0, 1060.0, 1100.0) # S3, S2, S1, P, R1, R2, R3 QUIET = (1035.0, 1036.0, 1034.0, 1035.0) def nse_day(day: str, start: str, rows: list[tuple[float, float, float, float]]) -> Session: o, h, lo, c = (np.array(col, dtype=float) for col in zip(*rows, strict=True)) minute = pd.date_range(f"{day} {start}", periods=len(rows), freq="min") clock = (minute.hour * 100 + minute.minute).to_numpy() candles = Candles(o, h, lo, c) return Session(pd.Timestamp(day), minute.to_numpy(), clock, candles, candles, np.full(len(rows), 100.0)) def candle(o: float, h: float, lo: float, c: float) -> list[tuple[float, float, float, float]]: return [(o, h, lo, o)] * 4 + [(o, h, lo, c)] YESTERDAY = nse_day("2020-07-24", "15:25", [QUIET] * 5) WARM_UP = candle(1035.0, 1036.0, 1034.0, 1035.0) # the 10:40 candle: inside the levels, before 10:45 BREAK_R1 = candle(1036.0, 1043.0, 1035.0, 1042.0) # the 10:45 candle closes through R1 (1040) def run(rows, *, vwap=1030.0, k_above_d=True, rules=EXACT, costs=NO_COSTS, start="10:40"): """Today's minutes from `start`; every 5-minute candle sees the same VWAP and Stoch RSI lines.""" today = nse_day("2020-07-27", start, rows) bars = session_bars([YESTERDAY, today], 5) n = len(bars) k, d = (np.full(n, 60.0), np.full(n, 50.0)) if k_above_d else (np.full(n, 50.0), np.full(n, 60.0)) return trade_day(today, range(1, n), bars, Indicators(np.full(n, vwap), k, d), LEVELS, rules, costs) def test_buys_the_close_through_a_pivot_above_vwap_with_stochastic_support(): [(trade, r)] = run(WARM_UP + BREAK_R1 + [(1042.0, 1052.5, 1041.0, 1052.0)]) assert (trade.side, trade.level, trade.entry_minute, trade.entry_price) == (LONG, 1040.0, 9, 1042.0) assert trade.stop == pytest.approx(1042 * 0.995) # "15 rupees stop" on ~3,000: 0.5% assert trade.target == pytest.approx(1042 * 1.01) # "30 rupees target": 1% assert (trade.exit_minute, trade.reason) == (10, "target") assert r == pytest.approx(2.0) def test_vwap_decides_the_side_and_the_stochastic_must_agree(): assert run(WARM_UP + BREAK_R1, vwap=1045.0) == [] # below VWAP: "plan to sell", and nothing broke down assert run(WARM_UP + BREAK_R1, k_above_d=False) == [] # blue line under the red one assert len(run(WARM_UP + BREAK_R1 + [QUIET], k_above_d=False, rules=vp.NO_STOCHASTIC)) == 1 def test_nothing_before_10_45(): assert run(BREAK_R1 + [QUIET] * 5, start="10:35") == [] # the break comes in the 10:40 candle def test_a_new_high_of_the_day_is_a_trigger_too(): day = WARM_UP + candle(1035.0, 1038.0, 1035.0, 1037.0) + [QUIET] [(trade, _)] = run(day) assert trade.level == 1036.0 # the 10:45 candle closed above the morning's high (1036), no pivot in between def test_stop_trading_after_a_target_and_never_more_than_two_trades(): up = [(1042.0, 1052.5, 1041.0, 1052.0)] day = WARM_UP + BREAK_R1 + up + candle(1052.0, 1062.0, 1051.0, 1061.0) + up assert len(run(day)) == 1 # "make a profit of 30 rupees, and stop trading for that particular day" down = candle(1042.0, 1042.0, 1036.0, 1036.0) # its first minute falls through the 0.5% stop day = WARM_UP + BREAK_R1 + down + BREAK_R1 + down + BREAK_R1 + down assert [t.reason for t, _ in run(day)] == ["stop", "stop"] # two stop-losses, then the day is over def test_the_vwap_stop_has_room_a_floor_and_a_limit(): [(trade, _)] = run(WARM_UP + BREAK_R1 + [QUIET], rules=vp.STOP_BELOW_VWAP, vwap=1037.0) assert trade.stop == pytest.approx(1037 * (1 - 0.0005)) # "below the VWAP ... space of Rupees 1 or 2" [(trade, _)] = run(WARM_UP + BREAK_R1 + [QUIET], rules=vp.STOP_BELOW_VWAP, vwap=1041.5) assert trade.stop == pytest.approx(1042 * (1 - 0.0025)) # never closer than 0.25%: "not very tight" assert run(WARM_UP + BREAK_R1, rules=vp.STOP_BELOW_VWAP, vwap=1020.0) == [] # further than the target: forgo def test_a_stop_order_fills_at_the_level_and_a_stop_touch_in_that_minute_is_a_loss(): calm = candle(1034.0, 1035.0, 1033.0, 1035.0) # the morning's high is the close: R1 is the nearest level above rows = calm + [(1036.0, 1041.0, 1034.0, 1040.5)] + [QUIET] * 4 [(trade, _)] = run(rows, rules=vp.STOP_ORDER_AT_LEVEL) assert (trade.entry_minute, trade.entry_price, trade.level) == (5, 1040.0, 1040.0) # the minute it traded through assert (trade.reason, trade.exit_minute) == ("stop", 5) # that minute also reached 0.5% below: which came first? gap = calm + [(1041.0, 1042.0, 1041.0, 1041.5)] + [QUIET] * 4 [(trade, _)] = run(gap, rules=vp.STOP_ORDER_AT_LEVEL) assert trade.entry_price == 1041.0 # it jumped past the level: filled at the minute's open def test_costs_charge_stt_on_the_sell_leg_and_slippage_on_market_fills(): costs = EquityCosts(0.00025, (("2015-01-01", 0.0000325),), 0.00003, 0.000001, 0.18, 0.0002, 0.0236) assert costs.points("2020-07-27", LONG, 1000.0, 1010.0) == pytest.approx( 0.00025 * 1010 + 0.0000325 * 2010 * 1.18 + 0.00003 * 1000 + 0.000001 * 2010 * 1.18 ) assert costs.points("2020-07-27", SHORT, 1010.0, 1000.0) == costs.points("2020-07-27", LONG, 1000.0, 1010.0) [(trade, r)] = run(WARM_UP + BREAK_R1 + [(1042.0, 1052.9, 1041.0, 1052.0)], costs=costs) assert trade.entry_price == pytest.approx(1042 * 1.0002) def test_vwap_restarts_every_session(): bars = session_bars([YESTERDAY, nse_day("2020-07-27", "09:15", [(10.0, 12.0, 8.0, 11.0)] * 5)], 5) vwap = session_vwap(bars.candles, bars.volume, bars.session) assert vwap[0] == pytest.approx((1036 + 1034 + 1035) / 3) and vwap[1] == pytest.approx((12 + 8 + 11) / 3) # --------------------------------------------------------------------------------------------------------------- # His own days and the whole history (tools/bb_backtest.py produced the published audit). needs_data = pytest.mark.skipif(not (RAW / "kaggle_hk7797").exists(), reason="stock minute data not downloaded") READINGS = [ vp.EXACT, vp.NO_STOCHASTIC, vp.STOP_BELOW_VWAP, vp.TARGET_NEXT_PIVOT, vp.STOP_ORDER_AT_LEVEL, vp.MOST_GENEROUS, ] def stock_parts(rules: dict, symbol: str) -> list[tuple[Path, float, str, str]]: parts = [] for part in rules["S1"]["data"][symbol]: name = part["source"].split()[-1].split("/")[-1] folder = "kaggle_hk7797" if "hk7797" in part["source"] else "kaggle" parts.append((RAW / folder / name, float(part.get("scale", 1.0)), part["from"], part["to"])) return parts def equity_costs(rules: dict, optimistic: bool) -> EquityCosts: c = rules["S1"]["costs"] brokerage = 2 * c["brokerage_per_order_inr"] * (1 + c["gst"]) / rules["money"]["fixed_risk_inr"]["S1"] exchange = tuple(map(tuple, c["nse_txn"])) return EquityCosts( c["stt_sell"], exchange, c["stamp_buy"], c["sebi_fee"], c["gst"], c["slippage_pct_per_fill"], brokerage, optimistic, ) @pytest.fixture(scope="module") def bajaj(): rules = json.loads(RULES_JSON.read_text()) return rules, load_nse_stock(stock_parts(rules, "BAJFINANCE")) @needs_data def test_his_own_bajaj_finance_days(bajaj): rules, sessions = bajaj trades = [ t for t, _ in vp.backtest(sessions, EXACT, equity_costs(rules, False)) if "2020-07-27" <= t.day <= "2020-08-14" ] shown = {(t.day, t.side): round(t.entry_price) for t in trades} # The trades he took on screen: short 3183 on 27 Jul, short 3482 on 10 Aug, long ~3450 after a stop on 11 Aug, # short ~3440 on 13 Aug, short 3400 after a stop on 14 Aug (8OurAiVyhE4, 561 s to 1840 s). assert shown[("2020-07-27", SHORT)] == 3183 assert shown[("2020-08-10", SHORT)] == 3482 assert shown[("2020-08-11", LONG)] == 3455 assert shown[("2020-08-13", SHORT)] == 3446 assert shown[("2020-08-14", SHORT)] == 3401 days = sorted({t.day for t in trades}) assert len(days) == 15 def published_engine(): sys.path.insert(0, str(ROOT / "tools")) import bb_backtest return bb_backtest @needs_data @pytest.mark.parametrize("symbol", ["BAJFINANCE", "HDFCBANK", "RELIANCE", "HDFC"]) @pytest.mark.parametrize("cost_case", ["costs", "zero costs", "optimistic"]) def test_every_trade_matches_the_audit_engine(symbol, cost_case): bb = published_engine() rules, _ = bb.load_rules() market = bb.prepare("S1", symbol, rules, "kaggle") sessions = load_nse_stock(stock_parts(rules, symbol)) assert [str(s.day.date()) for s in sessions] == [d.date for d in market.days] zero, optimistic = cost_case == "zero costs", cost_case == "optimistic" costs = replace(NO_COSTS, optimistic=optimistic) if zero else equity_costs(rules, optimistic) for variant, reading in zip(rules["S1"]["variants"], READINGS, strict=True): old = bb.run_s1(market, variant, rules, zero=zero, optimistic=optimistic) new = vp.backtest(sessions, reading, costs) assert len(new) == len(old), variant["name"] for o, (n, r) in zip(old, new, strict=True): assert (o["day"], o["side"], o["entry_i"], o["exit_i"], o["reason"], o["level"]) == ( n.day, n.side, n.entry_minute, n.exit_minute, n.reason, round(n.level, 4), ) assert (o["entry_px"], o["stop"], o["target"], o["exit_px"], o["R"]) == ( n.entry_price, n.stop, n.target, n.exit_price, r, )