Codex
101 lines
3.9 KiB
Python
101 lines
3.9 KiB
Python
import os
|
|
import unittest
|
|
from datetime import datetime, timezone
|
|
|
|
os.environ.setdefault("DATABASE_URL", "postgresql+psycopg://user:pass@localhost:5432/ai_ict_test")
|
|
|
|
from src.services.market_state.external_chart_provider import BinanceChartReferenceProvider
|
|
|
|
|
|
class ExternalChartProviderTests(unittest.TestCase):
|
|
def test_provider_builds_live_snapshot_from_payload(self) -> None:
|
|
provider = BinanceChartReferenceProvider(repository=_ChartSnapshotRepo())
|
|
|
|
snapshot = provider.fetch_snapshot(
|
|
instrument_id=1,
|
|
timeframe="1h",
|
|
symbol_hint="BTC-USDT",
|
|
now=datetime(2026, 5, 14, 4, 0, tzinfo=timezone.utc),
|
|
payload=[
|
|
[0, "100", "110", "90", "105", "10", 1715659200000],
|
|
[1, "105", "115", "95", "111", "12", 1715662800000],
|
|
],
|
|
)
|
|
|
|
self.assertEqual(snapshot["status"], "active")
|
|
self.assertEqual(snapshot["cache_state"], "fresh")
|
|
self.assertEqual(snapshot["symbol"], "BTC-USDT")
|
|
self.assertEqual(snapshot["interval"], "1h")
|
|
self.assertIn("binance.com", snapshot["chart_url"])
|
|
self.assertEqual(snapshot["reference_price"], 111.0)
|
|
|
|
def test_provider_uses_cached_snapshot_when_fetch_fails(self) -> None:
|
|
repository = _ChartSnapshotRepo(
|
|
{
|
|
"provider_name": "binance_public_chart_reference:BTCUSDT:1h",
|
|
"fetched_ts": "2026-05-14T02:00:00+00:00",
|
|
"event_count": 1,
|
|
"code": "active",
|
|
"active_events": [
|
|
{
|
|
"symbol": "BTC-USDT",
|
|
"interval": "1h",
|
|
"chart_url": "https://www.binance.com/en/trade/BTC_USDT?type=spot",
|
|
"reference_price": 111.0,
|
|
"reference_ts": "2026-05-14T02:00:00+00:00",
|
|
}
|
|
],
|
|
"raw_payload": {"data": []},
|
|
}
|
|
)
|
|
provider = BinanceChartReferenceProvider(repository=repository, cache_max_age_seconds=21600)
|
|
provider._fetch_payload = lambda **kwargs: (_ for _ in ()).throw(RuntimeError("api down")) # type: ignore[attr-defined]
|
|
|
|
snapshot = provider.fetch_snapshot(
|
|
instrument_id=1,
|
|
timeframe="1h",
|
|
symbol_hint="BTC-USDT",
|
|
now=datetime(2026, 5, 14, 2, 30, tzinfo=timezone.utc),
|
|
)
|
|
|
|
self.assertEqual(snapshot["status"], "active")
|
|
self.assertEqual(snapshot["cache_state"], "cached_fallback")
|
|
self.assertIn("api down", snapshot["warning"])
|
|
|
|
def test_provider_returns_empty_fallback_without_cache(self) -> None:
|
|
provider = BinanceChartReferenceProvider(repository=_ChartSnapshotRepo())
|
|
provider._fetch_payload = lambda **kwargs: (_ for _ in ()).throw(RuntimeError("api down")) # type: ignore[attr-defined]
|
|
|
|
snapshot = provider.fetch_snapshot(
|
|
instrument_id=1,
|
|
timeframe="1h",
|
|
symbol_hint="BTC-USDT",
|
|
now=datetime(2026, 5, 14, 2, 30, tzinfo=timezone.utc),
|
|
)
|
|
|
|
self.assertEqual(snapshot["status"], "idle")
|
|
self.assertEqual(snapshot["cache_state"], "empty_fallback")
|
|
self.assertIn("binance.com", snapshot["chart_url"])
|
|
self.assertIsNone(snapshot["error"])
|
|
|
|
|
|
class _ChartSnapshotRepo:
|
|
def __init__(self, row: dict | None = None) -> None:
|
|
self.row = row or {}
|
|
|
|
def fetch_instrument_by_id(self, instrument_id: int):
|
|
return {"id": instrument_id, "symbol": "BTC-USDT", "active": True}
|
|
|
|
def fetch_news_provider_snapshot(self, provider_name: str):
|
|
if self.row and provider_name == self.row.get("provider_name"):
|
|
return dict(self.row)
|
|
return None
|
|
|
|
def upsert_news_provider_snapshot(self, **kwargs):
|
|
self.row = dict(kwargs)
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|