feat: Add new fields and crud methods for Position

web_app/alembic/versions/1a6fada80369_add_postion_columns.py

@@ -0,0 +1,59 @@
+"""
+Add new columns to the `position` table.
+
+Revision ID: 1a6fada80369
+Revises: 0537a9a5e841
+Create Date: 2024-11-25 07:34:11.693095
+
+This migration introduces the following changes to the `position` table:
+- Adds `is_protection` (Boolean): Indicates whether the position has protection enabled.
+- Adds `liquidation_bonus` (Float): Represents any bonus applied during the liquidation process.
+- Adds `is_liquidated` (Boolean): Marks whether the position has been liquidated.
+- Adds `datetime_liquidation` (DateTime): Stores the timestamp of when the position was liquidated.
+These changes enhance the functionality of the `position` table by
+allowing better tracking of liquidation-related events and attributes.
+"""
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# Revision identifiers, used by Alembic.
+revision = "1a6fada80369"
+down_revision = "0537a9a5e841"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    """
+    Apply the upgrade.
+
+    This function adds four new columns to the `position` table:
+    - `is_protection`: A boolean field that indicates whether protection is enabled.
+    - `liquidation_bonus`: A float field to store any bonuses applied during liquidation.
+    - `is_liquidated`: A boolean field to indicate if the position has been liquidated.
+    - `datetime_liquidation`: A datetime field to record when the liquidation occurred.
+    """
+    op.add_column("position", sa.Column("is_protection", sa.Boolean(), nullable=True))
+    op.add_column("position", sa.Column("liquidation_bonus", sa.Float(), nullable=True))
+    op.add_column("position", sa.Column("is_liquidated", sa.Boolean(), nullable=True))
+    op.add_column(
+        "position", sa.Column("datetime_liquidation", sa.DateTime(), nullable=False)
+    )
+
+
+def downgrade() -> None:
+    """
+    Revert the upgrade.
+
+    This function removes the four columns added to the `position` table:
+    - `is_protection`
+    - `liquidation_bonus`
+    - `is_liquidated`
+    - `datetime_liquidation`
+    """
+    op.drop_column("position", "datetime_liquidation")
+    op.drop_column("position", "is_liquidated")
+    op.drop_column("position", "liquidation_bonus")
+    op.drop_column("position", "is_protection")

web_app/db/crud.py

@@ -472,6 +472,69 @@ def save_current_price(self, position: Position, price_dict: dict) -> None:
             self.write_to_db(position)
         except SQLAlchemyError as e:
             logger.error(f"Error while saving current_price for position: {e}")
+
+    def liquidate_position(self, position_id: int) -> bool:
+        """
+        Marks a position as liquidated by setting `is_liquidated` to True
+        and updating `datetime_liquidation` to the current timestamp.
+
+        :param position_id: ID of the position to be liquidated.
+        :return: True if the update was successful, False otherwise.
+        """
+        with self.Session() as db:
+            try:
+                # Fetch the position by ID
+                position = db.query(Position).filter(Position.id == position_id).first()
+
+                if not position:
+                    logger.warning(f"Position with ID {position_id} not found.")
+                    return False
+
+                position.is_liquidated = True
+                position.datetime_liquidation = datetime.now()
+
+                self.write_to_db(position)
+                logger.info(f"Position {position_id} successfully liquidated.")
+                return True
+
+            except SQLAlchemyError as e:
+                logger.error(f"Error liquidating position {position_id}: {str(e)}")
+                db.rollback()
+                return False
+
+    def get_all_liquidated_positions(self) -> list[dict]:
+        """
+        Retrieves all positions where `is_liquidated` is True.
+
+        :return: A list of dictionaries containing the liquidated positions.
+        """
+        with self.Session() as db:
+            try:
+                liquidated_positions = (
+                    db.query(Position)
+                    .filter(Position.is_liquidated == True)
+                    .all()
+                ).scalar()
+
+                # Convert ORM objects to dictionaries for return
+                return [
+                    {
+                        "user_id": position.user_id,
+                        "token_symbol": position.token_symbol,
+                        "amount": position.amount,
+                        "multiplier": position.multiplier,
+                        "created_at": position.created_at,
+                        "status": position.status.value,
+                        "start_price": position.start_price,
+                        "is_liquidated": position.is_liquidated,
+                        "datetime_liquidation": position.datetime_liquidation
+                    }
+                    for position in liquidated_positions
+                ]
+
+            except SQLAlchemyError as e:
+                logger.error(f"Error retrieving liquidated positions: {str(e)}")
+                return []
 
 
 class AirDropDBConnector(DBConnector):

web_app/db/models.py

@@ -17,6 +17,7 @@
     ForeignKey,
     Integer,
     String,
+    Float
 )
 from sqlalchemy.dialects.postgresql import UUID
 from sqlalchemy.sql import func
@@ -77,6 +78,10 @@ class Position(Base):
         default="pending",
     )
     start_price = Column(DECIMAL, nullable=False)
+    is_protection = Column(Boolean, default=False)
+    liquidation_bonus = Column(Float, default=0.0)
+    is_liquidated = Column(Boolean, default=False)
+    datetime_liquidation = Column(DateTime, nullable=False)
 
 
 class AirDrop(Base):