Feat/improve db management

database/connection.py

@@ -64,6 +64,11 @@ async def _run_migrations(self, conn):
                 "executors", "controller_id",
                 "ALTER TABLE executors ADD COLUMN controller_id TEXT NOT NULL DEFAULT 'main'"
             ),
+            # Add error_log to executors table for storing errors on failed executors
+            (
+                "executors", "error_log",
+                "ALTER TABLE executors ADD COLUMN error_log TEXT"
+            ),
         ]
         for table, column, sql in migrations:
             try:

database/models.py

@@ -1,4 +1,4 @@
-from sqlalchemy import TIMESTAMP, Column, ForeignKey, Integer, Numeric, String, Text, func
+from sqlalchemy import TIMESTAMP, Column, ForeignKey, Integer, Numeric, String, Text, UniqueConstraint, func
 from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy.orm import relationship
 
@@ -12,7 +12,7 @@ class AccountState(Base):
     timestamp = Column(TIMESTAMP(timezone=True), server_default=func.now(), nullable=False, index=True)
     account_name = Column(String, nullable=False, index=True)
     connector_name = Column(String, nullable=False, index=True)
-    
+
     token_states = relationship("TokenState", back_populates="account_state", cascade="all, delete-orphan")
 
 
@@ -26,142 +26,143 @@ class TokenState(Base):
     price = Column(Numeric(precision=30, scale=18), nullable=False)
     value = Column(Numeric(precision=30, scale=18), nullable=False)
     available_units = Column(Numeric(precision=30, scale=18), nullable=False)
-    
+
     account_state = relationship("AccountState", back_populates="token_states")
 
 
 class Order(Base):
     __tablename__ = "orders"
-    
+
     id = Column(Integer, primary_key=True, index=True)
     # Order identification
     client_order_id = Column(String, nullable=False, unique=True, index=True)
     exchange_order_id = Column(String, nullable=True, index=True)
-    
+
     # Timestamps
     created_at = Column(TIMESTAMP(timezone=True), server_default=func.now(), nullable=False, index=True)
     updated_at = Column(TIMESTAMP(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False)
-    
+
     # Account and connector info
     account_name = Column(String, nullable=False, index=True)
     connector_name = Column(String, nullable=False, index=True)
-    
+
     # Order details
     trading_pair = Column(String, nullable=False, index=True)
     trade_type = Column(String, nullable=False)  # BUY, SELL
     order_type = Column(String, nullable=False)  # LIMIT, MARKET, LIMIT_MAKER
     amount = Column(Numeric(precision=30, scale=18), nullable=False)
     price = Column(Numeric(precision=30, scale=18), nullable=True)  # Null for market orders
-    
+
     # Order status and execution
-    status = Column(String, nullable=False, default="SUBMITTED", index=True)  # SUBMITTED, OPEN, FILLED, CANCELLED, FAILED
+    status = Column(String, nullable=False, default="SUBMITTED",
+                    index=True)  # SUBMITTED, OPEN, FILLED, CANCELLED, FAILED
     filled_amount = Column(Numeric(precision=30, scale=18), nullable=False, default=0)
     average_fill_price = Column(Numeric(precision=30, scale=18), nullable=True)
-    
+
     # Fee information
     fee_paid = Column(Numeric(precision=30, scale=18), default=0, nullable=True)
     fee_currency = Column(String, nullable=True)
-    
+
     # Additional metadata
     error_message = Column(Text, nullable=True)
-    
+
     # Relationships for future enhancements
     trades = relationship("Trade", back_populates="order", cascade="all, delete-orphan")
 
 
 class Trade(Base):
     __tablename__ = "trades"
-    
+
     id = Column(Integer, primary_key=True, index=True)
     order_id = Column(Integer, ForeignKey("orders.id"), nullable=False)
-    
+
     # Trade identification
     trade_id = Column(String, nullable=False, unique=True, index=True)
-    
+
     # Timestamps
     timestamp = Column(TIMESTAMP(timezone=True), nullable=False, index=True)
-    
+
     # Trade details
     trading_pair = Column(String, nullable=False, index=True)
     trade_type = Column(String, nullable=False)  # BUY, SELL
     amount = Column(Numeric(precision=30, scale=18), nullable=False)
     price = Column(Numeric(precision=30, scale=18), nullable=False)
-    
+
     # Fee information
     fee_paid = Column(Numeric(precision=30, scale=18), nullable=False, default=0)
     fee_currency = Column(String, nullable=True)
-    
+
     # Relationship
     order = relationship("Order", back_populates="trades")
 
 
 class PositionSnapshot(Base):
     __tablename__ = "position_snapshots"
-    
+
     id = Column(Integer, primary_key=True, index=True)
-    
+
     # Position identification
     account_name = Column(String, nullable=False, index=True)
     connector_name = Column(String, nullable=False, index=True)
     trading_pair = Column(String, nullable=False, index=True)
-    
+
     # Timestamps
     timestamp = Column(TIMESTAMP(timezone=True), server_default=func.now(), nullable=False, index=True)
-    
+
     # Real-time exchange data (from connector.account_positions)
     side = Column(String, nullable=False)  # LONG, SHORT
     exchange_size = Column(Numeric(precision=30, scale=18), nullable=False)  # Size from exchange
     entry_price = Column(Numeric(precision=30, scale=18), nullable=True)  # Average entry price
     mark_price = Column(Numeric(precision=30, scale=18), nullable=True)  # Current mark price
-    
+
     # Real-time PnL data (can't be derived from trades alone)
     unrealized_pnl = Column(Numeric(precision=30, scale=18), nullable=True)  # From exchange
     percentage_pnl = Column(Numeric(precision=10, scale=6), nullable=True)  # PnL percentage
-    
+
     # Leverage and margin info
     leverage = Column(Numeric(precision=10, scale=2), nullable=True)  # Position leverage
     initial_margin = Column(Numeric(precision=30, scale=18), nullable=True)  # Initial margin
     maintenance_margin = Column(Numeric(precision=30, scale=18), nullable=True)  # Maintenance margin
-    
+
     # Fee tracking (exchange provides cumulative data)
     cumulative_funding_fees = Column(Numeric(precision=30, scale=18), nullable=False, default=0)  # Funding fees
     fee_currency = Column(String, nullable=True)  # Fee currency (usually USDT)
-    
+
     # Reconciliation fields (calculated from our trade data)
     calculated_size = Column(Numeric(precision=30, scale=18), nullable=True)  # Size from our trades
     calculated_entry_price = Column(Numeric(precision=30, scale=18), nullable=True)  # Entry from our trades
     size_difference = Column(Numeric(precision=30, scale=18), nullable=True)  # Difference for reconciliation
-    
+
     # Additional metadata
     exchange_position_id = Column(String, nullable=True, index=True)  # Exchange position ID
     is_reconciled = Column(String, nullable=False, default="PENDING")  # RECONCILED, MISMATCH, PENDING
 
 
 class FundingPayment(Base):
     __tablename__ = "funding_payments"
-    
+
     id = Column(Integer, primary_key=True, index=True)
-    
+
     # Payment identification
     funding_payment_id = Column(String, nullable=False, unique=True, index=True)
-    
+
     # Timestamps
     timestamp = Column(TIMESTAMP(timezone=True), nullable=False, index=True)
-    
+
     # Account and connector info
     account_name = Column(String, nullable=False, index=True)
     connector_name = Column(String, nullable=False, index=True)
-    
+
     # Funding details
     trading_pair = Column(String, nullable=False, index=True)
     funding_rate = Column(Numeric(precision=20, scale=18), nullable=False)  # Funding rate
     funding_payment = Column(Numeric(precision=30, scale=18), nullable=False)  # Payment amount
     fee_currency = Column(String, nullable=False)  # Payment currency (usually USDT)
-    
+
     # Position association
     position_size = Column(Numeric(precision=30, scale=18), nullable=True)  # Position size at time of payment
     position_side = Column(String, nullable=True)  # LONG, SHORT
-    
+
     # Additional metadata
     exchange_funding_id = Column(String, nullable=True, index=True)  # Exchange funding ID
 
@@ -277,7 +278,8 @@ class GatewayCLMMPosition(Base):
 
     # Price tracking for PnL calculation
     entry_price = Column(Numeric(precision=30, scale=18), nullable=True)  # Pool price when position opened
-    current_price = Column(Numeric(precision=30, scale=18), nullable=True)  # Latest price (becomes close price when closed)
+    current_price = Column(Numeric(precision=30, scale=18),
+                           nullable=True)  # Latest price (becomes close price when closed)
 
     # Initial deposit amounts (for PnL calculation)
     initial_base_token_amount = Column(Numeric(precision=30, scale=18), nullable=True)
@@ -322,7 +324,8 @@ class GatewayCLMMEvent(Base):
     timestamp = Column(TIMESTAMP(timezone=True), server_default=func.now(), nullable=False, index=True)
 
     # Event type
-    event_type = Column(String, nullable=False, index=True)  # OPEN, ADD_LIQUIDITY, REMOVE_LIQUIDITY, COLLECT_FEES, CLOSE
+    event_type = Column(String, nullable=False,
+                        index=True)  # OPEN, ADD_LIQUIDITY, REMOVE_LIQUIDITY, COLLECT_FEES, CLOSE
 
     # Event amounts
     base_token_amount = Column(Numeric(precision=30, scale=18), nullable=True)
@@ -374,6 +377,9 @@ class ExecutorRecord(Base):
     cum_fees_quote = Column(Numeric(precision=30, scale=18), nullable=False, default=0)
     filled_amount_quote = Column(Numeric(precision=30, scale=18), nullable=False, default=0)
 
+    # Error tracking
+    error_log = Column(Text, nullable=True)  # JSON: last errors captured during execution
+
     # Configuration (JSON)
     config = Column(Text, nullable=True)
 
@@ -384,6 +390,41 @@ class ExecutorRecord(Base):
     orders = relationship("ExecutorOrder", back_populates="executor", cascade="all, delete-orphan")
 
 
+class PositionHoldRecord(Base):
+    """Database model for position hold tracking (separate from executor lifecycle)."""
+    __tablename__ = "position_holds"
+    __table_args__ = (
+        UniqueConstraint(
+            "account_name", "connector_name", "trading_pair", "controller_id",
+            name="uq_position_hold_key"
+        ),
+    )
+
+    id = Column(Integer, primary_key=True, index=True)
+
+    # Position identification
+    account_name = Column(String, nullable=False, index=True)
+    connector_name = Column(String, nullable=False, index=True)
+    trading_pair = Column(String, nullable=False, index=True)
+    controller_id = Column(String, nullable=False, default="main", index=True)
+
+    # Aggregated amounts
+    buy_amount_base = Column(Numeric(precision=30, scale=18), nullable=False, default=0)
+    buy_amount_quote = Column(Numeric(precision=30, scale=18), nullable=False, default=0)
+    sell_amount_base = Column(Numeric(precision=30, scale=18), nullable=False, default=0)
+    sell_amount_quote = Column(Numeric(precision=30, scale=18), nullable=False, default=0)
+    realized_pnl_quote = Column(Numeric(precision=30, scale=18), nullable=False, default=0)
+
+    # Tracking
+    executor_ids = Column(Text, nullable=True)  # JSON array of executor IDs
+    status = Column(String, nullable=False, default="ACTIVE", index=True)  # ACTIVE, CLEARED
+
+    # Timestamps
+    created_at = Column(TIMESTAMP(timezone=True), server_default=func.now(), nullable=False, index=True)
+    last_updated = Column(TIMESTAMP(timezone=True), server_default=func.now(), onupdate=func.now(), nullable=False)
+    cleared_at = Column(TIMESTAMP(timezone=True), nullable=True)
+
+
 class ExecutorOrder(Base):
     """Database model for orders created by executors."""
     __tablename__ = "executor_orders"
@@ -414,5 +455,3 @@ class ExecutorOrder(Base):
 
     # Relationship
     executor = relationship("ExecutorRecord", back_populates="orders")
-
-