mirror of
https://github.com/ZhuLinsen/daily_stock_analysis
synced 2026-09-20 10:53:33 +08:00
972c314656
* feat: Web/API 指数入口与共享 canonical 去重基础 为 Issue #2303 Phase 2 PR1 落地 Web/API 指数入口适配： - API 使用 parse_analysis_target 构造结构化 AnalysisTarget 并贯通到 pipeline - API 与 TaskQueue 去重按 asset_type 分支（指数用 canonical_id，个股用 legacy code） - BatchTaskAcceptedResponse 追加可选 rejected 字段，未登记 CSI 单股 400、批量仅该目标失败 - TaskInfo 固化 dedupe_key，避免指数与同码个股折叠及 _analyzing_stocks 残留 - Web 移除 assetType=index 全局过滤，Chat 名称识别保护指数 canonical - 补齐 Pipeline 指数 DecisionSignal market_override=cn 真实分支测试 * fix: 收敛指数 canonical 身份与批量响应契约 PR #2312 review 修复：报告 meta 补充 asset_type 隐藏指数自选；/analyze 在解析前限制非空原始 token；is_single 统一驱动 metadata/409/单任务 202；HomePage 三元计数继续后续 chunk。 * fix: avoid double space in index news search query * fix: preserve canonical index identity * fix: validate legacy task asset type * fix: preserve canonical index identity in chat * fix: preserve index identity across chat backends
1619 lines
67 KiB
Python
1619 lines
67 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""
|
||
===================================
|
||
History Query Service Layer
|
||
===================================
|
||
|
||
Responsibilities:
|
||
1. Encapsulate history record query logic
|
||
2. Provide pagination and filtering functionality
|
||
3. Generate detailed reports in Markdown format
|
||
"""
|
||
from __future__ import annotations
|
||
import json
|
||
import logging
|
||
import re
|
||
from datetime import date, datetime, timedelta
|
||
from typing import Optional, Dict, Any, List, Tuple, TYPE_CHECKING
|
||
|
||
from src.config import get_config, resolve_news_window_days
|
||
from src.formatters import markdown_to_plain_text
|
||
from src.data.stock_index_loader import resolve_index_stock_code
|
||
from src.services.stock_code_utils import resolve_daily_stock_identity
|
||
from src.report_language import (
|
||
get_bias_status_emoji,
|
||
get_localized_stock_name,
|
||
get_report_labels,
|
||
get_signal_level,
|
||
get_chip_unavailable_reason,
|
||
is_chip_structure_unavailable,
|
||
localize_bias_status,
|
||
localize_chip_health,
|
||
localize_conflict_severity,
|
||
localize_consensus_level,
|
||
localize_strategy_signal,
|
||
localize_strategy_skill,
|
||
localize_strategy_synthesis_summary,
|
||
localize_trend_prediction,
|
||
normalize_report_language,
|
||
normalize_strategy_synthesis_payload,
|
||
strategy_invalid_opinion_count,
|
||
)
|
||
from src.storage import DatabaseManager
|
||
from src.services.run_diagnostics import build_run_diagnostic_summary
|
||
from src.services.empty_news import (
|
||
empty_news_disclosure,
|
||
empty_news_disclosure_from_stored,
|
||
persisted_news_evidence_present,
|
||
persisted_news_result_state,
|
||
)
|
||
from src.market_phase_summary import (
|
||
extract_market_phase_summary,
|
||
rebuild_market_phase_summary_for_stock_code,
|
||
)
|
||
from src.schemas.decision_action import (
|
||
display_action_fields,
|
||
display_action_fields_for_result,
|
||
display_operation_advice_for_result,
|
||
)
|
||
from src.schemas.decision_scale import extract_decision_guardrail_reason
|
||
from src.utils.sniper_points import find_sniper_points
|
||
from src.utils.data_processing import (
|
||
extract_realtime_detail_fields,
|
||
normalize_model_used,
|
||
parse_json_field,
|
||
signal_attribution_has_content,
|
||
signal_attribution_weight_items,
|
||
)
|
||
|
||
if TYPE_CHECKING:
|
||
from src.analyzer import AnalysisResult
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class MarkdownReportGenerationError(Exception):
|
||
"""Exception raised when Markdown report generation fails due to internal errors."""
|
||
|
||
def __init__(self, message: str, record_id: str = None):
|
||
self.message = message
|
||
self.record_id = record_id
|
||
super().__init__(self.message)
|
||
|
||
|
||
class HistoryService:
|
||
"""
|
||
History Query Service
|
||
|
||
Encapsulates query logic for historical analysis records.
|
||
"""
|
||
|
||
def __init__(self, db_manager: Optional[DatabaseManager] = None):
|
||
"""
|
||
Initialize the history query service.
|
||
|
||
Args:
|
||
db_manager: Database manager (optional, defaults to singleton instance)
|
||
"""
|
||
self.db = db_manager or DatabaseManager.get_instance()
|
||
|
||
@staticmethod
|
||
def _serialize_created_at(value: Optional[datetime]) -> Optional[str]:
|
||
"""Serialize stored server-local timestamps with an explicit offset."""
|
||
if value is None:
|
||
return None
|
||
return value.astimezone().isoformat()
|
||
|
||
@staticmethod
|
||
def _history_code_filter_candidates(
|
||
stock_code: str,
|
||
*,
|
||
market_hint: Optional[str] = None,
|
||
) -> List[str]:
|
||
raw_code = str(stock_code or "").strip()
|
||
if not raw_code:
|
||
return []
|
||
|
||
candidates: List[str] = []
|
||
|
||
def add(candidate: str) -> None:
|
||
candidate = str(candidate or "").strip().upper()
|
||
if candidate and candidate not in candidates:
|
||
candidates.append(candidate)
|
||
|
||
trusted_market = str(market_hint or "").strip().lower()
|
||
if trusted_market:
|
||
identity = resolve_daily_stock_identity(raw_code, market_hint=trusted_market)
|
||
if identity is None or identity.market != trusted_market:
|
||
return []
|
||
for candidate in identity.code_candidates:
|
||
candidate_text = str(candidate or "").strip()
|
||
if candidate_text.isdigit():
|
||
unhinted_identity = resolve_daily_stock_identity(candidate_text)
|
||
if unhinted_identity is None or unhinted_identity.market != trusted_market:
|
||
continue
|
||
add(candidate)
|
||
return candidates
|
||
|
||
try:
|
||
from data_provider.base import (
|
||
canonical_stock_code,
|
||
is_bse_code,
|
||
normalize_stock_code,
|
||
)
|
||
from src.services.stock_list_parser import ParseStatus, parse_analysis_target
|
||
|
||
# PR #2312: parser-aware index branch. ``parse_analysis_target``
|
||
# (with the index registry) is the single asset-type authority. When
|
||
# the query token is a registered INDEX, the persisted-read candidate
|
||
# set must be exactly:
|
||
# 1. the lowercase parser canonical (``sh000016`` / ``csi930955`` —
|
||
# the current storage form),
|
||
# 2. the legacy uppercase canonical (``SH000016`` / ``CSI930955`` —
|
||
# how pre-fix records were saved), and
|
||
# 3. every explicit alias/display form (``000016.SH`` /
|
||
# ``930955.CSI`` — ``IndexEntry.aliases``).
|
||
# The bare same-code stock (``000016`` / ``930955``) is deliberately
|
||
# excluded so an index record is never reachable through a stock
|
||
# query and vice versa. This unifies the previous CSI-only convergence
|
||
# (PR #2267) with the SH/SZ index identity so all three namespaces
|
||
# share one branch.
|
||
target = parse_analysis_target(raw_code)
|
||
if target.asset_type == ParseStatus.INDEX:
|
||
index_keys = [target.canonical_id]
|
||
legacy_upper = (
|
||
canonical_stock_code(target.canonical_id)
|
||
or target.canonical_id.upper()
|
||
)
|
||
if legacy_upper and legacy_upper not in index_keys:
|
||
index_keys.append(legacy_upper)
|
||
entry = target.matched_index
|
||
if entry is not None:
|
||
for alias in entry.aliases:
|
||
alias = str(alias or "").strip()
|
||
if not alias:
|
||
continue
|
||
if alias not in index_keys:
|
||
index_keys.append(alias)
|
||
# sqlite ``IN`` comparison is case-sensitive: a legacy
|
||
# record may have been persisted under the uppercase
|
||
# alias form (``SZ399300`` for registry alias
|
||
# ``sz399300``), so the persisted-read candidate set must
|
||
# carry both the raw alias and its uppercase form. The
|
||
# bare same-code stock (``000300``) is still never added.
|
||
alias_upper = alias.upper()
|
||
if alias_upper and alias_upper not in index_keys:
|
||
index_keys.append(alias_upper)
|
||
for key in index_keys:
|
||
if key and key not in candidates:
|
||
candidates.append(key)
|
||
return candidates
|
||
|
||
raw_canonical = canonical_stock_code(raw_code)
|
||
normalized = canonical_stock_code(normalize_stock_code(raw_canonical))
|
||
except Exception:
|
||
add(raw_code)
|
||
return candidates
|
||
|
||
def add_hk_variants(digits: str) -> None:
|
||
if not digits or not digits.isdigit():
|
||
return
|
||
|
||
normalized_digits = digits.zfill(5)
|
||
add(f"HK{normalized_digits}")
|
||
add(f"{normalized_digits}.HK")
|
||
|
||
unpadded_digits = digits.lstrip("0")
|
||
if unpadded_digits:
|
||
add(f"{unpadded_digits}.HK")
|
||
|
||
resolved = resolve_index_stock_code(raw_canonical) or resolve_index_stock_code(normalized)
|
||
resolved_normalized = ""
|
||
if resolved:
|
||
try:
|
||
resolved_normalized = canonical_stock_code(normalize_stock_code(resolved))
|
||
except Exception:
|
||
resolved_normalized = resolved
|
||
add(resolved)
|
||
add(resolved_normalized)
|
||
resolved_base = str(resolved_normalized or resolved).split(".", 1)[0]
|
||
if resolved_base and resolved_base.isdigit():
|
||
add(resolved_base)
|
||
|
||
add(raw_canonical)
|
||
add(normalized)
|
||
|
||
if normalized.startswith("HK") and normalized[2:].isdigit():
|
||
add_hk_variants(normalized[2:])
|
||
elif normalized.isdigit() and len(normalized) == 5:
|
||
add_hk_variants(normalized)
|
||
elif normalized.isdigit() and len(normalized) == 6:
|
||
exchange = None
|
||
if is_bse_code(normalized):
|
||
exchange = "BJ"
|
||
elif normalized.startswith(("5", "6", "9")):
|
||
exchange = "SH"
|
||
elif normalized.startswith(("0", "1", "2", "3")):
|
||
exchange = "SZ"
|
||
|
||
if exchange:
|
||
add(f"{normalized}.{exchange}")
|
||
add(f"{exchange}{normalized}")
|
||
if exchange == "SH":
|
||
add(f"{normalized}.SS")
|
||
add(f"SS{normalized}")
|
||
|
||
return candidates
|
||
|
||
def get_history_list(
|
||
self,
|
||
stock_code: Optional[str] = None,
|
||
report_type: Optional[str] = None,
|
||
start_date: Optional[str] = None,
|
||
end_date: Optional[str] = None,
|
||
page: int = 1,
|
||
limit: int = 20,
|
||
include_ambiguous_numeric_aliases: bool = True,
|
||
market_hint: Optional[str] = None,
|
||
) -> Dict[str, Any]:
|
||
"""
|
||
Get history analysis list.
|
||
|
||
Args:
|
||
stock_code: Stock code filter
|
||
report_type: Report type filter
|
||
start_date: Start date (YYYY-MM-DD)
|
||
end_date: End date (YYYY-MM-DD)
|
||
page: Page number
|
||
limit: Items per page
|
||
include_ambiguous_numeric_aliases: Whether to include bare numeric
|
||
aliases that may collide across offshore markets.
|
||
market_hint: Trusted market identity for candidate expansion. When
|
||
present, candidates from other indexed markets are excluded.
|
||
|
||
Returns:
|
||
Dictionary containing total count and items
|
||
"""
|
||
try:
|
||
if stock_code:
|
||
stock_code = self._history_code_filter_candidates(
|
||
stock_code,
|
||
market_hint=market_hint,
|
||
)
|
||
if not include_ambiguous_numeric_aliases and not market_hint:
|
||
stock_code = [candidate for candidate in stock_code if not candidate.isdigit()]
|
||
if not stock_code:
|
||
return {"total": 0, "items": []}
|
||
|
||
# Parse date parameters
|
||
start_dt = None
|
||
end_dt = None
|
||
|
||
if start_date:
|
||
try:
|
||
start_dt = datetime.strptime(start_date, "%Y-%m-%d").date()
|
||
except ValueError:
|
||
logger.warning(f"无效的 start_date 格式: {start_date}")
|
||
|
||
if end_date:
|
||
try:
|
||
end_dt = datetime.strptime(end_date, "%Y-%m-%d").date()
|
||
except ValueError:
|
||
logger.warning(f"无效的 end_date 格式: {end_date}")
|
||
|
||
# Calculate offset
|
||
offset = (page - 1) * limit
|
||
|
||
# Use new paginated query method
|
||
records, total = self.db.get_analysis_history_paginated(
|
||
code=stock_code,
|
||
report_type=report_type,
|
||
start_date=start_dt,
|
||
end_date=end_dt,
|
||
offset=offset,
|
||
limit=limit
|
||
)
|
||
|
||
# Convert to response format
|
||
items = []
|
||
for record in records:
|
||
items.append(self._record_to_list_item_dict(record))
|
||
|
||
return {
|
||
"total": total,
|
||
"items": items,
|
||
}
|
||
|
||
except Exception as e:
|
||
logger.error(f"查询历史列表失败: {e}", exc_info=True)
|
||
raise
|
||
|
||
@staticmethod
|
||
def _safe_float(value: Any) -> Optional[float]:
|
||
if value is None:
|
||
return None
|
||
try:
|
||
if isinstance(value, str):
|
||
value = value.strip().replace("%", "")
|
||
if not value:
|
||
return None
|
||
parsed = float(value)
|
||
return parsed if parsed == parsed else None
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
@staticmethod
|
||
def _first_present(*values: Any) -> Any:
|
||
for value in values:
|
||
if value is not None and value != "":
|
||
return value
|
||
return None
|
||
|
||
def _extract_history_market_fields(self, context_snapshot: Any) -> Dict[str, Optional[float]]:
|
||
snapshot_obj = parse_json_field(context_snapshot)
|
||
realtime_fields = extract_realtime_detail_fields(snapshot_obj)
|
||
|
||
volume_ratio = None
|
||
turnover_rate = None
|
||
if isinstance(snapshot_obj, dict):
|
||
enhanced = snapshot_obj.get("enhanced_context")
|
||
realtime = enhanced.get("realtime") if isinstance(enhanced, dict) else None
|
||
quote_raw = snapshot_obj.get("realtime_quote_raw")
|
||
quote = snapshot_obj.get("realtime_quote")
|
||
for source in (realtime, quote_raw, quote):
|
||
if not isinstance(source, dict):
|
||
continue
|
||
if volume_ratio is None:
|
||
volume_ratio = self._first_present(
|
||
source.get("volume_ratio"),
|
||
source.get("volumeRatio"),
|
||
)
|
||
if turnover_rate is None:
|
||
turnover_rate = self._first_present(
|
||
source.get("turnover_rate"),
|
||
source.get("turnoverRate"),
|
||
source.get("turnover"),
|
||
)
|
||
|
||
return {
|
||
"current_price": self._safe_float(realtime_fields.get("current_price")),
|
||
"change_pct": self._safe_float(realtime_fields.get("change_pct")),
|
||
"volume_ratio": self._safe_float(volume_ratio),
|
||
"turnover_rate": self._safe_float(turnover_rate),
|
||
}
|
||
|
||
@staticmethod
|
||
def _display_stock_code(raw_code: Any) -> str:
|
||
code = str(raw_code or "").strip()
|
||
if not code:
|
||
return code
|
||
from src.services.stock_list_parser import ParseStatus, parse_analysis_target
|
||
|
||
try:
|
||
target = parse_analysis_target(code)
|
||
except Exception:
|
||
return resolve_index_stock_code(code) or code
|
||
if target.asset_type == ParseStatus.INDEX:
|
||
# PR #2312: registered index records always surface the parser
|
||
# canonical (lowercase ``sh000300`` / ``csi930955``), even when the
|
||
# persisted row was saved under a legacy uppercase canonical or an
|
||
# explicit alias (``SH000016`` / ``000016.SH`` / ``sz399300`` /
|
||
# ``000300.CSI``). Web identity keys therefore only need a simple
|
||
# case fold for API/task/report codes — they must never guess a
|
||
# canonical from prefixes/suffixes.
|
||
return target.canonical_id
|
||
return resolve_index_stock_code(code) or code
|
||
|
||
def _display_market_phase_summary(self, stock_code: str, context_snapshot: Any) -> Any:
|
||
return rebuild_market_phase_summary_for_stock_code(
|
||
self._display_stock_code(stock_code),
|
||
context_snapshot,
|
||
)
|
||
|
||
@staticmethod
|
||
def _extract_market_review_region(context_snapshot: Any) -> Optional[str]:
|
||
snapshot = parse_json_field(context_snapshot)
|
||
if not isinstance(snapshot, dict):
|
||
return None
|
||
|
||
region = snapshot.get("market_review_region")
|
||
if not isinstance(region, str):
|
||
payload = snapshot.get("market_review_payload")
|
||
region = payload.get("region") if isinstance(payload, dict) else None
|
||
|
||
normalized = region.strip() if isinstance(region, str) else ""
|
||
return normalized or None
|
||
|
||
def _record_to_list_item_dict(self, record) -> Dict[str, Any]:
|
||
raw_result = parse_json_field(getattr(record, "raw_result", None))
|
||
model_used = raw_result.get("model_used") if isinstance(raw_result, dict) else None
|
||
display_code = self._display_stock_code(record.code)
|
||
market_fields = self._extract_history_market_fields(
|
||
getattr(record, "context_snapshot", None)
|
||
)
|
||
market_phase_summary = self._display_market_phase_summary(
|
||
record.code,
|
||
getattr(record, "context_snapshot", None),
|
||
)
|
||
action_fields = self._decision_action_fields_for_record(record, raw_result)
|
||
analysis_summary = record.analysis_summary
|
||
if getattr(record, "report_type", None) == "market_review":
|
||
market_review_content = self._extract_market_review_content(record, raw_result)
|
||
analysis_summary = self._market_review_summary(
|
||
record.analysis_summary,
|
||
market_review_content,
|
||
)
|
||
|
||
return {
|
||
"id": record.id,
|
||
"query_id": record.query_id,
|
||
"stock_code": display_code,
|
||
"stock_name": record.name,
|
||
"report_type": record.report_type,
|
||
"region": self._extract_market_review_region(
|
||
getattr(record, "context_snapshot", None)
|
||
),
|
||
"trend_prediction": record.trend_prediction,
|
||
"analysis_summary": analysis_summary,
|
||
"sentiment_score": record.sentiment_score,
|
||
"operation_advice": record.operation_advice,
|
||
"action": action_fields["action"],
|
||
"action_label": action_fields["action_label"],
|
||
"model_used": normalize_model_used(model_used),
|
||
"created_at": self._serialize_created_at(record.created_at),
|
||
"market_phase_summary": market_phase_summary,
|
||
"asset_type": self._asset_type_for_record(record),
|
||
**market_fields,
|
||
}
|
||
|
||
@staticmethod
|
||
def _asset_type_for_record(record) -> Optional[str]:
|
||
"""Return the parser-derived optional ``asset_type`` for a history row.
|
||
|
||
Uses the *persisted* ``record.code`` (never the display code) via
|
||
:func:`asset_type_from_canonical_code`, so ``sh000016`` is ``index``,
|
||
bare ``000016`` is ``stock`` and market review rows (``MARKET``) omit
|
||
the field. Optional by contract: legacy clients simply ignore it.
|
||
"""
|
||
from src.services.analysis_service import asset_type_from_canonical_code
|
||
|
||
return asset_type_from_canonical_code(getattr(record, "code", None))
|
||
|
||
def _resolve_record(
|
||
self,
|
||
record_id: str,
|
||
*,
|
||
code: Optional[str] = None,
|
||
report_type: Optional[str] = None,
|
||
):
|
||
"""
|
||
Resolve a record_id parameter to an AnalysisHistory object.
|
||
|
||
Tries integer primary key first; falls back to query_id string lookup
|
||
when the value is not a valid integer.
|
||
|
||
Args:
|
||
record_id: integer PK (as string) or query_id string
|
||
|
||
Returns:
|
||
AnalysisHistory object or None
|
||
"""
|
||
try:
|
||
int_id = int(record_id)
|
||
record = self.db.get_analysis_history_by_id(int_id)
|
||
if record:
|
||
return record
|
||
except (ValueError, TypeError):
|
||
pass
|
||
# Fall back to query_id lookup. Keep the old no-kwargs call for
|
||
# unfiltered paths so existing test doubles and integrations remain compatible.
|
||
if code is None and report_type is None:
|
||
return self.db.get_latest_analysis_by_query_id(record_id)
|
||
return self.db.get_latest_analysis_by_query_id(
|
||
record_id,
|
||
code=code,
|
||
report_type=report_type,
|
||
)
|
||
|
||
def resolve_and_get_detail(self, record_id: str) -> Optional[Dict[str, Any]]:
|
||
"""
|
||
Resolve record_id (int PK or query_id string) and return history detail.
|
||
|
||
Args:
|
||
record_id: integer PK (as string) or query_id string
|
||
|
||
Returns:
|
||
Complete analysis report dict, or None
|
||
"""
|
||
try:
|
||
record = self._resolve_record(record_id)
|
||
if not record:
|
||
return None
|
||
return self._record_to_detail_dict(record)
|
||
except Exception as e:
|
||
logger.error(f"resolve_and_get_detail failed for {record_id}: {e}", exc_info=True)
|
||
return None
|
||
|
||
def resolve_and_get_news(self, record_id: str, limit: int = 20) -> List[Dict[str, str]]:
|
||
"""
|
||
Resolve record_id (int PK or query_id string) and return associated news.
|
||
|
||
Args:
|
||
record_id: integer PK (as string) or query_id string
|
||
limit: max items to return
|
||
|
||
Returns:
|
||
List of news intel dicts
|
||
"""
|
||
try:
|
||
record = self._resolve_record(record_id)
|
||
if not record:
|
||
logger.warning(f"resolve_and_get_news: record not found for {record_id}")
|
||
return []
|
||
return self.get_news_intel(query_id=record.query_id, limit=limit)
|
||
except Exception as e:
|
||
logger.error(f"resolve_and_get_news failed for {record_id}: {e}", exc_info=True)
|
||
return []
|
||
|
||
def resolve_and_get_diagnostics(self, record_id: str) -> Optional[Dict[str, Any]]:
|
||
"""
|
||
Resolve record_id and return a user-facing run diagnostic summary.
|
||
|
||
Legacy records without diagnostic snapshots return an ``unknown``
|
||
summary instead of failing. Storage and JSON parsing errors are
|
||
propagated so callers can surface backend failures accurately.
|
||
"""
|
||
record = self._resolve_record(record_id)
|
||
if not record:
|
||
return None
|
||
|
||
return build_run_diagnostic_summary(
|
||
context_snapshot=self._parse_diagnostic_json_field(
|
||
getattr(record, "context_snapshot", None),
|
||
"context_snapshot",
|
||
),
|
||
raw_result=self._parse_diagnostic_json_field(
|
||
getattr(record, "raw_result", None),
|
||
"raw_result",
|
||
),
|
||
report_saved=True,
|
||
query_id=getattr(record, "query_id", None),
|
||
stock_code=getattr(record, "code", None),
|
||
)
|
||
|
||
def resolve_and_get_run_flow(
|
||
self,
|
||
record_id: str,
|
||
*,
|
||
code: Optional[str] = None,
|
||
report_type: Optional[str] = None,
|
||
):
|
||
"""
|
||
Resolve record_id and return a sanitized run-flow snapshot.
|
||
|
||
Uses the same strict JSON parsing behavior as diagnostics so malformed
|
||
persisted payloads surface as backend errors instead of partial graphs.
|
||
"""
|
||
record = self._resolve_record(record_id, code=code, report_type=report_type)
|
||
if not record:
|
||
return None
|
||
|
||
from src.services.run_flow import build_history_run_flow_snapshot
|
||
|
||
return build_history_run_flow_snapshot(
|
||
record,
|
||
context_snapshot=self._parse_diagnostic_json_field(
|
||
getattr(record, "context_snapshot", None),
|
||
"context_snapshot",
|
||
),
|
||
raw_result=self._parse_diagnostic_json_field(
|
||
getattr(record, "raw_result", None),
|
||
"raw_result",
|
||
),
|
||
)
|
||
|
||
@staticmethod
|
||
def _parse_diagnostic_json_field(value: Any, field_name: str) -> Any:
|
||
"""Strict JSON parser for persisted diagnostic inputs."""
|
||
if value is None:
|
||
return None
|
||
if isinstance(value, str):
|
||
if not value.strip():
|
||
return None
|
||
try:
|
||
return json.loads(value)
|
||
except (json.JSONDecodeError, TypeError, ValueError) as exc:
|
||
raise ValueError(f"invalid {field_name} JSON") from exc
|
||
return value
|
||
|
||
def get_history_detail_by_id(self, record_id: int) -> Optional[Dict[str, Any]]:
|
||
"""
|
||
Get history report detail.
|
||
|
||
Uses database primary key for precise query, avoiding returning incorrect records
|
||
due to duplicate query_id in batch analysis.
|
||
|
||
Args:
|
||
record_id: Analysis history record primary key ID
|
||
|
||
Returns:
|
||
Complete analysis report dictionary, or None if not exists
|
||
"""
|
||
try:
|
||
record = self.db.get_analysis_history_by_id(record_id)
|
||
if not record:
|
||
return None
|
||
return self._record_to_detail_dict(record)
|
||
except Exception as e:
|
||
logger.error(f"根据 ID 查询历史详情失败: {e}", exc_info=True)
|
||
return None
|
||
|
||
def get_latest_fundamental_snapshot(
|
||
self,
|
||
*,
|
||
query_id: str,
|
||
stock_code: str,
|
||
) -> Optional[Dict[str, Any]]:
|
||
"""Read the same persisted fundamental fallback used by history detail APIs."""
|
||
return self.db.get_latest_fundamental_snapshot(
|
||
query_id=query_id,
|
||
code=stock_code,
|
||
)
|
||
|
||
@staticmethod
|
||
def _normalize_display_sniper_value(value: Any) -> Optional[str]:
|
||
"""Normalize sniper point values for history display."""
|
||
if value is None:
|
||
return None
|
||
text = str(value).strip()
|
||
if not text or text in {"-", "—", "N/A"}:
|
||
return None
|
||
return text
|
||
|
||
def _get_display_sniper_points(self, record, raw_result: Any) -> Dict[str, Optional[str]]:
|
||
"""Prefer raw dashboard sniper strings for history display, then fall back to numeric DB columns."""
|
||
raw_points: Dict[str, Any] = {}
|
||
if isinstance(raw_result, dict):
|
||
for candidate in (raw_result.get("dashboard"), raw_result):
|
||
if not isinstance(candidate, dict):
|
||
continue
|
||
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
|
||
|
||
display_points: Dict[str, Optional[str]] = {}
|
||
for field in ("ideal_buy", "secondary_buy", "stop_loss", "take_profit"):
|
||
raw_value = self._normalize_display_sniper_value(raw_points.get(field))
|
||
if raw_value is not None:
|
||
display_points[field] = raw_value
|
||
continue
|
||
db_value = getattr(record, field, None)
|
||
display_points[field] = str(db_value) if db_value is not None else None
|
||
return display_points
|
||
|
||
@staticmethod
|
||
def _extract_market_review_content(record, raw_result: Any) -> Optional[str]:
|
||
"""Return persisted market review content from raw_result or news_content."""
|
||
if isinstance(raw_result, dict):
|
||
for field in ("raw_response", "market_review_report"):
|
||
content = raw_result.get(field)
|
||
if isinstance(content, str) and content.strip():
|
||
return content
|
||
|
||
news_content = getattr(record, "news_content", None)
|
||
if isinstance(news_content, str) and news_content.strip():
|
||
return news_content
|
||
return None
|
||
|
||
@staticmethod
|
||
def _market_review_summary(
|
||
persisted_summary: Any,
|
||
markdown: Any,
|
||
*,
|
||
limit: int = 120,
|
||
) -> Optional[str]:
|
||
"""Return a stable short summary without leaking report Markdown or metadata."""
|
||
persisted = str(persisted_summary or "").strip()
|
||
if persisted:
|
||
return persisted
|
||
|
||
source = str(markdown or "").strip()
|
||
if not source:
|
||
return None
|
||
|
||
# Full reports can contain internal reference metadata, tables, links and
|
||
# fenced diagnostic payloads. Keep readable prose only for summary fields.
|
||
source = re.sub(r"```[^\n]*\n.*?```", " ", source, flags=re.DOTALL)
|
||
source = re.sub(r"!\[([^\]]*)\]\([^)]*\)", r"\1", source)
|
||
source = re.sub(r"\[([^\]]+)\]\([^)]*\)", r"\1", source)
|
||
source = re.sub(r"<[^>]+>", " ", source)
|
||
text = markdown_to_plain_text(source)
|
||
text = re.sub(r"[`~|]", " ", text)
|
||
text = re.sub(r"^\s*:?-{3,}:?(?:\s+:?-{3,}:?)+\s*$", " ", text, flags=re.MULTILINE)
|
||
text = re.sub(r"^[+\d]+[.)]\s+", "", text, flags=re.MULTILINE)
|
||
text = re.sub(r"\s+", " ", text).strip(" •-—")
|
||
if not text:
|
||
return None
|
||
if len(text) > limit:
|
||
return text[:limit].rstrip() + "…"
|
||
return text
|
||
|
||
def _record_to_detail_dict(self, record) -> Dict[str, Any]:
|
||
"""
|
||
Convert an AnalysisHistory ORM record to a detail response dict.
|
||
"""
|
||
raw_result = parse_json_field(record.raw_result)
|
||
|
||
model_used = (raw_result or {}).get("model_used") if isinstance(raw_result, dict) else None
|
||
model_used = normalize_model_used(model_used)
|
||
sniper_points = self._get_display_sniper_points(record, raw_result)
|
||
|
||
context_snapshot = None
|
||
if record.context_snapshot:
|
||
try:
|
||
context_snapshot = json.loads(record.context_snapshot)
|
||
except json.JSONDecodeError:
|
||
context_snapshot = record.context_snapshot
|
||
|
||
report_language = normalize_report_language(
|
||
raw_result.get("report_language") if isinstance(raw_result, dict) else None
|
||
)
|
||
news_disclosure = empty_news_disclosure_from_stored(
|
||
raw_result,
|
||
context_snapshot,
|
||
report_language,
|
||
)
|
||
|
||
market_review_content = None
|
||
analysis_summary = record.analysis_summary
|
||
if getattr(record, "report_type", None) == "market_review":
|
||
market_review_content = self._extract_market_review_content(record, raw_result)
|
||
analysis_summary = self._market_review_summary(
|
||
record.analysis_summary,
|
||
market_review_content,
|
||
)
|
||
|
||
action_fields = self._decision_action_fields_for_record(record, raw_result)
|
||
display_code = self._display_stock_code(record.code)
|
||
market_phase_summary = self._display_market_phase_summary(record.code, context_snapshot)
|
||
return {
|
||
"id": record.id,
|
||
"query_id": record.query_id,
|
||
"stock_code": display_code,
|
||
"storage_stock_code": str(record.code or "").strip(),
|
||
"stock_name": record.name,
|
||
"report_type": record.report_type,
|
||
"created_at": self._serialize_created_at(record.created_at),
|
||
"model_used": model_used,
|
||
"analysis_summary": analysis_summary,
|
||
"operation_advice": record.operation_advice,
|
||
"action": action_fields["action"],
|
||
"action_label": action_fields["action_label"],
|
||
"trend_prediction": record.trend_prediction,
|
||
"sentiment_score": record.sentiment_score,
|
||
"sentiment_label": self._get_sentiment_label(record.sentiment_score or 50),
|
||
"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"),
|
||
"news_content": market_review_content or record.news_content,
|
||
"empty_news_disclosure": news_disclosure,
|
||
"raw_result": raw_result,
|
||
"context_snapshot": context_snapshot,
|
||
"market_phase_summary": market_phase_summary,
|
||
}
|
||
|
||
def _decision_action_fields_for_record(self, record, raw_result: Any) -> Dict[str, Any]:
|
||
raw = raw_result if isinstance(raw_result, dict) else {}
|
||
return display_action_fields(
|
||
operation_advice=raw.get("operation_advice") or getattr(record, "operation_advice", None),
|
||
explicit_action=raw.get("action"),
|
||
action_label=raw.get("action_label"),
|
||
report_type=getattr(record, "report_type", None),
|
||
report_language=normalize_report_language(raw.get("report_language")),
|
||
sentiment_score=getattr(record, "sentiment_score", None),
|
||
guardrail_reason=extract_decision_guardrail_reason(raw),
|
||
)
|
||
|
||
def delete_history_records(self, record_ids: List[int]) -> int:
|
||
"""
|
||
Delete specified analysis history records.
|
||
|
||
Args:
|
||
record_ids: List of history record primary key IDs
|
||
|
||
Returns:
|
||
Number of records actually deleted
|
||
|
||
Raises:
|
||
Exception: Re-raises any storage-layer exception so the API caller
|
||
receives a proper 500 error instead of a silent success.
|
||
"""
|
||
return self.db.delete_analysis_history_records(record_ids)
|
||
|
||
def get_news_intel(self, query_id: str, limit: int = 20) -> List[Dict[str, str]]:
|
||
"""
|
||
Get news intelligence associated with a specified query_id.
|
||
|
||
Args:
|
||
query_id: Unique analysis identifier
|
||
limit: Result limit
|
||
|
||
Returns:
|
||
List of news intelligence (containing title, snippet, and url)
|
||
"""
|
||
try:
|
||
records = self.db.get_news_intel_by_query_id(query_id=query_id, limit=limit)
|
||
|
||
if not records:
|
||
records = self._fallback_news_by_analysis_context(query_id=query_id, limit=limit)
|
||
|
||
items: List[Dict[str, str]] = []
|
||
for record in records:
|
||
snippet = (record.snippet or "").strip()
|
||
if len(snippet) > 200:
|
||
snippet = f"{snippet[:197]}..."
|
||
items.append({
|
||
"title": record.title,
|
||
"snippet": snippet,
|
||
"url": record.url,
|
||
})
|
||
|
||
return items
|
||
|
||
except Exception as e:
|
||
logger.error(f"查询新闻情报失败: {e}", exc_info=True)
|
||
return []
|
||
|
||
def get_news_intel_by_record_id(self, record_id: int, limit: int = 20) -> List[Dict[str, str]]:
|
||
"""
|
||
Get associated news intelligence based on analysis history record ID.
|
||
|
||
Parses record_id to query_id, then calls get_news_intel.
|
||
|
||
Args:
|
||
record_id: Analysis history primary key ID
|
||
limit: Result limit
|
||
|
||
Returns:
|
||
List of news intelligence (containing title, snippet, and url)
|
||
"""
|
||
try:
|
||
# Look up the corresponding AnalysisHistory record by record_id
|
||
record = self.db.get_analysis_history_by_id(record_id)
|
||
if not record:
|
||
logger.warning(f"No analysis record found for record_id={record_id}")
|
||
return []
|
||
|
||
# Get query_id from record, then call original method
|
||
return self.get_news_intel(query_id=record.query_id, limit=limit)
|
||
|
||
except Exception as e:
|
||
logger.error(f"根据 record_id 查询新闻情报失败: {e}", exc_info=True)
|
||
return []
|
||
|
||
def _fallback_news_by_analysis_context(self, query_id: str, limit: int) -> List[Any]:
|
||
"""
|
||
Fallback by analysis context when direct query_id lookup returns no news.
|
||
|
||
Typical scenarios:
|
||
- URL-level dedup keeps one canonical news row across repeated analyses.
|
||
- Legacy records may have different historical query_id strategies.
|
||
"""
|
||
records = self.db.get_analysis_history(query_id=query_id, limit=1)
|
||
if not records:
|
||
return []
|
||
|
||
analysis = records[0]
|
||
if not analysis.code or not analysis.created_at:
|
||
return []
|
||
|
||
# Narrow down to same-stock recent news, then filter by analysis time window.
|
||
days = max(1, (datetime.now() - analysis.created_at).days + 1)
|
||
candidates = self.db.get_recent_news(code=analysis.code, days=days, limit=max(limit * 5, 50))
|
||
|
||
start_time = analysis.created_at - timedelta(hours=6)
|
||
end_time = analysis.created_at + timedelta(hours=6)
|
||
matched = [
|
||
item for item in candidates
|
||
if item.fetched_at and start_time <= item.fetched_at <= end_time
|
||
]
|
||
|
||
# 历史兜底链路也做发布时间硬过滤，避免旧库脏数据重新冒出。
|
||
cfg = get_config()
|
||
window_days = resolve_news_window_days(
|
||
news_max_age_days=getattr(cfg, "news_max_age_days", 3),
|
||
news_strategy_profile=getattr(cfg, "news_strategy_profile", "short"),
|
||
)
|
||
# Anchor to analysis date instead of "today" to preserve historical context.
|
||
anchor_date = analysis.created_at.date()
|
||
latest_allowed = anchor_date + timedelta(days=1)
|
||
earliest_allowed = anchor_date - timedelta(days=max(0, window_days - 1))
|
||
|
||
filtered = []
|
||
for item in matched:
|
||
if not item.published_date:
|
||
continue
|
||
if isinstance(item.published_date, datetime):
|
||
published = item.published_date.date()
|
||
elif isinstance(item.published_date, date):
|
||
published = item.published_date
|
||
else:
|
||
continue
|
||
if earliest_allowed <= published <= latest_allowed:
|
||
filtered.append(item)
|
||
|
||
return filtered[:limit]
|
||
|
||
def _get_sentiment_label(self, score: int) -> str:
|
||
"""
|
||
Get sentiment label based on score.
|
||
|
||
Args:
|
||
score: Sentiment score (0-100)
|
||
|
||
Returns:
|
||
Sentiment label
|
||
"""
|
||
if score >= 80:
|
||
return "极度乐观"
|
||
elif score >= 60:
|
||
return "乐观"
|
||
elif score >= 40:
|
||
return "中性"
|
||
elif score >= 20:
|
||
return "悲观"
|
||
else:
|
||
return "极度悲观"
|
||
|
||
def get_markdown_report(self, record_id: str) -> Optional[str]:
|
||
"""
|
||
Generate a Markdown report for a single analysis history record.
|
||
|
||
This method reconstructs an AnalysisResult from the stored raw_result
|
||
and generates a detailed Markdown report similar to the push notifications.
|
||
|
||
Args:
|
||
record_id: integer PK (as string) or query_id string
|
||
|
||
Returns:
|
||
Markdown formatted report string, or None if record not found
|
||
|
||
Raises:
|
||
MarkdownReportGenerationError: If report generation fails due to internal errors
|
||
"""
|
||
record = self._resolve_record(record_id)
|
||
if not record:
|
||
logger.warning(f"get_markdown_report: record not found for {record_id}")
|
||
return None
|
||
|
||
# Rebuild AnalysisResult from raw_result
|
||
raw_result = parse_json_field(record.raw_result)
|
||
if not raw_result:
|
||
logger.error(f"get_markdown_report: raw_result is empty for {record_id}")
|
||
raise MarkdownReportGenerationError(
|
||
f"raw_result is empty or invalid for record {record_id}",
|
||
record_id=record_id
|
||
)
|
||
|
||
if getattr(record, "report_type", None) == "market_review":
|
||
markdown_report = self._extract_market_review_content(record, raw_result)
|
||
if markdown_report:
|
||
return markdown_report
|
||
logger.error(f"get_markdown_report: market review report is empty for {record_id}")
|
||
raise MarkdownReportGenerationError(
|
||
f"market review report is empty for record {record_id}",
|
||
record_id=record_id,
|
||
)
|
||
|
||
try:
|
||
result = self._rebuild_analysis_result(raw_result, record)
|
||
except Exception as e:
|
||
logger.error(f"get_markdown_report: failed to rebuild AnalysisResult for {record_id}: {e}", exc_info=True)
|
||
raise MarkdownReportGenerationError(
|
||
f"Failed to rebuild AnalysisResult: {str(e)}",
|
||
record_id=record_id
|
||
) from e
|
||
|
||
if not result:
|
||
logger.error(f"get_markdown_report: _rebuild_analysis_result returned None for {record_id}")
|
||
raise MarkdownReportGenerationError(
|
||
"Failed to rebuild AnalysisResult from raw_result",
|
||
record_id=record_id
|
||
)
|
||
|
||
# Generate Markdown report
|
||
try:
|
||
return self._generate_single_stock_markdown(result, record)
|
||
except Exception as e:
|
||
logger.error(f"get_markdown_report: failed to generate markdown for {record_id}: {e}", exc_info=True)
|
||
raise MarkdownReportGenerationError(
|
||
f"Failed to generate markdown report: {str(e)}",
|
||
record_id=record_id
|
||
) from e
|
||
|
||
def _rebuild_analysis_result(
|
||
self,
|
||
raw_result: Dict[str, Any],
|
||
record
|
||
) -> Optional[AnalysisResult]:
|
||
"""
|
||
Rebuild an AnalysisResult object from stored raw_result dict.
|
||
|
||
Args:
|
||
raw_result: The parsed raw_result JSON dict
|
||
record: The AnalysisHistory ORM record
|
||
|
||
Returns:
|
||
AnalysisResult object or None
|
||
"""
|
||
try:
|
||
from src.analyzer import AnalysisResult
|
||
# Extract dashboard data if available
|
||
dashboard = raw_result.get("dashboard", {})
|
||
context_snapshot = parse_json_field(getattr(record, "context_snapshot", None))
|
||
news_result_count, news_result_count_known = persisted_news_result_state(
|
||
raw_result,
|
||
context_snapshot,
|
||
)
|
||
|
||
# Build AnalysisResult with available data
|
||
result = AnalysisResult(
|
||
code=raw_result.get("code", record.code),
|
||
name=raw_result.get("name", record.name),
|
||
sentiment_score=raw_result.get("sentiment_score", record.sentiment_score or 50),
|
||
trend_prediction=raw_result.get("trend_prediction", record.trend_prediction or ""),
|
||
operation_advice=raw_result.get("operation_advice", record.operation_advice or ""),
|
||
decision_type=raw_result.get("decision_type", "hold"),
|
||
confidence_level=raw_result.get("confidence_level", "中"),
|
||
report_language=normalize_report_language(raw_result.get("report_language")),
|
||
action=raw_result.get("action"),
|
||
action_label=raw_result.get("action_label"),
|
||
dashboard=dashboard,
|
||
trend_analysis=raw_result.get("trend_analysis", ""),
|
||
short_term_outlook=raw_result.get("short_term_outlook", ""),
|
||
medium_term_outlook=raw_result.get("medium_term_outlook", ""),
|
||
technical_analysis=raw_result.get("technical_analysis", ""),
|
||
ma_analysis=raw_result.get("ma_analysis", ""),
|
||
volume_analysis=raw_result.get("volume_analysis", ""),
|
||
pattern_analysis=raw_result.get("pattern_analysis", ""),
|
||
fundamental_analysis=raw_result.get("fundamental_analysis", ""),
|
||
sector_position=raw_result.get("sector_position", ""),
|
||
company_highlights=raw_result.get("company_highlights", ""),
|
||
news_summary=raw_result.get("news_summary", record.news_content or ""),
|
||
market_sentiment=raw_result.get("market_sentiment", ""),
|
||
hot_topics=raw_result.get("hot_topics", ""),
|
||
analysis_summary=raw_result.get("analysis_summary", record.analysis_summary or ""),
|
||
key_points=raw_result.get("key_points", ""),
|
||
risk_warning=raw_result.get("risk_warning", ""),
|
||
buy_reason=raw_result.get("buy_reason", ""),
|
||
market_snapshot=raw_result.get("market_snapshot"),
|
||
search_performed=raw_result.get("search_performed", False),
|
||
news_result_count=news_result_count,
|
||
news_result_count_known=news_result_count_known,
|
||
news_evidence_present=persisted_news_evidence_present(
|
||
raw_result, news_result_count
|
||
),
|
||
data_sources=raw_result.get("data_sources", ""),
|
||
success=raw_result.get("success", True),
|
||
error_message=raw_result.get("error_message"),
|
||
current_price=raw_result.get("current_price"),
|
||
change_pct=raw_result.get("change_pct"),
|
||
model_used=raw_result.get("model_used"),
|
||
)
|
||
guardrail_reason = extract_decision_guardrail_reason(raw_result)
|
||
if guardrail_reason:
|
||
setattr(result, "guardrail_reason", guardrail_reason)
|
||
return result
|
||
except Exception as e:
|
||
logger.error(f"Failed to rebuild AnalysisResult: {e}", exc_info=True)
|
||
return None
|
||
|
||
def _generate_single_stock_markdown(
|
||
self,
|
||
result: AnalysisResult,
|
||
record
|
||
) -> str:
|
||
"""
|
||
Generate a Markdown report for a single stock analysis.
|
||
|
||
This follows the same format as NotificationService.generate_dashboard_report()
|
||
using dashboard structured data for detailed report.
|
||
|
||
Args:
|
||
result: The AnalysisResult object
|
||
record: The AnalysisHistory ORM record
|
||
|
||
Returns:
|
||
Markdown formatted report string
|
||
"""
|
||
report_date = record.created_at.strftime("%Y-%m-%d") if record.created_at else datetime.now().strftime("%Y-%m-%d")
|
||
report_time = record.created_at.strftime("%H:%M:%S") if record.created_at else datetime.now().strftime("%H:%M:%S")
|
||
report_language = normalize_report_language(getattr(result, "report_language", "zh"))
|
||
labels = get_report_labels(report_language)
|
||
|
||
def _label(en: str, zh: str, ko: str) -> str:
|
||
if report_language == "en":
|
||
return en
|
||
if report_language == "ko":
|
||
return ko
|
||
return zh
|
||
|
||
analysis_date_label = _label("Analysis Date", "分析日期", "분석일")
|
||
report_time_label = _label("Report Time", "报告生成时间", "생성 시각")
|
||
reason_label = _label("Rationale", "操作理由", "판단 근거")
|
||
risk_warning_label = _label("Risk Warning", "风险提示", "리스크 경고")
|
||
technical_heading = _label("Technicals", "技术面", "기술적 분석")
|
||
ma_label = _label("Moving Averages", "均线", "이동평균")
|
||
volume_analysis_label = _label("Volume", "量能", "거래량")
|
||
news_heading = _label("News Flow", "消息面", "뉴스 흐름")
|
||
|
||
# Escape markdown special characters in stock name
|
||
name_escaped = self._escape_md(
|
||
get_localized_stock_name(result.name, result.code, report_language)
|
||
) or result.code
|
||
|
||
# Get signal level
|
||
signal_text, signal_emoji, signal_tag = self._get_signal_level(result)
|
||
dashboard = result.dashboard if hasattr(result, 'dashboard') and result.dashboard else {}
|
||
|
||
report_lines = [
|
||
f"# 📊 {name_escaped} ({result.code}) {labels['report_title']}",
|
||
"",
|
||
f"> {analysis_date_label}: **{report_date}** | {report_time_label}: {report_time}",
|
||
"",
|
||
"---",
|
||
"",
|
||
]
|
||
|
||
news_disclosure = empty_news_disclosure(result, report_language)
|
||
if news_disclosure:
|
||
report_lines.extend([news_disclosure, ""])
|
||
|
||
# ========== 舆情与基本面概览（放在最前面）==========
|
||
intel = dashboard.get('intelligence', {}) if dashboard else {}
|
||
if intel:
|
||
report_lines.extend([
|
||
f"### 📰 {labels['info_heading']}",
|
||
"",
|
||
])
|
||
# 舆情情绪总结
|
||
if intel.get('sentiment_summary'):
|
||
report_lines.append(f"**💭 {labels['sentiment_summary_label']}**: {intel['sentiment_summary']}")
|
||
# 业绩预期
|
||
if intel.get('earnings_outlook'):
|
||
report_lines.append(f"**📊 {labels['earnings_outlook_label']}**: {intel['earnings_outlook']}")
|
||
# 风险警报（醒目显示）
|
||
risk_alerts = intel.get('risk_alerts', [])
|
||
if risk_alerts:
|
||
report_lines.append("")
|
||
report_lines.append(f"**🚨 {labels['risk_alerts_label']}**:")
|
||
for alert in risk_alerts:
|
||
report_lines.append(f"- {alert}")
|
||
# 利好催化
|
||
catalysts = intel.get('positive_catalysts', [])
|
||
if catalysts:
|
||
report_lines.append("")
|
||
report_lines.append(f"**✨ {labels['positive_catalysts_label']}**:")
|
||
for cat in catalysts:
|
||
report_lines.append(f"- {cat}")
|
||
# 最新消息
|
||
if intel.get('latest_news'):
|
||
report_lines.append("")
|
||
report_lines.append(f"**📢 {labels['latest_news_label']}**: {intel['latest_news']}")
|
||
report_lines.append("")
|
||
|
||
# ========== 核心结论 ==========
|
||
core = dashboard.get('core_conclusion', {}) if dashboard else {}
|
||
one_sentence = core.get('one_sentence', result.analysis_summary)
|
||
time_sense = core.get('time_sensitivity', labels['default_time_sensitivity'])
|
||
pos_advice = core.get('position_advice', {})
|
||
|
||
report_lines.extend([
|
||
f"### 📌 {labels['core_conclusion_heading']}",
|
||
"",
|
||
f"**{signal_emoji} {signal_text}** | {localize_trend_prediction(result.trend_prediction, report_language)}",
|
||
"",
|
||
f"> **{labels['one_sentence_label']}**: {one_sentence}",
|
||
"",
|
||
f"⏰ **{labels['time_sensitivity_label']}**: {time_sense}",
|
||
"",
|
||
])
|
||
# 持仓分类建议
|
||
if pos_advice:
|
||
report_lines.extend([
|
||
f"| {labels['position_status_label']} | {labels['action_advice_label']} |",
|
||
"|---------|---------|",
|
||
f"| 🆕 **{labels['no_position_label']}** | {pos_advice.get('no_position', self._get_display_operation_advice(result, report_language))} |",
|
||
f"| 💼 **{labels['has_position_label']}** | {pos_advice.get('has_position', labels['continue_holding'])} |",
|
||
"",
|
||
])
|
||
|
||
# ========== 行情快照 ==========
|
||
self._append_market_snapshot_to_report(report_lines, result, labels)
|
||
|
||
# ========== 数据透视 ==========
|
||
data_persp = dashboard.get('data_perspective', {}) if dashboard else {}
|
||
if data_persp:
|
||
trend_data = data_persp.get('trend_status', {})
|
||
price_data = data_persp.get('price_position', {})
|
||
vol_data = data_persp.get('volume_analysis', {})
|
||
chip_data = data_persp.get('chip_structure', {})
|
||
|
||
report_lines.extend([
|
||
f"### 📊 {labels['data_perspective_heading']}",
|
||
"",
|
||
])
|
||
# 趋势状态
|
||
if trend_data:
|
||
is_bullish = (
|
||
f"✅ {labels['yes_label']}"
|
||
if trend_data.get('is_bullish', False)
|
||
else f"❌ {labels['no_label']}"
|
||
)
|
||
report_lines.extend([
|
||
f"**{labels['ma_alignment_label']}**: {trend_data.get('ma_alignment', 'N/A')} | "
|
||
f"{labels['bullish_alignment_label']}: {is_bullish} | "
|
||
f"{labels['trend_strength_label']}: {trend_data.get('trend_score', 'N/A')}/100",
|
||
"",
|
||
])
|
||
# 价格位置
|
||
if price_data:
|
||
raw_bias_status = price_data.get('bias_status', 'N/A')
|
||
bias_status = localize_bias_status(raw_bias_status, report_language)
|
||
bias_emoji = get_bias_status_emoji(raw_bias_status)
|
||
report_lines.extend([
|
||
f"| {labels['price_metrics_label']} | {labels['current_price_label']} |",
|
||
"|---------|------|",
|
||
f"| {labels['current_price_label']} | {price_data.get('current_price', 'N/A')} |",
|
||
f"| {labels['ma5_label']} | {price_data.get('ma5', 'N/A')} |",
|
||
f"| {labels['ma10_label']} | {price_data.get('ma10', 'N/A')} |",
|
||
f"| {labels['ma20_label']} | {price_data.get('ma20', 'N/A')} |",
|
||
f"| {labels['bias_ma5_label']} | {price_data.get('bias_ma5', 'N/A')}% {bias_emoji}{bias_status} |",
|
||
f"| {labels['support_level_label']} | {price_data.get('support_level', 'N/A')} |",
|
||
f"| {labels['resistance_level_label']} | {price_data.get('resistance_level', 'N/A')} |",
|
||
"",
|
||
])
|
||
# 量能分析
|
||
if vol_data:
|
||
report_lines.extend([
|
||
f"**{labels['volume_label']}**: {labels['volume_ratio_label']} {vol_data.get('volume_ratio', 'N/A')} "
|
||
f"({vol_data.get('volume_status', '')}) | {labels['turnover_rate_label']} {vol_data.get('turnover_rate', 'N/A')}%",
|
||
f"💡 *{vol_data.get('volume_meaning', '')}*",
|
||
"",
|
||
])
|
||
# 筹码结构
|
||
if chip_data:
|
||
if is_chip_structure_unavailable(chip_data):
|
||
report_lines.extend([
|
||
f"**{labels['chip_label']}**: {get_chip_unavailable_reason(chip_data, report_language)}",
|
||
"",
|
||
])
|
||
else:
|
||
raw_chip_health = chip_data.get('chip_health', 'N/A')
|
||
chip_health = localize_chip_health(raw_chip_health, report_language)
|
||
normalized_chip_health = str(raw_chip_health or "").strip().lower()
|
||
if normalized_chip_health in {"健康", "healthy"}:
|
||
chip_emoji = "✅"
|
||
elif normalized_chip_health in {"一般", "average"}:
|
||
chip_emoji = "⚠️"
|
||
else:
|
||
chip_emoji = "🚨"
|
||
report_lines.extend([
|
||
f"**{labels['chip_label']}**: {chip_data.get('profit_ratio', 'N/A')} | {chip_data.get('avg_cost', 'N/A')} | "
|
||
f"{chip_data.get('concentration', 'N/A')} {chip_emoji}{chip_health}",
|
||
"",
|
||
])
|
||
else:
|
||
chip_unavailable_reason = get_chip_unavailable_reason(data_persp, report_language)
|
||
if chip_unavailable_reason:
|
||
report_lines.extend([
|
||
f"**{labels['chip_label']}**: {chip_unavailable_reason}",
|
||
"",
|
||
])
|
||
|
||
# ========== 作战计划 ==========
|
||
battle = dashboard.get('battle_plan', {}) if dashboard else {}
|
||
if battle:
|
||
report_lines.extend([
|
||
f"### 🎯 {labels['battle_plan_heading']}",
|
||
"",
|
||
])
|
||
# 狙击点位
|
||
sniper = battle.get('sniper_points', {})
|
||
if sniper:
|
||
report_lines.extend([
|
||
f"**📍 {labels['action_points_heading']}**",
|
||
"",
|
||
f"| {labels['action_points_heading']} | {labels['current_price_label']} |",
|
||
"|---------|------|",
|
||
f"| 🎯 {labels['ideal_buy_label']} | {self._clean_sniper_value(sniper.get('ideal_buy', 'N/A'))} |",
|
||
f"| 🔵 {labels['secondary_buy_label']} | {self._clean_sniper_value(sniper.get('secondary_buy', 'N/A'))} |",
|
||
f"| 🛑 {labels['stop_loss_label']} | {self._clean_sniper_value(sniper.get('stop_loss', 'N/A'))} |",
|
||
f"| 🎊 {labels['take_profit_label']} | {self._clean_sniper_value(sniper.get('take_profit', 'N/A'))} |",
|
||
"",
|
||
])
|
||
# 仓位策略
|
||
position = battle.get('position_strategy', {})
|
||
if position:
|
||
report_lines.extend([
|
||
f"**💰 {labels['suggested_position_label']}**: {position.get('suggested_position', 'N/A')}",
|
||
f"- {labels['entry_plan_label']}: {position.get('entry_plan', 'N/A')}",
|
||
f"- {labels['risk_control_label']}: {position.get('risk_control', 'N/A')}",
|
||
"",
|
||
])
|
||
# 检查清单
|
||
checklist = battle.get('action_checklist', []) if battle else []
|
||
if checklist:
|
||
report_lines.extend([
|
||
f"**✅ {labels['checklist_heading']}**",
|
||
"",
|
||
])
|
||
for item in checklist:
|
||
report_lines.append(f"- {item}")
|
||
report_lines.append("")
|
||
|
||
# ========== 信号归因分析 ==========
|
||
signal_attr = dashboard.get('signal_attribution', {}) if dashboard else {}
|
||
if signal_attribution_has_content(signal_attr):
|
||
report_lines.extend([
|
||
f"### 🎯 {labels.get('signal_attribution_heading', '信号归因分析')}",
|
||
"",
|
||
])
|
||
weight_items = signal_attribution_weight_items(signal_attr)
|
||
if weight_items:
|
||
report_lines.append(f"**{labels.get('attribution_weights_label', '归因权重')}**:")
|
||
weight_labels = {
|
||
"technical_indicators": ("📈", labels.get('technical_indicators_label', '技术指标')),
|
||
"news_sentiment": ("📰", labels.get('news_sentiment_label', '新闻舆情')),
|
||
"fundamentals": ("📊", labels.get('fundamentals_label', '基本面')),
|
||
"market_conditions": ("🌐", labels.get('market_conditions_label', '市场环境')),
|
||
}
|
||
for key, value in weight_items:
|
||
icon, label = weight_labels[key]
|
||
report_lines.append(f"- {icon} {label}: {value}%")
|
||
report_lines.append("")
|
||
bullish = signal_attr.get('strongest_bullish_signal')
|
||
bearish = signal_attr.get('strongest_bearish_signal')
|
||
if bullish:
|
||
report_lines.append(f"**🐂 {labels.get('strongest_bullish_signal_label', '最强看多信号')}**: {bullish}")
|
||
if bearish:
|
||
report_lines.append(f"**🐻 {labels.get('strongest_bearish_signal_label', '最强看空信号')}**: {bearish}")
|
||
report_lines.append("")
|
||
|
||
# ========== 多策略综合 ==========
|
||
strategy_synthesis = normalize_strategy_synthesis_payload(
|
||
dashboard.get('strategy_synthesis') if dashboard else None
|
||
)
|
||
if strategy_synthesis:
|
||
confidence = strategy_synthesis.get('confidence')
|
||
confidence_text = f"{confidence:.0%}" if isinstance(confidence, (int, float)) else "N/A"
|
||
report_lines.extend([
|
||
f"### 🧩 {labels.get('strategy_synthesis_heading', '多策略综合')}",
|
||
"",
|
||
(
|
||
f"- {labels.get('strategy_final_signal_label', '综合信号')}: "
|
||
f"{localize_strategy_signal(strategy_synthesis.get('final_signal', 'N/A'), report_language)} | "
|
||
f"{labels.get('strategy_consensus_level_label', '共识度')}: "
|
||
f"{localize_consensus_level(strategy_synthesis.get('consensus_level', 'N/A'), report_language)} | "
|
||
f"{labels.get('strategy_conflict_label', '冲突')}: "
|
||
f"{localize_conflict_severity(strategy_synthesis.get('conflict_severity', 'none'), report_language)} "
|
||
f"({strategy_synthesis.get('conflict_count', 0)}) | "
|
||
f"{labels.get('strategy_confidence_label', '置信度')}: {confidence_text}"
|
||
),
|
||
])
|
||
summary = localize_strategy_synthesis_summary(strategy_synthesis, report_language)
|
||
if summary:
|
||
report_lines.append(f"- {labels.get('strategy_summary_label', '综合说明')}: {summary}")
|
||
report_lines.append(
|
||
f"- {labels.get('strategy_supporting_skills_label', '支持策略')}: "
|
||
f"{self._format_strategy_skill_items(strategy_synthesis.get('supporting_skills'), report_language)}"
|
||
)
|
||
report_lines.append(
|
||
f"- {labels.get('strategy_opposing_skills_label', '反方策略')}: "
|
||
f"{self._format_strategy_skill_items(strategy_synthesis.get('opposing_skills'), report_language)}"
|
||
)
|
||
invalid_count = strategy_invalid_opinion_count(strategy_synthesis)
|
||
if invalid_count:
|
||
invalid_label_template = labels.get(
|
||
"strategy_invalid_opinions_label",
|
||
"另有 {count} 个策略解析失败",
|
||
)
|
||
try:
|
||
invalid_text = invalid_label_template.format(count=invalid_count)
|
||
except (KeyError, IndexError):
|
||
invalid_text = f"{invalid_label_template}: {invalid_count}"
|
||
report_lines.append(f"- {invalid_text}")
|
||
report_lines.append("")
|
||
|
||
# ========== 如果没有 dashboard，显示传统格式 ==========
|
||
if not dashboard:
|
||
# 操作理由
|
||
if result.buy_reason:
|
||
report_lines.extend([
|
||
f"**💡 {reason_label}**: {result.buy_reason}",
|
||
"",
|
||
])
|
||
# 风险提示
|
||
if result.risk_warning:
|
||
report_lines.extend([
|
||
f"**⚠️ {risk_warning_label}**: {result.risk_warning}",
|
||
"",
|
||
])
|
||
# 技术面分析
|
||
if result.ma_analysis or result.volume_analysis:
|
||
report_lines.extend([
|
||
f"### 📊 {technical_heading}",
|
||
"",
|
||
])
|
||
if result.ma_analysis:
|
||
report_lines.append(f"**{ma_label}**: {result.ma_analysis}")
|
||
if result.volume_analysis:
|
||
report_lines.append(f"**{volume_analysis_label}**: {result.volume_analysis}")
|
||
report_lines.append("")
|
||
# 消息面
|
||
if result.news_summary:
|
||
report_lines.extend([
|
||
f"### 📰 {news_heading}",
|
||
f"{result.news_summary}",
|
||
"",
|
||
])
|
||
|
||
# ========== 底部 ==========
|
||
report_lines.extend([
|
||
"---",
|
||
"",
|
||
f"*{labels['generated_at_label']}: {report_time}*",
|
||
])
|
||
|
||
return "\n".join(report_lines)
|
||
|
||
@staticmethod
|
||
def _format_strategy_skill_items(items: Any, report_language: str = "zh") -> str:
|
||
none_text = get_report_labels(report_language).get("none_label", "None")
|
||
if not isinstance(items, list):
|
||
return none_text
|
||
formatted: List[str] = []
|
||
for item in items:
|
||
if not isinstance(item, dict):
|
||
continue
|
||
skill_id = str(item.get("skill_id") or "").strip()
|
||
signal = str(item.get("signal") or "").strip()
|
||
confidence = item.get("confidence")
|
||
if not skill_id:
|
||
continue
|
||
suffix = f"/{localize_strategy_signal(signal, report_language)}" if signal else ""
|
||
if isinstance(confidence, (int, float)):
|
||
suffix += f"/{confidence:.0%}"
|
||
formatted.append(f"{localize_strategy_skill(skill_id, report_language)}{suffix}")
|
||
return "、".join(formatted) if formatted else none_text
|
||
|
||
@staticmethod
|
||
def _escape_md(text: Optional[str]) -> str:
|
||
"""Escape markdown special characters."""
|
||
if not text:
|
||
return ""
|
||
return text.replace('*', r'\*')
|
||
|
||
@staticmethod
|
||
def _clean_sniper_value(value: Any) -> str:
|
||
"""Clean sniper point value for display."""
|
||
if value is None:
|
||
return "N/A"
|
||
text = str(value).strip()
|
||
if not text or text in ("-", "—", "N/A", "None"):
|
||
return "N/A"
|
||
return text
|
||
|
||
def _get_display_operation_advice(
|
||
self,
|
||
result: AnalysisResult,
|
||
report_language: Optional[str] = None,
|
||
) -> str:
|
||
return display_operation_advice_for_result(
|
||
result,
|
||
report_language=report_language or getattr(result, "report_language", "zh"),
|
||
)
|
||
|
||
def _get_signal_level(self, result: AnalysisResult) -> Tuple[str, str, str]:
|
||
"""Get display text and signal metadata from the resolved action."""
|
||
report_language = getattr(result, "report_language", "zh")
|
||
display_fields = display_action_fields_for_result(
|
||
result,
|
||
report_language=report_language,
|
||
)
|
||
signal_advice = {
|
||
"buy": "buy",
|
||
"add": "buy",
|
||
"hold": "hold",
|
||
"reduce": "reduce",
|
||
"sell": "sell",
|
||
"watch": "watch",
|
||
"avoid": "hold",
|
||
"alert": "sell",
|
||
}.get(display_fields["action"])
|
||
_, emoji, signal_tag = get_signal_level(
|
||
signal_advice or self._get_display_operation_advice(result, report_language),
|
||
result.sentiment_score,
|
||
report_language,
|
||
)
|
||
return (
|
||
self._get_display_operation_advice(result, report_language),
|
||
emoji,
|
||
signal_tag,
|
||
)
|
||
|
||
@staticmethod
|
||
def _safe_format_number(value: Any, fmt: str = ".2f") -> str:
|
||
"""
|
||
Safely format a numeric value that may be a string.
|
||
|
||
Args:
|
||
value: The value to format (may be int, float, or string like "12.34" or "N/A")
|
||
fmt: Format string (default: ".2f")
|
||
|
||
Returns:
|
||
Formatted string or original string if not a valid number
|
||
"""
|
||
if value is None:
|
||
return "N/A"
|
||
if isinstance(value, (int, float)):
|
||
return f"{value:{fmt}}"
|
||
if isinstance(value, str):
|
||
value = value.strip()
|
||
if not value or value in ("N/A", "-", "—", "None"):
|
||
return "N/A"
|
||
try:
|
||
return f"{float(value):{fmt}}"
|
||
except (ValueError, TypeError):
|
||
return value
|
||
return str(value)
|
||
|
||
@staticmethod
|
||
def _append_market_snapshot_to_report(
|
||
lines: List[str],
|
||
result: AnalysisResult,
|
||
labels: Dict[str, str],
|
||
) -> None:
|
||
"""Append market snapshot data to report lines."""
|
||
snapshot = getattr(result, 'market_snapshot', None)
|
||
if not snapshot:
|
||
return
|
||
|
||
lines.extend([
|
||
f"### 📈 {labels['market_snapshot_heading']}",
|
||
"",
|
||
f"| {labels['price_metrics_label']} | {labels['current_price_label']} |",
|
||
"|------|------|",
|
||
])
|
||
|
||
# Price info
|
||
current_price = snapshot.get('price') or snapshot.get('current_price') or result.current_price
|
||
change_pct = snapshot.get('change_pct') or snapshot.get('pct_chg') or result.change_pct
|
||
if current_price is not None:
|
||
current_str = HistoryService._safe_format_number(current_price, ".2f")
|
||
if change_pct is not None:
|
||
if isinstance(change_pct, str) and change_pct.strip().endswith("%"):
|
||
change_str = change_pct.strip()
|
||
else:
|
||
change_str = f"{HistoryService._safe_format_number(change_pct, '+.2f')}%"
|
||
else:
|
||
change_str = "--"
|
||
lines.append(f"| {labels['current_price_label']} | **{current_str}** ({change_str}) |")
|
||
|
||
# Other metrics
|
||
metrics = [
|
||
(labels['open_label'], "open", ".2f"),
|
||
(labels['high_label'], "high", ".2f"),
|
||
(labels['low_label'], "low", ".2f"),
|
||
(labels['volume_label'], "volume", ",.0f"),
|
||
(labels['amount_label'], "amount", ",.0f"),
|
||
]
|
||
for label, key, fmt in metrics:
|
||
value = snapshot.get(key)
|
||
if value is not None:
|
||
formatted = HistoryService._safe_format_number(value, fmt)
|
||
lines.append(f"| {label} | {formatted} |")
|
||
|
||
lines.extend(["", "---", ""])
|