89 lines
3.1 KiB
Python
89 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from datetime import timedelta
|
|
from pathlib import Path
|
|
|
|
from crypto_spot_bot.models import Signal, utc_now
|
|
from crypto_spot_bot.storage import MAX_SIGNAL_DIAGNOSTICS_BYTES, PRUNE_BATCH_SIZE, Storage
|
|
from tools.compact_runtime_db import compact_database
|
|
|
|
|
|
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
|
|
|
|
|
|
def test_runtime_compaction_preserves_durable_state_and_bounds_telemetry(tmp_path) -> None:
|
|
database = tmp_path / "tradebot.sqlite3"
|
|
storage = Storage(database)
|
|
for index in range(10):
|
|
storage.insert_signal(
|
|
Signal("BTCUSDT", "BUY", 0.8, f"signal-{index}"),
|
|
hold_sample_seconds=0,
|
|
)
|
|
storage.set_runtime("active", {"value": 1})
|
|
|
|
result = compact_database(
|
|
database,
|
|
recent_rows={"signals": 3, "equity": 0, "events": 0, "llm_advice": 0},
|
|
)
|
|
|
|
compacted = Storage(database)
|
|
assert [row["reason"] for row in compacted.recent_signals(10)] == [
|
|
"signal-9",
|
|
"signal-8",
|
|
"signal-7",
|
|
]
|
|
assert compacted.get_runtime("active") == {"value": 1}
|
|
assert Path(result["backup"]).is_file()
|
|
assert result["rows"]["signals"] == 3
|