Cache leaderboard endpoints with lifecycle invalidation

quantara/web_app/api/leaderboard.py

@@ -5,9 +5,15 @@
 from web_app.db.crud.leaderboard import LeaderboardDBConnector
 from web_app.api.serializers.leaderboard import UserLeaderboardItem, TokenPositionStatistic
 from web_app.api.rate_limiter import limiter, READ_LIMIT
+from web_app.contract_tools.cache import get_cached_or_fetch
 
 router = APIRouter()
 leaderboard_db_connector = LeaderboardDBConnector()
+LEADERBOARD_CACHE_TTL_SECONDS = 30
+USER_LEADERBOARD_CACHE_NAME = ":".join(("leaderboard", "user", "top_positions"))
+POSITION_TOKEN_STATISTICS_CACHE_NAME = ":".join(
+    ("leaderboard", "position_tokens", "statistics")
+)
 
 @router.get(
     "/api/get-user-leaderboard",
@@ -21,8 +27,14 @@ async def get_user_leaderboard(request: Request) -> 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
+    async def fetch_leaderboard():
+        return leaderboard_db_connector.get_top_users_by_positions()
+
+    return await get_cached_or_fetch(
+        USER_LEADERBOARD_CACHE_NAME,
+        LEADERBOARD_CACHE_TTL_SECONDS,
+        fetch_leaderboard,
+    )
 
 
 @router.get(
@@ -38,4 +50,11 @@ async def get_position_tokens_statistic(request: Request) -> list[TokenPositionS
     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()
+    async def fetch_token_statistics():
+        return leaderboard_db_connector.get_position_token_statistics()
+
+    return await get_cached_or_fetch(
+        POSITION_TOKEN_STATISTICS_CACHE_NAME,
+        LEADERBOARD_CACHE_TTL_SECONDS,
+        fetch_token_statistics,
+    )

quantara/web_app/contract_tools/cache.py

@@ -4,6 +4,7 @@
 import os
 from typing import Awaitable, Callable, Optional
 
+import redis as redis_sync
 import redis.asyncio as redis
 
 logger = logging.getLogger(__name__)
@@ -52,3 +53,21 @@ async def get_cached_or_fetch(
         return value
     finally:
         await client.close()
+
+
+def delete_cache_pattern_sync(pattern: str) -> None:
+    """Best-effort synchronous cache invalidation for DB lifecycle hooks."""
+    client = redis_sync.Redis.from_url(_REDIS_URL, decode_responses=True)
+    try:
+        keys = list(client.scan_iter(match=pattern))
+        if keys:
+            client.delete(*keys)
+    except Exception as exc:
+        logger.warning("Cache invalidation failed for %s: %s", pattern, exc)
+    finally:
+        client.close()
+
+
+def invalidate_leaderboard_cache() -> None:
+    """Clear cached leaderboard aggregates after position lifecycle changes."""
+    delete_cache_pattern_sync("leaderboard:*")

quantara/web_app/db/crud/position.py

@@ -13,6 +13,7 @@
 from sqlalchemy.dialects.postgresql import insert
 from sqlalchemy.exc import SQLAlchemyError
 
+from web_app.contract_tools.cache import invalidate_leaderboard_cache
 from web_app.db.models import Base, ExtraDeposit, Position, Status, Transaction, User
 
 from .user import UserDBConnector
@@ -281,6 +282,7 @@ def close_position(self, position_id: uuid) -> Position | None:
             position.status = Status.CLOSED.value
             position.closed_at = datetime.now()
             self.write_to_db(position)
+            invalidate_leaderboard_cache()
         return position.status
 
     def open_position(self, position_id: uuid.UUID, current_prices: dict) -> str | None:
@@ -296,6 +298,7 @@ def open_position(self, position_id: uuid.UUID, current_prices: dict) -> str | N
             self.write_to_db(position)
             self.create_empty_claim(position.user_id)
             self.save_current_price(position, current_prices)
+            invalidate_leaderboard_cache()
             return position.status
         else:
             logger.error("db_open_position_not_found", position_id=str(position_id))
@@ -407,6 +410,7 @@ def liquidate_position(self, position_id: UUID) -> bool:
 
                 self.write_to_db(position)
                 logger.info("db_position_liquidated", position_id=str(position_id))
+                invalidate_leaderboard_cache()
                 return True
 
             except SQLAlchemyError as e:

quantara/web_app/tests/test_leaderboard_cache_static.py

@@ -0,0 +1,63 @@
+from pathlib import Path
+import unittest
+
+
+ROOT = Path(__file__).resolve().parents[1]
+LEADERBOARD_API = ROOT / "api" / "leaderboard.py"
+POSITION_CRUD = ROOT / "db" / "crud" / "position.py"
+CACHE_HELPER = ROOT / "contract_tools" / "cache.py"
+
+
+class LeaderboardCacheStaticTests(unittest.TestCase):
+    def test_leaderboard_endpoints_use_cache_names_and_ttl(self):
+        source = LEADERBOARD_API.read_text()
+
+        self.assertIn(
+            "from web_app.contract_tools.cache import get_cached_or_fetch",
+            source,
+        )
+        self.assertIn("LEADERBOARD_CACHE_TTL_SECONDS = 30", source)
+        self.assertIn("USER_LEADERBOARD_CACHE_NAME", source)
+        self.assertIn("POSITION_TOKEN_STATISTICS_CACHE_NAME", source)
+        self.assertIn(
+            '":".join(("leaderboard", "user", "top_positions"))',
+            source,
+        )
+        self.assertIn(
+            '("leaderboard", "position_tokens", "statistics")',
+            source,
+        )
+        self.assertIn("return await get_cached_or_fetch(", source)
+        self.assertIn(
+            "leaderboard_db_connector.get_top_users_by_positions()",
+            source,
+        )
+        self.assertIn(
+            "leaderboard_db_connector.get_position_token_statistics()",
+            source,
+        )
+
+    def test_position_lifecycle_invalidates_leaderboard_cache(self):
+        source = POSITION_CRUD.read_text()
+
+        self.assertIn(
+            "from web_app.contract_tools.cache import invalidate_leaderboard_cache",
+            source,
+        )
+        self.assertGreaterEqual(source.count("invalidate_leaderboard_cache()"), 3)
+        self.assertIn("def close_position", source)
+        self.assertIn("def open_position", source)
+        self.assertIn("def liquidate_position", source)
+
+    def test_cache_helper_has_sync_pattern_invalidation(self):
+        source = CACHE_HELPER.read_text()
+
+        self.assertIn("import redis as redis_sync", source)
+        self.assertIn("def delete_cache_pattern_sync(pattern: str) -> None:", source)
+        self.assertIn("client.scan_iter(match=pattern)", source)
+        self.assertIn("client.delete(*keys)", source)
+        self.assertIn('delete_cache_pattern_sync("leaderboard:*")', source)
+
+
+if __name__ == "__main__":
+    unittest.main()