refactor: move ifind pool helpers to feature service

app/ARCHITECTURE.md

@@ -28,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
@@ -51,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
 

app/backend/application.py

@@ -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()

app/backend/bootstrap/config.py

@@ -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):

app/backend/bootstrap/container.py

@@ -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

app/backend/data/numbers.py

@@ -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

app/backend/data/providers/tushare_client.py

@@ -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"

app/backend/features/heaven/agent.py

@@ -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}"

app/backend/features/market/charts.py

@@ -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)

app/backend/features/market/insights.py

@@ -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:

app/backend/features/mentor/agent.py

@@ -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}"

app/backend/features/mentor/http.py

@@ -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))

app/backend/features/pools/service.py

@@ -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()

app/backend/features/review/agent.py

@@ -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
 
 

app/backend/features/review/http.py

@@ -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:

app/backend/features/screener/compiler.py

@@ -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}"

app/backend/features/screener/engine.py

@@ -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

app/backend/features/sentiment/engine.py

@@ -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))
 

app/backend/http/handler.py

@@ -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}")

app/backend/llm/transport.py

@@ -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 ""

app/config/architecture-inventory.json

@@ -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",
@@ -269,13 +324,13 @@
     },
     {
       "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",
@@ -304,8 +359,8 @@
     },
     {
       "path": "backend/features/market/insights.py",
-      "bytes": 58150,
-      "lines": 1314
+      "bytes": 57998,
+      "lines": 1307
     },
     {
       "path": "frontend/pages/market/runtime.js",
@@ -319,8 +374,8 @@
     },
     {
       "path": "backend/application.py",
-      "bytes": 49784,
-      "lines": 1165
+      "bytes": 47552,
+      "lines": 1087
     },
     {
       "path": "frontend/styles/theme.css",

app/sync_data.py

@@ -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:

app/tests/preservation_helpers.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import ast
 import hashlib
 import re
 from pathlib import Path
@@ -33,6 +34,41 @@ 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"):

app/tests/test_data_gateway.py

@@ -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)

app/tests/test_feature_boundaries.py

@@ -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"))

app/tests/test_http_dispatch.py

@@ -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()

app/tests/test_http_streaming.py

@@ -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()

app/tests/test_llm_transport.py

@@ -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()

app/tests/test_market_symbol_normalization.py

@@ -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()

app/tests/test_mentor_stream.py

@@ -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(

app/tests/test_numeric_normalization.py

@@ -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()

app/tests/test_preservation_slice_accounts.py

@@ -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__)

app/tests/test_preservation_slice_heaven.py

@@ -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(

app/tests/test_preservation_slice_market.py

@@ -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)

app/tests/test_preservation_slice_market_insights.py

@@ -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"

app/tests/test_preservation_slice_mentor_llm.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"),

app/tests/test_preservation_slice_review_alerts.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)

app/tests/test_preservation_slice_screener.py

@@ -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)

app/tests/test_preservation_slice_sentiment_pools.py

@@ -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(

app/tests/test_review_assistant.py

@@ -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"

app/tools/build_api_registry.py

@@ -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',

app/tools/build_architecture_inventory.py

@@ -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}
 
@@ -127,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,
@@ -156,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"},
@@ -163,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(),
     }

docs/governance/code-reduction.md

@@ -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运行结果，确认未发现明显异常。本记录仅表示本轮可见功能与
+  页面使用未发现明显回归，不替代后续批次各自的自动测试和人工抽查。