61 lines
2.2 KiB
Python
61 lines
2.2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import timedelta
|
|
|
|
from crypto_spot_bot.models import Signal, utc_now
|
|
from crypto_spot_bot.storage import MAX_SIGNAL_DIAGNOSTICS_BYTES, PRUNE_BATCH_SIZE, Storage
|
|
|
|
|
|
def test_hold_sampling_is_independent_for_each_reason_and_diagnostics_are_bounded(tmp_path) -> None:
|
|
storage = Storage(tmp_path / "tradebot.sqlite3")
|
|
diagnostics = {
|
|
"strategy_mode": "torch_forecast",
|
|
"checks": {"model_fresh_ok": False},
|
|
"forecast": {
|
|
"model": "torch_lstm",
|
|
"expected_return_percent": 0.42,
|
|
"model_fresh": False,
|
|
"feature_snapshot": [
|
|
{"name": f"feature-{index}", "interpretation": "x" * 1000}
|
|
for index in range(100)
|
|
],
|
|
},
|
|
}
|
|
|
|
first = Signal("BTCUSDT", "HOLD", 0.2, "entry blocked", diagnostics)
|
|
second = Signal("BTCUSDT", "HOLD", 0.2, "position held", diagnostics)
|
|
|
|
assert storage.insert_signal(first, hold_sample_seconds=60) is True
|
|
assert storage.insert_signal(second, hold_sample_seconds=60) is True
|
|
assert storage.insert_signal(first, hold_sample_seconds=60) is False
|
|
|
|
rows = storage.recent_signals(10)
|
|
assert len(rows) == 2
|
|
stored = json.loads(rows[0]["diagnostics_json"])
|
|
assert len(rows[0]["diagnostics_json"].encode("utf-8")) <= MAX_SIGNAL_DIAGNOSTICS_BYTES
|
|
assert stored["forecast"]["model"] == "torch_lstm"
|
|
assert "feature_snapshot" not in stored["forecast"]
|
|
|
|
|
|
def test_prune_deletes_only_one_bounded_batch_per_table(tmp_path) -> None:
|
|
storage = Storage(tmp_path / "tradebot.sqlite3")
|
|
old_timestamp = (utc_now() - timedelta(days=90)).isoformat()
|
|
rows = [
|
|
("BTCUSDT", "HOLD", 0.0, "old", "{}", old_timestamp)
|
|
for _ in range(PRUNE_BATCH_SIZE + 5)
|
|
]
|
|
with storage.connect() as conn:
|
|
conn.executemany(
|
|
"""
|
|
INSERT INTO signals (symbol, action, confidence, reason, diagnostics_json, created_at)
|
|
VALUES (?, ?, ?, ?, ?, ?)
|
|
""",
|
|
rows,
|
|
)
|
|
|
|
deleted = storage.prune(30)
|
|
|
|
assert deleted["signals"] == PRUNE_BATCH_SIZE
|
|
assert len(storage.recent_signals(PRUNE_BATCH_SIZE + 10)) == 5
|