Merge pull request #541 from djeck1432/add_leaderboard

Add leaderboard

web_app/api/leaderboard.py

@@ -0,0 +1,38 @@
+"""
+This module handles leaderboard-related API endpoints.
+"""
+from fastapi import APIRouter
+from web_app.db.crud.leaderboard import LeaderboardDBConnector
+from web_app.api.serializers.leaderboard import UserLeaderboardItem, TokenPositionStatistic
+
+router = APIRouter()
+leaderboard_db_connector = LeaderboardDBConnector()
+
+@router.get(
+    "/api/get-user-leaderboard",
+    tags=["Leaderboard"],
+    response_model=list[UserLeaderboardItem],
+    summary="Get user leaderboard",
+    response_description="Returns the top 10 users ordered by closed/opened positions.",
+)
+async def get_user_leaderboard() -> list[UserLeaderboardItem]:
+    """
+    Get the top 10 users ordered by closed/opened positions.
+    """
+    leaderboard_data = leaderboard_db_connector.get_top_users_by_positions()
+    return leaderboard_data
+
+
+@router.get(
+    "/api/get-position-tokens-statistic",
+    tags=["Leaderboard"],
+    response_model=list[TokenPositionStatistic],
+    summary="Get statistics of positions by token",
+    response_description="Returns statistics of opened/closed positions by token",
+)
+async def get_position_tokens_statistic() -> list[TokenPositionStatistic]:
+    """
+    This endpoint retrieves statistics about positions grouped by token symbol.
+    Returns counts of opened and closed positions for each token.
+    """
+    return leaderboard_db_connector.get_position_token_statistics()

web_app/api/main.py

@@ -19,6 +19,7 @@
 from web_app.api.telegram import router as telegram_router
 from web_app.api.user import router as user_router
 from web_app.api.vault import router as vault_router
+from web_app.api.leaderboard import router as leaderboard_router
 from web_app.contract_tools.blockchain_call import CLIENT
 from web_app.contract_tools.constants import EKUBO_MAINNET_ADDRESS
 
@@ -72,3 +73,4 @@ async def startup_event():
 app.include_router(user_router)
 app.include_router(telegram_router)
 app.include_router(vault_router)
+app.include_router(leaderboard_router)

web_app/api/serializers/leaderboard.py

@@ -1,46 +1,19 @@
 """
-This module defines the API endpoints and serializers for the leaderboard functionality.
+Serializers for leaderboard data.
 """
-from fastapi import APIRouter, Depends
-from pydantic import BaseModel
-from typing import List
-from sqlalchemy.orm import Session
-from web_app.db.crud.leaderboard import LeaderboardCRUD
-from web_app.db.session import get_db
 
-router = APIRouter()
+from pydantic import BaseModel
 
 class UserLeaderboardItem(BaseModel):
     """
-    Args:
-        db (Session): Database session dependency.
-        
-    Returns:
-        UserLeaderboardResponse: Response containing the leaderboard data.
+    Represents statistics for positions of a specific user.
     """
     wallet_id: str
     positions_number: int
 
-class UserLeaderboardResponse(BaseModel):
-    """
-    UserLeaderboardResponse is a model representing the response for a user leaderboard.
-
-    Attributes:
-        leaderboard (List[UserLeaderboardItem]): A list of user leaderboard items.
-    """
-    leaderboard: List[UserLeaderboardItem]
-
-@router.get(
-    "/api/get-user-leaderboard",
-    tags=["Leaderboard"],
-    response_model=UserLeaderboardResponse,
-    summary="Get user leaderboard",
-    response_description="Returns the top 10 users ordered by closed/opened positions.",
-)
-async def get_user_leaderboard(db: Session = Depends(get_db)) -> UserLeaderboardResponse:
+class TokenPositionStatistic(BaseModel):
     """
-    Get the top 10 users ordered by closed/opened positions.
+    Represents statistics for positions of a specific token.
     """
-    leaderboard_crud = LeaderboardCRUD(db)
-    leaderboard_data = leaderboard_crud.get_top_users_by_positions()
-    return UserLeaderboardResponse(leaderboard=leaderboard_data)
+    token_symbol: str
+    total_positions: int

web_app/db/crud/leaderborad.py → web_app/db/crud/leaderboard.py

@@ -2,26 +2,19 @@
 This module provides CRUD operations for the leaderboard, retrieving the top users by positions.
 
 """
-from sqlalchemy.orm import Session
+from .base import DBConnector
 from sqlalchemy.exc import SQLAlchemyError
 from sqlalchemy import func
 from web_app.db.models import User, Position
 import logging
 
 logger = logging.getLogger(__name__)
 
-class LeaderboardCRUD:
+class LeaderboardDBConnector(DBConnector):
     """
-    A class used to perform CRUD operations related to the leaderboard.
+    Provides database connection and operations management using SQLAlchemy
+    in a FastAPI application context.
     """
-    def __init__(self, session: Session):
-        """
-        Initializes a new instance of the class.
-
-        Args:
-            session (Session): The database session to be used for database operations.
-        """
-        self.Session = session
 
     def get_top_users_by_positions(self) -> list[dict]:
         """
@@ -50,4 +43,33 @@ def get_top_users_by_positions(self) -> list[dict]:
 
             except SQLAlchemyError as e:
                 logger.error(f"Error retrieving top users by positions: {e}")
-                return []
+                return []
+
+    def get_position_token_statistics(self) -> list[dict]:
+        """
+        Retrieves closed/opened positions groupped by token_symbol.
+        :return: List of dictionaries containing token_symbol and total_positions.
+        """
+        with self.Session() as db:
+            try:
+                results = (
+                    db.query(
+                        Position.token_symbol,
+                        func.count(Position.id).label("total_positions")
+                    )
+                    .filter(Position.status.in_(["closed", "opened"]))
+                    .group_by(Position.token_symbol)
+                    .all()
+                )
+
+                return [
+                    {
+                        "token_symbol": result.token_symbol,
+                        "total_positions": result.total_positions
+                    }
+                    for result in results
+                ]
+
+            except SQLAlchemyError as e:
+                logger.error(f"Error retrieving position token statistics: {e}")
+                return []