From d81617961444b7b039348d8c27e4ab95beb90675 Mon Sep 17 00:00:00 2001 From: Zhaofeng Zhang <24791380+vcfgv@users.noreply.github.com> Date: Mon, 10 Nov 2025 17:34:27 +0800 Subject: [PATCH] feat: add strategy prompts API and model for managing reusable prompts --- .../valuecell/agents/strategy_agent/models.py | 9 ++- .../agents/strategy_agent/runtime.py | 65 +++++------------ .../server/api/routers/strategy_agent.py | 30 ++++++++ .../server/api/routers/strategy_api.py | 4 ++ .../server/api/routers/strategy_prompts.py | 69 +++++++++++++++++++ .../valuecell/server/api/schemas/strategy.py | 22 ++++++ .../server/db/models/strategy_prompt.py | 47 +++++++++++++ .../db/repositories/strategy_repository.py | 55 +++++++++++++++ 8 files changed, 251 insertions(+), 50 deletions(-) create mode 100644 python/valuecell/server/api/routers/strategy_prompts.py create mode 100644 python/valuecell/server/db/models/strategy_prompt.py diff --git a/python/valuecell/agents/strategy_agent/models.py b/python/valuecell/agents/strategy_agent/models.py index b8e559062..4bfc8069a 100644 --- a/python/valuecell/agents/strategy_agent/models.py +++ b/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( diff --git a/python/valuecell/agents/strategy_agent/runtime.py b/python/valuecell/agents/strategy_agent/runtime.py index b066ed194..f952ba7a0 100644 --- a/python/valuecell/agents/strategy_agent/runtime.py +++ b/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( diff --git a/python/valuecell/server/api/routers/strategy_agent.py b/python/valuecell/server/api/routers/strategy_agent.py index fe3c9fd6f..a6411036e 100644 --- a/python/valuecell/server/api/routers/strategy_agent.py +++ b/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" diff --git a/python/valuecell/server/api/routers/strategy_api.py b/python/valuecell/server/api/routers/strategy_api.py index 54e7814fa..a7d0799a5 100644 --- a/python/valuecell/server/api/routers/strategy_api.py +++ b/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 diff --git a/python/valuecell/server/api/routers/strategy_prompts.py b/python/valuecell/server/api/routers/strategy_prompts.py new file mode 100644 index 000000000..0b838dfcd --- /dev/null +++ b/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 diff --git a/python/valuecell/server/api/schemas/strategy.py b/python/valuecell/server/api/schemas/strategy.py index a7141b647..8e31d28b5 100644 --- a/python/valuecell/server/api/schemas/strategy.py +++ b/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] diff --git a/python/valuecell/server/db/models/strategy_prompt.py b/python/valuecell/server/db/models/strategy_prompt.py new file mode 100644 index 000000000..d68e0678d --- /dev/null +++ b/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"" + + 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, + } diff --git a/python/valuecell/server/db/repositories/strategy_repository.py b/python/valuecell/server/db/repositories/strategy_repository.py index 3123d233f..07c5d0143 100644 --- a/python/valuecell/server/db/repositories/strategy_repository.py +++ b/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