Compare commits

..
46 changed files with 1877 additions and 530 deletions
+64 -29
View File
@@ -1,40 +1,75 @@
# Architecture
# Candidate architecture
The normative governance contract is documented in
`docs/governance/architecture-standard.md`. This file describes the currently deployed
shape; the standard defines the target boundaries and the rules applied during migration.
`app/` is the behavior-preserving modular source tree accepted by the user on 2026-08-01.
The original `webapp/` runtime remains the deployment rollback baseline until an explicitly
approved switch. `next/` is a rejected, frozen implementation and is not a source for this
directory.
The application intentionally keeps a small deployment footprint: one Python process, one
SQLite database, and a build-free browser client. The internal boundaries are nevertheless
explicit so new features do not bypass account isolation or data-quality rules.
The application deliberately remains a modular monolith: one Python process, one SQLite WAL
database, and a build-free HTML/CSS/JavaScript client. The migration changed source ownership
and imports, not the technology stack or observable product behavior.
## Backend boundaries
## Runtime path
- `server.py`: application services and HTTP request/response wiring.
- `api_access.py`: the single authorization policy for authenticated, member, and admin APIs.
- `app_config.py`: runtime paths, local environment loading, and shared input validation.
- `database.py`: SQLite schema, migrations, and persistence operations.
- `tushare_client.py` and `realtime_aggregator.py`: external market-data adapters.
- `sentiment_engine.py`, `screener.py`, and `heaven_engine.py`: deterministic domain logic.
- `mentor_agent.py`, `heaven_agent.py`, and `llm_strategy.py`: bounded LLM adapters.
```text
browser
-> frontend/shared/api.js
-> backend HTTP transport and feature HTTP mixins
-> feature services
-> repositories / DataGateway / LLMGateway
-> SQLite / market providers / model providers
## Data ownership
background scheduler
-> backend/jobs
-> the same feature services and repositories
```
Public market snapshots, stock factors, built-in strategies, limit-up reasons, seat aliases,
and sector-element mappings are shared. Only administrators can modify shared knowledge.
## Source ownership
Watchlists, review notes, custom strategies, screener runs, mentor conversations, birth data,
alerts, trading journals, and assistant conversations are owned by a user ID and must be
queried with that ID. LLM features additionally require active membership.
- `server.py` is the stable command/import facade. Runtime composition lives in
`backend/application.py` and `backend/bootstrap/`.
- `backend/http/` owns common authentication, request IDs, responses, static delivery, and
error normalization. Feature-specific transport handlers live beside their feature.
Exact POST endpoints that only delegate to one of those handlers use the explicit maps in
`backend/application.py`; endpoints with path parameters, body handling, or special error
semantics remain visible control flow in `RequestHandler`.
- `backend/features/<feature>/` owns the mechanically moved service, repository, HTTP, agent,
or deterministic calculation code for that product area.
- `backend/data/` owns provider construction, source policy, provenance, units, freshness,
coverage, and display-versus-calculation eligibility.
- `backend/database/` owns connection management, ordered migrations, and narrow repository
adapters. Root `database.py` remains the legacy schema/composition anchor and combines the
feature repository mixins; do not add feature queries to it.
- `backend/jobs/` owns job definitions, locks, retries, idempotency, and persisted run state.
- `backend/llm/` owns model selection, membership/quota checks, fallback, provider transport,
streaming rules, and call audit. Feature agents only prepare messages and interpret
feature-specific results.
- `frontend/shared/` is the only browser API/state/Shell/component boundary.
- `frontend/pages/` owns page-local behavior. The original runtime was split mechanically;
source markers and preservation tests prove that the pieces reassemble to the audited
original, apart from explicitly registered trial retirements.
- `frontend/styles/`, `frontend/shared/tokens.css`, and the Wentian page stylesheet preserve
the approved cascade and light/dark/mobile behavior.
- `config/` is the versioned registry for pages, features, APIs, datasets, quality rules,
jobs, and the generated candidate architecture inventory.
## Data integrity
Root modules such as `screener.py`, `tushare_client.py`, and `mentor_agent.py` are compatibility
aliases to canonical modules. They contain no second implementation and remain only because
the original public import surface is part of the preservation contract.
Production reads never synthesize market prices. A failed live request may use the latest real
snapshot at or before the requested date. When no real snapshot exists, the API reports that
the data is unavailable. Demo builders remain test fixtures only.
## Non-negotiable maintenance rules
## Change contract
1. Preserve account ownership in every user-private query and test it with two accounts.
2. Browser requests go through `frontend/shared/api.js`; provider calls go through the data
boundary; model calls go through `backend/llm/`.
3. Calculation datasets fail closed when required source, date, unit, freshness, or coverage
evidence is missing. Display fallbacks do not silently enter calculations.
4. Do not implement logic in both a root compatibility module and a canonical module.
5. Do not remove compatibility or uncertain code without reference scanning, old/new
differential evidence, browser checks, and manual acceptance.
6. Run `python tools/verify_baseline.py` for every change and add `--e2e` when runtime or
frontend behavior can be affected.
New endpoints must be added to `api_access.required_role` when they need member or admin
access. New user-owned tables must include `user_id`, an ownership index, and cross-account
tests. API payload compatibility is protected by the Python and Playwright suites.
The authoritative migration constraints and handoff procedure are in
`../docs/migration/原版保真迁移总纲.md` and
`../docs/migration/人工维护与本地切换指南.md`.
+9 -5
View File
@@ -2,23 +2,27 @@
一个面向 A 股盘后复盘的本地 Web 工作台。后端使用 Python 访问 Tushare Pro,前端不依赖构建工具。
本目录是从原版源码逐项移动、机械拆分并完成差分验证与用户人工验收的模块化正式源码,
不是依据规格书重新开发的第二套产品。正式部署切换前,`webapp/`根目录继续作为当前部署与
回档基线;冻结的`next/`不得用于部署或后续开发。目录职责见[ARCHITECTURE.md](ARCHITECTURE.md)。
当前包含集合竞价、涨停池、炸板池、跌停板、昨日涨停、涨停表现、市场天梯、板块轮动、题材库、人气热榜、龙虎榜和个人复盘工作区。交易日快照与同步记录保存在本地 SQLite 数据库 `data/review.db`
集合竞价中心采用盘前生命周期:9:15 前显示预告,9:15–9:25 明确等待最终竞价,9:25–9:30 自动读取并重试最终竞价筛选,9:30 后停止更新并冻结为复盘归档。当前 Tushare 只提供 9:25 最终竞价快照,不将其表述为动态虚拟撮合行情。
第三阶段加入了机构席位、席位别名、个股复权日 K、资金流、自选股、涨停原因修订、个股笔记、每日复盘和历史数据回补。
股票代码在桌面端悬停后会显示分时与日 K 快速预览,默认优先展示日 K;移动端点击代码后从底部打开预览面板。股票详情以及板块、题材、指数详情均可在日 K 与最新分时之间切换。日 K 复用个股详情缓存;分时使用隔离的东方财富分钟图表源和短时内存缓存,只负责展示,不写入主行情、不参与情绪、选股或问天计算。图表源不可用时界面会明确显示“分时不可用”,不会使用日 K 数据模拟分时走势。
股票代码在桌面端悬停后会显示分时与日 K 快速预览,默认优先展示日 K;移动端点击代码后从底部打开预览面板。股票详情以及板块、题材、指数详情均可在日 K 与最新分时之间切换。日 K 复用个股详情缓存;分时优先使用 iFinD,东方财富仅作隔离的展示兜底,并使用短时内存缓存。图表数据不写入主行情、不参与情绪、选股或问天计算不可用时明确显示“分时不可用”,不会用日 K 模拟分时走势。
智能选股模块包含 45 日全市场因子库、六阶段市场识别、七套内置策略、受控公式 DSL、自然语言策略编译、候选排名和滚动回测。竞价涨幅、竞价成交额、竞价换手率与竞价量比随因子数据一并同步,可用于自定义公式和历史回测。首次使用需在页面点击“同步因子数据”。未配置 LLM 时使用本地策略模板;配置兼容 API 后自动切换为主模型编译,主模型失败时自动使用辅助模型,两者均支持独立连通性测试
智能选股包含六阶段盘后候选、29 套精选策略、自定义公式 DSL、自然语言公式编译、候选排名和滚动回测。阶段与精选策略在当日行情更新后由后台确定性计算;自定义选股由用户手动执行,LLM 只负责编译自然语言条件,不参与候选筛选。竞价、估值、财务、资金、人气和席位等字段按已登记的数据可用性进入因子库,缺失时明确显示覆盖问题
每次选股结果会自动进入五交易日持续跟踪,展示 T+1 开盘/收盘、T+3、T+5、最大涨幅与最大回撤。提醒中心支持手工日期提醒,并在策略首日反馈和五日跟踪完成时生成账号私有的站内提醒。
候选只有经用户手动加入后才进入五交易日持续跟踪,展示 T+1 开盘/收盘、T+3、T+5、最大涨幅与最大回撤。提醒中心支持手工日期提醒,并在策略首日反馈和五日跟踪完成时生成账号私有的站内提醒。
问师模块会读取当前复盘、近十日市场情绪、涨跌停、昨日反馈、板块轮动、市场阶段、龙虎榜和指定个股数据,再按选中的游资思维 Skill 进行单师对话。对话记录按账号、老师和交易日期保存在服务端;主模型不可用时自动切换辅助模型。
新增公开问师角色时,在 `游资skills` 下增加一个包含 `SKILL.md` 的独立目录,并在 `游资skills/mentor_catalog.json` 中登记素材等级与结构质检。管理员私有角色放在 `data/private-mentor-skills`,该目录不进入 Git 或 Docker 镜像,且只会出现在管理员的问师列表中。系统会从 Skill 的 frontmatter、一级标题、核心模型和引用语中自动生成角色信息,无需修改注册代码。
问天模块包含三个相互独立的部分:观势以市场数据生成三才六爻,用于观察“势”,行情缺失或自动取象明显偏差时可显式手动校准六爻,人工结果与自动来源严格区分;观气依据干支、精确节气、五运六气及客主加临关系观察“运”,行业五行仅作传统取象归类;观心通过30秒静心、六次三枚铜钱起卦、察念和解卦完成一次不输入问题的问心仪式。卦象、干支、节气与气机关系均由本地确定性程序计算,LLM只负责解释,不参与起卦或改动结果。
问天模块包含三个相互独立的部分:观势以市场数据生成三才六爻,用于观察“势”,行情缺失或自动取象明显偏差时可显式手动校准六爻,人工结果与自动来源严格区分;观气依据干支、精确节气、五运六气及客主加临关系观察“运”,行业五行仅作传统取象归类;观心先准备1秒,再完成5轮“吸3秒、顿2秒、呼4秒”,随后以六次三枚铜钱起卦、察念和解卦完成一次不输入问题的问心仪式。卦象、干支、节气与气机关系均由本地确定性程序计算,LLM只负责解释,不参与起卦或改动结果。
问天模块使用项目本地的 `lunar-python` 计算历法,并使用 `data/iching_zh.json` 中的固定六十四卦、卦辞和爻辞。第三方授权见 `THIRD_PARTY_NOTICES.md`
@@ -27,7 +31,7 @@
## 启动
```powershell
cd webapp
cd webapp\app
python -m pip install -r requirements.txt
python server.py
```
+43 -79
View File
@@ -507,6 +507,40 @@ class DashboardService(
SERVICE = DashboardService()
PUBLIC_POST_HANDLERS = {
"/api/auth/register": "auth_register",
"/api/auth/login": "auth_login",
}
AUTHENTICATED_POST_HANDLERS = {
"/api/auth/logout": "auth_logout",
"/api/account/birth-profile": "save_birth_profile",
"/api/account/password": "change_password",
"/api/alerts": "save_alert",
"/api/trades": "save_trade_entry",
"/api/assistant/chat": "stream_assistant_chat",
"/api/admin/settings": "save_system_settings",
"/api/admin/settings/test": "test_system_llm_settings",
"/api/admin/membership": "save_membership",
"/api/admin/refresh": "start_background_refresh",
"/api/watchlist": "save_watchlist",
"/api/notes": "save_note",
"/api/reasons": "save_reason",
"/api/seat-aliases": "save_seat_alias",
"/api/heaven/sector-phases": "save_sector_phase_override",
"/api/backfill": "backfill_data",
"/api/screener/sync": "sync_screener_data",
"/api/screener/compile": "compile_screener_strategy",
"/api/screener/strategies": "save_screener_strategy",
"/api/screener/run": "run_screener",
"/api/screener/tracking/refresh": "refresh_screener_tracking",
"/api/mentors/chat": "stream_mentor_chat",
"/api/heaven/hexagram": "heaven_hexagram",
"/api/heaven/personal": "heaven_personal",
"/api/heaven/interpret": "heaven_interpret",
}
class RequestHandler(
AccountHttpMixin,
SystemHttpMixin,
@@ -522,6 +556,13 @@ class RequestHandler(
application_service = SERVICE
route_registry = ROUTES
def _dispatch_named_handler(self, path: str, handlers: dict[str, str]) -> bool:
handler_name = handlers.get(path)
if handler_name is None:
return False
getattr(self, handler_name)()
return True
def do_GET(self) -> None:
parsed = urlparse(self.path)
if parsed.path == "/api/health":
@@ -864,24 +905,13 @@ class RequestHandler(
def do_POST(self) -> None:
parsed = urlparse(self.path)
if parsed.path == "/api/auth/register":
self.auth_register()
return
if parsed.path == "/api/auth/login":
self.auth_login()
if self._dispatch_named_handler(parsed.path, PUBLIC_POST_HANDLERS):
return
if not self.require_auth() or not self.require_csrf():
return
if not self.require_access("POST", parsed.path):
return
if parsed.path == "/api/auth/logout":
self.auth_logout()
return
if parsed.path == "/api/account/birth-profile":
self.save_birth_profile()
return
if parsed.path == "/api/account/password":
self.change_password()
if self._dispatch_named_handler(parsed.path, AUTHENTICATED_POST_HANDLERS):
return
alert_read_match = re.fullmatch(r"/api/alerts/(\d+)/read", parsed.path)
if alert_read_match:
@@ -895,57 +925,6 @@ class RequestHandler(
{"ok": True, **SERVICE.mark_all_alerts_read(str(body.get("as_of") or ""))}
)
return
if parsed.path == "/api/alerts":
self.save_alert()
return
if parsed.path == "/api/trades":
self.save_trade_entry()
return
if parsed.path == "/api/assistant/chat":
self.stream_assistant_chat()
return
if parsed.path == "/api/admin/settings":
self.save_system_settings()
return
if parsed.path == "/api/admin/settings/test":
self.test_system_llm_settings()
return
if parsed.path == "/api/admin/membership":
self.save_membership()
return
if parsed.path == "/api/admin/refresh":
self.start_background_refresh()
return
if parsed.path == "/api/watchlist":
self.save_watchlist()
return
if parsed.path == "/api/notes":
self.save_note()
return
if parsed.path == "/api/reasons":
self.save_reason()
return
if parsed.path == "/api/seat-aliases":
self.save_seat_alias()
return
if parsed.path == "/api/heaven/sector-phases":
self.save_sector_phase_override()
return
if parsed.path == "/api/backfill":
self.backfill_data()
return
if parsed.path == "/api/screener/sync":
self.sync_screener_data()
return
if parsed.path == "/api/screener/compile":
self.compile_screener_strategy()
return
if parsed.path == "/api/screener/strategies":
self.save_screener_strategy()
return
if parsed.path == "/api/screener/run":
self.run_screener()
return
if parsed.path == "/api/screener/tracking":
try:
result = SERVICE.add_screener_tracking(self.read_json_body())
@@ -953,9 +932,6 @@ class RequestHandler(
except (ValueError, json.JSONDecodeError) as exc:
self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)
return
if parsed.path == "/api/screener/tracking/refresh":
self.refresh_screener_tracking()
return
if parsed.path == "/api/mentors/preferences":
try:
result = SERVICE.save_mentor_preferences(self.read_json_body())
@@ -963,18 +939,6 @@ class RequestHandler(
except (ValueError, json.JSONDecodeError) as exc:
self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)
return
if parsed.path == "/api/mentors/chat":
self.stream_mentor_chat()
return
if parsed.path == "/api/heaven/hexagram":
self.heaven_hexagram()
return
if parsed.path == "/api/heaven/personal":
self.heaven_personal()
return
if parsed.path == "/api/heaven/interpret":
self.heaven_interpret()
return
self.send_json({"error": "Not found"}, HTTPStatus.NOT_FOUND)
def do_DELETE(self) -> None:
+20 -50
View File
@@ -1,11 +1,10 @@
from __future__ import annotations
import json
import time
import urllib.error
import urllib.request
from typing import Any
from backend.llm import transport as llm_transport
class HeavenAgentError(RuntimeError):
pass
@@ -24,45 +23,33 @@ def interpret_heaven(
if not api_key or not model:
raise HeavenAgentError("LLM API Key 或模型尚未配置。")
system_prompt = _system_prompt(mode)
payload = json.dumps(
messages = [
{"role": "system", "content": system_prompt},
{
"model": model,
"messages": [
{"role": "system", "content": system_prompt},
{
"role": "user",
"content": json.dumps(context, ensure_ascii=False, separators=(",", ":")),
},
],
"stream": False,
"role": "user",
"content": json.dumps(context, ensure_ascii=False, separators=(",", ":")),
},
ensure_ascii=False,
).encode("utf-8")
request = urllib.request.Request(
f"{base_url.rstrip('/')}/chat/completions",
data=payload,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
"User-Agent": "XiaobaiReviewWeb/0.7",
},
method="POST",
)
started = time.perf_counter()
]
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
result = json.loads(response.read().decode("utf-8"))
answer = str(result["choices"][0]["message"]["content"]).strip()
result = llm_transport.chat_completion(
api_key=api_key,
base_url=base_url,
model=model,
messages=messages,
timeout=timeout,
user_agent="XiaobaiReviewWeb/0.7",
)
answer = str(result.content).strip()
if not answer:
raise KeyError("empty response")
except urllib.error.HTTPError as exc:
raise HeavenAgentError(_http_error_message(exc)) from exc
except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, KeyError, IndexError) as exc:
except llm_transport.OpenAIHTTPError as exc:
raise HeavenAgentError(exc.describe("问天模型调用失败")) from exc
except (llm_transport.OpenAITransportError, KeyError) as exc:
raise HeavenAgentError(f"问天模型调用失败:{exc}") from exc
return {
"answer": answer,
"model": model,
"latency_ms": round((time.perf_counter() - started) * 1000),
"latency_ms": result.latency_ms,
}
@@ -99,20 +86,3 @@ def _system_prompt(mode: str) -> str:
全文控制在180至350个中文字符。只写一句卦意;一小段动爻与之卦;最后三句极短的问心句。
不要重述六条爻辞,不猜用户未说出口的问题,不以吉凶二字替代思考,不给出股票涨跌预测。语气安静、克制,越短越有余味。
""".strip()
def _http_error_message(exc: urllib.error.HTTPError) -> str:
detail = ""
try:
payload = json.loads(exc.read().decode("utf-8", errors="replace"))
error = payload.get("error")
if isinstance(error, dict):
detail = str(error.get("message") or error.get("code") or "")
elif error:
detail = str(error)
elif payload.get("message"):
detail = str(payload["message"])
except (json.JSONDecodeError, OSError):
detail = ""
suffix = f"{detail[:300]}" if detail else ""
return f"问天模型调用失败(HTTP {exc.code}{suffix}"
+1 -10
View File
@@ -12,6 +12,7 @@ from datetime import datetime, time as dt_time, timedelta
from threading import Lock
from typing import Any, ClassVar
from backend.bootstrap.config import tushare_code as _stock_market_code
from backend.data.providers.ifind_client import IfindError, IfindHttpClient
@@ -475,16 +476,6 @@ def _ifind_point(row: dict[str, Any]) -> dict[str, Any] | None:
}
def _stock_market_code(code: str) -> str:
if code.startswith(("4", "8", "9")):
suffix = "BJ"
elif code.startswith("6"):
suffix = "SH"
else:
suffix = "SZ"
return f"{code}.{suffix}"
def _number(value: Any) -> float:
try:
return float(value or 0)
+14 -63
View File
@@ -3,14 +3,12 @@ from __future__ import annotations
import json
import re
import time
import urllib.error
import urllib.request
from collections.abc import Iterator
from dataclasses import dataclass
from pathlib import Path
from typing import Any
from llm_stream import OpenAIStreamAccumulator
from backend.llm import transport as llm_transport
class MentorAgentError(RuntimeError):
@@ -195,50 +193,20 @@ def stream_with_mentor(
messages = [{"role": "system", "content": system_prompt}]
messages.extend(history[-10:])
messages.append({"role": "user", "content": question})
payload = json.dumps(
{"model": model, "messages": messages, "stream": True},
ensure_ascii=False,
).encode("utf-8")
request = urllib.request.Request(
f"{base_url.rstrip('/')}/chat/completions",
data=payload,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
"User-Agent": "XiaobaiReviewWeb/0.6",
"Accept": "text/event-stream",
},
method="POST",
)
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
yielded = False
accumulator = OpenAIStreamAccumulator()
for raw_line in response:
line = raw_line.decode("utf-8", errors="replace").strip()
if not line or line.startswith(":"):
continue
if line.startswith("data:"):
line = line[5:].strip()
if line == "[DONE]":
break
try:
result = json.loads(line)
except json.JSONDecodeError:
continue
choices = result.get("choices") or []
if not choices:
continue
choice = choices[0] or {}
content = accumulator.feed(choice)
if content:
yielded = True
yield str(content)
if not yielded:
raise MentorAgentError("问师模型未返回有效内容。")
except urllib.error.HTTPError as exc:
raise MentorAgentError(_http_error_message(exc)) from exc
except (urllib.error.URLError, TimeoutError, OSError) as exc:
yield from llm_transport.stream_chat_completion(
api_key=api_key,
base_url=base_url,
model=model,
messages=messages,
timeout=timeout,
user_agent="XiaobaiReviewWeb/0.6",
)
except llm_transport.OpenAIEmptyResponseError as exc:
raise MentorAgentError("问师模型未返回有效内容。") from exc
except llm_transport.OpenAIHTTPError as exc:
raise MentorAgentError(exc.describe("问师模型调用失败")) from exc
except llm_transport.OpenAITransportError as exc:
raise MentorAgentError(f"问师模型调用失败:{exc}") from exc
@@ -298,20 +266,3 @@ def _parse_frontmatter(content: str) -> dict[str, str]:
def _first_sentence(text: str) -> str:
compact = " ".join(line.strip() for line in text.splitlines() if line.strip())
return re.split(r"[。;]", compact, maxsplit=1)[0].strip()
def _http_error_message(exc: urllib.error.HTTPError) -> str:
detail = ""
try:
payload = json.loads(exc.read().decode("utf-8", errors="replace"))
error = payload.get("error")
if isinstance(error, dict):
detail = str(error.get("message") or error.get("code") or "")
elif error:
detail = str(error)
elif payload.get("message"):
detail = str(payload["message"])
except (json.JSONDecodeError, OSError):
detail = ""
suffix = f"{detail[:300]}" if detail else ""
return f"问师模型调用失败(HTTP {exc.code}{suffix}"
+13 -43
View File
@@ -1,12 +1,10 @@
from __future__ import annotations
import json
import urllib.error
import urllib.request
from collections.abc import Iterator
from typing import Any
from llm_stream import OpenAIStreamAccumulator
from backend.llm import transport as llm_transport
class ReviewAssistantError(RuntimeError):
@@ -27,48 +25,20 @@ def stream_review_assistant(
messages = [{"role": "system", "content": _system_prompt(context)}]
messages.extend(history[-12:])
messages.append({"role": "user", "content": question})
request = urllib.request.Request(
f"{base_url.rstrip('/')}/chat/completions",
data=json.dumps(
{"model": model, "messages": messages, "stream": True}, ensure_ascii=False
).encode("utf-8"),
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
"User-Agent": "XiaobaiReviewWeb/1.0",
"Accept": "text/event-stream",
},
method="POST",
)
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
yielded = False
accumulator = OpenAIStreamAccumulator()
for raw_line in response:
line = raw_line.decode("utf-8", errors="replace").strip()
if not line or line.startswith(":"):
continue
if line.startswith("data:"):
line = line[5:].strip()
if line == "[DONE]":
break
try:
payload = json.loads(line)
except json.JSONDecodeError:
continue
choices = payload.get("choices") or []
if not choices:
continue
choice = choices[0] or {}
content = accumulator.feed(choice)
if content:
yielded = True
yield str(content)
if not yielded:
raise ReviewAssistantError("智能解读未返回有效内容。")
except urllib.error.HTTPError as exc:
yield from llm_transport.stream_chat_completion(
api_key=api_key,
base_url=base_url,
model=model,
messages=messages,
timeout=timeout,
user_agent="XiaobaiReviewWeb/1.0",
)
except llm_transport.OpenAIEmptyResponseError as exc:
raise ReviewAssistantError("智能解读未返回有效内容。") from exc
except llm_transport.OpenAIHTTPError as exc:
raise ReviewAssistantError(f"智能解读服务暂不可用({exc.code})。") from exc
except (urllib.error.URLError, TimeoutError, OSError) as exc:
except llm_transport.OpenAITransportError as exc:
raise ReviewAssistantError("智能解读连接中断,请稍后重试。") from exc
+27 -73
View File
@@ -1,11 +1,9 @@
from __future__ import annotations
import json
import time
import urllib.error
import urllib.request
from typing import Any
from backend.llm import transport as llm_transport
from screener import FACTOR_FIELDS, REGIMES
@@ -21,39 +19,25 @@ def test_llm_connection(
) -> dict[str, Any]:
if not api_key or not model:
raise LLMCompilerError("API Key 或模型未配置。")
endpoint = f"{base_url.rstrip('/')}/chat/completions"
payload = json.dumps(
{
"model": model,
"messages": [{"role": "user", "content": "只回复 OK"}],
"stream": False,
},
ensure_ascii=False,
).encode("utf-8")
request = urllib.request.Request(
endpoint,
data=payload,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
"User-Agent": "XiaobaiReviewWeb/0.5",
},
method="POST",
)
started = time.perf_counter()
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
result = json.loads(response.read().decode("utf-8"))
reply = str(result["choices"][0]["message"]["content"]).strip()
except urllib.error.HTTPError as exc:
raise LLMCompilerError(_http_error_message(exc)) from exc
except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, KeyError, IndexError) as exc:
result = llm_transport.chat_completion(
api_key=api_key,
base_url=base_url,
model=model,
messages=[{"role": "user", "content": "只回复 OK"}],
timeout=timeout,
user_agent="XiaobaiReviewWeb/0.5",
)
reply = str(result.content).strip()
except llm_transport.OpenAIHTTPError as exc:
raise LLMCompilerError(exc.describe("模型连接测试失败")) from exc
except llm_transport.OpenAITransportError as exc:
raise LLMCompilerError(f"模型连接测试失败:{exc}") from exc
return {
"ok": True,
"model": model,
"reply": reply[:100],
"latency_ms": round((time.perf_counter() - started) * 1000),
"latency_ms": result.latency_ms,
}
@@ -67,7 +51,6 @@ def compile_strategy_with_llm(
) -> dict[str, Any]:
if not api_key or not model:
raise LLMCompilerError("尚未配置 LLM API Key 或模型。")
endpoint = f"{base_url.rstrip('/')}/chat/completions"
schema = {
"name": "策略名称",
"description": "策略说明",
@@ -90,57 +73,28 @@ def compile_strategy_with_llm(
"退潮和冰点策略必须提高门槛并允许结果为空。"
f"严格遵循以下结构:{json.dumps(schema, ensure_ascii=False)}"
)
payload = json.dumps(
{
"model": model,
"messages": [
try:
result = llm_transport.chat_completion(
api_key=api_key,
base_url=base_url,
model=model,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": prompt[:3000]},
],
"stream": False,
},
ensure_ascii=False,
).encode("utf-8")
request = urllib.request.Request(
endpoint,
data=payload,
headers={
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
"User-Agent": "XiaobaiReviewWeb/0.4",
},
method="POST",
)
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
result = json.loads(response.read().decode("utf-8"))
content = result["choices"][0]["message"]["content"].strip()
timeout=timeout,
user_agent="XiaobaiReviewWeb/0.4",
)
content = result.content.strip()
if content.startswith("```"):
content = content.strip("`")
if content.startswith("json"):
content = content[4:].strip()
compiled = json.loads(content)
except urllib.error.HTTPError as exc:
raise LLMCompilerError(_http_error_message(exc).replace("模型连接测试", "LLM 策略编译")) from exc
except (urllib.error.URLError, TimeoutError, json.JSONDecodeError, KeyError, IndexError) as exc:
except llm_transport.OpenAIHTTPError as exc:
raise LLMCompilerError(exc.describe("LLM 策略编译失败")) from exc
except (llm_transport.OpenAITransportError, json.JSONDecodeError) as exc:
raise LLMCompilerError(f"LLM 策略编译失败:{exc}") from exc
compiled["compiler"] = "llm"
compiled["model"] = model
return compiled
def _http_error_message(exc: urllib.error.HTTPError) -> str:
detail = ""
try:
payload = json.loads(exc.read().decode("utf-8", errors="replace"))
error = payload.get("error")
if isinstance(error, dict):
detail = str(error.get("message") or error.get("code") or "")
elif error:
detail = str(error)
elif payload.get("message"):
detail = str(payload["message"])
except (json.JSONDecodeError, OSError):
detail = ""
suffix = f"{detail[:300]}" if detail else ""
return f"模型连接测试失败(HTTP {exc.code}{suffix}"
+148
View File
@@ -0,0 +1,148 @@
from __future__ import annotations
import json
import time
import urllib.error
import urllib.request
from collections.abc import Iterator
from dataclasses import dataclass
from typing import Any
from .stream import OpenAIStreamAccumulator
class OpenAITransportError(RuntimeError):
pass
class OpenAIHTTPError(OpenAITransportError):
def __init__(self, code: int, detail: str = "") -> None:
super().__init__(f"HTTP {code}")
self.code = code
self.detail = detail
def describe(self, label: str) -> str:
suffix = f"{self.detail[:300]}" if self.detail else ""
return f"{label}HTTP {self.code}{suffix}"
class OpenAIEmptyResponseError(OpenAITransportError):
pass
@dataclass(frozen=True)
class OpenAIChatCompletion:
content: Any
latency_ms: int
def chat_completion(
*,
api_key: str,
base_url: str,
model: str,
messages: list[dict[str, Any]],
timeout: int,
user_agent: str,
) -> OpenAIChatCompletion:
request = _request(api_key, base_url, model, messages, user_agent, stream=False)
started = time.perf_counter()
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
result = json.loads(response.read().decode("utf-8"))
content = result["choices"][0]["message"]["content"]
except urllib.error.HTTPError as exc:
raise OpenAIHTTPError(exc.code, _http_error_detail(exc)) from exc
except (
urllib.error.URLError,
TimeoutError,
json.JSONDecodeError,
KeyError,
IndexError,
) as exc:
raise OpenAITransportError(str(exc)) from exc
return OpenAIChatCompletion(
content=content,
latency_ms=round((time.perf_counter() - started) * 1000),
)
def stream_chat_completion(
*,
api_key: str,
base_url: str,
model: str,
messages: list[dict[str, Any]],
timeout: int,
user_agent: str,
) -> Iterator[str]:
request = _request(api_key, base_url, model, messages, user_agent, stream=True)
yielded = False
try:
with urllib.request.urlopen(request, timeout=timeout) as response:
accumulator = OpenAIStreamAccumulator()
for raw_line in response:
line = raw_line.decode("utf-8", errors="replace").strip()
if not line or line.startswith(":"):
continue
if line.startswith("data:"):
line = line[5:].strip()
if line == "[DONE]":
break
try:
result = json.loads(line)
except json.JSONDecodeError:
continue
choices = result.get("choices") or []
if not choices:
continue
content = accumulator.feed(choices[0] or {})
if content:
yielded = True
yield str(content)
except urllib.error.HTTPError as exc:
raise OpenAIHTTPError(exc.code, _http_error_detail(exc)) from exc
except (urllib.error.URLError, TimeoutError, OSError) as exc:
raise OpenAITransportError(str(exc)) from exc
if not yielded:
raise OpenAIEmptyResponseError("empty response")
def _request(
api_key: str,
base_url: str,
model: str,
messages: list[dict[str, Any]],
user_agent: str,
*,
stream: bool,
) -> urllib.request.Request:
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {api_key}",
"User-Agent": user_agent,
}
if stream:
headers["Accept"] = "text/event-stream"
return urllib.request.Request(
f"{base_url.rstrip('/')}/chat/completions",
data=json.dumps(
{"model": model, "messages": messages, "stream": stream},
ensure_ascii=False,
).encode("utf-8"),
headers=headers,
method="POST",
)
def _http_error_detail(exc: urllib.error.HTTPError) -> str:
try:
payload = json.loads(exc.read().decode("utf-8", errors="replace"))
error = payload.get("error")
if isinstance(error, dict):
return str(error.get("message") or error.get("code") or "")
if error:
return str(error)
return str(payload.get("message") or "")
except (json.JSONDecodeError, OSError):
return ""
+11 -6
View File
@@ -1,12 +1,15 @@
# Governance Registries
These registries describe the approved product surface during architecture migration.
These registries describe the approved product surface of the modular preservation candidate.
- `pages.config.json`: primary page identity, navigation group, access expectation, scrolling,
and mobile composition policy.
- `features.config.json`: feature ownership, backend access class, data scope, and availability.
- `api.config.json`: transitional inventory of current routes, generated from `server.py` and
assigned to a feature owner.
- `api.config.json`: current routes generated from the preserved `server.py` API surface, with
one feature owner and backend access class per route. Its dispatcher implementation lives in
`backend/application.py`.
- `architecture-inventory.json`: generated inventory of candidate pages, routes, tables,
providers, model entry points, CSS layers, and remaining code hotspots.
- `data-fields.config.json`: canonical data products, provider eligibility, intended use, and
known blocked datasets.
- `data-quality.config.json`: freshness, coverage, units, adjustment, point-in-time, and
@@ -14,13 +17,15 @@ These registries describe the approved product surface during architecture migra
- `jobs.config.json`: background schedules, dependencies, lock keys, retry policy, timeouts,
and output versions.
During Stage 04 these files are contract inputs, not runtime replacements. Backend access in
`api_access.py` remains authoritative until the HTTP governance phase switches it atomically.
Frontend visibility remains a presentation concern and never grants backend access.
The registries are governance contracts, not substitutes for runtime authorization. Backend
access in `backend/http/routes.py` is authoritative; frontend visibility is only a presentation
concern and never grants access.
Regenerate the transitional API inventory after a route change:
```shell
python tools/build_api_registry.py
python tools/build_api_registry.py --check
python tools/build_architecture_inventory.py
python tools/build_architecture_inventory.py --check
```
+346
View File
@@ -0,0 +1,346 @@
{
"schema_version": 1,
"captured_from": "app accepted modular runtime",
"runtime": {
"http_server": "http.server.ThreadingHTTPServer",
"application_processes": 1,
"database": "SQLite WAL",
"frontend": "build-free HTML/CSS/JavaScript",
"container_port": 8765
},
"counts": {
"primary_pages": 16,
"api_exact_paths": 53,
"api_prefixes": 0,
"api_patterns": 11,
"database_tables": 36
},
"pages": [
{
"id": "sentimentCycleView",
"title": "情绪周期"
},
{
"id": "limitPool",
"title": "涨停池"
},
{
"id": "brokenView",
"title": "炸板池"
},
{
"id": "downView",
"title": "跌停板"
},
{
"id": "yesterdayView",
"title": "昨日涨停"
},
{
"id": "performanceView",
"title": "涨停表现"
},
{
"id": "ladderView",
"title": "市场天梯"
},
{
"id": "rotationView",
"title": "板块轮动"
},
{
"id": "auctionView",
"title": "集合竞价"
},
{
"id": "themeLibraryView",
"title": "题材库"
},
{
"id": "popularityView",
"title": "人气热榜"
},
{
"id": "dragonView",
"title": "龙虎榜"
},
{
"id": "screenerView",
"title": "智能选股"
},
{
"id": "mentorView",
"title": "问师"
},
{
"id": "heavenView",
"title": "问天"
},
{
"id": "reviewWorkspaceView",
"title": "我的复盘"
}
],
"api": {
"exact": [
"/api/account/birth-profile",
"/api/account/password",
"/api/account/status",
"/api/admin/membership",
"/api/admin/refresh",
"/api/admin/settings",
"/api/admin/settings/test",
"/api/alerts",
"/api/alerts/read-all",
"/api/assistant/chat",
"/api/assistant/messages",
"/api/auction",
"/api/auth/login",
"/api/auth/logout",
"/api/auth/me",
"/api/auth/register",
"/api/backfill",
"/api/chart/intraday",
"/api/dashboard",
"/api/dragon-tiger",
"/api/dragon-tiger/profiles",
"/api/health",
"/api/heaven/hexagram",
"/api/heaven/interpret",
"/api/heaven/personal",
"/api/heaven/readings",
"/api/heaven/sector-phases",
"/api/heaven/setup",
"/api/mentors/chat",
"/api/mentors/messages",
"/api/mentors/preferences",
"/api/mentors/setup",
"/api/notes",
"/api/popularity",
"/api/realtime-aggregate/health",
"/api/reasons",
"/api/rotation/history",
"/api/rotation/members",
"/api/screener/compile",
"/api/screener/run",
"/api/screener/setup",
"/api/screener/strategies",
"/api/screener/sync",
"/api/screener/tracking",
"/api/screener/tracking/refresh",
"/api/search",
"/api/search/detail",
"/api/seat-aliases",
"/api/sentiment/history",
"/api/themes",
"/api/themes/detail",
"/api/trades",
"/api/watchlist"
],
"prefixes": [],
"patterns": [
"/api/alerts/(\\d+)",
"/api/alerts/(\\d+)/read",
"/api/heaven/readings/(\\d+)",
"/api/heaven/sector-phases/(.+)",
"/api/notes/(\\d+)",
"/api/screener/strategies/(\\d+)",
"/api/screener/tracking/(\\d+)",
"/api/stock/(\\d{6})",
"/api/stock/(\\d{6})/preview",
"/api/trades/(\\d+)",
"/api/watchlist/(\\d{6})"
]
},
"database_tables": [
"users",
"user_sessions",
"user_credentials",
"user_birth_profiles",
"system_settings",
"llm_usage",
"dashboard_snapshots",
"sync_runs",
"data_snapshots",
"watchlist",
"review_notes",
"reason_overrides",
"seat_aliases",
"sector_phase_overrides",
"stock_master",
"daily_bars",
"benchmark_bars",
"daily_indicators",
"fundamental_indicators",
"moneyflow_daily",
"auction_factors",
"earnings_events",
"popularity_factors",
"lhb_institution_daily",
"screener_strategies",
"screener_runs",
"mentor_messages",
"mentor_preferences",
"wencai_saved_queries",
"strategy_tracks",
"alerts",
"trade_entries",
"assistant_messages",
"heaven_readings",
"job_runs",
"schema_migrations"
],
"background_job_methods": [
"_background_refresh_tick"
],
"external_data_adapters": [
{
"provider": "tushare",
"path": "backend/data/providers/tushare_client.py",
"runtime_role": "primary deterministic market data"
},
{
"provider": "ifind",
"path": "backend/data/providers/ifind_client.py",
"runtime_role": "realtime, charts, snapshots, enrichment"
},
{
"provider": "eastmoney",
"path": "backend/features/market/charts.py",
"runtime_role": "display chart fallback"
},
{
"provider": "eastmoney",
"path": "backend/data/realtime.py",
"runtime_role": "isolated realtime observation"
},
{
"provider": "tencent",
"path": "backend/data/realtime.py",
"runtime_role": "index observation fallback"
}
],
"llm_entrypoints": [
{
"function": "stream_with_mentor",
"path": "backend/features/mentor/agent.py"
},
{
"function": "interpret_heaven",
"path": "backend/features/heaven/agent.py"
},
{
"function": "stream_review_assistant",
"path": "backend/features/review/agent.py"
},
{
"function": "compile_strategy_with_llm",
"path": "backend/features/screener/compiler.py"
},
{
"function": "test_llm_connection",
"path": "backend/features/screener/compiler.py"
}
],
"llm_transport": [
{
"function": "chat_completion",
"path": "backend/llm/transport.py"
},
{
"function": "stream_chat_completion",
"path": "backend/llm/transport.py"
}
],
"css_layers": [
"/shared/tokens.css?v=20260729-1",
"/styles/styles.css",
"/styles/renovation.css?v=20260725-5",
"/styles/redesign-v2.css?v=20260728-1",
"/styles/design-system.css?v=20260728-4",
"/styles/theme.css?v=20260728-2",
"/pages/heaven/page.css?v=20260728-7"
],
"code_hotspots": [
{
"path": "frontend/styles/styles.css",
"bytes": 361776,
"lines": 15465
},
{
"path": "frontend/styles/redesign-v2.css",
"bytes": 263539,
"lines": 8570
},
{
"path": "frontend/index.html",
"bytes": 135019,
"lines": 1892
},
{
"path": "backend/features/screener/engine.py",
"bytes": 108552,
"lines": 2213
},
{
"path": "backend/data/providers/tushare_client.py",
"bytes": 94329,
"lines": 2175
},
{
"path": "frontend/app.js",
"bytes": 89213,
"lines": 1939
},
{
"path": "frontend/pages/heaven/page.js",
"bytes": 86493,
"lines": 1830
},
{
"path": "frontend/styles/renovation.css",
"bytes": 83949,
"lines": 1553
},
{
"path": "frontend/pages/heaven/page.css",
"bytes": 73222,
"lines": 1084
},
{
"path": "backend/features/heaven/service.py",
"bytes": 63123,
"lines": 1303
},
{
"path": "backend/features/market/insights.py",
"bytes": 58150,
"lines": 1314
},
{
"path": "frontend/pages/market/runtime.js",
"bytes": 55720,
"lines": 1333
},
{
"path": "backend/features/heaven/engine.py",
"bytes": 51670,
"lines": 1181
},
{
"path": "backend/application.py",
"bytes": 48749,
"lines": 1129
},
{
"path": "frontend/styles/theme.css",
"bytes": 36427,
"lines": 1253
},
{
"path": "database.py",
"bytes": 33284,
"lines": 746
}
]
}
+4 -5
View File
@@ -26,6 +26,7 @@ RETIRED_FRONTEND_SOURCE_RANGES = (
(9022, 9027),
(9052, 9055),
)
AUDITED_FRONTEND_SOURCE_LINE_COUNT = 9283
def sha256(path: Path) -> str:
@@ -51,12 +52,10 @@ def reassembled_frontend_runtime() -> str:
assembled.append(content)
next_line = end + 1
original_line_count = len(
(ORIGINAL_STATIC / "app.js").read_text(encoding="utf-8").splitlines()
)
if next_line != original_line_count + 1:
if next_line != AUDITED_FRONTEND_SOURCE_LINE_COUNT + 1:
raise AssertionError(
f"app.js source coverage ended at {next_line - 1}, expected {original_line_count}"
"app.js source coverage ended at "
f"{next_line - 1}, expected {AUDITED_FRONTEND_SOURCE_LINE_COUNT}"
)
return "".join(assembled)
+9
View File
@@ -4,6 +4,7 @@ import json
import re
import unittest
from pathlib import Path
from unittest.mock import patch
from tests.preservation_helpers import reassembled_frontend_runtime
@@ -46,6 +47,14 @@ class FrontendBoundaryTests(unittest.TestCase):
self.assertIn("const state = window.XiaobaiState.create({", app)
self.assertNotIn("const state = {", app)
def test_candidate_runtime_reassembly_does_not_require_original_static(self) -> None:
with patch(
"tests.preservation_helpers.ORIGINAL_STATIC",
ROOT / "missing-original-static",
):
app = reassembled_frontend_runtime()
self.assertIn("function openView(", app)
def test_runtime_page_registry_matches_governance_registry(self) -> None:
expected = json.loads(
(ROOT / "config" / "pages.config.json").read_text(encoding="utf-8")
+83
View File
@@ -0,0 +1,83 @@
from __future__ import annotations
import unittest
from backend.application import (
AUTHENTICATED_POST_HANDLERS,
PUBLIC_POST_HANDLERS,
RequestHandler,
)
class HttpDispatchContractTests(unittest.TestCase):
@staticmethod
def handler(path: str, calls: list[str]) -> RequestHandler:
handler = RequestHandler.__new__(RequestHandler)
handler.path = path
handler.require_auth = lambda: calls.append("auth") or True
handler.require_csrf = lambda: calls.append("csrf") or True
handler.require_access = lambda method, route: (
calls.append(f"access:{method}:{route}") or True
)
return handler
def test_named_handlers_are_real_registered_post_routes(self) -> None:
all_handlers = {**PUBLIC_POST_HANDLERS, **AUTHENTICATED_POST_HANDLERS}
self.assertEqual(
set(PUBLIC_POST_HANDLERS) & set(AUTHENTICATED_POST_HANDLERS), set()
)
for path, handler_name in all_handlers.items():
with self.subTest(path=path):
route = RequestHandler.route_registry.resolve("POST", path)
self.assertIsNotNone(route)
self.assertTrue(callable(getattr(RequestHandler, handler_name)))
expected_access = "public" if path in PUBLIC_POST_HANDLERS else None
if expected_access:
self.assertEqual(route.access, expected_access)
else:
self.assertNotEqual(route.access, "public")
def test_public_post_dispatches_without_authentication(self) -> None:
for path, handler_name in PUBLIC_POST_HANDLERS.items():
with self.subTest(path=path):
calls: list[str] = []
handler = self.handler(path, calls)
handler.require_auth = lambda: (_ for _ in ()).throw(
AssertionError("public route required authentication")
)
setattr(handler, handler_name, lambda: calls.append(handler_name))
RequestHandler.do_POST(handler)
self.assertEqual(calls, [handler_name])
def test_authenticated_post_preserves_guard_order(self) -> None:
for path, handler_name in AUTHENTICATED_POST_HANDLERS.items():
with self.subTest(path=path):
calls: list[str] = []
handler = self.handler(path, calls)
setattr(handler, handler_name, lambda: calls.append(handler_name))
RequestHandler.do_POST(handler)
self.assertEqual(
calls,
["auth", "csrf", f"access:POST:{path}", handler_name],
)
def test_failed_access_never_dispatches_protected_handler(self) -> None:
path, handler_name = next(iter(AUTHENTICATED_POST_HANDLERS.items()))
calls: list[str] = []
handler = self.handler(path, calls)
handler.require_access = lambda method, route: (
calls.append(f"access:{method}:{route}") or False
)
setattr(handler, handler_name, lambda: calls.append(handler_name))
RequestHandler.do_POST(handler)
self.assertEqual(calls, ["auth", "csrf", f"access:POST:{path}"])
if __name__ == "__main__":
unittest.main()
+187
View File
@@ -0,0 +1,187 @@
from __future__ import annotations
import io
import json
import unittest
import urllib.error
from pathlib import Path
from unittest.mock import patch
from assistant_agent import ReviewAssistantError, stream_review_assistant
from backend.llm import transport
from heaven_agent import HeavenAgentError, interpret_heaven
from llm_strategy import LLMCompilerError, test_llm_connection
from mentor_agent import MentorAgentError, MentorSkill, stream_with_mentor
ROOT = Path(__file__).resolve().parents[1]
class FakeResponse:
def __init__(self, *, payload: bytes = b"", lines: list[bytes] | None = None) -> None:
self.payload = payload
self.lines = lines or []
def __enter__(self):
return self
def __exit__(self, exc_type, exc_value, traceback):
return False
def read(self) -> bytes:
return self.payload
def __iter__(self):
return iter(self.lines)
class OpenAITransportTests(unittest.TestCase):
def test_chat_completion_builds_one_openai_compatible_request(self) -> None:
captured = {}
response = FakeResponse(
payload=json.dumps(
{"choices": [{"message": {"content": "OK"}}]}
).encode("utf-8")
)
def open_request(request, timeout):
captured["url"] = request.full_url
captured["headers"] = request.headers
captured["payload"] = json.loads(request.data.decode("utf-8"))
captured["timeout"] = timeout
return response
with patch("backend.llm.transport.urllib.request.urlopen", side_effect=open_request):
result = transport.chat_completion(
api_key="secret",
base_url="https://example.test/v1/",
model="model",
messages=[{"role": "user", "content": "ping"}],
timeout=17,
user_agent="XiaobaiReviewWeb/test",
)
self.assertEqual(result.content, "OK")
self.assertGreaterEqual(result.latency_ms, 0)
self.assertEqual(captured["url"], "https://example.test/v1/chat/completions")
self.assertEqual(captured["payload"]["stream"], False)
self.assertEqual(captured["headers"]["Authorization"], "Bearer secret")
self.assertEqual(captured["timeout"], 17)
def test_stream_completion_parses_deltas_and_ignores_final_snapshot(self) -> None:
response = FakeResponse(
lines=[
b'data: {"choices":[{"delta":{"content":"first"}}]}\n',
b'data: {"choices":[{"delta":{"content":" second"}}]}\n',
b'data: {"choices":[{"message":{"content":"first second"}}]}\n',
b"data: [DONE]\n",
]
)
with patch("backend.llm.transport.urllib.request.urlopen", return_value=response):
chunks = list(
transport.stream_chat_completion(
api_key="secret",
base_url="https://example.test/v1",
model="model",
messages=[],
timeout=17,
user_agent="XiaobaiReviewWeb/test",
)
)
self.assertEqual(chunks, ["first", " second"])
def test_empty_stream_has_a_stable_transport_error(self) -> None:
response = FakeResponse(lines=[b"data: [DONE]\n"])
with patch("backend.llm.transport.urllib.request.urlopen", return_value=response):
with self.assertRaises(transport.OpenAIEmptyResponseError):
list(
transport.stream_chat_completion(
api_key="secret",
base_url="https://example.test/v1",
model="model",
messages=[],
timeout=17,
user_agent="XiaobaiReviewWeb/test",
)
)
def test_http_error_keeps_code_and_sanitized_provider_detail(self) -> None:
error = urllib.error.HTTPError(
"https://example.test/v1/chat/completions",
429,
"rate limited",
{},
io.BytesIO(b'{"error":{"message":"capacity"}}'),
)
self.addCleanup(error.close)
with patch("backend.llm.transport.urllib.request.urlopen", side_effect=error):
with self.assertRaises(transport.OpenAIHTTPError) as caught:
transport.chat_completion(
api_key="secret",
base_url="https://example.test/v1",
model="model",
messages=[],
timeout=17,
user_agent="XiaobaiReviewWeb/test",
)
self.assertEqual(caught.exception.code, 429)
self.assertEqual(
caught.exception.describe("模型调用失败"),
"模型调用失败(HTTP 429):capacity",
)
def test_feature_agents_have_no_direct_provider_transport(self) -> None:
paths = (
"backend/features/mentor/agent.py",
"backend/features/heaven/agent.py",
"backend/features/review/agent.py",
"backend/features/screener/compiler.py",
)
for relative in paths:
source = (ROOT / relative).read_text(encoding="utf-8")
with self.subTest(path=relative):
self.assertNotIn("urllib.request", source)
self.assertNotIn("/chat/completions", source)
self.assertIn("llm_transport.", source)
class FeatureErrorMappingTests(unittest.TestCase):
def test_feature_specific_http_messages_are_preserved(self) -> None:
error = transport.OpenAIHTTPError(429, "capacity")
skill = MentorSkill(
skill_id="test",
name="测试老师",
description="",
tagline="",
focus=(),
content="",
path=Path("SKILL.md"),
)
with patch(
"mentor_agent.llm_transport.stream_chat_completion", side_effect=error
):
with self.assertRaisesRegex(
MentorAgentError, "问师模型调用失败(HTTP 429):capacity"
):
list(stream_with_mentor(skill, {}, "问题", [], "key", "https://x", "m"))
with patch("heaven_agent.llm_transport.chat_completion", side_effect=error):
with self.assertRaisesRegex(
HeavenAgentError, "问天模型调用失败(HTTP 429):capacity"
):
interpret_heaven("heart", {}, "key", "https://x", "m")
with patch(
"assistant_agent.llm_transport.stream_chat_completion", side_effect=error
):
with self.assertRaisesRegex(
ReviewAssistantError, "智能解读服务暂不可用(429"
):
list(stream_review_assistant({}, "问题", [], "key", "https://x", "m"))
with patch("llm_strategy.llm_transport.chat_completion", side_effect=error):
with self.assertRaisesRegex(
LLMCompilerError, "模型连接测试失败(HTTP 429):capacity"
):
test_llm_connection("key", "https://x", "m")
if __name__ == "__main__":
unittest.main()
+89
View File
@@ -0,0 +1,89 @@
from __future__ import annotations
import json
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path
from tools.build_api_registry import build as build_api_registry
from tools.build_architecture_inventory import (
build as build_architecture_inventory,
source_metrics,
)
from tools.verify_baseline import python_test_command
ROOT = Path(__file__).resolve().parents[1]
class MaintenanceToolTests(unittest.TestCase):
def test_generated_candidate_registries_are_current(self) -> None:
expected_api = json.loads(
(ROOT / "config" / "api.config.json").read_text(encoding="utf-8")
)
expected_architecture = json.loads(
(ROOT / "config" / "architecture-inventory.json").read_text(
encoding="utf-8"
)
)
self.assertEqual(expected_api, build_api_registry())
self.assertEqual(expected_architecture, build_architecture_inventory())
def test_baseline_verifier_uses_the_candidate_frontend(self) -> None:
source = (ROOT / "tools" / "verify_baseline.py").read_text(encoding="utf-8")
self.assertIn('FRONTEND_ROOT = ROOT / "frontend"', source)
self.assertIn('FRONTEND_ROOT.rglob("*.js")', source)
self.assertIn("start_e2e_server()", source)
self.assertIn("stop_e2e_server(server)", source)
self.assertNotIn('"static/app.js"', source)
def test_standalone_verifier_excludes_only_migration_comparison_modules(self) -> None:
repository_command = python_test_command(preservation_baseline=True)
standalone_command = python_test_command(preservation_baseline=False)
self.assertIn("discover", repository_command)
self.assertTrue(any(item == "tests.test_frontend_contract" for item in standalone_command))
self.assertFalse(any("test_preservation_" in item for item in standalone_command))
def test_architecture_metrics_are_independent_of_checkout_line_endings(self) -> None:
with tempfile.TemporaryDirectory() as directory:
root = Path(directory)
lf = root / "lf.js"
crlf = root / "crlf.js"
lf.write_bytes(b"const a = 1;\nconst b = 2;\n")
crlf.write_bytes(b"const a = 1;\r\nconst b = 2;\r\n")
self.assertEqual(source_metrics(lf), source_metrics(crlf))
def test_every_tool_has_a_non_mutating_help_path(self) -> None:
for path in sorted((ROOT / "tools").glob("*.py")):
if path.name.startswith("_"):
continue
with self.subTest(tool=path.name):
result = subprocess.run(
[sys.executable, str(path), "--help"],
cwd=ROOT,
capture_output=True,
text=True,
timeout=15,
check=False,
)
self.assertEqual(result.returncode, 0, result.stderr)
def test_migration_only_tools_are_explicitly_classified(self) -> None:
readme = (ROOT / "tools" / "README.md").read_text(encoding="utf-8")
for name in (
"build_preservation_manifest.py",
"move_class_methods.py",
"split_frontend_runtime.py",
):
self.assertIn(name, readme)
manifest = (ROOT / "tools" / "build_preservation_manifest.py").read_text(
encoding="utf-8"
)
self.assertNotIn('TARGET = ROOT / "app"', manifest)
self.assertIn('required=True', manifest)
if __name__ == "__main__":
unittest.main()
@@ -0,0 +1,26 @@
from __future__ import annotations
import unittest
from backend.bootstrap.config import tushare_code
from backend.features.market import charts
class MarketSymbolNormalizationTests(unittest.TestCase):
def test_chart_and_market_services_share_one_suffix_converter(self) -> None:
self.assertIs(charts._stock_market_code, tushare_code)
def test_existing_exchange_mapping_is_preserved(self) -> None:
cases = {
"000001": "000001.SZ",
"600000": "600000.SH",
"430047": "430047.BJ",
"830799": "830799.BJ",
}
for code, expected in cases.items():
with self.subTest(code=code):
self.assertEqual(charts._stock_market_code(code), expected)
if __name__ == "__main__":
unittest.main()
+4 -4
View File
@@ -44,7 +44,7 @@ class MentorStreamTests(unittest.TestCase):
captured["accept"] = request.headers.get("Accept")
return FakeStreamResponse(self.lines)
with patch("mentor_agent.urllib.request.urlopen", side_effect=open_request):
with patch("backend.llm.transport.urllib.request.urlopen", side_effect=open_request):
chunks = list(
stream_with_mentor(
self.skill, {"data_trade_date": "20260723"}, "怎么看?", [],
@@ -58,7 +58,7 @@ class MentorStreamTests(unittest.TestCase):
def test_non_streaming_compatibility_wrapper_collects_chunks(self):
with patch(
"mentor_agent.urllib.request.urlopen",
"backend.llm.transport.urllib.request.urlopen",
return_value=FakeStreamResponse(self.lines),
):
result = chat_with_mentor(
@@ -74,7 +74,7 @@ class MentorStreamTests(unittest.TestCase):
b"data: [DONE]\n",
]
with patch(
"mentor_agent.urllib.request.urlopen",
"backend.llm.transport.urllib.request.urlopen",
return_value=FakeStreamResponse(lines),
):
chunks = list(
@@ -92,7 +92,7 @@ class MentorStreamTests(unittest.TestCase):
b"data: [DONE]\n",
]
with patch(
"mentor_agent.urllib.request.urlopen",
"backend.llm.transport.urllib.request.urlopen",
return_value=FakeStreamResponse(lines),
):
chunks = list(
+6 -5
View File
@@ -91,11 +91,12 @@ class HeavenSliceSourceEquivalenceTests(unittest.TestCase):
for name in sorted(expected - (adapted or set())):
self.assertEqual(migrated[name], original[name], name)
def test_heaven_agent_is_an_exact_file(self) -> None:
self.assertEqual(
sha256(ORIGINAL_ROOT / "heaven_agent.py"),
sha256(APP_ROOT / "backend" / "features" / "heaven" / "agent.py"),
)
def test_heaven_agent_uses_shared_transport(self) -> None:
source = (
APP_ROOT / "backend" / "features" / "heaven" / "agent.py"
).read_text(encoding="utf-8")
self.assertIn("llm_transport.chat_completion", source)
self.assertNotIn("urllib.request", source)
def test_heaven_engine_definitions_are_exact_original_ast(self) -> None:
self.assertEqual(
+25 -1
View File
@@ -9,6 +9,7 @@ import chart_data_provider
import ifind_client
import realtime_aggregator
import tushare_client
from backend.bootstrap import config as bootstrap_config
from backend.data import realtime
from backend.data.providers import ifind_client as canonical_ifind
from backend.data.providers import tushare_client as canonical_tushare
@@ -101,6 +102,18 @@ def top_level_definitions(path: Path) -> dict[str, str]:
}
def function_contract(path: Path, name: str) -> tuple[str, str]:
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
function = next(
node for node in tree.body if isinstance(node, ast.FunctionDef) and node.name == name
)
body = ast.Module(body=function.body, type_ignores=[])
return (
ast.dump(function.args, include_attributes=False),
ast.dump(body, include_attributes=False),
)
class MarketSliceSourceEquivalenceTests(unittest.TestCase):
def test_market_service_methods_are_exact_original_ast(self) -> None:
original = class_methods(ORIGINAL_ROOT / "server.py", "DashboardService")
@@ -145,10 +158,21 @@ class MarketSliceSourceEquivalenceTests(unittest.TestCase):
top_level_definitions(ORIGINAL_ROOT / "tushare_client.py"),
top_level_definitions(APP_ROOT / "backend/data/providers/tushare_client.py"),
)
original_charts = top_level_definitions(ORIGINAL_ROOT / "chart_data_provider.py")
original_charts.pop("_stock_market_code")
self.assertEqual(
top_level_definitions(ORIGINAL_ROOT / "chart_data_provider.py"),
original_charts,
top_level_definitions(APP_ROOT / "backend/features/market/charts.py"),
)
self.assertEqual(
function_contract(
ORIGINAL_ROOT / "chart_data_provider.py", "_stock_market_code"
),
function_contract(
APP_ROOT / "backend/bootstrap/config.py", "tushare_code"
),
)
self.assertIs(charts._stock_market_code, bootstrap_config.tushare_code)
def test_relocated_frontend_preserves_original_market_runtime_and_styles(self) -> None:
assert_frontend_runtime_matches_audited_baseline(self)
@@ -105,11 +105,12 @@ class MentorLLMSliceSourceEquivalenceTests(unittest.TestCase):
for name in sorted(expected):
self.assertEqual(migrated[name], original[name], name)
def test_mentor_agent_and_stream_accumulator_are_exact_files(self) -> None:
self.assertEqual(
sha256(ORIGINAL_ROOT / "mentor_agent.py"),
sha256(APP_ROOT / "backend" / "features" / "mentor" / "agent.py"),
def test_mentor_agent_uses_shared_transport_and_stream_accumulator_is_exact(self) -> None:
mentor_source = (APP_ROOT / "backend" / "features" / "mentor" / "agent.py").read_text(
encoding="utf-8"
)
self.assertIn("llm_transport.stream_chat_completion", mentor_source)
self.assertNotIn("urllib.request", mentor_source)
self.assertEqual(
sha256(ORIGINAL_ROOT / "llm_stream.py"),
sha256(APP_ROOT / "backend" / "llm" / "stream.py"),
@@ -109,11 +109,12 @@ class ReviewAlertsSliceSourceEquivalenceTests(unittest.TestCase):
for name in sorted(expected):
self.assertEqual(migrated[name], original[name], name)
def test_review_assistant_agent_is_an_exact_file(self) -> None:
self.assertEqual(
sha256(ORIGINAL_ROOT / "assistant_agent.py"),
sha256(APP_ROOT / "backend" / "features" / "review" / "agent.py"),
)
def test_review_assistant_agent_uses_shared_transport(self) -> None:
source = (
APP_ROOT / "backend" / "features" / "review" / "agent.py"
).read_text(encoding="utf-8")
self.assertIn("llm_transport.stream_chat_completion", source)
self.assertNotIn("urllib.request", source)
def test_review_assistant_compatibility_module_is_canonical(self) -> None:
self.assertIs(assistant_agent, canonical_agent)
+10 -6
View File
@@ -170,12 +170,16 @@ class ScreenerSliceSourceEquivalenceTests(unittest.TestCase):
),
)
def test_library_and_compiler_files_are_exact_copies(self) -> None:
for original, migrated in (
("advanced_strategies.py", "backend/features/screener/strategies.py"),
("llm_strategy.py", "backend/features/screener/compiler.py"),
):
self.assertEqual(sha256(ORIGINAL_ROOT / original), sha256(APP_ROOT / migrated))
def test_library_is_exact_and_compiler_uses_shared_transport(self) -> None:
self.assertEqual(
sha256(ORIGINAL_ROOT / "advanced_strategies.py"),
sha256(APP_ROOT / "backend/features/screener/strategies.py"),
)
compiler_source = (
APP_ROOT / "backend/features/screener/compiler.py"
).read_text(encoding="utf-8")
self.assertEqual(compiler_source.count("llm_transport.chat_completion"), 2)
self.assertNotIn("urllib.request", compiler_source)
def test_compatibility_modules_export_the_canonical_objects(self) -> None:
self.assertIs(screener, engine)
+3 -3
View File
@@ -31,7 +31,7 @@ class ReviewAssistantStreamingTests(unittest.TestCase):
b'data: [DONE]\n',
]
)
with patch("assistant_agent.urllib.request.urlopen", return_value=response):
with patch("backend.llm.transport.urllib.request.urlopen", return_value=response):
chunks = list(
stream_review_assistant({}, "question", [], "key", "https://example.test/v1", "model")
)
@@ -39,7 +39,7 @@ class ReviewAssistantStreamingTests(unittest.TestCase):
def test_empty_stream_is_rejected(self):
response = StreamingResponse([b"data: [DONE]\n"])
with patch("assistant_agent.urllib.request.urlopen", return_value=response):
with patch("backend.llm.transport.urllib.request.urlopen", return_value=response):
with self.assertRaises(ReviewAssistantError):
list(
stream_review_assistant({}, "question", [], "key", "https://example.test/v1", "model")
@@ -54,7 +54,7 @@ class ReviewAssistantStreamingTests(unittest.TestCase):
b"data: [DONE]\n",
]
)
with patch("assistant_agent.urllib.request.urlopen", return_value=response):
with patch("backend.llm.transport.urllib.request.urlopen", return_value=response):
chunks = list(
stream_review_assistant(
{}, "question", [], "key", "https://example.test/v1", "model"
+2 -2
View File
@@ -46,7 +46,7 @@ class RealtimeClientStub:
class FixedMarketDatetime(datetime):
fixed_now = datetime.now().astimezone().replace(hour=10, minute=30, second=0, microsecond=0)
fixed_now = datetime(2026, 7, 31, 10, 30).astimezone()
@classmethod
def now(cls, tz=None):
@@ -54,7 +54,7 @@ class FixedMarketDatetime(datetime):
class FixedPreopenDatetime(datetime):
fixed_now = datetime.now().astimezone().replace(hour=8, minute=45, second=0, microsecond=0)
fixed_now = datetime(2026, 7, 31, 8, 45).astimezone()
@classmethod
def now(cls, tz=None):
+43
View File
@@ -0,0 +1,43 @@
# Candidate tools
Run these commands from `webapp/app/`. The tools are separated by responsibility so a
maintenance command cannot be mistaken for a historical migration rewrite.
## Normal maintenance
- `python tools/verify_baseline.py`: candidate unit tests, registry checks, every frontend
JavaScript syntax check, Git whitespace check, and read-only SQLite integrity check.
- `python tools/verify_baseline.py --e2e`: the same checks plus Playwright. The verifier owns
the local static-server lifecycle so the command exits cleanly on Windows.
- `python tools/build_api_registry.py [--check]`: generate or verify
`config/api.config.json` from `backend/application.py`.
- `python tools/build_architecture_inventory.py [--check]`: generate or verify
`config/architecture-inventory.json` from the candidate source tree.
Inside the canonical `webapp/app/` checkout, `verify_baseline.py` runs the full preservation
suite against the retained original baseline and enforces `git diff --check`. In a standalone
`app/` export where that baseline and Git checkout do not exist, the same command runs all
candidate-owned tests, skips only `test_preservation_*` comparison modules, and reports the Git
check as skipped. Product, registry, JavaScript, database, and optional Playwright checks remain
active in both modes.
## Acceptance and differential checks
- `run_preservation_runtime.py`: start an isolated original or candidate runtime with an
explicitly selected data directory and port.
- `compare_preservation_apis.py`: compare authenticated responses from two isolated runtimes.
- `compare_preservation_databases.py`: compare schema and selected table contents from two
SQLite copies.
These tools require explicit paths and do not select the production database automatically.
## Migration-only tools
- `build_preservation_manifest.py`: builds an exact-copy manifest for a specified source and
target. `--output` is mandatory so committed historical evidence is not overwritten.
- `move_class_methods.py`: mechanically moves named class methods between explicit files.
- `split_frontend_runtime.py`: reproduces the one-time Slice 10 split. It refuses to write
unless `--apply` is supplied and is not a normal maintenance command.
The migration-only tools are retained for audit and reproducibility. They are not part of
application startup, normal testing, or future feature development.
+26
View File
@@ -1,6 +1,7 @@
from __future__ import annotations
import argparse
import ast
import json
import re
import sys
@@ -59,8 +60,32 @@ def _role(method: str, path: str) -> str:
return required_role(method, sample)
def _mapped_paths(text: str) -> dict[str, set[str]]:
paths = {method: set() for method in ("GET", "POST", "DELETE")}
tree = ast.parse(text)
for node in tree.body:
if not isinstance(node, ast.Assign) or len(node.targets) != 1:
continue
target = node.targets[0]
if not isinstance(target, ast.Name):
continue
match = re.fullmatch(
r"(?:PUBLIC_|AUTHENTICATED_)?(GET|POST|DELETE)_HANDLERS", target.id
)
if not match:
continue
mapping = ast.literal_eval(node.value)
if not isinstance(mapping, dict) or not all(
isinstance(path, str) and path.startswith("/api/") for path in mapping
):
raise ValueError(f"Invalid route handler map: {target.id}")
paths[match.group(1)].update(mapping)
return paths
def build() -> dict:
text = (ROOT / "backend" / "application.py").read_text(encoding="utf-8")
mapped_paths = _mapped_paths(text)
method_matches = list(re.finditer(r"^ def do_(GET|POST|DELETE)\(", text, re.MULTILINE))
routes = []
for index, match in enumerate(method_matches):
@@ -68,6 +93,7 @@ def build() -> dict:
end = method_matches[index + 1].start() if index + 1 < len(method_matches) else len(text)
block = text[match.start():end]
exact_paths = set(re.findall(r'parsed\.path\s*==\s*"(/api/[^"]+)"', block))
exact_paths.update(mapped_paths[method])
patterns = set(
re.findall(
r're\.(?:fullmatch|match)\(\s*r?["\']([^"\']*?/api/[^"\']+)["\']\s*,\s*parsed\.path',
+63 -36
View File
@@ -9,7 +9,7 @@ from typing import Any
ROOT = Path(__file__).resolve().parents[1]
OUTPUT = ROOT / "docs" / "governance" / "architecture-inventory.json"
OUTPUT = ROOT / "config" / "architecture-inventory.json"
def relative(path: Path) -> str:
@@ -37,22 +37,28 @@ def page_inventory(html: str) -> list[dict[str, str]]:
def api_inventory(server: str) -> dict[str, list[str]]:
exact = sorted(set(re.findall(r'parsed\.path\s*==\s*"(/api/[^"]+)"', server)))
routes = json.loads(source("config/api.config.json"))["routes"]
exact = sorted(
{item["path"] for item in routes if item["match"] == "exact"}
)
prefixes = sorted(
set(re.findall(r'parsed\.path\.startswith\(\s*"(/api/[^"]+)"', server))
)
patterns = sorted(
set(
item
for item in re.findall(r'r?["\']([^"\']*?/api/[^"\']+)["\']', server)
if "\\d" in item or ".+" in item or "(?P" in item
)
{item["path"] for item in routes if item["match"] == "regex"}
)
return {"exact": exact, "prefixes": prefixes, "patterns": patterns}
def database_inventory(database: str) -> list[str]:
return re.findall(r"CREATE TABLE IF NOT EXISTS\s+([a-zA-Z0-9_]+)", database)
def database_inventory(sources: list[str]) -> list[str]:
tables: list[str] = []
for database in sources:
tables.extend(
re.findall(
r"CREATE TABLE IF NOT EXISTS\s+([a-zA-Z0-9_]+)", database
)
)
return list(dict.fromkeys(tables))
def python_functions(path: str, prefixes: tuple[str, ...]) -> list[str]:
@@ -68,13 +74,23 @@ def css_layers(html: str) -> list[str]:
return re.findall(r'<link[^>]+rel="stylesheet"[^>]+href="([^"]+)"', html)
def source_metrics(path: Path) -> dict[str, int]:
text = path.read_text(encoding="utf-8")
return {
"bytes": len(text.encode("utf-8")),
"lines": len(text.splitlines()),
}
def code_hotspots() -> list[dict[str, Any]]:
candidates = [
"server.py",
"backend/application.py",
"database.py",
"screener.py",
"market_insights.py",
"tushare_client.py",
"backend/features/screener/engine.py",
"backend/features/market/insights.py",
"backend/data/providers/tushare_client.py",
"backend/features/heaven/service.py",
"backend/features/heaven/engine.py",
"frontend/index.html",
"frontend/app.js",
"frontend/styles/styles.css",
@@ -88,26 +104,29 @@ def code_hotspots() -> list[dict[str, Any]]:
rows = []
for name in candidates:
path = ROOT / name
rows.append(
{
"path": name,
"bytes": path.stat().st_size,
"lines": len(path.read_text(encoding="utf-8").splitlines()),
}
)
if not path.is_file():
continue
rows.append({"path": name, **source_metrics(path)})
return sorted(rows, key=lambda item: item["bytes"], reverse=True)
def build() -> dict[str, Any]:
html = source("frontend/index.html")
server = source("server.py")
database = source("database.py")
server = source("backend/application.py")
database_sources = [source("database.py")]
database_sources.extend(
path.read_text(encoding="utf-8")
for path in sorted((ROOT / "backend" / "database" / "migrations").glob("m*.py"))
)
database_sources.append(
source("backend/database/migrations/runner.py")
)
pages = page_inventory(html)
api = api_inventory(server)
tables = database_inventory(database)
tables = database_inventory(database_sources)
return {
"schema_version": 1,
"captured_from": "governed source tree",
"captured_from": "app accepted modular runtime",
"runtime": {
"http_server": "http.server.ThreadingHTTPServer",
"application_processes": 1,
@@ -126,21 +145,26 @@ def build() -> dict[str, Any]:
"api": api,
"database_tables": tables,
"background_job_methods": python_functions(
"server.py", ("_background", "_run_background", "_schedule_", "run_automatic")
"backend/application.py",
("_background", "_run_background", "_schedule_", "run_automatic"),
),
"external_data_adapters": [
{"provider": "tushare", "path": "tushare_client.py", "runtime_role": "primary deterministic market data"},
{"provider": "ifind", "path": "ifind_client.py", "runtime_role": "realtime, charts, snapshots, enrichment"},
{"provider": "eastmoney", "path": "chart_data_provider.py", "runtime_role": "display chart fallback"},
{"provider": "eastmoney", "path": "realtime_aggregator.py", "runtime_role": "isolated realtime observation"},
{"provider": "tencent", "path": "realtime_aggregator.py", "runtime_role": "index observation fallback"},
{"provider": "tushare", "path": "backend/data/providers/tushare_client.py", "runtime_role": "primary deterministic market data"},
{"provider": "ifind", "path": "backend/data/providers/ifind_client.py", "runtime_role": "realtime, charts, snapshots, enrichment"},
{"provider": "eastmoney", "path": "backend/features/market/charts.py", "runtime_role": "display chart fallback"},
{"provider": "eastmoney", "path": "backend/data/realtime.py", "runtime_role": "isolated realtime observation"},
{"provider": "tencent", "path": "backend/data/realtime.py", "runtime_role": "index observation fallback"},
],
"llm_entrypoints": [
{"function": "stream_with_mentor", "path": "mentor_agent.py"},
{"function": "interpret_heaven", "path": "heaven_agent.py"},
{"function": "stream_review_assistant", "path": "assistant_agent.py"},
{"function": "compile_strategy_with_llm", "path": "llm_strategy.py"},
{"function": "test_llm_connection", "path": "llm_strategy.py"},
{"function": "stream_with_mentor", "path": "backend/features/mentor/agent.py"},
{"function": "interpret_heaven", "path": "backend/features/heaven/agent.py"},
{"function": "stream_review_assistant", "path": "backend/features/review/agent.py"},
{"function": "compile_strategy_with_llm", "path": "backend/features/screener/compiler.py"},
{"function": "test_llm_connection", "path": "backend/features/screener/compiler.py"},
],
"llm_transport": [
{"function": "chat_completion", "path": "backend/llm/transport.py"},
{"function": "stream_chat_completion", "path": "backend/llm/transport.py"},
],
"css_layers": css_layers(html),
"code_hotspots": code_hotspots(),
@@ -154,7 +178,10 @@ def main() -> int:
rendered = json.dumps(build(), ensure_ascii=False, indent=2) + "\n"
if args.check:
if not OUTPUT.exists() or OUTPUT.read_text(encoding="utf-8") != rendered:
raise SystemExit("architecture inventory is stale; run tools/build_architecture_inventory.py")
raise SystemExit(
"architecture inventory is stale; run "
"tools/build_architecture_inventory.py"
)
print("Architecture inventory is current.")
return 0
OUTPUT.parent.mkdir(parents=True, exist_ok=True)
+64 -21
View File
@@ -1,14 +1,13 @@
from __future__ import annotations
import argparse
import hashlib
import json
from datetime import datetime
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
TARGET = ROOT / "app"
OUTPUT = ROOT / "docs" / "migration" / "原版资产清单.json"
REPOSITORY_ROOT = Path(__file__).resolve().parents[2]
DIRECTORIES = (
"backend",
@@ -69,33 +68,57 @@ def digest(path: Path) -> str:
return checksum.hexdigest()
def source_files() -> list[Path]:
files = [ROOT / name for name in ROOT_FILES]
def display_path(path: Path, base: Path) -> str:
try:
return path.relative_to(base).as_posix() or "."
except ValueError:
return path.as_posix()
def source_files(source_root: Path) -> list[Path]:
files = [source_root / name for name in ROOT_FILES]
for directory in DIRECTORIES:
root = source_root / directory
if not root.is_dir():
continue
files.extend(
path
for path in (ROOT / directory).rglob("*")
for path in root.rglob("*")
if path.is_file() and "__pycache__" not in path.parts
)
return sorted(set(files), key=lambda path: path.relative_to(ROOT).as_posix())
return sorted(
set(files), key=lambda path: path.relative_to(source_root).as_posix()
)
def build_manifest() -> dict[str, object]:
def build_manifest(
source_root: Path,
target_root: Path,
source_commit: str,
) -> dict[str, object]:
assets = []
mismatches = []
for source in source_files():
relative = source.relative_to(ROOT)
target = TARGET / relative
source_hash = digest(source)
for source in source_files(source_root):
relative = source.relative_to(source_root)
target = target_root / relative
source_exists = source.is_file()
source_hash = digest(source) if source_exists else ""
target_hash = digest(target) if target.is_file() else ""
status = "identical" if source_hash == target_hash else "mismatch"
if not source_exists:
status = "missing_source"
elif not target.is_file():
status = "missing_target"
elif source_hash == target_hash:
status = "identical"
else:
status = "mismatch"
if status != "identical":
mismatches.append(relative.as_posix())
assets.append(
{
"source": relative.as_posix(),
"target": f"app/{relative.as_posix()}",
"bytes": source.stat().st_size,
"target": display_path(target, REPOSITORY_ROOT),
"bytes": source.stat().st_size if source_exists else 0,
"sha256": source_hash,
"disposition": "original_copy_pending_move",
"status": status,
@@ -104,9 +127,9 @@ def build_manifest() -> dict[str, object]:
return {
"schema_version": 1,
"generated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
"source_commit": "41329943c4878fc09ed82ec376eb93ab151e4092",
"source_root": ".",
"target_root": "app",
"source_commit": source_commit,
"source_root": display_path(source_root, REPOSITORY_ROOT),
"target_root": display_path(target_root, REPOSITORY_ROOT),
"excluded": [
"next/",
"data/review.db and SQLite sidecars",
@@ -123,14 +146,34 @@ def build_manifest() -> dict[str, object]:
def main() -> int:
manifest = build_manifest()
OUTPUT.write_text(
parser = argparse.ArgumentParser(
description=(
"Build an exact-copy manifest for a preservation migration stage. "
"This is a migration-only tool; it does not describe the final moved layout."
)
)
parser.add_argument("--source-root", type=Path, default=REPOSITORY_ROOT)
parser.add_argument("--target-root", type=Path, default=REPOSITORY_ROOT / "app")
parser.add_argument("--source-commit", default="")
parser.add_argument(
"--output",
type=Path,
required=True,
help="write to a new audit path; do not overwrite committed slice evidence",
)
args = parser.parse_args()
source_root = args.source_root.resolve()
target_root = args.target_root.resolve()
output = args.output.resolve()
manifest = build_manifest(source_root, target_root, args.source_commit)
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text(
json.dumps(manifest, ensure_ascii=False, indent=2) + "\n",
encoding="utf-8",
)
print(
f"assets={manifest['asset_count']} mismatches={manifest['mismatch_count']} "
f"output={OUTPUT}"
f"output={output}"
)
return 1 if manifest["mismatch_count"] else 0
+15
View File
@@ -1,5 +1,6 @@
from __future__ import annotations
import argparse
import hashlib
import json
import re
@@ -91,6 +92,20 @@ def extract_written_chunks(paths: list[Path]) -> dict[tuple[int, int], str]:
def main() -> None:
parser = argparse.ArgumentParser(
description="Reproduce the one-time Slice 10 frontend source split"
)
parser.add_argument(
"--apply",
action="store_true",
help="perform the historical split; this rewrites frontend runtime files",
)
args = parser.parse_args()
if not args.apply:
parser.error(
"this migration-only tool rewrites files; pass --apply only when "
"reproducing Slice 10 from its documented pre-split checkpoint"
)
source_path = ORIGINAL_STATIC / "app.js"
target_path = FRONTEND_ROOT / "app.js"
source_bytes = source_path.read_bytes()
+120 -9
View File
@@ -5,10 +5,15 @@ import shutil
import sqlite3
import subprocess
import sys
import time
import urllib.error
import urllib.request
from pathlib import Path
ROOT = Path(__file__).resolve().parents[1]
FRONTEND_ROOT = ROOT / "frontend"
E2E_URL = "http://127.0.0.1:8876/index.html"
def run(label: str, command: list[str]) -> None:
@@ -16,6 +21,39 @@ def run(label: str, command: list[str]) -> None:
subprocess.run(command, cwd=ROOT, check=True)
def python_test_command(preservation_baseline: bool | None = None) -> list[str]:
if preservation_baseline is None:
preservation_baseline = (ROOT.parent / "static" / "app.js").is_file()
if preservation_baseline:
return [sys.executable, "-m", "unittest", "discover", "-s", "tests"]
modules = [
f"tests.{path.stem}"
for path in sorted((ROOT / "tests").glob("test_*.py"))
if not path.stem.startswith("test_preservation_")
]
if not modules:
raise RuntimeError("no standalone candidate tests found")
return [sys.executable, "-m", "unittest", *modules]
def verify_git_diff() -> None:
result = subprocess.run(
["git", "rev-parse", "--show-toplevel"],
cwd=ROOT,
capture_output=True,
text=True,
check=False,
)
if result.returncode != 0:
print("\n[patch] skipped: standalone export is not a Git checkout")
return
repository_root = Path(result.stdout.strip()).resolve()
if ROOT != repository_root / "app":
print("\n[patch] skipped: standalone export is outside the canonical app path")
return
run("patch", ["git", "diff", "--check"])
def verify_database() -> None:
database = ROOT / "data" / "review.db"
if not database.exists():
@@ -28,8 +66,63 @@ def verify_database() -> None:
print(f"\n[database] integrity_check=ok size={database.stat().st_size}")
def url_reachable(url: str) -> bool:
try:
with urllib.request.urlopen(url, timeout=1) as response:
return response.status == 200
except (OSError, urllib.error.URLError):
return False
def start_e2e_server() -> subprocess.Popen[bytes] | None:
if url_reachable(E2E_URL):
print(f"\n[e2e-server] reusing {E2E_URL}")
return None
creation_flags = subprocess.CREATE_NO_WINDOW if sys.platform == "win32" else 0
process = subprocess.Popen(
[
sys.executable,
"-m",
"http.server",
"8876",
"--bind",
"127.0.0.1",
"--directory",
"frontend",
],
cwd=ROOT,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
creationflags=creation_flags,
)
deadline = time.monotonic() + 15
while time.monotonic() < deadline:
if process.poll() is not None:
raise RuntimeError("Playwright static server exited before becoming ready")
if url_reachable(E2E_URL):
print(f"\n[e2e-server] started {E2E_URL}")
return process
time.sleep(0.1)
stop_e2e_server(process)
raise RuntimeError("Playwright static server did not become ready within 15 seconds")
def stop_e2e_server(process: subprocess.Popen[bytes] | None) -> None:
if process is None:
return
process.terminate()
try:
process.wait(timeout=5)
except subprocess.TimeoutExpired:
process.kill()
process.wait(timeout=5)
print("\n[e2e-server] stopped")
def main() -> int:
parser = argparse.ArgumentParser(description="Verify the governance regression baseline")
parser = argparse.ArgumentParser(
description="Verify the modular preservation candidate"
)
parser.add_argument(
"--e2e",
action="store_true",
@@ -37,20 +130,38 @@ def main() -> int:
)
args = parser.parse_args()
run("python", [sys.executable, "-m", "unittest", "discover", "-s", "tests"])
run("python", python_test_command())
run(
"api-registry",
[sys.executable, "tools/build_api_registry.py", "--check"],
)
run(
"architecture-inventory",
[sys.executable, "tools/build_architecture_inventory.py", "--check"],
)
node = shutil.which("node")
if not node:
raise RuntimeError("node is required for JavaScript syntax checks")
for script in ("static/app.js", "static/heaven-loading-v2.js"):
run("javascript", [node, "--check", script])
run("patch", ["git", "diff", "--check"])
scripts = sorted(FRONTEND_ROOT.rglob("*.js"))
if not scripts:
raise RuntimeError(f"no JavaScript files found under {FRONTEND_ROOT}")
for script in scripts:
run(
"javascript",
[node, "--check", script.relative_to(ROOT).as_posix()],
)
verify_git_diff()
verify_database()
if args.e2e:
npm = shutil.which("npm.cmd" if sys.platform == "win32" else "npm")
if not npm:
raise RuntimeError("npm is required for the Playwright suite")
run("playwright", [npm, "run", "test:e2e"])
npx = shutil.which("npx.cmd" if sys.platform == "win32" else "npx")
if not npx:
raise RuntimeError("npx is required for the Playwright suite")
server = start_e2e_server()
try:
run("playwright", [npx, "playwright", "test", "--reporter=dot"])
finally:
stop_e2e_server(server)
print("\nBaseline verification passed.")
return 0
+91
View File
@@ -0,0 +1,91 @@
# `app/`代码减法账本
> 基线:`xiaobai-preservation-complete-20260801`
> 工作目录:只允许修改`webapp/app/`;原版根目录和冻结的`next/`只读
> 目标:删除重复实现和历史补丁,不改变功能、视觉、交互、动画、计算、权限、API或数据行为
## 固定规则
1. 每批只处理一个明确边界,先证明重复或无消费者,再修改。
2. 新共享实现必须在同一提交删除全部被替代实现;禁止只加一层包装。
3. 运行代码总量原则上不得增加;测试和证据代码单独统计。
4. 迁移期源码相等测试不得简单删除。发生已批准的结构重构时,必须替换成行为、错误语义和
唯一所有权契约。
5. 每批通过领域测试、全量候选测试、独立导出测试和受影响的浏览器流程后才建立Git检查点。
6. CSS最后处理;没有逐页日间、夜间和多视口截图证据,不删除视觉规则。
## 批次记录
| 批次 | 边界 | 基线问题 | 目标 | 状态 |
|---|---|---|---|---|
| CR-01 | LLM供应商传输 | 问师、问天、复盘助手和策略编译各自构造HTTP请求、解析响应和读取错误 | 只保留`backend/llm/transport.py`一个网络出口 | 已完成 |
| CR-02 | HTTP精确POST委托 | 27个端点重复使用“比较路径、调用无参数处理器、返回”三行分支 | 用公开/受保护两张显式映射统一委托,同时保留复杂路由的原控制流 | 已完成 |
| CR-03 | 股票市场后缀转换 | Tushare业务与iFinD图表各保留一份完全相同的沪深京代码转换函数 | 图表复用`bootstrap/config.py::tushare_code`,只保留一份函数体 | 已完成 |
## CR-01验收口径
- 四个功能模块不得包含`urllib.request``/chat/completions`
- 非流式与流式请求的URL、鉴权、User-Agent、SSE累积和空响应行为保持不变。
- 各功能原有错误类型和用户可见错误文案保持不变。
- `LLMGateway`的会员、额度、主辅回退、首字后不中途换模型和审计规则保持不变。
- 架构清单必须登记唯一传输入口;完整测试和真实主模型最小调用通过。
## CR-01结果
- 四个功能模块的运行代码由572行降至422行;新增唯一传输实现129行,生产代码净减少21行、
约1.4 KB。行数不是主要收益,关键是5处`/chat/completions`请求只剩1处。
- 三套重复HTTP错误正文解析合并为一套;各功能原有错误类型和用户可见文案由专项测试固定。
- 迁移期4个“Agent文件逐字相等”断言没有直接删除,而是替换为共享传输唯一所有权、提示词模块
归属、SSE行为和兼容模块对象契约。
- 候选311项、纯`app/`导出248项、45项Playwright通过;24个JavaScript文件、架构/API注册表和
SQLite完整性检查通过。
- 使用候选数据库中加密保存的主模型完成真实非流式与流式最小调用,分别成功返回完整响应和
4个流式分片。调用未输出密钥或模型正文。
- 本批不修改页面、CSS、提示词、业务计算、会员额度、模型回退、数据库或部署。
回档基线为`xiaobai-preservation-complete-20260801`;本批检查点为
`xiaobai-reduction-01-llm-transport-20260801`
## CR-02验收口径
- 仅纳入没有路径参数、请求体解析或专属异常分支的精确POST端点;其余路由保持原样。
- 公开注册/登录端点继续在鉴权前分发;受保护端点继续严格执行登录、CSRF、注册表权限、处理器。
- 27个映射路径必须都由权威API注册表解析,处理方法必须真实存在,公开与受保护集合不得重叠。
- API路径、功能归属、访问角色、状态码、错误正文和静态页面绕过鉴权行为保持不变。
- API清单生成器必须结构化读取显式映射;架构清查复用API清单,不再维护第二套路由发现规则。
## CR-02结果
- 2个公开端点和25个受保护端点改为显式委托映射,原来的79行重复分支被43行映射、分发与调用替代;
`backend/application.py`净减少36行,规范化源码约减少1.0 KB。
- 带正则路径参数、请求体读取、查询参数转换或特殊异常语义的GET、POST、DELETE端点未改动。
- 架构清查删除了自行扫描精确/正则API路径的第二套规则,改为消费权威`api.config.json`API注册表的
53个精确路径、11个正则路径、功能归属和权限均未变化。
- 原版231项、候选315项、纯`app/`导出252项、45项Playwright通过;24个JavaScript文件、
API/架构注册表、Git空白检查和SQLite完整性检查通过。
- 本批不修改前端、CSS、业务计算、数据源、数据库结构、LLM、会员规则或部署。
本批基线为`xiaobai-reduction-01-llm-transport-20260801`;检查点为
`xiaobai-reduction-02-http-dispatch-20260801`
## CR-03验收口径
- 图表模块不再定义第二份市场后缀转换函数,仍保留原局部名称和两个调用点。
- 深市、沪市、北交所的既有映射结果保持不变;不借本批修正或扩展代码规则。
- 原图表文件除该函数外的所有顶层定义继续与原版AST逐项相等。
- 原版`_stock_market_code`函数的参数和函数体必须与唯一共享实现AST相等,运行时别名必须指向
同一个函数对象。
## CR-03结果
- 删除`backend/features/market/charts.py`中第二份10行定义,以1行导入别名复用共享实现,生产代码
净减少9行;全仓后端只剩一份沪深京后缀转换函数体。
- 未合并实时聚合、东方财富图表、iFinD和Tushare的HTTP传输;它们的缓存、错误、重试和降级语义
不同,仅有外形相似,证据不足以安全抽象。
- 原有迁移期整文件相等断言被等价范围断言、共享函数AST断言和唯一对象断言替代,没有降低门禁。
- 原版231项、候选317项、纯`app/`导出254项、45项Playwright通过;API/架构注册表、
24个JavaScript文件、Git空白检查和SQLite完整性检查通过。
- 本批不修改图表请求、数据来源、缓存、时间范围、行情计算、前端、CSS、数据库或部署。
本批基线为`xiaobai-reduction-02-http-dispatch-20260801`;检查点为
`xiaobai-reduction-03-market-symbol-20260801`
+5 -1
View File
@@ -39,7 +39,7 @@
## 4. 真实浏览器差分
- 1920×1080日间模式检查情绪周期、集合竞价、智能选股、观势、观气、观心介绍及观心呼吸。
- 1920×1080日间模式检查情绪周期、智能选股、观势、观气、观心介绍及观心呼吸。
- 1920×1080夜间模式检查情绪周期的背景、字体、表格、几何和横向溢出。
- 390×844检查情绪周期与观势:移动Shell、底部五入口、页面纵向滚动、无横向溢出及问天特效均一致。
- 观势星空节点90个、观气100个、观心110个;三页八卦节点均为2个。
@@ -48,6 +48,10 @@
- 原版和迁移版检查流程均未产生新增控制台error或warn。
- 动画帧、焦点框和动态状态文字属于采样瞬时状态,因此截图文件哈希不要求相同;可见布局几何、计算样式、节点、动画名称和交互结果必须一致,本次均通过。
- 两个服务使用相同主机名、不同端口时会共享并覆盖登录Cookie;曾导致迁移版切页被误判为失效。逐服务重新登录后行为一致,该问题属于并行验收环境限制,不是产品回归。
- 2026-08-01像素复核发现`frontend-migrated-auction-light-1920x1080`实际截取了登录状态失效页,
不能证明集合竞价视觉等价,文件已重命名为`INVALID-login-session`。集合竞价仍有切片05真实页面、
本切片源码/样式保真、API及Playwright证据,但最终视觉明确留待人工验收,不以替代证据冒充
本切片截图差分。
- 机器可读记录见`browser-acceptance.json`,截图均保存在本目录。
## 5. 自动验证与保留边界
@@ -1,7 +1,7 @@
{
"schema_version": 1,
"tested_at": "2026-07-31T12:00:00+08:00",
"result": "passed",
"result": "passed_with_documented_evidence_gap",
"environments": {
"original": "temporary_original_runtime",
"migrated": "temporary_app_runtime",
@@ -15,7 +15,6 @@
"theme": "light",
"views": [
"sentiment",
"auction",
"screener",
"heaven_trend",
"heaven_fortune",
@@ -44,6 +43,14 @@
"equal": true
}
],
"invalid_captures": [
{
"view": "auction",
"file": "frontend-migrated-auction-light-1920x1080.INVALID-login-session.png",
"reason": "The migrated capture shows an expired login session and is not visual-equivalence evidence.",
"replacement_claim": "No replacement screenshot claim; final visual acceptance remains manual."
}
],
"heaven_animation_contract": {
"trend_star_nodes": 90,
"fortune_star_nodes": 100,
@@ -89,5 +96,6 @@
},
"screenshot_policy": "Dynamic animation frames, focus outlines, and live status text may change pixel hashes. Acceptance compares visible geometry, computed styles, DOM state, animation names, interaction results, and overflow behavior.",
"known_test_environment_constraint": "Original and migrated services on the same hostname share cookies across ports. Each service must be logged in separately immediately before comparison.",
"all_equal": true
"all_equal": true,
"manual_acceptance_required": true
}
+39 -11
View File
@@ -2,14 +2,18 @@
> 基线:`dec3cd12365df5e7d4c391369f14c457cf56d7d9`
> 候选回档标签:`xiaobai-preservation-slice-11-candidate-20260731`
> 当前结论:自动验收完成,等待用户人工验收;尚未执行正式切换、Docker或NAS部署
> 严格审计候选标签:`xiaobai-preservation-slice-11-audit-candidate-20260731`
> 跨日复验候选标签:`xiaobai-preservation-slice-11-audit-candidate-20260801`
> 独立维护候选标签:`xiaobai-preservation-slice-11-audit-candidate-standalone-20260801`
> 最终完成标签:`xiaobai-preservation-complete-20260801`
> 当前结论:自动与用户人工验收均已完成;本地迁移交付完成,尚未执行正式切换、Docker或NAS部署
## 1. 本切片做了什么
本切片不增加功能、不调整视觉,也不继续拆分业务代码。它逐项审计切片10保留的待定资产,
把有完整证据的无效实现放入可单独回档的试删候选,并为人工接管和后续切换准备文档。
试删候选
试删和人工验收后确认废弃
- `app/demo_data.py`:没有运行导入、动态注册、数据库或配置责任的旧演示数据构造器。
- `app/frontend/heaven-loading.js`:页面未加载的旧问天动画;正式入口继续使用
@@ -30,7 +34,7 @@
| 验证 | 结果 |
|---|---:|
| 原版`python -m unittest discover -s tests -q` | 231项通过 |
| 迁移版`python -m unittest discover -s tests -q` | 298项通过 |
| 迁移版`python -m unittest discover -s tests -q` | 305项通过 |
| 历史切片与前端/清理专项复核 | 通过 |
| 迁移版JavaScript语法检查 | 24个文件通过 |
| `npx.cmd playwright test --reporter=dot` | 45项通过(1.6分钟) |
@@ -38,12 +42,30 @@
| 数据库差分 | 62个schema对象、21张关键表全部一致 |
| `git diff --check` | 通过 |
2026-08-01 01:19Asia/Shanghai)跨日复验时,原版与迁移版共用的实时详情测试暴露出
测试夹具依赖运行日的问题:夹具在周六动态生成“今日”,而业务正确拒绝在非交易日合并实时
行情。两版测试夹具同步固定到明确的工作日盘中/盘前时点,产品代码与行情日期规则未改。
修复后统一验收命令再次通过302项测试、24个JavaScript文件、SQLite完整性和45项Playwright
2.0分钟)。
早期切片的五个前端源码测试原本要求与未试删的`static/app.js`逐字符一致。本切片把它们统一到
`preservation_helpers.assert_frontend_runtime_matches_audited_baseline`:只允许审计登记的五段原始
行号被删除,任何其他新增、删除、重排或内容变化仍会使全部历史测试失败。
完整API和数据库结果分别见`api-diff.json``database-diff.json`,二者`all_equal`均为`true`
严格完成审计另外验证了候选维护工具本身:API注册表和候选架构清单均可生成并执行
`--check`;24个前端脚本全部由统一命令发现;迁移期写入工具必须显式指定输出或`--apply`
Windows下Playwright托管Python静态服务器会在45项执行后不退出,统一验证器现改为显式启动、
探活和停止8876;最终同一命令正常退出并明确报告`45 passed`。四项维护工具契约测试使迁移版
总数从298增加到302。逐项目标矩阵见`completion-audit.md`
后续独立性审计把仅含`app/`的Git导出放到系统临时目录:运行模块全部解析到导出内部,统一
维护命令通过242项候选自有测试、两个注册表、24个JavaScript文件和SQLite完整性检查,并明确
跳过不存在的Git工作区。正式仓库继续执行全部保真差分,最终为305项Python测试和45项
Playwright通过。架构热点字节数已改为换行归一化后的UTF-8大小,Windows CRLF与Git/Linux LF
不会再制造假差异。
## 3. 真实浏览器验收
- 桌面端复核情绪周期、智能选股三个工作区、问天日间/夜间和加载资源;控制台无新增
@@ -56,25 +78,31 @@
- 同账号、同数据库快照、同主题和同视口下,原版与迁移版问天布局和计算样式一致。
机器可读记录见`browser-acceptance.json`。截图和更广的桌面/移动基线继续沿用切片10证据目录。
切片10截图的事后像素复核见`screenshot-pixel-audit.json`;九组有效截图平均色差低于0.3/255,
集合竞价候选图因登录状态失效被明确排除,未作为自动截图证据。2026-08-01用户随后在`8797`
真实登录状态下逐页检查全部页面并测试全部功能,确认视觉与功能迁移成功;所见问题几乎都
属于原版遗留问题,未发现阻止验收的迁移回归。完整人工结论见`manual-acceptance.md`
## 4. 数据与部署边界
- API/数据库比较使用两个隔离副本:`slice11-final-original``slice11-final-migrated`
- 没有写入根目录正式`data/review.db`,没有修改`.env`或私有Skill。
- 没有占用`8765`,没有执行Docker构建、NAS测试或服务器切换。
- 原版根目录仍是正式运行基线;`app/`只是等待人工验收的候选
- `app/`已经完成本地迁移验收;原版根目录仍是当前部署基线和切换前回档来源
- `next/`保持冻结,没有作为代码、样式、测试或文档来源。
## 5. 回档与最终确认
切片11提交和标签只代表“候选可验收”,不代表删除已被永久确认。人工验收前可按
`uncertain-code-audit.md`从切片10标签单独恢复任一候选;不需要回退其他已通过迁移切片。
2026-08-01用户完成全部页面和功能人工验收,确认迁移在视觉和功能上成功,并确认人工检查中
发现的问题几乎都属于原版遗留问题。切片11的2个文件和5个无消费者函数据此从“候选试删”改为
“确认废弃”;恢复标签继续保留作历史兼容应急,不代表删除结论仍待裁决。
用户人工确认后才允许
本地迁移收尾已执行
1. 试删候选状态改为“确认废弃”。
2. 建立最终完成标签。
3. 决定本地或Docker正式切换时间
4. 另行制定根目录兼容外壳和旧实现的清理计划。
1. 试删候选状态改为“确认废弃”。
2. 建立最终完成标签并推送Gitea
3. `app/`作为后续结构治理的唯一开发目录
正式数据库、原版部署、Docker和NAS仍未切换;切换时间与根目录清理必须由用户另行批准。
人工维护、验证、切换和回退步骤见`docs/migration/人工维护与本地切换指南.md`
@@ -18,8 +18,8 @@
"endpoint": "/api/account/status",
"original_status": 200,
"migrated_status": 200,
"original_sha256": "094530da01cbc633de4e2a09d1b3a58689ed9743d2ab2207c495e6c44781fc1e",
"migrated_sha256": "094530da01cbc633de4e2a09d1b3a58689ed9743d2ab2207c495e6c44781fc1e",
"original_sha256": "0ac1b88affa62671f14ba9ea0bd42155952fdc2e03dba3573c232bb452733bc6",
"migrated_sha256": "0ac1b88affa62671f14ba9ea0bd42155952fdc2e03dba3573c232bb452733bc6",
"equal": true,
"first_difference": null
},
@@ -62,8 +62,8 @@
"endpoint": "/api/auction?trade_date=2026-07-30",
"original_status": 200,
"migrated_status": 200,
"original_sha256": "5a6b643ab1d7e2285ae5667bca12e6ed887ba4aa7512ba989e0dcfb2032e755b",
"migrated_sha256": "5a6b643ab1d7e2285ae5667bca12e6ed887ba4aa7512ba989e0dcfb2032e755b",
"original_sha256": "bd75b100485c0afa0cdddc73d1880160eea8c0b8c91f556245c665d591285ed6",
"migrated_sha256": "bd75b100485c0afa0cdddc73d1880160eea8c0b8c91f556245c665d591285ed6",
"equal": true,
"first_difference": null
},
@@ -95,8 +95,8 @@
"endpoint": "/api/dragon-tiger?trade_date=2026-07-30",
"original_status": 200,
"migrated_status": 200,
"original_sha256": "9d19029f655813ed40a25deb0b3250423a40ae0ee779638ec90f85333f859c6f",
"migrated_sha256": "9d19029f655813ed40a25deb0b3250423a40ae0ee779638ec90f85333f859c6f",
"original_sha256": "1d6936e33aea2bb4258d1d4d007385626d0c638fc290b66271f00649dba59358",
"migrated_sha256": "1d6936e33aea2bb4258d1d4d007385626d0c638fc290b66271f00649dba59358",
"equal": true,
"first_difference": null
},
@@ -216,8 +216,8 @@
"endpoint": "/api/alerts?status=all&as_of=2026-07-30",
"original_status": 200,
"migrated_status": 200,
"original_sha256": "01e8f169fbc8700313dbdf8379e682b78d7b372cfbee556e1d68f46cd6cbeecf",
"migrated_sha256": "01e8f169fbc8700313dbdf8379e682b78d7b372cfbee556e1d68f46cd6cbeecf",
"original_sha256": "b611e936ada5d547c728e33e811630282d1eecd7c1ab1517e22b34902eda032a",
"migrated_sha256": "b611e936ada5d547c728e33e811630282d1eecd7c1ab1517e22b34902eda032a",
"equal": true,
"first_difference": null
},
@@ -0,0 +1,83 @@
# 保真迁移完成度逐项审计
> 审计对象:`webapp/app/`
> 审计基线:`xiaobai-preservation-slice-11-candidate-20260731`
> 结论口径:自动证据通过不替代用户人工验收,也不授权部署切换
本审计回答的不是“测试是否为绿色”,而是迁移总纲、批准目录、产品表面和人工维护目标是否
分别有可复验的证据。状态只使用:`自动闭环``人工闭环``明确保留``未执行`
## 1. 目标与证据矩阵
| 要求 | 当前实现 | 主要证据 | 状态 |
|---|---|---|---|
| 原版是唯一产品/视觉基线 | 所有切片从根目录原版复制或机械移动;规格书不覆盖真实行为 | 总纲;切片00-11源码映射 | 自动闭环 |
| `next/`完全隔离 | `next/`未进入候选导入、资产、测试或部署路径;切片10后Git差分为0 | `next失败冻结记录.md`Git路径审计 | 自动闭环 |
| 批准的模块化单体目录 | `backend/{bootstrap,http,features,data,database,jobs,llm}``frontend/{shared,pages,styles,vendor}``config/data/tests/tools`均已归位 | `app/config/architecture-inventory.json`;切片01-10 | 自动闭环 |
| 不制造第二套产品实现 | 根入口是兼容外壳;业务方法从原类机械移动,专项测试逐符号比较AST/哈希 | 切片01-10 preservation tests | 自动闭环 |
| 16个主工作区及内部能力完整 | 页面注册表有16个主工作区;策略跟踪、搜索、详情、提醒、账户和系统管理保留 | `pages.config.json`;前端边界测试;Playwright | 自动闭环 |
| API路径、权限、错误语义不变 | 53个精确路径和11个正则路径有唯一所有者;鉴权在后端执行 | `api.config.json`;HTTP/权限测试;21端点差分 | 自动闭环 |
| 数据源有统一边界和质量规则 | 18个数据集登记来源、用途、单位、新鲜度、覆盖率、复权和失败关闭规则 | data registriesDataGateway/quality tests | 自动闭环 |
| 数据库无损、用户隔离不变 | 36张表由初始schema和有序migration管理;62个schema对象、21张关键表差分相同 | 数据库差分;migration/repository/account-boundary tests | 自动闭环 |
| 后台任务可追踪和重试 | 3个任务登记调度、锁、超时、重试和输出版本;运行状态持久化 | `jobs.config.json`job runner tests | 自动闭环 |
| LLM唯一治理边界 | 问师、问天、复盘助手和策略编译经统一网关执行会员、额度、模型回退、流式和审计规则 | 切片07/09LLM gateway/stream tests | 自动闭环 |
| 前端唯一请求出口、Shell和页面职责 | 只有`frontend/shared/api.js`调用`fetch`;Shell、状态、弹窗、页面生命周期及页面模块已归位 | frontend boundary tests;切片10源码重组哈希 | 自动闭环 |
| CSS、主题、动画和移动行为不改写 | 原七层CSS和问天动画按字节/源码移动;令牌层与加载顺序受测试保护;用户逐页确认无迁移视觉回归 | CSS governance;切片10/11浏览器证据;`manual-acceptance.md` | 人工闭环 |
| 不确定代码先记录再试删 | 2个文件和5个无消费者函数经自动与人工验收确认废弃;历史表兼容责任继续保留 | `uncertain-code-audit.md`cleanup contract`manual-acceptance.md` | 人工闭环 |
| 可由人工维护者复验 | 候选README/架构说明、工具分类、API/架构生成器、统一验证命令均可从`app/`独立运行;系统临时目录导出实测242项候选测试通过 | `app/tools/README.md`maintenance tool tests;独立导出复验 | 自动闭环 |
| 正式数据库与部署不受影响 | 验收只使用隔离SQLite副本和非8765端口 | 各切片README;运行记录 | 自动闭环 |
| Docker/NAS和正式入口切换 | 用户已明确本轮不做NAS Docker测试;人工验收前禁止切换 | 迁移状态和切换指南 | 未执行 |
## 2. 结构减法的可量化结果
-`server.py`从原版约5,851行降为25行稳定入口;实现归入启动、HTTP和产品领域模块。
-`database.py`从原版2,839行降为746行,只保留初始schema、组合入口及有明确历史兼容责任的
方法;各领域查询已移入对应Repository。
- 20个其他根Python兼容模块均为1-15行导出/模块别名,不含第二套实现。
- 原9,283行前端运行时按原连续源码范围拆入共享层和13个页面目录;试删后仍由源码重组测试
保护,不能悄悄增加、丢失或重排原行为。
- 已确认没有消费者的`demo_data.py`、旧问天加载文件和5个函数进入可单项恢复的试删候选;
`wencai_saved_queries`因旧数据库兼容和账号隔离继续保留。
大文件不等于自动可删。筛选引擎、Tushare适配器、问天确定性计算和原CSS仍是单一有效实现;
在没有更细的同输入差分与人工视觉确认前继续拆分,反而违反“整理原件而非重拍”的迁移约束。
它们已登记在`config/architecture-inventory.json``code_hotspots`,供后续正常维护按领域处理。
## 3. 本次严格审计发现并关闭的缺口
候选版初次自动验收后,三个维护工具仍带有旧目录假设:基线验证器引用不存在的`static/`
架构清单写向不存在的候选文档目录,原样清单生成器假设`app/app`且会因试删文件直接异常。
这些问题不改变产品页面,却会让接管者无法按指南复验,因此不能视为可维护性完成。
关闭方式:
1. 基线验证器改为检查候选全量测试、两个生成注册表、`frontend/`下全部JavaScript、Git差异和
测试数据库完整性。
2. 候选架构清单固定生成到`config/architecture-inventory.json`,从正式候选路径统计16页、
64类路由匹配、36张表、供应商/LLM入口、CSS层和热点文件。
3. 原样资产清单、方法搬运和前端拆分明确标为迁移期工具;可能写文件的操作必须显式传路径或
`--apply`,默认帮助路径不改文件。
4. 新增维护工具契约测试,保证生成文件新鲜、所有工具`--help`可用,旧`static/`路径不会回归。
5. 跨到2026-08-01周六后,实时详情测试中用运行日构造的“固定时间”不再代表交易日;原版和
候选测试夹具同步固定到明确工作日,消除跨午夜/周末不确定性。产品实现与交易日规则未改,
随后统一验收再次通过302项Python测试和45项Playwright。
6. 独立导出`app/`后成功启动健康接口,236项非迁移业务/前端测试通过;同时发现六项日常前端
契约经辅助函数隐式读取旧`static/app.js`。候选运行时重组现使用已审计的9,283行覆盖边界,
只有明确的保真差分断言继续读取原版,避免未来清理旧目录后日常契约失效。
7. 对切片10十组截图重新做像素审计,九组平均色差均低于0.3/255;集合竞价候选图实际为登录
失效页,已明确标为无效并撤销其截图证明力。集合竞价最终视觉继续列为人工验收项。
8. 统一验证器现在按环境选择完整保真套件或候选自有套件;系统临时目录中的纯`app/`导出通过
242项测试、注册表、24个JavaScript文件和SQLite检查。架构热点大小按归一化UTF-8计算,
CRLF/LF不再导致清单失效;正式仓库最终通过305项测试和45项Playwright。
## 4. 最终人工裁决与部署边界
2026-08-01用户在隔离端口`8797`浏览全部页面并测试全部功能,确认视觉与功能迁移成功;发现的
问题几乎都属于原版遗留问题,未发现阻止验收的迁移回归。该结论已经关闭以下迁移裁决:
1. 切片11试删候选确认为废弃,恢复标签继续保留。
2. 页面视觉、功能、交互和动画的人工验收完成。
3. 允许建立本地迁移完成标签并推送Gitea。
该结论不授权切换正式数据库、Docker或NAS,也不授权删除根目录原版、数据库备份或冻结的
`next/`记录。部署切换继续以`人工维护与本地切换指南.md`为准,并须单独获得用户批准。
@@ -56,8 +56,8 @@
"table": "alerts",
"original_count": 2,
"migrated_count": 2,
"original_sha256": "a496857f915d8ba0870274455182a2d7f180a931836b7078e9676dc0e7bbe430",
"migrated_sha256": "a496857f915d8ba0870274455182a2d7f180a931836b7078e9676dc0e7bbe430",
"original_sha256": "3fd3b7ceb0950d543fba9cebdd2b4294fe187b37a6b8d4755a49428661b3ae1c",
"migrated_sha256": "3fd3b7ceb0950d543fba9cebdd2b4294fe187b37a6b8d4755a49428661b3ae1c",
"equal": true,
"excluded_columns": []
},
@@ -0,0 +1,33 @@
# 切片 11:用户人工验收记录
> 验收日期:2026-08-01
> 验收地址:`http://127.0.0.1:8797/`
> 数据边界:迁移审计数据库副本,未写正式数据库
> 结论:通过
## 验收范围
用户在迁移版中浏览了全部页面,并测试了全部可操作功能。验收覆盖页面视觉、布局、主题、
交互、动画、行情读取、LLM功能及主要写入流程。验收期间使用的Tushare、iFinD和主LLM均完成
真实最小调用验证,迁移版能够读取数据库中加密保存的原配置。
## 用户结论
迁移版在视觉和功能上迁移成功。人工检查中发现的问题几乎全部属于原版已经存在的遗留问题;
未发现阻止本次验收的`app/`目录迁移回归。
## 最终裁决
1. 接受`app/`与原版在功能、视觉、交互和动画上的保真结果。
2. 确认切片11登记的`demo_data.py`、旧问天加载文件和5个无消费者前端函数可以永久废弃。
3. 保留`wencai_saved_queries`及证据不足的相邻资产,继续承担旧数据库兼容责任。
4. 允许建立本地迁移完成提交和标签并推送Gitea。
5. 本次验收不授权切换Docker/NAS、删除原版或改写正式数据库;部署切换必须另行决定。
## 证据关系
- 自动完成度审计:`completion-audit.md`
- 不确定代码处置:`uncertain-code-audit.md`
- 浏览器自动记录:`browser-acceptance.json`
- API和数据库差分:`api-diff.json``database-diff.json`
- 回档基线:`xiaobai-preservation-slice-10-20260731`
@@ -0,0 +1,27 @@
{
"schema_version": 1,
"audited_at": "2026-08-01T01:55:00+08:00",
"source_directory": "docs/migration/evidence/slice-10",
"difference_threshold_per_channel": 8,
"valid_pairs": [
{"view": "dark-sentiment-1920x1080", "size": "1683x1080", "mean_absolute_difference": 0.065, "changed_pixel_percent": 0.123},
{"view": "dark-sentiment-390x844", "size": "375x812", "mean_absolute_difference": 0.009, "changed_pixel_percent": 0.012},
{"view": "heaven-fortune-light-1920x1080", "size": "1683x1080", "mean_absolute_difference": 0.043, "changed_pixel_percent": 0.046},
{"view": "heaven-heart-breath-light-1920x1080", "size": "1683x1080", "mean_absolute_difference": 0.256, "changed_pixel_percent": 0.545},
{"view": "heaven-heart-intro-light-1920x1080", "size": "1683x1080", "mean_absolute_difference": 0.047, "changed_pixel_percent": 0.068},
{"view": "heaven-trend-390x844", "size": "390x844", "mean_absolute_difference": 0.284, "changed_pixel_percent": 0.712},
{"view": "heaven-trend-light-1920x1080", "size": "1683x1080", "mean_absolute_difference": 0.040, "changed_pixel_percent": 0.063},
{"view": "light-1920x1080", "size": "1683x1080", "mean_absolute_difference": 0.000, "changed_pixel_percent": 0.000},
{"view": "screener-light-1920x1080", "size": "1683x1080", "mean_absolute_difference": 0.064, "changed_pixel_percent": 0.157}
],
"invalid_pairs": [
{
"view": "auction-light-1920x1080",
"mean_absolute_difference": 15.820,
"changed_pixel_percent": 77.932,
"reason": "The migrated image is an expired-login page, not the auction workspace.",
"disposition": "Renamed INVALID-login-session and excluded from visual-equivalence evidence."
}
],
"conclusion": "Nine valid pairs are geometrically and visually consistent within dynamic rendering noise. Auction remains a manual visual acceptance item."
}
@@ -3,9 +3,11 @@
> 审计基线:`dec3cd12365df5e7d4c391369f14c457cf56d7d9`
> 恢复标签:`xiaobai-preservation-slice-10-20260731`
> 原则:只有静态引用、动态注册、运行路径和兼容责任同时排除后才试删;其余继续保留
> 当前状态:试删后的自动、API、数据库和浏览器验收已通过,等待用户人工确认
> 当前状态:自动与人工验收已通过,试删项于2026-08-01确认废弃
本日志记录试删前的原位置、内容指纹、证据、决定和恢复方式。试删不等于永久废弃;自动回归通过后仍需用户在迁移版真实页面完成人工验收,才能把删除结论视为最终确认。
本日志记录试删前的原位置、内容指纹、证据、决定和恢复方式。试删阶段不等于永久废弃;
2026-08-01用户完成迁移版全部页面和功能人工检查,确认视觉与功能迁移成功,所见问题几乎都
属于原版遗留问题,未发现阻止验收的迁移回归。该人工结论关闭了本日志的最终确认门槛。
## 1. `app/demo_data.py`
@@ -15,7 +17,7 @@
- 动态/注册路径:没有模块名字符串、插件注册或反射加载。
- 产品事实:`app/README.md`明确主行情不再回退演示数据;行情服务和Repository只负责排除历史`source=demo`缓存,未依赖本文件。
- 兼容责任:不参与数据库schema、历史记录解释或配置读取。
- 决定:进入试删
- 最终决定:确认废弃;切片10标签继续保留单项恢复能力
- 恢复:从切片10标签恢复`app/demo_data.py`
## 2. `app/frontend/heaven-loading.js`
@@ -26,7 +28,7 @@
- 动态/注册路径:没有脚本清单、页面注册表或运行时代码引用旧路径;新旧文件虽然都导出`window.HeavenLoadingCanvas`,但浏览器只能执行v2。
- 运行证据:切片10已覆盖观势、观气和观心解读加载动画,v2节点、动画名及可见行为与原版基线一致。
- 兼容责任:不是数据库、配置或历史数据资产。
- 决定:进入试删
- 最终决定:确认废弃;切片10标签继续保留单项恢复能力
- 恢复:从切片10标签恢复该文件。
## 3. 五个疑似无引用前端函数
@@ -35,11 +37,11 @@
| 函数 | 位置 | 原用途线索 | 决定 |
|---|---|---|---|
| `commonReviewColumns` | `frontend/shared/export.js` | 旧通用导出列定义;当前每个导出入口均显式传列 | 试删 |
| `outcomeClass` | `frontend/app.js` | 旧昨日涨停结果样式映射;当前渲染不调用 | 试删 |
| `screenerResultMatchesSelection` | `frontend/pages/screener/page.js` | 旧选股结果存在性包装;已由`activeScreenerResultEntry`直接承担 | 试删 |
| `selectRegime` | `frontend/pages/screener/page.js` | 旧阶段手动切换;当前阶段由盘后数据写入且无按钮绑定 | 试删 |
| `showHeartRitualCurtain` | `frontend/pages/heaven/page.js` | 旧观心幕帘过场;当前流程直接调用`activateHeartRises` | 试删 |
| `commonReviewColumns` | `frontend/shared/export.js` | 旧通用导出列定义;当前每个导出入口均显式传列 | 确认废弃 |
| `outcomeClass` | `frontend/app.js` | 旧昨日涨停结果样式映射;当前渲染不调用 | 确认废弃 |
| `screenerResultMatchesSelection` | `frontend/pages/screener/page.js` | 旧选股结果存在性包装;已由`activeScreenerResultEntry`直接承担 | 确认废弃 |
| `selectRegime` | `frontend/pages/screener/page.js` | 旧阶段手动切换;当前阶段由盘后数据写入且无按钮绑定 | 确认废弃 |
| `showHeartRitualCurtain` | `frontend/pages/heaven/page.js` | 旧观心幕帘过场;当前流程直接调用`activateHeartRises` | 确认废弃 |
本次只删除这5个函数,不连带删除`selectedRegime``heartRitualCurtain`隐藏节点、`heartCurtainTimer`或相关CSS。后者与仍在使用的选股状态、观心节点和复合选择器相邻,尚不足以证明可独立删除,默认保留。
@@ -76,4 +78,14 @@
- 桌面日间/夜间及390x844移动端真实页面检查通过,问天三页和观气底部内容可达。
- `wencai_saved_queries`及三个方法保持存在,并由清理契约测试持续保护。
因此上述删除保持“候选试删”状态;只有用户人工验收后才改为“确认废弃”。
## 7. 人工最终确认
- 验收日期:2026-08-01。
- 验收运行时:隔离端口`8797`与迁移审计数据库副本。
- 验收范围:用户逐页浏览全部页面并测试全部可操作功能。
- 验收结论:视觉与功能迁移成功;发现的问题几乎都是原版遗留问题,未发现阻止验收的迁移回归。
- 删除结论:`demo_data.py`、旧问天加载文件和上述5个无消费者函数正式确认为废弃;
`wencai_saved_queries`及证据不足的相邻DOM、状态和CSS继续保留。
确认废弃不取消恢复证据。若后续发现历史兼容责任,可从
`xiaobai-preservation-slice-10-20260731`按本日志记录单项恢复,不回退其他迁移成果。
@@ -1,12 +1,13 @@
# 小白复盘人工维护与本地切换指南
> 适用目录:`webapp/app/`
> 当前状态:迁移候选已完成自动验收,尚未获得最终人工验收,不得替换正式部署
> 当前状态:本地迁移已完成自动与用户人工验收;正式数据库、Docker和NAS尚未切换
## 1. 先确认哪个版本是正式版本
- 当前正式基线仍是`webapp/`根目录及根目录`data/review.db`
- `webapp/app/`是保真迁移候选,运行代码来自原版的移动、机械拆分和去重,不是重新开发。
- `webapp/app/`已完成人工验收的保真迁移版本,运行代码来自原版的移动、机械拆分和去重,
不是重新开发。
- `webapp/next/`是已否决的冻结版本,禁止部署、继续开发或复制实现。
- 用户人工验收前,不要删除根级原版,不要让`app/`写入正式数据库,也不要修改NAS容器。
@@ -77,19 +78,13 @@ python -m unittest discover -s tests -q
```powershell
cd webapp\app
python -m unittest discover -s tests -q
npx.cmd playwright test --reporter=dot
python tools\verify_baseline.py
python tools\verify_baseline.py --e2e
```
JavaScript和Git差异:
```powershell
cd webapp\app
$files = Get-ChildItem frontend -Recurse -Filter *.js
foreach ($file in $files) { node --check $file.FullName }
cd ..
git diff --check
```
第一条命令已经包含迁移版全量单元测试、API/架构注册表新鲜度、全部前端JavaScript语法、
Git空白错误和测试数据库只读完整性检查。第二条额外运行Playwright。工具的日常/验收/迁移期
分类见`app/tools/README.md`;迁移期工具不是正常开发命令。
Playwright需要能够启动本机无头Edge。若测试停在浏览器启动前且没有`msedge`进程,先检查执行
环境是否禁止GUI/无头浏览器进程;这不是页面失败,不要通过删除测试或延长产品超时绕过。
@@ -114,6 +109,10 @@ app/tools/compare_preservation_databases.py
## 7. 人工验收清单
2026-08-01用户已在隔离端口`8797`完成全部页面和功能验收,确认视觉与功能迁移成功;所见问题
几乎都属于原版遗留问题,未发现阻止验收的迁移回归。以下清单继续作为后续结构修改和部署
切换时的回归标准:
- 用同一账号、日期和主题对照原版与迁移版全部页面。
- 检查日间/夜间、1080P/4K、390像素移动端和浏览器缩放后的滚动与弹窗。
- 检查图表悬浮、股票/板块/题材/指数详情、全局搜索和日期切换。
@@ -126,7 +125,7 @@ app/tools/compare_preservation_databases.py
人工验收发现差异时,记录页面、账号、日期、主题、视口、输入和截图;先对照原版复现,再判断
是迁移回归还是原版既有问题。
## 8. 获得批准后的本地切换方案
## 8. 获得部署批准后的本地切换方案
以下只是准备步骤,本次迁移没有执行:
@@ -150,5 +149,5 @@ Docker/NAS切换应以`app/`作为构建上下文,另行执行构建、卷挂
4. 从切换记录指定的原版提交重新启动根级`server.py`
5. 验证登录、健康接口、最近交易日、私有数据和模型配置后恢复使用。
切片11试删可从`xiaobai-preservation-slice-10-20260731`单项恢复;不要用破坏性的Git重置覆盖
正式数据或用户未提交的代码。
切片11已确认废弃项仍可从`xiaobai-preservation-slice-10-20260731`单项恢复;不要用破坏性的
Git重置覆盖正式数据或用户未提交的代码。
+16 -10
View File
@@ -1,7 +1,7 @@
{
"schema_version": 1,
"updated_at": "2026-07-31T16:10:00+08:00",
"status": "awaiting_manual_acceptance",
"updated_at": "2026-08-01T09:42:04+08:00",
"status": "completed_local_handoff",
"migration_mode": "behavior_preserving_source_migration",
"source_of_truth": "current_original_webapp_runtime_and_source",
"source_root": ".",
@@ -9,11 +9,11 @@
"failed_roots": [
"next"
],
"current_slice": "slice-11-uncertain-code-audit-final-acceptance-handoff",
"last_completed_slice": "slice-10-frontend-shell-pages-components-css-mobile",
"last_automated_slice": "slice-11-uncertain-code-audit-final-acceptance-handoff",
"last_checkpoint": "xiaobai-preservation-slice-11-candidate-20260731",
"next_action": "user_acceptance_on_local_port_8797_then_confirm_or_restore_each_trial_retirement; do_not_switch_docker_or_nas_before_approval",
"current_slice": "slice-11-strict-completion-audit-maintenance-handoff",
"last_completed_slice": "slice-11-strict-completion-audit-maintenance-handoff",
"last_automated_slice": "slice-11-strict-completion-audit-maintenance-handoff",
"last_checkpoint": "xiaobai-preservation-complete-20260801",
"next_action": "preservation_migration_complete; keep the original runtime as the deployment rollback baseline until a separately approved local or Docker/NAS switch",
"authoritative_documents": [
"AGENTS.md",
"docs/migration/原版保真迁移总纲.md",
@@ -35,11 +35,17 @@
"target_directory_name": "app",
"target_directory_structure": "approved_2026-07-30",
"migration_slice_order": "approved_2026-07-30",
"execution_mode": "autonomous_until_complete"
"execution_mode": "autonomous_until_complete",
"manual_visual_and_functional_acceptance": "approved_2026-08-01",
"slice_11_trial_retirements": "confirmed_retired_2026-08-01"
},
"manual_acceptance": {
"accepted_at": "2026-08-01T09:42:04+08:00",
"runtime": "http://127.0.0.1:8797/",
"scope": "all pages and all user-testable functions",
"result": "visual and functional preservation accepted; observed issues were predominantly original-version legacy issues, with no migration regression found that blocks acceptance"
},
"open_decisions": [
"user_acceptance_of_visual_interaction_and_animation_equivalence",
"final_confirmation_or_restore_of_slice_11_trial_retirements",
"local_and_docker_switch_timing"
]
}
+37 -8
View File
@@ -1,6 +1,6 @@
# 小白复盘保真迁移账本
> 当前状态:切片11自动验收完成,等待用户人工验收;正式切换尚未执行
> 当前状态:切片11自动与人工验收完成,本地保真迁移交付完成;正式部署切换尚未执行
本账本是上下文恢复和人工审计的连续记录。任何迁移提交必须在同一提交中更新本文件及
`保真迁移状态.json`
@@ -31,6 +31,9 @@
| 2026-07-31 | `xiaobai-preservation-slice-09-20260731` | 复盘、自选、交易日志、提醒与复盘助手原实现归位 | 自动、API、数据库、账户隔离与浏览器差分通过,进入切片10 |
| 2026-07-31 | `xiaobai-preservation-slice-10-20260731` | 前端Shell、页面、CSS、动画与移动端职责归位 | 源码重组、自动、API、数据库、桌面、夜间、移动端与动画差分通过,进入切片11 |
| 2026-07-31 | `xiaobai-preservation-slice-11-candidate-20260731` | 不确定代码审计、全量验收与人工交接准备 | 自动、API、数据库和浏览器验收通过;等待人工确认,未切换部署 |
| 2026-07-31 | `xiaobai-preservation-slice-11-audit-candidate-20260731` | 逐项目标审计并修复候选维护工具旧路径 | 302项测试、24个脚本、45项Playwright、21项API与62项schema差分通过;仍等待人工确认 |
| 2026-08-01 | `xiaobai-preservation-slice-11-audit-candidate-standalone-20260801` | 独立维护与截图证据复核 | 正式仓库305项、独立导出242项测试通过;撤销无效竞价截图;仍等待人工确认 |
| 2026-08-01 | `xiaobai-preservation-complete-20260801` | 用户逐页逐功能验收并完成迁移收尾 | 视觉与功能保真通过;试删确认废弃;本地交付完成,部署未切换 |
## 资产处置登记
@@ -40,9 +43,9 @@
|---|---|---|---|---|---|---|
| 原版运行源文件(排除`next/`、日志、缓存、构建产物和正式数据库) | 运行资产 | 全站 | 原样保留后逐项移动 | `app/` | 388项受控资产哈希一致;231项Python与45项Playwright测试通过 | 已复制 |
| `next/` | 失败实现 | 无正式消费者 | 原样保留但禁止迁移 | - | 用户冻结决定 | 已冻结 |
| `demo_data.py` | 旧演示代码 | 无运行、动态、配置或兼容消费者 | 候选试删 | 切片10标签可单项恢复 | 全量自动、API、数据库浏览器验收通过 | 待人工确认 |
| `static/heaven-loading.js` | 旧动画 | 页面只加载`heaven-loading-v2.js` | 候选试删 | 切片10标签可单项恢复 | 问天三页加载动画45项Playwright通过 | 待人工确认 |
| `commonReviewColumns`等5个前端函数 | 无引用符号 | 定义外引用为0 | 候选试删 | 切片10标签可按原行号恢复 | 审计白名单比较、24个JS语法浏览器流程通过 | 待人工确认 |
| `demo_data.py` | 旧演示代码 | 无运行、动态、配置或兼容消费者 | 确认废弃 | 切片10标签可单项恢复 | 全量自动、API、数据库浏览器及用户人工验收通过 | 已删除 |
| `static/heaven-loading.js` | 旧动画 | 页面只加载`heaven-loading-v2.js` | 确认废弃 | 切片10标签可单项恢复 | 问天三页加载动画45项Playwright及用户人工验收通过 | 已删除 |
| `commonReviewColumns`等5个前端函数 | 无引用符号 | 定义外引用为0 | 确认废弃 | 切片10标签可按原行号恢复 | 审计白名单比较、24个JS语法浏览器流程及用户人工验收通过 | 已删除 |
| `wencai_saved_queries`及其方法 | 历史兼容数据 | 旧库兼容与用户隔离 | 原样保留 | `app/backend/database/`兼容区 | schema及关键表差分一致;清理契约持续保护 | 保留 |
| 现有7层CSS | 视觉运行资产 | 全部页面和主题 | 原样保留后逐页归档 | `app/frontend/` | 必须通过截图与计算样式差分 | 保留 |
| `sentiment_engine.py` | 情绪周期计算 | 总览、轮动、选股 | 移动并保留兼容别名 | `app/backend/features/sentiment/engine.py` | 文件哈希与原版一致;248项Python与45项Playwright通过 | 已移动 |
@@ -190,19 +193,45 @@
- 回档:标签`xiaobai-preservation-slice-10-20260731`
- 完整证据:`docs/migration/evidence/slice-10/README.md`
自动验收完成、等待人工确认的切片:`slice-11-uncertain-code-audit-final-acceptance-handoff`
已完成切片:`slice-11-uncertain-code-audit-final-acceptance-handoff`
- 原版基线:提交`dec3cd12365df5e7d4c391369f14c457cf56d7d9`,即切片10回档点。
- 审计范围:`demo_data.py`、旧问天加载动画、5个无引用前端函数及`wencai_saved_queries`兼容责任。
- 处置:前3类进入可单项回档的候选试删`wencai_saved_queries`表和方法因旧库兼容与用户隔离继续保留。
- 处置:前3类进入可单项回档的候选试删,并在2026-08-01人工验收后确认废弃;
`wencai_saved_queries`表和方法因旧库兼容与用户隔离继续保留。
- 保护方式:全部历史前端源码测试统一使用审计白名单比较,只允许登记的原`app.js`行号缺失,其他差异仍会失败。
- API与数据库:21个全站只读API全部一致;62个schema对象和21张关键表逐行一致。
- 验收:原版231项、迁移版298项Python测试、24个JavaScript语法检查和45项Playwright1.6分钟)通过;桌面日间/夜间、390×844五个移动入口和问天三页真实检查通过。
- 验收:原版231项、迁移版302项Python测试、24个JavaScript语法检查和45项Playwright通过;桌面日间/夜间、390×844五个移动入口和问天三页真实检查通过。
- 部署边界:未改正式数据库、`.env``8765`、Docker或NAS;原版仍是正式运行基线。
- 候选回档:标签`xiaobai-preservation-slice-11-candidate-20260731`试删最终结论仍等待用户人工确认。
- 候选回档:标签`xiaobai-preservation-slice-11-candidate-20260731`最终完成标签
`xiaobai-preservation-complete-20260801`;试删恢复能力继续由切片10标签保留。
- 完整证据:`docs/migration/evidence/slice-11/README.md`
- 维护交接:`docs/migration/人工维护与本地切换指南.md`
严格完成审计补充:
- 发现候选`verify_baseline.py`、架构清单和原样资产清单仍带有旧`static/`、候选`docs/`
`app/app`路径假设;产品运行不受影响,但人工接管命令不可用,因此属于迁移目标缺口。
- 修复后统一验证命令检查302项Python测试、两个生成注册表、24个前端脚本、Git空白、SQLite
完整性和45项Playwright,并在Windows上显式管理8876测试服务器生命周期。
- 21个固定API使用同一数据库时点重新比较全部一致;数据库62个schema对象和21张关键表一致,
仅按既有规则排除内置策略启动校准的`updated_at`
- 候选自身的16页、64类路由匹配、36张表、数据/LLM/CSS入口和热点文件登记在
`app/config/architecture-inventory.json`;逐项结论见`completion-audit.md`
- 严格审计回档标签为`xiaobai-preservation-slice-11-audit-candidate-20260731`。该标签只代表自动
候选;最终人工验收与删除裁决记录在完成标签中。
- 2026-08-01跨日复验发现原版与候选共用的一项实时详情测试夹具依赖运行日,进入周六后会把
非交易日误作预期合并日;两版测试夹具同步固定到明确工作日,产品代码和日期规则未改。
- 跨日修复后统一验收再次通过302项Python测试、24个JavaScript文件、SQLite完整性和45项
Playwright;新增候选回档标签`xiaobai-preservation-slice-11-audit-candidate-20260801`
-`app/`导出在系统临时目录成功运行统一维护命令:242项候选测试、注册表、24个JavaScript
文件和SQLite检查通过;正式仓库保真套件为305项Python测试及45项Playwright通过。
- 日常前端契约不再隐式读取旧`static/app.js`,只有迁移差分断言保留原版依赖;架构热点大小
按归一化UTF-8统计,CRLF/LF跨环境结果一致。
- 切片10集合竞价候选截图实际为登录失效页,已重命名并撤销证明力;九组有效截图像素差异
低于动态渲染噪声阈值。2026-08-01用户随后在真实登录状态下逐页浏览并测试全部功能,确认
视觉与功能迁移成功;发现的问题几乎都属于原版遗留问题,未发现阻止验收的迁移回归。
## 决策记录
| 日期 | 决策 | 原因 |
+2 -2
View File
@@ -46,7 +46,7 @@ class RealtimeClientStub:
class FixedMarketDatetime(datetime):
fixed_now = datetime.now().astimezone().replace(hour=10, minute=30, second=0, microsecond=0)
fixed_now = datetime(2026, 7, 31, 10, 30).astimezone()
@classmethod
def now(cls, tz=None):
@@ -54,7 +54,7 @@ class FixedMarketDatetime(datetime):
class FixedPreopenDatetime(datetime):
fixed_now = datetime.now().astimezone().replace(hour=8, minute=45, second=0, microsecond=0)
fixed_now = datetime(2026, 7, 31, 8, 45).astimezone()
@classmethod
def now(cls, tz=None):