From d28bb5788b891081d10a62c82cfc9d7c57a19403 Mon Sep 17 00:00:00 2001 From: leefer Date: Fri, 31 Jul 2026 01:41:58 +0800 Subject: [PATCH] migration: preserve sentiment and pools slice --- backend/application.py | 176 +------ backend/data/providers/tushare_client.py | 2 +- backend/features/market/service.py | 2 +- backend/features/pools/__init__.py | 6 + backend/features/pools/repository.py | 27 + backend/features/pools/service.py | 150 ++++++ backend/features/sentiment/__init__.py | 19 + backend/features/sentiment/engine.py | 496 +++++++++++++++++ backend/features/sentiment/service.py | 39 ++ database.py | 23 +- demo_data.py | 2 +- screener.py | 2 +- sentiment_engine.py | 497 +----------------- tests/test_preservation_slice_market.py | 5 +- ...test_preservation_slice_sentiment_pools.py | 114 ++++ 15 files changed, 870 insertions(+), 690 deletions(-) create mode 100644 backend/features/pools/__init__.py create mode 100644 backend/features/pools/repository.py create mode 100644 backend/features/pools/service.py create mode 100644 backend/features/sentiment/__init__.py create mode 100644 backend/features/sentiment/engine.py create mode 100644 backend/features/sentiment/service.py create mode 100644 tests/test_preservation_slice_sentiment_pools.py diff --git a/backend/application.py b/backend/application.py index b38b31a..c7bb74d 100644 --- a/backend/application.py +++ b/backend/application.py @@ -53,14 +53,13 @@ from screener import ( from backend.features.accounts.http import AccountHttpMixin from backend.features.accounts.security import SecretVault from backend.features.accounts.service import AccountService -from backend.features.system import SystemHttpMixin -from sentiment_engine import ( - COMPONENT_WEIGHTS, - SENTIMENT_ENGINE_VERSION, - apply_sentiment_to_dashboard, +from backend.features.pools import PoolServiceMixin +from backend.features.sentiment import SentimentServiceMixin +from backend.features.sentiment.engine import ( build_sentiment_history, latest_contiguous_history, ) +from backend.features.system import SystemHttpMixin from backend.data.providers.tushare_client import TushareClient, TushareError, _sector_coverage_issue @@ -141,7 +140,7 @@ MENTOR_ETF_UNIVERSE = ( ) -class DashboardService(MarketServiceMixin): +class DashboardService(MarketServiceMixin, SentimentServiceMixin, PoolServiceMixin): def __init__(self) -> None: runtime = load_runtime_settings() self.vault = SecretVault(runtime.encryption_key) @@ -734,31 +733,6 @@ class DashboardService(MarketServiceMixin): return AccountService.public_personal_profile(personal) - def _enrich_dashboard_sentiment( - self, - dashboard: dict[str, Any], - end_date: str, - ) -> dict[str, Any]: - history = self.database.list_snapshot_payloads(end_date, 260) - return apply_sentiment_to_dashboard(dashboard, history) - - def sentiment_history(self, trade_date: str, limit: int = 20) -> dict[str, Any]: - normalized_date = normalize_date(trade_date) - limit = max(10, min(120, int(limit))) - full_series = build_sentiment_history( - self.database.list_snapshot_payloads(normalized_date, 240) - ) - series = latest_contiguous_history(full_series) - rows = series[-limit:] - return { - "trade_date": rows[-1]["trade_date"] if rows else normalized_date, - "available_days": len(series), - "stored_days": len(full_series), - "requested_days": limit, - "rows": rows, - "weights": COMPONENT_WEIGHTS, - "normalization": rows[-1]["normalization"] if rows else "固定锚点", - } def rotation_history(self, trade_date: str, limit: int = 9) -> dict[str, Any]: normalized_date = normalize_date(trade_date) @@ -3380,146 +3354,6 @@ class DashboardService(MarketServiceMixin): } - def save_reason(self, trade_date: str, code: str, reason: str) -> None: - normalized_date = normalize_date(trade_date) - code = validate_stock_code(code) - reason = reason.strip() - if not reason or len(reason) > 200: - raise ValueError("涨停原因应为 1 至 200 个字符。") - self.database.save_reason_override(normalized_date, code, reason) - - - def _apply_reason_overrides(self, dashboard: dict[str, Any]) -> dict[str, Any]: - trade_date = str(dashboard.get("meta", {}).get("trade_date", "")).replace("-", "") - enrichment = self.database.get_data_snapshot("ifind_event_enrichment_v1", trade_date) - if enrichment: - self._merge_ifind_event_enrichment(dashboard, enrichment) - else: - self._schedule_ifind_event_enrichment(trade_date) - overrides = self.database.reason_overrides(trade_date) - if not overrides: - return dashboard - for key in ("limits", "broken", "down_limits"): - for row in dashboard.get(key) or []: - if row.get("code") in overrides: - row["reason"] = overrides[row["code"]] - row["reason_source"] = "manual" - return dashboard - - def _schedule_ifind_event_enrichment(self, trade_date: str) -> None: - ifind = getattr(self, "ifind", None) - if not ifind or not ifind.configured or not re.fullmatch(r"\d{8}", trade_date): - return - now = datetime.now().astimezone() - if trade_date == now.strftime("%Y%m%d") and now.time().replace(tzinfo=None) < dt_time(15, 0): - return - self.jobs.submit( - "market.ifind-event-enrichment", - f"{trade_date}:v1", - lambda: self._refresh_ifind_event_enrichment(trade_date), - {"trade_date": trade_date, "trigger": "dashboard-enrichment"}, - ) - - def _refresh_ifind_event_enrichment(self, trade_date: str) -> None: - if not self._ifind_event_lock.acquire(blocking=False): - return - try: - if self.database.get_data_snapshot("ifind_event_enrichment_v1", trade_date): - return - ifind = getattr(self, "ifind", None) - if not ifind or not ifind.configured: - return - current = datetime.strptime(trade_date, "%Y%m%d") - display_date = f"{current.year}年{current.month}月{current.day}日" - requests = { - "limits": ( - f"{display_date}涨停股票，股票代码、股票简称、涨停原因、" - "首次涨停时间、最终涨停时间、开板次数" - ), - "broken": ( - f"{display_date}曾涨停但收盘未涨停的股票，股票代码、股票简称、" - "涨停原因、首次涨停时间、开板次数" - ), - "down_limits": ( - f"{display_date}跌停股票，股票代码、股票简称、跌停原因" - ), - } - result: dict[str, Any] = { - "trade_date": trade_date, - "generated_at": datetime.now().astimezone().isoformat(timespec="seconds"), - "limits": {}, "broken": {}, "down_limits": {}, "partial": False, - } - for kind, query in requests.items(): - try: - rows = ifind.wencai(query, "stock", cache_ttl=900) - except IfindError: - result["partial"] = True - continue - for raw in rows: - code = self._ifind_row_code(raw) - if not code: - continue - reason_tokens = ( - ("跌停原因", "风险线索", "原因") - if kind == "down_limits" - else ("涨停原因类别", "涨停原因", "触板逻辑", "原因") - ) - reason = str(self._ifind_field(raw, reason_tokens) or "").strip() - first_time = self._normalize_ifind_event_time( - self._ifind_field(raw, ("首次涨停时间", "首次触板时间", "首次封板时间")) - ) - last_time = self._normalize_ifind_event_time( - self._ifind_field(raw, ("最终涨停时间", "最后涨停时间", "最后封板时间")) - ) - open_times = self._ifind_field(raw, ("开板次数", "打开涨停次数")) - try: - open_count = max(0, int(float(open_times))) if open_times not in (None, "") else None - except (TypeError, ValueError): - open_count = None - result[kind][code] = { - "reason": reason, - "first_time": first_time, - "last_time": last_time, - "open_times": open_count, - } - if any(result[kind] for kind in ("limits", "broken", "down_limits")): - self.database.save_data_snapshot( - "ifind_event_enrichment_v1", trade_date, "ifind", result - ) - finally: - self._ifind_event_lock.release() - - @staticmethod - def _normalize_ifind_event_time(value: Any) -> str: - text = str(value or "").strip() - match = re.search(r"(?:^|\s)(\d{1,2}:\d{2}(?::\d{2})?)(?:$|\s)", text) - if not match: - match = re.search(r"(? None: - for kind in ("limits", "broken", "down_limits"): - records = enrichment.get(kind) or {} - for row in dashboard.get(kind) or []: - event = records.get(str(row.get("code") or "")) or {} - reason = str(event.get("reason") or "").strip() - if reason: - row["reason"] = reason - row["reason_source"] = "market_event" - if event.get("first_time"): - row["first_time"] = event["first_time"] - if event.get("last_time"): - row["last_time"] = event["last_time"] - if event.get("open_times") is not None: - row["open_times"] = event["open_times"] def _apply_seat_aliases(self, payload: dict[str, Any]) -> dict[str, Any]: aliases = self.database.list_seat_aliases() diff --git a/backend/data/providers/tushare_client.py b/backend/data/providers/tushare_client.py index a2f65b0..8e3c71e 100644 --- a/backend/data/providers/tushare_client.py +++ b/backend/data/providers/tushare_client.py @@ -11,7 +11,7 @@ from datetime import datetime, time as dt_time, timedelta from threading import Lock from typing import Any, ClassVar -from sentiment_engine import apply_sentiment_to_dashboard +from backend.features.sentiment.engine import apply_sentiment_to_dashboard TUSHARE_URL = "http://api.tushare.pro" diff --git a/backend/features/market/service.py b/backend/features/market/service.py index 3d5dccd..f4e29db 100644 --- a/backend/features/market/service.py +++ b/backend/features/market/service.py @@ -14,7 +14,7 @@ from backend.bootstrap.config import ( from backend.data.providers.ifind_client import IfindError from backend.data.providers.tushare_client import TushareClient, TushareError from backend.features.market.charts import ChartDataError -from sentiment_engine import SENTIMENT_ENGINE_VERSION +from backend.features.sentiment.engine import SENTIMENT_ENGINE_VERSION SEARCH_INDEXES = ( diff --git a/backend/features/pools/__init__.py b/backend/features/pools/__init__.py new file mode 100644 index 0000000..40bacfb --- /dev/null +++ b/backend/features/pools/__init__.py @@ -0,0 +1,6 @@ +"""Limit-up, broken-board, limit-down and prior-limit pool feature.""" + +from .repository import PoolRepositoryMixin +from .service import PoolServiceMixin + +__all__ = ["PoolRepositoryMixin", "PoolServiceMixin"] diff --git a/backend/features/pools/repository.py b/backend/features/pools/repository.py new file mode 100644 index 0000000..85b5dc0 --- /dev/null +++ b/backend/features/pools/repository.py @@ -0,0 +1,27 @@ +from __future__ import annotations + +from datetime import datetime + + +class PoolRepositoryMixin: + def save_reason_override(self, trade_date: str, code: str, reason: str) -> None: + now = datetime.now().astimezone().isoformat(timespec="seconds") + with self.connect() as connection: + connection.execute( + """ + INSERT INTO reason_overrides (trade_date, code, reason, updated_at) + VALUES (?, ?, ?, ?) + ON CONFLICT(trade_date, code) DO UPDATE SET + reason = excluded.reason, + updated_at = excluded.updated_at + """, + (trade_date, code, reason, now), + ) + + def reason_overrides(self, trade_date: str) -> dict[str, str]: + with self.connect() as connection: + rows = connection.execute( + "SELECT code, reason FROM reason_overrides WHERE trade_date = ?", + (trade_date,), + ).fetchall() + return {row["code"]: row["reason"] for row in rows} diff --git a/backend/features/pools/service.py b/backend/features/pools/service.py new file mode 100644 index 0000000..a4424f3 --- /dev/null +++ b/backend/features/pools/service.py @@ -0,0 +1,150 @@ +from __future__ import annotations + +import re +from datetime import datetime, time as dt_time +from typing import Any + +from backend.bootstrap.config import normalize_date, validate_stock_code +from backend.data.providers.ifind_client import IfindError + + +class PoolServiceMixin: + def save_reason(self, trade_date: str, code: str, reason: str) -> None: + normalized_date = normalize_date(trade_date) + code = validate_stock_code(code) + reason = reason.strip() + if not reason or len(reason) > 200: + raise ValueError("涨停原因应为 1 至 200 个字符。") + self.database.save_reason_override(normalized_date, code, reason) + + def _apply_reason_overrides(self, dashboard: dict[str, Any]) -> dict[str, Any]: + trade_date = str(dashboard.get("meta", {}).get("trade_date", "")).replace("-", "") + enrichment = self.database.get_data_snapshot("ifind_event_enrichment_v1", trade_date) + if enrichment: + self._merge_ifind_event_enrichment(dashboard, enrichment) + else: + self._schedule_ifind_event_enrichment(trade_date) + overrides = self.database.reason_overrides(trade_date) + if not overrides: + return dashboard + for key in ("limits", "broken", "down_limits"): + for row in dashboard.get(key) or []: + if row.get("code") in overrides: + row["reason"] = overrides[row["code"]] + row["reason_source"] = "manual" + return dashboard + + def _schedule_ifind_event_enrichment(self, trade_date: str) -> None: + ifind = getattr(self, "ifind", None) + if not ifind or not ifind.configured or not re.fullmatch(r"\d{8}", trade_date): + return + now = datetime.now().astimezone() + if trade_date == now.strftime("%Y%m%d") and now.time().replace(tzinfo=None) < dt_time(15, 0): + return + self.jobs.submit( + "market.ifind-event-enrichment", + f"{trade_date}:v1", + lambda: self._refresh_ifind_event_enrichment(trade_date), + {"trade_date": trade_date, "trigger": "dashboard-enrichment"}, + ) + + def _refresh_ifind_event_enrichment(self, trade_date: str) -> None: + if not self._ifind_event_lock.acquire(blocking=False): + return + try: + if self.database.get_data_snapshot("ifind_event_enrichment_v1", trade_date): + return + ifind = getattr(self, "ifind", None) + if not ifind or not ifind.configured: + return + current = datetime.strptime(trade_date, "%Y%m%d") + display_date = f"{current.year}年{current.month}月{current.day}日" + requests = { + "limits": ( + f"{display_date}涨停股票，股票代码、股票简称、涨停原因、" + "首次涨停时间、最终涨停时间、开板次数" + ), + "broken": ( + f"{display_date}曾涨停但收盘未涨停的股票，股票代码、股票简称、" + "涨停原因、首次涨停时间、开板次数" + ), + "down_limits": ( + f"{display_date}跌停股票，股票代码、股票简称、跌停原因" + ), + } + result: dict[str, Any] = { + "trade_date": trade_date, + "generated_at": datetime.now().astimezone().isoformat(timespec="seconds"), + "limits": {}, "broken": {}, "down_limits": {}, "partial": False, + } + for kind, query in requests.items(): + try: + rows = ifind.wencai(query, "stock", cache_ttl=900) + except IfindError: + result["partial"] = True + continue + for raw in rows: + code = self._ifind_row_code(raw) + if not code: + continue + reason_tokens = ( + ("跌停原因", "风险线索", "原因") + if kind == "down_limits" + else ("涨停原因类别", "涨停原因", "触板逻辑", "原因") + ) + reason = str(self._ifind_field(raw, reason_tokens) or "").strip() + first_time = self._normalize_ifind_event_time( + self._ifind_field(raw, ("首次涨停时间", "首次触板时间", "首次封板时间")) + ) + last_time = self._normalize_ifind_event_time( + self._ifind_field(raw, ("最终涨停时间", "最后涨停时间", "最后封板时间")) + ) + open_times = self._ifind_field(raw, ("开板次数", "打开涨停次数")) + try: + open_count = max(0, int(float(open_times))) if open_times not in (None, "") else None + except (TypeError, ValueError): + open_count = None + result[kind][code] = { + "reason": reason, + "first_time": first_time, + "last_time": last_time, + "open_times": open_count, + } + if any(result[kind] for kind in ("limits", "broken", "down_limits")): + self.database.save_data_snapshot( + "ifind_event_enrichment_v1", trade_date, "ifind", result + ) + finally: + self._ifind_event_lock.release() + + @staticmethod + def _normalize_ifind_event_time(value: Any) -> str: + text = str(value or "").strip() + match = re.search(r"(?:^|\s)(\d{1,2}:\d{2}(?::\d{2})?)(?:$|\s)", text) + if not match: + match = re.search(r"(? None: + for kind in ("limits", "broken", "down_limits"): + records = enrichment.get(kind) or {} + for row in dashboard.get(kind) or []: + event = records.get(str(row.get("code") or "")) or {} + reason = str(event.get("reason") or "").strip() + if reason: + row["reason"] = reason + row["reason_source"] = "market_event" + if event.get("first_time"): + row["first_time"] = event["first_time"] + if event.get("last_time"): + row["last_time"] = event["last_time"] + if event.get("open_times") is not None: + row["open_times"] = event["open_times"] diff --git a/backend/features/sentiment/__init__.py b/backend/features/sentiment/__init__.py new file mode 100644 index 0000000..5a5f280 --- /dev/null +++ b/backend/features/sentiment/__init__.py @@ -0,0 +1,19 @@ +"""Market sentiment cycle and history feature.""" + +from .engine import ( + COMPONENT_WEIGHTS, + SENTIMENT_ENGINE_VERSION, + apply_sentiment_to_dashboard, + build_sentiment_history, + latest_contiguous_history, +) +from .service import SentimentServiceMixin + +__all__ = [ + "COMPONENT_WEIGHTS", + "SENTIMENT_ENGINE_VERSION", + "SentimentServiceMixin", + "apply_sentiment_to_dashboard", + "build_sentiment_history", + "latest_contiguous_history", +] diff --git a/backend/features/sentiment/engine.py b/backend/features/sentiment/engine.py new file mode 100644 index 0000000..9345cd5 --- /dev/null +++ b/backend/features/sentiment/engine.py @@ -0,0 +1,496 @@ +from __future__ import annotations + +from copy import deepcopy +from statistics import mean, median +from typing import Any + + +COMPONENT_WEIGHTS = { + "breadth": 20, + "limit_ecology": 25, + "profit_effect": 30, + "ladder_structure": 15, + "liquidity": 10, +} + +SENTIMENT_ENGINE_VERSION = 2 + + +def _number(value: Any, default: float = 0.0) -> float: + try: + number = float(value) + return number if number == number else default + except (TypeError, ValueError): + return default + + +def _clamp(value: float, lower: float = 0.0, upper: float = 100.0) -> float: + return min(upper, max(lower, value)) + + +def _linear(value: float, low: float, high: float) -> float: + if high <= low: + return 50.0 + return _clamp((value - low) / (high - low) * 100) + + +def _percentile(value: float, history: list[float]) -> float: + if not history: + return 50.0 + below = sum(item < value for item in history) + equal = sum(item == value for item in history) + return _clamp((below + equal * 0.5) / len(history) * 100) + + +def _adaptive_score(value: float, fixed: float, history: list[float]) -> float: + if len(history) < 20: + return fixed + return fixed * 0.25 + _percentile(value, history[-250:]) * 0.75 + + +def _trade_date(payload: dict[str, Any]) -> str: + meta = payload.get("meta") or {} + return str(meta.get("trade_date") or payload.get("_snapshot_date") or "").replace("-", "") + + +def _deduplicate_snapshots(snapshots: list[dict[str, Any]]) -> list[dict[str, Any]]: + by_trade_date: dict[str, dict[str, Any]] = {} + for payload in snapshots: + trade_date = _trade_date(payload) + if trade_date: + by_trade_date[trade_date] = payload + return [by_trade_date[key] for key in sorted(by_trade_date)] + + +def _snapshot_stats(payload: dict[str, Any]) -> dict[str, Any]: + overview = payload.get("overview") or {} + meta = payload.get("meta") or {} + limits = list(payload.get("limits") or []) + broken = list(payload.get("broken") or []) + down_limits = list(payload.get("down_limits") or []) + yesterday = list(payload.get("yesterday_limits") or []) + + limit_up = len(limits) if limits else int(_number(overview.get("limit_up_count"))) + broken_count = len(broken) if broken else int(_number(overview.get("broken_count"))) + limit_down = len(down_limits) if down_limits else int(_number(overview.get("limit_down_count"))) + streaks = [max(1, int(_number(row.get("streak"), 1))) for row in limits] + first_board = sum(streak == 1 for streak in streaks) + second_board = sum(streak == 2 for streak in streaks) + three_plus = sum(streak >= 3 for streak in streaks) + max_height = max(streaks, default=0) + present_levels = set(streaks) + ladder_completeness = ( + sum(level in present_levels for level in range(1, max_height + 1)) / max_height * 100 + if max_height else 0.0 + ) + + up_count = int(_number(overview.get("up_count"))) + down_count = int(_number(overview.get("down_count"))) + flat_count = int(_number(overview.get("flat_count"))) + active_count = up_count + down_count + breadth_ratio = up_count / max(active_count, 1) * 100 + seal_rate = _number(overview.get("seal_rate")) + if not seal_rate and limit_up + broken_count: + seal_rate = limit_up / (limit_up + broken_count) * 100 + + previous_limit_count = len(yesterday) + previous_positive_count = sum(_number(row.get("current_change")) > 0 for row in yesterday) + previous_positive_rate = previous_positive_count / max(previous_limit_count, 1) * 100 + advanced_count = sum(row.get("outcome") == "晋级" for row in yesterday) + advance_rate = advanced_count / max(previous_limit_count, 1) * 100 + average_previous_change = ( + mean(_number(row.get("current_change")) for row in yesterday) if yesterday else 0.0 + ) + median_previous_change = ( + median(_number(row.get("current_change")) for row in yesterday) if yesterday else 0.0 + ) + severe_loss_count = sum(_number(row.get("current_change")) <= -5 for row in yesterday) + severe_loss_rate = severe_loss_count / max(previous_limit_count, 1) * 100 + previous_down_count = sum(row.get("outcome") == "跌停" for row in yesterday) + high_previous = [row for row in yesterday if int(_number(row.get("prior_streak"), 1)) >= 2] + high_positive_rate = ( + sum(_number(row.get("current_change")) > 0 for row in high_previous) + / max(len(high_previous), 1) + * 100 + ) + + amount_billion = _number(overview.get("amount_billion")) + limit_amount_billion = sum(_number(row.get("amount_billion")) for row in limits) + return { + "trade_date": _trade_date(payload), + "previous_trade_date": str(meta.get("previous_trade_date") or "").replace("-", ""), + "up_count": up_count, + "down_count": down_count, + "flat_count": flat_count, + "breadth_ratio": round(breadth_ratio, 1), + "limit_up_count": limit_up, + "first_board_count": first_board, + "second_board_count": second_board, + "three_plus_count": three_plus, + "max_height": max_height, + "ladder_completeness": round(ladder_completeness, 1), + "broken_count": broken_count, + "limit_down_count": limit_down, + "seal_rate": round(seal_rate, 1), + "previous_limit_count": previous_limit_count, + "previous_positive_count": previous_positive_count, + "previous_positive_rate": round(previous_positive_rate, 1), + "advance_rate": round(advance_rate, 1), + "average_previous_change": round(average_previous_change, 2), + "median_previous_change": round(median_previous_change, 2), + "severe_loss_count": severe_loss_count, + "severe_loss_rate": round(severe_loss_rate, 1), + "previous_down_count": previous_down_count, + "high_positive_rate": round(high_positive_rate, 1), + "amount_billion": round(amount_billion, 1), + "limit_amount_billion": round(limit_amount_billion, 2), + } + + +def _sentiment_label(score: float) -> str: + if score >= 80: + return "情绪高涨" + if score >= 60: + return "情绪偏强" + if score >= 40: + return "情绪中性" + if score >= 20: + return "情绪偏弱" + return "情绪冰点" + + +def _phase_signal(score: float, momentum: float, profit_score: float) -> str: + if score < 25: + return "修复" if momentum > 3 else "冰点" + if score < 45: + return "修复" if momentum > 3 else "退潮" + if score >= 80: + return "高潮" if momentum >= -2 and profit_score >= 60 else "分化" + if score >= 65: + return "分化" if momentum < -3 or profit_score < 50 else "发酵" + if momentum < -5: + return "退潮" + return "发酵" if momentum >= 0 and profit_score >= 45 else "分化" + + +def _confirmed_phase( + previous: dict[str, Any] | None, + score: float, + day_change: float, + systemic_health: float, + profit_score: float, + ecology_score: float, + phase_signal: str, + extreme_ice: bool, + fermentation_signal_count: int, +) -> tuple[str, str]: + if previous is None: + return phase_signal, "首个连续交易日，采用原始阶段信号" + previous_phase = str(previous.get("phase") or phase_signal) + if extreme_ice: + return "冰点", "市场宽度与跌停数量触发极端冰点" + + recovery = day_change >= 6 and score >= 25 and systemic_health >= 24 + fermentation_confirmed = fermentation_signal_count >= 2 + climax_ready = ( + score >= 80 + and profit_score >= 60 + and systemic_health >= 60 + and ecology_score >= 70 + ) + + if previous_phase == "冰点": + return ("修复", "冰点后首次有效回升") if recovery else ("冰点", "冰点尚未形成有效修复") + + if previous_phase == "退潮": + if score < 25: + return "冰点", "退潮继续下探至冰点区间" + return ("修复", "退潮后出现有效回升") if recovery else ("退潮", "退潮尚未形成有效修复") + + if previous_phase == "修复": + if score < 25: + return "冰点", "修复失败并重新跌入冰点区间" + if day_change <= -6 and score < 45: + return "退潮", "修复失败且温度显著回落" + if fermentation_confirmed: + return "发酵", "发酵条件连续两个交易日成立" + return "修复", "修复延续，等待发酵确认" + + if previous_phase == "发酵": + if score < 25: + return "冰点", "发酵阶段出现极端情绪坍塌" + if score < 45 and (day_change < 0 or systemic_health < 35): + return "退潮", "发酵阶段温度与系统健康度同步转弱" + if climax_ready: + return "高潮", "温度、赚钱效应与涨停生态共同达到高潮条件" + if phase_signal in {"分化", "退潮"} or day_change <= -6: + return "分化", "发酵阶段出现降温或赚钱效应弱化" + return "发酵", "发酵状态延续" + + if previous_phase == "高潮": + if score < 25: + return "冰点", "高潮后出现极端情绪坍塌" + if climax_ready: + return "高潮", "高潮条件继续成立" + if score < 45 or systemic_health < 30: + return "退潮", "高潮后风险快速释放" + return "分化", "高潮条件消退，进入分化" + + if previous_phase == "分化": + if score < 25: + return "冰点", "分化继续恶化至冰点区间" + if score < 45 or systemic_health < 30: + return "退潮", "分化后温度或系统健康度继续下降" + if fermentation_confirmed: + return "发酵", "分化转强条件连续两个交易日成立" + return "分化", "分化延续，等待方向确认" + + return phase_signal, "采用原始阶段信号" + + +def build_sentiment_history(snapshots: list[dict[str, Any]]) -> list[dict[str, Any]]: + payloads = _deduplicate_snapshots(snapshots) + raw_rows = [_snapshot_stats(payload) for payload in payloads] + results: list[dict[str, Any]] = [] + + for index, stats in enumerate(raw_rows): + previous = raw_rows[:index] + limit_history = [float(row["limit_up_count"]) for row in previous] + down_limit_history = [float(row["limit_down_count"]) for row in previous] + height_history = [float(row["max_height"]) for row in previous] + three_plus_history = [float(row["three_plus_count"]) for row in previous] + amount_history = [float(row["amount_billion"]) for row in previous[-20:] if row["amount_billion"]] + + breadth_score = _clamp(float(stats["breadth_ratio"])) + limit_strength = _adaptive_score( + float(stats["limit_up_count"]), + _linear(float(stats["limit_up_count"]), 10, 100), + limit_history, + ) + down_relief = 100 - _adaptive_score( + float(stats["limit_down_count"]), + _linear(float(stats["limit_down_count"]), 0, 50), + down_limit_history, + ) + seal_quality = _linear(float(stats["seal_rate"]), 35, 90) + systemic_health = breadth_score * 0.60 + down_relief * 0.40 + systemic_gate = 1.0 if systemic_health >= 35 else 0.35 + systemic_health / 35 * 0.65 + ecology_base_score = limit_strength * 0.35 + seal_quality * 0.35 + down_relief * 0.30 + # Systemic risk is applied once to the final temperature. Reapplying it here + # would count market breadth and limit-down pressure twice. + limit_ecology_score = ecology_base_score + + if stats["previous_limit_count"]: + positive_score = float(stats["previous_positive_rate"]) + average_change_score = _clamp(50 + float(stats["average_previous_change"]) * 6) + median_change_score = _clamp(50 + float(stats["median_previous_change"]) * 7) + advance_score = _clamp(float(stats["advance_rate"]) * 2.5) + severe_loss_safety = _clamp(100 - float(stats["severe_loss_rate"]) * 3) + down_safety = _clamp(100 - float(stats["previous_down_count"]) / stats["previous_limit_count"] * 700) + tail_safety_score = severe_loss_safety * 0.70 + down_safety * 0.30 + profit_effect_score = ( + positive_score * 0.30 + + median_change_score * 0.25 + + average_change_score * 0.10 + + advance_score * 0.20 + + tail_safety_score * 0.15 + ) + else: + profit_effect_score = 50.0 + + max_height_score = _adaptive_score( + float(stats["max_height"]), + _linear(float(stats["max_height"]), 1, 7), + height_history, + ) + continuation_rate = ( + (float(stats["second_board_count"]) + float(stats["three_plus_count"])) + / max(float(stats["limit_up_count"]), 1) + * 100 + ) + three_plus_density = float(stats["three_plus_count"]) / max(float(stats["limit_up_count"]), 1) * 100 + three_plus_score = _adaptive_score( + float(stats["three_plus_count"]), + _clamp(three_plus_density * 5), + three_plus_history, + ) + ladder_structure_score = ( + max_height_score * 0.30 + + _clamp(continuation_rate * 3) * 0.25 + + three_plus_score * 0.25 + + float(stats["ladder_completeness"]) * 0.20 + ) + + amount_baseline = mean(amount_history) if amount_history else float(stats["amount_billion"] or 1) + amount_ratio = float(stats["amount_billion"]) / max(amount_baseline, 1) + amount_score = _clamp(50 + (amount_ratio - 1) * 100) + limit_amount_share = float(stats["limit_amount_billion"]) / max(float(stats["amount_billion"]), 1) * 100 + liquidity_score = amount_score * 0.70 + _clamp(limit_amount_share * 20) * 0.30 + + component_scores = { + "breadth": breadth_score, + "limit_ecology": limit_ecology_score, + "profit_effect": profit_effect_score, + "ladder_structure": ladder_structure_score, + "liquidity": liquidity_score, + } + raw_score = sum(component_scores[key] * weight / 100 for key, weight in COMPONENT_WEIGHTS.items()) + score = round( + raw_score * systemic_gate + ) + extreme_ice = float(stats["breadth_ratio"]) <= 15 and float(stats["limit_down_count"]) >= 100 + if extreme_ice: + score = min(score, 15) + elif float(stats["breadth_ratio"]) <= 25 and float(stats["limit_down_count"]) >= 50: + score = min(score, 24) + previous_scores: list[float] = [] + expected_date = str(stats.get("previous_trade_date") or "") + for prior_result in reversed(results): + if not expected_date or str(prior_result.get("trade_date") or "") != expected_date: + break + previous_scores.append(float(prior_result["score"])) + expected_date = str(prior_result.get("previous_trade_date") or "") + if len(previous_scores) == 3: + break + momentum = score - mean(previous_scores) if previous_scores else 0.0 + direction = "升温" if momentum > 3 else "降温" if momentum < -3 else "持平" + normalization = "历史百分位" if len(previous) >= 20 else "固定锚点" + previous_result = ( + results[-1] + if results and str(stats.get("previous_trade_date") or "") == str(results[-1].get("trade_date") or "") + else None + ) + day_change = score - float(previous_result["score"]) if previous_result else 0.0 + ema_score = round( + score if not previous_result + else score * 0.5 + float(previous_result.get("ema_score", previous_result["score"])) * 0.5, + 1, + ) + phase_signal = _phase_signal(score, momentum, profit_effect_score) + fermentation_ready = ( + phase_signal == "发酵" + and score >= 45 + and profit_effect_score >= 45 + and systemic_health >= 35 + and not extreme_ice + ) + previous_fermentation_count = int(previous_result.get("fermentation_signal_count") or 0) if previous_result else 0 + fermentation_signal_count = previous_fermentation_count + 1 if fermentation_ready else 0 + phase, transition_reason = _confirmed_phase( + previous_result, + score, + day_change, + systemic_health, + profit_effect_score, + limit_ecology_score, + phase_signal, + extreme_ice, + fermentation_signal_count, + ) + previous_phase = str(previous_result.get("phase") or "") if previous_result else "" + if phase not in {"修复", "分化"}: + fermentation_signal_count = 0 + elif phase == "分化" and previous_phase != "分化": + fermentation_signal_count = 0 + + components = { + "breadth": { + "label": "市场宽度", + "score": round(breadth_score, 1), + "weight": COMPONENT_WEIGHTS["breadth"], + "summary": f"上涨占比 {stats['breadth_ratio']:.1f}%", + }, + "limit_ecology": { + "label": "涨停生态", + "score": round(limit_ecology_score, 1), + "weight": COMPONENT_WEIGHTS["limit_ecology"], + "summary": ( + f"涨停 {stats['limit_up_count']} · 跌停 {stats['limit_down_count']} · " + f"封板 {stats['seal_rate']:.1f}%" + ), + }, + "profit_effect": { + "label": "赚钱效应", + "score": round(profit_effect_score, 1), + "weight": COMPONENT_WEIGHTS["profit_effect"], + "summary": ( + f"昨涨停红盘 {stats['previous_positive_rate']:.1f}% · " + f"中位 {stats['median_previous_change']:+.2f}% · " + f"重亏 {stats['severe_loss_rate']:.1f}%" + if stats["previous_limit_count"] else "缺少前一交易日样本" + ), + }, + "ladder_structure": { + "label": "连板结构", + "score": round(ladder_structure_score, 1), + "weight": COMPONENT_WEIGHTS["ladder_structure"], + "summary": f"最高 {stats['max_height']} 板 · 三板以上 {stats['three_plus_count']} 家", + }, + "liquidity": { + "label": "成交活跃度", + "score": round(liquidity_score, 1), + "weight": COMPONENT_WEIGHTS["liquidity"], + "summary": f"成交 {stats['amount_billion']:.1f} 亿 · 均值比 {amount_ratio:.2f}", + }, + } + results.append( + { + **stats, + "score": score, + "ema_score": ema_score, + "label": _sentiment_label(score), + "phase": phase, + "phase_signal": phase_signal, + "transition_reason": transition_reason, + "fermentation_signal_count": fermentation_signal_count, + "day_change": round(day_change, 1), + "direction": direction, + "momentum": round(momentum, 1), + "normalization": "250日历史百分位" if len(previous) >= 20 else normalization, + "history_days": len(previous) + 1, + "systemic_health": round(systemic_health, 1), + "risk_multiplier": round(systemic_gate, 3), + "components": components, + } + ) + return results + + +def latest_contiguous_history(series: list[dict[str, Any]]) -> list[dict[str, Any]]: + if not series: + return [] + contiguous = [series[-1]] + for row in reversed(series[:-1]): + expected_previous = str(contiguous[0].get("previous_trade_date") or "") + if not expected_previous or expected_previous != str(row.get("trade_date") or ""): + break + contiguous.insert(0, row) + return contiguous + + +def apply_sentiment_to_dashboard( + dashboard: dict[str, Any], + historical_snapshots: list[dict[str, Any]] | None = None, +) -> dict[str, Any]: + result = deepcopy(dashboard) + history = list(historical_snapshots or []) + history.append(result) + series = build_sentiment_history(history) + target_date = _trade_date(result) + sentiment = next((row for row in reversed(series) if row["trade_date"] == target_date), None) + if not sentiment: + return result + overview = dict(result.get("overview") or {}) + overview.update( + { + "sentiment_score": sentiment["score"], + "sentiment_trend_score": sentiment["ema_score"], + "sentiment_label": sentiment["label"], + "sentiment_phase": sentiment["phase"], + "sentiment_direction": sentiment["direction"], + "sentiment_components": sentiment["components"], + "sentiment_engine_version": SENTIMENT_ENGINE_VERSION, + } + ) + result["overview"] = overview + return result diff --git a/backend/features/sentiment/service.py b/backend/features/sentiment/service.py new file mode 100644 index 0000000..0947d4c --- /dev/null +++ b/backend/features/sentiment/service.py @@ -0,0 +1,39 @@ +from __future__ import annotations + +from typing import Any + +from backend.bootstrap.config import normalize_date +from backend.features.sentiment.engine import ( + COMPONENT_WEIGHTS, + apply_sentiment_to_dashboard, + build_sentiment_history, + latest_contiguous_history, +) + + +class SentimentServiceMixin: + def _enrich_dashboard_sentiment( + self, + dashboard: dict[str, Any], + end_date: str, + ) -> dict[str, Any]: + history = self.database.list_snapshot_payloads(end_date, 260) + return apply_sentiment_to_dashboard(dashboard, history) + + def sentiment_history(self, trade_date: str, limit: int = 20) -> dict[str, Any]: + normalized_date = normalize_date(trade_date) + limit = max(10, min(120, int(limit))) + full_series = build_sentiment_history( + self.database.list_snapshot_payloads(normalized_date, 240) + ) + series = latest_contiguous_history(full_series) + rows = series[-limit:] + return { + "trade_date": rows[-1]["trade_date"] if rows else normalized_date, + "available_days": len(series), + "stored_days": len(full_series), + "requested_days": limit, + "rows": rows, + "weights": COMPONENT_WEIGHTS, + "normalization": rows[-1]["normalization"] if rows else "固定锚点", + } diff --git a/database.py b/database.py index 9a3b60b..8fe4266 100644 --- a/database.py +++ b/database.py @@ -9,6 +9,7 @@ from typing import Any from backend.database import MIGRATIONS, MigrationRunner, SQLiteConnectionFactory from backend.features.accounts.repository import AccountRepositoryMixin from backend.features.market.repository import MarketRepositoryMixin +from backend.features.pools.repository import PoolRepositoryMixin from backend.features.system.repository import SystemSettingsRepositoryMixin @@ -24,6 +25,7 @@ def _optional_float(value: Any) -> float | None: class ReviewDatabase( AccountRepositoryMixin, MarketRepositoryMixin, + PoolRepositoryMixin, SystemSettingsRepositoryMixin, ): def __init__(self, path: Path) -> None: @@ -836,27 +838,6 @@ class ReviewDatabase( ) return cursor.rowcount > 0 - def save_reason_override(self, trade_date: str, code: str, reason: str) -> None: - now = datetime.now().astimezone().isoformat(timespec="seconds") - with self.connect() as connection: - connection.execute( - """ - INSERT INTO reason_overrides (trade_date, code, reason, updated_at) - VALUES (?, ?, ?, ?) - ON CONFLICT(trade_date, code) DO UPDATE SET - reason = excluded.reason, - updated_at = excluded.updated_at - """, - (trade_date, code, reason, now), - ) - - def reason_overrides(self, trade_date: str) -> dict[str, str]: - with self.connect() as connection: - rows = connection.execute( - "SELECT code, reason FROM reason_overrides WHERE trade_date = ?", - (trade_date,), - ).fetchall() - return {row["code"]: row["reason"] for row in rows} def list_seat_aliases(self) -> dict[str, str]: with self.connect() as connection: diff --git a/demo_data.py b/demo_data.py index 9bf51a6..e1165b8 100644 --- a/demo_data.py +++ b/demo_data.py @@ -5,7 +5,7 @@ import math from datetime import datetime, timedelta from typing import Any -from sentiment_engine import apply_sentiment_to_dashboard +from backend.features.sentiment.engine import apply_sentiment_to_dashboard DEMO_LIMITS = [ diff --git a/screener.py b/screener.py index 1ee940e..d0796db 100644 --- a/screener.py +++ b/screener.py @@ -10,7 +10,7 @@ from typing import Any from advanced_strategies import ADVANCED_CURATED_STRATEGIES from database import ReviewDatabase -from sentiment_engine import build_sentiment_history, latest_contiguous_history +from backend.features.sentiment.engine import build_sentiment_history, latest_contiguous_history from tushare_client import TushareClient, TushareError diff --git a/sentiment_engine.py b/sentiment_engine.py index 9345cd5..0d462ac 100644 --- a/sentiment_engine.py +++ b/sentiment_engine.py @@ -1,496 +1,7 @@ -from __future__ import annotations +"""Compatibility alias for the canonical sentiment engine implementation.""" -from copy import deepcopy -from statistics import mean, median -from typing import Any +import sys +from backend.features.sentiment import engine as _implementation -COMPONENT_WEIGHTS = { - "breadth": 20, - "limit_ecology": 25, - "profit_effect": 30, - "ladder_structure": 15, - "liquidity": 10, -} - -SENTIMENT_ENGINE_VERSION = 2 - - -def _number(value: Any, default: float = 0.0) -> float: - try: - number = float(value) - return number if number == number else default - except (TypeError, ValueError): - return default - - -def _clamp(value: float, lower: float = 0.0, upper: float = 100.0) -> float: - return min(upper, max(lower, value)) - - -def _linear(value: float, low: float, high: float) -> float: - if high <= low: - return 50.0 - return _clamp((value - low) / (high - low) * 100) - - -def _percentile(value: float, history: list[float]) -> float: - if not history: - return 50.0 - below = sum(item < value for item in history) - equal = sum(item == value for item in history) - return _clamp((below + equal * 0.5) / len(history) * 100) - - -def _adaptive_score(value: float, fixed: float, history: list[float]) -> float: - if len(history) < 20: - return fixed - return fixed * 0.25 + _percentile(value, history[-250:]) * 0.75 - - -def _trade_date(payload: dict[str, Any]) -> str: - meta = payload.get("meta") or {} - return str(meta.get("trade_date") or payload.get("_snapshot_date") or "").replace("-", "") - - -def _deduplicate_snapshots(snapshots: list[dict[str, Any]]) -> list[dict[str, Any]]: - by_trade_date: dict[str, dict[str, Any]] = {} - for payload in snapshots: - trade_date = _trade_date(payload) - if trade_date: - by_trade_date[trade_date] = payload - return [by_trade_date[key] for key in sorted(by_trade_date)] - - -def _snapshot_stats(payload: dict[str, Any]) -> dict[str, Any]: - overview = payload.get("overview") or {} - meta = payload.get("meta") or {} - limits = list(payload.get("limits") or []) - broken = list(payload.get("broken") or []) - down_limits = list(payload.get("down_limits") or []) - yesterday = list(payload.get("yesterday_limits") or []) - - limit_up = len(limits) if limits else int(_number(overview.get("limit_up_count"))) - broken_count = len(broken) if broken else int(_number(overview.get("broken_count"))) - limit_down = len(down_limits) if down_limits else int(_number(overview.get("limit_down_count"))) - streaks = [max(1, int(_number(row.get("streak"), 1))) for row in limits] - first_board = sum(streak == 1 for streak in streaks) - second_board = sum(streak == 2 for streak in streaks) - three_plus = sum(streak >= 3 for streak in streaks) - max_height = max(streaks, default=0) - present_levels = set(streaks) - ladder_completeness = ( - sum(level in present_levels for level in range(1, max_height + 1)) / max_height * 100 - if max_height else 0.0 - ) - - up_count = int(_number(overview.get("up_count"))) - down_count = int(_number(overview.get("down_count"))) - flat_count = int(_number(overview.get("flat_count"))) - active_count = up_count + down_count - breadth_ratio = up_count / max(active_count, 1) * 100 - seal_rate = _number(overview.get("seal_rate")) - if not seal_rate and limit_up + broken_count: - seal_rate = limit_up / (limit_up + broken_count) * 100 - - previous_limit_count = len(yesterday) - previous_positive_count = sum(_number(row.get("current_change")) > 0 for row in yesterday) - previous_positive_rate = previous_positive_count / max(previous_limit_count, 1) * 100 - advanced_count = sum(row.get("outcome") == "晋级" for row in yesterday) - advance_rate = advanced_count / max(previous_limit_count, 1) * 100 - average_previous_change = ( - mean(_number(row.get("current_change")) for row in yesterday) if yesterday else 0.0 - ) - median_previous_change = ( - median(_number(row.get("current_change")) for row in yesterday) if yesterday else 0.0 - ) - severe_loss_count = sum(_number(row.get("current_change")) <= -5 for row in yesterday) - severe_loss_rate = severe_loss_count / max(previous_limit_count, 1) * 100 - previous_down_count = sum(row.get("outcome") == "跌停" for row in yesterday) - high_previous = [row for row in yesterday if int(_number(row.get("prior_streak"), 1)) >= 2] - high_positive_rate = ( - sum(_number(row.get("current_change")) > 0 for row in high_previous) - / max(len(high_previous), 1) - * 100 - ) - - amount_billion = _number(overview.get("amount_billion")) - limit_amount_billion = sum(_number(row.get("amount_billion")) for row in limits) - return { - "trade_date": _trade_date(payload), - "previous_trade_date": str(meta.get("previous_trade_date") or "").replace("-", ""), - "up_count": up_count, - "down_count": down_count, - "flat_count": flat_count, - "breadth_ratio": round(breadth_ratio, 1), - "limit_up_count": limit_up, - "first_board_count": first_board, - "second_board_count": second_board, - "three_plus_count": three_plus, - "max_height": max_height, - "ladder_completeness": round(ladder_completeness, 1), - "broken_count": broken_count, - "limit_down_count": limit_down, - "seal_rate": round(seal_rate, 1), - "previous_limit_count": previous_limit_count, - "previous_positive_count": previous_positive_count, - "previous_positive_rate": round(previous_positive_rate, 1), - "advance_rate": round(advance_rate, 1), - "average_previous_change": round(average_previous_change, 2), - "median_previous_change": round(median_previous_change, 2), - "severe_loss_count": severe_loss_count, - "severe_loss_rate": round(severe_loss_rate, 1), - "previous_down_count": previous_down_count, - "high_positive_rate": round(high_positive_rate, 1), - "amount_billion": round(amount_billion, 1), - "limit_amount_billion": round(limit_amount_billion, 2), - } - - -def _sentiment_label(score: float) -> str: - if score >= 80: - return "情绪高涨" - if score >= 60: - return "情绪偏强" - if score >= 40: - return "情绪中性" - if score >= 20: - return "情绪偏弱" - return "情绪冰点" - - -def _phase_signal(score: float, momentum: float, profit_score: float) -> str: - if score < 25: - return "修复" if momentum > 3 else "冰点" - if score < 45: - return "修复" if momentum > 3 else "退潮" - if score >= 80: - return "高潮" if momentum >= -2 and profit_score >= 60 else "分化" - if score >= 65: - return "分化" if momentum < -3 or profit_score < 50 else "发酵" - if momentum < -5: - return "退潮" - return "发酵" if momentum >= 0 and profit_score >= 45 else "分化" - - -def _confirmed_phase( - previous: dict[str, Any] | None, - score: float, - day_change: float, - systemic_health: float, - profit_score: float, - ecology_score: float, - phase_signal: str, - extreme_ice: bool, - fermentation_signal_count: int, -) -> tuple[str, str]: - if previous is None: - return phase_signal, "首个连续交易日，采用原始阶段信号" - previous_phase = str(previous.get("phase") or phase_signal) - if extreme_ice: - return "冰点", "市场宽度与跌停数量触发极端冰点" - - recovery = day_change >= 6 and score >= 25 and systemic_health >= 24 - fermentation_confirmed = fermentation_signal_count >= 2 - climax_ready = ( - score >= 80 - and profit_score >= 60 - and systemic_health >= 60 - and ecology_score >= 70 - ) - - if previous_phase == "冰点": - return ("修复", "冰点后首次有效回升") if recovery else ("冰点", "冰点尚未形成有效修复") - - if previous_phase == "退潮": - if score < 25: - return "冰点", "退潮继续下探至冰点区间" - return ("修复", "退潮后出现有效回升") if recovery else ("退潮", "退潮尚未形成有效修复") - - if previous_phase == "修复": - if score < 25: - return "冰点", "修复失败并重新跌入冰点区间" - if day_change <= -6 and score < 45: - return "退潮", "修复失败且温度显著回落" - if fermentation_confirmed: - return "发酵", "发酵条件连续两个交易日成立" - return "修复", "修复延续，等待发酵确认" - - if previous_phase == "发酵": - if score < 25: - return "冰点", "发酵阶段出现极端情绪坍塌" - if score < 45 and (day_change < 0 or systemic_health < 35): - return "退潮", "发酵阶段温度与系统健康度同步转弱" - if climax_ready: - return "高潮", "温度、赚钱效应与涨停生态共同达到高潮条件" - if phase_signal in {"分化", "退潮"} or day_change <= -6: - return "分化", "发酵阶段出现降温或赚钱效应弱化" - return "发酵", "发酵状态延续" - - if previous_phase == "高潮": - if score < 25: - return "冰点", "高潮后出现极端情绪坍塌" - if climax_ready: - return "高潮", "高潮条件继续成立" - if score < 45 or systemic_health < 30: - return "退潮", "高潮后风险快速释放" - return "分化", "高潮条件消退，进入分化" - - if previous_phase == "分化": - if score < 25: - return "冰点", "分化继续恶化至冰点区间" - if score < 45 or systemic_health < 30: - return "退潮", "分化后温度或系统健康度继续下降" - if fermentation_confirmed: - return "发酵", "分化转强条件连续两个交易日成立" - return "分化", "分化延续，等待方向确认" - - return phase_signal, "采用原始阶段信号" - - -def build_sentiment_history(snapshots: list[dict[str, Any]]) -> list[dict[str, Any]]: - payloads = _deduplicate_snapshots(snapshots) - raw_rows = [_snapshot_stats(payload) for payload in payloads] - results: list[dict[str, Any]] = [] - - for index, stats in enumerate(raw_rows): - previous = raw_rows[:index] - limit_history = [float(row["limit_up_count"]) for row in previous] - down_limit_history = [float(row["limit_down_count"]) for row in previous] - height_history = [float(row["max_height"]) for row in previous] - three_plus_history = [float(row["three_plus_count"]) for row in previous] - amount_history = [float(row["amount_billion"]) for row in previous[-20:] if row["amount_billion"]] - - breadth_score = _clamp(float(stats["breadth_ratio"])) - limit_strength = _adaptive_score( - float(stats["limit_up_count"]), - _linear(float(stats["limit_up_count"]), 10, 100), - limit_history, - ) - down_relief = 100 - _adaptive_score( - float(stats["limit_down_count"]), - _linear(float(stats["limit_down_count"]), 0, 50), - down_limit_history, - ) - seal_quality = _linear(float(stats["seal_rate"]), 35, 90) - systemic_health = breadth_score * 0.60 + down_relief * 0.40 - systemic_gate = 1.0 if systemic_health >= 35 else 0.35 + systemic_health / 35 * 0.65 - ecology_base_score = limit_strength * 0.35 + seal_quality * 0.35 + down_relief * 0.30 - # Systemic risk is applied once to the final temperature. Reapplying it here - # would count market breadth and limit-down pressure twice. - limit_ecology_score = ecology_base_score - - if stats["previous_limit_count"]: - positive_score = float(stats["previous_positive_rate"]) - average_change_score = _clamp(50 + float(stats["average_previous_change"]) * 6) - median_change_score = _clamp(50 + float(stats["median_previous_change"]) * 7) - advance_score = _clamp(float(stats["advance_rate"]) * 2.5) - severe_loss_safety = _clamp(100 - float(stats["severe_loss_rate"]) * 3) - down_safety = _clamp(100 - float(stats["previous_down_count"]) / stats["previous_limit_count"] * 700) - tail_safety_score = severe_loss_safety * 0.70 + down_safety * 0.30 - profit_effect_score = ( - positive_score * 0.30 - + median_change_score * 0.25 - + average_change_score * 0.10 - + advance_score * 0.20 - + tail_safety_score * 0.15 - ) - else: - profit_effect_score = 50.0 - - max_height_score = _adaptive_score( - float(stats["max_height"]), - _linear(float(stats["max_height"]), 1, 7), - height_history, - ) - continuation_rate = ( - (float(stats["second_board_count"]) + float(stats["three_plus_count"])) - / max(float(stats["limit_up_count"]), 1) - * 100 - ) - three_plus_density = float(stats["three_plus_count"]) / max(float(stats["limit_up_count"]), 1) * 100 - three_plus_score = _adaptive_score( - float(stats["three_plus_count"]), - _clamp(three_plus_density * 5), - three_plus_history, - ) - ladder_structure_score = ( - max_height_score * 0.30 - + _clamp(continuation_rate * 3) * 0.25 - + three_plus_score * 0.25 - + float(stats["ladder_completeness"]) * 0.20 - ) - - amount_baseline = mean(amount_history) if amount_history else float(stats["amount_billion"] or 1) - amount_ratio = float(stats["amount_billion"]) / max(amount_baseline, 1) - amount_score = _clamp(50 + (amount_ratio - 1) * 100) - limit_amount_share = float(stats["limit_amount_billion"]) / max(float(stats["amount_billion"]), 1) * 100 - liquidity_score = amount_score * 0.70 + _clamp(limit_amount_share * 20) * 0.30 - - component_scores = { - "breadth": breadth_score, - "limit_ecology": limit_ecology_score, - "profit_effect": profit_effect_score, - "ladder_structure": ladder_structure_score, - "liquidity": liquidity_score, - } - raw_score = sum(component_scores[key] * weight / 100 for key, weight in COMPONENT_WEIGHTS.items()) - score = round( - raw_score * systemic_gate - ) - extreme_ice = float(stats["breadth_ratio"]) <= 15 and float(stats["limit_down_count"]) >= 100 - if extreme_ice: - score = min(score, 15) - elif float(stats["breadth_ratio"]) <= 25 and float(stats["limit_down_count"]) >= 50: - score = min(score, 24) - previous_scores: list[float] = [] - expected_date = str(stats.get("previous_trade_date") or "") - for prior_result in reversed(results): - if not expected_date or str(prior_result.get("trade_date") or "") != expected_date: - break - previous_scores.append(float(prior_result["score"])) - expected_date = str(prior_result.get("previous_trade_date") or "") - if len(previous_scores) == 3: - break - momentum = score - mean(previous_scores) if previous_scores else 0.0 - direction = "升温" if momentum > 3 else "降温" if momentum < -3 else "持平" - normalization = "历史百分位" if len(previous) >= 20 else "固定锚点" - previous_result = ( - results[-1] - if results and str(stats.get("previous_trade_date") or "") == str(results[-1].get("trade_date") or "") - else None - ) - day_change = score - float(previous_result["score"]) if previous_result else 0.0 - ema_score = round( - score if not previous_result - else score * 0.5 + float(previous_result.get("ema_score", previous_result["score"])) * 0.5, - 1, - ) - phase_signal = _phase_signal(score, momentum, profit_effect_score) - fermentation_ready = ( - phase_signal == "发酵" - and score >= 45 - and profit_effect_score >= 45 - and systemic_health >= 35 - and not extreme_ice - ) - previous_fermentation_count = int(previous_result.get("fermentation_signal_count") or 0) if previous_result else 0 - fermentation_signal_count = previous_fermentation_count + 1 if fermentation_ready else 0 - phase, transition_reason = _confirmed_phase( - previous_result, - score, - day_change, - systemic_health, - profit_effect_score, - limit_ecology_score, - phase_signal, - extreme_ice, - fermentation_signal_count, - ) - previous_phase = str(previous_result.get("phase") or "") if previous_result else "" - if phase not in {"修复", "分化"}: - fermentation_signal_count = 0 - elif phase == "分化" and previous_phase != "分化": - fermentation_signal_count = 0 - - components = { - "breadth": { - "label": "市场宽度", - "score": round(breadth_score, 1), - "weight": COMPONENT_WEIGHTS["breadth"], - "summary": f"上涨占比 {stats['breadth_ratio']:.1f}%", - }, - "limit_ecology": { - "label": "涨停生态", - "score": round(limit_ecology_score, 1), - "weight": COMPONENT_WEIGHTS["limit_ecology"], - "summary": ( - f"涨停 {stats['limit_up_count']} · 跌停 {stats['limit_down_count']} · " - f"封板 {stats['seal_rate']:.1f}%" - ), - }, - "profit_effect": { - "label": "赚钱效应", - "score": round(profit_effect_score, 1), - "weight": COMPONENT_WEIGHTS["profit_effect"], - "summary": ( - f"昨涨停红盘 {stats['previous_positive_rate']:.1f}% · " - f"中位 {stats['median_previous_change']:+.2f}% · " - f"重亏 {stats['severe_loss_rate']:.1f}%" - if stats["previous_limit_count"] else "缺少前一交易日样本" - ), - }, - "ladder_structure": { - "label": "连板结构", - "score": round(ladder_structure_score, 1), - "weight": COMPONENT_WEIGHTS["ladder_structure"], - "summary": f"最高 {stats['max_height']} 板 · 三板以上 {stats['three_plus_count']} 家", - }, - "liquidity": { - "label": "成交活跃度", - "score": round(liquidity_score, 1), - "weight": COMPONENT_WEIGHTS["liquidity"], - "summary": f"成交 {stats['amount_billion']:.1f} 亿 · 均值比 {amount_ratio:.2f}", - }, - } - results.append( - { - **stats, - "score": score, - "ema_score": ema_score, - "label": _sentiment_label(score), - "phase": phase, - "phase_signal": phase_signal, - "transition_reason": transition_reason, - "fermentation_signal_count": fermentation_signal_count, - "day_change": round(day_change, 1), - "direction": direction, - "momentum": round(momentum, 1), - "normalization": "250日历史百分位" if len(previous) >= 20 else normalization, - "history_days": len(previous) + 1, - "systemic_health": round(systemic_health, 1), - "risk_multiplier": round(systemic_gate, 3), - "components": components, - } - ) - return results - - -def latest_contiguous_history(series: list[dict[str, Any]]) -> list[dict[str, Any]]: - if not series: - return [] - contiguous = [series[-1]] - for row in reversed(series[:-1]): - expected_previous = str(contiguous[0].get("previous_trade_date") or "") - if not expected_previous or expected_previous != str(row.get("trade_date") or ""): - break - contiguous.insert(0, row) - return contiguous - - -def apply_sentiment_to_dashboard( - dashboard: dict[str, Any], - historical_snapshots: list[dict[str, Any]] | None = None, -) -> dict[str, Any]: - result = deepcopy(dashboard) - history = list(historical_snapshots or []) - history.append(result) - series = build_sentiment_history(history) - target_date = _trade_date(result) - sentiment = next((row for row in reversed(series) if row["trade_date"] == target_date), None) - if not sentiment: - return result - overview = dict(result.get("overview") or {}) - overview.update( - { - "sentiment_score": sentiment["score"], - "sentiment_trend_score": sentiment["ema_score"], - "sentiment_label": sentiment["label"], - "sentiment_phase": sentiment["phase"], - "sentiment_direction": sentiment["direction"], - "sentiment_components": sentiment["components"], - "sentiment_engine_version": SENTIMENT_ENGINE_VERSION, - } - ) - result["overview"] = overview - return result +sys.modules[__name__] = _implementation diff --git a/tests/test_preservation_slice_market.py b/tests/test_preservation_slice_market.py index e0bbb29..d15d042 100644 --- a/tests/test_preservation_slice_market.py +++ b/tests/test_preservation_slice_market.py @@ -127,12 +127,15 @@ class MarketSliceSourceEquivalenceTests(unittest.TestCase): def test_provider_logic_is_the_original_implementation(self) -> None: exact_moves = ( - ("tushare_client.py", "backend/data/providers/tushare_client.py"), ("ifind_client.py", "backend/data/providers/ifind_client.py"), ("realtime_aggregator.py", "backend/data/realtime.py"), ) for original, migrated in exact_moves: self.assertEqual(sha256(ORIGINAL_ROOT / original), sha256(APP_ROOT / migrated)) + self.assertEqual( + top_level_definitions(ORIGINAL_ROOT / "tushare_client.py"), + top_level_definitions(APP_ROOT / "backend/data/providers/tushare_client.py"), + ) self.assertEqual( top_level_definitions(ORIGINAL_ROOT / "chart_data_provider.py"), top_level_definitions(APP_ROOT / "backend/features/market/charts.py"), diff --git a/tests/test_preservation_slice_sentiment_pools.py b/tests/test_preservation_slice_sentiment_pools.py new file mode 100644 index 0000000..69aec93 --- /dev/null +++ b/tests/test_preservation_slice_sentiment_pools.py @@ -0,0 +1,114 @@ +from __future__ import annotations + +import ast +import hashlib +import unittest +from pathlib import Path + +import sentiment_engine +from backend.features.sentiment import engine as canonical_engine + + +APP_ROOT = Path(__file__).resolve().parents[1] +ORIGINAL_ROOT = APP_ROOT.parent + +SENTIMENT_METHODS = { + "_enrich_dashboard_sentiment", + "sentiment_history", +} +POOL_METHODS = { + "save_reason", + "_apply_reason_overrides", + "_schedule_ifind_event_enrichment", + "_refresh_ifind_event_enrichment", + "_normalize_ifind_event_time", + "_merge_ifind_event_enrichment", +} +POOL_REPOSITORY_METHODS = { + "save_reason_override", + "reason_overrides", +} + + +def class_methods(path: Path, class_name: str) -> dict[str, str]: + tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path)) + owner = next( + node + for node in tree.body + if isinstance(node, ast.ClassDef) and node.name == class_name + ) + return { + node.name: ast.dump(node, include_attributes=False) + for node in owner.body + if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef)) + } + + +def sha256(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +class SentimentPoolSliceSourceEquivalenceTests(unittest.TestCase): + def test_sentiment_service_methods_are_exact_original_ast(self) -> None: + original = class_methods(ORIGINAL_ROOT / "server.py", "DashboardService") + migrated = class_methods( + APP_ROOT / "backend" / "features" / "sentiment" / "service.py", + "SentimentServiceMixin", + ) + self.assertEqual(set(migrated), SENTIMENT_METHODS) + for name in sorted(SENTIMENT_METHODS): + self.assertEqual(migrated[name], original[name], name) + + def test_pool_service_methods_are_exact_original_ast(self) -> None: + original = class_methods(ORIGINAL_ROOT / "server.py", "DashboardService") + migrated = class_methods( + APP_ROOT / "backend" / "features" / "pools" / "service.py", + "PoolServiceMixin", + ) + self.assertEqual(set(migrated), POOL_METHODS) + for name in sorted(POOL_METHODS): + self.assertEqual(migrated[name], original[name], name) + + def test_pool_repository_methods_are_exact_original_ast(self) -> None: + original = class_methods(ORIGINAL_ROOT / "database.py", "ReviewDatabase") + migrated = class_methods( + APP_ROOT / "backend" / "features" / "pools" / "repository.py", + "PoolRepositoryMixin", + ) + self.assertEqual(set(migrated), POOL_REPOSITORY_METHODS) + for name in sorted(POOL_REPOSITORY_METHODS): + self.assertEqual(migrated[name], original[name], name) + + def test_original_classes_no_longer_duplicate_moved_methods(self) -> None: + remaining_service = class_methods( + APP_ROOT / "backend" / "application.py", "DashboardService" + ) + remaining_database = class_methods(APP_ROOT / "database.py", "ReviewDatabase") + self.assertTrue((SENTIMENT_METHODS | POOL_METHODS).isdisjoint(remaining_service)) + self.assertTrue(POOL_REPOSITORY_METHODS.isdisjoint(remaining_database)) + + def test_sentiment_engine_is_exact_original_with_legacy_alias(self) -> None: + self.assertEqual( + sha256(ORIGINAL_ROOT / "sentiment_engine.py"), + sha256(APP_ROOT / "backend" / "features" / "sentiment" / "engine.py"), + ) + self.assertIs(sentiment_engine, canonical_engine) + + def test_api_and_frontend_assets_are_unchanged(self) -> None: + for relative in ( + "config/api.config.json", + "static/index.html", + "static/app.js", + "static/styles.css", + "static/pages/sentiment/page.js", + "static/pages/pools/page.js", + ): + self.assertEqual( + sha256(APP_ROOT / relative), + sha256(ORIGINAL_ROOT / relative), + relative, + ) + + +if __name__ == "__main__": + unittest.main()