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