fixing research tool

backend/app/agents/executor.py

@@ -171,8 +171,9 @@ def _build_agent_executor(self) -> AgentExecutor:
                 name="research_token",
                 func=self._execute_research_token_wrapper,
                 description=(
-                    "Research a token for news and any market sentiments, "
-                    "Input: token symbol and recency (e.g., 'ETH 7d' for last 7 days)"
+                    "Research a token for news and market sentiment. Uses a shared cache — "
+                    "calling it is cheap, so use it whenever you need research. "
+                    "Input: token symbol and optional recency (e.g., 'ETH 7d' for last 7 days)"
                 ),
             ),
             Tool(
@@ -249,7 +250,7 @@ def _build_agent_executor(self) -> AgentExecutor:
 - Always provide reasoning in your summary
 - This is a simulation - trades are not executed on-chain
 - Before making significant trades, research tokens using the research_token tool to check recent news and sentiment
-- If recent research already exists, you may reuse it instead of researching again
+- The research_token tool uses a shared cache automatically. Just call it when you need research.
 
 TOOLS:
 ------
@@ -347,40 +348,57 @@ def _execute_trade_wrapper(self, trade_input: str) -> str:
 
     def _execute_research_token_wrapper(self, input_str: str) -> str:
         """
-        Wrapper for executing research token from LangChain tool.
+        Cache-first research wrapper.
 
-        Args:
-            input_str: "ETH" or "ETH 7d"
-
-        Returns:
-            Research summary string
+        1. Parse input -> token + recency
+        2. Check shared cache via database_tool.get_fresh_research()
+        3. Cache HIT  -> return cached summary (zero API cost)
+        4. Cache MISS -> call Perplexity, derive opinion, upsert to cache
         """
-        from .tools.research_tool import research_result_to_dict
+        from .tools.research_tool import research_result_to_dict, derive_opinion
 
         try:
             parts = input_str.split()
             if len(parts) < 1:
                 return "Error: Invalid format. Expected 'TOKEN' or 'TOKEN RECENCY'"
 
             token = parts[0].upper()
-            recency = parts[1] if len(parts) > 1 else "7d"
+            _valid_recencies = {"1d", "7d", "30d", "365d"}
+            raw_recency = parts[1] if len(parts) > 1 else "7d"
+            recency = raw_recency if raw_recency in _valid_recencies else "7d"
+
+            # ── Cache check ──────────────────────────────────────────
+            if self.database_tool:
+                cached = self._run_async(
+                    self.database_tool.get_fresh_research(token, recency)
+                )
+                if cached is not None:
+                    logger.info(f"Research cache HIT for {token} — skipping API call")
+                    return cached["summary_markdown"]
 
-            # Call sync research method
+            # ── Cache MISS — call Perplexity ─────────────────────────
             research_result = self.research_tool.research_token(
                 token_symbol=token,
                 recency=recency,
             )
 
-            # Persist result to DB (convert dataclass to dict)
-            if self.database_tool and self.agent_uuid:
+            # Derive opinion from summary
+            opinion = derive_opinion(research_result.summary_markdown)
+
+            # Persist to shared cache (skip raw_results to keep rows lean)
+            if self.database_tool:
                 result_dict = research_result_to_dict(research_result)
                 self._run_async(
-                    self.database_tool.save_research_result(
-                        agent_uuid=self.agent_uuid,
+                    self.database_tool.upsert_research_result(
+                        crypto_token=token,
                         query=f"Research {token}",
-                        result=result_dict,
+                        summary_markdown=result_dict["summary_markdown"],
+                        citations=result_dict["citations"],
                         recency=recency,
+                        provider=result_dict.get("provider", "perplexity"),
                         related_tokens=[token],
+                        agent_opinion=opinion,
+                        last_researched_by=self.agent_uuid,
                     )
                 )
 

backend/app/agents/tools/__init__.py

@@ -17,6 +17,7 @@
 from .tweet_post_tool import TweetPostTool
 from .database_tool import DatabaseTool
 from .plan_tool import PlanTool
+from .research_tool import ResearchTool
 
 # Deprecated: These tools are kept for backwards compatibility
 # Use MakeTradeTool instead for simulated trading
@@ -29,6 +30,7 @@
     "TweetPostTool",
     "DatabaseTool",
     "PlanTool",
+    "ResearchTool",
     # Deprecated
     "TradeTool",
     "PortfolioTool",

backend/app/agents/tools/database_tool.py

@@ -8,18 +8,31 @@
 import logging
 from typing import Optional, Dict, Any, List
 from uuid import UUID
-from datetime import datetime, timezone, timedelta
+from datetime import datetime, timezone
 from decimal import Decimal
 
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy import select, update
 from sqlalchemy.dialects.postgresql import insert
 
-from ...db.models import Agent, AgentState, Trade, Tournament, ActionEnum, PlanActionEnum, PlanStatusEnum, PlanItem
+from ...db.models import Agent, AgentState, Trade, Tournament, ActionEnum, PlanActionEnum, PlanStatusEnum, PlanItem, AgentResearchArtifact
 from ..data_classes import Trade as TradeData, Portfolio
 
 logger = logging.getLogger(__name__)
 
+# Freshness thresholds per recency tier (seconds)
+FRESHNESS_THRESHOLDS = {
+    "1d":   2 * 3600,    # 2 hours
+    "7d":   6 * 3600,    # 6 hours
+    "30d":  24 * 3600,   # 24 hours
+    "365d": 72 * 3600,   # 72 hours
+}
+DEFAULT_FRESHNESS = 6 * 3600  # 6 hours
+
+# Tightness ordering — lower = tighter (more recent Perplexity search window).
+# A cache row fetched with "7d" should NOT satisfy a "1d" request.
+_RECENCY_TIGHTNESS = {"1d": 0, "7d": 1, "30d": 2, "365d": 3}
+
 
 class DatabaseTool:
     """
@@ -402,72 +415,139 @@ async def create_agent_if_not_exists(
             logger.error(f"Failed to create agent: {e}")
             raise
 
-    async def save_research_result(
-            self, 
-            agent_uuid: UUID, 
-            query: str, 
-            result: Dict, 
-            recency: Optional[str] = None, 
-            related_tokens: Optional[List[str]] = None
-    ) -> None: 
-        #Saving the research to Postgres (agent_research_artifact table)
+    # ── Shared research cache ──────────────────────────────────────────
+
+    async def get_fresh_research(
+        self, crypto_token: str, recency: str = "7d"
+    ) -> Optional[Dict]:
+        """
+        Return cached research for *crypto_token* if it is still fresh.
+
+        Freshness is determined by comparing the row's `updated_at` against
+        the threshold for the requested *recency* tier.
+
+        Returns:
+            Dict with cached research data if fresh, None if stale/missing.
+        """
         try:
-            stmt = insert(AgentResearchArtifact).values(
-                agent_id=agent_uuid,
-                query=query,
-                recency=recency,
-                provider=result.get("provider", "perplexity"),
-                summary_markdown=result["summary_markdown"],
-                citations=result.get("citations", []),
-                raw_results=result.get("raw_results", {}),
-                related_tokens=related_tokens or [],
-                created_at=datetime.now(timezone.utc),
+            stmt = select(AgentResearchArtifact).where(
+                AgentResearchArtifact.crypto_token == crypto_token
             )
-            await self.session.execute(stmt)
-            await self.session.commit() 
-            logger.info(f"Research result saved for agent={agent_uuid}, query='{query}'")
+            result = await self.session.execute(stmt)
+            artifact = result.scalar_one_or_none()
+
+            if artifact is None:
+                logger.info(f"Research cache MISS for {crypto_token} (no row)")
+                return None
+
+            # Tightness check: cached "7d" data can't satisfy a "1d" request
+            cached_tight = _RECENCY_TIGHTNESS.get(artifact.recency, 99)
+            requested_tight = _RECENCY_TIGHTNESS.get(recency, 1)
+            if cached_tight > requested_tight:
+                logger.info(
+                    f"Research cache MISS for {crypto_token} "
+                    f"(cached recency={artifact.recency} too loose for requested={recency})"
+                )
+                return None
+
+            age_seconds = (
+                datetime.now(timezone.utc) - artifact.updated_at
+            ).total_seconds()
+            threshold = FRESHNESS_THRESHOLDS.get(recency, DEFAULT_FRESHNESS)
+
+            if age_seconds > threshold:
+                logger.info(
+                    f"Research cache STALE for {crypto_token} "
+                    f"(age={age_seconds:.0f}s, threshold={threshold}s)"
+                )
+                return None
+
+            logger.info(
+                f"Research cache HIT for {crypto_token} "
+                f"(age={age_seconds:.0f}s, threshold={threshold}s)"
+            )
+            return {
+                "crypto_token": artifact.crypto_token,
+                "query": artifact.query,
+                "recency": artifact.recency,
+                "provider": artifact.provider,
+                "summary_markdown": artifact.summary_markdown,
+                "citations": artifact.citations,
+                "related_tokens": artifact.related_tokens,
+                "agent_opinion": artifact.agent_opinion,
+                "updated_at": artifact.updated_at.isoformat(),
+            }
+
         except Exception as e:
-            await self.session.rollback()
-            logger.error(f"Failed to save research result: {e}")
-            raise 
-
-    async def get_recent_research_results(
-            self,
-            agent_uuid: UUID,
-            token: Optional[str] = None,
-            days: int = 7
-    ) -> List[Dict]: 
-        #Fetch the recent research data for an agent from Postgres
+            logger.error(f"Failed to check research cache: {e}")
+            raise
+
+    async def upsert_research_result(
+        self,
+        crypto_token: str,
+        query: str,
+        summary_markdown: str,
+        citations: List[Dict],
+        recency: Optional[str] = None,
+        provider: str = "perplexity",
+        raw_results: Optional[Dict] = None,
+        related_tokens: Optional[List[str]] = None,
+        agent_opinion: Optional[str] = None,
+        last_researched_by: Optional[UUID] = None,
+    ) -> None:
+        """
+        Insert or update the shared research row for *crypto_token*.
+
+        Uses INSERT ... ON CONFLICT (crypto_token) DO UPDATE so every token
+        has exactly one row that gets refreshed in place.
+        """
+        now = datetime.now(timezone.utc)
         try:
-            cutoff_date = datetime.now(timezone.utc) - timedelta(days=days)
-            stmt = select(AgentResearchArtifact).where(
-                AgentResearchArtifact.agent_id == agent_uuid,
-                AgentResearchArtifact.created_at >= cutoff_date
+            stmt = (
+                insert(AgentResearchArtifact)
+                .values(
+                    crypto_token=crypto_token,
+                    last_researched_by=last_researched_by,
+                    query=query,
+                    recency=recency,
+                    provider=provider,
+                    summary_markdown=summary_markdown,
+                    citations=citations,
+                    raw_results=raw_results,
+                    related_tokens=related_tokens,
+                    agent_opinion=agent_opinion,
+                    created_at=now,
+                    updated_at=now,
+                )
+                .on_conflict_do_update(
+                    index_elements=["crypto_token"],
+                    set_={
+                        "last_researched_by": last_researched_by,
+                        "query": query,
+                        "recency": recency,
+                        "provider": provider,
+                        "summary_markdown": summary_markdown,
+                        "citations": citations,
+                        "raw_results": raw_results,
+                        "related_tokens": related_tokens,
+                        "agent_opinion": agent_opinion,
+                        "updated_at": now,
+                    },
+                )
             )
-            if token:
-                stmt = stmt.where(AgentResearchArtifact.related_tokens.contains([token]))
-
-            result = await self.session.execute(stmt)
-            artifacts = result.scalars().all()
-
-            research_results = []
-            for artifact in artifacts:
-                research_results.append({
-                    "query": artifact.query,
-                    "recency": artifact.recency,
-                    "provider": artifact.provider,
-                    "summary_markdown": artifact.summary_markdown,
-                    "citations": artifact.citations,
-                    "raw_results": artifact.raw_results,
-                    "created_at": artifact.created_at.isoformat(),
-                })
-
-            logger.info(f"Retrieved {len(research_results)} research results for agent={agent_uuid}")
-            return research_results
-
+
+            await self.session.execute(stmt)
+            await self.session.commit()
+            logger.info(
+                f"Research upserted for {crypto_token} "
+                f"(by={last_researched_by}, opinion={agent_opinion})"
+            )
+
         except Exception as e:
-            logger.error(f"Failed to retrieve research results: {e}")
+            await self.session.rollback()
+            logger.error(f"Failed to upsert research result: {e}")
             raise
+
     async def create_plan_item(self, agent_uuid: UUID, tournament_uuid: UUID, action_type: PlanActionEnum,
                                 execute_at: datetime, payload, idempotency_key: str, max_attempts: int = 3) -> PlanItem:
         stmt = (insert(PlanItem).values(agent_id = agent_uuid, 
@@ -526,7 +606,7 @@ async def list_plan_items(self, agent_uuid: UUID, tournament_uuid: UUID,
     async def cancel_plan_item(self, plan_item_id: UUID, reason: str | None = None) -> None:
         values = {
         "status": PlanStatusEnum.cancelled,
-        "updated_at": datetime.utcnow(),
+        "updated_at": datetime.now(timezone.utc),
         }
 
         if reason:
@@ -553,7 +633,7 @@ async def reschedule_plan_item(self, plan_item_id: UUID, new_execute_at: datetim
             )
             .values(
                 execute_at=new_execute_at,
-                updated_at=datetime.utcnow(),
+                updated_at=datetime.now(timezone.utc),
             )
             .returning(PlanItem)
         )