Liquidity

backend/alembic/env.py

@@ -8,6 +8,7 @@
 
 # Import model modules so metadata is fully populated.
 from app.models import api_key, dispute, htlc, order, swap, transaction, user  # noqa: F401
+from app.models import pool as pool_models  # noqa: F401
 
 config = context.config
 

backend/alembic/versions/20260601_0005_add_order_amendment_columns.py

@@ -0,0 +1,38 @@
+"""Add amendment_count and amendment_log to swap_orders."""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+revision: str = "20260601_0005"
+down_revision: Union[str, None] = "20260530_0004"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "swap_orders",
+        sa.Column(
+            "amendment_count",
+            sa.Integer(),
+            nullable=False,
+            server_default="0",
+        ),
+    )
+    op.add_column(
+        "swap_orders",
+        sa.Column(
+            "amendment_log",
+            postgresql.JSONB(astext_type=sa.Text()),
+            nullable=False,
+            server_default="[]",
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("swap_orders", "amendment_log")
+    op.drop_column("swap_orders", "amendment_count")

backend/alembic/versions/20260601_0006_add_advanced_order_conditions.py

@@ -0,0 +1,31 @@
+"""Add trigger_price and valid_from to swap_orders for advanced order conditions."""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+revision: str = "20260601_0006"
+down_revision: Union[str, None] = "20260601_0005"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "swap_orders",
+        sa.Column("trigger_price", sa.Numeric(precision=36, scale=18), nullable=True),
+    )
+    op.add_column(
+        "swap_orders",
+        sa.Column("valid_from", sa.BigInteger(), nullable=True),
+    )
+    op.create_index(
+        op.f("ix_swap_orders_valid_from"), "swap_orders", ["valid_from"], unique=False
+    )
+
+
+def downgrade() -> None:
+    op.drop_index(op.f("ix_swap_orders_valid_from"), table_name="swap_orders")
+    op.drop_column("swap_orders", "valid_from")
+    op.drop_column("swap_orders", "trigger_price")

backend/alembic/versions/20260601_0007_add_pool_health_tables.py

@@ -0,0 +1,82 @@
+"""Add liquidity_pool_snapshots and liquidity_pool_health tables for reserve drift monitoring (#507)."""
+
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+revision: str = "20260601_0007"
+down_revision: Union[str, None] = "20260601_0006"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "liquidity_pool_snapshots",
+        sa.Column(
+            "id",
+            postgresql.UUID(as_uuid=True),
+            primary_key=True,
+            server_default=sa.text("gen_random_uuid()"),
+        ),
+        sa.Column("pool_id", sa.BigInteger(), nullable=False),
+        sa.Column("asset_a", sa.String(), nullable=False),
+        sa.Column("asset_b", sa.String(), nullable=False),
+        sa.Column("reserve_a", sa.BigInteger(), nullable=False),
+        sa.Column("reserve_b", sa.BigInteger(), nullable=False),
+        sa.Column("ratio", sa.Float(), nullable=True),
+        sa.Column(
+            "captured_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+        sa.Column(
+            "created_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+        ),
+        sa.Column(
+            "updated_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            onupdate=sa.text("now()"),
+        ),
+    )
+    op.create_index("ix_liquidity_pool_snapshots_pool_id", "liquidity_pool_snapshots", ["pool_id"])
+    op.create_index(
+        "ix_liquidity_pool_snapshots_captured_at", "liquidity_pool_snapshots", ["captured_at"]
+    )
+    op.create_index(
+        "ix_pool_snapshots_pool_captured",
+        "liquidity_pool_snapshots",
+        ["pool_id", "captured_at"],
+    )
+
+    op.create_table(
+        "liquidity_pool_health",
+        sa.Column("pool_id", sa.BigInteger(), primary_key=True),
+        sa.Column("asset_a", sa.String(), nullable=False),
+        sa.Column("asset_b", sa.String(), nullable=False),
+        sa.Column("reserve_a", sa.BigInteger(), nullable=False),
+        sa.Column("reserve_b", sa.BigInteger(), nullable=False),
+        sa.Column("ratio", sa.Float(), nullable=True),
+        sa.Column("ratio_drift_pct", sa.Float(), nullable=True),
+        sa.Column("status", sa.String(), nullable=False, server_default="healthy"),
+        sa.Column(
+            "last_checked_at",
+            sa.DateTime(timezone=True),
+            server_default=sa.text("now()"),
+            nullable=False,
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("liquidity_pool_health")
+    op.drop_index("ix_pool_snapshots_pool_captured", table_name="liquidity_pool_snapshots")
+    op.drop_index("ix_liquidity_pool_snapshots_captured_at", table_name="liquidity_pool_snapshots")
+    op.drop_index("ix_liquidity_pool_snapshots_pool_id", table_name="liquidity_pool_snapshots")
+    op.drop_table("liquidity_pool_snapshots")

backend/app/models/__init__.py

@@ -8,3 +8,4 @@
 from .user import User
 from .asset import Asset
 from .protocol import GovernanceProposal, ReferralCampaign, ReferralReward
+from .pool import LiquidityPoolSnapshot, LiquidityPoolHealth

backend/app/models/order.py

@@ -1,6 +1,6 @@
 import uuid
-from sqlalchemy import Column, String, BigInteger, Integer, ForeignKey
-from sqlalchemy.dialects.postgresql import UUID
+from sqlalchemy import Column, String, BigInteger, Integer, Numeric, ForeignKey
+from sqlalchemy.dialects.postgresql import UUID, JSONB
 from .base import Base, TimestampMixin
 
 
@@ -27,3 +27,7 @@ class SwapOrder(Base, TimestampMixin):
     expiry = Column(BigInteger, nullable=False, index=True)
     status = Column(String, nullable=False, default="open", index=True)
     counterparty = Column(String, nullable=True)
+    amendment_count = Column(Integer, nullable=False, default=0)
+    amendment_log = Column(JSONB, nullable=False, default=list)
+    trigger_price = Column(Numeric(36, 18), nullable=True)
+    valid_from = Column(BigInteger, nullable=True, index=True)

backend/app/models/pool.py

@@ -0,0 +1,46 @@
+import uuid
+from sqlalchemy import Column, String, BigInteger, Float, DateTime, func, Index
+from sqlalchemy.dialects.postgresql import UUID
+from .base import Base, TimestampMixin
+
+
+class LiquidityPoolSnapshot(Base, TimestampMixin):
+    """Point-in-time record of a pool's reserves, used to track drift over time."""
+
+    __tablename__ = "liquidity_pool_snapshots"
+
+    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    pool_id = Column(BigInteger, nullable=False, index=True)
+    asset_a = Column(String, nullable=False)
+    asset_b = Column(String, nullable=False)
+    reserve_a = Column(BigInteger, nullable=False)
+    reserve_b = Column(BigInteger, nullable=False)
+    # ratio = reserve_a / reserve_b; stored as float for fast comparisons
+    ratio = Column(Float, nullable=True)
+    captured_at = Column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False, index=True
+    )
+
+    __table_args__ = (
+        Index("ix_pool_snapshots_pool_captured", "pool_id", "captured_at"),
+    )
+
+
+class LiquidityPoolHealth(Base):
+    """Latest computed health record for each tracked pool."""
+
+    __tablename__ = "liquidity_pool_health"
+
+    pool_id = Column(BigInteger, primary_key=True)
+    asset_a = Column(String, nullable=False)
+    asset_b = Column(String, nullable=False)
+    reserve_a = Column(BigInteger, nullable=False)
+    reserve_b = Column(BigInteger, nullable=False)
+    ratio = Column(Float, nullable=True)
+    # Percentage change in ratio since the oldest snapshot in the monitoring window
+    ratio_drift_pct = Column(Float, nullable=True)
+    # healthy | warning | critical
+    status = Column(String, nullable=False, default="healthy")
+    last_checked_at = Column(
+        DateTime(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False
+    )

backend/app/observability/pool_metrics.py

@@ -0,0 +1,51 @@
+"""Prometheus metrics for liquidity pool health monitoring (#507)."""
+
+from __future__ import annotations
+
+from typing import Optional
+
+from prometheus_client import Gauge
+
+POOL_RESERVE_A = Gauge(
+    "chainbridge_pool_reserve_a",
+    "Current reserve_a balance of a liquidity pool",
+    ["pool_id", "asset_a", "asset_b"],
+)
+
+POOL_RESERVE_B = Gauge(
+    "chainbridge_pool_reserve_b",
+    "Current reserve_b balance of a liquidity pool",
+    ["pool_id", "asset_a", "asset_b"],
+)
+
+POOL_RATIO = Gauge(
+    "chainbridge_pool_ratio",
+    "Current reserve_a / reserve_b ratio of a liquidity pool",
+    ["pool_id", "asset_a", "asset_b"],
+)
+
+# 0 = healthy, 1 = warning, 2 = critical
+POOL_HEALTH_STATUS = Gauge(
+    "chainbridge_pool_health_status",
+    "Health status of a liquidity pool (0=healthy, 1=warning, 2=critical)",
+    ["pool_id", "asset_a", "asset_b"],
+)
+
+_STATUS_CODES = {"healthy": 0, "warning": 1, "critical": 2}
+
+
+def update_pool_metrics(
+    pool_id: int,
+    asset_a: str,
+    asset_b: str,
+    reserve_a: int,
+    reserve_b: int,
+    ratio: Optional[float],
+    status: str,
+) -> None:
+    labels = {"pool_id": str(pool_id), "asset_a": asset_a, "asset_b": asset_b}
+    POOL_RESERVE_A.labels(**labels).set(reserve_a)
+    POOL_RESERVE_B.labels(**labels).set(reserve_b)
+    if ratio is not None:
+        POOL_RATIO.labels(**labels).set(ratio)
+    POOL_HEALTH_STATUS.labels(**labels).set(_STATUS_CODES.get(status, 2))

backend/app/routes/__init__.py

@@ -12,6 +12,7 @@
 from .assets import router as assets_router
 from .chains import router as chains_router
 from .protocol import router as protocol_router
+from .liquidity import router as liquidity_router
 
 api_router = APIRouter(prefix="/api/v1")
 api_router.include_router(htlc_router, prefix="/htlcs", tags=["HTLCs"])
@@ -27,3 +28,4 @@
 api_router.include_router(assets_router, prefix="/assets", tags=["Assets"])
 api_router.include_router(chains_router, prefix="/chains", tags=["Chains"])
 api_router.include_router(protocol_router, prefix="/protocol", tags=["Protocol"])
+api_router.include_router(liquidity_router, prefix="/liquidity", tags=["Liquidity"])