44 lines
1.5 KiB
Python
44 lines
1.5 KiB
Python
import json
|
|
from datetime import datetime
|
|
|
|
from app.db.intraday_frequency import get_intraday_frequency_health
|
|
|
|
|
|
def test_intraday_frequency_dedupes_repeated_gate_reject_events(pg_conn):
|
|
now = datetime.now().isoformat(timespec="seconds")
|
|
detail = {
|
|
"reason": "paper_order_gate_rejected",
|
|
"gate_reasons": ["too_far_from_entry"],
|
|
"recommendation_id": 101,
|
|
"gate_detail": {
|
|
"target_price": 10,
|
|
"distance_to_entry_pct": 8.5,
|
|
"max_distance_to_entry_pct": 8,
|
|
"rr1": 1.8,
|
|
"min_rr": 1.25,
|
|
"rec_score": 55,
|
|
"min_rec_score": 25,
|
|
},
|
|
}
|
|
for _ in range(3):
|
|
pg_conn.execute(
|
|
"""
|
|
INSERT INTO paper_trade_events (
|
|
trade_id, recommendation_id, symbol, event_type, event_time, detail_json, strategy_code
|
|
) VALUES (
|
|
0, 101, 'DASH/USDT', 'paper_gate_reject', %s, %s, 'long_second_wave_pullback_1h_v1'
|
|
)
|
|
""",
|
|
(now, json.dumps(detail, ensure_ascii=False)),
|
|
)
|
|
pg_conn.commit()
|
|
|
|
data = get_intraday_frequency_health(days=1)
|
|
|
|
reasons = {item["reason"]: item["count"] for item in data["gate_reasons"]}
|
|
assert reasons["paper_order_gate_rejected"] == 1
|
|
assert reasons["too_far_from_entry"] == 1
|
|
assert len(data["gate_samples"]) == 1
|
|
assert data["gate_samples"][0]["detail"]["distance_to_entry_pct"] == 8.5
|
|
assert "entry_plan" not in data["gate_samples"][0]["detail"]
|