Build 060.27: implement Persistent Market Data Storage
This commit is contained in:
103
app/tests/integration/market_data/storage/conftest.py
Normal file
103
app/tests/integration/market_data/storage/conftest.py
Normal file
@@ -0,0 +1,103 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from collections.abc import Iterator
|
||||
|
||||
import psycopg
|
||||
import pytest
|
||||
|
||||
from src.storage.migrations import StorageMigrationRunner
|
||||
from src.storage.postgres_pool import PostgresConnectionPool
|
||||
from tests.support.postgres_market_data import (
|
||||
PostgresTestSettings,
|
||||
acquire_postgres_test_lock,
|
||||
connect_postgres_test_database,
|
||||
load_postgres_test_settings,
|
||||
release_postgres_test_lock,
|
||||
reset_postgres_test_database,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
def postgres_test_settings() -> Iterator[PostgresTestSettings]:
|
||||
try:
|
||||
settings = load_postgres_test_settings(os.environ)
|
||||
except ValueError as error:
|
||||
pytest.fail(str(error), pytrace=False)
|
||||
|
||||
if settings is None:
|
||||
pytest.skip(
|
||||
"PostgreSQL integration is opt-in; set "
|
||||
"DZENTRA_RUN_POSTGRES_TESTS=1 and "
|
||||
"DZENTRA_TEST_POSTGRES_DSN explicitly."
|
||||
)
|
||||
|
||||
try:
|
||||
control = connect_postgres_test_database(settings)
|
||||
except psycopg.Error as error:
|
||||
pytest.fail(
|
||||
"Could not connect to the explicit PostgreSQL test database: "
|
||||
f"{type(error).__name__}.",
|
||||
pytrace=False,
|
||||
)
|
||||
|
||||
with control:
|
||||
if not acquire_postgres_test_lock(control):
|
||||
pytest.fail(
|
||||
"Another integration session already owns this test database.",
|
||||
pytrace=False,
|
||||
)
|
||||
|
||||
try:
|
||||
reset_postgres_test_database(
|
||||
control,
|
||||
expected_database_name=settings.database_name,
|
||||
)
|
||||
yield settings
|
||||
finally:
|
||||
reset_postgres_test_database(
|
||||
control,
|
||||
expected_database_name=settings.database_name,
|
||||
)
|
||||
release_postgres_test_lock(control)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def clean_postgres_test_database(
|
||||
postgres_test_settings: PostgresTestSettings,
|
||||
) -> Iterator[None]:
|
||||
with connect_postgres_test_database(postgres_test_settings) as control:
|
||||
reset_postgres_test_database(
|
||||
control,
|
||||
expected_database_name=postgres_test_settings.database_name,
|
||||
)
|
||||
|
||||
yield
|
||||
|
||||
with connect_postgres_test_database(postgres_test_settings) as control:
|
||||
reset_postgres_test_database(
|
||||
control,
|
||||
expected_database_name=postgres_test_settings.database_name,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def migrated_postgres_pool(
|
||||
postgres_test_settings: PostgresTestSettings,
|
||||
) -> Iterator[PostgresConnectionPool]:
|
||||
pool = PostgresConnectionPool(
|
||||
conninfo=postgres_test_settings.dsn,
|
||||
min_size=1,
|
||||
max_size=4,
|
||||
timeout_seconds=5.0,
|
||||
name="market-data-integration",
|
||||
)
|
||||
pool.open()
|
||||
StorageMigrationRunner(
|
||||
connection_provider=pool.connection,
|
||||
).run()
|
||||
|
||||
try:
|
||||
yield pool
|
||||
finally:
|
||||
pool.close()
|
||||
@@ -0,0 +1,257 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from decimal import Decimal
|
||||
import threading
|
||||
|
||||
import pytest
|
||||
|
||||
from src.market_data.acquisition.models.quote import Quote
|
||||
from src.market_data.acquisition.models.trade import (
|
||||
Trade,
|
||||
TradeAggressorSide,
|
||||
)
|
||||
from src.market_data.storage import (
|
||||
MARKET_DATA_PARTITION_ADVISORY_LOCK_ID,
|
||||
MarketDataPartitionType,
|
||||
MarketDataRetentionPolicy,
|
||||
MarketDataStorageOperationError,
|
||||
PostgresMarketDataPartitionManager,
|
||||
PostgresMarketDataRetentionService,
|
||||
PostgresQuoteRepository,
|
||||
PostgresTradeRepository,
|
||||
)
|
||||
from src.storage.postgres_pool import PostgresConnectionPool
|
||||
from tests.support.postgres_market_data import (
|
||||
PostgresTestSettings,
|
||||
connect_postgres_test_database,
|
||||
wait_for_postgres_advisory_lock_waiters,
|
||||
)
|
||||
|
||||
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
VENUE = "dzengi"
|
||||
SYMBOL = "BTC/USD_LEVERAGE"
|
||||
|
||||
|
||||
def _trade(*, trade_id: int, executed_at: datetime) -> Trade:
|
||||
return Trade(
|
||||
symbol=SYMBOL,
|
||||
trade_id=trade_id,
|
||||
price=Decimal("100"),
|
||||
quantity=Decimal("1"),
|
||||
executed_at=executed_at,
|
||||
aggressor_side=TradeAggressorSide.BUY,
|
||||
source="dzengi_websocket_trade",
|
||||
)
|
||||
|
||||
|
||||
def _quote(*, received_at: datetime) -> Quote:
|
||||
return Quote(
|
||||
symbol=SYMBOL,
|
||||
last_price=Decimal("100"),
|
||||
bid_price=Decimal("99"),
|
||||
ask_price=Decimal("101"),
|
||||
exchange_timestamp=received_at,
|
||||
received_at=received_at,
|
||||
source="dzengi",
|
||||
)
|
||||
|
||||
|
||||
def test_real_partition_creation_moves_default_row_and_is_idempotent(
|
||||
migrated_postgres_pool: PostgresConnectionPool,
|
||||
) -> None:
|
||||
event_time = datetime(2026, 7, 15, tzinfo=timezone.utc)
|
||||
repository = PostgresTradeRepository(
|
||||
connection_provider=migrated_postgres_pool.connection,
|
||||
)
|
||||
manager = PostgresMarketDataPartitionManager(
|
||||
connection_provider=migrated_postgres_pool.connection,
|
||||
)
|
||||
repository.store_trade(
|
||||
venue=VENUE,
|
||||
trade=_trade(trade_id=1, executed_at=event_time),
|
||||
observed_at=event_time,
|
||||
)
|
||||
|
||||
created = manager.ensure_month_partition(
|
||||
data_type=MarketDataPartitionType.TRADES,
|
||||
month=event_time,
|
||||
)
|
||||
repeated = manager.ensure_month_partition(
|
||||
data_type=MarketDataPartitionType.TRADES,
|
||||
month=event_time,
|
||||
)
|
||||
|
||||
with migrated_postgres_pool.connection() as connection:
|
||||
with connection.cursor() as cursor:
|
||||
cursor.execute(
|
||||
"SELECT tableoid::regclass::text FROM market_data.trades"
|
||||
)
|
||||
relation = cursor.fetchone()
|
||||
|
||||
assert created.created is True
|
||||
assert created.moved_row_count == 1
|
||||
assert repeated.created is False
|
||||
assert repeated.moved_row_count == 0
|
||||
assert relation == ("market_data.trades_2026_07",)
|
||||
|
||||
|
||||
def test_two_real_partition_callers_create_one_partition(
|
||||
migrated_postgres_pool: PostgresConnectionPool,
|
||||
postgres_test_settings: PostgresTestSettings,
|
||||
) -> None:
|
||||
month = datetime(2026, 8, 1, tzinfo=timezone.utc)
|
||||
manager = PostgresMarketDataPartitionManager(
|
||||
connection_provider=migrated_postgres_pool.connection,
|
||||
)
|
||||
start_barrier = threading.Barrier(3)
|
||||
caller_ids: set[int] = set()
|
||||
caller_ids_lock = threading.Lock()
|
||||
|
||||
def ensure_partition() -> tuple[bool, int]:
|
||||
with caller_ids_lock:
|
||||
caller_ids.add(threading.get_ident())
|
||||
start_barrier.wait(timeout=5.0)
|
||||
result = manager.ensure_month_partition(
|
||||
data_type=MarketDataPartitionType.TRADES,
|
||||
month=month,
|
||||
)
|
||||
return result.created, result.moved_row_count
|
||||
|
||||
with connect_postgres_test_database(
|
||||
postgres_test_settings,
|
||||
autocommit=False,
|
||||
) as control:
|
||||
with control.cursor() as cursor:
|
||||
cursor.execute(
|
||||
"SELECT pg_advisory_xact_lock(%s)",
|
||||
(MARKET_DATA_PARTITION_ADVISORY_LOCK_ID,),
|
||||
)
|
||||
|
||||
with ThreadPoolExecutor(max_workers=2) as executor:
|
||||
futures = tuple(executor.submit(ensure_partition) for _ in range(2))
|
||||
start_barrier.wait(timeout=5.0)
|
||||
|
||||
try:
|
||||
wait_for_postgres_advisory_lock_waiters(
|
||||
control,
|
||||
lock_id=MARKET_DATA_PARTITION_ADVISORY_LOCK_ID,
|
||||
expected_count=2,
|
||||
)
|
||||
finally:
|
||||
control.commit()
|
||||
|
||||
results = tuple(future.result(timeout=10.0) for future in futures)
|
||||
|
||||
assert len(caller_ids) == 2
|
||||
assert sorted(results) == [(False, 0), (True, 0)]
|
||||
|
||||
|
||||
def test_real_retention_uses_exact_cutoff(
|
||||
migrated_postgres_pool: PostgresConnectionPool,
|
||||
) -> None:
|
||||
now = datetime(2026, 8, 15, 12, 0, tzinfo=timezone.utc)
|
||||
cutoff = now - timedelta(days=10)
|
||||
repository = PostgresTradeRepository(
|
||||
connection_provider=migrated_postgres_pool.connection,
|
||||
)
|
||||
service = PostgresMarketDataRetentionService(
|
||||
connection_provider=migrated_postgres_pool.connection,
|
||||
)
|
||||
|
||||
for trade_id, executed_at in (
|
||||
(1, cutoff - timedelta(microseconds=1)),
|
||||
(2, cutoff),
|
||||
(3, cutoff + timedelta(microseconds=1)),
|
||||
):
|
||||
repository.store_trade(
|
||||
venue=VENUE,
|
||||
trade=_trade(trade_id=trade_id, executed_at=executed_at),
|
||||
observed_at=now,
|
||||
)
|
||||
|
||||
result = service.apply(
|
||||
policy=MarketDataRetentionPolicy(enabled=True, trade_days=10),
|
||||
now=now,
|
||||
)
|
||||
|
||||
with migrated_postgres_pool.connection() as connection:
|
||||
with connection.cursor() as cursor:
|
||||
cursor.execute(
|
||||
"SELECT trade_id FROM market_data.trades ORDER BY trade_id"
|
||||
)
|
||||
remaining = tuple(row[0] for row in cursor.fetchall())
|
||||
|
||||
assert result.total_removed_count == 1
|
||||
assert remaining == (2, 3)
|
||||
|
||||
|
||||
def test_real_retention_failure_rolls_back_all_data_types(
|
||||
migrated_postgres_pool: PostgresConnectionPool,
|
||||
) -> None:
|
||||
now = datetime(2026, 8, 15, 12, 0, tzinfo=timezone.utc)
|
||||
old_time = now - timedelta(days=30)
|
||||
trade_repository = PostgresTradeRepository(
|
||||
connection_provider=migrated_postgres_pool.connection,
|
||||
)
|
||||
quote_repository = PostgresQuoteRepository(
|
||||
connection_provider=migrated_postgres_pool.connection,
|
||||
)
|
||||
service = PostgresMarketDataRetentionService(
|
||||
connection_provider=migrated_postgres_pool.connection,
|
||||
)
|
||||
trade_repository.store_trade(
|
||||
venue=VENUE,
|
||||
trade=_trade(trade_id=1, executed_at=old_time),
|
||||
observed_at=now,
|
||||
)
|
||||
quote_repository.store_quote(
|
||||
venue=VENUE,
|
||||
quote=_quote(received_at=old_time),
|
||||
)
|
||||
|
||||
with migrated_postgres_pool.connection() as connection:
|
||||
with connection.cursor() as cursor:
|
||||
cursor.execute(
|
||||
"""
|
||||
CREATE FUNCTION market_data.reject_quote_delete()
|
||||
RETURNS trigger
|
||||
LANGUAGE plpgsql
|
||||
AS $$
|
||||
BEGIN
|
||||
RAISE EXCEPTION 'injected quote retention failure';
|
||||
END;
|
||||
$$
|
||||
"""
|
||||
)
|
||||
cursor.execute(
|
||||
"""
|
||||
CREATE TRIGGER reject_quote_delete
|
||||
BEFORE DELETE ON market_data.quotes
|
||||
FOR EACH ROW
|
||||
EXECUTE FUNCTION market_data.reject_quote_delete()
|
||||
"""
|
||||
)
|
||||
|
||||
with pytest.raises(MarketDataStorageOperationError):
|
||||
service.apply(
|
||||
policy=MarketDataRetentionPolicy(
|
||||
enabled=True,
|
||||
trade_days=10,
|
||||
quote_days=10,
|
||||
),
|
||||
now=now,
|
||||
)
|
||||
|
||||
with migrated_postgres_pool.connection() as connection:
|
||||
with connection.cursor() as cursor:
|
||||
cursor.execute("SELECT COUNT(*) FROM market_data.trades")
|
||||
trade_count = cursor.fetchone()
|
||||
cursor.execute("SELECT COUNT(*) FROM market_data.quotes")
|
||||
quote_count = cursor.fetchone()
|
||||
|
||||
assert trade_count == (1,)
|
||||
assert quote_count == (1,)
|
||||
@@ -0,0 +1,272 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
from dataclasses import replace
|
||||
from datetime import datetime, timedelta, timezone
|
||||
from decimal import Decimal
|
||||
import threading
|
||||
|
||||
import pytest
|
||||
|
||||
from src.market_data.acquisition.models.candle import Candle
|
||||
from src.market_data.acquisition.models.quote import Quote
|
||||
from src.market_data.acquisition.models.trade import (
|
||||
Trade,
|
||||
TradeAggressorSide,
|
||||
)
|
||||
from src.market_data.storage import (
|
||||
MarketDataStorageConflictError,
|
||||
MarketDataWriteStatus,
|
||||
PostgresCandleRepository,
|
||||
PostgresQuoteRepository,
|
||||
PostgresTradeRepository,
|
||||
)
|
||||
from src.storage.postgres_pool import PostgresConnectionPool
|
||||
|
||||
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
VENUE = "dzengi"
|
||||
SYMBOL = "BTC/USD_LEVERAGE"
|
||||
EVENT_TIME = datetime(2026, 7, 31, 12, 0, tzinfo=timezone.utc)
|
||||
OBSERVED_AT = EVENT_TIME + timedelta(seconds=1)
|
||||
|
||||
|
||||
def _trade(
|
||||
*,
|
||||
trade_id: int = 100,
|
||||
executed_at: datetime = EVENT_TIME,
|
||||
price: Decimal = Decimal("64159.45"),
|
||||
source: str = "dzengi_websocket_trade",
|
||||
symbol: str = SYMBOL,
|
||||
) -> Trade:
|
||||
return Trade(
|
||||
symbol=symbol,
|
||||
trade_id=trade_id,
|
||||
price=price,
|
||||
quantity=Decimal("0.125"),
|
||||
executed_at=executed_at,
|
||||
aggressor_side=TradeAggressorSide.BUY,
|
||||
source=source,
|
||||
)
|
||||
|
||||
|
||||
def _quote(*, source: str = "dzengi") -> Quote:
|
||||
return Quote(
|
||||
symbol=SYMBOL,
|
||||
last_price=Decimal("100"),
|
||||
bid_price=Decimal("99"),
|
||||
ask_price=Decimal("101"),
|
||||
exchange_timestamp=EVENT_TIME,
|
||||
received_at=OBSERVED_AT,
|
||||
source=source,
|
||||
)
|
||||
|
||||
|
||||
def _candle(*, source: str = "rest_klines:bid") -> Candle:
|
||||
return Candle(
|
||||
symbol=SYMBOL,
|
||||
interval="1m",
|
||||
open_time=EVENT_TIME,
|
||||
open_price=Decimal("100"),
|
||||
high_price=Decimal("110"),
|
||||
low_price=Decimal("90"),
|
||||
close_price=Decimal("105"),
|
||||
volume=Decimal("10"),
|
||||
source=source,
|
||||
)
|
||||
|
||||
|
||||
def test_real_trade_insert_duplicate_provenance_and_conflict(
|
||||
migrated_postgres_pool: PostgresConnectionPool,
|
||||
) -> None:
|
||||
repository = PostgresTradeRepository(
|
||||
connection_provider=migrated_postgres_pool.connection,
|
||||
)
|
||||
websocket_trade = _trade()
|
||||
|
||||
inserted = repository.store_trade(
|
||||
venue=VENUE,
|
||||
trade=websocket_trade,
|
||||
observed_at=OBSERVED_AT,
|
||||
)
|
||||
duplicate = repository.store_trade(
|
||||
venue=VENUE,
|
||||
trade=websocket_trade,
|
||||
observed_at=OBSERVED_AT,
|
||||
)
|
||||
provenance = repository.store_trade(
|
||||
venue=VENUE,
|
||||
trade=replace(websocket_trade, source="dzengi"),
|
||||
observed_at=OBSERVED_AT + timedelta(seconds=5),
|
||||
)
|
||||
|
||||
with pytest.raises(MarketDataStorageConflictError):
|
||||
repository.store_trade(
|
||||
venue=VENUE,
|
||||
trade=replace(websocket_trade, price=Decimal("999")),
|
||||
observed_at=OBSERVED_AT,
|
||||
)
|
||||
|
||||
with migrated_postgres_pool.connection() as connection:
|
||||
with connection.cursor() as cursor:
|
||||
cursor.execute(
|
||||
"""
|
||||
SELECT price, source, observation_sources,
|
||||
first_observed_at, last_observed_at
|
||||
FROM market_data.trades
|
||||
"""
|
||||
)
|
||||
rows = tuple(cursor.fetchall())
|
||||
|
||||
assert inserted.status is MarketDataWriteStatus.INSERTED
|
||||
assert duplicate.status is MarketDataWriteStatus.DUPLICATE
|
||||
assert provenance.status is MarketDataWriteStatus.PROVENANCE_UPDATED
|
||||
assert rows == (
|
||||
(
|
||||
Decimal("64159.45"),
|
||||
"dzengi_websocket_trade",
|
||||
["dzengi_websocket_trade", "dzengi"],
|
||||
OBSERVED_AT,
|
||||
OBSERVED_AT + timedelta(seconds=5),
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def test_real_trade_batch_rolls_back_preceding_insert_on_conflict(
|
||||
migrated_postgres_pool: PostgresConnectionPool,
|
||||
) -> None:
|
||||
repository = PostgresTradeRepository(
|
||||
connection_provider=migrated_postgres_pool.connection,
|
||||
)
|
||||
existing = _trade(trade_id=2, symbol="B")
|
||||
repository.store_trade(
|
||||
venue=VENUE,
|
||||
trade=existing,
|
||||
observed_at=OBSERVED_AT,
|
||||
)
|
||||
|
||||
with pytest.raises(MarketDataStorageConflictError):
|
||||
repository.store_trades(
|
||||
venue=VENUE,
|
||||
trades=(
|
||||
_trade(trade_id=1, symbol="A"),
|
||||
replace(existing, price=Decimal("999")),
|
||||
),
|
||||
observed_at=OBSERVED_AT,
|
||||
)
|
||||
|
||||
with migrated_postgres_pool.connection() as connection:
|
||||
with connection.cursor() as cursor:
|
||||
cursor.execute(
|
||||
"SELECT symbol, trade_id FROM market_data.trades"
|
||||
)
|
||||
rows = tuple(cursor.fetchall())
|
||||
|
||||
assert rows == (("B", 2),)
|
||||
|
||||
|
||||
def test_two_real_trade_writers_converge_on_one_canonical_row(
|
||||
migrated_postgres_pool: PostgresConnectionPool,
|
||||
) -> None:
|
||||
repository = PostgresTradeRepository(
|
||||
connection_provider=migrated_postgres_pool.connection,
|
||||
)
|
||||
trades = (
|
||||
_trade(source="dzengi_websocket_trade"),
|
||||
_trade(source="dzengi"),
|
||||
)
|
||||
start_barrier = threading.Barrier(2)
|
||||
caller_ids: set[int] = set()
|
||||
caller_ids_lock = threading.Lock()
|
||||
|
||||
def store(trade: Trade) -> MarketDataWriteStatus:
|
||||
with caller_ids_lock:
|
||||
caller_ids.add(threading.get_ident())
|
||||
start_barrier.wait(timeout=5.0)
|
||||
return repository.store_trade(
|
||||
venue=VENUE,
|
||||
trade=trade,
|
||||
observed_at=OBSERVED_AT,
|
||||
).status
|
||||
|
||||
with ThreadPoolExecutor(max_workers=2) as executor:
|
||||
statuses = tuple(executor.map(store, trades))
|
||||
|
||||
with migrated_postgres_pool.connection() as connection:
|
||||
with connection.cursor() as cursor:
|
||||
cursor.execute(
|
||||
"SELECT observation_sources FROM market_data.trades"
|
||||
)
|
||||
rows = tuple(cursor.fetchall())
|
||||
|
||||
assert set(statuses) == {
|
||||
MarketDataWriteStatus.INSERTED,
|
||||
MarketDataWriteStatus.PROVENANCE_UPDATED,
|
||||
}
|
||||
assert len(caller_ids) == 2
|
||||
assert len(rows) == 1
|
||||
assert set(rows[0][0]) == {"dzengi_websocket_trade", "dzengi"}
|
||||
|
||||
|
||||
def test_real_quote_and_candle_revision_provenance_and_identity(
|
||||
migrated_postgres_pool: PostgresConnectionPool,
|
||||
) -> None:
|
||||
quote_repository = PostgresQuoteRepository(
|
||||
connection_provider=migrated_postgres_pool.connection,
|
||||
)
|
||||
candle_repository = PostgresCandleRepository(
|
||||
connection_provider=migrated_postgres_pool.connection,
|
||||
)
|
||||
quote = _quote()
|
||||
candle = _candle()
|
||||
|
||||
assert quote_repository.store_quote(
|
||||
venue=VENUE,
|
||||
quote=quote,
|
||||
).status is MarketDataWriteStatus.INSERTED
|
||||
assert quote_repository.store_quote(
|
||||
venue=VENUE,
|
||||
quote=replace(quote, source="dzengi_websocket_quote"),
|
||||
).status is MarketDataWriteStatus.PROVENANCE_UPDATED
|
||||
assert candle_repository.store_candle_revision(
|
||||
venue=VENUE,
|
||||
candle=candle,
|
||||
observed_at=OBSERVED_AT,
|
||||
is_final=False,
|
||||
).status is MarketDataWriteStatus.INSERTED
|
||||
assert candle_repository.store_candle_revision(
|
||||
venue=VENUE,
|
||||
candle=replace(candle, source="secondary"),
|
||||
observed_at=OBSERVED_AT,
|
||||
is_final=False,
|
||||
).status is MarketDataWriteStatus.PROVENANCE_UPDATED
|
||||
assert candle_repository.store_candle_revision(
|
||||
venue=VENUE,
|
||||
candle=replace(candle, close_price=Decimal("106")),
|
||||
observed_at=OBSERVED_AT + timedelta(seconds=1),
|
||||
is_final=True,
|
||||
).status is MarketDataWriteStatus.INSERTED
|
||||
|
||||
with migrated_postgres_pool.connection() as connection:
|
||||
with connection.cursor() as cursor:
|
||||
cursor.execute(
|
||||
"SELECT source, observation_sources FROM market_data.quotes"
|
||||
)
|
||||
quote_rows = tuple(cursor.fetchall())
|
||||
cursor.execute(
|
||||
"""
|
||||
SELECT is_final, source, observation_sources
|
||||
FROM market_data.candle_revisions
|
||||
ORDER BY observed_at
|
||||
"""
|
||||
)
|
||||
candle_rows = tuple(cursor.fetchall())
|
||||
|
||||
assert quote_rows == (
|
||||
("dzengi", ["dzengi", "dzengi_websocket_quote"]),
|
||||
)
|
||||
assert candle_rows == (
|
||||
(False, "rest_klines:bid", ["rest_klines:bid", "secondary"]),
|
||||
(True, "rest_klines:bid", ["rest_klines:bid"]),
|
||||
)
|
||||
@@ -0,0 +1,204 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from concurrent.futures import ThreadPoolExecutor
|
||||
import threading
|
||||
|
||||
import pytest
|
||||
|
||||
from src.storage.exceptions import StorageMigrationError
|
||||
from src.storage.migrations import (
|
||||
STORAGE_MIGRATIONS,
|
||||
STORAGE_MIGRATION_ADVISORY_LOCK_ID,
|
||||
StorageMigrationRunner,
|
||||
)
|
||||
from src.storage.postgres_pool import PostgresConnectionPool
|
||||
from tests.support.postgres_market_data import (
|
||||
PostgresTestSettings,
|
||||
connect_postgres_test_database,
|
||||
count_other_test_connections,
|
||||
wait_for_postgres_advisory_lock_waiters,
|
||||
)
|
||||
|
||||
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
|
||||
def _open_pool(
|
||||
settings: PostgresTestSettings,
|
||||
*,
|
||||
name: str,
|
||||
) -> PostgresConnectionPool:
|
||||
pool = PostgresConnectionPool(
|
||||
conninfo=settings.dsn,
|
||||
min_size=1,
|
||||
max_size=2,
|
||||
timeout_seconds=5.0,
|
||||
name=name,
|
||||
)
|
||||
pool.open()
|
||||
return pool
|
||||
|
||||
|
||||
def test_real_migrations_create_expected_schema_and_are_idempotent(
|
||||
postgres_test_settings: PostgresTestSettings,
|
||||
) -> None:
|
||||
pool = _open_pool(
|
||||
postgres_test_settings,
|
||||
name="schema-integration",
|
||||
)
|
||||
runner = StorageMigrationRunner(
|
||||
connection_provider=pool.connection,
|
||||
)
|
||||
expected_versions = tuple(
|
||||
migration.version for migration in STORAGE_MIGRATIONS
|
||||
)
|
||||
|
||||
try:
|
||||
assert runner.run() == expected_versions
|
||||
assert runner.run() == ()
|
||||
|
||||
with pool.connection() as connection:
|
||||
with connection.cursor() as cursor:
|
||||
cursor.execute(
|
||||
"""
|
||||
SELECT version, name
|
||||
FROM public.storage_schema_migrations
|
||||
ORDER BY version
|
||||
"""
|
||||
)
|
||||
history = tuple(cursor.fetchall())
|
||||
cursor.execute(
|
||||
"""
|
||||
SELECT child.relname
|
||||
FROM pg_catalog.pg_inherits AS inheritance
|
||||
JOIN pg_catalog.pg_class AS parent
|
||||
ON parent.oid = inheritance.inhparent
|
||||
JOIN pg_catalog.pg_namespace AS parent_namespace
|
||||
ON parent_namespace.oid = parent.relnamespace
|
||||
JOIN pg_catalog.pg_class AS child
|
||||
ON child.oid = inheritance.inhrelid
|
||||
WHERE parent_namespace.nspname = 'market_data'
|
||||
AND parent.relname IN (
|
||||
'trades',
|
||||
'quotes',
|
||||
'candle_revisions'
|
||||
)
|
||||
ORDER BY child.relname
|
||||
"""
|
||||
)
|
||||
default_partitions = tuple(
|
||||
row[0] for row in cursor.fetchall()
|
||||
)
|
||||
cursor.execute(
|
||||
"""
|
||||
SELECT parent.relname
|
||||
FROM pg_catalog.pg_partitioned_table AS partitioned
|
||||
JOIN pg_catalog.pg_class AS parent
|
||||
ON parent.oid = partitioned.partrelid
|
||||
JOIN pg_catalog.pg_namespace AS namespace
|
||||
ON namespace.oid = parent.relnamespace
|
||||
WHERE namespace.nspname = 'market_data'
|
||||
ORDER BY parent.relname
|
||||
"""
|
||||
)
|
||||
partitioned_tables = tuple(
|
||||
row[0] for row in cursor.fetchall()
|
||||
)
|
||||
finally:
|
||||
pool.close()
|
||||
|
||||
assert history == tuple(
|
||||
(migration.version, migration.name)
|
||||
for migration in STORAGE_MIGRATIONS
|
||||
)
|
||||
assert default_partitions == (
|
||||
"candle_revisions_default",
|
||||
"quotes_default",
|
||||
"trades_default",
|
||||
)
|
||||
assert partitioned_tables == (
|
||||
"candle_revisions",
|
||||
"quotes",
|
||||
"trades",
|
||||
)
|
||||
|
||||
with connect_postgres_test_database(postgres_test_settings) as control:
|
||||
assert count_other_test_connections(control) == 0
|
||||
|
||||
|
||||
def test_two_real_migration_runners_apply_each_version_once(
|
||||
postgres_test_settings: PostgresTestSettings,
|
||||
) -> None:
|
||||
pools = (
|
||||
_open_pool(postgres_test_settings, name="migration-one"),
|
||||
_open_pool(postgres_test_settings, name="migration-two"),
|
||||
)
|
||||
runners = tuple(
|
||||
StorageMigrationRunner(connection_provider=pool.connection)
|
||||
for pool in pools
|
||||
)
|
||||
expected = tuple(
|
||||
migration.version for migration in STORAGE_MIGRATIONS
|
||||
)
|
||||
start_barrier = threading.Barrier(3)
|
||||
caller_ids: set[int] = set()
|
||||
caller_ids_lock = threading.Lock()
|
||||
|
||||
def run(runner: StorageMigrationRunner) -> tuple[int, ...]:
|
||||
with caller_ids_lock:
|
||||
caller_ids.add(threading.get_ident())
|
||||
start_barrier.wait(timeout=5.0)
|
||||
return runner.run()
|
||||
|
||||
try:
|
||||
with connect_postgres_test_database(
|
||||
postgres_test_settings,
|
||||
autocommit=False,
|
||||
) as control:
|
||||
with control.cursor() as cursor:
|
||||
cursor.execute(
|
||||
"SELECT pg_advisory_xact_lock(%s)",
|
||||
(STORAGE_MIGRATION_ADVISORY_LOCK_ID,),
|
||||
)
|
||||
|
||||
with ThreadPoolExecutor(max_workers=2) as executor:
|
||||
futures = tuple(executor.submit(run, runner) for runner in runners)
|
||||
start_barrier.wait(timeout=5.0)
|
||||
|
||||
try:
|
||||
wait_for_postgres_advisory_lock_waiters(
|
||||
control,
|
||||
lock_id=STORAGE_MIGRATION_ADVISORY_LOCK_ID,
|
||||
expected_count=2,
|
||||
)
|
||||
finally:
|
||||
control.commit()
|
||||
|
||||
results = tuple(future.result(timeout=10.0) for future in futures)
|
||||
finally:
|
||||
for pool in pools:
|
||||
pool.close()
|
||||
|
||||
assert len(caller_ids) == 2
|
||||
assert set(results) == {expected, ()}
|
||||
|
||||
|
||||
def test_real_migration_history_mismatch_is_fatal(
|
||||
migrated_postgres_pool: PostgresConnectionPool,
|
||||
) -> None:
|
||||
with migrated_postgres_pool.connection() as connection:
|
||||
with connection.cursor() as cursor:
|
||||
cursor.execute(
|
||||
"""
|
||||
UPDATE public.storage_schema_migrations
|
||||
SET name = 'tampered'
|
||||
WHERE version = 1
|
||||
"""
|
||||
)
|
||||
|
||||
runner = StorageMigrationRunner(
|
||||
connection_provider=migrated_postgres_pool.connection,
|
||||
)
|
||||
|
||||
with pytest.raises(StorageMigrationError, match="name mismatch"):
|
||||
runner.run()
|
||||
@@ -0,0 +1,445 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import asyncio
|
||||
import threading
|
||||
import time
|
||||
from dataclasses import replace
|
||||
from types import SimpleNamespace
|
||||
from typing import Any, cast
|
||||
|
||||
import pytest
|
||||
from aiogram import Bot, Dispatcher
|
||||
from psycopg.conninfo import conninfo_to_dict
|
||||
|
||||
from src.bootstrap.application import (
|
||||
ApplicationComposition,
|
||||
run_application,
|
||||
)
|
||||
from src.bootstrap.market_data_storage import build_market_data_storage
|
||||
from src.bootstrap.trade_stream_runtime import (
|
||||
build_trade_stream_production_runtime,
|
||||
)
|
||||
from src.core.config import MarketDataStorageSettings, Settings
|
||||
from src.market_data.storage import (
|
||||
MarketDataStorageOperationError,
|
||||
PostgresTradeRepository,
|
||||
TradeStorageObservationSink,
|
||||
)
|
||||
from src.storage.postgres_pool import PostgresConnectionPool
|
||||
from tests.integration.market_data.acquisition.runtime.loopback_trade_exchange import (
|
||||
LoopbackHttpResponse,
|
||||
LoopbackTradeEnvironment,
|
||||
LoopbackTradeRestServer,
|
||||
LoopbackTradeWebSocketServer,
|
||||
wait_until,
|
||||
)
|
||||
from tests.support.postgres_market_data import (
|
||||
PostgresTestSettings,
|
||||
connect_postgres_test_database,
|
||||
count_other_test_connections,
|
||||
)
|
||||
from tests.support.trade_stream_runtime import (
|
||||
SYMBOL,
|
||||
assert_no_owned_tasks,
|
||||
build_runtime,
|
||||
make_settings,
|
||||
run_scenario,
|
||||
start_runtime,
|
||||
state_store_from,
|
||||
stop_runtime,
|
||||
)
|
||||
|
||||
|
||||
pytestmark = pytest.mark.integration
|
||||
|
||||
|
||||
class FakeBotSession:
|
||||
def __init__(self) -> None:
|
||||
self.close_calls = 0
|
||||
|
||||
async def close(self) -> None:
|
||||
self.close_calls += 1
|
||||
|
||||
|
||||
class ControlledDispatcher:
|
||||
def __init__(self) -> None:
|
||||
self.started = asyncio.Event()
|
||||
self.release = asyncio.Event()
|
||||
self.cancelled = asyncio.Event()
|
||||
|
||||
async def start_polling(
|
||||
self,
|
||||
bot: object,
|
||||
*,
|
||||
close_bot_session: bool,
|
||||
) -> None:
|
||||
del bot
|
||||
assert close_bot_session is False
|
||||
self.started.set()
|
||||
|
||||
try:
|
||||
await self.release.wait()
|
||||
except asyncio.CancelledError:
|
||||
self.cancelled.set()
|
||||
raise
|
||||
|
||||
|
||||
def _sink(pool: PostgresConnectionPool) -> TradeStorageObservationSink:
|
||||
return TradeStorageObservationSink(
|
||||
trade_storage=PostgresTradeRepository(
|
||||
connection_provider=pool.connection,
|
||||
),
|
||||
venue="dzengi",
|
||||
)
|
||||
|
||||
|
||||
def _fetch_trade_rows(
|
||||
pool: PostgresConnectionPool,
|
||||
) -> tuple[tuple[Any, ...], ...]:
|
||||
with pool.connection() as connection:
|
||||
with connection.cursor() as cursor:
|
||||
cursor.execute(
|
||||
"""
|
||||
SELECT trade_id, source, observation_sources
|
||||
FROM market_data.trades
|
||||
ORDER BY executed_at, trade_id
|
||||
"""
|
||||
)
|
||||
return tuple(cursor.fetchall())
|
||||
|
||||
|
||||
def _recovered_trade(
|
||||
*,
|
||||
trade_id: int,
|
||||
timestamp_ms: int,
|
||||
price: str = "64555.56",
|
||||
quantity: str = "0.003",
|
||||
) -> dict[str, object]:
|
||||
return {
|
||||
"a": trade_id,
|
||||
"p": price,
|
||||
"q": quantity,
|
||||
"T": timestamp_ms,
|
||||
"m": False,
|
||||
}
|
||||
|
||||
|
||||
def _settings_for_database(
|
||||
*,
|
||||
postgres: PostgresTestSettings,
|
||||
websocket_url: str,
|
||||
rest_base_url: str,
|
||||
) -> Settings:
|
||||
parameters = conninfo_to_dict(postgres.dsn)
|
||||
host_value = parameters.get("host") or parameters.get("hostaddr")
|
||||
|
||||
if host_value is None:
|
||||
raise AssertionError("PostgreSQL integration DSN has no host")
|
||||
|
||||
port_value = parameters.get("port")
|
||||
|
||||
return replace(
|
||||
make_settings(
|
||||
websocket_url=websocket_url,
|
||||
rest_base_url=rest_base_url,
|
||||
),
|
||||
db_host=str(host_value),
|
||||
db_port=int(str(port_value if port_value is not None else 5432)),
|
||||
db_name=str(parameters["dbname"]),
|
||||
db_user=str(parameters.get("user", "")),
|
||||
db_password=str(parameters.get("password", "")),
|
||||
market_data_storage=MarketDataStorageSettings(
|
||||
enabled=True,
|
||||
pool_min_size=1,
|
||||
pool_max_size=4,
|
||||
pool_timeout_seconds=5.0,
|
||||
),
|
||||
)
|
||||
|
||||
|
||||
def test_live_runtime_persists_before_checkpoint(
|
||||
migrated_postgres_pool: PostgresConnectionPool,
|
||||
) -> None:
|
||||
async def scenario() -> None:
|
||||
websocket = LoopbackTradeWebSocketServer()
|
||||
rest = LoopbackTradeRestServer()
|
||||
|
||||
async with LoopbackTradeEnvironment(
|
||||
websocket=websocket,
|
||||
rest=rest,
|
||||
) as environment:
|
||||
runtime = build_runtime(
|
||||
websocket_url=environment.websocket_url,
|
||||
rest_base_url=rest.base_url,
|
||||
trade_observation_sink=_sink(migrated_postgres_pool),
|
||||
)
|
||||
runtime_task: asyncio.Task[None] | None = None
|
||||
|
||||
try:
|
||||
runtime_task = await start_runtime(runtime)
|
||||
await websocket.wait_for_subscriptions(1)
|
||||
timestamp_ms = time.time_ns() // 1_000_000
|
||||
await websocket.send_trade(
|
||||
0,
|
||||
symbol=SYMBOL,
|
||||
trade_id=600,
|
||||
timestamp_ms=timestamp_ms,
|
||||
)
|
||||
|
||||
state_store = state_store_from(runtime)
|
||||
await wait_until(
|
||||
lambda: (
|
||||
state_store.contains(SYMBOL)
|
||||
and state_store.get(SYMBOL).last_trade_id == 600
|
||||
)
|
||||
)
|
||||
|
||||
rows = await asyncio.to_thread(
|
||||
_fetch_trade_rows,
|
||||
migrated_postgres_pool,
|
||||
)
|
||||
assert rows == (
|
||||
(
|
||||
600,
|
||||
"dzengi_websocket_trade",
|
||||
["dzengi_websocket_trade"],
|
||||
),
|
||||
)
|
||||
finally:
|
||||
if runtime_task is not None:
|
||||
await stop_runtime(runtime, runtime_task)
|
||||
|
||||
await assert_no_owned_tasks()
|
||||
|
||||
run_scenario(scenario())
|
||||
|
||||
|
||||
def test_reconnect_recovery_and_live_share_real_repository(
|
||||
migrated_postgres_pool: PostgresConnectionPool,
|
||||
) -> None:
|
||||
async def scenario() -> None:
|
||||
base_time_ms = time.time_ns() // 1_000_000 - 1_000
|
||||
websocket = LoopbackTradeWebSocketServer()
|
||||
rest = LoopbackTradeRestServer(
|
||||
responses=(
|
||||
LoopbackHttpResponse(
|
||||
body=[
|
||||
_recovered_trade(
|
||||
trade_id=700,
|
||||
timestamp_ms=base_time_ms,
|
||||
price="64555.55",
|
||||
quantity="0.002",
|
||||
),
|
||||
_recovered_trade(
|
||||
trade_id=701,
|
||||
timestamp_ms=base_time_ms + 100,
|
||||
)
|
||||
]
|
||||
),
|
||||
)
|
||||
)
|
||||
|
||||
async with LoopbackTradeEnvironment(
|
||||
websocket=websocket,
|
||||
rest=rest,
|
||||
) as environment:
|
||||
runtime = build_runtime(
|
||||
websocket_url=environment.websocket_url,
|
||||
rest_base_url=rest.base_url,
|
||||
trade_observation_sink=_sink(migrated_postgres_pool),
|
||||
)
|
||||
runtime_task: asyncio.Task[None] | None = None
|
||||
|
||||
try:
|
||||
runtime_task = await start_runtime(runtime)
|
||||
await websocket.wait_for_subscriptions(1)
|
||||
await websocket.send_trade(
|
||||
0,
|
||||
symbol=SYMBOL,
|
||||
trade_id=700,
|
||||
timestamp_ms=base_time_ms,
|
||||
)
|
||||
state_store = state_store_from(runtime)
|
||||
await wait_until(
|
||||
lambda: (
|
||||
state_store.contains(SYMBOL)
|
||||
and state_store.get(SYMBOL).last_trade_id == 700
|
||||
)
|
||||
)
|
||||
|
||||
await websocket.abort_connection(0)
|
||||
await websocket.wait_for_subscriptions(2)
|
||||
await rest.wait_for_requests(1)
|
||||
await wait_until(
|
||||
lambda: state_store.get(SYMBOL).last_trade_id == 701
|
||||
)
|
||||
await websocket.send_trade(
|
||||
1,
|
||||
symbol=SYMBOL,
|
||||
trade_id=702,
|
||||
timestamp_ms=base_time_ms + 200,
|
||||
)
|
||||
await wait_until(
|
||||
lambda: state_store.get(SYMBOL).last_trade_id == 702
|
||||
)
|
||||
|
||||
rows = await asyncio.to_thread(
|
||||
_fetch_trade_rows,
|
||||
migrated_postgres_pool,
|
||||
)
|
||||
assert tuple(row[0] for row in rows) == (700, 701, 702)
|
||||
assert rows[0][1] == "dzengi_websocket_trade"
|
||||
assert rows[0][2] == [
|
||||
"dzengi_websocket_trade",
|
||||
"dzengi",
|
||||
]
|
||||
assert rows[1][1] == "dzengi"
|
||||
assert rows[1][2] == ["dzengi"]
|
||||
finally:
|
||||
if runtime_task is not None:
|
||||
await stop_runtime(runtime, runtime_task)
|
||||
|
||||
await assert_no_owned_tasks()
|
||||
|
||||
run_scenario(scenario())
|
||||
|
||||
|
||||
def test_real_persistence_failure_is_fatal_and_preserves_checkpoint(
|
||||
migrated_postgres_pool: PostgresConnectionPool,
|
||||
) -> None:
|
||||
async def scenario() -> None:
|
||||
base_time_ms = time.time_ns() // 1_000_000
|
||||
websocket = LoopbackTradeWebSocketServer()
|
||||
rest = LoopbackTradeRestServer()
|
||||
|
||||
async with LoopbackTradeEnvironment(
|
||||
websocket=websocket,
|
||||
rest=rest,
|
||||
) as environment:
|
||||
runtime = build_runtime(
|
||||
websocket_url=environment.websocket_url,
|
||||
rest_base_url=rest.base_url,
|
||||
trade_observation_sink=_sink(migrated_postgres_pool),
|
||||
)
|
||||
runtime_task: asyncio.Task[None] | None = None
|
||||
|
||||
try:
|
||||
runtime_task = await start_runtime(runtime)
|
||||
await websocket.wait_for_subscriptions(1)
|
||||
await websocket.send_trade(
|
||||
0,
|
||||
symbol=SYMBOL,
|
||||
trade_id=800,
|
||||
timestamp_ms=base_time_ms,
|
||||
)
|
||||
state_store = state_store_from(runtime)
|
||||
await wait_until(
|
||||
lambda: (
|
||||
state_store.contains(SYMBOL)
|
||||
and state_store.get(SYMBOL).last_trade_id == 800
|
||||
)
|
||||
)
|
||||
|
||||
await asyncio.to_thread(migrated_postgres_pool.close)
|
||||
await websocket.send_trade(
|
||||
0,
|
||||
symbol=SYMBOL,
|
||||
trade_id=801,
|
||||
timestamp_ms=base_time_ms + 1,
|
||||
)
|
||||
|
||||
with pytest.raises(MarketDataStorageOperationError):
|
||||
await asyncio.wait_for(runtime_task, timeout=3.0)
|
||||
|
||||
assert state_store.get(SYMBOL).last_trade_id == 800
|
||||
finally:
|
||||
if runtime_task is not None and not runtime_task.done():
|
||||
await stop_runtime(runtime, runtime_task)
|
||||
|
||||
await assert_no_owned_tasks()
|
||||
|
||||
run_scenario(scenario())
|
||||
|
||||
migrated_postgres_pool.open()
|
||||
assert tuple(row[0] for row in _fetch_trade_rows(migrated_postgres_pool)) == (
|
||||
800,
|
||||
)
|
||||
|
||||
|
||||
def test_application_owns_real_storage_before_and_after_runtime(
|
||||
postgres_test_settings: PostgresTestSettings,
|
||||
) -> None:
|
||||
storage_holder: list[Any] = []
|
||||
bot_session = FakeBotSession()
|
||||
|
||||
async def scenario() -> None:
|
||||
websocket = LoopbackTradeWebSocketServer()
|
||||
rest = LoopbackTradeRestServer()
|
||||
|
||||
async with LoopbackTradeEnvironment(
|
||||
websocket=websocket,
|
||||
rest=rest,
|
||||
) as environment:
|
||||
settings = _settings_for_database(
|
||||
postgres=postgres_test_settings,
|
||||
websocket_url=environment.websocket_url,
|
||||
rest_base_url=rest.base_url,
|
||||
)
|
||||
storage = build_market_data_storage(settings)
|
||||
assert storage is not None
|
||||
storage_holder.append(storage)
|
||||
runtime = build_trade_stream_production_runtime(
|
||||
settings,
|
||||
trade_observation_sink=storage.trade_observation_sink,
|
||||
)
|
||||
assert runtime is not None
|
||||
dispatcher = ControlledDispatcher()
|
||||
bot = SimpleNamespace(session=bot_session)
|
||||
application_task = asyncio.create_task(
|
||||
run_application(
|
||||
ApplicationComposition(
|
||||
bot=cast(Bot, cast(object, bot)),
|
||||
dispatcher=cast(
|
||||
Dispatcher,
|
||||
cast(object, dispatcher),
|
||||
),
|
||||
trade_stream_runtime=runtime,
|
||||
market_data_storage_lifecycle=storage.lifecycle,
|
||||
)
|
||||
)
|
||||
)
|
||||
|
||||
await dispatcher.started.wait()
|
||||
assert storage.connection_pool.is_open is True
|
||||
await websocket.wait_for_subscriptions(1)
|
||||
await websocket.send_trade(
|
||||
0,
|
||||
symbol=SYMBOL,
|
||||
trade_id=900,
|
||||
timestamp_ms=time.time_ns() // 1_000_000,
|
||||
)
|
||||
state_store = state_store_from(runtime)
|
||||
await wait_until(
|
||||
lambda: (
|
||||
state_store.contains(SYMBOL)
|
||||
and state_store.get(SYMBOL).last_trade_id == 900
|
||||
)
|
||||
)
|
||||
dispatcher.release.set()
|
||||
await asyncio.wait_for(application_task, timeout=5.0)
|
||||
|
||||
assert storage.connection_pool.is_open is False
|
||||
assert bot_session.close_calls == 1
|
||||
|
||||
await assert_no_owned_tasks()
|
||||
|
||||
run_scenario(scenario())
|
||||
|
||||
assert storage_holder
|
||||
with connect_postgres_test_database(postgres_test_settings) as control:
|
||||
with control.cursor() as cursor:
|
||||
cursor.execute(
|
||||
"SELECT trade_id FROM market_data.trades ORDER BY trade_id"
|
||||
)
|
||||
assert tuple(row[0] for row in cursor.fetchall()) == (900,)
|
||||
assert count_other_test_connections(control) == 0
|
||||
Reference in New Issue
Block a user