28 lines
1.8 KiB
Python
28 lines
1.8 KiB
Python
from fxhnt.application.pump_detection import pump_signal, evaluate_pump_edge
|
|
|
|
|
|
def test_signal_fires_on_volume_zspike_and_is_causal():
|
|
turnover = {"X": {d: 1_000_000.0 + (d % 7) * 20_000.0 for d in range(1, 40)}} # 39 calm days, mild variation
|
|
turnover["X"][40] = 50_000_000.0 # 50x spike on day 40
|
|
sig = pump_signal(turnover, z=3.0)
|
|
assert 40 in sig and sig[40].get("X") == -1.0 # fires on the spike day (short target)
|
|
assert all("X" not in sig.get(d, {}) for d in range(1, 40)) # never before (causal + calm)
|
|
|
|
|
|
def test_evaluate_pump_edge_gated_verdict_and_execution_gate_kills_it():
|
|
# A pump short is a ROUND TRIP: enter short (-1) on the spike day catching a 5%/2% drop, cover (flat) the
|
|
# next day. Turnover-basis slippage is paid on each position CHANGE (entry AND exit), not on a held book.
|
|
wbd = {}
|
|
for i in range(1, 80):
|
|
if i % 2 == 1: # spike day: short, realize the drop
|
|
wbd[i] = {"C": (-1.0, -(0.05 if i % 4 == 1 else 0.02))}
|
|
else: # next day: cover -> flat
|
|
wbd[i] = {"C": (0.0, 0.0)}
|
|
turnover = {"C": {i: 1_000_000.0 for i in range(1, 80)}}
|
|
v = evaluate_pump_edge(wbd, turnover, spread_bps={}, aum=1_000.0, impact_k=0.0)
|
|
assert set(v) >= {"sharpe", "survives_execution", "marginal_vs_positioning"}
|
|
assert v["survives_execution"] is True # ~3.5%/short gross, no cost -> positive
|
|
# the killer gate: crushing 8% spread each side (entry + exit ~16%/round-trip) turns it net-negative
|
|
v2 = evaluate_pump_edge(wbd, turnover, spread_bps={"C": 800.0}, aum=1_000.0, impact_k=0.0)
|
|
assert v2["survives_execution"] is False # round-trip slippage >> gross -> net negative
|