Fix agent fallback, portfolio pricing, and task queue tests (#1144)

Author: ZhuLinsen

api/v1/schemas/portfolio.py

@@ -160,6 +160,11 @@ class PortfolioPositionItem(BaseModel):
     market_value_base: float
     unrealized_pnl_base: float
     valuation_currency: str
+    price_source: str = "unknown"
+    price_provider: Optional[str] = None
+    price_date: Optional[str] = None
+    price_stale: bool = False
+    price_available: bool = True
 
 
 class PortfolioAccountSnapshot(BaseModel):

apps/dsa-web/src/pages/PortfolioPage.tsx

@@ -80,6 +80,31 @@ function formatPct(value: number | undefined | null): string {
   return `${value.toFixed(2)}%`;
 }
 
+function hasPositionPrice(row: PortfolioPositionItem): boolean {
+  return row.priceAvailable !== false && row.priceSource !== 'missing';
+}
+
+function formatPositionPrice(row: PortfolioPositionItem): string {
+  if (!hasPositionPrice(row)) return '--';
+  return row.lastPrice.toFixed(4);
+}
+
+function formatPositionMoney(value: number, row: PortfolioPositionItem): string {
+  if (!hasPositionPrice(row)) return '--';
+  return formatMoney(value, row.valuationCurrency);
+}
+
+function getPositionPriceLabel(row: PortfolioPositionItem): string {
+  if (!hasPositionPrice(row)) return '缺价';
+  if (row.priceSource === 'realtime_quote') {
+    return row.priceProvider ? `实时价 · ${row.priceProvider}` : '实时价';
+  }
+  if (row.priceSource === 'history_close') {
+    return row.priceStale && row.priceDate ? `收盘价 · ${row.priceDate}` : '收盘价';
+  }
+  return row.priceSource || '未知来源';
+}
+
 function formatSideLabel(value: PortfolioSide): string {
   return value === 'buy' ? '买入' : '卖出';
 }
@@ -979,10 +1004,23 @@ const PortfolioPage: React.FC = () => {
                       <td className="py-2 pr-2 font-mono text-foreground">{row.symbol}</td>
                       <td className="py-2 pr-2 text-right">{row.quantity.toFixed(2)}</td>
                       <td className="py-2 pr-2 text-right">{row.avgCost.toFixed(4)}</td>
-                      <td className="py-2 pr-2 text-right">{row.lastPrice.toFixed(4)}</td>
-                      <td className="py-2 pr-2 text-right">{formatMoney(row.marketValueBase, row.valuationCurrency)}</td>
-                      <td className={`py-2 text-right ${row.unrealizedPnlBase >= 0 ? 'text-success' : 'text-danger'}`}>
-                        {formatMoney(row.unrealizedPnlBase, row.valuationCurrency)}
+                      <td className="py-2 pr-2 text-right">
+                        <div>{formatPositionPrice(row)}</div>
+                        <div className={`text-[11px] ${hasPositionPrice(row) ? 'text-secondary' : 'text-warning'}`}>
+                          {getPositionPriceLabel(row)}
+                        </div>
+                      </td>
+                      <td className="py-2 pr-2 text-right">{formatPositionMoney(row.marketValueBase, row)}</td>
+                      <td
+                        className={`py-2 text-right ${
+                          hasPositionPrice(row)
+                            ? row.unrealizedPnlBase >= 0
+                              ? 'text-success'
+                              : 'text-danger'
+                            : 'text-secondary'
+                        }`}
+                      >
+                        {formatPositionMoney(row.unrealizedPnlBase, row)}
                       </td>
                     </tr>
                   ))}

apps/dsa-web/src/types/portfolio.ts

@@ -38,6 +38,11 @@ export interface PortfolioPositionItem {
   marketValueBase: number;
   unrealizedPnlBase: number;
   valuationCurrency: string;
+  priceSource?: 'realtime_quote' | 'history_close' | 'missing' | string;
+  priceProvider?: string | null;
+  priceDate?: string | null;
+  priceStale?: boolean;
+  priceAvailable?: boolean;
 }
 
 export interface PortfolioAccountSnapshot {

docs/CHANGELOG.md

@@ -16,6 +16,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 - [测试] 新增 `tests/test_bot_market_command.py`，覆盖 `MARKET_REVIEW_REGION=both` + open markets `{"cn","us"}` / `{"cn","hk"}` 的 `override_region` 透传断言，并覆盖全市场休市跳过与关闭交易日检查路径；新增 `tests/test_yfinance_hk_indices.py` 覆盖港股指数符号映射与部分/全部失败降级路径。
 - [修复] 问股 Agent 在未配置可用 LLM 时保留后端真实错误原因并维持 `done.success=false` 失败语义，避免前端把配置缺失误当成成功回答。
 - [文档] 补充 LLM 配置指南与 FAQ，明确问股 Agent 对 `LITELLM_CONFIG` / `LLM_CHANNELS` / legacy `GEMINI_*` `OPENAI_*` `ANTHROPIC_*` 的兼容优先级、回退路径与“不静默迁移旧配置”的结论。
+- [修复] Agent 模式未生成有效决策仪表盘时保留本地趋势分析的评分、趋势和操作建议，并将强买/强卖 fallback 归一到兼容的 `buy`/`sell` 决策类型，避免首页结果被 `50 / 观望 / 未知` 缺省值覆盖。
+- [修复] 持仓快照现价缺失时不再静默回退为持仓成本；当天快照优先使用历史收盘价，仅在缺失时使用实时价 fallback，缺价持仓不再污染市值与未实现盈亏汇总，并为持仓明细返回价格来源、日期、stale 与缺价状态。
+- [测试] 补齐 `task_queue` 轻量导入 stub 的股票代码规范化函数，恢复 `tests/test_task_queue_config_sync.py` 收集与运行。
 
 ## [3.14.1] - 2026-04-26
 - [测试] 修正大盘复盘 prompt 测试对“明日交易计划”标题的断言，并同步桌面端版本号，恢复发布 gate。

docs/full-guide.md

@@ -1189,6 +1189,7 @@ A: 检查是否启用了 Actions，以及 cron 表达式是否正确（注意是
 - 导入流程会先把 CSV 解析成标准化记录，再逐条提交到持仓账本；遇到忙碌行会计入 `failed_count`，不会因为单行冲突让整批请求整体失败。
 - 交易去重优先使用账户内唯一的 `trade_uid`，缺失时回退到基于日期、代码、方向、数量、价格、费用、税费、币种的确定性哈希。
 - 卖出会先校验可用数量，超卖返回 `409 portfolio_oversell`；并发写入冲突时可能返回 `409 portfolio_busy`。
+- 持仓快照的 `positions[]` 会返回 `price_source`、`price_date`、`price_stale`、`price_available` 等价格元信息；当天快照优先使用历史收盘价，仅在收盘价缺失时尝试实时价 fallback，历史 `as_of` 快照不会拉取实时价，也不会再把成本价静默当作现价；缺价持仓会标记 `price_available=false` 并从市值与未实现盈亏汇总中排除。
 - 汇率刷新会先尝试在线源；若在线获取失败，则回退到最近一次缓存并标记 `is_stale=true`，避免快照和风险页整体不可用。
 - 当 `PORTFOLIO_FX_UPDATE_ENABLED=false` 时，手动刷新接口会明确返回“在线刷新已禁用”，页面不会误导为“当前没有可刷新的汇率对”。
 - 风险摘要包含集中度、回撤、止损接近度等信息；`sector_concentration` 会优先尝试按板块归类，失败时降级到 `UNCLASSIFIED`，不会阻断风险结果返回。

docs/full-guide_EN.md

@@ -984,6 +984,7 @@ A: Check if Actions is enabled, and if cron expression is correct (note it's UTC
 - The button calls the existing `POST /api/v1/portfolio/fx/refresh` endpoint and reloads snapshot/risk data only.
 - If upstream FX fetch fails, the page may still remain stale after refresh and will explain the fallback result inline.
 - When `PORTFOLIO_FX_UPDATE_ENABLED=false`, the refresh API returns an explicit disabled status and the page shows that online FX refresh is disabled instead of implying that no refreshable pairs exist.
+- Portfolio snapshot `positions[]` now includes price metadata such as `price_source`, `price_date`, `price_stale`, and `price_available`. Today's snapshot uses the historical close first and only falls back to realtime quotes when no close exists, while historical `as_of` snapshots stay on historical-close semantics and no longer silently treat cost basis as the current price. Missing-price positions are marked with `price_available=false` and excluded from market value / unrealized PnL totals.
 
 ## Agent Tool Data Cache And Persistence
 

src/core/pipeline.py

@@ -822,7 +822,14 @@ def _analyze_with_agent(
             agent_result = executor.run(message, context=initial_context)
 
             # 转换为 AnalysisResult
-            result = self._agent_result_to_analysis_result(agent_result, code, stock_name, report_type, query_id)
+            result = self._agent_result_to_analysis_result(
+                agent_result,
+                code,
+                stock_name,
+                report_type,
+                query_id,
+                trend_result=trend_result,
+            )
             if result:
                 result.query_id = query_id
             # Agent weak integrity: placeholder fill only, no LLM retry
@@ -892,7 +899,13 @@ def _analyze_with_agent(
             return None
 
     def _agent_result_to_analysis_result(
-        self, agent_result, code: str, stock_name: str, report_type: ReportType, query_id: str
+        self,
+        agent_result,
+        code: str,
+        stock_name: str,
+        report_type: ReportType,
+        query_id: str,
+        trend_result: Optional[TrendAnalysisResult] = None,
     ) -> AnalysisResult:
         """
         将 AgentResult 转换为 AnalysisResult。
@@ -952,13 +965,57 @@ def _agent_result_to_analysis_result(
             # structure, so we unwrap it here.
             result.dashboard = dash.get("dashboard") or dash
         else:
-            result.sentiment_score = 50
-            result.operation_advice = "Watch" if report_language == "en" else "观望"
+            self._apply_trend_fallback(result, trend_result, report_language)
             if not result.error_message:
                 result.error_message = "Agent failed to generate a valid decision dashboard" if report_language == "en" else "Agent 未能生成有效的决策仪表盘"
 
         return result
 
+    @staticmethod
+    def _apply_trend_fallback(
+        result: AnalysisResult,
+        trend_result: Optional[TrendAnalysisResult],
+        report_language: str,
+    ) -> None:
+        if trend_result is None:
+            result.sentiment_score = 50
+            result.operation_advice = "Watch" if report_language == "en" else "观望"
+            return
+
+        score = getattr(trend_result, "signal_score", None)
+        try:
+            numeric_score = int(score)
+        except (TypeError, ValueError):
+            numeric_score = 50
+        result.sentiment_score = numeric_score if numeric_score > 0 else 50
+
+        trend_status = getattr(trend_result, "trend_status", None)
+        trend_label = getattr(trend_status, "value", None) or str(trend_status or "").strip()
+        if trend_label:
+            result.trend_prediction = trend_label
+
+        buy_signal = getattr(trend_result, "buy_signal", None)
+        signal_label = getattr(buy_signal, "value", None) or str(buy_signal or "").strip()
+        if signal_label:
+            result.operation_advice = signal_label
+        else:
+            result.operation_advice = "Watch" if report_language == "en" else "观望"
+
+        from src.agent.protocols import normalize_decision_signal
+
+        signal_name = getattr(buy_signal, "name", "").lower()
+        signal_to_decision = {
+            "strong_buy": "buy",
+            "buy": "buy",
+            "hold": "hold",
+            "wait": "hold",
+            "sell": "sell",
+            "strong_sell": "sell",
+        }
+        result.decision_type = signal_to_decision.get(signal_name, result.decision_type or "hold")
+        result.decision_type = normalize_decision_signal(result.decision_type)
+        result.data_sources = f"{result.data_sources},trend:fallback" if result.data_sources else "trend:fallback"
+
     @staticmethod
     def _is_placeholder_stock_name(name: str, code: str) -> bool:
         """Return True when the stock name is missing or placeholder-like."""

src/repositories/portfolio_repo.py

@@ -687,6 +687,10 @@ def query_corporate_actions(
     # Price / FX
     # ------------------------------------------------------------------
     def get_latest_close(self, symbol: str, as_of: date) -> Optional[float]:
+        close = self.get_latest_close_with_date(symbol=symbol, as_of=as_of)
+        return close[0] if close is not None else None
+
+    def get_latest_close_with_date(self, symbol: str, as_of: date) -> Optional[Tuple[float, date]]:
         with self.db.get_session() as session:
             row = session.execute(
                 select(StockDaily)
@@ -701,7 +705,7 @@ def get_latest_close(self, symbol: str, as_of: date) -> Optional[float]:
             ).scalar_one_or_none()
             if row is None or row.close is None:
                 return None
-            return float(row.close)
+            return float(row.close), row.date
 
     def save_fx_rate(
         self,

src/services/portfolio_service.py

@@ -70,6 +70,16 @@ class _AvgState:
     total_cost: float = 0.0
 
 
+@dataclass(frozen=True)
+class _ResolvedPositionPrice:
+    price: float
+    source: str
+    price_date: Optional[date]
+    is_stale: bool
+    is_available: bool
+    provider: Optional[str] = None
+
+
 class PortfolioService:
     """Business logic for account CRUD, event writes, and snapshot replay."""
 
@@ -987,25 +997,29 @@ def _build_positions(
                     }
                 )
 
-            last_price = self.repo.get_latest_close(symbol=symbol, as_of=as_of_date)
-            if last_price is None or last_price <= 0:
-                last_price = avg_cost
+            price_info = self._resolve_position_price(symbol=symbol, as_of_date=as_of_date)
+            last_price = price_info.price
 
-            local_market_value = qty * float(last_price)
-            market_base, stale_market, _ = self._convert_amount(
-                amount=local_market_value,
-                from_currency=currency,
-                to_currency=account.base_currency,
-                as_of_date=as_of_date,
-            )
-            cost_base, stale_cost, _ = self._convert_amount(
-                amount=total_cost,
-                from_currency=currency,
-                to_currency=account.base_currency,
-                as_of_date=as_of_date,
-            )
-            unrealized_base = market_base - cost_base
-            fx_stale = fx_stale or stale_market or stale_cost
+            if price_info.is_available:
+                local_market_value = qty * float(last_price)
+                market_base, stale_market, _ = self._convert_amount(
+                    amount=local_market_value,
+                    from_currency=currency,
+                    to_currency=account.base_currency,
+                    as_of_date=as_of_date,
+                )
+                cost_base, stale_cost, _ = self._convert_amount(
+                    amount=total_cost,
+                    from_currency=currency,
+                    to_currency=account.base_currency,
+                    as_of_date=as_of_date,
+                )
+                unrealized_base = market_base - cost_base
+                fx_stale = fx_stale or stale_market or stale_cost
+            else:
+                market_base = 0.0
+                cost_base = 0.0
+                unrealized_base = 0.0
 
             position_rows.append(
                 {
@@ -1019,6 +1033,11 @@ def _build_positions(
                     "market_value_base": round(market_base, 8),
                     "unrealized_pnl_base": round(unrealized_base, 8),
                     "valuation_currency": account.base_currency,
+                    "price_source": price_info.source,
+                    "price_provider": price_info.provider,
+                    "price_date": price_info.price_date.isoformat() if price_info.price_date else None,
+                    "price_stale": price_info.is_stale,
+                    "price_available": price_info.is_available,
                 }
             )
 
@@ -1027,6 +1046,67 @@ def _build_positions(
 
         return position_rows, lot_rows, market_value_base, total_cost_base, fx_stale
 
+    def _resolve_position_price(self, *, symbol: str, as_of_date: date) -> _ResolvedPositionPrice:
+        today = date.today()
+
+        close = self.repo.get_latest_close_with_date(symbol=symbol, as_of=as_of_date)
+        if close is not None:
+            close_price, close_date = close
+            if close_price > 0:
+                return _ResolvedPositionPrice(
+                    price=float(close_price),
+                    source="history_close",
+                    price_date=close_date,
+                    is_stale=close_date < as_of_date,
+                    is_available=True,
+                )
+
+        if as_of_date == today:
+            realtime_price, provider = self._fetch_realtime_position_price(symbol)
+            if realtime_price is not None and realtime_price > 0:
+                return _ResolvedPositionPrice(
+                    price=float(realtime_price),
+                    source="realtime_quote",
+                    price_date=today,
+                    is_stale=False,
+                    is_available=True,
+                    provider=provider,
+                )
+
+        return _ResolvedPositionPrice(
+            price=0.0,
+            source="missing",
+            price_date=None,
+            is_stale=True,
+            is_available=False,
+        )
+
+    @staticmethod
+    def _fetch_realtime_position_price(symbol: str) -> Tuple[Optional[float], Optional[str]]:
+        try:
+            from data_provider.base import DataFetcherManager
+
+            quote = DataFetcherManager().get_realtime_quote(symbol, log_final_failure=False)
+        except Exception as exc:
+            logger.warning("Failed to fetch realtime portfolio price for %s: %s", symbol, exc)
+            return None, None
+
+        if quote is None:
+            return None, None
+
+        price = getattr(quote, "price", None)
+        try:
+            numeric_price = float(price)
+        except (TypeError, ValueError):
+            return None, None
+
+        if numeric_price <= 0:
+            return None, None
+
+        source = getattr(quote, "source", None)
+        provider = getattr(source, "value", None) or (str(source) if source is not None else None)
+        return numeric_price, provider
+
     @staticmethod
     def _consume_fifo_lots(
         lots: List[Dict[str, Any]],