migration: preserve market insights slice

This commit is contained in:
leefer
2026-07-31 02:41:56 +08:00
parent 814e75730a
commit c184974bc4
30 changed files with 2487 additions and 1759 deletions
+8 -301
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,10 +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 backend.features.themes import ThemeServiceMixin
from backend.data.providers.tushare_client import TushareClient, TushareError, _sector_coverage_issue
@@ -142,6 +145,10 @@ class DashboardService(
SentimentServiceMixin,
PoolServiceMixin,
RotationServiceMixin,
AuctionServiceMixin,
ThemeServiceMixin,
PopularityServiceMixin,
DragonTigerServiceMixin,
):
def __init__(self) -> None:
runtime = load_runtime_settings()
@@ -753,28 +760,9 @@ class DashboardService(
}
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:
@@ -3074,287 +3062,6 @@ class DashboardService(
)
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 _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()
+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
+9
View File
@@ -14,6 +14,7 @@ from backend.bootstrap.config import (
from backend.data.providers.ifind_client import IfindError
from backend.data.providers.tushare_client import TushareClient, TushareError
from backend.features.market.charts import ChartDataError
from backend.features.market.insights import MarketInsightsService
from backend.features.sentiment.engine import SENTIMENT_ENGINE_VERSION
@@ -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:
@@ -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)
+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))
+6 -140
View File
@@ -8,8 +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
@@ -24,8 +27,11 @@ def _optional_float(value: Any) -> float | None:
class ReviewDatabase(
AccountRepositoryMixin,
AuctionRepositoryMixin,
DragonTigerRepositoryMixin,
MarketRepositoryMixin,
PoolRepositoryMixin,
PopularityRepositoryMixin,
SystemSettingsRepositoryMixin,
):
def __init__(self, path: Path) -> None:
@@ -839,24 +845,6 @@ class ReviewDatabase(
return cursor.rowcount > 0
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:
@@ -1056,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 = [
@@ -1130,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 ""
@@ -1227,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 -1312
View File
File diff suppressed because it is too large Load Diff
@@ -20,6 +20,7 @@ ORIGINAL_ROOT = APP_ROOT.parent
MARKET_METHODS = {
"_tushare_client",
"_market_insights",
"get_dashboard",
"_dashboard_sentiment_ready",
"_display_compact_date",
@@ -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()
+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,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-31T01:57: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-05-auction-themes-popularity-dragon-tiger",
"last_completed_slice": "slice-04-ladder-rotation",
"last_checkpoint": "xiaobai-preservation-slice-04-20260731",
"next_action": "capture_slice-05_auction_theme_popularity_dragon_tiger_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",
+16 -1
View File
@@ -1,6 +1,6 @@
# 小白复盘保真迁移账本
> 当前状态:正式迁移,切片04“市场天梯与板块轮动”已完成
> 当前状态:正式迁移,切片05“集合竞价、题材库、人气热榜与龙虎榜”已完成
本账本是上下文恢复和人工审计的连续记录。任何迁移提交必须在同一提交中更新本文件及
`保真迁移状态.json`
@@ -24,6 +24,7 @@
| 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 |
## 资产处置登记
@@ -43,6 +44,10 @@
| `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与原版一致 | 已归位 |
处置只允许:`原样保留``移动``合并重复``待定``确认废弃`
@@ -100,6 +105,16 @@
- 回档:标签`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`
## 决策记录
| 日期 | 决策 | 原因 |