68 lines
2.6 KiB
Python
68 lines
2.6 KiB
Python
"""CLI `fxhnt bybit-ingest-oi` — universe resolve + open_interest ingest into bybit_features. NO network: the
|
|
BybitOpenInterest adapter is monkeypatched with a fake, and the operational DSN points at a temp sqlite db."""
|
|
from __future__ import annotations
|
|
|
|
from typer.testing import CliRunner
|
|
|
|
from fxhnt.cli import app
|
|
|
|
|
|
class _FakeOI:
|
|
def oi_history(self, symbol, *, start_ms=None):
|
|
return {19000: 1_000_000.0, 19001: 950_000.0}
|
|
|
|
|
|
class _FakeFunding:
|
|
def active_universe(self):
|
|
return ["BTCUSDT", "ETHUSDT"]
|
|
|
|
|
|
def _patch(monkeypatch, tmp_path):
|
|
dsn = f"sqlite:///{tmp_path / 'op.db'}"
|
|
monkeypatch.setenv("FXHNT_OPERATIONAL_DSN", dsn)
|
|
from fxhnt import cli as climod
|
|
climod.get_settings.cache_clear()
|
|
monkeypatch.setattr("fxhnt.adapters.data.bybit_open_interest.BybitOpenInterest",
|
|
lambda *a, **k: _FakeOI())
|
|
monkeypatch.setattr("fxhnt.adapters.data.bybit_funding.BybitFunding", lambda *a, **k: _FakeFunding())
|
|
return dsn
|
|
|
|
|
|
def test_cli_all_active(monkeypatch, tmp_path):
|
|
dsn = _patch(monkeypatch, tmp_path)
|
|
res = CliRunner().invoke(app, ["bybit-ingest-oi", "--symbols", "all"])
|
|
assert res.exit_code == 0, res.output
|
|
assert "2 symbols (all active USDT perps)" in res.output
|
|
assert "done: 2 symbols, 4 day-rows" in res.output
|
|
|
|
from fxhnt.adapters.warehouse.timescale_feature_store import TimescaleFeatureStore
|
|
store = TimescaleFeatureStore(dsn, table="bybit_features")
|
|
assert store.read_panel(["BTCUSDT", "ETHUSDT"], "open_interest") == {
|
|
"BTCUSDT": {19000: 1_000_000.0, 19001: 950_000.0},
|
|
"ETHUSDT": {19000: 1_000_000.0, 19001: 950_000.0},
|
|
}
|
|
assert store.crypto_funding_panel() == {} # ONLY open_interest was written
|
|
store.close()
|
|
|
|
|
|
def test_cli_explicit_symbols_and_from(monkeypatch, tmp_path):
|
|
_patch(monkeypatch, tmp_path)
|
|
res = CliRunner().invoke(app, ["bybit-ingest-oi", "--symbols", "BTCUSDT", "--from", "2023-01"])
|
|
assert res.exit_code == 0, res.output
|
|
assert "1 symbols (explicit)" in res.output
|
|
assert "floor=2023-01" in res.output
|
|
|
|
|
|
def test_cli_liquid_empty_universe_errors(monkeypatch, tmp_path):
|
|
_patch(monkeypatch, tmp_path)
|
|
res = CliRunner().invoke(app, ["bybit-ingest-oi", "--symbols", "liquid"])
|
|
assert res.exit_code == 1
|
|
assert "empty symbol universe" in res.output
|
|
|
|
|
|
def test_cli_bad_from_month(monkeypatch, tmp_path):
|
|
_patch(monkeypatch, tmp_path)
|
|
res = CliRunner().invoke(app, ["bybit-ingest-oi", "--symbols", "BTCUSDT", "--from", "2023"])
|
|
assert res.exit_code == 1
|
|
assert "must be YYYY-MM" in res.output
|