Compare commits

..
49 changed files with 3774 additions and 2603 deletions
+17 -626
View File
@@ -42,7 +42,6 @@ from heaven_engine import (
from backend.data.providers.ifind_client import IfindError
from llm_strategy import LLMCompilerError, compile_strategy_with_llm, test_llm_connection
from mentor_agent import MentorAgentError, stream_with_mentor
from market_insights import MarketInsightsService
from screener import (
FACTOR_FIELDS,
FACTOR_GROUPS,
@@ -53,14 +52,14 @@ 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.auction import AuctionServiceMixin
from backend.features.dragon_tiger import DragonTigerServiceMixin
from backend.features.pools import PoolServiceMixin
from backend.features.popularity import PopularityServiceMixin
from backend.features.rotation import RotationServiceMixin
from backend.features.sentiment import SentimentServiceMixin
from backend.features.system import SystemHttpMixin
from sentiment_engine import (
COMPONENT_WEIGHTS,
SENTIMENT_ENGINE_VERSION,
apply_sentiment_to_dashboard,
build_sentiment_history,
latest_contiguous_history,
)
from backend.features.themes import ThemeServiceMixin
from backend.data.providers.tushare_client import TushareClient, TushareError, _sector_coverage_issue
@@ -141,7 +140,16 @@ MENTOR_ETF_UNIVERSE = (
)
class DashboardService(MarketServiceMixin):
class DashboardService(
MarketServiceMixin,
SentimentServiceMixin,
PoolServiceMixin,
RotationServiceMixin,
AuctionServiceMixin,
ThemeServiceMixin,
PopularityServiceMixin,
DragonTigerServiceMixin,
):
def __init__(self) -> None:
runtime = load_runtime_settings()
self.vault = SecretVault(runtime.encryption_key)
@@ -734,184 +742,7 @@ 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)
# 板块轮动固定展示最近 9 个交易日,按由近到远排列。
limit = 9
snapshots = self.database.list_snapshot_payloads(normalized_date, 240)
by_trade_date: dict[str, dict[str, Any]] = {}
for snapshot in snapshots:
meta = snapshot.get("meta") or {}
actual_date = str(meta.get("trade_date") or snapshot.get("_snapshot_date") or "")
compact_date = actual_date.replace("-", "")
if len(compact_date) == 8:
by_trade_date[compact_date] = snapshot
sentiment_dates = {
str(row.get("trade_date") or "").replace("-", "")
for row in latest_contiguous_history(build_sentiment_history(snapshots))
}
ordered_dates = sorted(
date_key for date_key in by_trade_date
if not sentiment_dates or date_key in sentiment_dates
)[-limit:][::-1]
rows = []
for date_key in ordered_dates:
snapshot = by_trade_date[date_key]
sector_context = {
str(item.get("name") or ""): item
for item in snapshot.get("sectors") or []
}
sectors = []
for item in (snapshot.get("sector_rotation") or [])[:12]:
name = str(item.get("name") or "").strip()
context = sector_context.get(name, {})
sectors.append(
{
"name": name,
"rank": int(item.get("rank") or len(sectors) + 1),
"trend": item.get("trend") or "持平",
"count": int(item.get("count") or 0),
"strength": float(item.get("strength") or context.get("strength") or 0),
"change": float(context.get("change") or 0),
"leader": item.get("leader") or context.get("leader") or "--",
}
)
rows.append(
{
"trade_date": f"{date_key[:4]}-{date_key[4:6]}-{date_key[6:]}",
"sectors": sectors,
}
)
return {
"trade_date": rows[0]["trade_date"] if rows else normalized_date,
"available_days": len(ordered_dates),
"requested_days": limit,
"rows": rows,
}
def rotation_sector_members(self, trade_date: str, sector_name: str) -> dict[str, Any]:
normalized_date = normalize_date(trade_date)
sector_name = validate_text(sector_name, "板块名称", 60, required=True)
dashboard = self.get_dashboard(normalized_date)
actual_date = normalize_date(
str((dashboard.get("meta") or {}).get("trade_date") or normalized_date)
)
cache_key = f"{actual_date}:{sector_name}"
cached = self.database.get_data_snapshot("rotation_sector_members_v1", cache_key)
if cached:
cached["meta"] = {**(cached.get("meta") or {}), "cached": True}
return cached
if not self.configured:
raise ValueError("板块成分数据暂不可用。")
representative = next(
(
item for item in dashboard.get("limits") or []
if str(item.get("sector") or "").strip() == sector_name
),
None,
)
if not representative:
raise ValueError("未找到该板块的代表股票,暂时无法核验成分股。")
raw_code = str(representative.get("ts_code") or representative.get("code") or "")
if "." in raw_code:
ts_code = raw_code
elif raw_code.startswith(("4", "8", "92")):
ts_code = f"{raw_code}.BJ"
elif raw_code.startswith(("6", "68", "90")):
ts_code = f"{raw_code}.SH"
else:
ts_code = f"{raw_code}.SZ"
client = self._tushare_client()
try:
industry = client.sw_stock_industry(ts_code, actual_date)
sector_code = str(industry.get("l2_code") or "")
members = client.sw_sector_members(sector_code, actual_date)
except TushareError as exc:
raise ValueError(f"该板块成分股暂不可用:{exc}") from exc
daily_rows = self.database.daily_bars_for_date(actual_date)
if len(daily_rows) < 1000:
try:
daily_rows = client.query(
"daily",
{"trade_date": actual_date},
"ts_code,trade_date,open,high,low,close,pct_chg,vol,amount",
)
if daily_rows:
self.database.upsert_daily_bars(daily_rows)
except TushareError:
daily_rows = self.database.daily_bars_for_date(actual_date)
daily_map = {str(item.get("ts_code") or ""): item for item in daily_rows}
rows = []
for member in members:
member_code = str(member.get("ts_code") or "")
quote = daily_map.get(member_code) or {}
rows.append(
{
"code": member_code.split(".")[0],
"ts_code": member_code,
"name": str(member.get("name") or "--"),
"change": quote.get("pct_chg"),
"open": quote.get("open"),
"close": quote.get("close"),
"amount_billion": (
round(float(quote.get("amount") or 0) / 100000, 2)
if quote else None
),
"quoted": bool(quote),
}
)
rows.sort(
key=lambda item: (
bool(item.get("quoted")),
float(item.get("change") or -999),
float(item.get("amount_billion") or 0),
),
reverse=True,
)
result = {
"meta": {
"trade_date": self._display_compact_date(actual_date),
"sector_name": str(industry.get("l2_name") or sector_name),
"sector_code": sector_code,
"member_count": len(rows),
"quoted_count": sum(bool(item.get("quoted")) for item in rows),
"cached": False,
},
"rows": rows,
}
self.database.save_data_snapshot(
"rotation_sector_members_v1", cache_key, "tushare", result
)
return result
def status(self) -> dict[str, Any]:
llm_access = self.llm_access_status()
@@ -929,28 +760,9 @@ class DashboardService(MarketServiceMixin):
}
def _market_insights(self) -> MarketInsightsService:
if not self.configured:
raise ValueError("行情数据尚未配置。")
return MarketInsightsService(
self.database,
self._tushare_client(),
ifind=self.ifind,
)
def auction_center(self, trade_date: str, force: bool = False) -> dict[str, Any]:
return self._market_insights().auction_center(
normalize_date(trade_date), force, self.current_user_id
)
def theme_library(self, trade_date: str, force: bool = False) -> dict[str, Any]:
return self._market_insights().theme_library(normalize_date(trade_date), force)
def theme_detail(self, code: str, trade_date: str) -> dict[str, Any]:
return self._market_insights().theme_detail(code, normalize_date(trade_date))
def popularity(self, trade_date: str, force: bool = False) -> dict[str, Any]:
return self._market_insights().popularity(normalize_date(trade_date), force)
@staticmethod
def _ifind_field(row: dict[str, Any], tokens: tuple[str, ...]) -> Any:
@@ -3250,427 +3062,6 @@ class DashboardService(MarketServiceMixin):
)
return result
def get_hot_money_profiles(self, force: bool = False) -> dict[str, Any]:
cache_kind = "hot_money_profiles_v1"
cache_key = "directory"
cached = self.database.get_data_snapshot(cache_kind, cache_key)
if cached and not force:
cached["meta"] = {**cached.get("meta", {}), "cached": True}
return cached
if self.configured:
try:
payload = self._tushare_client().hot_money_profiles()
except TushareError:
if cached:
cached["meta"] = {
**cached.get("meta", {}),
"cached": True,
"stale": True,
"notice": "名录暂未完成更新,当前展示最近一次收录结果。",
}
return cached
return {
"meta": {
"source": "unavailable",
"status": "unavailable",
"schema_version": 1,
"cached": False,
"updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
"notice": "游资名录暂不可用,请稍后重试。",
},
"summary": {
"profile_count": 0,
"described_count": 0,
"organization_count": 0,
},
"profiles": [],
}
payload["meta"]["cached"] = False
if payload.get("meta", {}).get("status") == "success":
self.database.save_data_snapshot(cache_kind, cache_key, "tushare", payload)
return payload
if cached:
cached["meta"] = {**cached.get("meta", {}), "cached": True}
return cached
return {
"meta": {
"source": "unavailable",
"status": "unavailable",
"schema_version": 1,
"cached": False,
"updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
"notice": "游资名录暂不可用,请联系管理员检查行情配置。",
},
"summary": {
"profile_count": 0,
"described_count": 0,
"organization_count": 0,
},
"profiles": [],
}
def get_dragon_tiger(self, trade_date: str, force: bool = False) -> dict[str, Any]:
normalized_date = normalize_date(trade_date)
cache_kind = "hot_money_detail_v3"
if not force:
cached = self.database.get_data_snapshot(cache_kind, normalized_date)
if (
cached
and cached.get("meta", {}).get("source") == "tushare"
and cached.get("meta", {}).get("status") == "success"
and int(cached.get("meta", {}).get("schema_version") or 0) == 3
):
cached["meta"] = {**cached.get("meta", {}), "cached": True}
return cached
if self.configured:
try:
payload = self._tushare_client().dragon_tiger(normalized_date)
except TushareError as exc:
return {
"meta": {
"requested_date": f"{normalized_date[:4]}-{normalized_date[4:6]}-{normalized_date[6:8]}",
"trade_date": f"{normalized_date[:4]}-{normalized_date[4:6]}-{normalized_date[6:8]}",
"source": "tushare_error",
"status": "error",
"schema_version": 3,
"cached": False,
"updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
"notice": "龙虎榜数据暂不可用,请稍后重试。",
},
"summary": {
"trader_count": 0,
"identity_count": 0,
"operation_count": 0,
"active_stock_count": 0,
"seat_net_buy_million": 0,
"unclassified_count": 0,
"directory_count": 0,
},
"traders": [],
"unclassified_seats": [],
"rows": [],
}
payload["meta"]["cached"] = False
if payload.get("meta", {}).get("status") == "success":
self.database.save_data_snapshot(cache_kind, normalized_date, "tushare", payload)
return payload
return {
"meta": {
"requested_date": f"{normalized_date[:4]}-{normalized_date[4:6]}-{normalized_date[6:8]}",
"trade_date": f"{normalized_date[:4]}-{normalized_date[4:6]}-{normalized_date[6:8]}",
"source": "unavailable",
"status": "unavailable",
"schema_version": 3,
"cached": False,
"notice": "龙虎榜数据暂不可用,请联系管理员检查行情配置。",
},
"summary": {
"trader_count": 0,
"identity_count": 0,
"operation_count": 0,
"active_stock_count": 0,
"seat_net_buy_million": 0,
"unclassified_count": 0,
"directory_count": 0,
},
"traders": [],
"unclassified_seats": [],
"rows": [],
}
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"(?<!\d)(\d{6})(?!\d)", text)
if match:
compact = match.group(1)
return f"{compact[:2]}:{compact[2:4]}:{compact[4:]}"
return ""
parts = match.group(1).split(":")
return ":".join(part.zfill(2) for part in parts)
@staticmethod
def _merge_ifind_event_enrichment(
dashboard: dict[str, Any], enrichment: dict[str, Any]
) -> 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()
result = dict(payload)
rows = payload.get("rows") or []
for row in rows:
for institution in row.get("institutions") or []:
institution["alias"] = aliases.get(institution.get("seat_name", ""), "")
traders: dict[tuple[str, str], dict[str, Any]] = {}
unclassified: dict[str, dict[str, Any]] = {}
seen_operations: set[tuple[Any, ...]] = set()
builtin_aliases = {
"国泰海通证券股份有限公司南京太平南路证券营业部": "作手新一",
}
for row in rows:
for institution in row.get("institutions") or []:
seat_name = str(institution.get("seat_name") or "未知席位").strip()
saved_alias = str(institution.get("alias") or "").strip()
builtin_alias = builtin_aliases.get(seat_name, "")
if saved_alias or builtin_alias:
identity_name = saved_alias or builtin_alias
identity_type = "trader"
recognized = True
identity_source = "manual" if saved_alias else "builtin"
elif "机构专用" in seat_name:
identity_name = "机构专用"
identity_type = "institution"
recognized = True
identity_source = "system"
elif "沪股通专用" in seat_name or "深股通专用" in seat_name:
identity_name = "北向资金"
identity_type = "channel"
recognized = True
identity_source = "system"
else:
identity_name = seat_name
identity_type = "unclassified"
recognized = False
identity_source = "raw"
buy = round(float(institution.get("buy_million") or 0), 2)
sell = round(float(institution.get("sell_million") or 0), 2)
net_buy = round(float(institution.get("net_buy_million") or 0), 2)
operation_key = (row.get("code"), seat_name, buy, sell, net_buy)
if operation_key in seen_operations:
continue
seen_operations.add(operation_key)
group_key = (identity_type, identity_name)
group = traders.setdefault(
group_key,
{
"name": identity_name,
"identity_type": identity_type,
"identity_source": identity_source,
"recognized": recognized,
"buy_million": 0.0,
"sell_million": 0.0,
"net_buy_million": 0.0,
"seat_names": set(),
"stock_codes": set(),
"operations": [],
},
)
group["buy_million"] += buy
group["sell_million"] += sell
group["net_buy_million"] += net_buy
group["seat_names"].add(seat_name)
group["stock_codes"].add(str(row.get("code") or ""))
group["operations"].append(
{
"code": row.get("code") or "",
"name": row.get("name") or "--",
"change": row.get("change") or 0,
"direction": "买入" if net_buy > 0 else "卖出" if net_buy < 0 else "持平",
"buy_million": buy,
"sell_million": sell,
"net_buy_million": net_buy,
"reason": row.get("reason") or "--",
"seat_name": seat_name,
"seat_alias": identity_name if recognized else "",
}
)
if not recognized:
pending = unclassified.setdefault(
seat_name,
{
"seat_name": seat_name,
"stock_codes": set(),
"operation_count": 0,
"buy_million": 0.0,
"sell_million": 0.0,
"net_buy_million": 0.0,
},
)
pending["stock_codes"].add(str(row.get("code") or ""))
pending["operation_count"] += 1
pending["buy_million"] += buy
pending["sell_million"] += sell
pending["net_buy_million"] += net_buy
type_order = {"trader": 0, "institution": 1, "channel": 2, "unclassified": 3}
aggregated = list(traders.values())
aggregated.sort(
key=lambda item: (
type_order.get(item["identity_type"], 9),
-abs(item["net_buy_million"]),
item["name"],
)
)
for index, group in enumerate(aggregated, start=1):
group["id"] = f"identity-{index}"
group["buy_million"] = round(group["buy_million"], 2)
group["sell_million"] = round(group["sell_million"], 2)
group["net_buy_million"] = round(group["net_buy_million"], 2)
group["seat_count"] = len(group.pop("seat_names"))
group["stock_count"] = len(group.pop("stock_codes"))
group["operation_count"] = len(group["operations"])
group["operations"].sort(
key=lambda item: abs(float(item.get("net_buy_million") or 0)), reverse=True
)
pending_seats = list(unclassified.values())
for pending in pending_seats:
pending["stock_count"] = len(pending.pop("stock_codes"))
pending["buy_million"] = round(pending["buy_million"], 2)
pending["sell_million"] = round(pending["sell_million"], 2)
pending["net_buy_million"] = round(pending["net_buy_million"], 2)
pending_seats.sort(key=lambda item: abs(item["net_buy_million"]), reverse=True)
operation_count = sum(item["operation_count"] for item in aggregated)
active_stocks = {
operation["code"] for item in aggregated for operation in item["operations"]
}
seat_net_buy = round(sum(item["net_buy_million"] for item in aggregated), 2)
result["rows"] = rows
result["traders"] = aggregated
result["unclassified_seats"] = pending_seats
result["summary"] = {
**(payload.get("summary") or {}),
"trader_count": sum(item["identity_type"] == "trader" for item in aggregated),
"identity_count": len(aggregated),
"operation_count": operation_count,
"active_stock_count": len(active_stocks),
"seat_net_buy_million": seat_net_buy,
"unclassified_count": len(pending_seats),
}
return result
SERVICE = DashboardService()
+1 -1
View File
@@ -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"
+4
View File
@@ -0,0 +1,4 @@
from .repository import AuctionRepositoryMixin
from .service import AuctionServiceMixin
__all__ = ["AuctionRepositoryMixin", "AuctionServiceMixin"]
@@ -0,0 +1,63 @@
from __future__ import annotations
from typing import Any
class AuctionRepositoryMixin:
def upsert_auction_factors(self, rows: list[dict[str, Any]]) -> int:
values = []
for row in rows:
trade_date = str(row.get("trade_date") or "")
ts_code = str(row.get("ts_code") or "")
price = float(row.get("price") or 0)
pre_close = float(row.get("pre_close") or 0)
if not trade_date or not ts_code or price <= 0 or pre_close <= 0:
continue
values.append(
(
trade_date,
ts_code,
price,
pre_close,
(price / pre_close - 1) * 100,
float(row.get("vol") or 0),
float(row.get("amount") or 0),
float(row.get("turnover_rate") or 0),
float(row.get("volume_ratio") or 0),
)
)
with self.connect() as connection:
connection.executemany(
"""
INSERT INTO auction_factors
(trade_date, ts_code, price, pre_close, change, vol, amount,
turnover_rate, volume_ratio)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(trade_date, ts_code) DO UPDATE SET
price=excluded.price, pre_close=excluded.pre_close,
change=excluded.change, vol=excluded.vol, amount=excluded.amount,
turnover_rate=excluded.turnover_rate,
volume_ratio=excluded.volume_ratio
""",
values,
)
return len(values)
def auction_factor_dates(self, end_date: str = "", limit: int = 80) -> list[str]:
where = "WHERE trade_date <= ?" if end_date else ""
parameters: tuple[Any, ...] = (end_date, limit) if end_date else (limit,)
with self.connect() as connection:
rows = connection.execute(
f"SELECT DISTINCT trade_date FROM auction_factors {where} "
"ORDER BY trade_date DESC LIMIT ?",
parameters,
).fetchall()
return [row["trade_date"] for row in reversed(rows)]
def auction_factors_for_date(self, trade_date: str) -> list[dict[str, Any]]:
with self.connect() as connection:
rows = connection.execute(
"SELECT * FROM auction_factors WHERE trade_date = ? ORDER BY ts_code",
(trade_date,),
).fetchall()
return [dict(row) for row in rows]
+13
View File
@@ -0,0 +1,13 @@
from __future__ import annotations
from typing import Any
from backend.bootstrap.config import normalize_date
from backend.features.market.insights import MarketInsightsService
class AuctionServiceMixin:
def auction_center(self, trade_date: str, force: bool = False) -> dict[str, Any]:
return self._market_insights().auction_center(
normalize_date(trade_date), force, self.current_user_id
)
@@ -0,0 +1,4 @@
from .repository import DragonTigerRepositoryMixin
from .service import DragonTigerServiceMixin
__all__ = ["DragonTigerRepositoryMixin", "DragonTigerServiceMixin"]
@@ -0,0 +1,61 @@
from __future__ import annotations
from datetime import datetime
from typing import Any
class DragonTigerRepositoryMixin:
def list_seat_aliases(self) -> dict[str, str]:
with self.connect() as connection:
rows = connection.execute("SELECT seat_name, alias FROM seat_aliases").fetchall()
return {row["seat_name"]: row["alias"] for row in rows}
def save_seat_alias(self, seat_name: str, alias: str) -> None:
now = datetime.now().astimezone().isoformat(timespec="seconds")
with self.connect() as connection:
connection.execute(
"""
INSERT INTO seat_aliases (seat_name, alias, updated_at)
VALUES (?, ?, ?)
ON CONFLICT(seat_name) DO UPDATE SET
alias = excluded.alias,
updated_at = excluded.updated_at
""",
(seat_name, alias, now),
)
def upsert_lhb_institutions(self, rows: list[dict[str, Any]]) -> int:
grouped: dict[tuple[str, str], dict[str, float | int]] = {}
for row in rows:
trade_date = str(row.get("trade_date") or "")
ts_code = str(row.get("ts_code") or "")
seat_name = str(row.get("exalter") or row.get("seat_name") or "")
if not trade_date or not ts_code or "机构专用" not in seat_name:
continue
group = grouped.setdefault(
(trade_date, ts_code),
{"net": 0.0, "buy": 0.0, "sell": 0.0, "seats": 0},
)
group["net"] = float(group["net"]) + float(row.get("net_buy") or row.get("net_amount") or 0)
group["buy"] = float(group["buy"]) + float(row.get("buy") or row.get("buy_amount") or 0)
group["sell"] = float(group["sell"]) + float(row.get("sell") or row.get("sell_amount") or 0)
group["seats"] = int(group["seats"]) + 1
values = [
(trade_date, ts_code, item["net"], item["buy"], item["sell"], item["seats"])
for (trade_date, ts_code), item in grouped.items()
]
with self.connect() as connection:
connection.executemany(
"""
INSERT INTO lhb_institution_daily
(trade_date, ts_code, net_buy_amount, buy_amount, sell_amount, seat_count)
VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT(trade_date, ts_code) DO UPDATE SET
net_buy_amount=excluded.net_buy_amount,
buy_amount=excluded.buy_amount,
sell_amount=excluded.sell_amount,
seat_count=excluded.seat_count
""",
values,
)
return len(values)
@@ -0,0 +1,288 @@
from __future__ import annotations
from datetime import datetime
from typing import Any
from backend.bootstrap.config import normalize_date
from backend.data.providers.tushare_client import TushareError
class DragonTigerServiceMixin:
def get_hot_money_profiles(self, force: bool = False) -> dict[str, Any]:
cache_kind = "hot_money_profiles_v1"
cache_key = "directory"
cached = self.database.get_data_snapshot(cache_kind, cache_key)
if cached and not force:
cached["meta"] = {**cached.get("meta", {}), "cached": True}
return cached
if self.configured:
try:
payload = self._tushare_client().hot_money_profiles()
except TushareError:
if cached:
cached["meta"] = {
**cached.get("meta", {}),
"cached": True,
"stale": True,
"notice": "名录暂未完成更新,当前展示最近一次收录结果。",
}
return cached
return {
"meta": {
"source": "unavailable",
"status": "unavailable",
"schema_version": 1,
"cached": False,
"updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
"notice": "游资名录暂不可用,请稍后重试。",
},
"summary": {
"profile_count": 0,
"described_count": 0,
"organization_count": 0,
},
"profiles": [],
}
payload["meta"]["cached"] = False
if payload.get("meta", {}).get("status") == "success":
self.database.save_data_snapshot(cache_kind, cache_key, "tushare", payload)
return payload
if cached:
cached["meta"] = {**cached.get("meta", {}), "cached": True}
return cached
return {
"meta": {
"source": "unavailable",
"status": "unavailable",
"schema_version": 1,
"cached": False,
"updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
"notice": "游资名录暂不可用,请联系管理员检查行情配置。",
},
"summary": {
"profile_count": 0,
"described_count": 0,
"organization_count": 0,
},
"profiles": [],
}
def get_dragon_tiger(self, trade_date: str, force: bool = False) -> dict[str, Any]:
normalized_date = normalize_date(trade_date)
cache_kind = "hot_money_detail_v3"
if not force:
cached = self.database.get_data_snapshot(cache_kind, normalized_date)
if (
cached
and cached.get("meta", {}).get("source") == "tushare"
and cached.get("meta", {}).get("status") == "success"
and int(cached.get("meta", {}).get("schema_version") or 0) == 3
):
cached["meta"] = {**cached.get("meta", {}), "cached": True}
return cached
if self.configured:
try:
payload = self._tushare_client().dragon_tiger(normalized_date)
except TushareError as exc:
return {
"meta": {
"requested_date": f"{normalized_date[:4]}-{normalized_date[4:6]}-{normalized_date[6:8]}",
"trade_date": f"{normalized_date[:4]}-{normalized_date[4:6]}-{normalized_date[6:8]}",
"source": "tushare_error",
"status": "error",
"schema_version": 3,
"cached": False,
"updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
"notice": "龙虎榜数据暂不可用,请稍后重试。",
},
"summary": {
"trader_count": 0,
"identity_count": 0,
"operation_count": 0,
"active_stock_count": 0,
"seat_net_buy_million": 0,
"unclassified_count": 0,
"directory_count": 0,
},
"traders": [],
"unclassified_seats": [],
"rows": [],
}
payload["meta"]["cached"] = False
if payload.get("meta", {}).get("status") == "success":
self.database.save_data_snapshot(cache_kind, normalized_date, "tushare", payload)
return payload
return {
"meta": {
"requested_date": f"{normalized_date[:4]}-{normalized_date[4:6]}-{normalized_date[6:8]}",
"trade_date": f"{normalized_date[:4]}-{normalized_date[4:6]}-{normalized_date[6:8]}",
"source": "unavailable",
"status": "unavailable",
"schema_version": 3,
"cached": False,
"notice": "龙虎榜数据暂不可用,请联系管理员检查行情配置。",
},
"summary": {
"trader_count": 0,
"identity_count": 0,
"operation_count": 0,
"active_stock_count": 0,
"seat_net_buy_million": 0,
"unclassified_count": 0,
"directory_count": 0,
},
"traders": [],
"unclassified_seats": [],
"rows": [],
}
def _apply_seat_aliases(self, payload: dict[str, Any]) -> dict[str, Any]:
aliases = self.database.list_seat_aliases()
result = dict(payload)
rows = payload.get("rows") or []
for row in rows:
for institution in row.get("institutions") or []:
institution["alias"] = aliases.get(institution.get("seat_name", ""), "")
traders: dict[tuple[str, str], dict[str, Any]] = {}
unclassified: dict[str, dict[str, Any]] = {}
seen_operations: set[tuple[Any, ...]] = set()
builtin_aliases = {
"国泰海通证券股份有限公司南京太平南路证券营业部": "作手新一",
}
for row in rows:
for institution in row.get("institutions") or []:
seat_name = str(institution.get("seat_name") or "未知席位").strip()
saved_alias = str(institution.get("alias") or "").strip()
builtin_alias = builtin_aliases.get(seat_name, "")
if saved_alias or builtin_alias:
identity_name = saved_alias or builtin_alias
identity_type = "trader"
recognized = True
identity_source = "manual" if saved_alias else "builtin"
elif "机构专用" in seat_name:
identity_name = "机构专用"
identity_type = "institution"
recognized = True
identity_source = "system"
elif "沪股通专用" in seat_name or "深股通专用" in seat_name:
identity_name = "北向资金"
identity_type = "channel"
recognized = True
identity_source = "system"
else:
identity_name = seat_name
identity_type = "unclassified"
recognized = False
identity_source = "raw"
buy = round(float(institution.get("buy_million") or 0), 2)
sell = round(float(institution.get("sell_million") or 0), 2)
net_buy = round(float(institution.get("net_buy_million") or 0), 2)
operation_key = (row.get("code"), seat_name, buy, sell, net_buy)
if operation_key in seen_operations:
continue
seen_operations.add(operation_key)
group_key = (identity_type, identity_name)
group = traders.setdefault(
group_key,
{
"name": identity_name,
"identity_type": identity_type,
"identity_source": identity_source,
"recognized": recognized,
"buy_million": 0.0,
"sell_million": 0.0,
"net_buy_million": 0.0,
"seat_names": set(),
"stock_codes": set(),
"operations": [],
},
)
group["buy_million"] += buy
group["sell_million"] += sell
group["net_buy_million"] += net_buy
group["seat_names"].add(seat_name)
group["stock_codes"].add(str(row.get("code") or ""))
group["operations"].append(
{
"code": row.get("code") or "",
"name": row.get("name") or "--",
"change": row.get("change") or 0,
"direction": "买入" if net_buy > 0 else "卖出" if net_buy < 0 else "持平",
"buy_million": buy,
"sell_million": sell,
"net_buy_million": net_buy,
"reason": row.get("reason") or "--",
"seat_name": seat_name,
"seat_alias": identity_name if recognized else "",
}
)
if not recognized:
pending = unclassified.setdefault(
seat_name,
{
"seat_name": seat_name,
"stock_codes": set(),
"operation_count": 0,
"buy_million": 0.0,
"sell_million": 0.0,
"net_buy_million": 0.0,
},
)
pending["stock_codes"].add(str(row.get("code") or ""))
pending["operation_count"] += 1
pending["buy_million"] += buy
pending["sell_million"] += sell
pending["net_buy_million"] += net_buy
type_order = {"trader": 0, "institution": 1, "channel": 2, "unclassified": 3}
aggregated = list(traders.values())
aggregated.sort(
key=lambda item: (
type_order.get(item["identity_type"], 9),
-abs(item["net_buy_million"]),
item["name"],
)
)
for index, group in enumerate(aggregated, start=1):
group["id"] = f"identity-{index}"
group["buy_million"] = round(group["buy_million"], 2)
group["sell_million"] = round(group["sell_million"], 2)
group["net_buy_million"] = round(group["net_buy_million"], 2)
group["seat_count"] = len(group.pop("seat_names"))
group["stock_count"] = len(group.pop("stock_codes"))
group["operation_count"] = len(group["operations"])
group["operations"].sort(
key=lambda item: abs(float(item.get("net_buy_million") or 0)), reverse=True
)
pending_seats = list(unclassified.values())
for pending in pending_seats:
pending["stock_count"] = len(pending.pop("stock_codes"))
pending["buy_million"] = round(pending["buy_million"], 2)
pending["sell_million"] = round(pending["sell_million"], 2)
pending["net_buy_million"] = round(pending["net_buy_million"], 2)
pending_seats.sort(key=lambda item: abs(item["net_buy_million"]), reverse=True)
operation_count = sum(item["operation_count"] for item in aggregated)
active_stocks = {
operation["code"] for item in aggregated for operation in item["operations"]
}
seat_net_buy = round(sum(item["net_buy_million"] for item in aggregated), 2)
result["rows"] = rows
result["traders"] = aggregated
result["unclassified_seats"] = pending_seats
result["summary"] = {
**(payload.get("summary") or {}),
"trader_count": sum(item["identity_type"] == "trader" for item in aggregated),
"identity_count": len(aggregated),
"operation_count": operation_count,
"active_stock_count": len(active_stocks),
"seat_net_buy_million": seat_net_buy,
"unclassified_count": len(pending_seats),
}
return result
File diff suppressed because it is too large Load Diff
+10 -1
View File
@@ -14,7 +14,8 @@ 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.market.insights import MarketInsightsService
from backend.features.sentiment.engine import SENTIMENT_ENGINE_VERSION
SEARCH_INDEXES = (
@@ -36,6 +37,14 @@ THS_SEARCH_TYPES = {
class MarketServiceMixin:
def _market_insights(self) -> MarketInsightsService:
if not self.configured:
raise ValueError("行情数据尚未配置。")
return MarketInsightsService(
self.database,
self._tushare_client(),
ifind=self.ifind,
)
def _tushare_client(self) -> TushareClient:
gateway = getattr(self, "data_gateway", None)
if gateway is not None:
+6
View File
@@ -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"]
+27
View File
@@ -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}
+150
View File
@@ -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"(?<!\d)(\d{6})(?!\d)", text)
if match:
compact = match.group(1)
return f"{compact[:2]}:{compact[2:4]}:{compact[4:]}"
return ""
parts = match.group(1).split(":")
return ":".join(part.zfill(2) for part in parts)
@staticmethod
def _merge_ifind_event_enrichment(
dashboard: dict[str, Any], enrichment: dict[str, Any]
) -> 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"]
@@ -0,0 +1,4 @@
from .repository import PopularityRepositoryMixin
from .service import PopularityServiceMixin
__all__ = ["PopularityRepositoryMixin", "PopularityServiceMixin"]
@@ -0,0 +1,37 @@
from __future__ import annotations
from typing import Any
class PopularityRepositoryMixin:
def upsert_popularity_factors(self, rows: list[dict[str, Any]]) -> int:
values = [
(
str(row.get("trade_date") or ""),
str(row.get("ts_code") or ""),
int(row["ths_rank"]) if row.get("ths_rank") not in (None, "") else None,
int(row["dc_rank"]) if row.get("dc_rank") not in (None, "") else None,
float(row.get("combined_score") or 0),
int(row["rank_change"]) if row.get("rank_change") not in (None, "") else None,
int(bool(row.get("dual_source"))),
)
for row in rows
if row.get("trade_date") and row.get("ts_code")
]
with self.connect() as connection:
connection.executemany(
"""
INSERT INTO popularity_factors
(trade_date, ts_code, ths_rank, dc_rank, combined_score,
rank_change, dual_source)
VALUES (?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(trade_date, ts_code) DO UPDATE SET
ths_rank=excluded.ths_rank,
dc_rank=excluded.dc_rank,
combined_score=excluded.combined_score,
rank_change=excluded.rank_change,
dual_source=excluded.dual_source
""",
values,
)
return len(values)
@@ -0,0 +1,11 @@
from __future__ import annotations
from typing import Any
from backend.bootstrap.config import normalize_date
from backend.features.market.insights import MarketInsightsService
class PopularityServiceMixin:
def popularity(self, trade_date: str, force: bool = False) -> dict[str, Any]:
return self._market_insights().popularity(normalize_date(trade_date), force)
@@ -0,0 +1,5 @@
"""Sector rotation history and constituent detail feature."""
from .service import RotationServiceMixin
__all__ = ["RotationServiceMixin"]
+165
View File
@@ -0,0 +1,165 @@
from __future__ import annotations
from typing import Any
from backend.bootstrap.config import normalize_date, validate_text
from backend.data.providers.tushare_client import TushareError
from backend.features.sentiment.engine import (
build_sentiment_history,
latest_contiguous_history,
)
class RotationServiceMixin:
def rotation_history(self, trade_date: str, limit: int = 9) -> dict[str, Any]:
normalized_date = normalize_date(trade_date)
# 板块轮动固定展示最近 9 个交易日,按由近到远排列。
limit = 9
snapshots = self.database.list_snapshot_payloads(normalized_date, 240)
by_trade_date: dict[str, dict[str, Any]] = {}
for snapshot in snapshots:
meta = snapshot.get("meta") or {}
actual_date = str(meta.get("trade_date") or snapshot.get("_snapshot_date") or "")
compact_date = actual_date.replace("-", "")
if len(compact_date) == 8:
by_trade_date[compact_date] = snapshot
sentiment_dates = {
str(row.get("trade_date") or "").replace("-", "")
for row in latest_contiguous_history(build_sentiment_history(snapshots))
}
ordered_dates = sorted(
date_key for date_key in by_trade_date
if not sentiment_dates or date_key in sentiment_dates
)[-limit:][::-1]
rows = []
for date_key in ordered_dates:
snapshot = by_trade_date[date_key]
sector_context = {
str(item.get("name") or ""): item
for item in snapshot.get("sectors") or []
}
sectors = []
for item in (snapshot.get("sector_rotation") or [])[:12]:
name = str(item.get("name") or "").strip()
context = sector_context.get(name, {})
sectors.append(
{
"name": name,
"rank": int(item.get("rank") or len(sectors) + 1),
"trend": item.get("trend") or "持平",
"count": int(item.get("count") or 0),
"strength": float(item.get("strength") or context.get("strength") or 0),
"change": float(context.get("change") or 0),
"leader": item.get("leader") or context.get("leader") or "--",
}
)
rows.append(
{
"trade_date": f"{date_key[:4]}-{date_key[4:6]}-{date_key[6:]}",
"sectors": sectors,
}
)
return {
"trade_date": rows[0]["trade_date"] if rows else normalized_date,
"available_days": len(ordered_dates),
"requested_days": limit,
"rows": rows,
}
def rotation_sector_members(self, trade_date: str, sector_name: str) -> dict[str, Any]:
normalized_date = normalize_date(trade_date)
sector_name = validate_text(sector_name, "板块名称", 60, required=True)
dashboard = self.get_dashboard(normalized_date)
actual_date = normalize_date(
str((dashboard.get("meta") or {}).get("trade_date") or normalized_date)
)
cache_key = f"{actual_date}:{sector_name}"
cached = self.database.get_data_snapshot("rotation_sector_members_v1", cache_key)
if cached:
cached["meta"] = {**(cached.get("meta") or {}), "cached": True}
return cached
if not self.configured:
raise ValueError("板块成分数据暂不可用。")
representative = next(
(
item for item in dashboard.get("limits") or []
if str(item.get("sector") or "").strip() == sector_name
),
None,
)
if not representative:
raise ValueError("未找到该板块的代表股票,暂时无法核验成分股。")
raw_code = str(representative.get("ts_code") or representative.get("code") or "")
if "." in raw_code:
ts_code = raw_code
elif raw_code.startswith(("4", "8", "92")):
ts_code = f"{raw_code}.BJ"
elif raw_code.startswith(("6", "68", "90")):
ts_code = f"{raw_code}.SH"
else:
ts_code = f"{raw_code}.SZ"
client = self._tushare_client()
try:
industry = client.sw_stock_industry(ts_code, actual_date)
sector_code = str(industry.get("l2_code") or "")
members = client.sw_sector_members(sector_code, actual_date)
except TushareError as exc:
raise ValueError(f"该板块成分股暂不可用:{exc}") from exc
daily_rows = self.database.daily_bars_for_date(actual_date)
if len(daily_rows) < 1000:
try:
daily_rows = client.query(
"daily",
{"trade_date": actual_date},
"ts_code,trade_date,open,high,low,close,pct_chg,vol,amount",
)
if daily_rows:
self.database.upsert_daily_bars(daily_rows)
except TushareError:
daily_rows = self.database.daily_bars_for_date(actual_date)
daily_map = {str(item.get("ts_code") or ""): item for item in daily_rows}
rows = []
for member in members:
member_code = str(member.get("ts_code") or "")
quote = daily_map.get(member_code) or {}
rows.append(
{
"code": member_code.split(".")[0],
"ts_code": member_code,
"name": str(member.get("name") or "--"),
"change": quote.get("pct_chg"),
"open": quote.get("open"),
"close": quote.get("close"),
"amount_billion": (
round(float(quote.get("amount") or 0) / 100000, 2)
if quote else None
),
"quoted": bool(quote),
}
)
rows.sort(
key=lambda item: (
bool(item.get("quoted")),
float(item.get("change") or -999),
float(item.get("amount_billion") or 0),
),
reverse=True,
)
result = {
"meta": {
"trade_date": self._display_compact_date(actual_date),
"sector_name": str(industry.get("l2_name") or sector_name),
"sector_code": sector_code,
"member_count": len(rows),
"quoted_count": sum(bool(item.get("quoted")) for item in rows),
"cached": False,
},
"rows": rows,
}
self.database.save_data_snapshot(
"rotation_sector_members_v1", cache_key, "tushare", result
)
return result
@@ -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",
]
+496
View File
@@ -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
+39
View File
@@ -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 "固定锚点",
}
+3
View File
@@ -0,0 +1,3 @@
from .service import ThemeServiceMixin
__all__ = ["ThemeServiceMixin"]
+14
View File
@@ -0,0 +1,14 @@
from __future__ import annotations
from typing import Any
from backend.bootstrap.config import normalize_date
from backend.features.market.insights import MarketInsightsService
class ThemeServiceMixin:
def theme_library(self, trade_date: str, force: bool = False) -> dict[str, Any]:
return self._market_insights().theme_library(normalize_date(trade_date), force)
def theme_detail(self, code: str, trade_date: str) -> dict[str, Any]:
return self._market_insights().theme_detail(code, normalize_date(trade_date))
+8 -161
View File
@@ -8,7 +8,11 @@ from typing import Any
from backend.database import MIGRATIONS, MigrationRunner, SQLiteConnectionFactory
from backend.features.accounts.repository import AccountRepositoryMixin
from backend.features.auction.repository import AuctionRepositoryMixin
from backend.features.dragon_tiger.repository import DragonTigerRepositoryMixin
from backend.features.market.repository import MarketRepositoryMixin
from backend.features.pools.repository import PoolRepositoryMixin
from backend.features.popularity.repository import PopularityRepositoryMixin
from backend.features.system.repository import SystemSettingsRepositoryMixin
@@ -23,7 +27,11 @@ def _optional_float(value: Any) -> float | None:
class ReviewDatabase(
AccountRepositoryMixin,
AuctionRepositoryMixin,
DragonTigerRepositoryMixin,
MarketRepositoryMixin,
PoolRepositoryMixin,
PopularityRepositoryMixin,
SystemSettingsRepositoryMixin,
):
def __init__(self, path: Path) -> None:
@@ -836,46 +844,7 @@ 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:
rows = connection.execute("SELECT seat_name, alias FROM seat_aliases").fetchall()
return {row["seat_name"]: row["alias"] for row in rows}
def save_seat_alias(self, seat_name: str, alias: str) -> None:
now = datetime.now().astimezone().isoformat(timespec="seconds")
with self.connect() as connection:
connection.execute(
"""
INSERT INTO seat_aliases (seat_name, alias, updated_at)
VALUES (?, ?, ?)
ON CONFLICT(seat_name) DO UPDATE SET
alias = excluded.alias,
updated_at = excluded.updated_at
""",
(seat_name, alias, now),
)
def list_sector_phase_overrides(self) -> dict[str, str]:
with self.connect() as connection:
@@ -1075,44 +1044,6 @@ class ReviewDatabase(
)
return len(values)
def upsert_auction_factors(self, rows: list[dict[str, Any]]) -> int:
values = []
for row in rows:
trade_date = str(row.get("trade_date") or "")
ts_code = str(row.get("ts_code") or "")
price = float(row.get("price") or 0)
pre_close = float(row.get("pre_close") or 0)
if not trade_date or not ts_code or price <= 0 or pre_close <= 0:
continue
values.append(
(
trade_date,
ts_code,
price,
pre_close,
(price / pre_close - 1) * 100,
float(row.get("vol") or 0),
float(row.get("amount") or 0),
float(row.get("turnover_rate") or 0),
float(row.get("volume_ratio") or 0),
)
)
with self.connect() as connection:
connection.executemany(
"""
INSERT INTO auction_factors
(trade_date, ts_code, price, pre_close, change, vol, amount,
turnover_rate, volume_ratio)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(trade_date, ts_code) DO UPDATE SET
price=excluded.price, pre_close=excluded.pre_close,
change=excluded.change, vol=excluded.vol, amount=excluded.amount,
turnover_rate=excluded.turnover_rate,
volume_ratio=excluded.volume_ratio
""",
values,
)
return len(values)
def upsert_earnings_events(self, rows: list[dict[str, Any]]) -> int:
values = [
@@ -1149,84 +1080,7 @@ class ReviewDatabase(
)
return len(values)
def upsert_popularity_factors(self, rows: list[dict[str, Any]]) -> int:
values = [
(
str(row.get("trade_date") or ""),
str(row.get("ts_code") or ""),
int(row["ths_rank"]) if row.get("ths_rank") not in (None, "") else None,
int(row["dc_rank"]) if row.get("dc_rank") not in (None, "") else None,
float(row.get("combined_score") or 0),
int(row["rank_change"]) if row.get("rank_change") not in (None, "") else None,
int(bool(row.get("dual_source"))),
)
for row in rows
if row.get("trade_date") and row.get("ts_code")
]
with self.connect() as connection:
connection.executemany(
"""
INSERT INTO popularity_factors
(trade_date, ts_code, ths_rank, dc_rank, combined_score,
rank_change, dual_source)
VALUES (?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(trade_date, ts_code) DO UPDATE SET
ths_rank=excluded.ths_rank,
dc_rank=excluded.dc_rank,
combined_score=excluded.combined_score,
rank_change=excluded.rank_change,
dual_source=excluded.dual_source
""",
values,
)
return len(values)
def upsert_lhb_institutions(self, rows: list[dict[str, Any]]) -> int:
grouped: dict[tuple[str, str], dict[str, float | int]] = {}
for row in rows:
trade_date = str(row.get("trade_date") or "")
ts_code = str(row.get("ts_code") or "")
seat_name = str(row.get("exalter") or row.get("seat_name") or "")
if not trade_date or not ts_code or "机构专用" not in seat_name:
continue
group = grouped.setdefault(
(trade_date, ts_code),
{"net": 0.0, "buy": 0.0, "sell": 0.0, "seats": 0},
)
group["net"] = float(group["net"]) + float(row.get("net_buy") or row.get("net_amount") or 0)
group["buy"] = float(group["buy"]) + float(row.get("buy") or row.get("buy_amount") or 0)
group["sell"] = float(group["sell"]) + float(row.get("sell") or row.get("sell_amount") or 0)
group["seats"] = int(group["seats"]) + 1
values = [
(trade_date, ts_code, item["net"], item["buy"], item["sell"], item["seats"])
for (trade_date, ts_code), item in grouped.items()
]
with self.connect() as connection:
connection.executemany(
"""
INSERT INTO lhb_institution_daily
(trade_date, ts_code, net_buy_amount, buy_amount, sell_amount, seat_count)
VALUES (?, ?, ?, ?, ?, ?)
ON CONFLICT(trade_date, ts_code) DO UPDATE SET
net_buy_amount=excluded.net_buy_amount,
buy_amount=excluded.buy_amount,
sell_amount=excluded.sell_amount,
seat_count=excluded.seat_count
""",
values,
)
return len(values)
def auction_factor_dates(self, end_date: str = "", limit: int = 80) -> list[str]:
where = "WHERE trade_date <= ?" if end_date else ""
parameters: tuple[Any, ...] = (end_date, limit) if end_date else (limit,)
with self.connect() as connection:
rows = connection.execute(
f"SELECT DISTINCT trade_date FROM auction_factors {where} "
"ORDER BY trade_date DESC LIMIT ?",
parameters,
).fetchall()
return [row["trade_date"] for row in reversed(rows)]
def daily_indicator_dates(self, end_date: str = "", limit: int = 400) -> list[str]:
where = "WHERE trade_date <= ?" if end_date else ""
@@ -1246,13 +1100,6 @@ class ReviewDatabase(
).fetchall()
return [str(row["end_date"]) for row in rows]
def auction_factors_for_date(self, trade_date: str) -> list[dict[str, Any]]:
with self.connect() as connection:
rows = connection.execute(
"SELECT * FROM auction_factors WHERE trade_date = ? ORDER BY ts_code",
(trade_date,),
).fetchall()
return [dict(row) for row in rows]
def daily_bars_for_date(self, trade_date: str) -> list[dict[str, Any]]:
with self.connect() as connection:
+1 -1
View File
@@ -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 = [
+1 -1312
View File
File diff suppressed because it is too large Load Diff
+1 -1
View File
@@ -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
+4 -493
View File
@@ -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
@@ -0,0 +1,92 @@
from __future__ import annotations
import ast
import hashlib
import unittest
from pathlib import Path
APP_ROOT = Path(__file__).resolve().parents[1]
ORIGINAL_ROOT = APP_ROOT.parent
ROTATION_METHODS = {
"rotation_history",
"rotation_sector_members",
}
LADDER_ROTATION_BUILDERS = {
"_build_ladders",
"_build_sector_rotation",
}
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 top_level_functions(path: Path) -> dict[str, str]:
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
return {
node.name: ast.dump(node, include_attributes=False)
for node in tree.body
if isinstance(node, (ast.FunctionDef, ast.AsyncFunctionDef))
and node.name in LADDER_ROTATION_BUILDERS
}
def sha256(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
class LadderRotationSliceSourceEquivalenceTests(unittest.TestCase):
def test_rotation_service_methods_are_exact_original_ast(self) -> None:
original = class_methods(ORIGINAL_ROOT / "server.py", "DashboardService")
migrated = class_methods(
APP_ROOT / "backend" / "features" / "rotation" / "service.py",
"RotationServiceMixin",
)
self.assertEqual(set(migrated), ROTATION_METHODS)
for name in sorted(ROTATION_METHODS):
self.assertEqual(migrated[name], original[name], name)
def test_dashboard_service_no_longer_duplicates_rotation_methods(self) -> None:
remaining = class_methods(
APP_ROOT / "backend" / "application.py", "DashboardService"
)
self.assertTrue(ROTATION_METHODS.isdisjoint(remaining))
def test_ladder_and_rotation_builders_are_exact_original_ast(self) -> None:
self.assertEqual(
top_level_functions(ORIGINAL_ROOT / "tushare_client.py"),
top_level_functions(
APP_ROOT / "backend" / "data" / "providers" / "tushare_client.py"
),
)
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/ladder/page.js",
"static/pages/rotation/page.js",
):
self.assertEqual(
sha256(APP_ROOT / relative),
sha256(ORIGINAL_ROOT / relative),
relative,
)
if __name__ == "__main__":
unittest.main()
+5 -1
View File
@@ -20,6 +20,7 @@ ORIGINAL_ROOT = APP_ROOT.parent
MARKET_METHODS = {
"_tushare_client",
"_market_insights",
"get_dashboard",
"_dashboard_sentiment_ready",
"_display_compact_date",
@@ -127,12 +128,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"),
@@ -0,0 +1,195 @@
from __future__ import annotations
import ast
import hashlib
import unittest
from pathlib import Path
import market_insights
from backend.features.market import insights as canonical_insights
APP_ROOT = Path(__file__).resolve().parents[1]
ORIGINAL_ROOT = APP_ROOT.parent
MARKET_INSIGHT_METHODS = {
"__init__",
"_trade_context",
"_latest_feature_snapshot",
"_auction_session",
"_stock_master",
"_expectation_label",
"_auction_confirmation",
"_attention_score",
"_auction_candidates",
"_auction_theme_evidence",
"_auction_amount_history",
"_ensure_auction_amount_history",
"_with_auction_watchlist",
"_dynamic_auction_rows",
"auction_center",
"_theme_directory",
"theme_library",
"theme_detail",
"_parse_concepts",
"popularity",
"_hot_rows",
"_normalize_hot",
}
MARKET_SERVICE_METHODS = {"_market_insights"}
AUCTION_SERVICE_METHODS = {"auction_center"}
THEME_SERVICE_METHODS = {"theme_library", "theme_detail"}
POPULARITY_SERVICE_METHODS = {"popularity"}
DRAGON_TIGER_SERVICE_METHODS = {
"get_hot_money_profiles",
"get_dragon_tiger",
"_apply_seat_aliases",
}
AUCTION_REPOSITORY_METHODS = {
"upsert_auction_factors",
"auction_factor_dates",
"auction_factors_for_date",
}
POPULARITY_REPOSITORY_METHODS = {"upsert_popularity_factors"}
DRAGON_TIGER_REPOSITORY_METHODS = {
"list_seat_aliases",
"save_seat_alias",
"upsert_lhb_institutions",
}
TUSHARE_METHODS = {"hot_money_profiles", "dragon_tiger"}
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 MarketInsightsSliceSourceEquivalenceTests(unittest.TestCase):
def assert_methods_equal(
self,
original_path: Path,
original_class: str,
migrated_path: Path,
migrated_class: str,
names: set[str],
) -> None:
original = class_methods(original_path, original_class)
migrated = class_methods(migrated_path, migrated_class)
self.assertEqual(set(migrated), names)
for name in sorted(names):
self.assertEqual(migrated[name], original[name], name)
def test_shared_market_insight_service_is_exact_original_ast(self) -> None:
self.assert_methods_equal(
ORIGINAL_ROOT / "market_insights.py",
"MarketInsightsService",
APP_ROOT / "backend" / "features" / "market" / "insights.py",
"MarketInsightsService",
MARKET_INSIGHT_METHODS,
)
self.assertIs(market_insights.MarketInsightsService, canonical_insights.MarketInsightsService)
def test_dashboard_service_methods_are_exact_original_ast(self) -> None:
original = ORIGINAL_ROOT / "server.py"
mappings = (
("auction/service.py", "AuctionServiceMixin", AUCTION_SERVICE_METHODS),
("themes/service.py", "ThemeServiceMixin", THEME_SERVICE_METHODS),
("popularity/service.py", "PopularityServiceMixin", POPULARITY_SERVICE_METHODS),
("dragon_tiger/service.py", "DragonTigerServiceMixin", DRAGON_TIGER_SERVICE_METHODS),
)
for relative, class_name, names in mappings:
with self.subTest(relative=relative):
self.assert_methods_equal(
original,
"DashboardService",
APP_ROOT / "backend" / "features" / relative,
class_name,
names,
)
original_methods = class_methods(original, "DashboardService")
market_methods = class_methods(
APP_ROOT / "backend" / "features" / "market" / "service.py",
"MarketServiceMixin",
)
for name in MARKET_SERVICE_METHODS:
self.assertEqual(market_methods[name], original_methods[name], name)
def test_repository_methods_are_exact_original_ast(self) -> None:
original = ORIGINAL_ROOT / "database.py"
mappings = (
("auction/repository.py", "AuctionRepositoryMixin", AUCTION_REPOSITORY_METHODS),
("popularity/repository.py", "PopularityRepositoryMixin", POPULARITY_REPOSITORY_METHODS),
("dragon_tiger/repository.py", "DragonTigerRepositoryMixin", DRAGON_TIGER_REPOSITORY_METHODS),
)
for relative, class_name, names in mappings:
with self.subTest(relative=relative):
self.assert_methods_equal(
original,
"ReviewDatabase",
APP_ROOT / "backend" / "features" / relative,
class_name,
names,
)
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")
moved_service = (
MARKET_SERVICE_METHODS
| AUCTION_SERVICE_METHODS
| THEME_SERVICE_METHODS
| POPULARITY_SERVICE_METHODS
| DRAGON_TIGER_SERVICE_METHODS
)
moved_repository = (
AUCTION_REPOSITORY_METHODS
| POPULARITY_REPOSITORY_METHODS
| DRAGON_TIGER_REPOSITORY_METHODS
)
self.assertTrue(moved_service.isdisjoint(remaining_service))
self.assertTrue(moved_repository.isdisjoint(remaining_database))
def test_tushare_dragon_tiger_implementations_are_exact_original_ast(self) -> None:
original = class_methods(ORIGINAL_ROOT / "tushare_client.py", "TushareClient")
migrated = class_methods(
APP_ROOT / "backend" / "data" / "providers" / "tushare_client.py",
"TushareClient",
)
for name in sorted(TUSHARE_METHODS):
self.assertEqual(migrated[name], original[name], name)
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/auction/page.js",
"static/pages/themes/page.js",
"static/pages/popularity/page.js",
"static/pages/dragon-tiger/page.js",
):
self.assertEqual(
sha256(APP_ROOT / relative),
sha256(ORIGINAL_ROOT / relative),
relative,
)
if __name__ == "__main__":
unittest.main()
@@ -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()
+111
View File
@@ -0,0 +1,111 @@
from __future__ import annotations
import argparse
import hashlib
import http.cookiejar
import json
import urllib.error
import urllib.request
from pathlib import Path
from typing import Any
def request_json(
opener: urllib.request.OpenerDirector,
url: str,
payload: dict[str, Any] | None = None,
) -> tuple[int, Any]:
data = None
headers = {"Accept": "application/json"}
if payload is not None:
data = json.dumps(payload, ensure_ascii=False).encode("utf-8")
headers["Content-Type"] = "application/json"
request = urllib.request.Request(url, data=data, headers=headers)
try:
with opener.open(request, timeout=90) as response:
return response.status, json.loads(response.read().decode("utf-8"))
except urllib.error.HTTPError as exc:
return exc.code, json.loads(exc.read().decode("utf-8"))
def session(base_url: str, username: str, password: str) -> urllib.request.OpenerDirector:
opener = urllib.request.build_opener(
urllib.request.HTTPCookieProcessor(http.cookiejar.CookieJar())
)
status, body = request_json(
opener,
f"{base_url.rstrip('/')}/api/auth/login",
{"username": username, "password": password},
)
if status != 200 or not body.get("ok"):
raise RuntimeError(f"Login failed for {base_url}: HTTP {status} {body}")
return opener
def digest(value: Any) -> str:
content = json.dumps(
value, ensure_ascii=False, sort_keys=True, separators=(",", ":")
).encode("utf-8")
return hashlib.sha256(content).hexdigest()
def comparable(value: Any) -> Any:
if isinstance(value, dict):
return {
key: comparable(item)
for key, item in value.items()
if key != "request_id"
}
if isinstance(value, list):
return [comparable(item) for item in value]
return value
def main() -> None:
parser = argparse.ArgumentParser(description="Compare authenticated preservation APIs")
parser.add_argument("--original", required=True)
parser.add_argument("--migrated", required=True)
parser.add_argument("--username", required=True)
parser.add_argument("--password", required=True)
parser.add_argument("--output", type=Path, required=True)
parser.add_argument("endpoints", nargs="+")
args = parser.parse_args()
original = session(args.original, args.username, args.password)
migrated = session(args.migrated, args.username, args.password)
rows = []
all_equal = True
for endpoint in args.endpoints:
original_status, original_body = request_json(
original, f"{args.original.rstrip('/')}{endpoint}"
)
migrated_status, migrated_body = request_json(
migrated, f"{args.migrated.rstrip('/')}{endpoint}"
)
original_comparable = comparable(original_body)
migrated_comparable = comparable(migrated_body)
equal = original_status == migrated_status and original_comparable == migrated_comparable
all_equal = all_equal and equal
rows.append(
{
"endpoint": endpoint,
"original_status": original_status,
"migrated_status": migrated_status,
"original_sha256": digest(original_comparable),
"migrated_sha256": digest(migrated_comparable),
"equal": equal,
}
)
result = {"all_equal": all_equal, "endpoints": rows}
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(
json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
)
print(json.dumps(result, ensure_ascii=False, indent=2))
if not all_equal:
raise SystemExit(1)
if __name__ == "__main__":
main()
@@ -0,0 +1,92 @@
from __future__ import annotations
import argparse
import hashlib
import json
import sqlite3
from pathlib import Path
from typing import Any
def digest(value: Any) -> str:
content = json.dumps(
value, ensure_ascii=False, sort_keys=True, separators=(",", ":"), default=str
).encode("utf-8")
return hashlib.sha256(content).hexdigest()
def schema(connection: sqlite3.Connection) -> list[dict[str, Any]]:
rows = connection.execute(
"""
SELECT type, name, tbl_name, sql
FROM sqlite_master
WHERE name NOT LIKE 'sqlite_%'
ORDER BY type, name
"""
).fetchall()
return [dict(row) for row in rows]
def table_rows(connection: sqlite3.Connection, table: str) -> list[dict[str, Any]]:
quoted = '"' + table.replace('"', '""') + '"'
rows = [dict(row) for row in connection.execute(f"SELECT * FROM {quoted}").fetchall()]
return sorted(rows, key=lambda row: json.dumps(row, ensure_ascii=False, sort_keys=True, default=str))
def main() -> None:
parser = argparse.ArgumentParser(description="Compare preservation SQLite databases")
parser.add_argument("--original", type=Path, required=True)
parser.add_argument("--migrated", type=Path, required=True)
parser.add_argument("--output", type=Path, required=True)
parser.add_argument("tables", nargs="+")
args = parser.parse_args()
original = sqlite3.connect(args.original)
migrated = sqlite3.connect(args.migrated)
original.row_factory = sqlite3.Row
migrated.row_factory = sqlite3.Row
try:
original_schema = schema(original)
migrated_schema = schema(migrated)
tables = []
all_equal = original_schema == migrated_schema
for table in args.tables:
original_rows = table_rows(original, table)
migrated_rows = table_rows(migrated, table)
equal = original_rows == migrated_rows
all_equal = all_equal and equal
tables.append(
{
"table": table,
"original_count": len(original_rows),
"migrated_count": len(migrated_rows),
"original_sha256": digest(original_rows),
"migrated_sha256": digest(migrated_rows),
"equal": equal,
}
)
result = {
"all_equal": all_equal,
"schema": {
"object_count": len(original_schema),
"original_sha256": digest(original_schema),
"migrated_sha256": digest(migrated_schema),
"equal": original_schema == migrated_schema,
},
"tables": tables,
}
finally:
original.close()
migrated.close()
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(
json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8"
)
print(json.dumps(result, ensure_ascii=False, indent=2))
if not all_equal:
raise SystemExit(1)
if __name__ == "__main__":
main()
+46
View File
@@ -0,0 +1,46 @@
from __future__ import annotations
import argparse
import sys
from http.server import ThreadingHTTPServer
from pathlib import Path
def main() -> None:
parser = argparse.ArgumentParser(description="Run an isolated preservation runtime")
parser.add_argument("--runtime-root", type=Path, required=True)
parser.add_argument("--data-dir", type=Path, required=True)
parser.add_argument("--port", type=int, required=True)
args = parser.parse_args()
runtime_root = args.runtime_root.resolve()
data_dir = args.data_dir.resolve()
data_dir.mkdir(parents=True, exist_ok=True)
sys.path.insert(0, str(runtime_root))
if (runtime_root / "backend" / "bootstrap" / "config.py").is_file():
from backend.bootstrap import config
config.DATA_DIR = data_dir
config.PRIVATE_MENTOR_SKILLS_DIR = data_dir / "private-mentor-skills"
else:
import app_config as config
config.DATA_DIR = data_dir
config.PRIVATE_MENTOR_SKILLS_DIR = data_dir / "private-mentor-skills"
from server import RequestHandler, SERVICE
server = ThreadingHTTPServer(("127.0.0.1", args.port), RequestHandler)
print(f"Preservation runtime is running at http://127.0.0.1:{args.port}", flush=True)
try:
server.serve_forever()
except KeyboardInterrupt:
pass
finally:
SERVICE._background_stop.set()
server.server_close()
if __name__ == "__main__":
main()
@@ -0,0 +1,66 @@
# 切片 03:情绪周期、五类股池与涨停表现
> 基线:`a426432`(切片 02
> 回档标签:`xiaobai-preservation-slice-03-20260731`
> 结论:源码、API、数据库、真实页面和浏览器回归通过;最终视觉仍等待全站人工验收
## 1. 原实现归位
本切片只移动原版副本中的真实实现,没有从 `next/` 取用代码,也没有改写情绪公式、股池数据、
原因补全、表格、样式或交互。
| 原位置 | 新的唯一实现位置 | 原位置兼容 |
|---|---|---|
| `app/backend/application.py` 的 2 个情绪服务方法 | `app/backend/features/sentiment/service.py` | `DashboardService` 继承 `SentimentServiceMixin` |
| `app/backend/application.py` 的 6 个股池原因及事件补全方法 | `app/backend/features/pools/service.py` | `DashboardService` 继承 `PoolServiceMixin` |
| `app/database.py` 的 2 个原因覆盖方法 | `app/backend/features/pools/repository.py` | `ReviewDatabase` 继承 `PoolRepositoryMixin` |
| `app/sentiment_engine.py` | `app/backend/features/sentiment/engine.py` | 根模块为同一模块对象的兼容别名 |
五类股池、涨停梯队和涨停表现仍由切片 02 已归位的原 Tushare 总览实现生成,本切片没有建立第二套
计算或数据来源。
## 2. 等价证据
- `test_preservation_slice_sentiment_pools.py` 对 8 个业务方法和 2 个 Repository 方法逐项执行无位置
信息 AST 比较,全部与根目录原版 `server.py``database.py` 完全相同。
- 新的情绪引擎文件与原版 `sentiment_engine.py` SHA-256 完全相同;根级兼容模块与新模块是同一模块对象。
- 已归位的应用、行情服务、Tushare Provider、演示数据和选股模块直接导入新的唯一实现;Tushare
Provider 仅调整该导入,其全部类和函数 AST 继续与原版一致。
- 原版 `8784` 与迁移版 `8785` 在相同账号、日期和数据库副本上请求 `/api/dashboard`
`/api/sentiment/history`,JSON 状态、字段、值和顺序完全相同。
- 2026-07-30 的同请求结果均为:涨停 56、炸板 23、跌停 83、昨日涨停 81、情绪历史 20 日。
- 原版和迁移版数据库均为 62 个 schema 对象,schema 哈希均为
`17918327f8b919496e6630458293f9f777c7c24662625bb3fc0b64ff0a8fbeef`
- `config/api.config.json`、API 路径、鉴权和 `app/static/` 未修改。
- `app-light-1920x1080.png` 是真实迁移服务载入完成后的情绪周期页面,SHA-256 为
`e387417abbe0667e00875a8d4061b5546748ecf2452a692d06d078d516330dab`
## 3. 真实运行检查
- 迁移副本:`http://127.0.0.1:8785/`,管理员会话与缓存行情载入正常。
- 情绪周期:20 个连续交易日、当前阶段、评分构成和交易日明细均完整显示。
- 股池:涨停池 56 行、炸板池 23 行、跌停池 83 行、昨日涨停 81 行。
- 涨停表现:四档晋级率、市场宽度和今日结论均显示原版结果。
- 1920×1080 下六个页面横向溢出均为 0;日间、夜间背景与面板状态正常;浏览器控制台无迁移错误。
## 4. 自动验证
| 验证 | 结果 |
|---|---:|
| `python -m unittest discover -s tests -q` | 248 项通过 |
| `python -m unittest tests.test_preservation_slice_sentiment_pools -q` | 6 项通过 |
| 情绪、总览、缓存、iFinD 与前端契约专项集合 | 48 项通过 |
| `npx.cmd playwright test --reporter=dot` | 45 项通过 |
| `python -m compileall -q ...` | 通过 |
| `git diff --check` | 通过 |
Windows 下由 Playwright 自行创建临时静态服务器时,45 项完成后子进程无法回收;改为预先启动同一个
`8876` 静态服务器并让 Playwright 复用后,测试以零退出码正常结束,结果为 `45 passed (2.1m)`
## 5. 保留边界
- 板块轮动仍调用情绪历史公共函数,待切片 04 与市场天梯一并归位。
- 竞价、题材、人气和龙虎榜对股池数据的消费保持原调用路径,待切片 05 迁移。
- 根级情绪引擎兼容模块、`DashboardService``ReviewDatabase` 兼容面继续保留;数据库内尚未迁移的
选股统计方法仍走兼容别名,待切片 06 随完整方法一并归位。
- 没有删除待定代码、没有改动根目录正式数据库、没有切换 Docker/NAS。
Binary file not shown.

After

Width:  |  Height:  |  Size: 151 KiB

@@ -0,0 +1,58 @@
# 切片 04:市场天梯与板块轮动
> 基线:`b3555d2`(切片 03
> 回档标签:`xiaobai-preservation-slice-04-20260731`
> 结论:源码、API、真实页面和浏览器回归通过;最终视觉仍等待全站人工验收
## 1. 原实现归位
本切片从原版副本机械移动板块轮动服务,没有从 `next/` 取用代码,也没有修改天梯、轮动的计算、
排序、展开、配色、页面结构或交互。
| 原位置 | 新的唯一实现位置 | 原位置兼容 |
|---|---|---|
| `app/backend/application.py` 的 2 个轮动方法 | `app/backend/features/rotation/service.py` | `DashboardService` 继承 `RotationServiceMixin` |
| Tushare Provider 的天梯与轮动构造函数 | 保持 `app/backend/data/providers/tushare_client.py` | 切片 02 已归位的公共数据实现 |
市场天梯没有独立后端 API 或第二套计算,直接展示 `/api/dashboard` 中原 Tushare 实现生成的
`ladders`;因此没有为目录形式建立空的天梯服务。
## 2. 等价证据
- `test_preservation_slice_ladder_rotation.py` 对 2 个轮动服务方法逐项执行无位置信息 AST 比较,
全部与根目录原版 `server.py` 完全相同。
- `_build_ladders``_build_sector_rotation` 两个原数据构造函数的 AST 与根目录原版完全相同。
- 原版 `8784` 与迁移版 `8785` 在相同账号、日期和数据库副本上返回的天梯数据及 9 日轮动历史
JSON 逐字段完全相同。
- 成分股接口在当前外部网络状态下两版均返回 HTTP 400、`bad_request` 和相同的
`该板块成分股暂不可用:Tushare request failed:`,没有改变错误或增加静默降级。
- `config/api.config.json`、API 路径、鉴权、数据库 schema 和 `app/static/` 未修改。
## 3. 真实运行检查
- 市场天梯:8 个层级(含断层)、18 个首屏股票单元格、3 个结构分析模块正常;1920×1080 下
页面宽度无溢出,首板展开入口保留。
- 板块轮动:9 个交易日、每日 Top 12 共 108 个板块单元格、由远到近/由近到远两个排序入口正常;
1920×1080 下页面宽度无溢出并保持全页滚动。
- 日间模式页面控制台没有错误或警告。
- `app-light-ladder-1920x1080.png` SHA-256
`9e57d18d92e745fd92131f7bf08f21faaaa745476dd942cdaa2a703b9a7a303a`
- `app-light-rotation-1920x1080.png` SHA-256
`03c092bb40bd0eb672136dff853abffc87e9790840107c2f22e6cf31d5f83c09`
## 4. 自动验证
| 验证 | 结果 |
|---|---:|
| `python -m unittest discover -s tests -q` | 252 项通过 |
| `python -m unittest tests.test_preservation_slice_ladder_rotation -q` | 4 项通过 |
| 切片 02 至 04 与总览缓存专项集合 | 24 项通过 |
| `npx.cmd playwright test --reporter=dot` | 45 项通过 |
| `git diff --check` | 通过 |
## 5. 保留边界
- 成分股接口依赖的日行情与因子持久化方法仍由原 `ReviewDatabase` 提供,因其同时服务智能选股,
待切片 06 随完整共享职责归位。
- 天梯和轮动前端资产保持原位,切片 10 再按页面职责归档;当前没有复制或改写。
- 没有删除待定代码、没有改动根目录正式数据库、没有切换 Docker/NAS。
Binary file not shown.

After

Width:  |  Height:  |  Size: 139 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 198 KiB

@@ -0,0 +1,69 @@
# 切片 05:集合竞价、题材库、人气热榜与龙虎榜
> 基线:`814e757`(切片 04
> 回档标签:`xiaobai-preservation-slice-05-20260731`
> 结论:源码、API、数据库、真实页面和全量回归通过;最终视觉仍等待全站人工验收
## 1. 原实现归位
本切片没有从`next/`取用代码,也没有重写计算、页面或接口。集合竞价、题材库和人气热榜原本
共享`MarketInsightsService`,其中竞价候选会直接调用人气榜热度数据,因此整体移动到公共行情
领域,避免拆出互相复制的实现;各页面入口仍按功能目录归位。
| 原位置 | 新的唯一实现位置 | 兼容方式 |
|---|---|---|
| `app/market_insights.py` | `app/backend/features/market/insights.py` | 根级模块导出同一类对象 |
| `DashboardService`竞价入口 | `app/backend/features/auction/service.py` | `AuctionServiceMixin` |
| `DashboardService`题材入口 | `app/backend/features/themes/service.py` | `ThemeServiceMixin` |
| `DashboardService`人气入口 | `app/backend/features/popularity/service.py` | `PopularityServiceMixin` |
| `DashboardService`龙虎榜及游资档案 | `app/backend/features/dragon_tiger/service.py` | `DragonTigerServiceMixin` |
| 竞价、人气、龙虎榜持久化方法 | 对应功能目录的`repository.py` | `ReviewDatabase`继承原接口 |
Tushare Provider 中`hot_money_profiles``dragon_tiger`继续保持切片02归位的唯一实现,没有为目录
形式再制造一套数据构造逻辑。
## 2. 源码与接口等价
- `test_preservation_slice_market_insights.py`逐项比较22个市场洞察方法、8个页面服务方法、7个
Repository方法和2个Tushare方法,全部与根目录原版无位置信息AST一致。
- `DashboardService``ReviewDatabase`不再重复保留已移动方法;根级`market_insights`与新模块
暴露同一个`MarketInsightsService`类对象。
- 原版`8784`和迁移版`8785`使用同一数据库的独立副本,集合竞价、题材库、题材详情、人气热榜、
龙虎榜、游资档案和席位别名共7个真实API状态码及JSON一致。
- 题材详情在当前外部网络条件下两版均返回HTTP 400;差分只排除每次请求随机生成的
`request_id`,错误码与错误内容仍完全一致。
- 完整接口摘要见`api-diff.json`
## 3. 数据库差分
- 两个副本均为62个schema对象,哈希均为
`60a4e044f0ddb6502e44b45f65bedc1a0a31d4bd596f386d4ccfe153a6c8ddd1`
- `auction_factors` 511914行、`popularity_factors` 232行、`lhb_institution_daily` 47行、
`seat_aliases` 0行、`stock_master` 5535行均逐行一致。
- 完整表计数与哈希见`database-diff.json`;运行数据库副本已在验收后删除,未提交凭据或正式数据。
## 4. 真实浏览器检查
- 1920×1080日间模式检查集合竞价、题材库、人气热榜和龙虎榜四页;均无横向溢出,控制台无
错误或警告。
- 集合竞价载入30行重点候选;题材库载入394个题材及选中题材成分股;人气热榜载入3个摘要
模块和200行综合榜;龙虎榜按当前缓存显示既有不可用空态。
- 四页HTML、主JS、CSS及各自页面JS与根目录原版字节哈希一致。
- 截图SHA-256
- `app-light-auction-1920x1080.png``8065086c8f2b360aeb1004bd60429f3e2d7f1b8c872ec4a965e9a5d0c016b915`
- `app-light-themes-1920x1080.png``80e1e41101497ee7213e1dadfaa4c9572a1c47b3495edd09e36745ccdb639402`
- `app-light-popularity-1920x1080.png``614d6b7770f4e9ec72c059ab8a4129486df9eff5c4dd7e8279cc68ebcb80a36b`
- `app-light-dragon-tiger-1920x1080.png``2f1e2ad3a9bc3874c73bae884fc8744177cef354c7176559da1453fab1f85993`
## 5. 自动验证与保留边界
| 验证 | 结果 |
|---|---:|
| `python -m unittest discover -s tests -q` | 258项通过 |
| `python -m unittest tests.test_preservation_slice_market_insights -q` | 6项通过 |
| `npx.cmd playwright test --reporter=dot` | 45项通过 |
| `git diff --check` | 通过 |
- 竞价、人气和龙虎榜因子同时服务切片06智能选股,迁移后仍由`ReviewDatabase`原方法名暴露。
- 前端资产保持原位置,切片10再按页面职责归档;本切片没有改DOM、CSS、动画或交互。
- 没有删除待定代码、没有修改根目录正式数据库、没有切换Docker/NAS。
@@ -0,0 +1,61 @@
{
"all_equal": true,
"endpoints": [
{
"endpoint": "/api/auction?trade_date=2026-07-29",
"original_status": 200,
"migrated_status": 200,
"original_sha256": "523144cc14d876577b7d518cdf38fd2722b13a8f01ed5d2e20dcc38f6a2624ce",
"migrated_sha256": "523144cc14d876577b7d518cdf38fd2722b13a8f01ed5d2e20dcc38f6a2624ce",
"equal": true
},
{
"endpoint": "/api/themes?trade_date=2026-07-29",
"original_status": 200,
"migrated_status": 200,
"original_sha256": "95c2ad418f18d877d94ec2a71fe6fafd5e329b069d87a9300f7fcac92d4ba5d1",
"migrated_sha256": "95c2ad418f18d877d94ec2a71fe6fafd5e329b069d87a9300f7fcac92d4ba5d1",
"equal": true
},
{
"endpoint": "/api/themes/detail?code=885001.TI&trade_date=2026-07-29",
"original_status": 400,
"migrated_status": 400,
"original_sha256": "b11a3314b172d3ad6ba28d969fcd6a9a2a4b49e7d29ed804496a4dfecd2a364a",
"migrated_sha256": "b11a3314b172d3ad6ba28d969fcd6a9a2a4b49e7d29ed804496a4dfecd2a364a",
"equal": true
},
{
"endpoint": "/api/popularity?trade_date=2026-07-29",
"original_status": 200,
"migrated_status": 200,
"original_sha256": "e62a93c41f7c3f95c3d47f8ccaedaa809563c8dd65c04dd54b164f73e6014e94",
"migrated_sha256": "e62a93c41f7c3f95c3d47f8ccaedaa809563c8dd65c04dd54b164f73e6014e94",
"equal": true
},
{
"endpoint": "/api/dragon-tiger?trade_date=2026-07-29",
"original_status": 200,
"migrated_status": 200,
"original_sha256": "a3998b935377d5fd0673ec5b9214d1b0a64680d155c61e6cff49d0d5fcf0e843",
"migrated_sha256": "a3998b935377d5fd0673ec5b9214d1b0a64680d155c61e6cff49d0d5fcf0e843",
"equal": true
},
{
"endpoint": "/api/dragon-tiger/profiles",
"original_status": 200,
"migrated_status": 200,
"original_sha256": "dfb1e534c018ec12fd8d8ee0e7fe0f234e73dc7715d7f0a948f74ef8d12d779a",
"migrated_sha256": "dfb1e534c018ec12fd8d8ee0e7fe0f234e73dc7715d7f0a948f74ef8d12d779a",
"equal": true
},
{
"endpoint": "/api/seat-aliases",
"original_status": 200,
"migrated_status": 200,
"original_sha256": "2b0fb0a6b3e353c69158d61221c2200e4199d0d60dd0b9d99702a22eaa917a78",
"migrated_sha256": "2b0fb0a6b3e353c69158d61221c2200e4199d0d60dd0b9d99702a22eaa917a78",
"equal": true
}
]
}
Binary file not shown.

After

Width:  |  Height:  |  Size: 162 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 66 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 136 KiB

Binary file not shown.

After

Width:  |  Height:  |  Size: 136 KiB

@@ -0,0 +1,51 @@
{
"all_equal": true,
"schema": {
"object_count": 62,
"original_sha256": "60a4e044f0ddb6502e44b45f65bedc1a0a31d4bd596f386d4ccfe153a6c8ddd1",
"migrated_sha256": "60a4e044f0ddb6502e44b45f65bedc1a0a31d4bd596f386d4ccfe153a6c8ddd1",
"equal": true
},
"tables": [
{
"table": "auction_factors",
"original_count": 511914,
"migrated_count": 511914,
"original_sha256": "3d0470787adaf7c4cf5f15f5ad9fa1d67c8fcd4807285ac264eeacdcf5054cd1",
"migrated_sha256": "3d0470787adaf7c4cf5f15f5ad9fa1d67c8fcd4807285ac264eeacdcf5054cd1",
"equal": true
},
{
"table": "popularity_factors",
"original_count": 232,
"migrated_count": 232,
"original_sha256": "3f4c61a13ceaa1ed9ecb28f86241a8a478a6757d6a44b46f432f73c0226bba7f",
"migrated_sha256": "3f4c61a13ceaa1ed9ecb28f86241a8a478a6757d6a44b46f432f73c0226bba7f",
"equal": true
},
{
"table": "lhb_institution_daily",
"original_count": 47,
"migrated_count": 47,
"original_sha256": "1f847eac34d2ba576b429da208667f3b67b591d36ee66800803605bbc447970c",
"migrated_sha256": "1f847eac34d2ba576b429da208667f3b67b591d36ee66800803605bbc447970c",
"equal": true
},
{
"table": "seat_aliases",
"original_count": 0,
"migrated_count": 0,
"original_sha256": "4f53cda18c2baa0c0354bb5f9a3ecbe5ed12ab4d8e11ba873c2f11161202b945",
"migrated_sha256": "4f53cda18c2baa0c0354bb5f9a3ecbe5ed12ab4d8e11ba873c2f11161202b945",
"equal": true
},
{
"table": "stock_master",
"original_count": 5535,
"migrated_count": 5535,
"original_sha256": "8656e2d189d3520433fb3552e17998e4bf17bcf6838403cddc6719282b23e792",
"migrated_sha256": "8656e2d189d3520433fb3552e17998e4bf17bcf6838403cddc6719282b23e792",
"equal": true
}
]
}
+5 -5
View File
@@ -1,6 +1,6 @@
{
"schema_version": 1,
"updated_at": "2026-07-31T00:56:00+08:00",
"updated_at": "2026-07-31T02:42:00+08:00",
"status": "active",
"migration_mode": "behavior_preserving_source_migration",
"source_of_truth": "current_original_webapp_runtime_and_source",
@@ -9,10 +9,10 @@
"failed_roots": [
"next"
],
"current_slice": "slice-03-sentiment-pools-performance",
"last_completed_slice": "slice-02-market-search-charts-data",
"last_checkpoint": "xiaobai-preservation-slice-02-20260731",
"next_action": "capture_slice-03_sentiment_pool_performance_contracts_then_move_original_implementations",
"current_slice": "slice-06-screener-custom-tracking",
"last_completed_slice": "slice-05-auction-themes-popularity-dragon-tiger",
"last_checkpoint": "xiaobai-preservation-slice-05-20260731",
"next_action": "capture_slice-06_screener_custom_selection_and_tracking_contracts_then_move_original_implementations",
"authoritative_documents": [
"AGENTS.md",
"docs/migration/原版保真迁移总纲.md",
+43 -1
View File
@@ -1,6 +1,6 @@
# 小白复盘保真迁移账本
> 当前状态:正式迁移,切片02“公共行情、搜索、详情、图表与数据网关”已完成
> 当前状态:正式迁移,切片05“集合竞价、题材库、人气热榜与龙虎榜”已完成
本账本是上下文恢复和人工审计的连续记录。任何迁移提交必须在同一提交中更新本文件及
`保真迁移状态.json`
@@ -22,6 +22,9 @@
| 2026-07-30 | `41329943c4878fc09ed82ec376eb93ab151e4092` | 完成只读资产清查并由用户批准`app/`结构 | 开始切片00 |
| 2026-07-31 | `xiaobai-preservation-slice-01-20260731` | 启动、HTTP、账号、会员与系统管理原实现归位 | 自动差分通过,进入切片02 |
| 2026-07-31 | `xiaobai-preservation-slice-02-20260731` | 公共行情、搜索、详情、图表与数据适配原实现归位 | 自动与浏览器差分通过,进入切片03 |
| 2026-07-31 | `xiaobai-preservation-slice-03-20260731` | 情绪周期、五类股池与涨停表现原实现归位 | 自动、API与浏览器差分通过,进入切片04 |
| 2026-07-31 | `xiaobai-preservation-slice-04-20260731` | 市场天梯与板块轮动原实现归位 | 自动、API与浏览器差分通过,进入切片05 |
| 2026-07-31 | `xiaobai-preservation-slice-05-20260731` | 集合竞价、题材库、人气热榜与龙虎榜原实现归位 | 自动、API、数据库与浏览器差分通过,进入切片06 |
## 资产处置登记
@@ -36,6 +39,15 @@
| `commonReviewColumns`等5个前端函数 | 疑似无引用符号 | 未发现静态调用 | 待定 | 待删隔离账本 | 仍需动态注册与浏览器覆盖 | 保留 |
| `wencai_saved_queries`及其方法 | 历史兼容数据 | 当前无前端入口 | 待定 | 数据库兼容区 | 不允许在迁移期破坏旧库 | 保留 |
| 现有7层CSS | 视觉运行资产 | 全部页面和主题 | 原样保留后逐页归档 | `app/frontend/` | 必须通过截图与计算样式差分 | 保留 |
| `sentiment_engine.py` | 情绪周期计算 | 总览、轮动、选股 | 移动并保留兼容别名 | `app/backend/features/sentiment/engine.py` | 文件哈希与原版一致;248项Python与45项Playwright通过 | 已移动 |
| `DashboardService`情绪及股池原因方法 | 业务服务 | 情绪页、五类股池、涨停表现 | 按职责机械移动 | `app/backend/features/sentiment/``app/backend/features/pools/` | 8个方法AST与原版一致;真实API完全一致 | 已移动 |
| `ReviewDatabase`原因覆盖方法 | 持久化 | 股池原因人工覆盖 | 按职责机械移动 | `app/backend/features/pools/repository.py` | 2个方法AST与原版一致;数据库schema哈希一致 | 已移动 |
| `DashboardService`板块轮动方法 | 业务服务 | 板块轮动页 | 按职责机械移动 | `app/backend/features/rotation/service.py` | 2个方法AST、真实API与原版一致 | 已移动 |
| Tushare天梯与轮动构造函数 | 公共数据计算 | 市场天梯、板块轮动 | 原位置保持唯一实现 | `app/backend/data/providers/tushare_client.py` | 2个构造函数AST与原版一致 | 已归位 |
| `MarketInsightsService` | 共享市场洞察服务 | 集合竞价、题材库、人气热榜 | 整体机械移动,保留唯一共享实现 | `app/backend/features/market/insights.py` | 22个方法AST与原版一致;根级模块为同一类对象别名 | 已移动 |
| `DashboardService`竞价、题材、人气与龙虎榜方法 | 业务服务 | 切片05四类页面与API | 按职责机械移动 | `app/backend/features/auction/``themes/``popularity/``dragon_tiger/` | 8个方法AST、7个真实API与原版一致 | 已移动 |
| `ReviewDatabase`竞价、人气与龙虎榜方法 | 持久化 | 市场洞察及后续智能选股 | 按职责机械移动并保持Mixin原接口 | `app/backend/features/auction/repository.py``popularity/repository.py``dragon_tiger/repository.py` | 7个方法AST一致;62个schema对象及5张关键表逐行一致 | 已移动 |
| Tushare游资名录与龙虎榜实现 | 公共数据计算 | 龙虎榜与游资档案 | 原位置保持唯一实现 | `app/backend/data/providers/tushare_client.py` | 2个方法AST与原版一致 | 已归位 |
处置只允许:`原样保留``移动``合并重复``待定``确认废弃`
@@ -73,6 +85,36 @@
- 回档:标签`xiaobai-preservation-slice-02-20260731`
- 完整证据:`docs/migration/evidence/slice-02/README.md`
已完成切片:`slice-03-sentiment-pools-performance`
- 原版基线:提交`a426432`,即切片02回档点。
- 迁移范围:情绪计算引擎、2个情绪服务方法、6个股池原因与iFinD事件补全方法、2个原因覆盖持久化方法。
- 兼容边界:根级`sentiment_engine.py`保留同一模块对象别名;股池生成仍使用切片02的原Tushare总览实现。
- API与数据库:原版`8784`和迁移版`8785`的总览、情绪历史JSON完全一致;两库schema均为62项且哈希一致。
- 验收:248项Python测试、6项切片源码等价测试、45项Playwright测试及六个真实页面流程通过。
- 回档:标签`xiaobai-preservation-slice-03-20260731`
- 完整证据:`docs/migration/evidence/slice-03/README.md`
已完成切片:`slice-04-ladder-rotation`
- 原版基线:提交`b3555d2`,即切片03回档点。
- 迁移范围:2个板块轮动服务方法;市场天梯继续使用切片02已归位的原Tushare数据构造实现。
- 兼容边界:`DashboardService`通过`RotationServiceMixin`保持所有原调用;天梯不制造空服务或第二套计算。
- API与错误:天梯与9日轮动历史JSON完全一致;成分股两版均返回同一Tushare外部失败语义。
- 验收:252项Python测试、4项切片源码等价测试、45项Playwright测试及两个真实页面流程通过。
- 回档:标签`xiaobai-preservation-slice-04-20260731`
- 完整证据:`docs/migration/evidence/slice-04/README.md`
已完成切片:`slice-05-auction-themes-popularity-dragon-tiger`
- 原版基线:提交`814e757`,即切片04回档点。
- 迁移范围:共享市场洞察服务、竞价/题材/人气入口、龙虎榜与游资档案服务、7个相关持久化方法。
- 兼容边界:根级`market_insights.py`保留同一类对象别名;竞价与人气共用候选热度逻辑,不复制第二套实现。
- API与数据库:7个真实API逐字段一致,仅排除每次请求必然变化的`request_id`;62个schema对象与5张关键表完全一致。
- 验收:258项Python测试、6项切片源码等价测试、45项Playwright测试及四个真实页面流程通过。
- 回档:标签`xiaobai-preservation-slice-05-20260731`
- 完整证据:`docs/migration/evidence/slice-05/README.md`
## 决策记录
| 日期 | 决策 | 原因 |