spb/crypto-trade-bot

1from hypothesis import given, settings2from hypothesis import strategies as st34from ktrade.risk.gateway import MarketState, OrderIntent, PortfolioSnapshot, RiskGateway, RiskLimits567def base():8    return (9        OrderIntent(symbol="BTC/USD", side="buy", qty=0.01),10        PortfolioSnapshot(equity=10_000, cash=10_000, positions={}, marks={}, peak_equity=10_000, day_start_equity=10_000),11        MarketState(last_price=30_000, data_age_seconds=1, ordermin=0.0001, costmin=0.5),12    )131415def test_allows_a_sane_order():16    i, p, m = base()17    assert RiskGateway(RiskLimits(max_order_notional=1000, max_position_notional=5000)).evaluate(i, p, m).allowed181920def test_default_deny_on_missing_inputs():21    i, _, _ = base()22    d = RiskGateway().evaluate(i, None, None)23    assert not d.allowed and "default-deny" in " ".join(d.reasons)242526def test_kill_switch_blocks_everything():27    i, p, m = base()28    assert not RiskGateway(kill_switch=True).evaluate(i, p, m).allowed293031def test_stale_data_blocks():32    i, p, m = base()33    m.data_age_seconds = 99934    d = RiskGateway().evaluate(i, p, m)35    assert not d.allowed and d.checks["data_fresh"] is False363738def test_min_size_and_notional_limits():39    i, p, m = base()40    i.qty = 0.0000141    assert RiskGateway().evaluate(i, p, m).checks["min_size"] is False42    i.qty = 1.0  # 30k notional43    assert RiskGateway(RiskLimits(max_order_notional=1000)).evaluate(i, p, m).checks["order_notional"] is False444546def test_no_short_on_spot():47    i, p, m = base()48    i.side = "sell"49    assert RiskGateway().evaluate(i, p, m).checks["no_short"] is False505152def test_reducing_sell_allowed_under_drawdown_but_buy_not():53    i, p, m = base()54    p.positions = {"BTC/USD": 0.05}55    p.equity = 8_00056    p.cash = 6_50057    gw = RiskGateway(RiskLimits(max_drawdown_limit=0.10, max_order_notional=5000, max_position_notional=5000))58    assert not gw.evaluate(i, p, m).allowed59    sell = OrderIntent(symbol="BTC/USD", side="sell", qty=0.05)60    assert gw.evaluate(sell, p, m).allowed616263def test_unreconciled_ledger_blocks():64    i, p, m = base()65    p.reconciled = False66    assert not RiskGateway().evaluate(i, p, m).allowed676869@settings(max_examples=200, deadline=None)70@given(qty=st.floats(min_value=0, max_value=10) | st.just(float('nan')) | st.just(float('inf')), cash=st.floats(0, 1e6), price=st.floats(1, 1e6), age=st.floats(0, 1000))71def test_allowed_implies_all_checks_pass(qty, cash, price, age):72    i = OrderIntent(symbol="X", side="buy", qty=qty)73    p = PortfolioSnapshot(equity=cash, cash=cash)74    m = MarketState(last_price=price, data_age_seconds=age)75    d = RiskGateway().evaluate(i, p, m)76    if d.allowed:77        assert all(d.checks.values()) and not d.reasons78    else:79        assert d.reasons80