feat: add strategy prompts API and model for managing reusable prompts

python/valuecell/agents/strategy_agent/models.py

@@ -192,11 +192,14 @@ class TradingConfig(BaseModel):
         gt=0,
     )
     template_id: Optional[str] = Field(
-        default=None, description="Strategy template identifier to guide the agent"
+        default=None, description="Saved prompt template id to use for this strategy"
     )
-    custom_prompt: Optional[str] = Field(
+    prompt_text: Optional[str] = Field(
         default=None,
-        description="Optional custom prompt to customize strategy behavior",
+        description="Direct prompt text to use (overrides template_id if provided)",
+    )
+    custom_prompt: Optional[str] = Field(
+        default=None, description="Custom prompt text to use alongside prompt_text"
     )
 
     cap_factor: float = Field(

python/valuecell/agents/strategy_agent/runtime.py

@@ -1,5 +1,4 @@
 from dataclasses import dataclass
-from pathlib import Path
 from typing import Optional
 
 from valuecell.utils.uuid import generate_uuid
@@ -16,53 +15,25 @@
 from .trading_history.recorder import InMemoryHistoryRecorder
 
 
-def _make_prompt_provider(template_dir: Optional[Path] = None):
-    """Return a prompt_provider callable that builds prompts from templates.
+def _simple_prompt_provider(request: UserRequest) -> str:
+    """Return a resolved prompt text by fusing custom_prompt and prompt_text.
 
-    Behavior:
-    - If request.trading_config.template_id matches a file under templates dir
-      (try extensions .txt, .md, or exact name), the file content is used.
-    - If request.trading_config.custom_prompt is present, it is appended after
-      the template content (separated by two newlines).
-    - If neither is present, fall back to a simple generated prompt mentioning
-      the symbols.
+    Fusion logic:
+    - If custom_prompt exists, use it as base
+    - If prompt_text also exists, append it after custom_prompt
+    - If only prompt_text exists, use it
+    - Fallback: simple generated mention of symbols
     """
-    base = Path(__file__).parent / "templates" if template_dir is None else template_dir
-
-    def provider(request: UserRequest) -> str:
-        tid = request.trading_config.template_id
-        custom = request.trading_config.custom_prompt
-
-        template_text = ""
-        if tid:
-            # safe-resolve candidate files
-            candidates = [tid, f"{tid}.txt", f"{tid}.md"]
-            for name in candidates:
-                try_path = base / name
-                try:
-                    resolved = try_path.resolve()
-                    # ensure resolved path is inside base
-                    if base.resolve() in resolved.parents or resolved == base.resolve():
-                        if resolved.exists() and resolved.is_file():
-                            template_text = resolved.read_text(encoding="utf-8")
-                            break
-                except Exception:
-                    continue
-
-        parts = []
-        if template_text:
-            parts.append(template_text.strip())
-        if custom:
-            parts.append(custom.strip())
-
-        if parts:
-            return "\n\n".join(parts)
-
-        # fallback: simple generated prompt referencing symbols
-        symbols = ", ".join(request.trading_config.symbols)
-        return f"Compose trading instructions for symbols: {symbols}."
-
-    return provider
+    custom = request.trading_config.custom_prompt
+    prompt = request.trading_config.prompt_text
+    if custom and prompt:
+        return f"{prompt}\n\n{custom}"
+    elif custom:
+        return custom
+    elif prompt:
+        return prompt
+    symbols = ", ".join(request.trading_config.symbols)
+    return f"Compose trading instructions for symbols: {symbols}."
 
 
 @dataclass
@@ -141,7 +112,7 @@ def create_strategy_runtime(
         execution_gateway=execution_gateway,
         history_recorder=history_recorder,
         digest_builder=digest_builder,
-        prompt_provider=_make_prompt_provider(),
+        prompt_provider=_simple_prompt_provider,
     )
 
     return StrategyRuntime(

python/valuecell/server/api/routers/strategy_agent.py

@@ -64,6 +64,36 @@ async def create_strategy_agent(
                 # Best-effort override; continue even if config update fails
                 pass
 
+            # Prepare repository with injected session (used below and for prompt resolution)
+            repo = get_strategy_repository(db_session=db)
+
+            # If a prompt_id (previously template_id) is provided but prompt_text is empty,
+            # attempt to resolve it from the prompts table and populate trading_config.prompt_text.
+            try:
+                prompt_id = user_request.trading_config.template_id
+                if prompt_id and not user_request.trading_config.prompt_text:
+                    try:
+                        prompt_item = repo.get_prompt_by_id(prompt_id)
+                        if prompt_item is not None:
+                            # prompt_item may be an ORM object or dict-like; use attribute or key access
+                            content = prompt_item.content
+                            if content:
+                                user_request.trading_config.prompt_text = content
+                                logger.info(
+                                    "Resolved prompt_id=%s to prompt_text for strategy creation",
+                                    prompt_id,
+                                )
+                    except Exception:
+                        logger.exception(
+                            "Failed to load prompt for prompt_id=%s; continuing without resolved prompt",
+                            prompt_id,
+                        )
+            except Exception:
+                # Defensive: any unexpected error here should not block strategy creation
+                logger.exception(
+                    "Unexpected error while resolving prompt_id before strategy creation"
+                )
+
             query = user_request.model_dump_json()
 
             agent_name = "StrategyAgent"

python/valuecell/server/api/routers/strategy_api.py

@@ -8,6 +8,7 @@
 
 from .strategy import create_strategy_router
 from .strategy_agent import create_strategy_agent_router
+from .strategy_prompts import create_strategy_prompts_router
 
 
 def create_strategy_api_router() -> APIRouter:
@@ -19,4 +20,7 @@ def create_strategy_api_router() -> APIRouter:
     # Include StrategyAgent endpoints (prefix: /strategies)
     router.include_router(create_strategy_agent_router())
 
+    # Include strategy prompts endpoints (prefix: /strategies/prompts)
+    router.include_router(create_strategy_prompts_router())
+
     return router

python/valuecell/server/api/routers/strategy_prompts.py

@@ -0,0 +1,69 @@
+"""Strategy Prompts API Router
+
+Provides minimal endpoints to list and create strategy prompts.
+Design: simple, no versioning, permissions, or pagination.
+"""
+
+from fastapi import APIRouter, Depends, HTTPException
+from sqlalchemy.orm import Session
+
+from valuecell.server.api.schemas.base import SuccessResponse
+from valuecell.server.api.schemas.strategy import (
+    PromptCreateRequest,
+    PromptCreateResponse,
+    PromptItem,
+    PromptListResponse,
+)
+from valuecell.server.db import get_db
+from valuecell.server.db.repositories import get_strategy_repository
+
+
+def create_strategy_prompts_router() -> APIRouter:
+    router = APIRouter(
+        prefix="/strategies/prompts",
+        tags=["strategies"],  # keep under strategy namespace
+        responses={404: {"description": "Not found"}},
+    )
+
+    @router.get(
+        "/",
+        response_model=PromptListResponse,
+        summary="List strategy prompts",
+        description="Return all available strategy prompts (unordered by recency).",
+    )
+    async def list_prompts(db: Session = Depends(get_db)) -> PromptListResponse:
+        try:
+            repo = get_strategy_repository(db_session=db)
+            items = repo.list_prompts()
+            prompt_items = [PromptItem(**p.to_dict()) for p in items]
+            return SuccessResponse.create(
+                data=prompt_items, msg=f"Fetched {len(prompt_items)} prompts"
+            )
+        except HTTPException:
+            raise
+        except Exception as e:  # noqa: BLE001
+            raise HTTPException(status_code=500, detail=f"Failed to list prompts: {e}")
+
+    @router.post(
+        "/create",
+        response_model=PromptCreateResponse,
+        summary="Create a strategy prompt",
+        description="Create a new strategy prompt with name and content.",
+    )
+    async def create_prompt(
+        payload: PromptCreateRequest, db: Session = Depends(get_db)
+    ) -> PromptCreateResponse:
+        try:
+            repo = get_strategy_repository(db_session=db)
+            item = repo.create_prompt(name=payload.name, content=payload.content)
+            if item is None:
+                raise HTTPException(status_code=500, detail="Failed to create prompt")
+            return SuccessResponse.create(
+                data=PromptItem(**item.to_dict()), msg="Prompt created"
+            )
+        except HTTPException:
+            raise
+        except Exception as e:  # noqa: BLE001
+            raise HTTPException(status_code=500, detail=f"Failed to create prompt: {e}")
+
+    return router

python/valuecell/server/api/schemas/strategy.py

@@ -136,3 +136,25 @@ class StrategyStatusUpdateResponse(BaseModel):
 
 
 StrategyStatusSuccessResponse = SuccessResponse[StrategyStatusUpdateResponse]
+
+
+# =====================
+# Prompt Schemas (strategy namespace)
+# =====================
+
+
+class PromptItem(BaseModel):
+    id: str = Field(..., description="Prompt UUID")
+    name: str = Field(..., description="Prompt name")
+    content: str = Field(..., description="Prompt content text")
+    created_at: Optional[datetime] = Field(None, description="Creation timestamp")
+    updated_at: Optional[datetime] = Field(None, description="Update timestamp")
+
+
+class PromptCreateRequest(BaseModel):
+    name: str = Field(..., description="Prompt name")
+    content: str = Field(..., description="Prompt content text")
+
+
+PromptListResponse = SuccessResponse[list[PromptItem]]
+PromptCreateResponse = SuccessResponse[PromptItem]

python/valuecell/server/db/models/strategy_prompt.py

@@ -0,0 +1,47 @@
+"""Strategy Prompt Model
+
+Minimal table to store reusable strategy prompt texts.
+No versioning, ownership, or permissions at this stage.
+"""
+
+import uuid
+from typing import Any, Dict
+
+from sqlalchemy import Column, DateTime, String, Text
+from sqlalchemy.sql import func
+
+from .base import Base
+
+
+class StrategyPrompt(Base):
+    """Reusable prompt text for strategies."""
+
+    __tablename__ = "strategy_prompts"
+
+    id = Column(
+        String(100), primary_key=True, default=lambda: "prompt-" + str(uuid.uuid4())
+    )
+    name = Column(String(200), nullable=False, comment="Prompt name (display)")
+    content = Column(Text, nullable=False, comment="Full prompt text")
+
+    created_at = Column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False
+    )
+    updated_at = Column(
+        DateTime(timezone=True),
+        server_default=func.now(),
+        onupdate=func.now(),
+        nullable=False,
+    )
+
+    def __repr__(self) -> str:  # pragma: no cover - debug aid
+        return f"<StrategyPrompt(id={self.id}, name='{self.name}')>"
+
+    def to_dict(self) -> Dict[str, Any]:
+        return {
+            "id": str(self.id),
+            "name": self.name,
+            "content": self.content,
+            "created_at": self.created_at.isoformat() if self.created_at else None,
+            "updated_at": self.updated_at.isoformat() if self.updated_at else None,
+        }

python/valuecell/server/db/repositories/strategy_repository.py

@@ -16,6 +16,7 @@
 from ..models.strategy_detail import StrategyDetail
 from ..models.strategy_holding import StrategyHolding
 from ..models.strategy_portfolio import StrategyPortfolioView
+from ..models.strategy_prompt import StrategyPrompt
 
 
 class StrategyRepository:
@@ -315,6 +316,60 @@ def get_details(
             if not self.db_session:
                 session.close()
 
+    # Prompts operations (kept under strategy namespace)
+    def list_prompts(self) -> List[StrategyPrompt]:
+        """Return all prompts ordered by updated_at desc."""
+        session = self._get_session()
+        try:
+            items = (
+                session.query(StrategyPrompt)
+                .order_by(StrategyPrompt.updated_at.desc())
+                .all()
+            )
+            for item in items:
+                session.expunge(item)
+            return items
+        finally:
+            if not self.db_session:
+                session.close()
+
+    def create_prompt(self, name: str, content: str) -> Optional[StrategyPrompt]:
+        """Create a new prompt."""
+        session = self._get_session()
+        try:
+            item = StrategyPrompt(name=name, content=content)
+            session.add(item)
+            session.commit()
+            session.refresh(item)
+            session.expunge(item)
+            return item
+        except Exception:
+            session.rollback()
+            return None
+        finally:
+            if not self.db_session:
+                session.close()
+
+    def get_prompt_by_id(self, prompt_id: str) -> Optional[StrategyPrompt]:
+        """Fetch one prompt by UUID string."""
+        session = self._get_session()
+        try:
+            try:
+                # Rely on DB to cast string to UUID
+                item = (
+                    session.query(StrategyPrompt)
+                    .filter(StrategyPrompt.id == prompt_id)
+                    .first()
+                )
+            except Exception:
+                item = None
+            if item:
+                session.expunge(item)
+            return item
+        finally:
+            if not self.db_session:
+                session.close()
+
 
 # Global repository instance
 _strategy_repository: Optional[StrategyRepository] = None