Compare commits

..
60 changed files with 1678 additions and 599 deletions
+21 -9
View File
@@ -1,8 +1,9 @@
# Candidate architecture
`app/` is the behavior-preserving modular candidate. The original `webapp/` runtime remains
the product and visual baseline until manual acceptance. `next/` is a rejected, frozen
implementation and is not a source for this directory.
`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 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
@@ -27,18 +28,25 @@ background scheduler
- `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.
- `backend/bootstrap/` owns process configuration, dependency construction, startup, and
shared input/display-format contracts. It does not own feature behavior.
- `backend/http/` owns common authentication, request IDs, JSON/NDJSON responses, static
delivery, streaming connection lifecycle, 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.
coverage, display-versus-calculation eligibility, and shared numeric normalization policies.
- `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, streaming rules,
and call audit. Feature agents only prepare context and provider payloads.
- `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
@@ -50,7 +58,11 @@ background scheduler
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.
the original public import surface is part of the preservation contract. Canonical backend
modules must import other canonical modules directly rather than routing through these aliases.
The remaining `api_access` import in `backend/application.py` and preserved lazy
`sentiment_engine` import in the screener repository are registered transition boundaries;
the root `database.py` remains the documented schema/composition anchor.
## Non-negotiable maintenance rules
+3 -3
View File
@@ -2,9 +2,9 @@
一个面向 A 股盘后复盘的本地 Web 工作台。后端使用 Python 访问 Tushare Pro,前端不依赖构建工具。
本目录是从原版源码逐项移动、机械拆分并完成差分验证的模块化候选,不是依据规格书重新开发的
第二套产品。人工验收和正式切换前,`webapp/`根目录仍是唯一正式基线;冻结的`next/`不得用于
部署或后续开发。目录职责见[ARCHITECTURE.md](ARCHITECTURE.md)。
本目录是从原版源码逐项移动、机械拆分并完成差分验证与用户人工验收的模块化正式源码,
不是依据规格书重新开发的第二套产品。正式部署切换前,`webapp/`根目录继续作为当前部署与
回档基线;冻结的`next/`不得用于部署或后续开发。目录职责见[ARCHITECTURE.md](ARCHITECTURE.md)。
当前包含集合竞价、涨停池、炸板池、跌停板、昨日涨停、涨停表现、市场天梯、板块轮动、题材库、人气热榜、龙虎榜和个人复盘工作区。交易日快照与同步记录保存在本地 SQLite 数据库 `data/review.db`
+43 -121
View File
@@ -472,41 +472,43 @@ class DashboardService(
**self.database.status(),
}
@staticmethod
def _ifind_field(row: dict[str, Any], tokens: tuple[str, ...]) -> Any:
for key, value in row.items():
label = str(key or "")
if any(token.casefold() == label.casefold() for token in tokens):
return value
for key, value in row.items():
label = str(key or "")
if any(token in label for token in tokens):
return value
return None
@classmethod
def _ifind_row_code(cls, row: dict[str, Any]) -> str:
value = cls._ifind_field(row, ("股票代码", "证券代码", "代码", "thscode"))
match = re.search(r"(?<!\d)(\d{6})(?!\d)", str(value or ""))
if match:
return match.group(1)
for value in row.values():
match = re.search(r"(?<!\d)(\d{6})\.(?:SH|SZ|BJ)(?![A-Z])", str(value or ""), re.I)
if match:
return match.group(1)
return ""
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 +524,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 +873,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 +893,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 +900,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 +907,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:
@@ -1056,16 +988,6 @@ class RequestHandler(
return
self.send_json({"error": "Not found"}, HTTPStatus.NOT_FOUND)
def _write_stream_event(self, payload: dict[str, Any]) -> None:
self.wfile.write(
(json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + "\n").encode("utf-8")
)
self.wfile.flush()
def save_reason(self) -> None:
try:
body = self.read_json_body()
+4
View File
@@ -71,6 +71,10 @@ def normalize_date(value: str) -> str:
return parsed.strftime("%Y%m%d")
def display_compact_date(value: str) -> str:
return f"{value[:4]}-{value[4:6]}-{value[6:8]}" if len(value) == 8 else value
def validate_stock_code(value: str) -> str:
code = value.strip()
if not re.fullmatch(r"\d{6}", code):
+1 -1
View File
@@ -9,10 +9,10 @@ from backend.database.repositories import RepositoryBundle, build_repository_bun
from backend.features.alerts import AlertService
from backend.features.mentor.agent import MentorSkillRegistry
from backend.features.review import TradeJournalService
from backend.features.screener.engine import ScreenerEngine
from backend.features.screener.tracking import StrategyTrackingService
from backend.jobs import InProcessJobRunner, JobRegistry, SQLiteJobRunRepository
from database import ReviewDatabase
from screener import ScreenerEngine
from backend.data.providers.ifind_client import IfindHttpClient
from backend.data.realtime import WebRealtimeAggregator
from backend.features.market.charts import MarketChartClient
+20
View File
@@ -0,0 +1,20 @@
from __future__ import annotations
import math
from typing import Any
def finite_number(value: Any, default: float = 0.0) -> float:
try:
number = float(value)
return number if math.isfinite(number) else default
except (TypeError, ValueError):
return default
def non_nan_number(value: Any, default: float = 0.0) -> float:
try:
number = float(value)
return number if number == number else default
except (TypeError, ValueError):
return default
+2 -12
View File
@@ -11,6 +11,8 @@ from datetime import datetime, time as dt_time, timedelta
from threading import Lock
from typing import Any, ClassVar
from backend.bootstrap.config import display_compact_date as _display_date
from backend.data.numbers import finite_number as _number
from backend.features.sentiment.engine import apply_sentiment_to_dashboard
@@ -1812,14 +1814,6 @@ class TushareClient:
}
def _number(value: Any, default: float = 0.0) -> float:
try:
number = float(value)
return number if math.isfinite(number) else default
except (TypeError, ValueError):
return default
def _text(value: Any) -> str:
if isinstance(value, (list, tuple, set)):
return "".join(str(item).strip() for item in value if str(item).strip())
@@ -2014,10 +2008,6 @@ def _display_time(value: Any) -> str:
return f"{raw[:2]}:{raw[2:4]}:{raw[4:6]}"
def _display_date(value: str) -> str:
return f"{value[:4]}-{value[4:6]}-{value[6:8]}" if len(value) == 8 else value
def _realtime_market_status(current_time: dt_time) -> str:
if current_time < dt_time(9, 25):
return "pre_open"
+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)
+1 -8
View File
@@ -6,6 +6,7 @@ from datetime import datetime, time as dt_time, timedelta, timezone
from statistics import median
from typing import TYPE_CHECKING, Any, Callable
from backend.data.numbers import non_nan_number as _number
from backend.data.providers.ifind_client import IfindError, IfindHttpClient
from backend.data.providers.tushare_client import TushareClient, TushareError
@@ -16,14 +17,6 @@ if TYPE_CHECKING:
CHINA_TIMEZONE = timezone(timedelta(hours=8))
def _number(value: Any, default: float = 0.0) -> float:
try:
number = float(value)
return number if number == number else default
except (TypeError, ValueError):
return default
def _display_date(value: str) -> str:
text = str(value or "").replace("-", "")
if len(text) != 8:
+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}"
+1 -16
View File
@@ -14,19 +14,4 @@ class MentorHttpMixin:
except (ValueError, json.JSONDecodeError) as exc:
self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)
return
self.send_response(HTTPStatus.OK)
self.send_header("Content-Type", "application/x-ndjson; charset=utf-8")
self.send_header("Cache-Control", "no-cache, no-transform")
self.send_header("X-Accel-Buffering", "no")
self.send_header("Connection", "close")
self.end_headers()
try:
for event in stream:
self._write_stream_event(event)
self._write_stream_event({"type": "done"})
except (ValueError, MentorAgentError) as exc:
self._write_stream_event({"type": "error", "error": str(exc)})
except (BrokenPipeError, ConnectionResetError):
pass
finally:
self.close_connection = True
self.send_ndjson_stream(stream, (ValueError, MentorAgentError))
+24
View File
@@ -117,6 +117,30 @@ class PoolServiceMixin:
finally:
self._ifind_event_lock.release()
@staticmethod
def _ifind_field(row: dict[str, Any], tokens: tuple[str, ...]) -> Any:
for key, value in row.items():
label = str(key or "")
if any(token.casefold() == label.casefold() for token in tokens):
return value
for key, value in row.items():
label = str(key or "")
if any(token in label for token in tokens):
return value
return None
@classmethod
def _ifind_row_code(cls, row: dict[str, Any]) -> str:
value = cls._ifind_field(row, ("股票代码", "证券代码", "代码", "thscode"))
match = re.search(r"(?<!\d)(\d{6})(?!\d)", str(value or ""))
if match:
return match.group(1)
for value in row.values():
match = re.search(r"(?<!\d)(\d{6})\.(?:SH|SZ|BJ)(?![A-Z])", str(value or ""), re.I)
if match:
return match.group(1)
return ""
@staticmethod
def _normalize_ifind_event_time(value: Any) -> str:
text = str(value or "").strip()
+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
+2 -16
View File
@@ -26,22 +26,8 @@ class ReviewHttpMixin:
except (ValueError, json.JSONDecodeError) as exc:
self.send_json({"error": str(exc)}, HTTPStatus.BAD_REQUEST)
return
self.send_response(HTTPStatus.OK)
self.send_header("Content-Type", "application/x-ndjson; charset=utf-8")
self.send_header("Cache-Control", "no-cache, no-transform")
self.send_header("X-Accel-Buffering", "no")
self.send_header("Connection", "close")
self.end_headers()
try:
for chunk in stream:
self._write_stream_event({"type": "delta", "content": chunk})
self._write_stream_event({"type": "done"})
except (ValueError, ReviewAssistantError) as exc:
self._write_stream_event({"type": "error", "error": str(exc)})
except (BrokenPipeError, ConnectionResetError):
pass
finally:
self.close_connection = True
events = ({"type": "delta", "content": chunk} for chunk in stream)
self.send_ndjson_stream(events, (ValueError, ReviewAssistantError))
def save_watchlist(self) -> None:
try:
+28 -74
View File
@@ -1,12 +1,10 @@
from __future__ import annotations
import json
import time
import urllib.error
import urllib.request
from typing import Any
from screener import FACTOR_FIELDS, REGIMES
from backend.llm import transport as llm_transport
from backend.features.screener.engine import FACTOR_FIELDS, REGIMES
class LLMCompilerError(RuntimeError):
@@ -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}"
+4 -14
View File
@@ -8,10 +8,12 @@ from collections import defaultdict
from datetime import datetime, timedelta
from typing import Any
from advanced_strategies import ADVANCED_CURATED_STRATEGIES
from backend.bootstrap.config import display_compact_date as _display_date
from backend.data.numbers import finite_number as _number
from backend.data.providers.tushare_client import TushareClient, TushareError
from backend.features.screener.strategies import ADVANCED_CURATED_STRATEGIES
from database import ReviewDatabase
from backend.features.sentiment.engine import build_sentiment_history, latest_contiguous_history
from tushare_client import TushareClient, TushareError
REGIMES = {
@@ -2199,15 +2201,3 @@ def _regime_reason(regime: str) -> str:
"divergence": "指数或核心仍强,但广度、封板质量开始分化。",
"retreat": "情绪指标继续走弱,应提高筛选门槛并接受无候选结果。",
}.get(regime, "市场阶段待确认。")
def _number(value: Any, default: float = 0.0) -> float:
try:
number = float(value)
return number if math.isfinite(number) else default
except (TypeError, ValueError):
return default
def _display_date(value: str) -> str:
return f"{value[:4]}-{value[4:6]}-{value[6:8]}" if len(value) == 8 else value
+2 -8
View File
@@ -4,6 +4,8 @@ from copy import deepcopy
from statistics import mean, median
from typing import Any
from backend.data.numbers import non_nan_number as _number
COMPONENT_WEIGHTS = {
"breadth": 20,
@@ -16,14 +18,6 @@ COMPONENT_WEIGHTS = {
SENTIMENT_ENGINE_VERSION = 2
def _number(value: Any, default: float = 0.0) -> float:
try:
number = float(value)
return number if number == number else default
except (TypeError, ValueError):
return default
def _clamp(value: float, lower: float = 0.0, upper: float = 100.0) -> float:
return min(upper, max(lower, value))
+29
View File
@@ -3,6 +3,7 @@ from __future__ import annotations
import json
import mimetypes
import secrets
from collections.abc import Iterable
from http import HTTPStatus
from http.cookies import SimpleCookie
from typing import Any
@@ -142,5 +143,33 @@ class HttpTransportMixin:
self.end_headers()
self.wfile.write(content)
def _write_stream_event(self, payload: dict[str, Any]) -> None:
self.wfile.write(
(json.dumps(payload, ensure_ascii=False, separators=(",", ":")) + "\n").encode("utf-8")
)
self.wfile.flush()
def send_ndjson_stream(
self,
events: Iterable[dict[str, Any]],
error_types: tuple[type[Exception], ...],
) -> None:
self.send_response(HTTPStatus.OK)
self.send_header("Content-Type", "application/x-ndjson; charset=utf-8")
self.send_header("Cache-Control", "no-cache, no-transform")
self.send_header("X-Accel-Buffering", "no")
self.send_header("Connection", "close")
self.end_headers()
try:
for event in events:
self._write_stream_event(event)
self._write_stream_event({"type": "done"})
except error_types as exc:
self._write_stream_event({"type": "error", "error": str(exc)})
except (BrokenPipeError, ConnectionResetError):
pass
finally:
self.close_connection = True
def log_message(self, format_string: str, *args: Any) -> None:
print(f"[{self.log_date_time_string()}] {format_string % args}")
+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 ""
+70 -15
View File
@@ -1,6 +1,6 @@
{
"schema_version": 1,
"captured_from": "app modular preservation candidate",
"captured_from": "app accepted modular runtime",
"runtime": {
"http_server": "http.server.ThreadingHTTPServer",
"application_processes": 1,
@@ -220,6 +220,41 @@
"runtime_role": "index observation fallback"
}
],
"provider_construction": [
{
"client": "TushareClient",
"owner": "backend/data/providers/tushare.py",
"compatibility_fallback": "backend/features/market/service.py"
},
{
"client": "IfindHttpClient",
"owner": "backend/data/gateway.py"
},
{
"client": "MarketChartClient",
"owner": "backend/data/gateway.py"
},
{
"client": "WebRealtimeAggregator",
"owner": "backend/data/gateway.py"
}
],
"numeric_normalization": [
{
"function": "finite_number",
"path": "backend/data/numbers.py"
},
{
"function": "non_nan_number",
"path": "backend/data/numbers.py"
}
],
"date_formatting": [
{
"function": "display_compact_date",
"path": "backend/bootstrap/config.py"
}
],
"llm_entrypoints": [
{
"function": "stream_with_mentor",
@@ -242,6 +277,26 @@
"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"
}
],
"http_transport": [
{
"function": "send_json",
"path": "backend/http/handler.py"
},
{
"function": "send_ndjson_stream",
"path": "backend/http/handler.py"
}
],
"css_layers": [
"/shared/tokens.css?v=20260729-1",
"/styles/styles.css",
@@ -254,7 +309,7 @@
"code_hotspots": [
{
"path": "frontend/styles/styles.css",
"bytes": 361780,
"bytes": 361776,
"lines": 15465
},
{
@@ -269,22 +324,22 @@
},
{
"path": "backend/features/screener/engine.py",
"bytes": 108552,
"lines": 2213
"bytes": 108387,
"lines": 2203
},
{
"path": "backend/data/providers/tushare_client.py",
"bytes": 94329,
"lines": 2175
"bytes": 94124,
"lines": 2165
},
{
"path": "frontend/app.js",
"bytes": 91151,
"bytes": 89213,
"lines": 1939
},
{
"path": "frontend/pages/heaven/page.js",
"bytes": 88322,
"bytes": 86493,
"lines": 1830
},
{
@@ -299,17 +354,17 @@
},
{
"path": "backend/features/heaven/service.py",
"bytes": 64421,
"bytes": 63123,
"lines": 1303
},
{
"path": "backend/features/market/insights.py",
"bytes": 58150,
"lines": 1314
"bytes": 57998,
"lines": 1307
},
{
"path": "frontend/pages/market/runtime.js",
"bytes": 57053,
"bytes": 55720,
"lines": 1333
},
{
@@ -319,8 +374,8 @@
},
{
"path": "backend/application.py",
"bytes": 50934,
"lines": 1165
"bytes": 47552,
"lines": 1087
},
{
"path": "frontend/styles/theme.css",
@@ -329,7 +384,7 @@
},
{
"path": "database.py",
"bytes": 34013,
"bytes": 33284,
"lines": 746
}
]
+2 -1
View File
@@ -3,7 +3,8 @@ from __future__ import annotations
import argparse
from datetime import date
from server import SERVICE, normalize_date
from backend.application import SERVICE
from backend.bootstrap.config import normalize_date
def main() -> None:
+40 -5
View File
@@ -1,5 +1,6 @@
from __future__ import annotations
import ast
import hashlib
import re
from pathlib import Path
@@ -26,12 +27,48 @@ RETIRED_FRONTEND_SOURCE_RANGES = (
(9022, 9027),
(9052, 9055),
)
AUDITED_FRONTEND_SOURCE_LINE_COUNT = 9283
def sha256(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
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),
)
def module_contract(
path: Path,
*,
excluded_definitions: set[str] | None = None,
excluded_import_modules: set[str] | None = None,
exclude_imports: bool = False,
) -> str:
excluded_definitions = excluded_definitions or set()
excluded_import_modules = excluded_import_modules or set()
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
tree.body = [
node
for node in tree.body
if not (exclude_imports and isinstance(node, (ast.Import, ast.ImportFrom)))
and not (
isinstance(node, ast.ImportFrom)
and node.module in excluded_import_modules
)
and getattr(node, "name", None) not in excluded_definitions
]
return ast.dump(tree, include_attributes=False)
def reassembled_frontend_runtime() -> str:
chunks: dict[tuple[int, int], str] = {}
for path in FRONTEND_ROOT.rglob("*.js"):
@@ -51,12 +88,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)
+29 -2
View File
@@ -1,7 +1,9 @@
from __future__ import annotations
import ast
import unittest
from datetime import datetime, timedelta
from pathlib import Path
from backend.data import (
DataPolicyError,
@@ -45,8 +47,6 @@ class DataGatewayTests(unittest.TestCase):
self.assertIs(gateway.chart_data.ifind, gateway.ifind)
def test_server_has_no_direct_runtime_tushare_construction(self) -> None:
from pathlib import Path
source = (
Path(__file__).resolve().parents[1]
/ "backend"
@@ -57,6 +57,33 @@ class DataGatewayTests(unittest.TestCase):
self.assertEqual(source.count("TushareClient(self.token)"), 1)
self.assertIn("return gateway.tushare()", source)
def test_provider_construction_has_unique_declared_owners(self) -> None:
root = Path(__file__).resolve().parents[1]
owners = {
"EastmoneyChartClient": {"backend/data/gateway.py"},
"IfindHttpClient": {"backend/data/gateway.py"},
"IfindProvider": {"backend/data/gateway.py"},
"MarketChartClient": {"backend/data/gateway.py"},
"TushareClient": {"backend/features/market/service.py"},
"TushareProvider": {"backend/data/gateway.py"},
"WebRealtimeAggregator": {"backend/data/gateway.py"},
}
found = {name: set() for name in owners}
for path in (root / "backend").rglob("*.py"):
relative = path.relative_to(root).as_posix()
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
for node in ast.walk(tree):
if not isinstance(node, ast.Call):
continue
name = getattr(node.func, "id", None) or getattr(node.func, "attr", None)
if name in found:
found[name].add(relative)
self.assertEqual(found, owners)
provider_source = (root / "backend/data/providers/tushare.py").read_text(
encoding="utf-8"
)
self.assertIn("client_factory: Callable[[str], TushareClient] = TushareClient", provider_source)
def test_quality_gate_accepts_matching_daily_evidence(self) -> None:
timezone = market_timezone()
now = datetime(2026, 7, 29, 16, 0, tzinfo=timezone)
+47 -6
View File
@@ -20,12 +20,6 @@ class FeatureBoundaryTests(unittest.TestCase):
}
violations = []
for path in FEATURES.rglob("*.py"):
# The screener engine is an exact-preservation move of the legacy
# calculation module. Its provider dependency is covered by the
# slice equivalence tests and will be addressed only after the
# behavior-preserving migration is complete.
if path.relative_to(FEATURES).as_posix() == "screener/engine.py":
continue
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
for node in ast.walk(tree):
names = []
@@ -38,6 +32,53 @@ class FeatureBoundaryTests(unittest.TestCase):
violations.append(f"{path.relative_to(ROOT)} -> {name}")
self.assertEqual(violations, [])
def test_backend_uses_root_compatibility_modules_only_at_declared_boundaries(self) -> None:
compatibility_modules = {
"advanced_strategies",
"alert_service",
"api_access",
"app_config",
"assistant_agent",
"chart_data_provider",
"heaven_agent",
"heaven_engine",
"ifind_client",
"llm_strategy",
"llm_stream",
"market_insights",
"mentor_agent",
"realtime_aggregator",
"screener",
"security",
"sentiment_engine",
"server",
"strategy_tracking",
"trade_journal",
"tushare_client",
}
allowed = {
"backend/application.py": {"api_access"},
"backend/features/screener/repository.py": {"sentiment_engine"},
}
violations = []
for path in (ROOT / "backend").rglob("*.py"):
relative = path.relative_to(ROOT).as_posix()
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
for node in ast.walk(tree):
names = []
if isinstance(node, ast.Import):
names = [alias.name for alias in node.names]
elif isinstance(node, ast.ImportFrom) and node.module:
names = [node.module]
for name in names:
root_name = name.split(".")[0]
if (
root_name in compatibility_modules
and root_name not in allowed.get(relative, set())
):
violations.append(f"{relative} -> {name}")
self.assertEqual(violations, [])
def test_legacy_service_modules_are_compatibility_exports_only(self) -> None:
for filename in ("alert_service.py", "trade_journal.py", "strategy_tracking.py"):
tree = ast.parse((ROOT / filename).read_text(encoding="utf-8"))
+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()
+85
View File
@@ -0,0 +1,85 @@
from __future__ import annotations
import io
import json
import unittest
from backend.http.handler import HttpTransportMixin
class StreamError(RuntimeError):
pass
class TransportStub(HttpTransportMixin):
def __init__(self) -> None:
self.statuses: list[int] = []
self.response_headers: list[tuple[str, str]] = []
self.wfile = io.BytesIO()
self.close_connection = False
def send_response(self, status: int) -> None:
self.statuses.append(int(status))
def send_header(self, name: str, value: str) -> None:
self.response_headers.append((name, value))
def end_headers(self) -> None:
pass
class HttpStreamingTests(unittest.TestCase):
def test_stream_transport_preserves_headers_events_and_completion(self) -> None:
handler = TransportStub()
handler.send_ndjson_stream(
({"type": "delta", "content": value} for value in ("", "")),
(StreamError,),
)
self.assertEqual(handler.statuses, [200])
self.assertEqual(
dict(handler.response_headers),
{
"Content-Type": "application/x-ndjson; charset=utf-8",
"Cache-Control": "no-cache, no-transform",
"X-Accel-Buffering": "no",
"Connection": "close",
},
)
events = [
json.loads(line)
for line in handler.wfile.getvalue().decode("utf-8").splitlines()
]
self.assertEqual(
events,
[
{"type": "delta", "content": ""},
{"type": "delta", "content": ""},
{"type": "done"},
],
)
self.assertTrue(handler.close_connection)
def test_stream_transport_preserves_feature_error_event(self) -> None:
def events():
yield {"type": "delta", "content": "partial"}
raise StreamError("stream failed")
handler = TransportStub()
handler.send_ndjson_stream(events(), (StreamError,))
payloads = [
json.loads(line)
for line in handler.wfile.getvalue().decode("utf-8").splitlines()
]
self.assertEqual(
payloads,
[
{"type": "delta", "content": "partial"},
{"type": "error", "error": "stream failed"},
],
)
self.assertTrue(handler.close_connection)
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()
+22 -1
View File
@@ -3,11 +3,16 @@ 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
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]
@@ -34,6 +39,22 @@ class MaintenanceToolTests(unittest.TestCase):
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("_"):
@@ -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(
+38
View File
@@ -0,0 +1,38 @@
from __future__ import annotations
import math
import unittest
from backend.data.numbers import finite_number, non_nan_number
from backend.data.providers import tushare_client
from backend.features.market import insights
from backend.features.screener import engine as screener_engine
from backend.features.sentiment import engine as sentiment_engine
class NumericNormalizationTests(unittest.TestCase):
def test_consumers_use_their_declared_shared_policy(self) -> None:
self.assertIs(tushare_client._number, finite_number)
self.assertIs(screener_engine._number, finite_number)
self.assertIs(insights._number, non_nan_number)
self.assertIs(sentiment_engine._number, non_nan_number)
def test_finite_policy_preserves_existing_results(self) -> None:
self.assertEqual(finite_number("12.5"), 12.5)
self.assertEqual(finite_number(None), 0.0)
self.assertEqual(finite_number("invalid", 7.0), 7.0)
self.assertEqual(finite_number(math.nan, 7.0), 7.0)
self.assertEqual(finite_number(math.inf, 7.0), 7.0)
self.assertEqual(finite_number(-math.inf, 7.0), 7.0)
def test_non_nan_policy_keeps_infinity_but_rejects_nan(self) -> None:
self.assertEqual(non_nan_number("12.5"), 12.5)
self.assertEqual(non_nan_number(None), 0.0)
self.assertEqual(non_nan_number("invalid", 7.0), 7.0)
self.assertEqual(non_nan_number(math.nan, 7.0), 7.0)
self.assertEqual(non_nan_number(math.inf, 7.0), math.inf)
self.assertEqual(non_nan_number(-math.inf, 7.0), -math.inf)
if __name__ == "__main__":
unittest.main()
@@ -40,6 +40,8 @@ class AccountSliceStructureTests(unittest.TestCase):
"require_access",
"serve_static",
"send_json",
"_write_stream_event",
"send_ndjson_stream",
):
self.assertNotIn(method, RequestHandler.__dict__)
self.assertIn(method, HttpTransportMixin.__dict__)
+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(
+31 -2
View File
@@ -9,13 +9,16 @@ 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.numbers import finite_number
from backend.data.providers import ifind_client as canonical_ifind
from backend.data.providers import tushare_client as canonical_tushare
from backend.features.market import charts
from tests.preservation_helpers import (
assert_frontend_runtime_matches_audited_baseline,
assert_moved_asset_matches,
function_contract,
)
@@ -141,14 +144,40 @@ class MarketSliceSourceEquivalenceTests(unittest.TestCase):
)
for original, migrated in exact_moves:
self.assertEqual(sha256(ORIGINAL_ROOT / original), sha256(APP_ROOT / migrated))
original_tushare = top_level_definitions(ORIGINAL_ROOT / "tushare_client.py")
original_tushare.pop("_number")
original_tushare.pop("_display_date")
self.assertEqual(
top_level_definitions(ORIGINAL_ROOT / "tushare_client.py"),
original_tushare,
top_level_definitions(APP_ROOT / "backend/data/providers/tushare_client.py"),
)
self.assertEqual(
top_level_definitions(ORIGINAL_ROOT / "chart_data_provider.py"),
function_contract(ORIGINAL_ROOT / "tushare_client.py", "_number"),
function_contract(APP_ROOT / "backend/data/numbers.py", "finite_number"),
)
self.assertIs(canonical_tushare._number, finite_number)
self.assertEqual(
function_contract(ORIGINAL_ROOT / "tushare_client.py", "_display_date"),
function_contract(
APP_ROOT / "backend/bootstrap/config.py", "display_compact_date"
),
)
self.assertIs(canonical_tushare._display_date, bootstrap_config.display_compact_date)
original_charts = top_level_definitions(ORIGINAL_ROOT / "chart_data_provider.py")
original_charts.pop("_stock_market_code")
self.assertEqual(
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)
@@ -6,11 +6,13 @@ import unittest
from pathlib import Path
import market_insights
from backend.data.numbers import non_nan_number
from backend.features.market import insights as canonical_insights
from tests.preservation_helpers import (
assert_frontend_runtime_matches_audited_baseline,
assert_moved_asset_matches,
assert_page_prefix_matches,
function_contract,
)
@@ -106,6 +108,11 @@ class MarketInsightsSliceSourceEquivalenceTests(unittest.TestCase):
MARKET_INSIGHT_METHODS,
)
self.assertIs(market_insights.MarketInsightsService, canonical_insights.MarketInsightsService)
self.assertEqual(
function_contract(ORIGINAL_ROOT / "market_insights.py", "_number"),
function_contract(APP_ROOT / "backend/data/numbers.py", "non_nan_number"),
)
self.assertIs(canonical_insights._number, non_nan_number)
def test_dashboard_service_methods_are_exact_original_ast(self) -> None:
original = ORIGINAL_ROOT / "server.py"
@@ -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)
+35 -17
View File
@@ -9,12 +9,16 @@ import advanced_strategies
import llm_strategy
import screener
import strategy_tracking
from backend.bootstrap import config as bootstrap_config
from backend.data.numbers import finite_number
from backend.features.screener import compiler, engine, strategies, tracking
from backend.features.screener import service as screener_service
from tests.preservation_helpers import (
assert_frontend_runtime_matches_audited_baseline,
assert_moved_asset_matches,
assert_page_prefix_matches,
function_contract,
module_contract,
)
@@ -91,14 +95,6 @@ def top_level_definition(path: Path, name: str) -> str:
return ast.dump(node, include_attributes=False)
def module_without_imports(path: Path) -> str:
tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
tree.body = [
node for node in tree.body if not isinstance(node, (ast.Import, ast.ImportFrom))
]
return ast.dump(tree, include_attributes=False)
def sha256(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
@@ -154,11 +150,29 @@ class ScreenerSliceSourceEquivalenceTests(unittest.TestCase):
def test_engine_and_tracking_logic_match_the_original(self) -> None:
self.assertEqual(
module_without_imports(ORIGINAL_ROOT / "screener.py"),
module_without_imports(
APP_ROOT / "backend" / "features" / "screener" / "engine.py"
module_contract(
ORIGINAL_ROOT / "screener.py",
excluded_definitions={"_display_date", "_number"},
exclude_imports=True,
),
module_contract(
APP_ROOT / "backend" / "features" / "screener" / "engine.py",
excluded_definitions={"_display_date", "_number"},
exclude_imports=True,
),
)
self.assertEqual(
function_contract(ORIGINAL_ROOT / "screener.py", "_number"),
function_contract(APP_ROOT / "backend/data/numbers.py", "finite_number"),
)
self.assertIs(engine._number, finite_number)
self.assertEqual(
function_contract(ORIGINAL_ROOT / "screener.py", "_display_date"),
function_contract(
APP_ROOT / "backend/bootstrap/config.py", "display_compact_date"
),
)
self.assertIs(engine._display_date, bootstrap_config.display_compact_date)
self.assertEqual(
class_methods(
ORIGINAL_ROOT / "backend" / "features" / "screener" / "tracking.py",
@@ -170,12 +184,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)
@@ -6,11 +6,14 @@ import unittest
from pathlib import Path
import sentiment_engine
from backend.data.numbers import non_nan_number
from backend.features.sentiment import engine as canonical_engine
from tests.preservation_helpers import (
assert_frontend_runtime_matches_audited_baseline,
assert_moved_asset_matches,
assert_page_prefix_matches,
function_contract,
module_contract,
)
@@ -26,6 +29,8 @@ POOL_METHODS = {
"_apply_reason_overrides",
"_schedule_ifind_event_enrichment",
"_refresh_ifind_event_enrichment",
"_ifind_field",
"_ifind_row_code",
"_normalize_ifind_event_time",
"_merge_ifind_event_enrichment",
}
@@ -94,10 +99,22 @@ class SentimentPoolSliceSourceEquivalenceTests(unittest.TestCase):
def test_sentiment_engine_is_exact_original_with_legacy_alias(self) -> None:
self.assertEqual(
sha256(ORIGINAL_ROOT / "sentiment_engine.py"),
sha256(APP_ROOT / "backend" / "features" / "sentiment" / "engine.py"),
module_contract(
ORIGINAL_ROOT / "sentiment_engine.py",
excluded_definitions={"_number"},
),
module_contract(
APP_ROOT / "backend" / "features" / "sentiment" / "engine.py",
excluded_definitions={"_number"},
excluded_import_modules={"backend.data.numbers"},
),
)
self.assertEqual(
function_contract(ORIGINAL_ROOT / "sentiment_engine.py", "_number"),
function_contract(APP_ROOT / "backend/data/numbers.py", "non_nan_number"),
)
self.assertIs(sentiment_engine, canonical_engine)
self.assertIs(canonical_engine._number, non_nan_number)
def test_api_and_frontend_assets_are_unchanged(self) -> None:
self.assertEqual(
+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):
+7
View File
@@ -14,6 +14,13 @@ maintenance command cannot be mistaken for a historical migration rewrite.
- `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
+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',
+36 -14
View File
@@ -37,16 +37,15 @@ 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}
@@ -75,6 +74,14 @@ 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 = [
"backend/application.py",
@@ -99,13 +106,7 @@ def code_hotspots() -> list[dict[str, Any]]:
path = ROOT / name
if not path.is_file():
continue
rows.append(
{
"path": name,
"bytes": path.stat().st_size,
"lines": len(path.read_text(encoding="utf-8").splitlines()),
}
)
rows.append({"path": name, **source_metrics(path)})
return sorted(rows, key=lambda item: item["bytes"], reverse=True)
@@ -125,7 +126,7 @@ def build() -> dict[str, Any]:
tables = database_inventory(database_sources)
return {
"schema_version": 1,
"captured_from": "app modular preservation candidate",
"captured_from": "app accepted modular runtime",
"runtime": {
"http_server": "http.server.ThreadingHTTPServer",
"application_processes": 1,
@@ -154,6 +155,19 @@ def build() -> dict[str, Any]:
{"provider": "eastmoney", "path": "backend/data/realtime.py", "runtime_role": "isolated realtime observation"},
{"provider": "tencent", "path": "backend/data/realtime.py", "runtime_role": "index observation fallback"},
],
"provider_construction": [
{"client": "TushareClient", "owner": "backend/data/providers/tushare.py", "compatibility_fallback": "backend/features/market/service.py"},
{"client": "IfindHttpClient", "owner": "backend/data/gateway.py"},
{"client": "MarketChartClient", "owner": "backend/data/gateway.py"},
{"client": "WebRealtimeAggregator", "owner": "backend/data/gateway.py"},
],
"numeric_normalization": [
{"function": "finite_number", "path": "backend/data/numbers.py"},
{"function": "non_nan_number", "path": "backend/data/numbers.py"},
],
"date_formatting": [
{"function": "display_compact_date", "path": "backend/bootstrap/config.py"},
],
"llm_entrypoints": [
{"function": "stream_with_mentor", "path": "backend/features/mentor/agent.py"},
{"function": "interpret_heaven", "path": "backend/features/heaven/agent.py"},
@@ -161,6 +175,14 @@ def build() -> dict[str, Any]:
{"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"},
],
"http_transport": [
{"function": "send_json", "path": "backend/http/handler.py"},
{"function": "send_ndjson_stream", "path": "backend/http/handler.py"},
],
"css_layers": css_layers(html),
"code_hotspots": code_hotspots(),
}
+35 -2
View File
@@ -21,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():
@@ -97,7 +130,7 @@ 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"],
@@ -117,7 +150,7 @@ def main() -> int:
"javascript",
[node, "--check", script.relative_to(ROOT).as_posix()],
)
run("patch", ["git", "diff", "--check"])
verify_git_diff()
verify_database()
if args.e2e:
+232
View File
@@ -0,0 +1,232 @@
# `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-04 | 数值归一化策略 | 四个业务模块分别保留两组完全相同的数值转换函数体 | 由`backend/data/numbers.py`集中拥有两种既有语义,消费者保留原局部别名 | 已完成 |
| CR-05 | 根级兼容入口 | 正式后端仍有五处通过迁移兼容模块反向导入规范实现 | 正式代码改用规范路径;兼容入口只服务原公开导入契约 | 已完成 |
| CR-06 | 紧凑日期显示 | Tushare与选股引擎各保留一份完全相同的`YYYYMMDD`显示转换 | 由`bootstrap/config.py`拥有唯一格式策略,消费者保留原局部别名 | 已完成 |
| CR-07 | 数据Provider组装 | 核查网关、容器和业务服务是否重复创建外部数据客户端 | 固化唯一创建位置及兼容例外,不改数据源语义 | 已完成 |
| CR-08 | NDJSON流式传输 | 问师与复盘助手重复维护响应头、事件写入、完成、断线及关闭流程 | HTTP共享层拥有唯一流式连接生命周期 | 已完成 |
| CR-09 | 应用服务门面 | 两个仅服务股池iFinD补全的方法仍错位在全局`DashboardService` | 原函数体机械归位到`PoolServiceMixin` | 已完成 |
## 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`
## CR-04验收口径
- 只合并参数、函数体和运行结果完全一致的数值转换函数,不借本批改变任何业务计算或异常默认值。
- `finite_number`继续拒绝`NaN`与正负无穷;`non_nan_number`继续只拒绝`NaN`并保留正负无穷。
- Tushare与智能选股必须复用有限数策略;市场洞察与情绪引擎必须复用非NaN策略,并继续暴露原局部
`_number`名称以保持兼容。
- 实时行情和图表转换器的空值、默认值或参数签名语义不同,必须继续独立保留,不能因名称相同而合并。
- 原版函数参数和函数体分别与共享实现AST相等;四个消费者的局部别名必须指向对应的唯一函数对象。
## CR-04结果
- 删除Tushare、智能选股、市场洞察和情绪引擎中的四份重复函数体,新建两种明确命名的共享策略;生产
代码净减少约12行,全仓AST扫描不再发现完全相同的函数定义。
- 将可复用的函数及模块AST契约归入测试辅助层,原有迁移保持性测试改为“未改范围保持相等、被替换
函数与共享实现相等、运行时唯一对象”三重断言,没有降低门禁。
- 实时行情`backend/data/realtime.py::_number`与图表`backend/features/market/charts.py::_number`
被明确保留;它们不是本批重复实现,也未改变行为。
- 58项定向测试、41项保持性/治理测试、原版231项、候选320项、纯`app/`导出257项和45项
Playwright通过;24个JavaScript文件、API/架构注册表、Git空白检查和SQLite完整性检查通过。
- 本批不修改前端、CSS、接口、数据来源、行情口径、选股条件、数据库、LLM、权限或部署。
本批基线为`xiaobai-reduction-03-market-symbol-20260801`;检查点为
`xiaobai-reduction-04-numeric-normalization-20260801`
## CR-05验收口径
- 逐项扫描根级Python入口、生产代码、测试、工具和动态导入;没有消费者或兼容责任的入口才能删除。
- 规范后端不得经由`screener``advanced_strategies``tushare_client``server`兼容入口
间接访问已经归位的实现。
- 所有根级模块继续保持原导入名称、导出对象及模块对象身份,既有启动命令和第三方维护脚本不受影响。
- `api_access`、选股Repository的惰性`sentiment_engine`导入及根级`database.py`属于已登记边界,
分别留到HTTP、Repository阶段处理,不在本批跨边界修改。
## CR-05结果
- 审计确认21个根级兼容入口均有测试、工具、启动或原公开导入契约消费者,因此本批没有冒险删除文件。
- 容器、策略编译器、选股引擎及数据同步命令的五处导入改为规范模块路径,正式代码不再通过四个根级
兼容模块反向进入实现;运行代码行数未增加。
- 特性边界测试取消选股引擎旧例外,并新增全后端兼容导入门禁;只允许两项已登记过渡边界,后续代码
无法重新引入隐式根级依赖。
- 候选321项、纯`app/`导出258项、24个JavaScript文件、API/架构注册表、Git空白检查和SQLite
完整性检查通过;本批不涉及页面、CSS或浏览器行为。
- 本批不修改业务计算、策略公式、数据源、API、数据库、LLM、权限、前端或部署。
本批基线为`xiaobai-reduction-04-numeric-normalization-20260801`;检查点为
`xiaobai-reduction-05-compatibility-boundaries-20260801`
## CR-06验收口径
- 只合并参数、函数体和异常行为完全相同的日期/文本转换;名称相似但空值、未来日期、错误文案或
输入格式不同的函数不得合并。
- Tushare与选股引擎继续暴露局部`_display_date`名称,并分别指向唯一共享实现。
- 原版两个`_display_date`函数必须分别与共享实现AST相等,所有原调用结果保持不变。
- 市场洞察的日期显示函数会清理连字符并容忍空值,语义不同,必须继续独立保留。
## CR-06结果
- 删除Tushare与选股引擎内两个重复日期函数体,新增`display_compact_date`唯一策略;生产代码总
行数不增加,重复函数体由两份降为一份。
- 架构清单登记日期格式唯一所有权;保持性测试改为未改范围AST相等、共享函数AST相等和运行时
对象身份三重契约,没有放宽原迁移门禁。
- `normalize_date`、市场洞察日期显示、实时行情时间格式和会员日期边界因语义不同均原样保留。
- 候选321项、纯`app/`导出258项、24个JavaScript文件、API/架构注册表、Git空白检查和SQLite
完整性检查通过;本批不涉及页面、CSS或浏览器行为。
- 本批不修改日期输入规则、业务计算、选股结果、接口、数据库、数据源、LLM、权限或部署。
本批基线为`xiaobai-reduction-05-compatibility-boundaries-20260801`;检查点为
`xiaobai-reduction-06-date-formatting-20260801`
## CR-07验收口径
- iFinD、图表、实时观察器和Provider适配器必须只在`build_data_gateway`创建,并由容器共享。
- Tushare必须继续通过实时Token供应器按需创建,不能为了减少对象数量缓存过期Token。
- 市场服务中为原版隔离测试桩保留的一处`TushareClient(self.token)`是明确兼容例外,不得被误判为
第二条正式数据链路。
- 不得合并Tushare、iFinD、东方财富和腾讯的传输、缓存、重试或降级逻辑。
## CR-07结果
- 全后端构造点扫描确认iFinD、MarketChart、东方财富图表和实时观察器均只有网关一个创建位置;
`ApplicationContainer`暴露的是同一对象引用,没有第二份客户端。
- Tushare Provider使用动态Token供应器,市场服务只有一处已登记测试兼容回退;本批没有发现可安全
删除的生产实现,因此不为追求行数强行修改运行代码。
- 架构清单新增Provider创建所有权,自动测试会在未来出现第二个未登记构造点时失败。
- 候选322项、纯`app/`导出259项、24个JavaScript文件、API/架构注册表、Git空白检查和SQLite
完整性检查通过;本批不涉及页面、CSS或浏览器行为。
- 本批不修改请求频率、缓存、重试、Token更新、数据源选择、计算口径、API、数据库或前端。
本批基线为`xiaobai-reduction-06-date-formatting-20260801`;检查点为
`xiaobai-reduction-07-provider-ownership-20260801`
## CR-08验收口径
- 只合并NDJSON响应头、事件序列化、完成事件、业务错误事件、客户端断线和连接关闭这些传输行为。
- 问师继续直接输出原事件字典;复盘助手继续把文本分片包装为`delta/content`事件。
- 两个功能各自的业务异常类型、请求体错误状态码、错误正文和流创建时机保持不变。
- `_write_stream_event`与连接生命周期必须只由`backend/http/handler.py`拥有,应用大类和功能HTTP
模块不再保留第二份实现。
## CR-08结果
- 删除问师和复盘助手各16行重复流式控制流,并将应用大类中的10行事件写入方法归入HTTP共享层;
新共享实现29行、两个调用适配共3行,生产代码净减少10行。
- 新增专项测试固定四个响应头、中文NDJSON序列化、增量顺序、完成事件、业务错误事件和关闭状态。
- 候选324项、纯`app/`导出261项及45项Playwright通过;24个JavaScript文件、API/架构注册表、
Git空白检查和SQLite完整性检查通过。
- 本批不修改提示词、模型选择、会员计次、流式正文、前端解析、API路径、数据库或数据源。
本批基线为`xiaobai-reduction-07-provider-ownership-20260801`;检查点为
`xiaobai-reduction-08-ndjson-transport-20260801`
## CR-09验收口径
- 只有消费者全部属于单一领域、且能够按原函数体机械移动的方法才从应用门面移出。
- `_ifind_field``_ifind_row_code`继续保持静态/类方法签名、字段优先级、大小写规则和代码正则。
- 账号委托属于稳定公开门面;系统设置属于跨领域协调;后台刷新留到CR-11,本批均不得删除或重写。
- 移动后`DashboardService`必须继续通过Mixin解析同名方法,调用点和返回值不变。
## CR-09结果
- 将iFinD字段匹配和股票代码提取两个方法从应用大类机械移动到股池服务,原版与迁移方法AST逐项
相等;应用大类不再直接拥有股池专属实现。
- 连同迁移期遗留空行,`backend/application.py`减少32行,股池服务增加24行,生产代码净减少8行。
- 候选324项、纯`app/`导出261项、24个JavaScript文件、API/架构注册表、Git空白检查和SQLite
完整性检查通过;本批不涉及页面、CSS或浏览器行为。
- 本批不修改字段匹配、涨跌停原因补全、接口、数据源、缓存、数据库、权限或前端。
本批基线为`xiaobai-reduction-08-ndjson-transport-20260801`;检查点为
`xiaobai-reduction-09-service-facade-20260801`
## 人工验收记录
- 2026-08-01:用户检查CR-02与CR-03运行结果,确认未发现明显异常。本记录仅表示本轮可见功能与
页面使用未发现明显回归,不替代后续批次各自的自动测试和人工抽查。
+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
}
+32 -11
View File
@@ -3,14 +3,17 @@
> 基线:`dec3cd12365df5e7d4c391369f14c457cf56d7d9`
> 候选回档标签:`xiaobai-preservation-slice-11-candidate-20260731`
> 严格审计候选标签:`xiaobai-preservation-slice-11-audit-candidate-20260731`
> 当前结论:自动验收完成,等待用户人工验收;尚未执行正式切换、Docker或NAS部署
> 跨日复验候选标签:`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`:页面未加载的旧问天动画;正式入口继续使用
@@ -31,7 +34,7 @@
| 验证 | 结果 |
|---|---:|
| 原版`python -m unittest discover -s tests -q` | 231项通过 |
| 迁移版`python -m unittest discover -s tests -q` | 302项通过 |
| 迁移版`python -m unittest discover -s tests -q` | 305项通过 |
| 历史切片与前端/清理专项复核 | 通过 |
| 迁移版JavaScript语法检查 | 24个文件通过 |
| `npx.cmd playwright test --reporter=dot` | 45项通过(1.6分钟) |
@@ -39,6 +42,12 @@
| 数据库差分 | 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`:只允许审计登记的五段原始
行号被删除,任何其他新增、删除、重排或内容变化仍会使全部历史测试失败。
@@ -51,6 +60,12 @@ 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. 真实浏览器验收
- 桌面端复核情绪周期、智能选股三个工作区、问天日间/夜间和加载资源;控制台无新增
@@ -63,25 +78,31 @@ Windows下Playwright托管Python静态服务器会在45项执行后不退出,
- 同账号、同数据库快照、同主题和同视口下,原版与迁移版问天布局和计算样式一致。
机器可读记录见`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`
@@ -5,7 +5,7 @@
> 结论口径:自动证据通过不替代用户人工验收,也不授权部署切换
本审计回答的不是“测试是否为绿色”,而是迁移总纲、批准目录、产品表面和人工维护目标是否
分别有可复验的证据。状态只使用:`自动闭环``人工待验``明确保留``未执行`
分别有可复验的证据。状态只使用:`自动闭环``人工闭环``明确保留``未执行`
## 1. 目标与证据矩阵
@@ -22,9 +22,9 @@
| 后台任务可追踪和重试 | 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浏览器证据 | 自动闭环 |
| 不确定代码先记录再试删 | 2个文件和5个无消费者函数仅为候选试删;历史表兼容责任继续保留 | `uncertain-code-audit.md`cleanup contract | 人工待验 |
| 可由人工维护者复验 | 候选README/架构说明、工具分类、API/架构生成器、统一验证命令均可从`app/`独立运行 | `app/tools/README.md`maintenance tool tests | 自动闭环 |
| 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测试;人工验收前禁止切换 | 迁移状态和切换指南 | 未执行 |
@@ -58,15 +58,26 @@
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. 仍需用户完成的最终裁决
## 4. 最终人工裁决与部署边界
自动部分完成后仍不能执行以下动作:
2026-08-01用户在隔离端口`8797`浏览全部页面并测试全部功能,确认视觉与功能迁移成功;发现的
问题几乎都属于原版遗留问题,未发现阻止验收的迁移回归。该结论已经关闭以下迁移裁决:
1. 切片11试删候选从“待人工确认”改成“确认废弃”
2. 声称所有页面视觉、交互手感和动画已经由用户确认等价
3. 建立最终完成标签,或把`app/`切换为本地/容器正式入口
4. 删除根目录原版、正式数据库备份或冻结的失败记录。
1. 切片11试删候选确认为废弃,恢复标签继续保留
2. 页面视觉、功能、交互和动画的人工验收完成
3. 允许建立本地迁移完成标签并推送Gitea
本地人工验收入口及逐页清单以`人工维护与本地切换指南.md`为准。发现差异时只回退对应候选,
不使用破坏性Git操作覆盖原版或正式数据
该结论不授权切换正式数据库、Docker或NAS,也不授权删除根目录原版、数据库备份或冻结的
`next/`记录。部署切换继续以`人工维护与本地切换指南.md`为准,并须单独获得用户批准
@@ -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容器。
@@ -108,6 +109,10 @@ app/tools/compare_preservation_databases.py
## 7. 人工验收清单
2026-08-01用户已在隔离端口`8797`完成全部页面和功能验收,确认视觉与功能迁移成功;所见问题
几乎都属于原版遗留问题,未发现阻止验收的迁移回归。以下清单继续作为后续结构修改和部署
切换时的回归标准:
- 用同一账号、日期和主题对照原版与迁移版全部页面。
- 检查日间/夜间、1080P/4K、390像素移动端和浏览器缩放后的滚动与弹窗。
- 检查图表悬浮、股票/板块/题材/指数详情、全局搜索和日期切换。
@@ -120,7 +125,7 @@ app/tools/compare_preservation_databases.py
人工验收发现差异时,记录页面、账号、日期、主题、视口、输入和截图;先对照原版复现,再判断
是迁移回归还是原版既有问题。
## 8. 获得批准后的本地切换方案
## 8. 获得部署批准后的本地切换方案
以下只是准备步骤,本次迁移没有执行:
@@ -144,5 +149,5 @@ Docker/NAS切换应以`app/`作为构建上下文,另行执行构建、卷挂
4. 从切换记录指定的原版提交重新启动根级`server.py`
5. 验证登录、健康接口、最近交易日、私有数据和模型配置后恢复使用。
切片11试删可从`xiaobai-preservation-slice-10-20260731`单项恢复;不要用破坏性的Git重置覆盖
正式数据或用户未提交的代码。
切片11已确认废弃项仍可从`xiaobai-preservation-slice-10-20260731`单项恢复;不要用破坏性的
Git重置覆盖正式数据或用户未提交的代码。
+14 -8
View File
@@ -1,7 +1,7 @@
{
"schema_version": 1,
"updated_at": "2026-07-31T20:57:42+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": ".",
@@ -10,10 +10,10 @@
"next"
],
"current_slice": "slice-11-strict-completion-audit-maintenance-handoff",
"last_completed_slice": "slice-10-frontend-shell-pages-components-css-mobile",
"last_completed_slice": "slice-11-strict-completion-audit-maintenance-handoff",
"last_automated_slice": "slice-11-strict-completion-audit-maintenance-handoff",
"last_checkpoint": "xiaobai-preservation-slice-11-audit-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",
"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"
]
}
+24 -9
View File
@@ -1,6 +1,6 @@
# 小白复盘保真迁移账本
> 当前状态:切片11严格完成审计通过,等待用户人工验收;正式切换尚未执行
> 当前状态:切片11自动与人工验收完成,本地保真迁移交付完成;正式部署切换尚未执行
本账本是上下文恢复和人工审计的连续记录。任何迁移提交必须在同一提交中更新本文件及
`保真迁移状态.json`
@@ -32,6 +32,8 @@
| 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` | 用户逐页逐功能验收并完成迁移收尾 | 视觉与功能保真通过;试删确认废弃;本地交付完成,部署未切换 |
## 资产处置登记
@@ -41,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通过 | 已移动 |
@@ -191,16 +193,18 @@
- 回档:标签`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项、迁移版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`
@@ -214,8 +218,19 @@
仅按既有规则排除内置策略启动校准的`updated_at`
- 候选自身的16页、64类路由匹配、36张表、数据/LLM/CSS入口和热点文件登记在
`app/config/architecture-inventory.json`;逐项结论见`completion-audit.md`
- 严格审计回档标签为`xiaobai-preservation-slice-11-audit-candidate-20260731`。该标签仍不是最终
完成标签,未改变试删候选、人工验收和部署切换边界
- 严格审计回档标签为`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):