fix: add A-share financial/dividend context outputs (#710) (#731)

Co-authored-by: mumu <42829555+ZhuLinsen@users.noreply.github.com>

README.md

@@ -33,7 +33,7 @@
 | AI | 决策仪表盘 | 一句话核心结论 + 精确买卖点位 + 操作检查清单 |
 | 分析 | 多维度分析 | 技术面（盘中实时 MA/多头排列）+ 筹码分布 + 舆情情报 + 实时行情 |
 | 市场 | 全球市场 | 支持 A股、港股、美股及美股指数（SPX、DJI、IXIC 等） |
-| 基本面 | 结构化聚合 | 新增 `fundamental_context`（valuation/growth/earnings/institution/capital_flow/dragon_tiger/boards，其中 `boards` 表示板块涨跌榜），主链路 fail-open 降级 |
+| 基本面 | 结构化聚合 | 新增 `fundamental_context`（valuation/growth/earnings/institution/capital_flow/dragon_tiger/boards，其中 `earnings.data` 新增 `financial_report` 与 `dividend`，`boards` 表示板块涨跌榜），主链路 fail-open 降级 |
 | 策略 | 市场策略系统 | 内置 A股「三段式复盘策略」与美股「Regime Strategy」，输出进攻/均衡/防守或 risk-on/neutral/risk-off 计划，并附“仅供参考，不构成投资建议”提示 |
 | 复盘 | 大盘复盘 | 每日市场概览、板块涨跌；支持 cn(A股)/us(美股)/both(两者) 切换 |
 | 智能导入 | 多源导入 | 支持图片、CSV/Excel 文件、剪贴板粘贴；Vision LLM 提取代码+名称；置信度分层确认；名称→代码解析（本地+拼音+AkShare） |
@@ -188,6 +188,8 @@
 > - 若业务需要硬 SLA，可在后续阶段升级为“子进程隔离 + kill”的硬超时方案。
 > - 字段契约：
 >   - `fundamental_context.boards.data` = `sector_rankings`（板块涨跌榜，结构 `{top, bottom}`）；
+>   - `fundamental_context.earnings.data.financial_report` = 财报摘要（报告期、营收、归母净利润、经营现金流、ROE）；
+>   - `fundamental_context.earnings.data.dividend` = 分红指标（仅现金分红税前口径，含 `events`、`ttm_cash_dividend_per_share`、`ttm_dividend_yield_pct`）；
 >   - `get_stock_info.belong_boards` = 个股所属板块列表；
 >   - `get_stock_info.boards` 为兼容别名，值与 `belong_boards` 相同（未来仅在大版本考虑移除）；
 >   - `get_stock_info.sector_rankings` 与 `fundamental_context.boards.data` 保持一致。

api/v1/endpoints/analysis.py

@@ -53,7 +53,11 @@ from src.services.task_queue import (
     DuplicateTaskError,
     TaskStatus as TaskStatusEnum,
 )
-from src.utils.data_processing import normalize_model_used, parse_json_field
+from src.utils.data_processing import (
+    normalize_model_used,
+    parse_json_field,
+    extract_fundamental_detail_fields,
+)
 
 logger = logging.getLogger(__name__)
 
@@ -261,8 +265,17 @@ def _handle_sync_analysis(
 
         # 构建报告结构
         report_data = result.get("report", {})
+        context_snapshot, fundamental_snapshot = _load_sync_fundamental_sources(
+            query_id=query_id,
+            stock_code=result.get("stock_code", stock_code),
+        )
         report = _build_analysis_report(
-            report_data, query_id, stock_code, result.get("stock_name")
+            report_data,
+            query_id,
+            stock_code,
+            result.get("stock_name"),
+            context_snapshot=context_snapshot,
+            fallback_fundamental_payload=fundamental_snapshot,
         )
 
         return AnalysisResultResponse(
@@ -555,11 +568,44 @@ def get_analysis_status(task_id: str) -> TaskStatus:
 # 辅助函数
 # ============================================================
 
+def _load_sync_fundamental_sources(
+    query_id: str,
+    stock_code: str,
+) -> tuple[Optional[Any], Optional[Dict[str, Any]]]:
+    """
+    Load context_snapshot and fallback fundamental snapshot for sync analyze response.
+    """
+    try:
+        from src.storage import DatabaseManager
+
+        db = DatabaseManager.get_instance()
+        records = db.get_analysis_history(query_id=query_id, code=stock_code, limit=1)
+        context_snapshot = None
+        if records:
+            context_snapshot = parse_json_field(getattr(records[0], "context_snapshot", None))
+
+        fallback_fundamental = db.get_latest_fundamental_snapshot(
+            query_id=query_id,
+            code=stock_code,
+        )
+        return context_snapshot, fallback_fundamental
+    except Exception as e:
+        logger.debug(
+            "load sync fundamental sources failed (fail-open): query_id=%s stock_code=%s err=%s",
+            query_id,
+            stock_code,
+            e,
+        )
+        return None, None
+
+
 def _build_analysis_report(
         report_data: Dict[str, Any],
         query_id: str,
         stock_code: str,
-        stock_name: Optional[str] = None
+        stock_name: Optional[str] = None,
+        context_snapshot: Optional[Any] = None,
+        fallback_fundamental_payload: Optional[Dict[str, Any]] = None,
 ) -> AnalysisReport:
     """
     构建符合 API 规范的分析报告
@@ -569,6 +615,8 @@ def _build_analysis_report(
         query_id: 查询 ID
         stock_code: 股票代码
         stock_name: 股票名称
+        context_snapshot: 上下文快照（可选）
+        fallback_fundamental_payload: 基本面快照 payload（可选）
         
     Returns:
         AnalysisReport: 结构化的分析报告
@@ -606,12 +654,18 @@ def _build_analysis_report(
             take_profit=strategy_data.get("take_profit")
         )
 
+    extracted_fundamental = extract_fundamental_detail_fields(
+        context_snapshot=context_snapshot,
+        fallback_fundamental_payload=fallback_fundamental_payload,
+    )
     details = None
-    if details_data:
+    if details_data or any(extracted_fundamental.values()) or context_snapshot is not None:
         details = ReportDetails(
             news_content=details_data.get("news_summary") or details_data.get("news_content"),
             raw_result=details_data,
-            context_snapshot=None
+            context_snapshot=context_snapshot,
+            financial_report=extracted_fundamental.get("financial_report"),
+            dividend_metrics=extracted_fundamental.get("dividend_metrics"),
         )
 
     return AnalysisReport(

api/v1/endpoints/history.py

@@ -32,7 +32,7 @@ from api.v1.schemas.history import (
 from api.v1.schemas.common import ErrorResponse
 from src.storage import DatabaseManager
 from src.services.history_service import HistoryService, MarkdownReportGenerationError
-from src.utils.data_processing import normalize_model_used
+from src.utils.data_processing import normalize_model_used, extract_fundamental_detail_fields
 
 logger = logging.getLogger(__name__)
 
@@ -254,10 +254,21 @@ def get_history_detail(
             take_profit=result.get("take_profit")
         )
         
+        fallback_fundamental = db_manager.get_latest_fundamental_snapshot(
+            query_id=result.get("query_id", ""),
+            code=result.get("stock_code", ""),
+        )
+        extracted_fundamental = extract_fundamental_detail_fields(
+            context_snapshot=result.get("context_snapshot"),
+            fallback_fundamental_payload=fallback_fundamental,
+        )
+
         details = ReportDetails(
             news_content=result.get("news_content"),
             raw_result=result.get("raw_result"),
-            context_snapshot=result.get("context_snapshot")
+            context_snapshot=result.get("context_snapshot"),
+            financial_report=extracted_fundamental.get("financial_report"),
+            dividend_metrics=extracted_fundamental.get("dividend_metrics"),
         )
         
         return AnalysisReport(

api/v1/schemas/history.py

@@ -155,6 +155,8 @@ class ReportDetails(BaseModel):
     news_content: Optional[str] = Field(None, description="新闻摘要")
     raw_result: Optional[Any] = Field(None, description="原始分析结果（JSON）")
     context_snapshot: Optional[Any] = Field(None, description="分析时上下文快照（JSON）")
+    financial_report: Optional[Any] = Field(None, description="结构化财报摘要（来自 fundamental_context）")
+    dividend_metrics: Optional[Any] = Field(None, description="结构化分红指标（含 TTM 口径）")
 
 
 class AnalysisReport(BaseModel):

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

@@ -53,6 +53,8 @@ export interface ReportDetails {
   newsContent?: string;
   rawResult?: Record<string, unknown>;
   contextSnapshot?: Record<string, unknown>;
+  financialReport?: Record<string, unknown>;
+  dividendMetrics?: Record<string, unknown>;
 }
 
 /** 完整分析报告 */

data_provider/base.py

@@ -1793,8 +1793,59 @@ class DataFetcherManager:
         growth_payload = bundle_payload.get("growth", {}) if isinstance(bundle_payload, dict) else {}
         earnings_payload = bundle_payload.get("earnings", {}) if isinstance(bundle_payload, dict) else {}
         institution_payload = bundle_payload.get("institution", {}) if isinstance(bundle_payload, dict) else {}
+        if not isinstance(growth_payload, dict):
+            growth_payload = {}
+        else:
+            growth_payload = dict(growth_payload)
+        if not isinstance(earnings_payload, dict):
+            earnings_payload = {}
+        else:
+            earnings_payload = dict(earnings_payload)
+        if not isinstance(institution_payload, dict):
+            institution_payload = {}
+        else:
+            institution_payload = dict(institution_payload)
+
+        # Derive TTM dividend yield from already-fetched quote price; avoid extra quote calls.
+        earnings_extra_errors: List[str] = []
+        dividend_payload = earnings_payload.get("dividend")
+        if isinstance(dividend_payload, dict):
+            dividend_payload = dict(dividend_payload)
+            ttm_cash_raw = dividend_payload.get("ttm_cash_dividend_per_share")
+            ttm_cash = None
+            if ttm_cash_raw is not None:
+                try:
+                    ttm_cash = float(ttm_cash_raw)
+                except (TypeError, ValueError):
+                    earnings_extra_errors.append("invalid_ttm_cash_dividend_per_share")
+            if isinstance(quote_payload, dict):
+                latest_price_raw = quote_payload.get("price")
+            else:
+                latest_price_raw = getattr(quote_payload, "price", None) if quote_payload else None
+            latest_price = None
+            if latest_price_raw is not None:
+                try:
+                    latest_price = float(latest_price_raw)
+                except (TypeError, ValueError):
+                    latest_price = None
+            ttm_yield = None
+            if ttm_cash is not None:
+                if latest_price is not None and latest_price > 0:
+                    ttm_yield = round(ttm_cash / latest_price * 100.0, 4)
+                else:
+                    earnings_extra_errors.append("invalid_price_for_ttm_dividend_yield")
+
+            dividend_payload["ttm_dividend_yield_pct"] = ttm_yield
+            if ttm_yield is not None:
+                dividend_payload["yield_formula"] = "ttm_cash_dividend_per_share / latest_price * 100"
+            earnings_payload["dividend"] = dividend_payload
+
         adapter_errors = list(bundle_payload.get("errors", [])) if isinstance(bundle_payload, dict) else []
         adapter_errors.extend(bundle_errors)
+        growth_errors = list(adapter_errors)
+        earnings_errors = list(adapter_errors)
+        earnings_errors.extend(earnings_extra_errors)
+        institution_errors = list(adapter_errors)
 
         growth_status = self._infer_block_status(growth_payload, bundle_status)
         earnings_status = self._infer_block_status(earnings_payload, bundle_status)
@@ -1804,19 +1855,19 @@ class DataFetcherManager:
             growth_status,
             growth_payload,
             bundle_chain,
-            adapter_errors,
+            growth_errors,
         )
         result_ctx["earnings"] = self._build_fundamental_block(
             earnings_status,
             earnings_payload,
             bundle_chain,
-            adapter_errors,
+            earnings_errors,
         )
         result_ctx["institution"] = self._build_fundamental_block(
             institution_status,
             institution_payload,
             bundle_chain,
-            adapter_errors,
+            institution_errors,
         )
 
         # capital flow

data_provider/fundamental_adapter.py

@@ -17,6 +17,32 @@ import pandas as pd
 
 logger = logging.getLogger(__name__)
 
+_DIVIDEND_KEYWORD_MAP: Dict[str, List[str]] = {
+    "per_share": [
+        "每股派息",
+        "每股现金红利",
+        "每股分红",
+        "每股派现",
+        "派现(元/股)",
+        "派息(元/股)",
+        "税前派息(元/股)",
+        "现金分红(税前)",
+    ],
+    "plan_text": [
+        "分配方案",
+        "分红方案",
+        "实施方案",
+        "派息方案",
+        "方案",
+        "预案",
+        "方案说明",
+    ],
+    "ex_dividend_date": ["除权除息日", "除息日", "除权日", "除权除息", "除息日期"],
+    "record_date": ["股权登记日", "登记日"],
+    "announce_date": ["公告日期", "公告日", "实施公告日", "预案公告日"],
+    "report_date": ["报告期", "报告日期", "截止日期", "统计截止日期"],
+}
+
 
 def _safe_float(value: Any) -> Optional[float]:
     """Best-effort float conversion."""
@@ -42,6 +68,21 @@ def _safe_str(value: Any) -> str:
     return str(value).strip()
 
 
+def _safe_datetime(value: Any) -> Optional[datetime]:
+    if value is None:
+        return None
+    try:
+        parsed = pd.to_datetime(value)
+    except Exception:
+        return None
+    if pd.isna(parsed):
+        return None
+    try:
+        return parsed.to_pydatetime()
+    except Exception:
+        return None
+
+
 def _normalize_code(raw: Any) -> str:
     s = _safe_str(raw).upper()
     if "." in s:
@@ -63,6 +104,139 @@ def _pick_by_keywords(row: pd.Series, keywords: List[str]) -> Optional[Any]:
     return None
 
 
+def _parse_dividend_plan_to_per_share(plan_text: str) -> Optional[float]:
+    """Parse per-share cash dividend from Chinese plan text."""
+    text = _safe_str(plan_text)
+    if not text:
+        return None
+
+    for pattern in (
+        r"(?:每)?\s*10\s*股?\s*派(?:发)?\s*([0-9]+(?:\.[0-9]+)?)\s*元",
+        r"10\s*派\s*([0-9]+(?:\.[0-9]+)?)\s*元",
+    ):
+        match = re.search(pattern, text)
+        if match:
+            parsed = _safe_float(match.group(1))
+            if parsed is not None and parsed > 0:
+                return parsed / 10.0
+
+    match_per_share = re.search(r"每\s*股\s*派(?:发)?\s*([0-9]+(?:\.[0-9]+)?)\s*元", text)
+    if match_per_share:
+        parsed = _safe_float(match_per_share.group(1))
+        if parsed is not None and parsed > 0:
+            return parsed
+    return None
+
+
+def _extract_cash_dividend_per_share(row: pd.Series) -> Optional[float]:
+    """Extract pre-tax cash dividend per share from a row."""
+    plan_text = _safe_str(_pick_by_keywords(row, _DIVIDEND_KEYWORD_MAP["plan_text"]))
+    # Keep pre-tax semantics; skip explicit after-tax plans unless pre-tax marker exists.
+    if "税后" in plan_text and "税前" not in plan_text and "含税" not in plan_text:
+        return None
+
+    direct = _safe_float(_pick_by_keywords(row, _DIVIDEND_KEYWORD_MAP["per_share"]))
+    if direct is not None and direct > 0:
+        return direct
+    return _parse_dividend_plan_to_per_share(plan_text)
+
+
+def _filter_rows_by_code(df: pd.DataFrame, stock_code: str) -> pd.DataFrame:
+    if df is None or df.empty:
+        return pd.DataFrame()
+    code_cols = [c for c in df.columns if any(k in str(c) for k in ("代码", "股票代码", "证券代码", "symbol", "ts_code"))]
+    if not code_cols:
+        return df
+
+    target = _normalize_code(stock_code)
+    for col in code_cols:
+        try:
+            series = df[col].astype(str).map(_normalize_code)
+            filtered = df[series == target]
+            if not filtered.empty:
+                return filtered
+        except Exception:
+            continue
+    return pd.DataFrame()
+
+
+def _normalize_report_date(value: Any) -> Optional[str]:
+    parsed = _safe_datetime(value)
+    return parsed.date().isoformat() if parsed else None
+
+
+def _build_dividend_payload(
+    dividend_df: pd.DataFrame,
+    stock_code: str,
+    max_events: int = 5,
+) -> Dict[str, Any]:
+    work_df = _filter_rows_by_code(dividend_df, stock_code)
+    if work_df.empty:
+        return {}
+
+    now_date = datetime.now().date()
+    ttm_start_date = now_date - timedelta(days=365)
+    dedupe_keys = set()
+    events: List[Dict[str, Any]] = []
+
+    for _, row in work_df.iterrows():
+        if not isinstance(row, pd.Series):
+            continue
+        ex_dt = _safe_datetime(_pick_by_keywords(row, _DIVIDEND_KEYWORD_MAP["ex_dividend_date"]))
+        record_dt = _safe_datetime(_pick_by_keywords(row, _DIVIDEND_KEYWORD_MAP["record_date"]))
+        announce_dt = _safe_datetime(_pick_by_keywords(row, _DIVIDEND_KEYWORD_MAP["announce_date"]))
+        event_dt = ex_dt or record_dt or announce_dt
+        if event_dt is None:
+            continue
+        event_date = event_dt.date()
+        if event_date > now_date:
+            continue
+
+        per_share = _extract_cash_dividend_per_share(row)
+        if per_share is None or per_share <= 0:
+            continue
+
+        dedupe_key = (event_date.isoformat(), round(per_share, 6))
+        if dedupe_key in dedupe_keys:
+            continue
+        dedupe_keys.add(dedupe_key)
+
+        events.append(
+            {
+                "event_date": event_date.isoformat(),
+                "ex_dividend_date": ex_dt.date().isoformat() if ex_dt else None,
+                "record_date": record_dt.date().isoformat() if record_dt else None,
+                "announcement_date": announce_dt.date().isoformat() if announce_dt else None,
+                "cash_dividend_per_share": round(per_share, 6),
+                "is_pre_tax": True,
+            }
+        )
+
+    if not events:
+        return {}
+
+    events.sort(key=lambda item: item.get("event_date") or "", reverse=True)
+    ttm_events: List[Dict[str, Any]] = []
+    for item in events:
+        event_dt = _safe_datetime(item.get("event_date"))
+        if event_dt is None:
+            continue
+        event_date = event_dt.date()
+        if ttm_start_date <= event_date <= now_date:
+            ttm_events.append(item)
+
+    return {
+        "events": events[:max(1, max_events)],
+        "ttm_event_count": len(ttm_events),
+        "ttm_cash_dividend_per_share": (
+            round(sum(float(item.get("cash_dividend_per_share") or 0.0) for item in ttm_events), 6)
+            if ttm_events else None
+        ),
+        "coverage": "cash_dividend_pre_tax",
+        "as_of": now_date.isoformat(),
+    }
+
+
 def _extract_latest_row(df: pd.DataFrame, stock_code: str) -> Optional[pd.Series]:
     """
     Select the most relevant row for the given stock.
@@ -142,12 +316,27 @@ class AkshareFundamentalAdapter:
                 profit_yoy = _safe_float(_pick_by_keywords(row, ["净利润同比", "净利同比", "归母净利润同比"]))
                 roe = _safe_float(_pick_by_keywords(row, ["净资产收益率", "ROE", "净资产收益"]))
                 gross_margin = _safe_float(_pick_by_keywords(row, ["毛利率"]))
+                report_date = _normalize_report_date(_pick_by_keywords(row, _DIVIDEND_KEYWORD_MAP["report_date"]))
+                revenue = _safe_float(_pick_by_keywords(row, ["营业总收入", "营业收入", "营收"]))
+                net_profit_parent = _safe_float(_pick_by_keywords(row, ["归母净利润", "母公司股东净利润", "净利润"]))
+                operating_cash_flow = _safe_float(
+                    _pick_by_keywords(row, ["经营活动产生的现金流量净额", "经营现金流", "经营活动现金流"])
+                )
                 result["growth"] = {
                     "revenue_yoy": revenue_yoy,
                     "net_profit_yoy": profit_yoy,
                     "roe": roe,
                     "gross_margin": gross_margin,
                 }
+                financial_report_payload = {
+                    "report_date": report_date,
+                    "revenue": revenue,
+                    "net_profit_parent": net_profit_parent,
+                    "operating_cash_flow": operating_cash_flow,
+                    "roe": roe,
+                }
+                if any(v is not None for v in financial_report_payload.values()):
+                    result["earnings"]["financial_report"] = financial_report_payload
                 result["source_chain"].append(f"growth:{fin_source}")
 
         # Earnings forecast
@@ -180,6 +369,19 @@ class AkshareFundamentalAdapter:
                 )[:200]
                 result["source_chain"].append(f"earnings_quick:{quick_source}")
 
+        # Dividend details (cash dividend, pre-tax)
+        dividend_df, dividend_source, dividend_errors = self._call_df_candidates([
+            ("stock_fhps_detail_em", {"symbol": stock_code}),
+            ("stock_history_dividend_detail", {"symbol": stock_code, "indicator": "分红", "date": ""}),
+            ("stock_dividend_cninfo", {"symbol": stock_code}),
+        ])
+        result["errors"].extend(dividend_errors)
+        if dividend_df is not None:
+            dividend_payload = _build_dividend_payload(dividend_df, stock_code, max_events=5)
+            if dividend_payload:
+                result["earnings"]["dividend"] = dividend_payload
+                result["source_chain"].append(f"dividend:{dividend_source}")
+
         # Institution / top shareholders
         inst_df, inst_source, inst_errors = self._call_df_candidates([
             ("stock_institute_hold", {}),

docs/CHANGELOG.md

@@ -12,6 +12,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
 ### 新功能
 
 - 📱 **Social Sentiment Intelligence (US stocks)** — 新增 Reddit / X (Twitter) / Polymarket 社交媒体情绪数据源，为美股分析提供实时社交舆情情报。数据来自 api.adanos.org，包含 Buzz Score、情绪评分、提及量等指标。完全可选（需配置 `SOCIAL_SENTIMENT_API_KEY`），仅对美股生效，A 股 / 港股不受影响。
+- 📊 **A股财报与分红结构化增强（Issue #710）** — `fundamental_context.earnings.data` 新增 `financial_report` 与 `dividend` 字段；分红采用“仅现金分红、税前口径”并支持 `10派X元 -> 每股X/10` 换算；新增 `ttm_cash_dividend_per_share` 与 `ttm_dividend_yield_pct`，同时在分析/历史 API 的 `details` 中新增 `financial_report`、`dividend_metrics` 可选字段（fail-open，向后兼容）。
 - 🔍 **接入Tushare筹码、行业板块涨跌接口** — 新增 Tushare筹码分布、行业板块涨跌接口获取；修复筹码分布判断部分错误；将筹码分布与行业板块的数据源优先级改为与实时行情获取优先级一致，均在.env配置；默认在上海时间19点之后才能用Tushare获取当天交易日的筹码分布，盘后获取当天交易日的行业板块涨跌，否则用前一个交易日的；行业板块涨跌受数据源影响，结果会不一致，优先使用Tushare的同花顺接口、其次东财接口。
 
 ### 文档

docs/full-guide.md

@@ -256,6 +256,8 @@ daily_stock_analysis/
 > - 任何异常走 fail-open，仅记录错误，不影响技术面/新闻/筹码主链路。
 > - 字段契约：
 >   - `fundamental_context.boards.data` = `sector_rankings`（板块涨跌榜，结构 `{top, bottom}`）；
+>   - `fundamental_context.earnings.data.financial_report` = 财报摘要（报告期、营收、归母净利润、经营现金流、ROE）；
+>   - `fundamental_context.earnings.data.dividend` = 分红指标（仅现金分红税前口径，含 `events`、`ttm_cash_dividend_per_share`、`ttm_dividend_yield_pct`）；
 >   - `get_stock_info.belong_boards` = 个股所属板块列表；
 >   - `get_stock_info.boards` 为兼容别名，值与 `belong_boards` 相同（未来仅在大版本考虑移除）；
 >   - `get_stock_info.sector_rankings` 与 `fundamental_context.boards.data` 保持一致。

setup.cfg

@@ -6,6 +6,7 @@ exclude =
     .env,
     venv,
     .venv,
+    .venv*,
     build,
     dist,
     *.egg-info

src/analyzer.py

@@ -1117,7 +1117,52 @@ class GeminiAnalyzer:
 | 流通市值 | {self._format_amount(rt.get('circ_mv'))} | |
 | 60日涨跌幅 | {rt.get('change_60d', 'N/A')}% | 中期表现 |
 """
-        
+
+        # 添加财报与分红（价值投资口径）
+        fundamental_context = context.get("fundamental_context") if isinstance(context, dict) else None
+        earnings_block = (
+            fundamental_context.get("earnings", {})
+            if isinstance(fundamental_context, dict)
+            else {}
+        )
+        earnings_data = (
+            earnings_block.get("data", {})
+            if isinstance(earnings_block, dict)
+            else {}
+        )
+        financial_report = (
+            earnings_data.get("financial_report", {})
+            if isinstance(earnings_data, dict)
+            else {}
+        )
+        dividend_metrics = (
+            earnings_data.get("dividend", {})
+            if isinstance(earnings_data, dict)
+            else {}
+        )
+        if isinstance(financial_report, dict) or isinstance(dividend_metrics, dict):
+            financial_report = financial_report if isinstance(financial_report, dict) else {}
+            dividend_metrics = dividend_metrics if isinstance(dividend_metrics, dict) else {}
+            ttm_yield = dividend_metrics.get("ttm_dividend_yield_pct", "N/A")
+            ttm_cash = dividend_metrics.get("ttm_cash_dividend_per_share", "N/A")
+            ttm_count = dividend_metrics.get("ttm_event_count", "N/A")
+            report_date = financial_report.get("report_date", "N/A")
+            prompt += f"""
+### 财报与分红（价值投资口径）
+| 指标 | 数值 | 说明 |
+|------|------|------|
+| 最近报告期 | {report_date} | 来自结构化财报字段 |
+| 营业收入 | {financial_report.get('revenue', 'N/A')} | |
+| 归母净利润 | {financial_report.get('net_profit_parent', 'N/A')} | |
+| 经营现金流 | {financial_report.get('operating_cash_flow', 'N/A')} | |
+| ROE | {financial_report.get('roe', 'N/A')} | |
+| 近12个月每股现金分红 | {ttm_cash} | 仅现金分红、税前口径 |
+| TTM 股息率 | {ttm_yield} | 公式：近12个月每股现金分红 / 当前价格 × 100% |
+| TTM 分红事件数 | {ttm_count} | |
+
+> 若上述字段为 N/A 或缺失，请明确写“数据缺失，无法判断”，禁止编造。
+"""
+
         # 添加筹码分布数据
         if 'chip' in context:
             chip = context['chip']

src/storage.py

@@ -965,6 +965,49 @@ class DatabaseManager:
                 )
                 return 0
 
+    def get_latest_fundamental_snapshot(
+        self,
+        query_id: str,
+        code: str,
+    ) -> Optional[Dict[str, Any]]:
+        """
+        获取指定 query_id + code 的最新基本面快照 payload。
+
+        读取失败或不存在时返回 None（fail-open）。
+        """
+        if not query_id or not code:
+            return None
+
+        with self.get_session() as session:
+            try:
+                row = session.execute(
+                    select(FundamentalSnapshot)
+                    .where(
+                        and_(
+                            FundamentalSnapshot.query_id == query_id,
+                            FundamentalSnapshot.code == code,
+                        )
+                    )
+                    .order_by(desc(FundamentalSnapshot.created_at))
+                    .limit(1)
+                ).scalar_one_or_none()
+            except Exception as e:
+                logger.debug(
+                    "基本面快照读取失败（fail-open）: query_id=%s code=%s err=%s",
+                    query_id,
+                    code,
+                    e,
+                )
+                return None
+
+            if row is None:
+                return None
+            try:
+                payload = json.loads(row.payload or "{}")
+                return payload if isinstance(payload, dict) else None
+            except Exception:
+                return None
+
     def get_recent_news(self, code: str, days: int = 7, limit: int = 20) -> List[NewsIntel]:
         """
         获取指定股票最近 N 天的新闻情报

src/utils/data_processing.py

@@ -4,7 +4,7 @@ Shared data parsing and normalization helpers.
 """
 
 import json
-from typing import Any, Optional
+from typing import Any, Dict, Optional
 
 
 _MODEL_PLACEHOLDER_VALUES = {"unknown", "error", "none", "null", "n/a"}
@@ -32,3 +32,57 @@ def parse_json_field(value: Any) -> Any:
         except (json.JSONDecodeError, TypeError, ValueError):
             return value
     return value
+
+
+def _non_empty_dict(value: Any) -> Optional[Dict[str, Any]]:
+    if not isinstance(value, dict):
+        return None
+    return value if value else None
+
+
+def extract_fundamental_context(
+    context_snapshot: Any,
+    fallback_fundamental_payload: Any = None,
+) -> Optional[Dict[str, Any]]:
+    """
+    Resolve fundamental_context from context snapshot, with optional fallback payload.
+    """
+    snapshot_obj = parse_json_field(context_snapshot)
+    if isinstance(snapshot_obj, dict):
+        enhanced = snapshot_obj.get("enhanced_context")
+        if isinstance(enhanced, dict):
+            fundamental = enhanced.get("fundamental_context")
+            if isinstance(fundamental, dict):
+                return fundamental
+
+    fallback_obj = parse_json_field(fallback_fundamental_payload)
+    if isinstance(fallback_obj, dict):
+        return fallback_obj
+    return None
+
+
+def extract_fundamental_detail_fields(
+    context_snapshot: Any,
+    fallback_fundamental_payload: Any = None,
+) -> Dict[str, Optional[Dict[str, Any]]]:
+    """
+    Extract stable API-facing financial and dividend blocks from fundamental_context.
+    """
+    fundamental_ctx = extract_fundamental_context(
+        context_snapshot=context_snapshot,
+        fallback_fundamental_payload=fallback_fundamental_payload,
+    )
+    if not isinstance(fundamental_ctx, dict):
+        return {"financial_report": None, "dividend_metrics": None}
+
+    earnings_block = fundamental_ctx.get("earnings")
+    earnings_data = earnings_block.get("data") if isinstance(earnings_block, dict) else None
+    if not isinstance(earnings_data, dict):
+        return {"financial_report": None, "dividend_metrics": None}
+
+    financial_report = _non_empty_dict(earnings_data.get("financial_report"))
+    dividend_metrics = _non_empty_dict(earnings_data.get("dividend"))
+    return {
+        "financial_report": financial_report,
+        "dividend_metrics": dividend_metrics,
+    }

tests/test_analysis_api_contract.py

@@ -16,10 +16,16 @@ ensure_litellm_stub()
 
 try:
     from api.app import create_app
-    from api.v1.endpoints.analysis import trigger_analysis
+    from api.v1.endpoints.analysis import (
+        trigger_analysis,
+        _build_analysis_report,
+        _load_sync_fundamental_sources,
+    )
 except Exception:  # pragma: no cover - optional dependency environments
     create_app = None
     trigger_analysis = None
+    _build_analysis_report = None
+    _load_sync_fundamental_sources = None
 
 from src.enums import ReportType
 from src.services.analysis_service import AnalysisService
@@ -70,6 +76,72 @@ class AnalysisApiContractTestCase(unittest.TestCase):
 
         self.assertEqual(result["report"]["meta"]["report_type"], "full")
 
+    def test_build_analysis_report_extracts_fundamental_fields_from_snapshot(self) -> None:
+        if _build_analysis_report is None:
+            self.skipTest("analysis endpoint helpers unavailable in this environment")
+
+        report = _build_analysis_report(
+            report_data={
+                "meta": {},
+                "summary": {},
+                "strategy": {},
+                "details": {"news_summary": "news"},
+            },
+            query_id="q1",
+            stock_code="600519",
+            stock_name="贵州茅台",
+            context_snapshot={
+                "enhanced_context": {
+                    "fundamental_context": {
+                        "earnings": {
+                            "data": {
+                                "financial_report": {"report_date": "2025-12-31", "revenue": 1000},
+                                "dividend": {"ttm_dividend_yield_pct": 2.5},
+                            }
+                        }
+                    }
+                }
+            },
+            fallback_fundamental_payload=None,
+        )
+
+        self.assertEqual(report.details.financial_report["report_date"], "2025-12-31")
+        self.assertEqual(report.details.dividend_metrics["ttm_dividend_yield_pct"], 2.5)
+
+    def test_load_sync_fundamental_sources_uses_query_and_code_for_fallback(self) -> None:
+        if _load_sync_fundamental_sources is None:
+            self.skipTest("analysis endpoint helpers unavailable in this environment")
+
+        mock_db = MagicMock()
+        mock_db.get_analysis_history.return_value = [SimpleNamespace(context_snapshot=None)]
+        fallback_payload = {
+            "earnings": {
+                "data": {
+                    "financial_report": {"report_date": "2025-12-31"},
+                    "dividend": {"ttm_dividend_yield_pct": 2.1},
+                }
+            }
+        }
+        mock_db.get_latest_fundamental_snapshot.return_value = fallback_payload
+
+        with patch("src.storage.DatabaseManager.get_instance", return_value=mock_db):
+            context_snapshot, fundamental_snapshot = _load_sync_fundamental_sources(
+                query_id="q_sync_001",
+                stock_code="600519",
+            )
+
+        self.assertIsNone(context_snapshot)
+        self.assertEqual(fundamental_snapshot, fallback_payload)
+        mock_db.get_analysis_history.assert_called_once_with(
+            query_id="q_sync_001",
+            code="600519",
+            limit=1,
+        )
+        mock_db.get_latest_fundamental_snapshot.assert_called_once_with(
+            query_id="q_sync_001",
+            code="600519",
+        )
+
     def test_openapi_declares_single_and_batch_async_202_payloads(self) -> None:
         if create_app is None:
             self.skipTest("fastapi is not installed in this test environment")

tests/test_analysis_history.py

@@ -26,9 +26,11 @@ except ModuleNotFoundError:
 try:
     from fastapi.testclient import TestClient
     from api.app import create_app
+    from api.v1.endpoints.history import get_history_detail
 except ModuleNotFoundError:
     TestClient = None
     create_app = None
+    get_history_detail = None
 
 from src.config import Config
 from src.storage import DatabaseManager, AnalysisHistory, BacktestResult
@@ -292,6 +294,72 @@ class AnalysisHistoryTestCase(unittest.TestCase):
         self.assertEqual(detail.get("stop_loss"), "110.0")
         self.assertEqual(detail.get("take_profit"), "150.0")
 
+    def test_history_detail_uses_fundamental_snapshot_fallback_when_context_missing(self) -> None:
+        """When context_snapshot is disabled, detail API should fallback to fundamental_snapshot."""
+        if get_history_detail is None:
+            self.skipTest("fastapi is not installed in this test environment")
+
+        result = self._build_result()
+        query_id = "query_fundamental_fallback_001"
+        saved = self.db.save_analysis_history(
+            result=result,
+            query_id=query_id,
+            report_type="simple",
+            news_content="新闻摘要",
+            context_snapshot=None,
+            save_snapshot=False,
+        )
+        self.assertEqual(saved, 1)
+
+        self.db.save_fundamental_snapshot(
+            query_id=query_id,
+            code="600519",
+            payload={
+                "earnings": {
+                    "data": {
+                        "financial_report": {"report_date": "2025-12-31", "revenue": 1000},
+                        "dividend": {"ttm_dividend_yield_pct": 2.6, "ttm_cash_dividend_per_share": 1.3},
+                    }
+                }
+            },
+        )
+
+        with self.db.get_session() as session:
+            row = session.query(AnalysisHistory).filter(AnalysisHistory.query_id == query_id).first()
+            if row is None:
+                self.fail("未找到保存的历史记录")
+            record_id = row.id
+
+        report = get_history_detail(str(record_id), db_manager=self.db)
+        self.assertEqual(report.details.financial_report["report_date"], "2025-12-31")
+        self.assertEqual(report.details.dividend_metrics["ttm_dividend_yield_pct"], 2.6)
+
+    def test_history_detail_returns_null_fundamental_fields_when_snapshot_absent(self) -> None:
+        """Detail API should keep new fields nullable when no context/fundamental snapshot exists."""
+        if get_history_detail is None:
+            self.skipTest("fastapi is not installed in this test environment")
+
+        query_id = "query_fundamental_fallback_002"
+        saved = self.db.save_analysis_history(
+            result=self._build_result(),
+            query_id=query_id,
+            report_type="simple",
+            news_content="新闻摘要",
+            context_snapshot=None,
+            save_snapshot=False,
+        )
+        self.assertEqual(saved, 1)
+
+        with self.db.get_session() as session:
+            row = session.query(AnalysisHistory).filter(AnalysisHistory.query_id == query_id).first()
+            if row is None:
+                self.fail("未找到保存的历史记录")
+            record_id = row.id
+
+        report = get_history_detail(str(record_id), db_manager=self.db)
+        self.assertIsNone(report.details.financial_report)
+        self.assertIsNone(report.details.dividend_metrics)
+
     def test_delete_analysis_history_records_also_cleans_backtests(self) -> None:
         """删除历史记录时应一并清理关联回测结果。"""
         record_id = self._save_history("query_delete_001")

tests/test_analyzer_news_prompt.py

@@ -24,6 +24,14 @@ class AnalyzerNewsPromptTestCase(unittest.TestCase):
             "stock_name": "贵州茅台",
             "date": "2026-03-16",
             "today": {},
+            "fundamental_context": {
+                "earnings": {
+                    "data": {
+                        "financial_report": {"report_date": "2025-12-31", "revenue": 1000},
+                        "dividend": {"ttm_cash_dividend_per_share": 1.2, "ttm_dividend_yield_pct": 2.4},
+                    }
+                }
+            },
         }
         fake_cfg = SimpleNamespace(
             news_max_age_days=30,
@@ -36,6 +44,8 @@ class AnalyzerNewsPromptTestCase(unittest.TestCase):
         self.assertIn("每一条都必须带具体日期（YYYY-MM-DD）", prompt)
         self.assertIn("超出近7日窗口的新闻一律忽略", prompt)
         self.assertIn("时间未知、无法确定发布日期的新闻一律忽略", prompt)
+        self.assertIn("财报与分红（价值投资口径）", prompt)
+        self.assertIn("禁止编造", prompt)
 
     def test_prompt_prefers_context_news_window_days(self) -> None:
         with patch.object(GeminiAnalyzer, "_init_litellm", return_value=None):

tests/test_fundamental_adapter.py

@@ -6,16 +6,28 @@ Tests for fundamental adapter helpers.
 import os
 import sys
 import unittest
+from datetime import datetime, timedelta
 from unittest.mock import patch
 
 import pandas as pd
 
 sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), "..")))
 
-from data_provider.fundamental_adapter import AkshareFundamentalAdapter, _extract_latest_row
+from data_provider.fundamental_adapter import (
+    AkshareFundamentalAdapter,
+    _build_dividend_payload,
+    _extract_latest_row,
+    _parse_dividend_plan_to_per_share,
+)
 
 
 class TestFundamentalAdapter(unittest.TestCase):
+    def test_parse_dividend_plan_to_per_share_supports_cn_patterns(self) -> None:
+        self.assertAlmostEqual(_parse_dividend_plan_to_per_share("10派3元(含税)"), 0.3, places=6)
+        self.assertAlmostEqual(_parse_dividend_plan_to_per_share("每10股派发2.5元"), 0.25, places=6)
+        self.assertAlmostEqual(_parse_dividend_plan_to_per_share("每股派0.8元"), 0.8, places=6)
+        self.assertIsNone(_parse_dividend_plan_to_per_share("仅送股，不现金分红"))
+
     def test_extract_latest_row_returns_none_when_code_mismatch(self) -> None:
         df = pd.DataFrame(
             {
@@ -61,6 +73,103 @@ class TestFundamentalAdapter(unittest.TestCase):
         self.assertTrue(result["is_on_list"])
         self.assertGreaterEqual(result["recent_count"], 1)
 
+    def test_fundamental_bundle_includes_financial_report_and_dividend_payload(self) -> None:
+        adapter = AkshareFundamentalAdapter()
+        now = datetime.now()
+        within_ttm = (now - timedelta(days=30)).strftime("%Y-%m-%d")
+        future_day = (now + timedelta(days=10)).strftime("%Y-%m-%d")
+        old_day = (now - timedelta(days=500)).strftime("%Y-%m-%d")
+        fin_df = pd.DataFrame(
+            {
+                "股票代码": ["600519"],
+                "报告期": [within_ttm],
+                "营业总收入": [1000.0],
+                "归母净利润": [300.0],
+                "经营活动产生的现金流量净额": [500.0],
+                "净资产收益率": [18.2],
+                "营业收入同比": [12.0],
+                "净利润同比": [9.5],
+            }
+        )
+        forecast_df = pd.DataFrame({"股票代码": ["600519"], "预告": ["预增"]})
+        quick_df = pd.DataFrame({"股票代码": ["600519"], "快报": ["快报摘要"]})
+        dividend_df = pd.DataFrame(
+            {
+                "股票代码": ["600519", "600519", "600519", "600519"],
+                "除息日": [within_ttm, within_ttm, future_day, old_day],
+                "分配方案": ["10派3元(含税)", "10派3元(含税)", "10派5元", "10派1元"],
+            }
+        )
+
+        with patch.object(
+            adapter,
+            "_call_df_candidates",
+            side_effect=[
+                (fin_df, "stock_financial_abstract", []),
+                (forecast_df, "stock_yjyg_em", []),
+                (quick_df, "stock_yjkb_em", []),
+                (dividend_df, "stock_fhps_detail_em", []),
+                (None, None, []),
+                (None, None, []),
+            ],
+        ):
+            result = adapter.get_fundamental_bundle("600519")
+
+        financial_report = result["earnings"].get("financial_report", {})
+        self.assertEqual(financial_report.get("report_date"), within_ttm)
+        self.assertEqual(financial_report.get("revenue"), 1000.0)
+        self.assertEqual(financial_report.get("net_profit_parent"), 300.0)
+        self.assertEqual(financial_report.get("operating_cash_flow"), 500.0)
+        self.assertEqual(financial_report.get("roe"), 18.2)
+
+        dividend_payload = result["earnings"].get("dividend", {})
+        events = dividend_payload.get("events", [])
+        self.assertEqual(len(events), 2)  # duplicate + future day filtered
+        self.assertEqual(dividend_payload.get("ttm_event_count"), 1)
+        self.assertAlmostEqual(dividend_payload.get("ttm_cash_dividend_per_share"), 0.3, places=6)
+
+    def test_build_dividend_payload_returns_empty_when_code_not_matched(self) -> None:
+        now = datetime.now().strftime("%Y-%m-%d")
+        df = pd.DataFrame(
+            {
+                "股票代码": ["000001"],
+                "除息日": [now],
+                "分配方案": ["10派3元(含税)"],
+            }
+        )
+
+        payload = _build_dividend_payload(df, stock_code="600519")
+        self.assertEqual(payload, {})
+
+    def test_build_dividend_payload_skips_after_tax_plan(self) -> None:
+        now = datetime.now().strftime("%Y-%m-%d")
+        df = pd.DataFrame(
+            {
+                "股票代码": ["600519"],
+                "除息日": [now],
+                "分配方案": ["10派3元(税后)"],
+            }
+        )
+
+        payload = _build_dividend_payload(df, stock_code="600519")
+        self.assertEqual(payload, {})
+
+    def test_build_dividend_payload_ttm_window_boundary(self) -> None:
+        now = datetime.now()
+        day_365 = (now - timedelta(days=365)).strftime("%Y-%m-%d")
+        day_366 = (now - timedelta(days=366)).strftime("%Y-%m-%d")
+        df = pd.DataFrame(
+            {
+                "股票代码": ["600519", "600519"],
+                "除息日": [day_365, day_366],
+                "分配方案": ["10派3元(含税)", "10派5元(含税)"],
+            }
+        )
+
+        payload = _build_dividend_payload(df, stock_code="600519")
+        self.assertEqual(payload.get("ttm_event_count"), 1)
+        self.assertAlmostEqual(payload.get("ttm_cash_dividend_per_share"), 0.3, places=6)
+
 
 if __name__ == "__main__":
     unittest.main()

tests/test_fundamental_context.py

@@ -152,6 +152,89 @@ class TestFundamentalContext(unittest.TestCase):
         self.assertIn("capital_flow", ctx)
         self.assertIn("dragon_tiger", ctx)
 
+    def test_fundamental_context_derives_ttm_dividend_yield_from_quote_price(self) -> None:
+        manager = DataFetcherManager(fetchers=[])
+        cfg = SimpleNamespace(
+            enable_fundamental_pipeline=True,
+            fundamental_cache_ttl_seconds=120,
+            fundamental_stage_timeout_seconds=1.5,
+            fundamental_fetch_timeout_seconds=0.8,
+            fundamental_retry_max=1,
+        )
+        quote = SimpleNamespace(
+            price=50.0,
+            pe_ratio=12.3,
+            pb_ratio=2.1,
+            total_mv=1.0e11,
+            circ_mv=7.0e10,
+            source=SimpleNamespace(value="tencent"),
+        )
+        with patch("src.config.get_config", return_value=cfg), \
+                patch.object(manager, "get_realtime_quote", return_value=quote), \
+                patch("data_provider.fundamental_adapter.AkshareFundamentalAdapter.get_fundamental_bundle", return_value={
+                    "status": "partial",
+                    "growth": {},
+                    "earnings": {
+                        "dividend": {
+                            "ttm_cash_dividend_per_share": 2.5,
+                            "ttm_event_count": 1,
+                            "events": [{"event_date": "2026-01-01", "cash_dividend_per_share": 2.5}],
+                        }
+                    },
+                    "institution": {},
+                    "source_chain": [],
+                    "errors": [],
+                }), \
+                patch.object(manager, "get_capital_flow_context", return_value={"status": "not_supported", "source_chain": []}), \
+                patch.object(manager, "get_dragon_tiger_context", return_value={"status": "not_supported", "source_chain": []}), \
+                patch.object(manager, "get_board_context", return_value={"status": "not_supported", "source_chain": []}):
+            ctx = manager.get_fundamental_context("600519", budget_seconds=1.5)
+
+        dividend_payload = ctx["earnings"]["data"]["dividend"]
+        self.assertAlmostEqual(dividend_payload["ttm_dividend_yield_pct"], 5.0, places=6)
+        self.assertIn("yield_formula", dividend_payload)
+
+    def test_fundamental_context_dividend_yield_keeps_null_when_price_invalid(self) -> None:
+        manager = DataFetcherManager(fetchers=[])
+        cfg = SimpleNamespace(
+            enable_fundamental_pipeline=True,
+            fundamental_cache_ttl_seconds=120,
+            fundamental_stage_timeout_seconds=1.5,
+            fundamental_fetch_timeout_seconds=0.8,
+            fundamental_retry_max=1,
+        )
+        quote = SimpleNamespace(
+            price=None,
+            pe_ratio=12.3,
+            pb_ratio=2.1,
+            total_mv=1.0e11,
+            circ_mv=7.0e10,
+            source=SimpleNamespace(value="tencent"),
+        )
+        with patch("src.config.get_config", return_value=cfg), \
+                patch.object(manager, "get_realtime_quote", return_value=quote), \
+                patch("data_provider.fundamental_adapter.AkshareFundamentalAdapter.get_fundamental_bundle", return_value={
+                    "status": "partial",
+                    "growth": {},
+                    "earnings": {
+                        "dividend": {
+                            "ttm_cash_dividend_per_share": 1.2,
+                            "events": [{"event_date": "2026-01-01", "cash_dividend_per_share": 1.2}],
+                        }
+                    },
+                    "institution": {},
+                    "source_chain": [],
+                    "errors": [],
+                }), \
+                patch.object(manager, "get_capital_flow_context", return_value={"status": "not_supported", "source_chain": []}), \
+                patch.object(manager, "get_dragon_tiger_context", return_value={"status": "not_supported", "source_chain": []}), \
+                patch.object(manager, "get_board_context", return_value={"status": "not_supported", "source_chain": []}):
+            ctx = manager.get_fundamental_context("600519", budget_seconds=1.5)
+
+        dividend_payload = ctx["earnings"]["data"]["dividend"]
+        self.assertIsNone(dividend_payload.get("ttm_dividend_yield_pct"))
+        self.assertIn("invalid_price_for_ttm_dividend_yield", ctx["earnings"]["errors"])
+
     def test_non_etf_board_budget_not_forced_to_zero(self) -> None:
         manager = DataFetcherManager(fetchers=[])
         cfg = SimpleNamespace(