Stability (#5)

Author: kongzii

labs_api/insights.py

@@ -1,3 +1,5 @@
+import fastapi
+from prediction_market_agent_tooling.loggers import logger
 from prediction_market_agent_tooling.markets.omen.omen_subgraph_handler import (
     HexAddress,
     OmenSubgraphHandler,
@@ -19,17 +21,28 @@ def market_insights_cached(
 
     else:
         new = market_insights(market_id)
-        cache.save(new)
+        if new.has_insights:
+            cache.save(new)
         return new
 
 
 def market_insights(market_id: HexAddress) -> MarketInsightsResponse:
     """Returns market insights for a given market on Omen."""
-    market = OmenSubgraphHandler().get_omen_market_by_market_id(market_id)
+    try:
+        market = OmenSubgraphHandler().get_omen_market_by_market_id(market_id)
+    except ValueError:
+        raise fastapi.HTTPException(
+            status_code=404, detail=f"Market with id `{market_id}` not found."
+        )
+    try:
+        insights = tavily_insights(market.question_title)
+    except Exception as e:
+        logger.error(f"Failed to get insights for market `{market_id}`: {e}")
+        insights = None
     return MarketInsightsResponse.from_tavily_response(
         market_id=market_id,
         created_at=utcnow(),
-        tavily_response=tavily_insights(market.question_title),
+        tavily_response=insights,
     )
 
 

labs_api/main.py

@@ -1,3 +1,5 @@
+import typing as t
+
 import fastapi
 import uvicorn
 from config import Config
@@ -9,6 +11,15 @@
 from labs_api.insights import MarketInsightsResponse, market_insights_cached
 from labs_api.insights_cache import MarketInsightsResponseCache
 
+HEX_ADDRESS_VALIDATOR = t.Annotated[
+    HexAddress,
+    fastapi.Query(
+        ...,
+        description="Hex address of the market on Omen.",
+        pattern="^0x[a-fA-F0-9]{40}$",
+    ),
+]
+
 
 def create_app() -> fastapi.FastAPI:
     app = fastapi.FastAPI()
@@ -28,7 +39,7 @@ def _ping() -> str:
         return "pong"
 
     @app.get("/market-insights/")
-    def _market_insights(market_id: HexAddress) -> MarketInsightsResponse:
+    def _market_insights(market_id: HEX_ADDRESS_VALIDATOR) -> MarketInsightsResponse:
         """Returns market insights for a given market on Omen."""
         insights = market_insights_cached(market_id, market_insights_cache)
         logger.info(f"Insights for `{market_id}`: {insights.model_dump()}")

labs_api/models.py

@@ -1,3 +1,4 @@
+import typing as t
 from datetime import datetime
 
 from prediction_market_agent_tooling.gtypes import HexAddress
@@ -16,23 +17,31 @@ def from_tavily_result(tavily_result: "TavilyResult") -> "MarketInsightResult":
 class MarketInsightsResponse(BaseModel):
     market_id: HexAddress
     created_at: datetime
-    summary: str
+    summary: str | None
     results: list[MarketInsightResult]
 
+    @property
+    def has_insights(self) -> bool:
+        return bool(self.summary or self.results)
+
     @staticmethod
     def from_tavily_response(
         market_id: HexAddress,
         created_at: datetime,
-        tavily_response: "TavilyResponse",
+        tavily_response: t.Union["TavilyResponse", None],
     ) -> "MarketInsightsResponse":
         return MarketInsightsResponse(
             market_id=market_id,
             created_at=created_at,
-            summary=tavily_response.answer,
-            results=[
-                MarketInsightResult.from_tavily_result(result)
-                for result in tavily_response.results
-            ],
+            summary=tavily_response.answer if tavily_response else None,
+            results=(
+                [
+                    MarketInsightResult.from_tavily_result(result)
+                    for result in tavily_response.results
+                ]
+                if tavily_response
+                else []
+            ),
         )