feat: 新增 DecisionSignal 持久化 API（#1390 P1） (#1645)

* feat: add decision signal persistence api

Refs #1390

* fix: handle HK decision signal filters without market

api/app.py

@@ -221,7 +221,8 @@ def create_app(static_dir: Optional[Path] = None) -> FastAPI:
             "- 历史记录：查询历史分析报告\n"
             "- 股票数据：获取行情数据\n\n"
             "## 认证方式\n"
-            "支持可选的运行时认证（通过 WebUI 设置页面启用/关闭）"
+            "支持可选管理员认证：ADMIN_AUTH_ENABLED=true 时，除登录、状态、健康检查和 "
+            "OpenAPI 文档外，/api/v1/* 需要有效管理员会话 Cookie；关闭时不强制认证。"
         ),
         version="1.0.0",
         lifespan=app_lifespan,

api/v1/endpoints/__init__.py

@@ -20,6 +20,7 @@ from api.v1.endpoints import (
     usage,
     portfolio,
     alerts,
+    decision_signals,
     alphasift,
 )
 __all__ = [
@@ -34,5 +35,6 @@ __all__ = [
     "usage",
     "portfolio",
     "alerts",
+    "decision_signals",
     "alphasift",
 ]

api/v1/endpoints/decision_signals.py

@@ -0,0 +1,258 @@
+# -*- coding: utf-8 -*-
+"""DecisionSignal API endpoints."""
+
+from __future__ import annotations
+
+import logging
+from typing import Optional
+
+from fastapi import APIRouter, HTTPException, Query, Security
+from fastapi.security import APIKeyCookie
+
+from api.v1.schemas.common import ErrorResponse
+from api.v1.schemas.decision_signals import (
+    DecisionSignalCreateRequest,
+    DecisionSignalItem,
+    DecisionSignalListResponse,
+    DecisionSignalMutationResponse,
+    DecisionSignalStatusUpdateRequest,
+)
+from src.auth import COOKIE_NAME
+from src.services.decision_signal_service import (
+    DecisionSignalNotFoundError,
+    DecisionSignalService,
+    DecisionSignalStorageError,
+)
+
+
+logger = logging.getLogger(__name__)
+
+admin_session_cookie = APIKeyCookie(
+    name=COOKIE_NAME,
+    scheme_name="AdminSessionCookie",
+    auto_error=False,
+)
+router = APIRouter(dependencies=[Security(admin_session_cookie)])
+
+AUTH_RESPONSE = {
+    401: {
+        "model": ErrorResponse,
+        "description": "未登录或管理员会话无效（ADMIN_AUTH_ENABLED=true 时）",
+    },
+}
+
+
+def _bad_request(exc: Exception, *, error: str = "validation_error") -> HTTPException:
+    return HTTPException(
+        status_code=400,
+        detail={"error": error, "message": str(exc)},
+    )
+
+
+def _not_found(exc: Exception) -> HTTPException:
+    return HTTPException(
+        status_code=404,
+        detail={"error": "not_found", "message": str(exc)},
+    )
+
+
+def _internal_error(message: str, exc: Exception) -> HTTPException:
+    logger.error("%s: %s", message, exc, exc_info=True)
+    return HTTPException(
+        status_code=500,
+        detail={"error": "internal_error", "message": message},
+    )
+
+
+@router.post(
+    "",
+    response_model=DecisionSignalMutationResponse,
+    responses={
+        **AUTH_RESPONSE,
+        400: {"model": ErrorResponse, "description": "请求字段非法"},
+        422: {"model": ErrorResponse, "description": "请求体或路径参数校验失败"},
+        500: {"model": ErrorResponse, "description": "创建失败"},
+    },
+    summary="创建或去重决策信号",
+    description=(
+        "显式写入 DecisionSignal。命中同源去重键时返回已有记录和 created=false；"
+        "若已有记录为 expired 且新请求为 active 并携带未来 expires_at，则原地刷新该记录；"
+        "P1 不保证并发绝对幂等。"
+    ),
+    operation_id="createDecisionSignal",
+)
+def create_signal(request: DecisionSignalCreateRequest) -> DecisionSignalMutationResponse:
+    service = DecisionSignalService()
+    try:
+        payload = request.model_dump()
+        return DecisionSignalMutationResponse(**service.create_signal(payload))
+    except DecisionSignalStorageError as exc:
+        raise _internal_error("Create decision signal failed", exc)
+    except ValueError as exc:
+        raise _bad_request(exc)
+    except Exception as exc:
+        raise _internal_error("Create decision signal failed", exc)
+
+
+@router.get(
+    "",
+    response_model=DecisionSignalListResponse,
+    responses={
+        **AUTH_RESPONSE,
+        400: {"model": ErrorResponse, "description": "查询参数非法"},
+        422: {"model": ErrorResponse, "description": "查询参数校验失败"},
+        500: {"model": ErrorResponse, "description": "查询失败"},
+    },
+    summary="查询决策信号列表",
+    description=(
+        "分页查询 DecisionSignal；读取前会懒过期已到 expires_at 的 active 信号。"
+        "holding_only=true 只读取 active 账户的 portfolio_positions 缓存持仓，不触发 portfolio snapshot replay。"
+    ),
+    operation_id="listDecisionSignals",
+)
+def list_signals(
+    market: Optional[str] = Query(None, description="Optional market filter: cn/hk/us"),
+    stock_code: Optional[str] = Query(None, description="Optional stock code filter"),
+    action: Optional[str] = Query(None, description="Optional decision action filter"),
+    market_phase: Optional[str] = Query(None, description="Optional market phase filter"),
+    source_type: Optional[str] = Query(None, description="Optional source type filter"),
+    source_report_id: Optional[int] = Query(None, description="Optional source report id filter"),
+    trace_id: Optional[str] = Query(None, description="Optional trace id filter"),
+    trigger_source: Optional[str] = Query(None, description="Optional trigger source filter"),
+    status: Optional[str] = Query(None, description="Optional status filter"),
+    created_from: Optional[str] = Query(None, description="Inclusive created_at lower bound"),
+    created_to: Optional[str] = Query(None, description="Inclusive created_at upper bound"),
+    expires_from: Optional[str] = Query(None, description="Inclusive expires_at lower bound"),
+    expires_to: Optional[str] = Query(None, description="Inclusive expires_at upper bound"),
+    holding_only: bool = Query(False, description="Filter to active cached portfolio holdings only"),
+    account_id: Optional[int] = Query(
+        None,
+        description="Optional active portfolio account id for holding_only",
+    ),
+    page: int = Query(1, ge=1),
+    page_size: int = Query(20, ge=1, le=100),
+) -> DecisionSignalListResponse:
+    service = DecisionSignalService()
+    try:
+        return DecisionSignalListResponse(
+            **service.list_signals(
+                market=market,
+                stock_code=stock_code,
+                action=action,
+                market_phase=market_phase,
+                source_type=source_type,
+                source_report_id=source_report_id,
+                trace_id=trace_id,
+                trigger_source=trigger_source,
+                status=status,
+                created_from=created_from,
+                created_to=created_to,
+                expires_from=expires_from,
+                expires_to=expires_to,
+                holding_only=holding_only,
+                account_id=account_id,
+                page=page,
+                page_size=page_size,
+            )
+        )
+    except DecisionSignalStorageError as exc:
+        raise _internal_error("List decision signals failed", exc)
+    except ValueError as exc:
+        raise _bad_request(exc)
+    except Exception as exc:
+        raise _internal_error("List decision signals failed", exc)
+
+
+@router.get(
+    "/latest/{stock_code}",
+    response_model=DecisionSignalListResponse,
+    responses={
+        **AUTH_RESPONSE,
+        400: {"model": ErrorResponse, "description": "请求参数非法"},
+        422: {"model": ErrorResponse, "description": "路径或查询参数校验失败"},
+        500: {"model": ErrorResponse, "description": "查询失败"},
+    },
+    summary="查询股票最新 active 决策信号",
+    description="返回指定股票最新 active 信号列表；读取前会执行懒过期。",
+    operation_id="getLatestDecisionSignals",
+)
+def get_latest_active(
+    stock_code: str,
+    market: Optional[str] = Query(None, description="Optional market filter: cn/hk/us"),
+    limit: int = Query(1, ge=1, le=100),
+) -> DecisionSignalListResponse:
+    service = DecisionSignalService()
+    try:
+        return DecisionSignalListResponse(
+            **service.get_latest_active(
+                stock_code=stock_code,
+                market=market,
+                limit=limit,
+            )
+        )
+    except DecisionSignalStorageError as exc:
+        raise _internal_error("Get latest decision signals failed", exc)
+    except ValueError as exc:
+        raise _bad_request(exc)
+    except Exception as exc:
+        raise _internal_error("Get latest decision signals failed", exc)
+
+
+@router.get(
+    "/{signal_id}",
+    response_model=DecisionSignalItem,
+    responses={
+        **AUTH_RESPONSE,
+        404: {"model": ErrorResponse, "description": "信号不存在"},
+        422: {"model": ErrorResponse, "description": "路径参数校验失败"},
+        500: {"model": ErrorResponse, "description": "查询失败"},
+    },
+    summary="查询单条决策信号",
+    description="按 ID 查询单条 DecisionSignal；读取前会执行懒过期。",
+    operation_id="getDecisionSignal",
+)
+def get_signal(signal_id: int) -> DecisionSignalItem:
+    service = DecisionSignalService()
+    try:
+        return DecisionSignalItem(**service.get_signal(signal_id))
+    except DecisionSignalNotFoundError as exc:
+        raise _not_found(exc)
+    except DecisionSignalStorageError as exc:
+        raise _internal_error("Get decision signal failed", exc)
+    except Exception as exc:
+        raise _internal_error("Get decision signal failed", exc)
+
+
+@router.patch(
+    "/{signal_id}/status",
+    response_model=DecisionSignalItem,
+    responses={
+        **AUTH_RESPONSE,
+        400: {"model": ErrorResponse, "description": "状态非法"},
+        404: {"model": ErrorResponse, "description": "信号不存在"},
+        422: {"model": ErrorResponse, "description": "请求体或路径参数校验失败"},
+        500: {"model": ErrorResponse, "description": "更新失败"},
+    },
+    summary="更新决策信号状态",
+    description="只更新合法状态和可选 metadata；传入 metadata 时按整包替换保存，P1 不实现复杂状态机。",
+    operation_id="updateDecisionSignalStatus",
+)
+def update_status(signal_id: int, request: DecisionSignalStatusUpdateRequest) -> DecisionSignalItem:
+    service = DecisionSignalService()
+    try:
+        return DecisionSignalItem(
+            **service.update_status(
+                signal_id,
+                status=request.status,
+                metadata=request.metadata,
+                replace_metadata="metadata" in request.model_fields_set,
+            )
+        )
+    except DecisionSignalNotFoundError as exc:
+        raise _not_found(exc)
+    except DecisionSignalStorageError as exc:
+        raise _internal_error("Update decision signal status failed", exc)
+    except ValueError as exc:
+        raise _bad_request(exc)
+    except Exception as exc:
+        raise _internal_error("Update decision signal status failed", exc)

api/v1/router.py

@@ -11,7 +11,21 @@ API v1 路由聚合
 
 from fastapi import APIRouter
 
-from api.v1.endpoints import alerts, analysis, auth, history, stocks, backtest, system_config, agent, usage, portfolio, alphasift, health
+from api.v1.endpoints import (
+    agent,
+    alerts,
+    alphasift,
+    analysis,
+    auth,
+    backtest,
+    decision_signals,
+    health,
+    history,
+    portfolio,
+    stocks,
+    system_config,
+    usage,
+)
 
 # 创建 v1 版本主路由
 router = APIRouter(prefix="/api/v1")
@@ -76,6 +90,12 @@ router.include_router(
     tags=["Alerts"]
 )
 
+router.include_router(
+    decision_signals.router,
+    prefix="/decision-signals",
+    tags=["DecisionSignals"]
+)
+
 router.include_router(
     alphasift.router,
     prefix="/alphasift",

api/v1/schemas/__init__.py

@@ -107,6 +107,13 @@ from api.v1.schemas.alerts import (
     AlertTriggerItem,
     AlertTriggerListResponse,
 )
+from api.v1.schemas.decision_signals import (
+    DecisionSignalCreateRequest,
+    DecisionSignalItem,
+    DecisionSignalListResponse,
+    DecisionSignalMutationResponse,
+    DecisionSignalStatusUpdateRequest,
+)
 
 __all__ = [
     # common
@@ -201,4 +208,10 @@ __all__ = [
     "AlertRuleUpdateRequest",
     "AlertTriggerItem",
     "AlertTriggerListResponse",
+    # decision signals
+    "DecisionSignalCreateRequest",
+    "DecisionSignalItem",
+    "DecisionSignalListResponse",
+    "DecisionSignalMutationResponse",
+    "DecisionSignalStatusUpdateRequest",
 ]

api/v1/schemas/decision_signals.py

@@ -0,0 +1,104 @@
+# -*- coding: utf-8 -*-
+"""DecisionSignal API schemas."""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any, Dict, List, Literal, Optional
+
+from pydantic import BaseModel, Field
+
+from api.v1.schemas.market_phase import MarketPhaseValue
+from src.schemas.decision_action import DecisionAction
+
+
+DecisionSignalSourceType = Literal["analysis", "agent", "alert", "market_review", "manual"]
+DecisionSignalStatus = Literal["active", "expired", "invalidated", "closed", "archived"]
+DecisionSignalPlanQuality = Literal["complete", "partial", "minimal", "unknown"]
+DecisionSignalHorizon = Literal["intraday", "1d", "3d", "5d", "10d", "swing", "long"]
+DecisionSignalMarket = Literal["cn", "hk", "us"]
+
+
+class DecisionSignalCreateRequest(BaseModel):
+    stock_code: str = Field(..., min_length=1, max_length=32)
+    stock_name: Optional[str] = Field(None, json_schema_extra={"maxLength": 64})
+    market: DecisionSignalMarket
+    source_type: DecisionSignalSourceType
+    source_agent: Optional[str] = Field(None, json_schema_extra={"maxLength": 64})
+    source_report_id: Optional[int] = None
+    trace_id: Optional[str] = Field(None, json_schema_extra={"maxLength": 64})
+    market_phase: Optional[MarketPhaseValue] = None
+    trigger_source: str = Field(..., min_length=1, json_schema_extra={"maxLength": 64})
+    action: DecisionAction
+    action_label: Optional[str] = Field(None, json_schema_extra={"maxLength": 32})
+    confidence: Optional[float] = Field(None, ge=0.0, le=1.0)
+    score: Optional[int] = Field(None, ge=0, le=100)
+    horizon: Optional[DecisionSignalHorizon] = None
+    entry_low: Optional[float] = Field(None, gt=0, allow_inf_nan=False)
+    entry_high: Optional[float] = Field(None, gt=0, allow_inf_nan=False)
+    stop_loss: Optional[float] = Field(None, gt=0, allow_inf_nan=False)
+    target_price: Optional[float] = Field(None, gt=0, allow_inf_nan=False)
+    invalidation: Optional[Any] = None
+    watch_conditions: Optional[Any] = None
+    reason: Optional[Any] = None
+    risk_summary: Optional[Any] = None
+    catalyst_summary: Optional[Any] = None
+    evidence: Optional[Any] = None
+    data_quality_summary: Optional[Any] = None
+    plan_quality: Optional[DecisionSignalPlanQuality] = None
+    status: Optional[DecisionSignalStatus] = None
+    expires_at: Optional[datetime] = None
+    metadata: Optional[Dict[str, Any]] = None
+    report_language: Optional[Literal["zh", "en"]] = None
+
+
+class DecisionSignalStatusUpdateRequest(BaseModel):
+    status: DecisionSignalStatus
+    metadata: Optional[Dict[str, Any]] = None
+
+
+class DecisionSignalItem(BaseModel):
+    id: int
+    stock_code: str
+    stock_name: Optional[str] = None
+    market: str
+    source_type: str
+    source_agent: Optional[str] = None
+    source_report_id: Optional[int] = None
+    trace_id: Optional[str] = None
+    market_phase: Optional[str] = None
+    trigger_source: str
+    action: str
+    action_label: Optional[str] = None
+    confidence: Optional[float] = None
+    score: Optional[int] = None
+    horizon: Optional[str] = None
+    entry_low: Optional[float] = None
+    entry_high: Optional[float] = None
+    stop_loss: Optional[float] = None
+    target_price: Optional[float] = None
+    invalidation: Optional[str] = None
+    watch_conditions: Optional[str] = None
+    reason: Optional[str] = None
+    risk_summary: Optional[str] = None
+    catalyst_summary: Optional[str] = None
+    evidence: Optional[Any] = None
+    data_quality_summary: Optional[Any] = None
+    plan_quality: str
+    status: str
+    expires_at: Optional[str] = None
+    created_at: Optional[str] = None
+    updated_at: Optional[str] = None
+    metadata: Optional[Any] = None
+
+
+class DecisionSignalMutationResponse(BaseModel):
+    item: DecisionSignalItem
+    created: bool
+
+
+class DecisionSignalListResponse(BaseModel):
+    items: List[DecisionSignalItem] = Field(default_factory=list)
+    total: int
+    page: int
+    page_size: int

docs/CHANGELOG.md

@@ -12,6 +12,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 - [修复] 问股从历史报告进入后的追问会持续携带当前标的，切回或重载已有会话时可从历史消息恢复基础当前标的，并由后端阻断未明确切换时的错误股票工具调用、交易所片段和指标缩写误路由。
 - [修复] 自选股加入和删除按等价股票代码匹配港股及大小写美股变体，避免 `00700`、`HK00700`、`00700.HK` 或 `aapl`、`AAPL` 被误判为不同标的。
 - [改进] #1390 P0 为个股分析与历史/回测展示新增可选八态 `action` / `action_label` 建议动作字段，保留 `operation_advice` 自由文本和 `decision_type=buy|hold|sell` 统计口径，不新增迁移或配置项。
+- [新功能] #1390 P1 新增独立 `DecisionSignal` 存储、Repository、Service 与 `/api/v1/decision-signals` API，支持按来源类型/市场/股票/动作/期限/阶段去重、按 `source_report_id` / `trace_id` 查询、同源过期信号续期且保留来源身份字段、禁止 expired 直接 PATCH 复活、价格计划校验、状态更新、懒过期、cache-only 持仓过滤、敏感信息脱敏、敏感 `trace_id` 拒绝和仅清理 `source_type=analysis` 历史绑定信号的历史删除联动。
 - [修复] #1390 收紧建议动作 legacy fallback：英文 `not to ...` 与 `avoid selling/reducing/trimming ...` 等否定/回避表达不再误判为买卖动作，Web 旧记录不再把中文金融上下文、`buy or sell`、多 guard 歧义文本或 `buyback` / `buy-back` / `buy back` / `selloff` / `sell-off` / `sell off` 等英文复合词渲染成 action badge，并在有结构化 `action` 时让回测/历史趋势等入口按界面语言显示 action 标签。
 - [改进] 完善运行时日志上下文，补充 logger name、触发来源、市场统计与实时行情预取链路状态，便于排查调度、API、Bot 和数据源降级路径。
 <!-- 新条目格式：- [类型] 描述（类型取值：新功能/改进/修复/文档/测试/chore）-->

docs/full-guide.md

@@ -1297,7 +1297,29 @@ python main.py --debug
 
 未知或歧义建议不会兜底成 `watch` 或 `hold`，而是返回空 `action/action_label`。Web 历史卡片、StockBar、同股历史抽屉和回测结果行会在旧记录缺少 `action/action_label` 时从 `operation_advice` 做展示级 fallback；该 fallback 只影响前端标签，不等价于稳定 API action 或后续信号资产。Web 展示层在同时收到 `action` 与 `action_label` 时，会优先按当前界面语言从 `action` 生成标签；API 中的 `action_label` 仍按报告语言生成，供非 Web 客户端或无 `action` 的兼容展示使用。大盘复盘和其他非个股报告不会产生交易 `action`，只保留 `operation_advice` 文本。`dashboard.phase_decision.immediate_action` 属于市场阶段护栏报告字段，不参与 #1390 P0 的八态 action 派生；最终市场阶段仍来自 `report.meta.market_phase_summary.phase`。
 
-#1390 P0 不定义或输出后续信号资产字段；`horizon`、`plan_quality`、`status` 等更细粒度计划字段留待后续独立设计。本阶段不平铺到现有 summary、历史列表、StockBar 或回测响应，不做 DB migration、不回填历史、不新增配置项。
+#1390 P0 不会把后续信号资产字段平铺到现有 summary、历史列表、StockBar 或回测响应。#1390 P1 开始通过独立 `DecisionSignal` 资源承接 `horizon`、`plan_quality`、`status` 等更细粒度计划字段，仍不改变既有报告主契约、不回填历史、不新增配置项。
+
+### 决策信号资产（#1390 P1）
+
+`DecisionSignal` 是独立后端资源，用于把 AI 建议沉淀为可查询、可去重、可更新状态的信号资产。它不替换 `operation_advice`、不扩展 `decision_type=buy|hold|sell`，也不会自动从现有报告提取；P2 之前只有显式调用 API 或 service 的路径会写入信号。
+
+核心字段包括 `stock_code`、`stock_name`、`market`、`source_type`、`source_agent`、`source_report_id`、`trace_id`、`market_phase`、`trigger_source`、`action`、`action_label`、`confidence`、`score`、`horizon`、`entry_low`、`entry_high`、`stop_loss`、`target_price`、`invalidation`、`watch_conditions`、`reason`、`risk_summary`、`catalyst_summary`、`evidence`、`data_quality_summary`、`plan_quality`、`status`、`expires_at`、`created_at`、`updated_at` 和 `metadata`。`action` 复用八态建议动作；`market_phase` 复用市场阶段枚举；`source_type` 支持 `analysis|agent|alert|market_review|manual`；`status` 支持 `active|expired|invalidated|closed|archived`；`horizon` 支持 `intraday|1d|3d|5d|10d|swing|long`。
+
+`confidence` 为 `0.0-1.0`，`score` 为 `0-100`，与历史报告的 `sentiment_score` 解耦。价格计划字段 `entry_low`、`entry_high`、`stop_loss`、`target_price` 必须是有限正数，且同时传入 `entry_low` 和 `entry_high` 时要求 `entry_low <= entry_high`。`plan_quality` 支持 `complete|partial|minimal|unknown`：调用方显式传入合法值时直接保存；未传时由 service 计算，入场区间（`entry_low` 或 `entry_high` 任一有值）算 1 项，`stop_loss`、`target_price`、`invalidation`、`watch_conditions` 各算 1 项，满足 2 项为 `partial`，满足 4 项及以上为 `complete`，仅有 action/reason 为 `minimal`。
+
+新增 API：
+
+- `POST /api/v1/decision-signals`：创建或按同源键去重，返回 `{ item, created }`，HTTP 200。去重键为 `(source_report_id, source_type, market, stock_code, action, horizon, market_phase)`；没有 report 但有 `trace_id` 时使用 `(trace_id, source_type, market, stock_code, action, horizon, market_phase)`；两者皆无则不去重。`source_type` 是来源命名空间，manual/pre-report 弱引用不会与真实 analysis 绑定信号互相去重；`horizon` 和 `market_phase` 同为 `NULL` 时才互相去重，不同来源类型、不同市场、不同期限或不同市场阶段允许保存多条信号。若命中同源 expired 记录，且新请求为 active 并携带未来 `expires_at`，会原地刷新该记录并返回 `created=false`。P1 不提供并发唯一性保证。
+- `GET /api/v1/decision-signals`：分页查询，支持 `market`、`stock_code`、`action`、`market_phase`、`source_type`、`source_report_id`、`trace_id`、`trigger_source`、`status`、时间范围、`holding_only`、`account_id`。
+- `GET /api/v1/decision-signals/{signal_id}`：查询单条，不存在返回 404。
+- `PATCH /api/v1/decision-signals/{signal_id}/status`：更新合法状态和可选 `metadata`；传入 `metadata` 时按整包替换保存，不实现复杂状态机。
+- `GET /api/v1/decision-signals/latest/{stock_code}`：按股票查询最新 active 信号，默认 `limit=1`。
+
+读取入口会懒过期：列表、详情和 latest 查询前会把已到 `expires_at` 的 active 信号标为 expired；创建时已过期的 active 信号会直接保存为 expired；同源 expired 信号只能通过重新 `POST` active + 未来 `expires_at` 的方式延展，`PATCH /status` 不接受 `expires_at`。`closed|invalidated|archived` 不会被 create 路径复活。时间字段按 UTC 归一化为无时区 `datetime` 保存和比较；带时区输入会先转为 UTC 后去掉 `tzinfo`，无时区输入按 UTC 处理，API 响应继续返回不带时区后缀的 ISO 字符串。股票代码入库与查询按 `market` 确定性归一化：A 股 `600519`、`SH600519`、`600519.SH` 等常见变体按同一代码匹配；港股 `00700`、`HK00700`、`00700.HK` 按 `HK00700` 匹配；美股 ticker 统一大写。`holding_only=true` 只读取 active 账户下 `portfolio_positions` 中 `quantity > 0` 的缓存持仓，并按持仓 `(market, stock_code)` 匹配信号，可选 active `account_id`；该查询不会调用组合 snapshot replay，无缓存时返回空结果，需先通过 portfolio snapshot API 刷新缓存。
+
+`source_report_id` 可为空且不强制校验历史记录存在；删除历史记录时只显式清理 `source_type=analysis` 且 `source_report_id` 命中实际删除 ID 的历史绑定信号，`manual/agent/alert/market_review` 等弱引用信号不会仅因 ID 碰撞被删除；列表接口支持按 `source_report_id` 和 `trace_id` 做 typed filter。`task_id`、`alert_trigger_id` 等后续关联字段先放入 `metadata`，P1 不新增独立列，也不提供 typed filter，后续联动阶段再提升为独立契约。JSON 字段、长文本字段和展示型短文本字段（`stock_name/source_agent/trigger_source/action_label`）会在写入前执行信号专用脱敏，覆盖敏感 key、Bearer、Authorization/Cookie header 或赋值、token-like 字符串、其他敏感赋值、webhook URL、URL userinfo 以及带敏感 query/fragment 参数的 URL；普通证据 URL 会保留以保证来源可追溯，且长文本不会套用诊断文本的 300 字符截断。`trace_id` 是同源去重身份字段，若包含会被脱敏的敏感 credential，API 会拒绝请求而不是保存有损 redaction 后的值。
+
+这些接口继承现有 `/api/v1/*` 管理员鉴权：`ADMIN_AUTH_ENABLED=true` 时必须携带有效管理员会话 Cookie；本功能不新增独立认证方式。
 
 ## 回测功能
 
@@ -1402,6 +1424,11 @@ FastAPI 提供 RESTful API 服务，支持配置管理和触发分析。
 | `/api/v1/alphasift/screen/tasks/{task_id}` | GET | 查询 AlphaSift 选股任务状态与完成结果 |
 | `/api/v1/history` | GET | 查询分析历史 |
 | `/api/v1/history/{record_id}/diagnostics` | GET | 查询历史报告运行诊断摘要与脱敏复制文本 |
+| `/api/v1/decision-signals` | POST | 显式创建或按同源键去重决策信号，返回 `{ item, created }` |
+| `/api/v1/decision-signals` | GET | 分页查询决策信号，支持股票、市场、动作、阶段、来源、状态、时间范围和 cache-only 持仓过滤 |
+| `/api/v1/decision-signals/{signal_id}` | GET | 查询单条决策信号，读取前执行懒过期 |
+| `/api/v1/decision-signals/{signal_id}/status` | PATCH | 更新决策信号状态和可选 metadata |
+| `/api/v1/decision-signals/latest/{stock_code}` | GET | 查询指定股票最新 active 决策信号 |
 | `/api/v1/usage/summary?period=today|month|all` | GET | 按调用类型与模型维度汇总 LLM 调用次数和 Token 用量 |
 | `/api/v1/backtest/run` | POST | 触发回测 |
 | `/api/v1/backtest/results` | GET | 查询回测结果（分页） |

docs/full-guide_EN.md

@@ -1128,7 +1128,29 @@ The `decision_type` bridge in the table only documents compatibility between the
 
 Unknown or ambiguous advice is not coerced into `watch` or `hold`; it returns empty `action/action_label`. Web history cards, StockBar, same-stock history drawers, and backtest result rows use `operation_advice` as a display-only fallback when old records do not have `action/action_label`; that fallback affects only the UI label and is not a stable API action or future signal asset. When Web receives both `action` and `action_label`, it first renders the label from `action` in the current UI language; API `action_label` remains report-language display metadata for non-Web clients or compatibility display when `action` is absent. Market review and other non-stock reports do not emit trading `action` values and keep only the `operation_advice` text. `dashboard.phase_decision.immediate_action` belongs to the market-phase guardrail report block and is not used by the #1390 P0 eight-state action derivation. The final market phase still comes from `report.meta.market_phase_summary.phase`.
 
-#1390 P0 does not define or emit future signal-asset fields. More granular plan fields such as `horizon`, `plan_quality`, and `status` are left for a separate follow-up design. This phase does not flatten them into current report summaries, history lists, StockBar rows, or backtest responses; it adds no DB migration, no historical backfill, and no new configuration.
+#1390 P0 does not flatten future signal-asset fields into current report summaries, history lists, StockBar rows, or backtest responses. #1390 P1 now carries more granular plan fields such as `horizon`, `plan_quality`, and `status` through an independent `DecisionSignal` resource; it still does not change the existing report contract, backfill history, or add configuration.
+
+### Decision Signal Asset (#1390 P1)
+
+`DecisionSignal` is an independent backend resource for persisting AI recommendations as queryable, deduplicated, status-updatable signal assets. It does not replace `operation_advice`, does not expand the legacy `decision_type=buy|hold|sell` contract, and does not auto-extract from existing reports yet; before P2, signals are written only through explicit API or service calls.
+
+Core fields include `stock_code`, `stock_name`, `market`, `source_type`, `source_agent`, `source_report_id`, `trace_id`, `market_phase`, `trigger_source`, `action`, `action_label`, `confidence`, `score`, `horizon`, `entry_low`, `entry_high`, `stop_loss`, `target_price`, `invalidation`, `watch_conditions`, `reason`, `risk_summary`, `catalyst_summary`, `evidence`, `data_quality_summary`, `plan_quality`, `status`, `expires_at`, `created_at`, `updated_at`, and `metadata`. `action` reuses the eight-state action taxonomy; `market_phase` reuses the market phase enum; `source_type` supports `analysis|agent|alert|market_review|manual`; `status` supports `active|expired|invalidated|closed|archived`; `horizon` supports `intraday|1d|3d|5d|10d|swing|long`.
+
+`confidence` is `0.0-1.0`, and `score` is `0-100`, separate from historical `sentiment_score`. Price-plan fields `entry_low`, `entry_high`, `stop_loss`, and `target_price` must be finite positive numbers; when both `entry_low` and `entry_high` are present, `entry_low <= entry_high` is required. `plan_quality` supports `complete|partial|minimal|unknown`: a valid explicit value is saved as-is; otherwise the service computes it. The entry range (`entry_low` or `entry_high`) counts as one slot, and `stop_loss`, `target_price`, `invalidation`, and `watch_conditions` each count as one slot. Two slots produce `partial`, four or more produce `complete`, and action/reason without enough slots produces `minimal`.
+
+New API endpoints:
+
+- `POST /api/v1/decision-signals`: create or deduplicate a signal and return `{ item, created }` with HTTP 200. Deduplication uses `(source_report_id, source_type, market, stock_code, action, horizon, market_phase)` when `source_report_id` is present, or `(trace_id, source_type, market, stock_code, action, horizon, market_phase)` when only `trace_id` is present. Signals without either source identifier are not deduplicated. `source_type` is a source namespace, so manual/pre-report weak references do not deduplicate against real analysis-bound signals. `NULL` `horizon` and `NULL` `market_phase` deduplicate only against the same `NULL` dimensions; different source types, markets, horizons, or market phases may persist as separate signals. When the same source key matches an expired signal and the new request is active with a future `expires_at`, the existing row is refreshed in place and still returns `created=false`. P1 does not guarantee concurrent idempotency.
+- `GET /api/v1/decision-signals`: paginated query with `market`, `stock_code`, `action`, `market_phase`, `source_type`, `source_report_id`, `trace_id`, `trigger_source`, `status`, time ranges, `holding_only`, and `account_id`.
+- `GET /api/v1/decision-signals/{signal_id}`: fetch one signal; missing IDs return 404.
+- `PATCH /api/v1/decision-signals/{signal_id}/status`: update a valid status and optional `metadata`; when `metadata` is provided it replaces the whole stored metadata object, and no complex state machine is enforced.
+- `GET /api/v1/decision-signals/latest/{stock_code}`: return latest active signals for a stock, default `limit=1`.
+
+Read paths lazily expire active signals whose `expires_at` has passed before list, detail, and latest queries; creating an already expired active signal stores it as `expired`; the same-source expired signal can only be extended by re-posting active data with a future `expires_at`, and `PATCH /status` does not accept `expires_at`. `closed|invalidated|archived` signals are not reactivated by the create path. Time fields are normalized to UTC naive datetimes for storage and comparison; timezone-aware inputs are converted to UTC and stripped of `tzinfo`, naive inputs are treated as UTC, and API responses continue to return ISO strings without timezone suffixes. Stock codes are normalized deterministically by `market`: CN variants such as `600519`, `SH600519`, and `600519.SH` match the same stored code; HK variants such as `00700`, `HK00700`, and `00700.HK` match `HK00700`; US tickers are uppercased. `holding_only=true` reads only cached `portfolio_positions` rows with `quantity > 0` under active accounts and matches signals by the held `(market, stock_code)`, optionally scoped by an active `account_id`; it does not call portfolio snapshot replay. When no cache exists, it returns an empty result and callers should refresh the cache through the portfolio snapshot API first.
+
+`source_report_id` is nullable and is not required to reference an existing history row; deleting history records explicitly removes only history-bound signals with `source_type=analysis` whose `source_report_id` matches actually deleted IDs, so `manual/agent/alert/market_review` weak-reference signals are not deleted solely because of an ID collision. The list endpoint supports typed filters for `source_report_id` and `trace_id`. Follow-up association fields such as `task_id` and `alert_trigger_id` should be stored in `metadata` for P1; P1 does not add dedicated columns or typed filters for them, which are deferred to the later integration phase. JSON fields, long text fields, and public short text fields (`stock_name/source_agent/trigger_source/action_label`) are sanitized before persistence with a signal-specific sanitizer that redacts sensitive keys, Bearer values, Authorization/Cookie headers or assignments, token-like strings, other sensitive assignments, webhook URLs, URL userinfo, and URLs with sensitive query or fragment parameters. Ordinary evidence URLs are preserved for source traceability, and long text does not use the diagnostics 300-character truncation. `trace_id` is a same-source identity field; if it contains sensitive credentials that would be redacted, the API rejects the request instead of storing a lossy redacted value.
+
+These endpoints inherit the existing `/api/v1/*` admin authentication middleware: when `ADMIN_AUTH_ENABLED=true`, callers must send a valid admin session cookie. DecisionSignal does not add a separate auth scheme.
 
 ## Backtesting
 
@@ -1234,6 +1256,11 @@ For this feature, the product behavior is:
 | `/api/v1/alphasift/screen/tasks/{task_id}` | GET | Query AlphaSift screening task status and completed result |
 | `/api/v1/history` | GET | Query analysis history |
 | `/api/v1/history/{record_id}/diagnostics` | GET | Query a historical report run diagnostic summary and sanitized copy text |
+| `/api/v1/decision-signals` | POST | Explicitly create or deduplicate a decision signal and return `{ item, created }` |
+| `/api/v1/decision-signals` | GET | Paginated decision-signal query with stock, market, action, phase, source, status, time-range, and cache-only holdings filters |
+| `/api/v1/decision-signals/{signal_id}` | GET | Fetch one decision signal and apply lazy expiration before reading |
+| `/api/v1/decision-signals/{signal_id}/status` | PATCH | Update a decision signal status and optional metadata |
+| `/api/v1/decision-signals/latest/{stock_code}` | GET | Query the latest active decision signals for a stock |
 | `/api/v1/usage/summary?period=today|month|all` | GET | Query LLM call counts and token usage grouped by call type and model |
 | `/api/v1/backtest/run` | POST | Trigger backtest |
 | `/api/v1/backtest/results` | GET | Query backtest results (paginated) |

src/repositories/__init__.py

@@ -10,10 +10,12 @@
 
 from src.repositories.analysis_repo import AnalysisRepository
 from src.repositories.backtest_repo import BacktestRepository
+from src.repositories.decision_signal_repo import DecisionSignalRepository
 from src.repositories.stock_repo import StockRepository
 
 __all__ = [
     "AnalysisRepository",
     "BacktestRepository",
+    "DecisionSignalRepository",
     "StockRepository",
 ]

src/repositories/decision_signal_repo.py

@@ -0,0 +1,322 @@
+# -*- coding: utf-8 -*-
+"""Decision signal repository for Issue #1390 P1."""
+
+from __future__ import annotations
+
+from datetime import datetime
+from typing import Any, Dict, List, Optional, Tuple
+
+from sqlalchemy import and_, desc, func, or_, select
+
+from src.storage import (
+    DatabaseManager,
+    DecisionSignalRecord,
+    to_utc_naive_datetime,
+    utc_naive_now,
+)
+
+
+class DecisionSignalRepository:
+    """DB access layer for persisted AI decision signals."""
+
+    _IMMUTABLE_REFRESH_FIELDS = frozenset({
+        "id",
+        "created_at",
+        "source_report_id",
+        "source_type",
+        "source_agent",
+        "trace_id",
+        "trigger_source",
+        "market",
+        "stock_code",
+        "action",
+        "horizon",
+        "market_phase",
+    })
+
+    def __init__(self, db_manager: Optional[DatabaseManager] = None):
+        self.db = db_manager or DatabaseManager.get_instance()
+
+    def create(self, fields: Dict[str, Any]) -> DecisionSignalRecord:
+        fields = self._normalize_datetime_fields(fields)
+        with self.db.get_session() as session:
+            row = DecisionSignalRecord(**fields)
+            session.add(row)
+            session.commit()
+            session.refresh(row)
+            return row
+
+    def create_if_absent(self, fields: Dict[str, Any]) -> Tuple[DecisionSignalRecord, bool]:
+        self.expire_due_signals()
+        fields = self._normalize_datetime_fields(fields)
+        with self.db.get_session() as session:
+            existing = self._find_existing_in_session(session=session, fields=fields)
+            if existing is not None:
+                if self._should_refresh_existing(existing, fields):
+                    self._refresh_existing_in_session(existing, fields)
+                    session.commit()
+                    session.refresh(existing)
+                return existing, False
+            row = DecisionSignalRecord(**fields)
+            session.add(row)
+            session.commit()
+            session.refresh(row)
+            return row, True
+
+    def get(self, signal_id: int) -> Optional[DecisionSignalRecord]:
+        self.expire_due_signals()
+        with self.db.get_session() as session:
+            return session.execute(
+                select(DecisionSignalRecord).where(DecisionSignalRecord.id == signal_id).limit(1)
+            ).scalar_one_or_none()
+
+    def list(
+        self,
+        *,
+        stock_codes: Optional[List[str]] = None,
+        stock_identities: Optional[List[Tuple[str, str]]] = None,
+        market: Optional[str] = None,
+        action: Optional[str] = None,
+        market_phase: Optional[str] = None,
+        source_type: Optional[str] = None,
+        source_report_id: Optional[int] = None,
+        trace_id: Optional[str] = None,
+        trigger_source: Optional[str] = None,
+        status: Optional[str] = None,
+        created_from: Optional[datetime] = None,
+        created_to: Optional[datetime] = None,
+        expires_from: Optional[datetime] = None,
+        expires_to: Optional[datetime] = None,
+        page: int = 1,
+        page_size: int = 20,
+    ) -> Tuple[List[DecisionSignalRecord], int]:
+        self.expire_due_signals()
+        created_from = self._normalize_optional_datetime(created_from)
+        created_to = self._normalize_optional_datetime(created_to)
+        expires_from = self._normalize_optional_datetime(expires_from)
+        expires_to = self._normalize_optional_datetime(expires_to)
+        conditions = self._build_conditions(
+            stock_codes=stock_codes,
+            stock_identities=stock_identities,
+            market=market,
+            action=action,
+            market_phase=market_phase,
+            source_type=source_type,
+            source_report_id=source_report_id,
+            trace_id=trace_id,
+            trigger_source=trigger_source,
+            status=status,
+            created_from=created_from,
+            created_to=created_to,
+            expires_from=expires_from,
+            expires_to=expires_to,
+        )
+        where_clause = and_(*conditions) if conditions else True
+        safe_page = max(1, int(page))
+        safe_page_size = max(1, min(int(page_size), 100))
+        offset = (safe_page - 1) * safe_page_size
+
+        with self.db.get_session() as session:
+            total = session.execute(
+                select(func.count(DecisionSignalRecord.id))
+                .select_from(DecisionSignalRecord)
+                .where(where_clause)
+            ).scalar() or 0
+            rows = session.execute(
+                select(DecisionSignalRecord)
+                .where(where_clause)
+                .order_by(desc(DecisionSignalRecord.created_at), desc(DecisionSignalRecord.id))
+                .offset(offset)
+                .limit(safe_page_size)
+            ).scalars().all()
+            return list(rows), int(total)
+
+    def get_latest_active(
+        self,
+        *,
+        stock_codes: List[str],
+        market: Optional[str] = None,
+        limit: int = 1,
+    ) -> List[DecisionSignalRecord]:
+        self.expire_due_signals()
+        safe_limit = max(1, min(int(limit), 100))
+        conditions = [
+            DecisionSignalRecord.status == "active",
+            DecisionSignalRecord.stock_code.in_(stock_codes),
+        ]
+        if market:
+            conditions.append(DecisionSignalRecord.market == market)
+        with self.db.get_session() as session:
+            rows = session.execute(
+                select(DecisionSignalRecord)
+                .where(and_(*conditions))
+                .order_by(desc(DecisionSignalRecord.created_at), desc(DecisionSignalRecord.id))
+                .limit(safe_limit)
+            ).scalars().all()
+            return list(rows)
+
+    def update_status(
+        self,
+        signal_id: int,
+        *,
+        status: str,
+        metadata_json: Optional[str] = None,
+        replace_metadata: bool = False,
+    ) -> Optional[DecisionSignalRecord]:
+        with self.db.get_session() as session:
+            row = session.execute(
+                select(DecisionSignalRecord).where(DecisionSignalRecord.id == signal_id).limit(1)
+            ).scalar_one_or_none()
+            if row is None:
+                return None
+            row.status = status
+            if replace_metadata:
+                row.metadata_json = metadata_json
+            row.updated_at = utc_naive_now()
+            session.commit()
+            session.refresh(row)
+            return row
+
+    def expire_due_signals(self, now: Optional[datetime] = None) -> int:
+        now_value = to_utc_naive_datetime(now) if now is not None else utc_naive_now()
+        with self.db.get_session() as session:
+            rows = session.execute(
+                select(DecisionSignalRecord).where(
+                    DecisionSignalRecord.status == "active",
+                    DecisionSignalRecord.expires_at.is_not(None),
+                    DecisionSignalRecord.expires_at <= now_value,
+                )
+            ).scalars().all()
+            for row in rows:
+                row.status = "expired"
+                row.updated_at = now_value
+            session.commit()
+            return len(rows)
+
+    @staticmethod
+    def _normalize_datetime_fields(fields: Dict[str, Any]) -> Dict[str, Any]:
+        normalized = dict(fields)
+        for field_name in ("expires_at", "created_at", "updated_at"):
+            value = normalized.get(field_name)
+            if isinstance(value, datetime):
+                normalized[field_name] = to_utc_naive_datetime(value)
+        return normalized
+
+    @staticmethod
+    def _normalize_optional_datetime(value: Optional[datetime]) -> Optional[datetime]:
+        if value is None:
+            return None
+        return to_utc_naive_datetime(value)
+
+    @classmethod
+    def _should_refresh_existing(cls, existing: DecisionSignalRecord, fields: Dict[str, Any]) -> bool:
+        expires_at = fields.get("expires_at")
+        return (
+            existing.status == "expired"
+            and fields.get("status") == "active"
+            and expires_at is not None
+            and expires_at > utc_naive_now()
+        )
+
+    @classmethod
+    def _refresh_existing_in_session(cls, existing: DecisionSignalRecord, fields: Dict[str, Any]) -> None:
+        for field_name, value in fields.items():
+            if field_name in cls._IMMUTABLE_REFRESH_FIELDS:
+                continue
+            setattr(existing, field_name, value)
+        existing.updated_at = utc_naive_now()
+
+    @staticmethod
+    def _find_existing_in_session(*, session: Any, fields: Dict[str, Any]) -> Optional[DecisionSignalRecord]:
+        source_report_id = fields.get("source_report_id")
+        trace_id = fields.get("trace_id")
+        source_type = fields.get("source_type")
+        stock_code = fields.get("stock_code")
+        market = fields.get("market")
+        action = fields.get("action")
+        horizon = fields.get("horizon")
+        market_phase = fields.get("market_phase")
+        if source_report_id is not None:
+            conditions = [
+                DecisionSignalRecord.source_report_id == source_report_id,
+                DecisionSignalRecord.source_type == source_type,
+                DecisionSignalRecord.market == market,
+                DecisionSignalRecord.stock_code == stock_code,
+                DecisionSignalRecord.action == action,
+                DecisionSignalRecord.horizon == horizon,
+                DecisionSignalRecord.market_phase == market_phase,
+            ]
+        elif trace_id:
+            conditions = [
+                DecisionSignalRecord.trace_id == trace_id,
+                DecisionSignalRecord.source_type == source_type,
+                DecisionSignalRecord.market == market,
+                DecisionSignalRecord.stock_code == stock_code,
+                DecisionSignalRecord.action == action,
+                DecisionSignalRecord.horizon == horizon,
+                DecisionSignalRecord.market_phase == market_phase,
+            ]
+        else:
+            return None
+        return session.execute(
+            select(DecisionSignalRecord)
+            .where(and_(*conditions))
+            .order_by(DecisionSignalRecord.id.asc())
+            .limit(1)
+        ).scalar_one_or_none()
+
+    @staticmethod
+    def _build_conditions(
+        *,
+        stock_codes: Optional[List[str]],
+        stock_identities: Optional[List[Tuple[str, str]]],
+        market: Optional[str],
+        action: Optional[str],
+        market_phase: Optional[str],
+        source_type: Optional[str],
+        source_report_id: Optional[int],
+        trace_id: Optional[str],
+        trigger_source: Optional[str],
+        status: Optional[str],
+        created_from: Optional[datetime],
+        created_to: Optional[datetime],
+        expires_from: Optional[datetime],
+        expires_to: Optional[datetime],
+    ) -> List[Any]:
+        conditions: List[Any] = []
+        if stock_identities:
+            identity_conditions = [
+                and_(
+                    DecisionSignalRecord.market == identity_market,
+                    DecisionSignalRecord.stock_code == identity_code,
+                )
+                for identity_market, identity_code in stock_identities
+            ]
+            conditions.append(or_(*identity_conditions))
+        elif stock_codes:
+            conditions.append(DecisionSignalRecord.stock_code.in_(stock_codes))
+        if market:
+            conditions.append(DecisionSignalRecord.market == market)
+        if action:
+            conditions.append(DecisionSignalRecord.action == action)
+        if market_phase:
+            conditions.append(DecisionSignalRecord.market_phase == market_phase)
+        if source_type:
+            conditions.append(DecisionSignalRecord.source_type == source_type)
+        if source_report_id is not None:
+            conditions.append(DecisionSignalRecord.source_report_id == source_report_id)
+        if trace_id:
+            conditions.append(DecisionSignalRecord.trace_id == trace_id)
+        if trigger_source:
+            conditions.append(DecisionSignalRecord.trigger_source == trigger_source)
+        if status:
+            conditions.append(DecisionSignalRecord.status == status)
+        if created_from:
+            conditions.append(DecisionSignalRecord.created_at >= created_from)
+        if created_to:
+            conditions.append(DecisionSignalRecord.created_at <= created_to)
+        if expires_from:
+            conditions.append(DecisionSignalRecord.expires_at >= expires_from)
+        if expires_to:
+            conditions.append(DecisionSignalRecord.expires_at <= expires_to)
+        return conditions

src/repositories/portfolio_repo.py

@@ -797,6 +797,40 @@ class PortfolioRepository:
             cutoff_ordinal = as_of.toordinal() - lookback_days
             return [row for row in rows if row.snapshot_date.toordinal() >= cutoff_ordinal]
 
+    def list_cached_position_identities(
+        self,
+        *,
+        account_id: Optional[int] = None,
+    ) -> List[Tuple[str, str]]:
+        """Return market/symbol identities from cached non-zero positions only."""
+        with self.db.get_session() as session:
+            query = (
+                select(PortfolioPosition.market, PortfolioPosition.symbol)
+                .join(PortfolioAccount, PortfolioPosition.account_id == PortfolioAccount.id)
+                .where(
+                    PortfolioPosition.quantity > 0,
+                    PortfolioAccount.is_active.is_(True),
+                )
+            )
+            if account_id is not None:
+                query = query.where(PortfolioPosition.account_id == account_id)
+            rows = session.execute(
+                query.order_by(
+                    PortfolioPosition.market.asc(),
+                    PortfolioPosition.symbol.asc(),
+                )
+            ).all()
+            seen = set()
+            identities: List[Tuple[str, str]] = []
+            for market, symbol in rows:
+                market_text = str(market or "").strip().lower()
+                symbol_text = str(symbol or "").strip()
+                identity = (market_text, symbol_text)
+                if market_text and symbol_text and identity not in seen:
+                    seen.add(identity)
+                    identities.append(identity)
+            return identities
+
     # ------------------------------------------------------------------
     # Snapshot / position cache
     # ------------------------------------------------------------------

src/services/decision_signal_service.py

@@ -0,0 +1,547 @@
+# -*- coding: utf-8 -*-
+"""Service layer for persisted DecisionSignal assets."""
+
+from __future__ import annotations
+
+import json
+import logging
+import math
+from datetime import datetime
+from typing import Any, Dict, List, Optional, Tuple, get_args
+
+from data_provider.base import canonical_stock_code, normalize_stock_code
+from src.core.trading_calendar import MarketPhase
+from src.repositories.decision_signal_repo import DecisionSignalRepository
+from src.repositories.portfolio_repo import PortfolioRepository
+from src.report_language import normalize_report_language
+from src.schemas.decision_action import DecisionAction, localize_action_label
+from src.services.portfolio_service import VALID_MARKETS
+from src.storage import (
+    DatabaseManager,
+    DecisionSignalRecord,
+    to_utc_naive_datetime,
+    utc_naive_now,
+)
+from src.utils.sanitize import sanitize_decision_signal_payload, sanitize_decision_signal_text
+
+
+SOURCE_TYPES = frozenset({"analysis", "agent", "alert", "market_review", "manual"})
+SIGNAL_STATUSES = frozenset({"active", "expired", "invalidated", "closed", "archived"})
+PLAN_QUALITIES = frozenset({"complete", "partial", "minimal", "unknown"})
+HORIZONS = frozenset({"intraday", "1d", "3d", "5d", "10d", "swing", "long"})
+MARKET_PHASES = frozenset(phase.value for phase in MarketPhase)
+DECISION_ACTIONS = frozenset(get_args(DecisionAction))
+REDACTION_MARKERS = ("[REDACTED]", "[REDACTED_URL]")
+
+logger = logging.getLogger(__name__)
+
+
+class DecisionSignalNotFoundError(ValueError):
+    """Raised when a requested decision signal does not exist."""
+
+
+class DecisionSignalStorageError(RuntimeError):
+    """Raised when persisted decision-signal data is internally inconsistent."""
+
+
+class DecisionSignalService:
+    """Business logic for DecisionSignal storage, querying, and serialization."""
+
+    def __init__(
+        self,
+        repo: Optional[DecisionSignalRepository] = None,
+        portfolio_repo: Optional[PortfolioRepository] = None,
+        db_manager: Optional[DatabaseManager] = None,
+    ):
+        self.repo = repo or DecisionSignalRepository(db_manager)
+        self.portfolio_repo = portfolio_repo or PortfolioRepository(db_manager)
+
+    def create_signal(self, payload: Dict[str, Any]) -> Dict[str, Any]:
+        fields = self._normalize_payload(payload)
+        row, created = self.repo.create_if_absent(fields)
+        return {"item": self._serialize(row), "created": created}
+
+    def get_signal(self, signal_id: int) -> Dict[str, Any]:
+        row = self.repo.get(signal_id)
+        if row is None:
+            raise DecisionSignalNotFoundError(f"Decision signal not found: {signal_id}")
+        return self._serialize(row)
+
+    def list_signals(
+        self,
+        *,
+        stock_code: Optional[str] = None,
+        market: Optional[str] = None,
+        action: Optional[str] = None,
+        market_phase: Optional[str] = None,
+        source_type: Optional[str] = None,
+        source_report_id: Optional[Any] = None,
+        trace_id: Optional[str] = None,
+        trigger_source: Optional[str] = None,
+        status: Optional[str] = None,
+        created_from: Optional[Any] = None,
+        created_to: Optional[Any] = None,
+        expires_from: Optional[Any] = None,
+        expires_to: Optional[Any] = None,
+        holding_only: bool = False,
+        account_id: Optional[int] = None,
+        page: int = 1,
+        page_size: int = 20,
+    ) -> Dict[str, Any]:
+        safe_page = max(1, int(page))
+        safe_page_size = max(1, min(int(page_size), 100))
+        market_norm = self._normalize_optional_market(market)
+        action_norm = self._normalize_optional_action(action)
+        market_phase_norm = self._normalize_optional_enum(market_phase, MARKET_PHASES, "market_phase")
+        source_type_norm = self._normalize_optional_enum(source_type, SOURCE_TYPES, "source_type")
+        source_report_id_norm = self._optional_int(source_report_id, "source_report_id")
+        trace_id_norm = self._optional_identity_text(trace_id, "trace_id", max_length=64)
+        status_norm = self._normalize_optional_enum(status, SIGNAL_STATUSES, "status")
+        trigger_source_norm = self._normalize_optional_trigger_source(trigger_source)
+        created_from_dt = self._parse_datetime(created_from)
+        created_to_dt = self._parse_datetime(created_to)
+        expires_from_dt = self._parse_datetime(expires_from)
+        expires_to_dt = self._parse_datetime(expires_to)
+        stock_codes = self._stock_filter_codes(stock_code, market=market_norm)
+        stock_identities = None
+
+        if holding_only:
+            held_identities = self._cached_holding_identities(account_id=account_id)
+            if market_norm:
+                held_identities = {
+                    identity for identity in held_identities if identity[0] == market_norm
+                }
+            if stock_codes:
+                requested_codes = set(stock_codes)
+                held_identities = {
+                    identity for identity in held_identities if identity[1] in requested_codes
+                }
+            stock_identities = sorted(held_identities)
+            stock_codes = None
+            if not stock_identities:
+                return {"items": [], "total": 0, "page": safe_page, "page_size": safe_page_size}
+
+        rows, total = self.repo.list(
+            stock_codes=stock_codes,
+            stock_identities=stock_identities,
+            market=market_norm,
+            action=action_norm,
+            market_phase=market_phase_norm,
+            source_type=source_type_norm,
+            source_report_id=source_report_id_norm,
+            trace_id=trace_id_norm,
+            trigger_source=trigger_source_norm,
+            status=status_norm,
+            created_from=created_from_dt,
+            created_to=created_to_dt,
+            expires_from=expires_from_dt,
+            expires_to=expires_to_dt,
+            page=safe_page,
+            page_size=safe_page_size,
+        )
+        return {
+            "items": [self._serialize(row) for row in rows],
+            "total": total,
+            "page": safe_page,
+            "page_size": safe_page_size,
+        }
+
+    def get_latest_active(
+        self,
+        *,
+        stock_code: str,
+        market: Optional[str] = None,
+        limit: int = 1,
+    ) -> Dict[str, Any]:
+        market_norm = self._normalize_optional_market(market)
+        rows = self.repo.get_latest_active(
+            stock_codes=self._stock_filter_codes(stock_code, market=market_norm) or [
+                self._normalize_stock_code(stock_code)
+            ],
+            market=market_norm,
+            limit=limit,
+        )
+        return {
+            "items": [self._serialize(row) for row in rows],
+            "total": len(rows),
+            "page": 1,
+            "page_size": max(1, min(int(limit), 100)),
+        }
+
+    def update_status(
+        self,
+        signal_id: int,
+        *,
+        status: str,
+        metadata: Optional[Any] = None,
+        replace_metadata: bool = False,
+    ) -> Dict[str, Any]:
+        status_norm = self._normalize_enum(status, SIGNAL_STATUSES, "status")
+        metadata_json = self._json_dumps(metadata) if replace_metadata else None
+        existing = self.repo.get(signal_id)
+        if existing is None:
+            raise DecisionSignalNotFoundError(f"Decision signal not found: {signal_id}")
+        if status_norm == "active" and (
+            existing.status == "expired" or self._is_expired(existing.expires_at)
+        ):
+            raise ValueError("expired decision signal cannot be reactivated without extending expires_at")
+        row = self.repo.update_status(
+            signal_id,
+            status=status_norm,
+            metadata_json=metadata_json,
+            replace_metadata=replace_metadata,
+        )
+        if row is None:
+            raise DecisionSignalNotFoundError(f"Decision signal not found: {signal_id}")
+        return self._serialize(row)
+
+    def _normalize_payload(self, payload: Dict[str, Any]) -> Dict[str, Any]:
+        market = self._normalize_market(payload.get("market"))
+        stock_code = self._normalize_stock_code(payload.get("stock_code"), market=market)
+        action = self._normalize_action(payload.get("action"))
+        report_language = normalize_report_language(payload.get("report_language"))
+        action_label = self._optional_public_text(payload.get("action_label"), "action_label", max_length=32)
+        if not action_label:
+            action_label = localize_action_label(action, report_language)
+
+        confidence = self._optional_float(payload.get("confidence"), "confidence")
+        if confidence is not None and not 0.0 <= confidence <= 1.0:
+            raise ValueError("confidence must be between 0.0 and 1.0")
+        score = self._optional_int(payload.get("score"), "score")
+        if score is not None and not 0 <= score <= 100:
+            raise ValueError("score must be between 0 and 100")
+
+        fields: Dict[str, Any] = {
+            "stock_code": stock_code,
+            "stock_name": self._optional_public_text(payload.get("stock_name"), "stock_name", max_length=64),
+            "market": market,
+            "source_type": self._normalize_enum(payload.get("source_type"), SOURCE_TYPES, "source_type"),
+            "source_agent": self._optional_public_text(payload.get("source_agent"), "source_agent", max_length=64),
+            "source_report_id": self._optional_int(payload.get("source_report_id"), "source_report_id"),
+            "trace_id": self._optional_identity_text(payload.get("trace_id"), "trace_id", max_length=64),
+            "market_phase": self._normalize_optional_enum(payload.get("market_phase"), MARKET_PHASES, "market_phase"),
+            "trigger_source": self._normalize_trigger_source(payload.get("trigger_source")),
+            "action": action,
+            "action_label": action_label,
+            "confidence": confidence,
+            "score": score,
+            "horizon": self._normalize_optional_enum(payload.get("horizon"), HORIZONS, "horizon"),
+            "entry_low": self._optional_price_float(payload.get("entry_low"), "entry_low"),
+            "entry_high": self._optional_price_float(payload.get("entry_high"), "entry_high"),
+            "stop_loss": self._optional_price_float(payload.get("stop_loss"), "stop_loss"),
+            "target_price": self._optional_price_float(payload.get("target_price"), "target_price"),
+            "invalidation": self._optional_signal_text(payload.get("invalidation")),
+            "watch_conditions": self._optional_signal_text(payload.get("watch_conditions")),
+            "reason": self._optional_signal_text(payload.get("reason")),
+            "risk_summary": self._optional_signal_text(payload.get("risk_summary")),
+            "catalyst_summary": self._optional_signal_text(payload.get("catalyst_summary")),
+            "evidence_json": self._json_dumps(payload.get("evidence")),
+            "data_quality_summary_json": self._json_dumps(payload.get("data_quality_summary")),
+            "status": self._normalize_optional_enum(payload.get("status"), SIGNAL_STATUSES, "status") or "active",
+            "expires_at": self._parse_datetime(payload.get("expires_at")),
+            "metadata_json": self._json_dumps(payload.get("metadata")),
+        }
+        if fields["status"] == "active" and self._is_expired(fields["expires_at"]):
+            fields["status"] = "expired"
+        self._validate_entry_range(fields)
+        fields["plan_quality"] = self._normalize_plan_quality(
+            payload.get("plan_quality"),
+            fields=fields,
+        )
+        return fields
+
+    def _normalize_plan_quality(self, value: Any, *, fields: Dict[str, Any]) -> str:
+        if value is not None:
+            return self._normalize_enum(value, PLAN_QUALITIES, "plan_quality")
+        has_action_or_reason = bool(fields.get("action") or fields.get("reason"))
+        if not has_action_or_reason:
+            return "unknown"
+        slots = 0
+        if fields.get("entry_low") is not None or fields.get("entry_high") is not None:
+            slots += 1
+        for key in ("stop_loss", "target_price", "invalidation", "watch_conditions"):
+            if fields.get(key) not in (None, ""):
+                slots += 1
+        if slots >= 4:
+            return "complete"
+        if slots >= 2:
+            return "partial"
+        return "minimal"
+
+    def _cached_holding_identities(self, *, account_id: Optional[int]) -> set[Tuple[str, str]]:
+        identities = self.portfolio_repo.list_cached_position_identities(account_id=account_id)
+        normalized: set[Tuple[str, str]] = set()
+        for market, symbol in identities:
+            if not str(symbol or "").strip():
+                continue
+            market_norm = self._normalize_market(market)
+            normalized.add((market_norm, self._normalize_stock_code(symbol, market=market_norm)))
+        return normalized
+
+    @classmethod
+    def _stock_filter_codes(
+        cls,
+        stock_code: Optional[str],
+        *,
+        market: Optional[str] = None,
+    ) -> Optional[List[str]]:
+        if not stock_code:
+            return None
+        normalized = cls._normalize_stock_code(stock_code, market=market)
+        if market is not None:
+            return [normalized]
+
+        hk_normalized = cls._normalize_hk_stock_code(str(stock_code).strip())
+        return list(dict.fromkeys([normalized, hk_normalized]))
+
+    @classmethod
+    def _normalize_stock_code(cls, value: Any, *, market: Optional[str] = None) -> str:
+        raw = str(value or "").strip()
+        if market == "us":
+            code = canonical_stock_code(raw)
+        elif market == "hk":
+            code = cls._normalize_hk_stock_code(raw)
+        else:
+            code = canonical_stock_code(normalize_stock_code(raw))
+        if not code:
+            raise ValueError("stock_code is required")
+        return code
+
+    @staticmethod
+    def _normalize_hk_stock_code(value: str) -> str:
+        normalized = canonical_stock_code(normalize_stock_code(value))
+        digits = ""
+        if normalized.startswith("HK"):
+            digits = normalized[2:]
+        elif normalized.isdigit():
+            digits = normalized
+        if digits.isdigit() and 1 <= len(digits) <= 5:
+            return f"HK{digits.zfill(5)}"
+        return normalized
+
+    @staticmethod
+    def _normalize_market(value: Any) -> str:
+        market = str(value or "").strip().lower()
+        if market not in VALID_MARKETS:
+            raise ValueError("market must be one of cn, hk, us")
+        return market
+
+    @classmethod
+    def _normalize_optional_market(cls, value: Any) -> Optional[str]:
+        if value in (None, ""):
+            return None
+        return cls._normalize_market(value)
+
+    @staticmethod
+    def _normalize_action(value: Any) -> str:
+        action = str(value or "").strip().lower()
+        if not action or action not in DECISION_ACTIONS:
+            raise ValueError("action must be one of buy/add/hold/reduce/sell/watch/avoid/alert")
+        return action
+
+    @classmethod
+    def _normalize_optional_action(cls, value: Any) -> Optional[str]:
+        if value in (None, ""):
+            return None
+        return cls._normalize_action(value)
+
+    @staticmethod
+    def _normalize_enum(value: Any, allowed: frozenset[str], field_name: str) -> str:
+        text = str(value or "").strip()
+        if text not in allowed:
+            allowed_text = ", ".join(sorted(allowed))
+            raise ValueError(f"{field_name} must be one of {allowed_text}")
+        return text
+
+    @classmethod
+    def _normalize_optional_enum(
+        cls,
+        value: Any,
+        allowed: frozenset[str],
+        field_name: str,
+    ) -> Optional[str]:
+        if value in (None, ""):
+            return None
+        return cls._normalize_enum(value, allowed, field_name)
+
+    @staticmethod
+    def _normalize_trigger_source(value: Any) -> str:
+        text = DecisionSignalService._public_text(value, "trigger_source", max_length=64, required=True)
+        if not text:
+            raise ValueError("trigger_source is required")
+        return text
+
+    @classmethod
+    def _normalize_optional_trigger_source(cls, value: Any) -> Optional[str]:
+        if value in (None, ""):
+            return None
+        return cls._normalize_trigger_source(value)
+
+    @staticmethod
+    def _optional_text(value: Any, field_name: str, *, max_length: int) -> Optional[str]:
+        if value is None:
+            return None
+        text = str(value).strip()
+        if not text:
+            return None
+        if len(text) > max_length:
+            raise ValueError(f"{field_name} must be at most {max_length} characters")
+        return text
+
+    @classmethod
+    def _optional_public_text(cls, value: Any, field_name: str, *, max_length: int) -> Optional[str]:
+        return cls._public_text(value, field_name, max_length=max_length, required=False)
+
+    @staticmethod
+    def _public_text(value: Any, field_name: str, *, max_length: int, required: bool) -> Optional[str]:
+        if value is None:
+            if required:
+                raise ValueError(f"{field_name} is required")
+            return None
+        text = sanitize_decision_signal_text(value)
+        if not text:
+            if required:
+                raise ValueError(f"{field_name} is required")
+            return None
+        if len(text) > max_length:
+            raise ValueError(f"{field_name} must be at most {max_length} characters")
+        return text
+
+    @classmethod
+    def _optional_identity_text(cls, value: Any, field_name: str, *, max_length: int) -> Optional[str]:
+        text = cls._optional_text(value, field_name, max_length=max_length)
+        if text is None:
+            return None
+        sanitized = sanitize_decision_signal_text(text)
+        if any(marker in sanitized for marker in REDACTION_MARKERS):
+            raise ValueError(f"{field_name} must not contain sensitive credentials")
+        return text
+
+    @staticmethod
+    def _optional_signal_text(value: Any) -> Optional[str]:
+        if value is None:
+            return None
+        if isinstance(value, (dict, list)):
+            return json.dumps(sanitize_decision_signal_payload(value), ensure_ascii=False, sort_keys=True)
+        text = sanitize_decision_signal_text(value)
+        return text or None
+
+    @staticmethod
+    def _optional_float(value: Any, field_name: str) -> Optional[float]:
+        if value in (None, ""):
+            return None
+        try:
+            return float(value)
+        except (TypeError, ValueError) as exc:
+            raise ValueError(f"{field_name} must be a number") from exc
+
+    @classmethod
+    def _optional_price_float(cls, value: Any, field_name: str) -> Optional[float]:
+        number = cls._optional_float(value, field_name)
+        if number is None:
+            return None
+        if not math.isfinite(number) or number <= 0:
+            raise ValueError(f"{field_name} must be a finite positive number")
+        return number
+
+    @staticmethod
+    def _validate_entry_range(fields: Dict[str, Any]) -> None:
+        entry_low = fields.get("entry_low")
+        entry_high = fields.get("entry_high")
+        if entry_low is not None and entry_high is not None and entry_low > entry_high:
+            raise ValueError("entry_low must be less than or equal to entry_high")
+
+    @staticmethod
+    def _optional_int(value: Any, field_name: str) -> Optional[int]:
+        if value in (None, ""):
+            return None
+        try:
+            return int(value)
+        except (TypeError, ValueError) as exc:
+            raise ValueError(f"{field_name} must be an integer") from exc
+
+    @staticmethod
+    def _parse_datetime(value: Any) -> Optional[datetime]:
+        if value in (None, ""):
+            return None
+        if isinstance(value, datetime):
+            return to_utc_naive_datetime(value)
+        if isinstance(value, str):
+            text = value.strip()
+            if not text:
+                return None
+            try:
+                parsed = datetime.fromisoformat(text.replace("Z", "+00:00"))
+            except ValueError as exc:
+                raise ValueError(f"invalid datetime value: {value}") from exc
+            return to_utc_naive_datetime(parsed)
+        raise ValueError(f"invalid datetime value: {value}")
+
+    @classmethod
+    def _is_expired(cls, expires_at: Optional[datetime]) -> bool:
+        normalized_expires_at = cls._parse_datetime(expires_at)
+        return normalized_expires_at is not None and normalized_expires_at <= utc_naive_now()
+
+    @staticmethod
+    def _json_dumps(value: Any) -> Optional[str]:
+        if value in (None, ""):
+            return None
+        sanitized = sanitize_decision_signal_payload(value)
+        return json.dumps(sanitized, ensure_ascii=False, sort_keys=True, default=str)
+
+    @staticmethod
+    def _json_loads(value: Optional[str], *, signal_id: int, field_name: str) -> Any:
+        if not value:
+            return None
+        try:
+            return json.loads(value)
+        except json.JSONDecodeError as exc:
+            logger.warning(
+                "Invalid decision signal JSON: id=%s field=%s error=%s",
+                signal_id,
+                field_name,
+                exc,
+            )
+            raise DecisionSignalStorageError(
+                f"invalid persisted JSON for decision signal {signal_id} field {field_name}"
+            ) from exc
+
+    def _serialize(self, row: DecisionSignalRecord) -> Dict[str, Any]:
+        return {
+            "id": row.id,
+            "stock_code": row.stock_code,
+            "stock_name": row.stock_name,
+            "market": row.market,
+            "source_type": row.source_type,
+            "source_agent": row.source_agent,
+            "source_report_id": row.source_report_id,
+            "trace_id": row.trace_id,
+            "market_phase": row.market_phase,
+            "trigger_source": row.trigger_source,
+            "action": row.action,
+            "action_label": row.action_label,
+            "confidence": row.confidence,
+            "score": row.score,
+            "horizon": row.horizon,
+            "entry_low": row.entry_low,
+            "entry_high": row.entry_high,
+            "stop_loss": row.stop_loss,
+            "target_price": row.target_price,
+            "invalidation": row.invalidation,
+            "watch_conditions": row.watch_conditions,
+            "reason": row.reason,
+            "risk_summary": row.risk_summary,
+            "catalyst_summary": row.catalyst_summary,
+            "evidence": self._json_loads(row.evidence_json, signal_id=row.id, field_name="evidence_json"),
+            "data_quality_summary": self._json_loads(
+                row.data_quality_summary_json,
+                signal_id=row.id,
+                field_name="data_quality_summary_json",
+            ),
+            "plan_quality": row.plan_quality,
+            "status": row.status,
+            "expires_at": row.expires_at.isoformat() if row.expires_at else None,
+            "created_at": row.created_at.isoformat() if row.created_at else None,
+            "updated_at": row.updated_at.isoformat() if row.updated_at else None,
+            "metadata": self._json_loads(row.metadata_json, signal_id=row.id, field_name="metadata_json"),
+        }

src/storage.py

@@ -19,7 +19,7 @@ import logging
 import re
 import threading
 import time
-from datetime import datetime, date, timedelta
+from datetime import datetime, date, timedelta, timezone
 from typing import Optional, List, Dict, Any, TYPE_CHECKING, Tuple, Callable, TypeVar, Union
 
 import pandas as pd
@@ -66,6 +66,18 @@ if TYPE_CHECKING:
     from src.search_service import SearchResponse
 
 
+def utc_naive_now() -> datetime:
+    """Return current UTC time without tzinfo for SQLite DateTime columns."""
+    return datetime.now(timezone.utc).replace(tzinfo=None)
+
+
+def to_utc_naive_datetime(value: datetime) -> datetime:
+    """Normalize aware datetimes to UTC-naive; treat naive values as UTC-naive."""
+    if value.tzinfo is not None and value.utcoffset() is not None:
+        return value.astimezone(timezone.utc).replace(tzinfo=None)
+    return value
+
+
 # === 数据模型定义 ===
 
 class DatabaseSchemaMigration(Base):
@@ -774,6 +786,70 @@ class AlertCooldownRecord(Base):
     )
 
 
+class DecisionSignalRecord(Base):
+    """Persisted AI decision signal asset for Issue #1390 P1."""
+
+    __tablename__ = 'decision_signals'
+
+    id = Column(Integer, primary_key=True, autoincrement=True)
+    stock_code = Column(String(16), nullable=False, index=True)
+    stock_name = Column(String(64))
+    market = Column(String(8), nullable=False, index=True)
+    source_type = Column(String(32), nullable=False, index=True)
+    source_agent = Column(String(64))
+    source_report_id = Column(Integer, index=True)
+    trace_id = Column(String(64), index=True)
+    market_phase = Column(String(24), index=True)
+    trigger_source = Column(String(64), nullable=False, index=True)
+    action = Column(String(16), nullable=False, index=True)
+    action_label = Column(String(32))
+    confidence = Column(Float)
+    score = Column(Integer)
+    horizon = Column(String(16), index=True)
+    entry_low = Column(Float)
+    entry_high = Column(Float)
+    stop_loss = Column(Float)
+    target_price = Column(Float)
+    invalidation = Column(Text)
+    watch_conditions = Column(Text)
+    reason = Column(Text)
+    risk_summary = Column(Text)
+    catalyst_summary = Column(Text)
+    evidence_json = Column(Text)
+    data_quality_summary_json = Column(Text)
+    plan_quality = Column(String(16), nullable=False, default='unknown', index=True)
+    status = Column(String(16), nullable=False, default='active', index=True)
+    expires_at = Column(DateTime, index=True)
+    created_at = Column(DateTime, default=utc_naive_now, index=True)
+    updated_at = Column(DateTime, default=utc_naive_now, onupdate=utc_naive_now, index=True)
+    metadata_json = Column(Text)
+
+    __table_args__ = (
+        Index('ix_decision_signal_stock_status_time', 'stock_code', 'status', 'created_at'),
+        Index('ix_decision_signal_market_status_time', 'market', 'status', 'created_at'),
+        Index(
+            'ix_decision_signal_report_type_market_stock_action_horizon_phase',
+            'source_report_id',
+            'source_type',
+            'market',
+            'stock_code',
+            'action',
+            'horizon',
+            'market_phase',
+        ),
+        Index(
+            'ix_decision_signal_trace_type_market_stock_action_horizon_phase',
+            'trace_id',
+            'source_type',
+            'market',
+            'stock_code',
+            'action',
+            'horizon',
+            'market_phase',
+        ),
+    )
+
+
 class _DatabaseManagerMeta(type):
     """Serialize DatabaseManager construction across __new__ and __init__."""
 
@@ -1633,7 +1709,9 @@ class DatabaseManager(metaclass=_DatabaseManagerMeta):
         """
         删除指定的分析历史记录。
 
-        同时清理依赖这些历史记录的回测结果，避免外键约束失败。
+        同时清理依赖这些历史记录的回测结果和分析来源决策信号，避免
+        依赖历史记录的派生数据残留。DecisionSignal 的 source_report_id
+        允许弱引用，因此这里只清理 source_type=analysis 的真实历史绑定信号。
 
         Args:
             record_ids: 要删除的历史记录主键 ID 列表
@@ -1646,11 +1724,27 @@ class DatabaseManager(metaclass=_DatabaseManagerMeta):
             return 0
 
         with self.session_scope() as session:
+            existing_ids = sorted(
+                session.execute(
+                    select(AnalysisHistory.id).where(AnalysisHistory.id.in_(ids))
+                ).scalars().all()
+            )
+            if not existing_ids:
+                return 0
+
             session.execute(
-                delete(BacktestResult).where(BacktestResult.analysis_history_id.in_(ids))
+                delete(DecisionSignalRecord).where(
+                    and_(
+                        DecisionSignalRecord.source_type == "analysis",
+                        DecisionSignalRecord.source_report_id.in_(existing_ids),
+                    )
+                )
+            )
+            session.execute(
+                delete(BacktestResult).where(BacktestResult.analysis_history_id.in_(existing_ids))
             )
             result = session.execute(
-                delete(AnalysisHistory).where(AnalysisHistory.id.in_(ids))
+                delete(AnalysisHistory).where(AnalysisHistory.id.in_(existing_ids))
             )
             return result.rowcount or 0
 

src/utils/sanitize.py

@@ -5,6 +5,7 @@ from __future__ import annotations
 
 import re
 from typing import Any
+from urllib.parse import parse_qsl, urlsplit
 
 
 _REDACTED = "[REDACTED]"
@@ -49,6 +50,28 @@ _SENSITIVE_COMPACT_KEY_PHRASES = {
 _SENSITIVE_COMPACT_KEY_PATTERN = re.compile(
     r"authorization|cookie|password|secret|sendkey|token(?!s)|webhook"
 )
+_URL_PATTERN = re.compile(r"https?://[^\s,;)\]}]+", re.IGNORECASE)
+_BEARER_PATTERN = re.compile(r"\b(bearer\s+)[^\s,;&]+", re.IGNORECASE)
+_AUTHORIZATION_HEADER_PATTERN = re.compile(
+    r"\b(authorization|proxy[_-]?authorization)(\s*[:=]\s*)"
+    r"(?:(?:Bearer|Basic|Token|Digest)\s+)?[^\s,;&]+",
+    re.IGNORECASE,
+)
+_COOKIE_HEADER_PATTERN = re.compile(
+    r"\b(cookie|set[_-]?cookie)(\s*[:=]\s*)[^\s,;&]+",
+    re.IGNORECASE,
+)
+_SECRET_ASSIGNMENT_PATTERN = re.compile(
+    r"\b(token|secret|password|sendkey|api[_-]?key|apikey|api[_-]?token|auth[_-]?token|"
+    r"access[_-]?token|refresh[_-]?token|session[_-]?token|license[_-]?key|private[_-]?key|"
+    r"secret[_-]?key|webhook[_-]?url|authorization|proxy[_-]?authorization|cookie|set[_-]?cookie)"
+    r"([=:]\s*)[^\s,;&]+",
+    re.IGNORECASE,
+)
+_TOKEN_LIKE_PATTERN = re.compile(
+    r"\b(?:sk-[a-z0-9_\-]{16,}|xox[baprs]-[a-z0-9\-]{16,}|gh[pousr]_[a-z0-9_]{20,})\b",
+    re.IGNORECASE,
+)
 
 
 def sanitize_diagnostic_text(text: Any, *, max_length: int = 300) -> str:
@@ -56,7 +79,9 @@ def sanitize_diagnostic_text(text: Any, *, max_length: int = 300) -> str:
     sanitized = str(text or "").strip()
     if not sanitized:
         return ""
-    sanitized = re.sub(r"(?i)(bearer\s+)[a-z0-9._\-:]+", r"\1[REDACTED]", sanitized)
+    sanitized = _AUTHORIZATION_HEADER_PATTERN.sub(r"\1\2[REDACTED]", sanitized)
+    sanitized = _COOKIE_HEADER_PATTERN.sub(r"\1\2[REDACTED]", sanitized)
+    sanitized = _BEARER_PATTERN.sub(r"\1[REDACTED]", sanitized)
     sanitized = re.sub(r"(?i)(token|secret|password|sendkey)([=:]\s*)[^\s,;&]+", r"\1\2[REDACTED]", sanitized)
     sanitized = re.sub(r"https?://[^\s]+", "[REDACTED_URL]", sanitized)
     return " ".join(sanitized.split())[:max_length]
@@ -81,6 +106,103 @@ def redact_sensitive_mapping(obj: Any) -> Any:
     return obj
 
 
+def sanitize_decision_signal_text(text: Any) -> str:
+    """Redact obvious secrets from persisted decision-signal text without truncating."""
+    sanitized = str(text or "").strip()
+    if not sanitized:
+        return ""
+    sanitized = _URL_PATTERN.sub(_redact_sensitive_url_match, sanitized)
+    sanitized = _AUTHORIZATION_HEADER_PATTERN.sub(r"\1\2[REDACTED]", sanitized)
+    sanitized = _COOKIE_HEADER_PATTERN.sub(r"\1\2[REDACTED]", sanitized)
+    sanitized = _BEARER_PATTERN.sub(r"\1[REDACTED]", sanitized)
+    sanitized = _SECRET_ASSIGNMENT_PATTERN.sub(r"\1\2[REDACTED]", sanitized)
+    sanitized = _TOKEN_LIKE_PATTERN.sub("[REDACTED]", sanitized)
+    return " ".join(sanitized.split())
+
+
+def sanitize_decision_signal_payload(obj: Any) -> Any:
+    """Redact decision-signal JSON payloads by sensitive keys and string values."""
+    redacted = redact_sensitive_mapping(obj)
+    return _sanitize_decision_signal_payload_values(redacted)
+
+
+def _sanitize_decision_signal_payload_values(obj: Any) -> Any:
+    if isinstance(obj, dict):
+        return {
+            key: _sanitize_decision_signal_payload_values(value)
+            for key, value in obj.items()
+        }
+    if isinstance(obj, list):
+        return [_sanitize_decision_signal_payload_values(item) for item in obj]
+    if isinstance(obj, str):
+        return sanitize_decision_signal_text(obj)
+    return obj
+
+
+def _redact_sensitive_url_match(match: re.Match[str]) -> str:
+    url = match.group(0)
+    if _is_sensitive_url(url):
+        return "[REDACTED_URL]"
+    return url
+
+
+def _is_sensitive_url(url: str) -> bool:
+    if _TOKEN_LIKE_PATTERN.search(url):
+        return True
+    try:
+        parsed = urlsplit(url)
+    except ValueError:
+        return False
+    if parsed.username or parsed.password:
+        return True
+    if _is_webhook_url(parsed.hostname or "", parsed.path):
+        return True
+    return (
+        _has_sensitive_url_params(parsed.query)
+        or _has_sensitive_url_params(parsed.fragment)
+    )
+
+
+def _is_webhook_url(hostname: str, path: str) -> bool:
+    hostname = str(hostname or "").lower().strip(".")
+    normalized_path = f"/{path.lstrip('/').lower()}"
+    path_segments = [segment for segment in normalized_path.split("/") if segment]
+
+    if hostname == "hooks.slack.com" and normalized_path.startswith("/services/"):
+        return True
+    if hostname in {"discord.com", "discordapp.com"} and "/api/webhooks/" in normalized_path:
+        return True
+    if hostname == "open.feishu.cn" and "/open-apis/bot/" in normalized_path and "/hook/" in normalized_path:
+        return True
+    if hostname == "oapi.dingtalk.com" and normalized_path.startswith("/robot/send"):
+        return True
+    if hostname == "qyapi.weixin.qq.com" and normalized_path.startswith("/cgi-bin/webhook/send"):
+        return True
+    if hostname in {"sctapi.ftqq.com", "sc.ftqq.com"}:
+        return True
+    if hostname.startswith("hooks."):
+        return True
+    if {"hook", "webhook", "webhooks"} & set(path_segments):
+        return True
+    return False
+
+
+def _has_sensitive_url_params(params_text: str) -> bool:
+    if not params_text:
+        return False
+    try:
+        params = parse_qsl(params_text, keep_blank_values=True)
+    except ValueError:
+        return False
+    for key, value in params:
+        key_text = str(key or "").strip().lower()
+        if _is_sensitive_mapping_key(key_text):
+            return True
+        if _TOKEN_LIKE_PATTERN.search(str(value or "")):
+            return True
+    return False
+
+
 def _is_sensitive_mapping_key(key: Any) -> bool:
     key_text = str(key or "").strip()
     if not key_text:

tests/test_analysis_history.py

@@ -34,7 +34,7 @@ except ModuleNotFoundError:
     get_stock_bar = None
 
 from src.config import Config
-from src.storage import DatabaseManager, AnalysisHistory, BacktestResult
+from src.storage import DatabaseManager, AnalysisHistory, BacktestResult, DecisionSignalRecord
 from src.analyzer import AnalysisResult
 from src.services.history_service import HistoryService
 import src.auth as auth
@@ -1395,8 +1395,8 @@ class AnalysisHistoryTestCase(unittest.TestCase):
         self.assertIn("✅Safe", markdown)
         self.assertNotIn("🚨Safe", markdown)
 
-    def test_delete_analysis_history_records_also_cleans_backtests(self) -> None:
-        """删除历史记录时应一并清理关联回测结果。"""
+    def test_delete_analysis_history_records_also_cleans_backtests_and_decision_signals(self) -> None:
+        """删除历史记录时应一并清理关联回测结果和决策信号。"""
         record_id = self._save_history("query_delete_001")
 
         with self.db.session_scope() as session:
@@ -1408,6 +1408,36 @@ class AnalysisHistoryTestCase(unittest.TestCase):
                 engine_version="v1",
                 eval_status="pending",
             ))
+            session.add(DecisionSignalRecord(
+                stock_code="600519",
+                stock_name="贵州茅台",
+                market="cn",
+                source_type="analysis",
+                source_report_id=record_id,
+                trace_id="trace-delete-linked",
+                market_phase="intraday",
+                trigger_source="api",
+                action="buy",
+                action_label="买入",
+                reason="linked",
+                plan_quality="minimal",
+                status="active",
+            ))
+            session.add(DecisionSignalRecord(
+                stock_code="000001",
+                stock_name="平安银行",
+                market="cn",
+                source_type="analysis",
+                source_report_id=record_id + 999,
+                trace_id="trace-delete-unrelated",
+                market_phase="intraday",
+                trigger_source="api",
+                action="watch",
+                action_label="观望",
+                reason="unrelated",
+                plan_quality="minimal",
+                status="active",
+            ))
 
         deleted = self.db.delete_analysis_history_records([record_id])
         self.assertEqual(deleted, 1)
@@ -1418,6 +1448,166 @@ class AnalysisHistoryTestCase(unittest.TestCase):
                 session.query(BacktestResult).filter(BacktestResult.analysis_history_id == record_id).count(),
                 0,
             )
+            self.assertEqual(
+                session.query(DecisionSignalRecord).filter(DecisionSignalRecord.source_report_id == record_id).count(),
+                0,
+            )
+            self.assertEqual(
+                session.query(DecisionSignalRecord).filter(DecisionSignalRecord.trace_id == "trace-delete-unrelated").count(),
+                1,
+            )
+
+    def test_delete_analysis_history_records_keeps_signals_for_nonexistent_history_id(self) -> None:
+        """不存在的历史 ID 不应触发弱关联 DecisionSignal 清理。"""
+        missing_id = 987654321
+
+        with self.db.session_scope() as session:
+            session.add(DecisionSignalRecord(
+                stock_code="600519",
+                stock_name="贵州茅台",
+                market="cn",
+                source_type="manual",
+                source_report_id=missing_id,
+                trace_id="trace-delete-missing-history",
+                market_phase="intraday",
+                trigger_source="api",
+                action="watch",
+                action_label="观望",
+                reason="manual signal with unverified report id",
+                plan_quality="minimal",
+                status="active",
+            ))
+
+        deleted = self.db.delete_analysis_history_records([missing_id])
+        self.assertEqual(deleted, 0)
+
+        with self.db.get_session() as session:
+            self.assertEqual(
+                session.query(DecisionSignalRecord).filter(
+                    DecisionSignalRecord.trace_id == "trace-delete-missing-history"
+                ).count(),
+                1,
+            )
+
+    def test_delete_analysis_history_records_keeps_manual_signal_with_same_report_id(self) -> None:
+        """source_report_id 是弱引用，真实 history 删除不应误删 manual/pre-report 信号。"""
+        record_id = self._save_history("query_delete_manual_collision")
+
+        with self.db.session_scope() as session:
+            session.add(DecisionSignalRecord(
+                stock_code="600519",
+                stock_name="贵州茅台",
+                market="cn",
+                source_type="analysis",
+                source_report_id=record_id,
+                trace_id="trace-delete-analysis-bound",
+                market_phase="intraday",
+                trigger_source="api",
+                action="buy",
+                action_label="买入",
+                reason="history-bound signal",
+                plan_quality="minimal",
+                status="active",
+            ))
+            session.add(DecisionSignalRecord(
+                stock_code="600519",
+                stock_name="贵州茅台",
+                market="cn",
+                source_type="manual",
+                source_report_id=record_id,
+                trace_id="trace-delete-manual-weak-ref",
+                market_phase="intraday",
+                trigger_source="api",
+                action="watch",
+                action_label="观望",
+                reason="manual signal with caller-supplied report id",
+                plan_quality="minimal",
+                status="active",
+            ))
+
+        deleted = self.db.delete_analysis_history_records([record_id])
+        self.assertEqual(deleted, 1)
+
+        with self.db.get_session() as session:
+            self.assertEqual(
+                session.query(DecisionSignalRecord).filter(
+                    DecisionSignalRecord.trace_id == "trace-delete-analysis-bound"
+                ).count(),
+                0,
+            )
+            self.assertEqual(
+                session.query(DecisionSignalRecord).filter(
+                    DecisionSignalRecord.trace_id == "trace-delete-manual-weak-ref"
+                ).count(),
+                1,
+            )
+
+    def test_delete_analysis_history_records_cleans_only_existing_ids_in_mixed_batch(self) -> None:
+        """混合存在/不存在 ID 时，只清理实际存在历史记录的关联数据。"""
+        record_id = self._save_history("query_delete_mixed")
+        missing_id = record_id + 987654
+
+        with self.db.session_scope() as session:
+            session.add(BacktestResult(
+                analysis_history_id=record_id,
+                code="600519",
+                analysis_date=None,
+                eval_window_days=10,
+                engine_version="v1",
+                eval_status="pending",
+            ))
+            session.add(DecisionSignalRecord(
+                stock_code="600519",
+                stock_name="贵州茅台",
+                market="cn",
+                source_type="analysis",
+                source_report_id=record_id,
+                trace_id="trace-delete-mixed-linked",
+                market_phase="intraday",
+                trigger_source="api",
+                action="buy",
+                action_label="买入",
+                reason="linked",
+                plan_quality="minimal",
+                status="active",
+            ))
+            session.add(DecisionSignalRecord(
+                stock_code="000001",
+                stock_name="平安银行",
+                market="cn",
+                source_type="manual",
+                source_report_id=missing_id,
+                trace_id="trace-delete-mixed-missing",
+                market_phase="intraday",
+                trigger_source="api",
+                action="watch",
+                action_label="观望",
+                reason="weak report id collision",
+                plan_quality="minimal",
+                status="active",
+            ))
+
+        deleted = self.db.delete_analysis_history_records([record_id, missing_id])
+        self.assertEqual(deleted, 1)
+
+        with self.db.get_session() as session:
+            self.assertIsNone(session.query(AnalysisHistory).filter(AnalysisHistory.id == record_id).first())
+            self.assertEqual(
+                session.query(BacktestResult).filter(BacktestResult.analysis_history_id == record_id).count(),
+                0,
+            )
+            self.assertEqual(
+                session.query(DecisionSignalRecord).filter(
+                    DecisionSignalRecord.trace_id == "trace-delete-mixed-linked"
+                ).count(),
+                0,
+            )
+            self.assertEqual(
+                session.query(DecisionSignalRecord).filter(
+                    DecisionSignalRecord.trace_id == "trace-delete-mixed-missing"
+                ).count(),
+                1,
+            )
 
     @patch("src.auth.is_auth_enabled", return_value=False)
     def test_delete_history_api_deletes_selected_records(self, mock_auth) -> None:

tests/test_api_schema_pydantic.py

@@ -1,11 +1,30 @@
 """Regression tests for API schema metadata under Pydantic v2."""
 
+import json
+from pathlib import Path
+
+from api.app import create_app
 from api.v1.schemas.analysis import AnalyzeRequest, MarketReviewRequest
 from api.v1.schemas.common import RootResponse
 from api.v1.schemas.history import HistoryItem
 from api.v1.schemas.stocks import StockQuote
 
 
+DECISION_SIGNAL_PATHS = (
+    "/api/v1/decision-signals",
+    "/api/v1/decision-signals/latest/{stock_code}",
+    "/api/v1/decision-signals/{signal_id}",
+    "/api/v1/decision-signals/{signal_id}/status",
+)
+DECISION_SIGNAL_SCHEMAS = (
+    "DecisionSignalCreateRequest",
+    "DecisionSignalItem",
+    "DecisionSignalListResponse",
+    "DecisionSignalMutationResponse",
+    "DecisionSignalStatusUpdateRequest",
+)
+
+
 def test_schema_examples_remain_in_openapi_schema() -> None:
     root_schema = RootResponse.model_json_schema()
     analyze_schema = AnalyzeRequest.model_json_schema()
@@ -66,3 +85,25 @@ def test_analyze_request_rejects_invalid_analysis_phase() -> None:
         assert "analysis_phase" in str(exc)
     else:
         raise AssertionError("invalid analysis_phase should be rejected")
+
+
+def test_decision_signal_static_api_spec_matches_runtime_paths() -> None:
+    static_spec_path = Path(__file__).resolve().parents[1] / "docs" / "architecture" / "api_spec.json"
+    static_spec = json.loads(static_spec_path.read_text(encoding="utf-8"))
+    runtime_spec = create_app().openapi()
+
+    assert static_spec["openapi"] == runtime_spec["openapi"]
+    assert static_spec["info"]["description"] == runtime_spec["info"]["description"]
+    assert "暂无认证要求" not in static_spec["info"]["description"]
+    assert "ADMIN_AUTH_ENABLED=true" in static_spec["info"]["description"]
+    for path in DECISION_SIGNAL_PATHS:
+        assert static_spec["paths"][path] == runtime_spec["paths"][path]
+        for operation in static_spec["paths"][path].values():
+            assert "401" in operation["responses"]
+            assert operation["security"] == [{"AdminSessionCookie": []}]
+    assert static_spec["components"]["securitySchemes"] == runtime_spec["components"]["securitySchemes"]
+    for schema_name in DECISION_SIGNAL_SCHEMAS:
+        assert static_spec["components"]["schemas"][schema_name] == runtime_spec["components"]["schemas"][schema_name]
+
+    status_schema = static_spec["components"]["schemas"]["DecisionSignalStatusUpdateRequest"]["properties"]["status"]
+    assert status_schema["enum"] == ["active", "expired", "invalidated", "closed", "archived"]

tests/test_decision_signal_repo.py

@@ -0,0 +1,370 @@
+# -*- coding: utf-8 -*-
+"""Repository tests for DecisionSignal P1."""
+
+from __future__ import annotations
+
+import os
+from datetime import datetime, timedelta, timezone
+
+import pytest
+from sqlalchemy import inspect
+
+from src.config import Config
+from src.repositories.decision_signal_repo import DecisionSignalRepository
+from src.storage import Base, DatabaseManager, DecisionSignalRecord, utc_naive_now
+
+
+@pytest.fixture()
+def isolated_db(tmp_path):
+    old_database_path = os.environ.get("DATABASE_PATH")
+    db_path = tmp_path / "decision_signal_repo.db"
+    os.environ["DATABASE_PATH"] = str(db_path)
+    Config.reset_instance()
+    DatabaseManager.reset_instance()
+    db = DatabaseManager.get_instance()
+    try:
+        yield db
+    finally:
+        DatabaseManager.reset_instance()
+        Config.reset_instance()
+        if old_database_path is None:
+            os.environ.pop("DATABASE_PATH", None)
+        else:
+            os.environ["DATABASE_PATH"] = old_database_path
+
+
+def _fields(**overrides):
+    fields = {
+        "stock_code": "600519",
+        "stock_name": "贵州茅台",
+        "market": "cn",
+        "source_type": "analysis",
+        "source_agent": "test-agent",
+        "source_report_id": 1001,
+        "trace_id": "trace-1001",
+        "market_phase": "intraday",
+        "trigger_source": "api",
+        "action": "buy",
+        "action_label": "买入",
+        "confidence": 0.8,
+        "score": 88,
+        "horizon": "3d",
+        "entry_low": 1680.0,
+        "entry_high": 1700.0,
+        "stop_loss": 1600.0,
+        "target_price": 1850.0,
+        "invalidation": "跌破 1600",
+        "watch_conditions": "量能继续放大",
+        "reason": "趋势增强",
+        "risk_summary": "波动加大",
+        "catalyst_summary": "业绩披露",
+        "evidence_json": '{"items":[]}',
+        "data_quality_summary_json": '{"level":"good"}',
+        "plan_quality": "complete",
+        "status": "active",
+        "metadata_json": '{"task_id":"task-1"}',
+    }
+    fields.update(overrides)
+    return fields
+
+
+def test_create_if_absent_deduplicates_report_and_trace_keys(isolated_db) -> None:
+    repo = DecisionSignalRepository(isolated_db)
+
+    row1, created1 = repo.create_if_absent(_fields())
+    row2, created2 = repo.create_if_absent(_fields(reason="new reason"))
+    assert created1 is True
+    assert created2 is False
+    assert row2.id == row1.id
+    assert row2.reason == "趋势增强"
+
+    different_horizon, horizon_created = repo.create_if_absent(_fields(horizon="10d", target_price=1900))
+    assert horizon_created is True
+    assert different_horizon.id != row1.id
+
+    different_phase, phase_created = repo.create_if_absent(_fields(market_phase="premarket", target_price=1800))
+    assert phase_created is True
+    assert different_phase.id != row1.id
+
+    different_market, market_created = repo.create_if_absent(
+        _fields(market="hk", stock_code="600519", target_price=1810)
+    )
+    assert market_created is True
+    assert different_market.id != row1.id
+
+    different_source_type, source_type_created = repo.create_if_absent(
+        _fields(source_type="manual", trace_id="trace-manual", target_price=1820)
+    )
+    assert source_type_created is True
+    assert different_source_type.id != row1.id
+
+    duplicate_manual, duplicate_manual_created = repo.create_if_absent(
+        _fields(source_type="manual", trace_id="trace-manual-new", target_price=1830)
+    )
+    assert duplicate_manual_created is False
+    assert duplicate_manual.id == different_source_type.id
+
+    trace_row1, trace_created1 = repo.create_if_absent(
+        _fields(source_report_id=None, trace_id="trace-only", stock_code="000001")
+    )
+    trace_row2, trace_created2 = repo.create_if_absent(
+        _fields(source_report_id=None, trace_id="trace-only", stock_code="000001", reason="ignored")
+    )
+    assert trace_created1 is True
+    assert trace_created2 is False
+    assert trace_row2.id == trace_row1.id
+
+    trace_horizon_row, trace_horizon_created = repo.create_if_absent(
+        _fields(
+            source_report_id=None,
+            trace_id="trace-only",
+            stock_code="000001",
+            horizon="10d",
+        )
+    )
+    assert trace_horizon_created is True
+    assert trace_horizon_row.id != trace_row1.id
+
+    trace_source_type_row, trace_source_type_created = repo.create_if_absent(
+        _fields(
+            source_type="manual",
+            source_report_id=None,
+            trace_id="trace-only",
+            stock_code="000001",
+        )
+    )
+    assert trace_source_type_created is True
+    assert trace_source_type_row.id != trace_row1.id
+
+    none_dim_row1, none_dim_created1 = repo.create_if_absent(
+        _fields(source_report_id=1002, trace_id="trace-none-dim", horizon=None, market_phase=None)
+    )
+    none_dim_row2, none_dim_created2 = repo.create_if_absent(
+        _fields(source_report_id=1002, trace_id="trace-none-dim", horizon=None, market_phase=None)
+    )
+    assert none_dim_created1 is True
+    assert none_dim_created2 is False
+    assert none_dim_row2.id == none_dim_row1.id
+
+    no_key_row1, no_key_created1 = repo.create_if_absent(
+        _fields(source_report_id=None, trace_id=None, stock_code="000002")
+    )
+    no_key_row2, no_key_created2 = repo.create_if_absent(
+        _fields(source_report_id=None, trace_id=None, stock_code="000002")
+    )
+    assert no_key_created1 is True
+    assert no_key_created2 is True
+    assert no_key_row2.id != no_key_row1.id
+
+
+def test_list_latest_status_update_and_lazy_expire(isolated_db) -> None:
+    repo = DecisionSignalRepository(isolated_db)
+    old_row = repo.create(_fields(source_report_id=2001, trace_id="trace-2001", action="watch"))
+    new_row = repo.create(_fields(source_report_id=2002, trace_id="trace-2002", action="buy"))
+    expired_row = repo.create(
+        _fields(
+            source_report_id=2003,
+            trace_id="trace-2003",
+            action="alert",
+            expires_at=utc_naive_now() - timedelta(minutes=1),
+        )
+    )
+
+    with isolated_db.session_scope() as session:
+        session.query(DecisionSignalRecord).filter_by(id=old_row.id).update(
+            {"created_at": utc_naive_now() - timedelta(days=1)}
+        )
+
+    rows, total = repo.list(stock_codes=["600519"], action="buy", page=1, page_size=10)
+    assert total == 1
+    assert rows[0].id == new_row.id
+
+    latest = repo.get_latest_active(stock_codes=["600519"], limit=2)
+    assert [row.id for row in latest] == [new_row.id, old_row.id]
+    assert repo.get(expired_row.id).status == "expired"
+    assert repo.expire_due_signals() == 0
+
+    latest_after_expire = repo.get_latest_active(stock_codes=["600519"], limit=2)
+    assert [row.id for row in latest_after_expire] == [new_row.id, old_row.id]
+
+    updated = repo.update_status(
+        new_row.id,
+        status="closed",
+        metadata_json='{"closed_by":"test"}',
+        replace_metadata=True,
+    )
+    assert updated.status == "closed"
+    assert updated.metadata_json == '{"closed_by":"test"}'
+    assert repo.update_status(999999, status="closed") is None
+
+
+def test_expire_due_signals_normalizes_aware_now(isolated_db) -> None:
+    repo = DecisionSignalRepository(isolated_db)
+    now_utc = datetime.now(timezone.utc)
+    expired_row = repo.create(
+        _fields(
+            source_report_id=2101,
+            trace_id="trace-aware-now-expired",
+            expires_at=now_utc - timedelta(minutes=1),
+        )
+    )
+    future_row = repo.create(
+        _fields(
+            source_report_id=2102,
+            trace_id="trace-aware-now-future",
+            expires_at=now_utc + timedelta(minutes=1),
+        )
+    )
+
+    assert repo.expire_due_signals(now=now_utc) == 1
+    assert repo.get(expired_row.id).status == "expired"
+    assert repo.get(future_row.id).status == "active"
+
+
+def test_create_and_list_normalize_aware_datetimes(isolated_db) -> None:
+    repo = DecisionSignalRepository(isolated_db)
+
+    row = repo.create(
+        _fields(
+            source_report_id=2201,
+            trace_id="trace-aware-fields",
+            created_at=datetime(2026, 6, 8, 20, 0, tzinfo=timezone.utc),
+            updated_at=datetime(2026, 6, 8, 20, 0, tzinfo=timezone.utc),
+            expires_at=datetime(2099, 1, 1, 0, 0, tzinfo=timezone(timedelta(hours=8))),
+        )
+    )
+
+    assert row.created_at == datetime(2026, 6, 8, 20, 0)
+    assert row.updated_at == datetime(2026, 6, 8, 20, 0)
+    assert row.expires_at == datetime(2098, 12, 31, 16, 0)
+    assert row.created_at.tzinfo is None
+    assert row.updated_at.tzinfo is None
+    assert row.expires_at.tzinfo is None
+
+    rows, total = repo.list(
+        created_from=datetime(2026, 6, 8, 19, 59, tzinfo=timezone.utc),
+        created_to=datetime(2026, 6, 8, 20, 1, tzinfo=timezone.utc),
+        expires_from=datetime(2098, 12, 31, 15, 59, tzinfo=timezone.utc),
+        expires_to=datetime(2098, 12, 31, 16, 1, tzinfo=timezone.utc),
+    )
+    assert total == 1
+    assert rows[0].id == row.id
+
+
+def test_create_if_absent_refreshes_expired_same_key_only_with_future_active(isolated_db) -> None:
+    repo = DecisionSignalRepository(isolated_db)
+    expired_row, expired_created = repo.create_if_absent(
+        _fields(
+            source_report_id=2301,
+            trace_id="trace-refresh-original",
+            status="expired",
+            expires_at=utc_naive_now() - timedelta(days=1),
+            reason="old reason",
+            target_price=1800,
+        )
+    )
+    original_created_at = expired_row.created_at
+
+    refreshed_row, refreshed_created = repo.create_if_absent(
+        _fields(
+            source_report_id=2301,
+            trace_id="trace-refresh-new",
+            source_agent="new-agent",
+            trigger_source="alert",
+            status="active",
+            expires_at=utc_naive_now() + timedelta(days=2),
+            reason="fresh reason",
+            target_price=1900,
+        )
+    )
+
+    assert expired_created is True
+    assert refreshed_created is False
+    assert refreshed_row.id == expired_row.id
+    assert refreshed_row.status == "active"
+    assert refreshed_row.reason == "fresh reason"
+    assert refreshed_row.target_price == 1900
+    assert refreshed_row.source_type == "analysis"
+    assert refreshed_row.source_agent == "test-agent"
+    assert refreshed_row.trace_id == "trace-refresh-original"
+    assert refreshed_row.trigger_source == "api"
+    assert refreshed_row.created_at == original_created_at
+
+    different_source_type_row, different_source_type_created = repo.create_if_absent(
+        _fields(
+            source_report_id=2301,
+            trace_id="trace-refresh-agent",
+            source_type="agent",
+            source_agent="new-agent",
+            trigger_source="alert",
+            status="active",
+            expires_at=utc_naive_now() + timedelta(days=2),
+            reason="agent reason",
+            target_price=1950,
+        )
+    )
+    assert different_source_type_created is True
+    assert different_source_type_row.id != expired_row.id
+    assert different_source_type_row.source_type == "agent"
+
+    past_row, past_created = repo.create_if_absent(
+        _fields(
+            source_report_id=2302,
+            trace_id="trace-refresh-past",
+            status="expired",
+            expires_at=utc_naive_now() - timedelta(days=1),
+            reason="past old",
+        )
+    )
+    still_expired, still_expired_created = repo.create_if_absent(
+        _fields(
+            source_report_id=2302,
+            trace_id="trace-refresh-past-new",
+            status="active",
+            expires_at=utc_naive_now() - timedelta(minutes=1),
+            reason="past fresh",
+        )
+    )
+    assert past_created is True
+    assert still_expired_created is False
+    assert still_expired.id == past_row.id
+    assert still_expired.status == "expired"
+    assert still_expired.reason == "past old"
+
+    closed_row, closed_created = repo.create_if_absent(
+        _fields(
+            source_report_id=2303,
+            trace_id="trace-refresh-closed",
+            status="closed",
+            expires_at=utc_naive_now() - timedelta(days=1),
+            reason="closed old",
+        )
+    )
+    still_closed, still_closed_created = repo.create_if_absent(
+        _fields(
+            source_report_id=2303,
+            trace_id="trace-refresh-closed-new",
+            status="active",
+            expires_at=utc_naive_now() + timedelta(days=2),
+            reason="closed fresh",
+        )
+    )
+    assert closed_created is True
+    assert still_closed_created is False
+    assert still_closed.id == closed_row.id
+    assert still_closed.status == "closed"
+    assert still_closed.reason == "closed old"
+
+
+def test_create_all_is_idempotent_and_indexes_exist(isolated_db) -> None:
+    Base.metadata.create_all(isolated_db._engine)
+    Base.metadata.create_all(isolated_db._engine)
+
+    index_names = {
+        item["name"]
+        for item in inspect(isolated_db._engine).get_indexes("decision_signals")
+    }
+    assert "ix_decision_signal_stock_status_time" in index_names
+    assert "ix_decision_signal_market_status_time" in index_names
+    assert "ix_decision_signal_report_type_market_stock_action_horizon_phase" in index_names
+    assert "ix_decision_signal_trace_type_market_stock_action_horizon_phase" in index_names

tests/test_decision_signal_service.py

@@ -0,0 +1,477 @@
+# -*- coding: utf-8 -*-
+"""Service tests for DecisionSignal P1."""
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+from math import inf, nan
+from pathlib import Path
+
+import pytest
+
+from src.config import Config
+from src.services.decision_signal_service import DecisionSignalService, DecisionSignalStorageError
+from src.storage import DatabaseManager, DecisionSignalRecord
+from src.utils.sanitize import sanitize_decision_signal_text, sanitize_diagnostic_text
+
+
+def test_service_imports_without_api_bootstrap() -> None:
+    repo_root = Path(__file__).resolve().parents[1]
+    result = subprocess.run(
+        [
+            sys.executable,
+            "-c",
+            "from src.services.decision_signal_service import DecisionSignalService; "
+            "print(DecisionSignalService.__name__)",
+        ],
+        cwd=repo_root,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+
+    assert result.returncode == 0, result.stderr
+    assert "DecisionSignalService" in result.stdout
+
+
+@pytest.fixture()
+def isolated_db(tmp_path):
+    old_database_path = os.environ.get("DATABASE_PATH")
+    db_path = tmp_path / "decision_signal_service.db"
+    os.environ["DATABASE_PATH"] = str(db_path)
+    Config.reset_instance()
+    DatabaseManager.reset_instance()
+    db = DatabaseManager.get_instance()
+    try:
+        yield db
+    finally:
+        DatabaseManager.reset_instance()
+        Config.reset_instance()
+        if old_database_path is None:
+            os.environ.pop("DATABASE_PATH", None)
+        else:
+            os.environ["DATABASE_PATH"] = old_database_path
+
+
+def _payload(**overrides):
+    payload = {
+        "stock_code": "SH600519",
+        "stock_name": "贵州茅台",
+        "market": "cn",
+        "source_type": "analysis",
+        "source_report_id": 101,
+        "trace_id": "trace-101",
+        "market_phase": "intraday",
+        "trigger_source": "api",
+        "action": "buy",
+        "confidence": 0.72,
+        "score": 83,
+        "horizon": "3d",
+        "reason": "放量突破",
+    }
+    payload.update(overrides)
+    return payload
+
+
+def test_service_normalizes_fields_and_partial_plan_quality(isolated_db) -> None:
+    service = DecisionSignalService(db_manager=isolated_db)
+
+    result = service.create_signal(
+        _payload(
+            entry_low="1680.5",
+            stop_loss="1600",
+        )
+    )
+
+    item = result["item"]
+    assert result["created"] is True
+    assert item["stock_code"] == "600519"
+    assert item["market"] == "cn"
+    assert item["action"] == "buy"
+    assert item["action_label"] == "买入"
+    assert item["confidence"] == 0.72
+    assert item["score"] == 83
+    assert item["entry_low"] == 1680.5
+    assert item["stop_loss"] == 1600.0
+    assert item["plan_quality"] == "partial"
+
+
+def test_service_plan_quality_slots_and_explicit_override(isolated_db) -> None:
+    service = DecisionSignalService(db_manager=isolated_db)
+
+    minimal = service.create_signal(_payload(source_report_id=201, trace_id="trace-201", entry_low=1680))
+    assert minimal["item"]["plan_quality"] == "minimal"
+
+    complete = service.create_signal(
+        _payload(
+            source_report_id=202,
+            trace_id="trace-202",
+            entry_low=1680,
+            entry_high=1700,
+            stop_loss=1600,
+            target_price=1850,
+            invalidation="跌破 1600",
+        )
+    )
+    assert complete["item"]["plan_quality"] == "complete"
+
+    explicit = service.create_signal(
+        _payload(
+            source_report_id=203,
+            trace_id="trace-203",
+            plan_quality="unknown",
+            entry_low=1680,
+            stop_loss=1600,
+            target_price=1850,
+            invalidation="跌破 1600",
+        )
+    )
+    assert explicit["item"]["plan_quality"] == "unknown"
+
+
+def test_service_rejects_invalid_enums_and_ranges(isolated_db) -> None:
+    service = DecisionSignalService(db_manager=isolated_db)
+
+    with pytest.raises(ValueError, match="market"):
+        service.create_signal(_payload(market="jp"))
+    with pytest.raises(ValueError, match="action"):
+        service.create_signal(_payload(action="strong buy"))
+    with pytest.raises(ValueError, match="confidence"):
+        service.create_signal(_payload(confidence=1.1))
+    with pytest.raises(ValueError, match="score"):
+        service.create_signal(_payload(score=101))
+    with pytest.raises(ValueError, match="trigger_source"):
+        service.create_signal(_payload(trigger_source="x" * 65))
+    with pytest.raises(ValueError, match="trace_id"):
+        service.create_signal(_payload(trace_id="x" * 65))
+    with pytest.raises(ValueError, match="source_agent"):
+        service.create_signal(_payload(source_agent="x" * 65))
+    with pytest.raises(ValueError, match="stock_name"):
+        service.create_signal(_payload(stock_name="x" * 65))
+    with pytest.raises(ValueError, match="action_label"):
+        service.create_signal(_payload(action_label="x" * 33))
+
+
+def test_service_rejects_invalid_price_plan_values(isolated_db) -> None:
+    service = DecisionSignalService(db_manager=isolated_db)
+
+    invalid_cases = [
+        {"entry_low": -1},
+        {"entry_high": 0},
+        {"stop_loss": nan},
+        {"target_price": inf},
+        {"entry_low": "not-a-number"},
+    ]
+    for index, overrides in enumerate(invalid_cases, start=1):
+        with pytest.raises(ValueError):
+            service.create_signal(_payload(source_report_id=300 + index, trace_id=f"trace-price-{index}", **overrides))
+
+    with pytest.raises(ValueError, match="entry_low"):
+        service.create_signal(_payload(source_report_id=306, trace_id="trace-price-range", entry_low=1700, entry_high=1600))
+
+
+def test_decision_signal_sanitizer_redacts_sensitive_url_queries_without_url_tail_leaks() -> None:
+    sanitized = sanitize_decision_signal_text(
+        "plain https://news.example.com/article?id=1 "
+        "signed https://news.example.com/article?token=abc&id=1 "
+        "auth https://news.example.com/article?auth_token=abc&id=2 "
+        "api https://news.example.com/article?api-token=abc&id=3 "
+        "userinfo https://user:pass@example.com/path "
+        "fragment https://news.example.com/cb#access_token=abc "
+        "slack https://hooks.slack.com/services/T000/B000/abc123 "
+        "feishu https://open.feishu.cn/open-apis/bot/v2/hook/abcdef123456 "
+        "wecom https://qyapi.weixin.qq.com/cgi-bin/webhook/send?key=abcdef"
+    )
+
+    assert "https://news.example.com/article?id=1" in sanitized
+    assert sanitized.count("[REDACTED_URL]") == 8
+    assert "token=abc" not in sanitized
+    assert "auth_token=abc" not in sanitized
+    assert "api-token=abc" not in sanitized
+    assert "user:pass" not in sanitized
+    assert "hooks.slack.com" not in sanitized
+    assert "open.feishu.cn" not in sanitized
+    assert "qyapi.weixin.qq.com" not in sanitized
+    assert "]&id=" not in sanitized
+
+
+@pytest.mark.parametrize(
+    ("raw_text", "expected_text", "leaked_fragments"),
+    [
+        (
+            "auth Bearer abcdef0123456789 next",
+            "auth Bearer [REDACTED] next",
+            ("abcdef0123456789", "0123456789"),
+        ),
+        (
+            "jwt Bearer header.payload:signature next",
+            "jwt Bearer [REDACTED] next",
+            ("header.payload:signature", "payload:signature"),
+        ),
+        (
+            "base64 Bearer abc+/def==, next",
+            "base64 Bearer [REDACTED], next",
+            ("abc+/def==", "+/def==", "def=="),
+        ),
+        (
+            "semicolon Bearer abc+/def==; next",
+            "semicolon Bearer [REDACTED]; next",
+            ("abc+/def==", "+/def==", "def=="),
+        ),
+        (
+            "ampersand Bearer abc+/def==&next=1",
+            "ampersand Bearer [REDACTED]&next=1",
+            ("abc+/def==", "+/def==", "def=="),
+        ),
+    ],
+)
+def test_decision_signal_sanitizer_redacts_entire_bearer_token_matrix(
+    raw_text,
+    expected_text,
+    leaked_fragments,
+) -> None:
+    sanitized = sanitize_decision_signal_text(raw_text)
+
+    assert expected_text in sanitized
+    for leaked in leaked_fragments:
+        assert leaked not in sanitized
+
+
+@pytest.mark.parametrize(
+    ("raw_text", "expected_text", "leaked_fragments"),
+    [
+        (
+            "basic Authorization: Basic dXNlcjpwYXNz next",
+            "basic Authorization: [REDACTED] next",
+            ("dXNlcjpwYXNz", "pwYXNz"),
+        ),
+        (
+            "token Authorization: Token abc+/def==; next",
+            "token Authorization: [REDACTED]; next",
+            ("abc+/def==", "+/def==", "def=="),
+        ),
+        (
+            "assignment authorization=secret-value next",
+            "assignment authorization=[REDACTED] next",
+            ("secret-value",),
+        ),
+        (
+            "cookie Cookie: session=abc123; next",
+            "cookie Cookie: [REDACTED]; next",
+            ("session=abc123", "abc123"),
+        ),
+        (
+            "set-cookie Set-Cookie: session=abc123; Path=/ next",
+            "set-cookie Set-Cookie: [REDACTED]; Path=/ next",
+            ("session=abc123", "abc123"),
+        ),
+        (
+            "cookie assignment cookie=session=abc123 next",
+            "cookie assignment cookie=[REDACTED] next",
+            ("session=abc123", "abc123"),
+        ),
+    ],
+)
+def test_decision_signal_sanitizer_redacts_authorization_and_cookie_matrix(
+    raw_text,
+    expected_text,
+    leaked_fragments,
+) -> None:
+    sanitized = sanitize_decision_signal_text(raw_text)
+
+    assert expected_text in sanitized
+    for leaked in leaked_fragments:
+        assert leaked not in sanitized
+
+
+def test_shared_diagnostic_sanitizer_uses_same_auth_credential_boundary() -> None:
+    sanitized = sanitize_diagnostic_text(
+        "Authorization: Bearer abc+/def==; next "
+        "Authorization: Basic dXNlcjpwYXNz "
+        "Cookie: session=abc123"
+    )
+
+    assert "Authorization: [REDACTED]; next" in sanitized
+    assert "Authorization: [REDACTED]" in sanitized
+    assert "Cookie: [REDACTED]" in sanitized
+    for leaked in (
+        "abc+/def==",
+        "+/def==",
+        "def==",
+        "dXNlcjpwYXNz",
+        "pwYXNz",
+        "session=abc123",
+        "abc123",
+    ):
+        assert leaked not in sanitized
+
+
+def test_trace_id_identity_is_not_silently_truncated(isolated_db) -> None:
+    service = DecisionSignalService(db_manager=isolated_db)
+    trace_a = f"{'x' * 63}a"
+    trace_b = f"{'x' * 63}b"
+
+    first = service.create_signal(_payload(source_report_id=None, trace_id=trace_a))
+    second = service.create_signal(_payload(source_report_id=None, trace_id=trace_b))
+
+    assert first["created"] is True
+    assert second["created"] is True
+    assert first["item"]["id"] != second["item"]["id"]
+
+
+def test_trace_id_rejects_sensitive_identity_text(isolated_db) -> None:
+    service = DecisionSignalService(db_manager=isolated_db)
+
+    with pytest.raises(ValueError, match="trace_id"):
+        service.create_signal(_payload(trace_id="Bearer abc+/def=="))
+
+    with pytest.raises(ValueError, match="trace_id"):
+        service.create_signal(_payload(trace_id="Authorization: Basic dXNlcjpwYXNz"))
+
+    with pytest.raises(ValueError, match="trace_id"):
+        service.create_signal(_payload(trace_id="cookie=session=abc123"))
+
+    with pytest.raises(ValueError, match="trace_id"):
+        service.create_signal(_payload(trace_id="https://hooks.example.com/send"))
+
+
+def test_service_sanitizes_public_short_fields_before_persisting(isolated_db) -> None:
+    service = DecisionSignalService(db_manager=isolated_db)
+
+    result = service.create_signal(
+        _payload(
+            stock_name="secret=plain-secret",
+            source_agent="Bearer abc+/def==",
+            trigger_source="Bearer abc+/def==",
+            action_label="token=abc",
+        )
+    )
+
+    item = result["item"]
+    assert item["stock_name"] == "secret=[REDACTED]"
+    assert item["source_agent"] == "Bearer [REDACTED]"
+    assert item["trigger_source"] == "Bearer [REDACTED]"
+    assert item["action_label"] == "token=[REDACTED]"
+    assert "plain-secret" not in str(item)
+    assert "abc+/def==" not in str(item)
+
+    with isolated_db.get_session() as session:
+        row = session.query(DecisionSignalRecord).filter_by(id=item["id"]).one()
+        stored_blob = " ".join(
+            str(value or "")
+            for value in (
+                row.stock_name,
+                row.source_agent,
+                row.trigger_source,
+                row.action_label,
+            )
+        )
+    assert "plain-secret" not in stored_blob
+    assert "abc+/def==" not in stored_blob
+    assert "Bearer [REDACTED]" in stored_blob
+
+
+def test_service_sanitizes_text_and_json_before_persisting(isolated_db) -> None:
+    service = DecisionSignalService(db_manager=isolated_db)
+    long_text = "x" * 450
+
+    result = service.create_signal(
+        _payload(
+            reason=f"{long_text} Bearer abc.def.ghi https://hooks.example.com/send",
+            risk_summary="api_key=sk-1234567890abcdef123456",
+            invalidation={"token": "plain-secret", "note": "secret=keepout"},
+            watch_conditions=["watch https://example.com/path"],
+            evidence={
+                "webhook_url": "https://secret.example.com/hook",
+                "source_url": "https://news.example.com/article?id=1",
+                "signed_url": "https://news.example.com/article?token=abc&id=1",
+                "auth_url": "https://news.example.com/article?auth_token=abc&id=2",
+                "hyphen_signed_url": "https://news.example.com/article?api-key=abc",
+                "slack": "https://hooks.slack.com/services/T000/B000/abcdef",
+                "feishu": "https://open.feishu.cn/open-apis/bot/v2/hook/abcdef",
+                "userinfo": "https://user:pass@example.com/path",
+                "fragment": "https://news.example.com/cb#access_token=abc",
+                "note": "Bearer abc+/def==",
+                "auth_header": "Authorization: Basic dXNlcjpwYXNz",
+                "cookie_header": "Cookie: session=abc123",
+            },
+            metadata={
+                "access_token": "abc",
+                "callback": "https://example.com/cb",
+                "auth_assignment": "authorization=secret-value",
+            },
+        )
+    )
+
+    item = result["item"]
+    assert len(item["reason"]) > 300
+    response_blob = str(item)
+    assert "hooks.example.com" not in response_blob
+    assert "news.example.com/article?id=1" in response_blob
+    assert "example.com/cb" in response_blob
+    assert "secret.example.com" not in response_blob
+    assert "hooks.slack.com" not in response_blob
+    assert "open.feishu.cn" not in response_blob
+    assert "user:pass" not in response_blob
+    assert "access_token=abc" not in response_blob
+    assert "token=abc" not in response_blob
+    assert "auth_token=abc" not in response_blob
+    assert "api-key=abc" not in response_blob
+    assert "]&id=" not in response_blob
+    assert "plain-secret" not in response_blob
+    assert "abc+/def==" not in response_blob
+    assert "+/def==" not in response_blob
+    assert "dXNlcjpwYXNz" not in response_blob
+    assert "pwYXNz" not in response_blob
+    assert "session=abc123" not in response_blob
+    assert "secret-value" not in response_blob
+    assert "sk-1234567890abcdef123456" not in response_blob
+    assert "[REDACTED" in response_blob
+
+    with isolated_db.get_session() as session:
+        row = session.query(DecisionSignalRecord).filter_by(id=item["id"]).one()
+        stored_blob = " ".join(
+            str(value or "")
+            for value in (
+                row.reason,
+                row.risk_summary,
+                row.invalidation,
+                row.watch_conditions,
+                row.evidence_json,
+                row.metadata_json,
+            )
+        )
+    assert "hooks.example.com" not in stored_blob
+    assert "news.example.com/article?id=1" in stored_blob
+    assert "hooks.slack.com" not in stored_blob
+    assert "open.feishu.cn" not in stored_blob
+    assert "user:pass" not in stored_blob
+    assert "access_token=abc" not in stored_blob
+    assert "token=abc" not in stored_blob
+    assert "auth_token=abc" not in stored_blob
+    assert "api-key=abc" not in stored_blob
+    assert "]&id=" not in stored_blob
+    assert "plain-secret" not in stored_blob
+    assert "abc+/def==" not in stored_blob
+    assert "+/def==" not in stored_blob
+    assert "dXNlcjpwYXNz" not in stored_blob
+    assert "pwYXNz" not in stored_blob
+    assert "session=abc123" not in stored_blob
+    assert "secret-value" not in stored_blob
+    assert "sk-1234567890abcdef123456" not in stored_blob
+
+
+def test_service_raises_on_corrupt_persisted_json(isolated_db) -> None:
+    service = DecisionSignalService(db_manager=isolated_db)
+    result = service.create_signal(_payload(source_report_id=351, trace_id="trace-351"))
+    signal_id = result["item"]["id"]
+
+    with isolated_db.get_session() as session:
+        row = session.get(DecisionSignalRecord, signal_id)
+        row.evidence_json = "{not valid json"
+        session.commit()
+
+    with pytest.raises(DecisionSignalStorageError, match="invalid persisted JSON"):
+        service.get_signal(signal_id)