fanhuadesenlinnn/daily_stock_analysis
1
0
Fork 0
mirror of https://github.com/ZhuLinsen/daily_stock_analysis synced 2026-09-20 10:53:33 +08:00
Code Issues Packages Projects Releases Wiki Activity

feat: extract decision signals from analysis reports (#1686)

Browse Source
This commit is contained in:
Alfred
2026-06-14 18:26:30 +08:00
committed by GitHub
parent 145ee6b5a5
commit 6d4c93d2b7
16 changed files with 1127 additions and 263 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
docs/CHANGELOG.md
View File

@@ -30,6 +30,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/).
<!-- 每条独立一行追加到本段末尾，无需分类标题，合并时冲突最小 -->
- [修复] 发布说明生成查询 PR 作者失败时保留降级并输出包含 PR 编号和异常类型的 warning，便于排查 token、权限、网络或 GitHub API 异常。
- [新功能] 个股分析历史成功保存后会从最终报告 best-effort 提取 `DecisionSignal` 决策信号，复用现有信号去重、计划质量计算和脱敏契约。
## [3.22.0] - 2026-06-13
### 发布亮点

8
docs/full-guide.md
View File

@@ -1302,9 +1302,13 @@ python main.py --debug
#1390 P0 不会把后续信号资产字段平铺到现有 summary、历史列表、StockBar 或回测响应。#1390 P1 开始通过独立 `DecisionSignal` 资源承接 `horizon`、`plan_quality`、`status` 等更细粒度计划字段，仍不改变既有报告主契约、不回填历史、不新增配置项。
### 决策信号资产（#1390 P1）
### 决策信号资产（#1390 P1/P2）
`DecisionSignal` 是独立后端资源，用于把 AI 建议沉淀为可查询、可去重、可更新状态的信号资产。它不替换 `operation_advice`、不扩展 `decision_type=buy|hold|sell`，也不会自动从现有报告提取；P2 之前只有显式调用 API 或 service 的路径会写入信号。
`DecisionSignal` 是独立后端资源，用于把 AI 建议沉淀为可查询、可去重、可更新状态的信号资产。它不替换 `operation_advice`、不扩展 `decision_type=buy|hold|sell`。#1390 P2 开始，普通个股分析和 Agent 个股分析在分析历史保存成功后，会从最终 `AnalysisResult` best-effort 提取一条 `source_type=analysis` 的信号；显式 API 或 service 调用仍然保留。
自动提取只消费已生成报告中的结构化字段，不重新解析 Markdown，也不回填旧历史、不新增配置项、不改变报告主契约。提取失败、建议动作未知或歧义、非个股报告、无法识别市场时会跳过写入，不影响分析报告保存。`source_report_id` 使用刚保存的 `AnalysisHistory.id`；`trace_id` 优先使用运行诊断 trace，缺失时降级到 pipeline trace 或 `query_id`；`stock_name` 来自 `AnalysisResult.name`；`trigger_source` 来自运行入口，缺失时为 `system`。
P2 自动提取的市场阶段优先读取保存快照中的 `market_phase_summary.phase`，其次读取 `AnalysisResult.market_phase_summary.phase`；数据质量优先读取保存快照中的 `analysis_context_pack_overview.data_quality`，其次读取 `AnalysisResult.analysis_context_pack_overview.data_quality`。价格计划复用历史保存的狙击点解析规则，从 `dashboard.battle_plan.sniper_points.ideal_buy/secondary_buy/stop_loss/take_profit` 映射到 `entry_low/entry_high/stop_loss/target_price`；只有 `ideal_buy` 时写入 `entry_low`，只有 `secondary_buy` 时写入 `entry_high`，两者同时存在时按有效价格排序为 `entry_low <= entry_high`。缺失止损或目标价只会降低 service 自动计算的 `plan_quality`，不会编造字段。`watch_conditions` 优先读取 `dashboard.phase_decision.watch_conditions`，没有时才读取 `dashboard.battle_plan.action_checklist`；`catalyst_summary` 仅在 `dashboard.intelligence.positive_catalysts` 存在且为列表时写入。`confidence` 由报告置信等级做保守映射：`高/high=0.8`、`中/medium/mid=0.6`、`低/low=0.4`，原始置信等级保留在 `metadata`。
核心字段包括 `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`。

8
docs/full-guide_EN.md
View File

@@ -1131,9 +1131,13 @@ Unknown or ambiguous advice is not coerced into `watch` or `hold`; it returns em
#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)
### Decision Signal Asset (#1390 P1/P2)
`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.
`DecisionSignal` is an independent backend resource for persisting AI recommendations as queryable, deduplicated, status-updatable signal assets. It does not replace `operation_advice` or expand the legacy `decision_type=buy|hold|sell` contract. Starting with #1390 P2, regular stock analysis and Agent stock analysis best-effort extract one `source_type=analysis` signal from the final `AnalysisResult` after analysis history is saved successfully; explicit API and service calls remain supported.
Automatic extraction consumes structured fields from the completed report only. It does not parse Markdown, backfill old history, add configuration, or change the main report contract. Extraction failures, unknown or ambiguous advice, non-stock reports, and unrecognized markets skip signal writes without affecting report persistence. `source_report_id` is the just-saved `AnalysisHistory.id`; `trace_id` prefers the runtime diagnostics trace and falls back to the pipeline trace or `query_id`; `stock_name` comes from `AnalysisResult.name`; `trigger_source` comes from the runtime entrypoint and falls back to `system`.
For P2 automatic extraction, `market_phase` first reads `market_phase_summary.phase` from the saved context snapshot and then falls back to `AnalysisResult.market_phase_summary.phase`; data quality first reads `analysis_context_pack_overview.data_quality` from the saved context snapshot and then falls back to `AnalysisResult.analysis_context_pack_overview.data_quality`. Price-plan extraction reuses the same sniper-point parser used by history persistence, mapping `dashboard.battle_plan.sniper_points.ideal_buy/secondary_buy/stop_loss/take_profit` to `entry_low/entry_high/stop_loss/target_price`; `ideal_buy` alone writes `entry_low`, `secondary_buy` alone writes `entry_high`, and when both are present they are sorted into `entry_low <= entry_high`. Missing stop-loss or target prices only lower the service-computed `plan_quality` instead of inventing fields. `watch_conditions` first reads `dashboard.phase_decision.watch_conditions` and then falls back to `dashboard.battle_plan.action_checklist`. `catalyst_summary` is written only when `dashboard.intelligence.positive_catalysts` exists and is a list. `confidence` uses a conservative report-level mapping: `高/high=0.8`, `中/medium/mid=0.6`, `低/low=0.4`; the original report confidence level remains in `metadata`.
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`.

53
src/core/market_review.py
View File

@@ -553,7 +553,7 @@ def _persist_market_review_history(
)
db = DatabaseManager.get_instance()
saved = db.save_analysis_history(
saved_history_id = db.save_analysis_history(
result=result,
query_id=history_query_id,
report_type=MARKET_REVIEW_REPORT_TYPE,
@@ -561,21 +561,26 @@ def _persist_market_review_history(
context_snapshot=context_snapshot,
save_snapshot=True,
)
saved_history_id = _resolve_saved_market_review_history_id(
db=db,
query_id=history_query_id,
) if saved else None
valid_saved_history_id = (
saved_history_id
if (
isinstance(saved_history_id, int)
and not isinstance(saved_history_id, bool)
and saved_history_id > 0
)
else None
)
record_history_run(
report_saved=bool(saved),
metadata_saved=bool(saved),
analysis_history_id=saved_history_id,
report_saved=bool(saved_history_id),
metadata_saved=bool(saved_history_id),
analysis_history_id=valid_saved_history_id,
)
_refresh_market_review_history_diagnostics(query_id=history_query_id)
if saved:
if saved_history_id:
logger.info("大盘复盘历史记录已保存: query_id=%s", history_query_id)
else:
logger.warning("大盘复盘历史记录保存失败: query_id=%s", history_query_id)
return saved
return saved_history_id
except Exception as exc:
record_history_run(
report_saved=False,
@@ -645,34 +650,6 @@ def _build_market_review_context_overview(
}
def _resolve_saved_market_review_history_id(
*,
db: object,
query_id: str,
) -> Optional[int]:
"""Resolve the real AnalysisHistory primary key after save_analysis_history returns row count."""
try:
resolver = getattr(db, "get_latest_analysis_by_query_id", None)
if not callable(resolver):
return None
record = resolver(
query_id,
code=MARKET_REVIEW_HISTORY_CODE,
report_type=MARKET_REVIEW_REPORT_TYPE,
)
record_id = getattr(record, "id", None)
return record_id if isinstance(record_id, int) and not isinstance(record_id, bool) else None
except TypeError:
try:
record = db.get_latest_analysis_by_query_id(query_id)
record_id = getattr(record, "id", None)
return record_id if isinstance(record_id, int) and not isinstance(record_id, bool) else None
except Exception:
return None
except Exception:
return None
def _summarize_market_review(review_report: str, report_language: str) -> str:
for line in (review_report or "").splitlines():
text = line.strip().lstrip("#").strip()

90
src/core/pipeline.py
View File

@@ -70,6 +70,7 @@ from src.services.run_diagnostics import (
reset_run_diagnostic_context,
sanitize_diagnostic_text,
)
from src.services.decision_signal_extractor import extract_and_persist_from_analysis_result
from src.enums import ReportType
from src.stock_analyzer import StockTrendAnalyzer, TrendAnalysisResult
from src.core.trading_calendar import (
@@ -707,7 +708,7 @@ class StockAnalysisPipeline:
market_phase_summary=market_phase_summary,
)
result.diagnostic_context_snapshot = context_snapshot
saved_count = self.db.save_analysis_history(
saved_history_id = self.db.save_analysis_history(
result=result,
query_id=query_id,
report_type=report_type.value,
@@ -715,10 +716,26 @@ class StockAnalysisPipeline:
context_snapshot=context_snapshot,
save_snapshot=self.save_context_snapshot
)
record_history_run(
report_saved=bool(saved_count),
metadata_saved=bool(saved_count),
valid_saved_history_id = (
isinstance(saved_history_id, int)
and not isinstance(saved_history_id, bool)
and saved_history_id > 0
)
record_history_run(
report_saved=bool(saved_history_id),
metadata_saved=bool(saved_history_id),
analysis_history_id=(
saved_history_id if valid_saved_history_id else None
),
)
if valid_saved_history_id:
self._extract_decision_signal_after_history_save(
result=result,
query_id=query_id,
source_report_id=saved_history_id,
report_type=report_type.value,
context_snapshot=context_snapshot,
)
except Exception as e:
record_history_run(
report_saved=False,
@@ -1274,7 +1291,7 @@ class StockAnalysisPipeline:
)
result.diagnostic_context_snapshot = agent_context_snapshot
agent_context_snapshot["stock_name"] = resolved_stock_name
saved_count = self.db.save_analysis_history(
saved_history_id = self.db.save_analysis_history(
result=result,
query_id=query_id,
report_type=report_type.value,
@@ -1282,10 +1299,26 @@ class StockAnalysisPipeline:
context_snapshot=agent_context_snapshot,
save_snapshot=self.save_context_snapshot,
)
record_history_run(
report_saved=bool(saved_count),
metadata_saved=bool(saved_count),
valid_saved_history_id = (
isinstance(saved_history_id, int)
and not isinstance(saved_history_id, bool)
and saved_history_id > 0
)
record_history_run(
report_saved=bool(saved_history_id),
metadata_saved=bool(saved_history_id),
analysis_history_id=(
saved_history_id if valid_saved_history_id else None
),
)
if valid_saved_history_id:
self._extract_decision_signal_after_history_save(
result=result,
query_id=query_id,
source_report_id=saved_history_id,
report_type=report_type.value,
context_snapshot=agent_context_snapshot,
)
latest_diagnostic_snapshot = current_diagnostic_snapshot()
if latest_diagnostic_snapshot is not None:
agent_context_snapshot["diagnostics"] = latest_diagnostic_snapshot
@@ -2067,6 +2100,47 @@ class StockAnalysisPipeline:
snapshot["skills"] = list(self.analysis_skills)
return snapshot
def _extract_decision_signal_after_history_save(
self,
*,
result: AnalysisResult,
query_id: str,
source_report_id: int,
report_type: str,
context_snapshot: Dict[str, Any],
) -> None:
"""Best-effort DecisionSignal extraction after analysis history is saved."""
assert (
isinstance(source_report_id, int)
and not isinstance(source_report_id, bool)
and source_report_id > 0
)
try:
diagnostic_context = get_current_diagnostic_context()
trace_id = (
getattr(diagnostic_context, "trace_id", None)
or getattr(self, "trace_id", None)
or query_id
)
extract_and_persist_from_analysis_result(
result,
context_snapshot=context_snapshot,
source_report_id=source_report_id,
trace_id=str(trace_id),
query_source=getattr(self, "query_source", None) or "system",
report_type=report_type,
)
except Exception as exc:
logger.warning(
"Decision signal extraction skipped after history save: query_id=%s stock_code=%s error=%s",
query_id,
getattr(result, "code", None),
exc,
exc_info=True,
)
@staticmethod
def _build_notification_run_snapshot(
*,

2
src/repositories/analysis_repo.py
View File

@@ -97,7 +97,7 @@ class AnalysisRepository:
context_snapshot: 上下文快照
Returns:
保存的记录数
新保存的 AnalysisHistory.id；保存失败返回 0。
"""
try:
return self.db.save_analysis_history(

238
src/services/decision_signal_extractor.py Normal file
View File

@@ -0,0 +1,238 @@
# -*- coding: utf-8 -*-
"""Extract DecisionSignal payloads from completed analysis reports."""
from __future__ import annotations
import logging
import math
from typing import Any, Dict, Mapping, Optional
from data_provider.base import normalize_stock_code
from src.analyzer import AnalysisResult
from src.core.trading_calendar import get_market_for_stock
from src.schemas.decision_action import build_action_fields
from src.services.decision_signal_service import DecisionSignalService
from src.utils.sniper_points import extract_sniper_points
logger = logging.getLogger(__name__)
_CONFIDENCE_MAP = {
"高": 0.8,
"high": 0.8,
"中": 0.6,
"medium": 0.6,
"mid": 0.6,
"低": 0.4,
"low": 0.4,
}
def build_decision_signal_payload_from_report(
result: AnalysisResult,
*,
context_snapshot: Dict[str, Any] | None = None,
source_report_id: int | None = None,
trace_id: str,
query_source: str,
report_type: str,
) -> Dict[str, Any] | None:
"""Build a DecisionSignal payload from a completed stock analysis report."""
if result is None or not getattr(result, "success", True):
return None
action_fields = build_action_fields(
operation_advice=getattr(result, "operation_advice", None),
explicit_action=getattr(result, "action", None),
report_type=report_type,
report_language=getattr(result, "report_language", None),
)
action = action_fields.get("action")
if not action:
return None
raw_code = str(getattr(result, "code", "") or "").strip()
market = get_market_for_stock(normalize_stock_code(raw_code))
if not market:
logger.warning("Skip decision signal extraction: unrecognized market stock_code=%s", raw_code)
return None
dashboard = _as_mapping(getattr(result, "dashboard", None))
sniper_points = extract_sniper_points(result)
entry_low, entry_high = _entry_range(
sniper_points.get("ideal_buy"),
sniper_points.get("secondary_buy"),
)
payload: Dict[str, Any] = {
"stock_code": raw_code,
"stock_name": getattr(result, "name", None),
"market": market,
"source_type": "analysis",
"source_report_id": source_report_id,
"trace_id": trace_id,
"market_phase": _extract_market_phase(context_snapshot, result),
"trigger_source": str(query_source or "").strip() or "system",
"action": action,
"action_label": action_fields.get("action_label"),
"confidence": _confidence_from_level(getattr(result, "confidence_level", None)),
"score": _score_from_result(getattr(result, "sentiment_score", None)),
"entry_low": entry_low,
"entry_high": entry_high,
"stop_loss": sniper_points.get("stop_loss"),
"target_price": sniper_points.get("take_profit"),
"reason": _first_text(
getattr(result, "analysis_summary", None),
getattr(result, "buy_reason", None),
getattr(result, "key_points", None),
),
"risk_summary": _risk_summary(result, dashboard),
"catalyst_summary": _catalyst_summary(dashboard),
"watch_conditions": _watch_conditions(dashboard),
"evidence": _evidence(result, sniper_points),
"data_quality_summary": _extract_data_quality(context_snapshot, result),
"metadata": {
"report_type": report_type,
"decision_type": getattr(result, "decision_type", None),
"report_confidence_level": getattr(result, "confidence_level", None),
"report_language": getattr(result, "report_language", None),
},
"report_language": getattr(result, "report_language", None),
}
return {key: value for key, value in payload.items() if value not in (None, "", [], {})}
def extract_and_persist_from_analysis_result(
result: AnalysisResult,
*,
context_snapshot: Dict[str, Any] | None = None,
source_report_id: int | None = None,
trace_id: str,
query_source: str,
report_type: str,
service: Optional[DecisionSignalService] = None,
) -> Dict[str, Any] | None:
"""Best-effort extract and persist a DecisionSignal from an analysis result."""
try:
payload = build_decision_signal_payload_from_report(
result,
context_snapshot=context_snapshot,
source_report_id=source_report_id,
trace_id=trace_id,
query_source=query_source,
report_type=report_type,
)
if payload is None:
return None
writer = service or DecisionSignalService()
return writer.create_signal(payload)
except Exception as exc:
logger.warning(
"Decision signal extraction failed: query_id=%s stock_code=%s error=%s",
trace_id,
getattr(result, "code", None),
exc,
exc_info=True,
)
return None
def _as_mapping(value: Any) -> Dict[str, Any]:
return dict(value) if isinstance(value, Mapping) else {}
def _first_text(*values: Any) -> Optional[str]:
for value in values:
text = str(value or "").strip()
if text:
return text
return None
def _score_from_result(value: Any) -> Optional[int]:
try:
score = int(float(value))
except (TypeError, ValueError):
return None
return score if 0 <= score <= 100 else None
def _confidence_from_level(value: Any) -> Optional[float]:
key = str(value or "").strip().lower()
return _CONFIDENCE_MAP.get(key)
def _entry_range(ideal_buy: Optional[float], secondary_buy: Optional[float]) -> tuple[Optional[float], Optional[float]]:
"""Return numeric entry bounds while preserving single-value source semantics."""
low = ideal_buy if ideal_buy is not None and math.isfinite(ideal_buy) and ideal_buy > 0 else None
high = secondary_buy if secondary_buy is not None and math.isfinite(secondary_buy) and secondary_buy > 0 else None
if low is not None and high is not None and low > high:
return high, low
return low, high
def _extract_market_phase(context_snapshot: Optional[Mapping[str, Any]], result: AnalysisResult) -> Optional[str]:
snapshot_phase = _as_mapping(_as_mapping(context_snapshot).get("market_phase_summary")).get("phase")
if snapshot_phase:
return str(snapshot_phase)
result_phase = _as_mapping(getattr(result, "market_phase_summary", None)).get("phase")
return str(result_phase) if result_phase else None
def _extract_data_quality(context_snapshot: Optional[Mapping[str, Any]], result: AnalysisResult) -> Optional[Any]:
snapshot_quality = _as_mapping(
_as_mapping(context_snapshot).get("analysis_context_pack_overview")
).get("data_quality")
if snapshot_quality:
return snapshot_quality
return _as_mapping(getattr(result, "analysis_context_pack_overview", None)).get("data_quality")
def _risk_summary(result: AnalysisResult, dashboard: Mapping[str, Any]) -> Optional[Any]:
risks = []
risk_warning = getattr(result, "risk_warning", None)
if risk_warning:
risks.append(str(risk_warning))
intelligence = _as_mapping(dashboard.get("intelligence"))
risk_alerts = intelligence.get("risk_alerts")
if isinstance(risk_alerts, list):
risks.extend(str(item) for item in risk_alerts if str(item or "").strip())
return risks[:5] or None
def _catalyst_summary(dashboard: Mapping[str, Any]) -> Optional[Any]:
catalysts = _as_mapping(dashboard.get("intelligence")).get("positive_catalysts")
if not isinstance(catalysts, list):
return None
out = [str(item) for item in catalysts if str(item or "").strip()]
return out[:5] or None
def _watch_conditions(dashboard: Mapping[str, Any]) -> Optional[Any]:
phase_decision = _as_mapping(dashboard.get("phase_decision"))
watch_conditions = phase_decision.get("watch_conditions")
if isinstance(watch_conditions, list) and watch_conditions:
return [str(item) for item in watch_conditions if str(item or "").strip()] or None
battle_plan = _as_mapping(dashboard.get("battle_plan"))
checklist = battle_plan.get("action_checklist")
if isinstance(checklist, list) and checklist:
return [str(item) for item in checklist if str(item or "").strip()] or None
return None
def _evidence(result: AnalysisResult, sniper_points: Mapping[str, Any]) -> Dict[str, Any]:
evidence = {
"operation_advice": getattr(result, "operation_advice", None),
"decision_type": getattr(result, "decision_type", None),
"trend_prediction": getattr(result, "trend_prediction", None),
"confidence_level": getattr(result, "confidence_level", None),
"current_price": getattr(result, "current_price", None),
"change_pct": getattr(result, "change_pct", None),
"sniper_points": dict(sniper_points),
}
return {key: value for key, value in evidence.items() if value not in (None, "", [], {})}

3
src/services/history_service.py
View File

@@ -33,6 +33,7 @@ from src.storage import DatabaseManager
from src.services.run_diagnostics import build_run_diagnostic_summary
from src.market_phase_summary import extract_market_phase_summary
from src.schemas.decision_action import build_action_fields
from src.utils.sniper_points import find_sniper_points
from src.utils.data_processing import (
extract_realtime_detail_fields,
normalize_model_used,
@@ -470,7 +471,7 @@ class HistoryService:
for candidate in (raw_result.get("dashboard"), raw_result):
if not isinstance(candidate, dict):
continue
raw_points = DatabaseManager._find_sniper_in_dashboard(candidate) or raw_points
raw_points = find_sniper_points(candidate) or raw_points
if any(raw_points.get(k) is not None for k in ("ideal_buy", "secondary_buy", "stop_loss", "take_profit")):
break

215
src/storage.py
View File

@@ -16,7 +16,6 @@ from contextlib import contextmanager
import hashlib
import json
import logging
import re
import threading
import time
from datetime import datetime, date, timedelta, timezone
@@ -54,6 +53,7 @@ from sqlalchemy.exc import IntegrityError, OperationalError
from src.agent.provider_trace import PROVIDER_TRACE_RETENTION_LIMIT
from src.config import get_config
from src.utils.sniper_points import extract_sniper_points, parse_sniper_value
logger = logging.getLogger(__name__)
T = TypeVar("T")
@@ -1460,7 +1460,10 @@ class DatabaseManager(metaclass=_DatabaseManagerMeta):
save_snapshot: bool = True
) -> int:
"""
保存分析结果历史记录
保存分析结果历史记录。
Returns:
新保存的 AnalysisHistory.id；保存失败返回 0。
"""
if result is None:
return 0
@@ -1473,27 +1476,27 @@ class DatabaseManager(metaclass=_DatabaseManagerMeta):
try:
def _write(session: Session) -> int:
session.add(
AnalysisHistory(
query_id=query_id,
code=result.code,
name=result.name,
report_type=report_type,
sentiment_score=result.sentiment_score,
operation_advice=result.operation_advice,
trend_prediction=result.trend_prediction,
analysis_summary=result.analysis_summary,
raw_result=self._safe_json_dumps(raw_result),
news_content=news_content,
context_snapshot=context_text,
ideal_buy=sniper_points.get("ideal_buy"),
secondary_buy=sniper_points.get("secondary_buy"),
stop_loss=sniper_points.get("stop_loss"),
take_profit=sniper_points.get("take_profit"),
created_at=datetime.now(),
)
history = AnalysisHistory(
query_id=query_id,
code=result.code,
name=result.name,
report_type=report_type,
sentiment_score=result.sentiment_score,
operation_advice=result.operation_advice,
trend_prediction=result.trend_prediction,
analysis_summary=result.analysis_summary,
raw_result=self._safe_json_dumps(raw_result),
news_content=news_content,
context_snapshot=context_text,
ideal_buy=sniper_points.get("ideal_buy"),
secondary_buy=sniper_points.get("secondary_buy"),
stop_loss=sniper_points.get("stop_loss"),
take_profit=sniper_points.get("take_profit"),
created_at=datetime.now(),
)
return 1
session.add(history)
session.flush()
return int(history.id or 0)
return self._run_write_transaction(
f"save_analysis_history[{result.code}]",
_write,
@@ -1622,6 +1625,34 @@ class DatabaseManager(metaclass=_DatabaseManagerMeta):
).scalars().all()
return list(results)
def get_latest_analysis_history_id(
self,
*,
query_id: str,
code: str,
report_type: str,
) -> Optional[int]:
"""Return the latest matching history id for read-only lookups.
P2 automatic DecisionSignal extraction receives the freshly saved id
directly from ``save_analysis_history()`` and does not use this helper.
"""
if not query_id or not code or not report_type:
return None
with self.get_session() as session:
return session.execute(
select(AnalysisHistory.id)
.where(
AnalysisHistory.query_id == query_id,
AnalysisHistory.code == code,
AnalysisHistory.report_type == report_type,
)
.order_by(desc(AnalysisHistory.created_at), desc(AnalysisHistory.id))
.limit(1)
).scalar_one_or_none()
def get_analysis_history_paginated(
self,
@@ -2177,146 +2208,12 @@ class DatabaseManager(metaclass=_DatabaseManagerMeta):
@staticmethod
def _parse_sniper_value(value: Any) -> Optional[float]:
"""
Parse a sniper point value from various formats to float.
Handles: numeric types, plain number strings, Chinese price formats
like "18.50元", range formats like "18.50-19.00", and text with
embedded numbers while filtering out MA indicators.
"""
if value is None:
return None
if isinstance(value, (int, float)):
v = float(value)
return v if v > 0 else None
text = str(value).replace(',', '').replace('，', '').strip()
if not text or text == '-' or text == '—' or text == 'N/A':
return None
# 尝试直接解析纯数字字符串
try:
return float(text)
except ValueError:
pass
# 优先截取 "：" 到 "元" 之间的价格，避免误提取 MA5/MA10 等技术指标数字
colon_pos = max(text.rfind("："), text.rfind(":"))
yuan_pos = text.find("元", colon_pos + 1 if colon_pos != -1 else 0)
if yuan_pos != -1:
segment_start = colon_pos + 1 if colon_pos != -1 else 0
segment = text[segment_start:yuan_pos]
# 使用 finditer 并过滤掉 MA 开头的数字
matches = list(re.finditer(r"-?\d+(?:\.\d+)?", segment))
valid_numbers = []
for m in matches:
# 检查前面是否是 "MA" (忽略大小写)
start_idx = m.start()
if start_idx >= 2:
prefix = segment[start_idx-2:start_idx].upper()
if prefix == "MA":
continue
valid_numbers.append(m.group())
if valid_numbers:
try:
return abs(float(valid_numbers[-1]))
except ValueError:
pass
# 兜底：无"元"字时，先截去第一个括号后的内容，避免误提取括号内技术指标数字
# 例如 "1.52-1.53 (回踩MA5/10附近)" → 仅在 "1.52-1.53 " 中搜索
paren_pos = len(text)
for paren_char in ('(', '（'):
pos = text.find(paren_char)
if pos != -1:
paren_pos = min(paren_pos, pos)
search_text = text[:paren_pos].strip() or text # 括号前为空时降级用全文
valid_numbers = []
for m in re.finditer(r"\d+(?:\.\d+)?", search_text):
start_idx = m.start()
if start_idx >= 2 and search_text[start_idx-2:start_idx].upper() == "MA":
continue
valid_numbers.append(m.group())
if valid_numbers:
try:
return float(valid_numbers[-1])
except ValueError:
pass
return None
return parse_sniper_value(value)
def _extract_sniper_points(self, result: Any) -> Dict[str, Optional[float]]:
"""
Extract sniper point values from an AnalysisResult.
"""Extract normalized sniper point values from an AnalysisResult."""
Tries multiple extraction paths to handle different dashboard structures:
1. result.get_sniper_points() (standard path)
2. Direct dashboard dict traversal with various nesting levels
3. Fallback from raw_result dict if available
"""
raw_points = {}
# Path 1: standard method
if hasattr(result, "get_sniper_points"):
raw_points = result.get_sniper_points() or {}
# Path 2: direct dashboard traversal when standard path yields empty values
if not any(raw_points.get(k) for k in ("ideal_buy", "secondary_buy", "stop_loss", "take_profit")):
dashboard = getattr(result, "dashboard", None)
if isinstance(dashboard, dict):
raw_points = self._find_sniper_in_dashboard(dashboard) or raw_points
# Path 3: try raw_result for agent mode results
if not any(raw_points.get(k) for k in ("ideal_buy", "secondary_buy", "stop_loss", "take_profit")):
raw_response = getattr(result, "raw_response", None)
if isinstance(raw_response, dict):
raw_points = self._find_sniper_in_dashboard(raw_response) or raw_points
return {
"ideal_buy": self._parse_sniper_value(raw_points.get("ideal_buy")),
"secondary_buy": self._parse_sniper_value(raw_points.get("secondary_buy")),
"stop_loss": self._parse_sniper_value(raw_points.get("stop_loss")),
"take_profit": self._parse_sniper_value(raw_points.get("take_profit")),
}
@staticmethod
def _find_sniper_in_dashboard(d: dict) -> Optional[Dict[str, Any]]:
"""
Recursively search for sniper_points in a dashboard dict.
Handles various nesting: dashboard.battle_plan.sniper_points,
dashboard.dashboard.battle_plan.sniper_points, etc.
"""
if not isinstance(d, dict):
return None
# Direct: d has sniper_points keys at top level
if "ideal_buy" in d:
return d
# d.sniper_points
sp = d.get("sniper_points")
if isinstance(sp, dict) and sp:
return sp
# d.battle_plan.sniper_points
bp = d.get("battle_plan")
if isinstance(bp, dict):
sp = bp.get("sniper_points")
if isinstance(sp, dict) and sp:
return sp
# d.dashboard.battle_plan.sniper_points (double-nested)
inner = d.get("dashboard")
if isinstance(inner, dict):
bp = inner.get("battle_plan")
if isinstance(bp, dict):
sp = bp.get("sniper_points")
if isinstance(sp, dict) and sp:
return sp
return None
return extract_sniper_points(result)
@staticmethod
def _build_fallback_url_key(

123
src/utils/sniper_points.py Normal file
View File

@@ -0,0 +1,123 @@
# -*- coding: utf-8 -*-
"""Helpers for parsing report sniper-point price values."""
from __future__ import annotations
import re
from collections.abc import Mapping
from typing import Any, Dict, Optional
SNIPER_KEYS = ("ideal_buy", "secondary_buy", "stop_loss", "take_profit")
def parse_sniper_value(value: Any) -> Optional[float]:
"""Parse a sniper point value from report text into a positive price."""
if value is None:
return None
if isinstance(value, (int, float)):
parsed = float(value)
return parsed if parsed > 0 else None
text = str(value).replace(",", "").replace("，", "").strip()
if not text or text in {"-", "—", "N/A"}:
return None
try:
parsed = float(text)
return parsed if parsed > 0 else None
except ValueError:
pass
colon_pos = max(text.rfind("："), text.rfind(":"))
yuan_pos = text.find("元", colon_pos + 1 if colon_pos != -1 else 0)
if yuan_pos != -1:
segment_start = colon_pos + 1 if colon_pos != -1 else 0
segment = text[segment_start:yuan_pos]
valid_numbers = []
for match in re.finditer(r"-?\d+(?:\.\d+)?", segment):
start_idx = match.start()
if start_idx >= 2 and segment[start_idx - 2:start_idx].upper() == "MA":
continue
valid_numbers.append(match.group())
if valid_numbers:
try:
parsed = abs(float(valid_numbers[-1]))
return parsed if parsed > 0 else None
except ValueError:
pass
paren_pos = len(text)
for paren_char in ("(", "（"):
pos = text.find(paren_char)
if pos != -1:
paren_pos = min(paren_pos, pos)
search_text = text[:paren_pos].strip() or text
valid_numbers = []
for match in re.finditer(r"\d+(?:\.\d+)?", search_text):
start_idx = match.start()
if start_idx >= 2 and search_text[start_idx - 2:start_idx].upper() == "MA":
continue
valid_numbers.append(match.group())
if valid_numbers:
try:
parsed = float(valid_numbers[-1])
return parsed if parsed > 0 else None
except ValueError:
pass
return None
def extract_sniper_points(result: Any) -> Dict[str, Optional[float]]:
"""Extract normalized sniper-point prices from a completed analysis result."""
raw_points: Mapping[str, Any] = {}
if hasattr(result, "get_sniper_points"):
candidate = result.get_sniper_points() or {}
if isinstance(candidate, Mapping):
raw_points = candidate
if not _has_any_sniper_value(raw_points):
dashboard = getattr(result, "dashboard", None)
if isinstance(dashboard, Mapping):
raw_points = find_sniper_points(dashboard) or raw_points
if not _has_any_sniper_value(raw_points):
raw_response = getattr(result, "raw_response", None)
if isinstance(raw_response, Mapping):
raw_points = find_sniper_points(raw_response) or raw_points
return {key: parse_sniper_value(raw_points.get(key)) for key in SNIPER_KEYS}
def _has_any_sniper_value(points: Mapping[str, Any]) -> bool:
return any(points.get(key) not in (None, "") for key in SNIPER_KEYS)
def find_sniper_points(data: Mapping[str, Any]) -> Optional[Mapping[str, Any]]:
if not isinstance(data, Mapping):
return None
if any(key in data for key in SNIPER_KEYS):
return data
sniper_points = data.get("sniper_points")
if isinstance(sniper_points, Mapping) and sniper_points:
return sniper_points
battle_plan = data.get("battle_plan")
if isinstance(battle_plan, Mapping):
sniper_points = battle_plan.get("sniper_points")
if isinstance(sniper_points, Mapping) and sniper_points:
return sniper_points
inner_dashboard = data.get("dashboard")
if isinstance(inner_dashboard, Mapping):
found = find_sniper_points(inner_dashboard)
if found:
return found
return None

175
tests/test_analysis_history.py
View File

@@ -150,12 +150,13 @@ class AnalysisHistoryTestCase(unittest.TestCase):
context_snapshot=None,
save_snapshot=False,
)
self.assertEqual(saved, 1)
self.assertGreater(saved, 0)
with self.db.get_session() as session:
row = session.query(AnalysisHistory).filter(AnalysisHistory.query_id == query_id).first()
if row is None:
self.fail("未找到保存的历史记录")
self.assertEqual(row.id, saved)
return row.id
def test_save_analysis_history_with_snapshot(self) -> None:
@@ -182,7 +183,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
save_snapshot=True
)
self.assertEqual(saved, 1)
self.assertGreater(saved, 0)
history = self.db.get_analysis_history(code="600519", days=7, limit=10)
self.assertEqual(len(history), 1)
@@ -191,6 +192,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
row = session.query(AnalysisHistory).first()
if row is None:
self.fail("未找到保存的历史记录")
self.assertEqual(row.id, saved)
self.assertEqual(row.query_id, "query_001")
self.assertIsNotNone(row.context_snapshot)
self.assertEqual(row.ideal_buy, 125.5)
@@ -198,6 +200,75 @@ class AnalysisHistoryTestCase(unittest.TestCase):
self.assertEqual(row.stop_loss, 110.0)
self.assertEqual(row.take_profit, 150.0)
def test_save_analysis_history_persists_sniper_columns_via_shared_parser(self) -> None:
"""迁出 sniper parser 后历史狙击点位列仍按原规则保存。"""
result = self._build_result()
result.dashboard = {
"battle_plan": {
"sniper_points": {
"ideal_buy": "理想买入点：125.5元",
"secondary_buy": "1.52-1.53 (回踩MA5/10附近)",
"stop_loss": "—",
"take_profit": "目标位：150.0元",
}
}
}
saved = self.db.save_analysis_history(
result=result,
query_id="query_shared_sniper_parser",
report_type="simple",
news_content="新闻摘要",
context_snapshot=None,
save_snapshot=False,
)
self.assertGreater(saved, 0)
with self.db.get_session() as session:
row = session.query(AnalysisHistory).filter(
AnalysisHistory.query_id == "query_shared_sniper_parser"
).first()
if row is None:
self.fail("未找到保存的历史记录")
self.assertEqual(row.id, saved)
self.assertEqual(row.ideal_buy, 125.5)
self.assertEqual(row.secondary_buy, 1.53)
self.assertIsNone(row.stop_loss)
self.assertEqual(row.take_profit, 150.0)
def test_get_latest_analysis_history_id_filters_by_report_type_and_latest_record(self) -> None:
"""按 query/code/report_type 返回最新真实历史主键。"""
for report_type in ("simple", "full", "simple"):
saved = self.db.save_analysis_history(
result=self._build_result(),
query_id="query_latest_id",
report_type=report_type,
news_content="新闻摘要",
context_snapshot=None,
save_snapshot=False,
)
self.assertGreater(saved, 0)
simple_id = self.db.get_latest_analysis_history_id(
query_id="query_latest_id",
code="600519",
report_type="simple",
)
full_id = self.db.get_latest_analysis_history_id(
query_id="query_latest_id",
code="600519",
report_type="full",
)
self.assertIsNotNone(simple_id)
self.assertIsNotNone(full_id)
self.assertGreater(simple_id, full_id)
def test_get_latest_analysis_history_id_requires_report_type(self) -> None:
"""report_type 是必传参数，避免误取同 query/code 的其他报告。"""
with self.assertRaises(TypeError):
self.db.get_latest_analysis_history_id(query_id="query", code="600519")
def test_save_analysis_history_without_snapshot(self) -> None:
"""关闭快照保存时不写入 context_snapshot"""
result = self._build_result()
@@ -211,12 +282,13 @@ class AnalysisHistoryTestCase(unittest.TestCase):
save_snapshot=False
)
self.assertEqual(saved, 1)
self.assertGreater(saved, 0)
with self.db.get_session() as session:
row = session.query(AnalysisHistory).first()
if row is None:
self.fail("未找到保存的历史记录")
self.assertEqual(row.id, saved)
self.assertIsNone(row.context_snapshot)
def test_save_analysis_history_persists_model_used(self) -> None:
@@ -232,12 +304,13 @@ class AnalysisHistoryTestCase(unittest.TestCase):
context_snapshot=None,
save_snapshot=False
)
self.assertEqual(saved, 1)
self.assertGreater(saved, 0)
with self.db.get_session() as session:
row = session.query(AnalysisHistory).filter(AnalysisHistory.query_id == "query_003").first()
if row is None:
self.fail("未找到保存的历史记录")
self.assertEqual(row.id, saved)
payload = json.loads(row.raw_result or "{}")
self.assertEqual(payload.get("model_used"), "gemini/gemini-2.0-flash")
@@ -259,7 +332,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
},
save_snapshot=True,
)
self.assertEqual(saved, 1)
self.assertGreater(saved, 0)
updated = self.db.update_analysis_history_diagnostics(
query_id="query_diag_patch",
@@ -280,6 +353,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
).first()
if row is None:
self.fail("未找到保存的历史记录")
self.assertEqual(row.id, saved)
snapshot = json.loads(row.context_snapshot or "{}")
self.assertEqual(snapshot["enhanced_context"]["code"], "600519")
notification_run = snapshot["diagnostics"]["notification_runs"][-1]
@@ -299,12 +373,13 @@ class AnalysisHistoryTestCase(unittest.TestCase):
context_snapshot=None,
save_snapshot=False
)
self.assertEqual(saved, 1)
self.assertGreater(saved, 0)
with self.db.get_session() as session:
row = session.query(AnalysisHistory).filter(AnalysisHistory.query_id == "query_004").first()
if row is None:
self.fail("未找到保存的历史记录")
self.assertEqual(row.id, saved)
record_id = row.id
service = HistoryService(self.db)
@@ -336,7 +411,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
context_snapshot=context_snapshot,
save_snapshot=True,
)
self.assertEqual(saved, 1)
self.assertGreater(saved, 0)
service = HistoryService(self.db)
payload = service.get_history_list(stock_code="600519.SH", page=1, limit=5)
@@ -382,7 +457,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
context_snapshot=None,
save_snapshot=False,
)
self.assertEqual(saved, 1)
self.assertGreater(saved, 0)
service = HistoryService(self.db)
payload = service.get_history_list(stock_code="600519", page=1, limit=10)
@@ -397,6 +472,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
).first()
if row is None:
self.fail("未找到保存的历史记录")
self.assertEqual(row.id, saved)
self.assertEqual(row.operation_advice, "观望")
def test_market_review_history_can_be_filtered_without_stock_records(self) -> None:
@@ -411,7 +487,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
analysis_summary="大盘复盘摘要",
)
self.assertEqual(
self.assertGreater(
self.db.save_analysis_history(
result=stock_result,
query_id="query_stock_history",
@@ -420,9 +496,9 @@ class AnalysisHistoryTestCase(unittest.TestCase):
context_snapshot=None,
save_snapshot=False,
),
1,
0,
)
self.assertEqual(
self.assertGreater(
self.db.save_analysis_history(
result=market_result,
query_id="query_market_review_history",
@@ -437,7 +513,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
},
save_snapshot=True,
),
1,
0,
)
service = HistoryService(self.db)
@@ -466,7 +542,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
analysis_summary="大盘复盘摘要",
)
self.assertEqual(
self.assertGreater(
self.db.save_analysis_history(
result=stock_result,
query_id="query_stock_bar_stock",
@@ -475,9 +551,9 @@ class AnalysisHistoryTestCase(unittest.TestCase):
context_snapshot=None,
save_snapshot=False,
),
1,
0,
)
self.assertEqual(
self.assertGreater(
self.db.save_analysis_history(
result=market_result,
query_id="query_stock_bar_market",
@@ -486,7 +562,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
context_snapshot=None,
save_snapshot=False,
),
1,
0,
)
records = self.db.get_distinct_stocks_from_history(limit=10)
@@ -508,7 +584,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
context_snapshot=None,
save_snapshot=False,
)
self.assertEqual(saved, 1)
self.assertGreater(saved, 0)
response = get_stock_bar(
start_date=None,
@@ -571,7 +647,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
context_snapshot=None,
save_snapshot=False,
)
self.assertEqual(saved, 1)
self.assertGreater(saved, 0)
save_record("600519.SH", "query_cn_suffix")
save_record("600519", "query_cn_plain")
@@ -624,7 +700,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
context_snapshot=None,
save_snapshot=False,
)
self.assertEqual(saved, 1)
self.assertGreater(saved, 0)
save_record("1810.HK", "query_hk_unpadded")
save_record("01810.HK", "query_hk_padded")
@@ -660,7 +736,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
context_snapshot=None,
save_snapshot=False,
)
self.assertEqual(saved, 1)
self.assertGreater(saved, 0)
save_record("600519.SH", "query_cn_sh")
save_record("600519.SS", "query_cn_ss")
@@ -704,12 +780,13 @@ class AnalysisHistoryTestCase(unittest.TestCase):
context_snapshot=context_snapshot,
save_snapshot=True,
)
self.assertEqual(saved, 1)
self.assertGreater(saved, 0)
with self.db.get_session() as session:
row = session.query(AnalysisHistory).filter(AnalysisHistory.query_id == query_id).first()
if row is None:
self.fail("未找到保存的历史记录")
self.assertEqual(row.id, saved)
record_id = row.id
report = get_history_detail(str(record_id), db_manager=self.db)
@@ -741,12 +818,13 @@ class AnalysisHistoryTestCase(unittest.TestCase):
context_snapshot=context_snapshot,
save_snapshot=True,
)
self.assertEqual(saved, 1)
self.assertGreater(saved, 0)
with self.db.get_session() as session:
row = session.query(AnalysisHistory).filter(AnalysisHistory.query_id == query_id).first()
if row is None:
self.fail("未找到保存的历史记录")
self.assertEqual(row.id, saved)
record_id = row.id
report = get_history_detail(str(record_id), db_manager=self.db)
@@ -774,12 +852,13 @@ class AnalysisHistoryTestCase(unittest.TestCase):
context_snapshot=context_snapshot,
save_snapshot=True,
)
self.assertEqual(saved, 1)
self.assertGreater(saved, 0)
with self.db.get_session() as session:
row = session.query(AnalysisHistory).filter(AnalysisHistory.query_id == query_id).first()
if row is None:
self.fail("未找到保存的历史记录")
self.assertEqual(row.id, saved)
record_id = row.id
static_dir = Path(self._temp_dir.name) / "empty-static"
@@ -805,12 +884,13 @@ class AnalysisHistoryTestCase(unittest.TestCase):
context_snapshot=None,
save_snapshot=False
)
self.assertEqual(saved, 1)
self.assertGreater(saved, 0)
with self.db.get_session() as session:
row = session.query(AnalysisHistory).filter(AnalysisHistory.query_id == "query_005").first()
if row is None:
self.fail("未找到保存的历史记录")
self.assertEqual(row.id, saved)
row.raw_result = {"model_used": "unknown", "extra": "v"}
service = HistoryService(self.db)
@@ -842,12 +922,13 @@ class AnalysisHistoryTestCase(unittest.TestCase):
context_snapshot=None,
save_snapshot=False
)
self.assertEqual(saved, 1)
self.assertGreater(saved, 0)
with self.db.get_session() as session:
row = session.query(AnalysisHistory).filter(AnalysisHistory.query_id == "query_006").first()
if row is None:
self.fail("未找到保存的历史记录")
self.assertEqual(row.id, saved)
record_id = row.id
service = HistoryService(self.db)
@@ -869,7 +950,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
context_snapshot=None,
save_snapshot=False
)
self.assertEqual(saved, 1)
self.assertGreater(saved, 0)
with self.db.get_session() as session:
row = session.query(AnalysisHistory).filter(AnalysisHistory.query_id == "query_007").first()
@@ -881,6 +962,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
row.take_profit = 150.0
row.raw_result = json.dumps({"model_used": "gemini/gemini-2.0-flash"})
session.commit()
self.assertEqual(row.id, saved)
record_id = row.id
service = HistoryService(self.db)
@@ -906,7 +988,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
context_snapshot=None,
save_snapshot=False,
)
self.assertEqual(saved, 1)
self.assertGreater(saved, 0)
self.db.save_fundamental_snapshot(
query_id=query_id,
@@ -932,6 +1014,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
row = session.query(AnalysisHistory).filter(AnalysisHistory.query_id == query_id).first()
if row is None:
self.fail("未找到保存的历史记录")
self.assertEqual(row.id, saved)
record_id = row.id
report = get_history_detail(str(record_id), db_manager=self.db)
@@ -954,7 +1037,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
context_snapshot=None,
save_snapshot=False,
)
self.assertEqual(saved, 1)
self.assertGreater(saved, 0)
fallback_fundamental = {
"belong_boards": [{"name": "白酒", "type": "行业"}],
@@ -968,12 +1051,13 @@ class AnalysisHistoryTestCase(unittest.TestCase):
code="600519",
payload=fallback_fundamental,
)
self.assertEqual(saved_snapshot, 1)
self.assertGreater(saved_snapshot, 0)
with self.db.get_session() as session:
row = session.query(AnalysisHistory).filter(AnalysisHistory.query_id == query_id).first()
if row is None:
self.fail("未找到保存的历史记录")
self.assertEqual(row.id, saved)
record_id = row.id
report = get_history_detail(str(record_id), db_manager=self.db)
@@ -994,12 +1078,13 @@ class AnalysisHistoryTestCase(unittest.TestCase):
context_snapshot=None,
save_snapshot=False,
)
self.assertEqual(saved, 1)
self.assertGreater(saved, 0)
with self.db.get_session() as session:
row = session.query(AnalysisHistory).filter(AnalysisHistory.query_id == query_id).first()
if row is None:
self.fail("未找到保存的历史记录")
self.assertEqual(row.id, saved)
record_id = row.id
report = get_history_detail(str(record_id), db_manager=self.db)
@@ -1029,12 +1114,13 @@ class AnalysisHistoryTestCase(unittest.TestCase):
context_snapshot=None,
save_snapshot=False,
)
self.assertEqual(saved, 1)
self.assertGreater(saved, 0)
with self.db.get_session() as session:
row = session.query(AnalysisHistory).filter(AnalysisHistory.query_id == query_id).first()
if row is None:
self.fail("未找到保存的历史记录")
self.assertEqual(row.id, saved)
record_id = row.id
report = get_history_detail(str(record_id), db_manager=self.db)
@@ -1070,12 +1156,13 @@ class AnalysisHistoryTestCase(unittest.TestCase):
context_snapshot=context_snapshot,
save_snapshot=True,
)
self.assertEqual(saved, 1)
self.assertGreater(saved, 0)
with self.db.get_session() as session:
row = session.query(AnalysisHistory).filter(AnalysisHistory.query_id == query_id).first()
if row is None:
self.fail("未找到保存的历史记录")
self.assertEqual(row.id, saved)
record_id = row.id
report = get_history_detail(str(record_id), db_manager=self.db)
@@ -1107,12 +1194,13 @@ class AnalysisHistoryTestCase(unittest.TestCase):
},
save_snapshot=True,
)
self.assertEqual(saved, 1)
self.assertGreater(saved, 0)
with self.db.get_session() as session:
row = session.query(AnalysisHistory).filter(AnalysisHistory.query_id == query_id).first()
if row is None:
self.fail("未找到保存的历史记录")
self.assertEqual(row.id, saved)
record_id = row.id
report = get_history_detail(str(record_id), db_manager=self.db)
@@ -1154,12 +1242,13 @@ class AnalysisHistoryTestCase(unittest.TestCase):
},
save_snapshot=False,
)
self.assertEqual(saved, 1)
self.assertGreater(saved, 0)
with self.db.get_session() as session:
row = session.query(AnalysisHistory).filter(AnalysisHistory.query_id == query_id).first()
if row is None:
self.fail("未找到保存的历史记录")
self.assertEqual(row.id, saved)
record_id = row.id
self.assertIsNone(row.context_snapshot)
@@ -1207,7 +1296,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
context_snapshot=None,
save_snapshot=False,
)
self.assertEqual(saved, 1)
self.assertGreater(saved, 0)
with self.db.get_session() as session:
row = session.query(AnalysisHistory).filter(
@@ -1215,6 +1304,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
).first()
if row is None:
self.fail("未找到保存的历史记录")
self.assertEqual(row.id, saved)
record_id = row.id
markdown = HistoryService(self.db).get_markdown_report(str(record_id))
@@ -1245,7 +1335,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
context_snapshot=None,
save_snapshot=False,
)
self.assertEqual(saved, 1)
self.assertGreater(saved, 0)
with self.db.get_session() as session:
row = session.query(AnalysisHistory).filter(
@@ -1253,6 +1343,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
).first()
if row is None:
self.fail("未找到保存的历史记录")
self.assertEqual(row.id, saved)
record_id = row.id
markdown = HistoryService(self.db).get_markdown_report(str(record_id))
@@ -1287,7 +1378,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
context_snapshot=None,
save_snapshot=False,
)
self.assertEqual(saved, 1)
self.assertGreater(saved, 0)
with self.db.get_session() as session:
row = session.query(AnalysisHistory).filter(
@@ -1295,6 +1386,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
).first()
if row is None:
self.fail("未找到保存的历史记录")
self.assertEqual(row.id, saved)
record_id = row.id
markdown = HistoryService(self.db).get_markdown_report(str(record_id))
@@ -1327,7 +1419,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
context_snapshot=None,
save_snapshot=False,
)
self.assertEqual(saved, 1)
self.assertGreater(saved, 0)
with self.db.get_session() as session:
row = session.query(AnalysisHistory).filter(
@@ -1335,6 +1427,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
).first()
if row is None:
self.fail("未找到保存的历史记录")
self.assertEqual(row.id, saved)
record_id = row.id
report = get_history_detail(str(record_id), db_manager=self.db)
@@ -1368,7 +1461,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
context_snapshot=None,
save_snapshot=False,
)
self.assertEqual(saved, 1)
self.assertGreater(saved, 0)
with self.db.get_session() as session:
row = session.query(AnalysisHistory).filter(
@@ -1376,6 +1469,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
).first()
if row is None:
self.fail("未找到保存的历史记录")
self.assertEqual(row.id, saved)
record_id = row.id
report = get_history_detail(str(record_id), db_manager=self.db)
@@ -1422,7 +1516,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
context_snapshot=None,
save_snapshot=False,
)
self.assertEqual(saved, 1)
self.assertGreater(saved, 0)
with self.db.get_session() as session:
row = session.query(AnalysisHistory).filter(
@@ -1430,6 +1524,7 @@ class AnalysisHistoryTestCase(unittest.TestCase):
).first()
if row is None:
self.fail("未找到保存的历史记录")
self.assertEqual(row.id, saved)
record_id = row.id
markdown = HistoryService(self.db).get_markdown_report(str(record_id))

295
tests/test_decision_signal_extractor.py Normal file
View File

@@ -0,0 +1,295 @@
# -*- coding: utf-8 -*-
"""Tests for extracting DecisionSignal assets from completed reports."""
from __future__ import annotations
import os
import pytest
from src.analyzer import AnalysisResult
from src.config import Config
from src.services.decision_signal_extractor import (
build_decision_signal_payload_from_report,
extract_and_persist_from_analysis_result,
)
from src.services.decision_signal_service import DecisionSignalService
from src.storage import DatabaseManager
@pytest.fixture()
def isolated_db(tmp_path):
old_database_path = os.environ.get("DATABASE_PATH")
db_path = tmp_path / "decision_signal_extractor.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 _result(**overrides) -> AnalysisResult:
result = AnalysisResult(
code="600519",
name="贵州茅台",
sentiment_score=82,
trend_prediction="看多",
operation_advice="买入",
decision_type="buy",
confidence_level="高",
analysis_summary="趋势确认，量价配合。",
risk_warning="跌破支撑需止损",
report_language="zh",
)
result.dashboard = {
"battle_plan": {
"sniper_points": {
"ideal_buy": "理想买入点：1700元",
"secondary_buy": "1680-1690（回踩MA5附近）",
"stop_loss": "止损位：1600元",
"take_profit": "目标位：1850元",
},
"action_checklist": ["放量突破前高", "回踩不破MA10"],
},
"phase_decision": {
"watch_conditions": ["盘中量能继续放大"],
},
"intelligence": {
"risk_alerts": ["估值偏高"],
"positive_catalysts": ["业绩超预期"],
},
}
for key, value in overrides.items():
setattr(result, key, value)
return result
def test_build_payload_maps_report_context_and_price_plan() -> None:
result = _result()
result.market_phase_summary = {"phase": "postmarket"}
result.analysis_context_pack_overview = {"data_quality": {"overall_score": 55, "level": "fair"}}
context_snapshot = {
"market_phase_summary": {"phase": "intraday"},
"analysis_context_pack_overview": {
"data_quality": {"overall_score": 91, "level": "good"},
},
}
payload = build_decision_signal_payload_from_report(
result,
context_snapshot=context_snapshot,
source_report_id=88,
trace_id="trace-88",
query_source="api",
report_type="full",
)
assert payload is not None
assert payload["stock_code"] == "600519"
assert payload["stock_name"] == "贵州茅台"
assert payload["market"] == "cn"
assert payload["source_type"] == "analysis"
assert payload["source_report_id"] == 88
assert payload["trace_id"] == "trace-88"
assert payload["trigger_source"] == "api"
assert payload["action"] == "buy"
assert payload["confidence"] == 0.8
assert payload["score"] == 82
assert payload["market_phase"] == "intraday"
assert payload["entry_low"] == 1690.0
assert payload["entry_high"] == 1700.0
assert payload["stop_loss"] == 1600.0
assert payload["target_price"] == 1850.0
assert payload["data_quality_summary"]["overall_score"] == 91
assert payload["watch_conditions"] == ["盘中量能继续放大"]
assert payload["risk_summary"] == ["跌破支撑需止损", "估值偏高"]
assert payload["catalyst_summary"] == ["业绩超预期"]
assert payload["metadata"]["report_confidence_level"] == "高"
def test_build_payload_uses_result_fallbacks_and_optional_catalysts() -> None:
result = _result(confidence_level="低")
result.dashboard = {
"battle_plan": {
"sniper_points": {"ideal_buy": "1700"},
"action_checklist": ["等待回踩确认"],
},
"intelligence": {},
}
result.market_phase_summary = {"phase": "postmarket"}
result.analysis_context_pack_overview = {"data_quality": {"level": "limited"}}
payload = build_decision_signal_payload_from_report(
result,
context_snapshot=None,
source_report_id=None,
trace_id="trace-fallback",
query_source="",
report_type="simple",
)
assert payload is not None
assert payload["market_phase"] == "postmarket"
assert payload["data_quality_summary"] == {"level": "limited"}
assert payload["entry_low"] == 1700.0
assert "entry_high" not in payload
assert payload["watch_conditions"] == ["等待回踩确认"]
assert "catalyst_summary" not in payload
assert payload["trigger_source"] == "system"
assert payload["confidence"] == 0.4
def test_build_payload_maps_secondary_only_entry_to_entry_high() -> None:
result = _result()
result.dashboard = {
"battle_plan": {
"sniper_points": {"secondary_buy": "次优买入点：1680元"},
},
}
payload = build_decision_signal_payload_from_report(
result,
trace_id="trace-secondary-only",
query_source="api",
report_type="simple",
)
assert payload is not None
assert "entry_low" not in payload
assert payload["entry_high"] == 1680.0
def test_build_payload_reuses_shared_sniper_fallback_paths(isolated_db) -> None:
result = _result()
result.dashboard = {}
result.raw_response = {
"dashboard": {
"battle_plan": {
"sniper_points": {
"ideal_buy": "1690",
"secondary_buy": "1705",
"stop_loss": "1620",
"take_profit": "1880",
}
}
}
}
payload = build_decision_signal_payload_from_report(
result,
trace_id="trace-raw-sniper",
query_source="api",
report_type="simple",
)
stored_points = isolated_db._extract_sniper_points(result)
assert payload is not None
assert stored_points == {
"ideal_buy": 1690.0,
"secondary_buy": 1705.0,
"stop_loss": 1620.0,
"take_profit": 1880.0,
}
assert payload["entry_low"] == 1690.0
assert payload["entry_high"] == 1705.0
assert payload["stop_loss"] == 1620.0
assert payload["target_price"] == 1880.0
def test_build_payload_skips_ambiguous_action_non_stock_and_unknown_market() -> None:
ambiguous = _result(operation_advice="买盘增强，继续观察", action=None)
assert build_decision_signal_payload_from_report(
ambiguous,
trace_id="trace-1",
query_source="api",
report_type="simple",
) is None
market_review = _result(operation_advice="买入", action="buy")
assert build_decision_signal_payload_from_report(
market_review,
trace_id="trace-2",
query_source="api",
report_type="market_review",
) is None
unknown_market = _result(code="UNKNOWN", operation_advice="买入", action="buy")
assert build_decision_signal_payload_from_report(
unknown_market,
trace_id="trace-3",
query_source="api",
report_type="simple",
) is None
def test_extract_and_persist_reuses_service_dedup_and_sanitization(isolated_db) -> None:
service = DecisionSignalService(db_manager=isolated_db)
result = _result(
analysis_summary="趋势确认 token=super-secret",
)
first = extract_and_persist_from_analysis_result(
result,
context_snapshot={"market_phase_summary": {"phase": "intraday"}},
source_report_id=901,
trace_id="trace-901",
query_source="api",
report_type="full",
service=service,
)
second = extract_and_persist_from_analysis_result(
result,
context_snapshot={"market_phase_summary": {"phase": "intraday"}},
source_report_id=901,
trace_id="trace-901",
query_source="api",
report_type="full",
service=service,
)
assert first is not None
assert second is not None
assert first["created"] is True
assert second["created"] is False
assert first["item"]["reason"] == "趋势确认 token=[REDACTED]"
assert first["item"]["plan_quality"] == "complete"
listed = service.list_signals(source_report_id=901)
assert listed["total"] == 1
persisted = listed["items"][0]
assert persisted["source_report_id"] == 901
assert persisted["reason"] == "趋势确认 token=[REDACTED]"
assert persisted["entry_low"] == 1690.0
assert persisted["entry_high"] == 1700.0
def test_extract_and_persist_missing_price_plan_does_not_fabricate_fields(isolated_db) -> None:
service = DecisionSignalService(db_manager=isolated_db)
result = _result()
result.dashboard = {"battle_plan": {"sniper_points": {}}, "intelligence": {}}
created = extract_and_persist_from_analysis_result(
result,
context_snapshot={"market_phase_summary": {"phase": "postmarket"}},
source_report_id=902,
trace_id="trace-902",
query_source="schedule",
report_type="simple",
service=service,
)
assert created is not None
item = created["item"]
assert item["plan_quality"] == "minimal"
assert item["entry_low"] is None
assert item["entry_high"] is None
assert item["stop_loss"] is None
assert item["target_price"] is None

12
tests/test_market_review.py
View File

@@ -430,20 +430,14 @@ class MarketReviewLocalizationTestCase(unittest.TestCase):
},
)
self.assertEqual(saved, 1)
with DatabaseManager.get_instance().get_session() as session:
row = session.query(AnalysisHistory).filter(
AnalysisHistory.query_id == "market-task-001"
).first()
self.assertIsNotNone(row)
snapshot = json.loads(row.context_snapshot or "{}")
self.assertIn("analysis_context_pack_overview", snapshot)
self.assertGreater(saved, 0)
db = DatabaseManager.get_instance()
with db.get_session() as session:
row = session.query(AnalysisHistory).filter(
AnalysisHistory.query_id == "market-task-001"
).first()
self.assertIsNotNone(row)
self.assertEqual(row.id, saved)
self.assertEqual(row.code, market_review_module.MARKET_REVIEW_HISTORY_CODE)
self.assertEqual(row.name, "大盘复盘")
self.assertEqual(row.report_type, market_review_module.MARKET_REVIEW_REPORT_TYPE)
@@ -452,6 +446,8 @@ class MarketReviewLocalizationTestCase(unittest.TestCase):
self.assertIn('"market_light_snapshots"', row.context_snapshot)
self.assertIn('"market_review_payload"', row.context_snapshot)
self.assertIn('"trade_date": "2026-03-06"', row.context_snapshot)
snapshot = json.loads(row.context_snapshot or "{}")
self.assertIn("analysis_context_pack_overview", snapshot)
finally:
DatabaseManager.reset_instance()
Config._instance = None

4
tests/test_news_intel.py
View File

@@ -183,7 +183,7 @@ class NewsIntelStorageTestCase(unittest.TestCase):
with patch.object(self.db, "get_session", side_effect=[first_session, second_session]):
with patch.object(first_session, "execute", side_effect=stmt_exc):
with patch("src.storage.time.sleep") as mock_sleep:
saved = self.db.save_news_intel(
saved_count = self.db.save_news_intel(
code="600519",
name="贵州茅台",
dimension="latest_news",
@@ -191,7 +191,7 @@ class NewsIntelStorageTestCase(unittest.TestCase):
response=response,
)
self.assertEqual(saved, 1)
self.assertEqual(saved_count, 1)
self.assertEqual(mock_sleep.call_count, 1)
self.assertAlmostEqual(mock_sleep.call_args.args[0], self.db._sqlite_write_retry_base_delay, places=6)

158
tests/test_pipeline_market_phase_context.py
View File

@@ -721,6 +721,164 @@ class PipelineMarketPhaseContextTestCase(unittest.TestCase):
finally:
reset_run_diagnostic_context(token)
def test_decision_signal_helper_uses_saved_history_id(self):
pipeline = _make_pipeline(agent_mode=False, save_context_snapshot=True)
pipeline.trace_id = "trace-helper"
pipeline.query_source = "api"
result = _analysis_result()
context_snapshot = {"market_phase_summary": _phase_payload()}
with patch("src.core.pipeline.extract_and_persist_from_analysis_result") as mock_extract:
pipeline._extract_decision_signal_after_history_save(
result=result,
query_id="q-helper",
source_report_id=42,
report_type=ReportType.SIMPLE.value,
context_snapshot=context_snapshot,
)
pipeline.db.get_latest_analysis_history_id.assert_not_called()
mock_extract.assert_called_once()
self.assertIs(mock_extract.call_args.args[0], result)
kwargs = mock_extract.call_args.kwargs
self.assertIs(kwargs["context_snapshot"], context_snapshot)
self.assertEqual(kwargs["source_report_id"], 42)
self.assertEqual(kwargs["trace_id"], "trace-helper")
self.assertEqual(kwargs["query_source"], "api")
self.assertEqual(kwargs["report_type"], ReportType.SIMPLE.value)
def test_decision_signal_helper_failure_does_not_raise(self):
pipeline = _make_pipeline(agent_mode=False, save_context_snapshot=True)
with patch(
"src.core.pipeline.extract_and_persist_from_analysis_result",
side_effect=RuntimeError("boom"),
):
pipeline._extract_decision_signal_after_history_save(
result=_analysis_result(),
query_id="q-helper-fail",
source_report_id=42,
report_type=ReportType.SIMPLE.value,
context_snapshot={"market_phase_summary": _phase_payload()},
)
def test_legacy_pipeline_extracts_decision_signal_with_saved_history_id(self):
pipeline = _make_pipeline(agent_mode=False, save_context_snapshot=True)
pipeline.trace_id = "trace-runtime"
pipeline.query_source = "api"
pipeline.db.save_analysis_history.return_value = 42
phase_context = SimpleNamespace(to_dict=MagicMock(return_value=_phase_payload()))
with (
patch("src.core.pipeline.build_market_phase_context", return_value=phase_context),
patch("src.core.pipeline.extract_and_persist_from_analysis_result") as mock_extract,
):
result = pipeline.analyze_stock(
"600519",
ReportType.SIMPLE,
"q-runtime-signal",
current_time=datetime(2026, 3, 27, 10, 0),
)
self.assertIsNotNone(result)
mock_extract.assert_called_once()
kwargs = mock_extract.call_args.kwargs
self.assertEqual(kwargs["source_report_id"], 42)
self.assertEqual(kwargs["trace_id"], "trace-runtime")
self.assertEqual(kwargs["query_source"], "api")
self.assertEqual(kwargs["report_type"], ReportType.SIMPLE.value)
def test_legacy_pipeline_does_not_extract_when_history_save_fails(self):
pipeline = _make_pipeline(agent_mode=False, save_context_snapshot=True)
pipeline.db.save_analysis_history.return_value = 0
phase_context = SimpleNamespace(to_dict=MagicMock(return_value=_phase_payload()))
with (
patch("src.core.pipeline.build_market_phase_context", return_value=phase_context),
patch("src.core.pipeline.extract_and_persist_from_analysis_result") as mock_extract,
):
result = pipeline.analyze_stock(
"600519",
ReportType.SIMPLE,
"q-runtime-no-signal",
current_time=datetime(2026, 3, 27, 10, 0),
)
self.assertIsNotNone(result)
mock_extract.assert_not_called()
def test_legacy_pipeline_extract_failure_does_not_mark_history_save_failed(self):
pipeline = _make_pipeline(agent_mode=False, save_context_snapshot=True)
pipeline.db.save_analysis_history.return_value = 42
phase_context = SimpleNamespace(to_dict=MagicMock(return_value=_phase_payload()))
with (
patch("src.core.pipeline.build_market_phase_context", return_value=phase_context),
patch(
"src.core.pipeline.extract_and_persist_from_analysis_result",
side_effect=RuntimeError("boom"),
) as mock_extract,
patch("src.core.pipeline.record_history_run") as mock_record,
):
result = pipeline.analyze_stock(
"600519",
ReportType.SIMPLE,
"q-runtime-extract-fail",
current_time=datetime(2026, 3, 27, 10, 0),
)
self.assertIsNotNone(result)
mock_extract.assert_called_once()
self.assertEqual(mock_record.call_args.kwargs["analysis_history_id"], 42)
self.assertTrue(mock_record.call_args.kwargs["report_saved"])
def test_agent_pipeline_extracts_decision_signal_with_saved_history_id(self):
pipeline = _make_pipeline(agent_mode=True, save_context_snapshot=True)
pipeline.db.save_analysis_history.return_value = 84
pipeline._ensure_agent_history = MagicMock()
phase_payload = _phase_payload()
from src.agent.executor import AgentResult
agent_result = AgentResult(
success=True,
content="{}",
dashboard={
"stock_name": "贵州茅台",
"sentiment_score": 66,
"trend_prediction": "震荡",
"operation_advice": "持有",
"decision_type": "hold",
},
provider="test",
)
executor = MagicMock()
executor.run.return_value = agent_result
with (
patch("src.agent.factory.build_agent_executor", return_value=executor),
patch("src.core.pipeline.extract_and_persist_from_analysis_result") as mock_extract,
):
result = pipeline._analyze_with_agent(
code="600519",
report_type=ReportType.SIMPLE,
query_id="q-agent-signal",
stock_name="贵州茅台",
realtime_quote=None,
chip_data=None,
fundamental_context={"market": "cn"},
trend_result=None,
market_phase_context=phase_payload,
market_phase_summary=phase_payload,
)
self.assertIsNotNone(result)
mock_extract.assert_called_once()
kwargs = mock_extract.call_args.kwargs
self.assertEqual(kwargs["source_report_id"], 84)
self.assertEqual(kwargs["report_type"], ReportType.SIMPLE.value)
self.assertIs(kwargs["context_snapshot"], pipeline.db.save_analysis_history.call_args.kwargs["context_snapshot"])
if __name__ == "__main__":
unittest.main()

4
tests/test_run_flow.py
View File

@@ -1071,10 +1071,10 @@ class RunFlowTestCase(unittest.TestCase):
notification = next(node for node in snapshot.nodes if node.id.startswith("notification_report"))
self.assertEqual(notification.attempts, 0)
def test_market_review_persist_records_diagnostics_without_bool_history_id(self) -> None:
def test_market_review_persist_records_diagnostics_with_saved_history_id(self) -> None:
from src.core.market_review import _persist_market_review_history
fake_db = _FakeMarketReviewDb(save_result=True)
fake_db = _FakeMarketReviewDb(save_result=42)
config = SimpleNamespace(report_language="zh")
token = activate_run_diagnostic_context(
trace_id="trace-market",