spb/crypto-trade-bot

1import numpy as np2import pandas as pd34from ktrade.backtest.costs import CostModel, round_lot5from ktrade.backtest.engine import run_backtest6from ktrade.backtest.metrics import compute_metrics7from ktrade.strategy.base import ParamSpec, Strategy, StrategyMeta8from ktrade.strategy.benchmarks import BuyAndHold9from tests.conftest import synthetic_candles101112class StepAt(Strategy):13    meta = StrategyMeta(name="step", label="step", family="test", hypothesis="go long at bar k", min_bars=1, params=[ParamSpec(name="k", type="int", default=10)])1415    def features(self, df):16        return pd.DataFrame({"ts": df["ts"].values}, index=df.index)1718    def target(self, feats):19        t = pd.Series(0.0, index=feats.index)20        t.iloc[self.params["k"]:] = 1.021        return t222324def test_execution_happens_at_next_open_not_same_bar():25    df = synthetic_candles(n=50, seed=2)26    k = 1027    res = run_backtest(StepAt({"k": k}), df, interval_min=60, cost_model=CostModel(spread_bps=0, slippage_bps=0, impact_coeff=0, taker_fee=0, max_participation=10))28    assert len(res.trades) == 129    t = res.trades[0]30    assert t.ts == int(df["ts"].iloc[k + 1])          # filled on bar k+131    assert t.ref_price == df["open"].iloc[k + 1]       # at its OPEN32    assert res.equity["position"].iloc[k] == 0.0       # no position on the decision bar33    assert res.equity["position"].iloc[k + 1] > 0343536def test_latency_adds_bars():37    df = synthetic_candles(n=50, seed=2)38    res = run_backtest(StepAt({"k": 10}), df, interval_min=60, cost_model=CostModel(latency_bars=2, taker_fee=0, spread_bps=0, slippage_bps=0, impact_coeff=0, max_participation=10))39    assert res.trades[0].ts == int(df["ts"].iloc[13])404142def test_fees_and_slippage_are_charged():43    df = synthetic_candles(n=50, seed=2)44    cm = CostModel(taker_fee=0.004, spread_bps=10, slippage_bps=5, impact_coeff=0, max_participation=10)45    res = run_backtest(StepAt({"k": 5}), df, interval_min=60, cost_model=cm, initial_cash=10_000)46    t = res.trades[0]47    assert t.price > t.ref_price  # buy pays half-spread + slippage48    assert abs(t.fee - t.notional * 0.004) < 1e-949    assert res.fees_paid > 0 and res.slippage_paid > 050    assert abs(t.notional + t.fee - (10_000 - res.equity["cash"].iloc[6])) < 1e-6515253def test_partial_fills_respect_participation_cap():54    df = synthetic_candles(n=60, seed=2)55    df["volume"] = 0.01  # tiny bar volume → participation cap binds56    cm = CostModel(max_participation=0.5, taker_fee=0, spread_bps=0, slippage_bps=0, impact_coeff=0)57    res = run_backtest(StepAt({"k": 5}), df, interval_min=60, cost_model=cm, initial_cash=10_000)58    assert all(t.qty <= 0.005 + 1e-12 for t in res.trades)59    assert any(t.partial for t in res.trades)60    assert len(res.trades) > 1  # remainder keeps being worked on following bars616263def test_lot_rounding_and_min_size():64    assert round_lot(1.23456789123, 8) == 1.2345678965    assert round_lot(0.000000001, 8) == 0.066    df = synthetic_candles(n=40, seed=2)67    res = run_backtest(StepAt({"k": 5}), df, interval_min=60, initial_cash=1.0, ordermin=0.0001, costmin=0.5, cost_model=CostModel(max_participation=10))68    assert res.trades == [] and res.rejected  # 1 USD cannot buy ≥ 0.0001 BTC at 30k → rejected by the gateway697071def test_buy_and_hold_equity_tracks_price():72    df = synthetic_candles(n=300, seed=4)73    cm = CostModel(taker_fee=0, spread_bps=0, slippage_bps=0, impact_coeff=0, max_participation=10)74    res = run_backtest(BuyAndHold(), df, interval_min=60, cost_model=cm, initial_cash=10_000)75    entry = df["open"].iloc[1]76    expected = 10_000 / entry * df["close"].iloc[-1]77    assert abs(res.equity["equity"].iloc[-1] - expected) / expected < 1e-678    m = compute_metrics(res.equity, res.trades, 60, 10_000, 0, 0, res.turnover)79    assert m["bars"] == 300 and np.isfinite(m["sharpe"]) and -1 <= m["max_drawdown"] <= 0808182def test_never_short_and_never_negative_cash():83    from ktrade.strategy.momentum import TimeSeriesMomentum8485    df = synthetic_candles(n=1500, seed=9, vol=0.03)86    res = run_backtest(TimeSeriesMomentum(), df, interval_min=60)87    assert (res.equity["position"] >= -1e-12).all()88    assert (res.equity["cash"] >= -1e-6).all()89