diff --git a/app/algorithms/simulation/scenarios.py b/app/algorithms/simulation/scenarios.py
index 579ca86..794f027 100644
--- a/app/algorithms/simulation/scenarios.py
+++ b/app/algorithms/simulation/scenarios.py
@@ -1,3 +1,5 @@
+from __future__ import annotations
+
 import json
 from datetime import datetime
 from functools import wraps
@@ -647,6 +649,7 @@ def pressure_regulation(
     modify_fixed_pump_pattern: dict[str, list] = None,
     modify_variable_pump_pattern: dict[str, list] = None,
     scheme_name: str = None,
+    scada_mappings: simulation.ScadaElementMappings | None = None,
     _temporary_project: str | None = None,
 ) -> None:
     """
@@ -704,5 +707,6 @@ def pressure_regulation(
         scheme_type="pressure_regulation",
         scheme_name=scheme_name,
         result_db_name=name,
+        scada_mappings=scada_mappings,
     )
     # return result
diff --git a/app/api/v1/endpoints/scada.py b/app/api/v1/endpoints/scada.py
index 4c2f3db..c050c46 100644
--- a/app/api/v1/endpoints/scada.py
+++ b/app/api/v1/endpoints/scada.py
@@ -13,21 +13,21 @@ router = APIRouter()
 
 
 @router.get("/network-schemas/scada-device", summary="获取 SCADA 设备结构")
-async def get_scada_device_schema(
+def get_scada_device_schema(
     network: str = Query(..., description="管网名称（或数据库名称）"),
 ) -> dict[str, dict[str, Any]]:
     return get_scada_info_schema(network)
 
 
 @router.get("/scada-devices", summary="获取 SCADA 设备列表")
-async def get_scada_devices(
+def get_scada_devices(
     network: str = Query(..., description="管网名称（或数据库名称）"),
 ) -> list[dict[str, Any]]:
     return get_all_scada_info(network)
 
 
 @router.get("/scada-devices/detail", summary="获取 SCADA 设备")
-async def get_scada_device(
+def get_scada_device(
     network: str = Query(..., description="管网名称（或数据库名称）"),
     device_id: str = Query(..., description="SCADA 设备 ID"),
 ) -> dict[str, Any]:
diff --git a/app/api/v1/endpoints/sensor_placement.py b/app/api/v1/endpoints/sensor_placement.py
index e2f0861..cc5d0a5 100644
--- a/app/api/v1/endpoints/sensor_placement.py
+++ b/app/api/v1/endpoints/sensor_placement.py
@@ -185,7 +185,7 @@ async def get_sensor_placement_runs(
     response_model=SensorPlacementSchemeResponse,
     summary="获取监测点方案详情",
 )
-async def get_sensor_placement_run_detail(
+def get_sensor_placement_run_detail(
     run_id: UUID,
     network: str = Query(..., min_length=1),
     project_context: ProjectContext = Depends(get_project_context),
@@ -204,7 +204,7 @@ async def get_sensor_placement_run_detail(
     response_model=SensorPlacementSchemeResponse,
     summary="覆盖保存监测点方案",
 )
-async def overwrite_sensor_placement_run(
+def overwrite_sensor_placement_run(
     run_id: UUID,
     payload: SensorPlacementUpdateRequest,
     network: str = Query(..., min_length=1),
diff --git a/app/api/v1/endpoints/simulation.py b/app/api/v1/endpoints/simulation.py
index ff200ba..53f2e2b 100644
--- a/app/api/v1/endpoints/simulation.py
+++ b/app/api/v1/endpoints/simulation.py
@@ -6,7 +6,6 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Path, Body
 from fastapi.responses import PlainTextResponse
 from app.auth.keycloak_dependencies import get_current_keycloak_username
 import app.services.simulation as simulation
-import app.services.globals as globals
 from app.services.tjnetwork import (
     run_project,
     run_project_return_dict,
@@ -115,12 +114,16 @@ def run_simulation_manually_by_date(
     if hydraulic_step_seconds <= 0:
         raise ValueError("HYDRAULIC TIMESTEP must be greater than 0.")
     hydraulic_step = timedelta(seconds=hydraulic_step_seconds)
+    scada_mappings = simulation.query_corresponding_element_id_and_query_id(
+        network_name
+    )
     current_time = start_time
     while current_time < end_datetime:
         simulation.run_simulation(
             name=network_name,
             simulation_type="realtime",
             modify_pattern_start_time=current_time.isoformat(timespec="seconds"),
+            scada_mappings=scada_mappings,
         )
         current_time += hydraulic_step
 
@@ -497,9 +500,11 @@ def fastapi_pressure_regulation(data: PressureRegulation = Body(..., description
     支持固定泵和变速泵的独立控制。
     """
     item = data.model_dump()
-    simulation.query_corresponding_element_id_and_query_id(item["network"])
-    fixed_pumps = set(globals.fixed_pumps_id.keys())
-    variable_pumps = set(globals.variable_pumps_id.keys())
+    scada_mappings = simulation.query_corresponding_element_id_and_query_id(
+        item["network"]
+    )
+    fixed_pumps = set(scada_mappings.fixed_pumps)
+    variable_pumps = set(scada_mappings.variable_pumps)
     fixed_pump_pattern: dict[str, list] = {}
     variable_pump_pattern: dict[str, list] = {}
     for pump_id, values in item["pump_control"].items():
@@ -515,6 +520,7 @@ def fastapi_pressure_regulation(data: PressureRegulation = Body(..., description
         modify_fixed_pump_pattern=fixed_pump_pattern or None,
         modify_variable_pump_pattern=variable_pump_pattern or None,
         scheme_name=item["scheme_name"],
+        scada_mappings=scada_mappings,
     )
     return "success"
 
@@ -667,7 +673,6 @@ def fastapi_run_simulation_manually_by_date(
     """
     item = data.model_dump()
     try:
-        simulation.query_corresponding_element_id_and_query_id(item["name"])
         start_time = parse_utc_time(item["start_time"], field_name="start_time")
         run_simulation_manually_by_date(
             item["name"], start_time, item["duration"]
diff --git a/app/api/v1/endpoints/timeseries/analysis.py b/app/api/v1/endpoints/timeseries/analysis.py
index 763146c..5093684 100644
--- a/app/api/v1/endpoints/timeseries/analysis.py
+++ b/app/api/v1/endpoints/timeseries/analysis.py
@@ -1,7 +1,7 @@
 from datetime import datetime
 from uuid import UUID
 
-from fastapi import APIRouter, Body, Depends, HTTPException, Path, Query
+from fastapi import APIRouter, Depends, HTTPException, Query
 from psycopg import AsyncConnection
 
 from app.infra.db.timescaledb.repositories.analysis import AnalysisResultsRepository
@@ -11,27 +11,6 @@ from .dependencies import get_timescale_connection
 router = APIRouter()
 
 
-@router.post("/timeseries/analysis/runs/{run_id}/results", status_code=201)
-async def store_analysis_results(
-    run_id: UUID = Path(..., description="分析运行 ID"),
-    payload: dict = Body(...),
-    conn: AsyncConnection = Depends(get_timescale_connection),
-):
-    try:
-        node_rows = payload.get("node_results", [])
-        link_rows = payload.get("link_results", [])
-        await AnalysisResultsRepository.store_results(
-            conn, run_id, node_rows, link_rows
-        )
-        return {
-            "run_id": run_id,
-            "node_count": len(node_rows),
-            "link_count": len(link_rows),
-        }
-    except ValueError as exc:
-        raise HTTPException(status_code=409, detail=str(exc)) from exc
-
-
 @router.get("/timeseries/analysis/runs/{run_id}/nodes/{node_id}")
 async def get_analysis_node_series(
     run_id: UUID,
diff --git a/app/api/v1/endpoints/timeseries/composite.py b/app/api/v1/endpoints/timeseries/composite.py
index 06ea8dd..7b1e469 100644
--- a/app/api/v1/endpoints/timeseries/composite.py
+++ b/app/api/v1/endpoints/timeseries/composite.py
@@ -226,8 +226,8 @@ async def clean_scada_data(
 @router.get("/pipeline-health-predictions", summary="预测管道健康状况")
 async def predict_pipeline_health(
     query_time: datetime = Query(..., description="查询时间"),
-    network_name: str = Query(..., description="管网名称（或数据库名称）"),
     timescale_conn: AsyncConnection = Depends(get_timescale_connection),
+    postgres_conn: AsyncConnection = Depends(get_postgres_connection),
 ):
     """
     预测管道健康状况
@@ -237,7 +237,6 @@ async def predict_pipeline_health(
 
     Args:
         query_time: 查询时间
-        network_name: 管网名称（或数据库名称）
         timescale_conn: TimescaleDB连接
 
     Returns:
@@ -248,7 +247,7 @@ async def predict_pipeline_health(
     """
     try:
         return await CompositeQueries.predict_pipeline_health(
-            timescale_conn, network_name, query_time
+            timescale_conn, postgres_conn, query_time
         )
     except ValueError as e:
         raise HTTPException(status_code=400, detail=str(e))
diff --git a/app/api/v1/endpoints/timeseries/scada.py b/app/api/v1/endpoints/timeseries/scada.py
index 3f75b98..0fbeccd 100644
--- a/app/api/v1/endpoints/timeseries/scada.py
+++ b/app/api/v1/endpoints/timeseries/scada.py
@@ -2,17 +2,41 @@ from fastapi import APIRouter, Depends, HTTPException, Query, Path, Body
 from typing import List
 from datetime import datetime
 from psycopg import AsyncConnection
+from pydantic import BaseModel, Field, field_validator
 
+from app.infra.db.postgresql.scada import ScadaInfoRepository
 from app.infra.db.timescaledb.repositories.scada import ScadaRepository
-from .dependencies import get_timescale_connection
+from .dependencies import get_postgres_connection, get_timescale_connection
 
 router = APIRouter()
+SCADA_BATCH_MAX_ITEMS = 10_000
+
+
+class ScadaReadingBatchItem(BaseModel):
+    time: datetime
+    device_id: str = Field(min_length=1)
+    monitored_value: float | None = None
+    cleaned_value: float | None = None
+
+    @field_validator("device_id")
+    @classmethod
+    def normalize_device_id(cls, value: str) -> str:
+        normalized = value.strip()
+        if not normalized:
+            raise ValueError("device_id must not be blank")
+        return normalized
 
 
 @router.post("/timeseries/scada-readings/batches", status_code=201, summary="批量插入SCADA监测数据")
 async def insert_scada_data(
-    data: List[dict] = Body(..., description="SCADA设备监测数据列表"),
+    data: List[ScadaReadingBatchItem] = Body(
+        ...,
+        min_length=1,
+        max_length=SCADA_BATCH_MAX_ITEMS,
+        description="SCADA设备监测数据列表",
+    ),
     conn: AsyncConnection = Depends(get_timescale_connection),
+    postgres_conn: AsyncConnection = Depends(get_postgres_connection),
 ):
     """
     批量插入SCADA监测数据
@@ -25,7 +49,20 @@ async def insert_scada_data(
     Returns:
         插入成功的记录数
     """
-    await ScadaRepository.insert_scada_batch(conn, data)
+    rows = [item.model_dump() for item in data]
+    requested_ids = list(dict.fromkeys(item["device_id"] for item in rows))
+    existing_ids = await ScadaInfoRepository.get_existing_device_ids(
+        postgres_conn, requested_ids
+    )
+    missing_ids = [
+        device_id for device_id in requested_ids if device_id not in existing_ids
+    ]
+    if missing_ids:
+        raise HTTPException(
+            status_code=422,
+            detail=f"SCADA devices do not exist in BizDB: {', '.join(missing_ids)}",
+        )
+    await ScadaRepository.insert_scada_batch(conn, rows)
     return {"message": f"Inserted {len(data)} records"}
 
 
diff --git a/app/infra/db/postgresql/analysis.py b/app/infra/db/postgresql/analysis.py
index 7dd405a..03da960 100644
--- a/app/infra/db/postgresql/analysis.py
+++ b/app/infra/db/postgresql/analysis.py
@@ -1,9 +1,105 @@
-from uuid import UUID
+from datetime import datetime
+from typing import Any
+from uuid import UUID, uuid4
 
-from psycopg import AsyncConnection
+from psycopg import AsyncConnection, Connection
+from psycopg.types.json import Jsonb
 
 
 class AnalysisRepository:
+    @staticmethod
+    def create_run_sync(
+        conn: Connection,
+        *,
+        name: str,
+        run_type: str,
+        created_by: str,
+        started_at: datetime,
+        status: str,
+        parameters: dict[str, Any],
+        run_id: UUID | None = None,
+    ) -> dict[str, Any]:
+        execution_id = run_id or uuid4()
+        with conn.cursor() as cur:
+            cur.execute(
+                """
+                INSERT INTO analysis.runs
+                    (run_id, name, run_type, created_by, created_at,
+                     started_at, status, parameters)
+                VALUES (%s, %s, %s, %s, now(), %s, %s, %s)
+                RETURNING run_id, name, run_type, created_by, created_at,
+                          started_at, status, parameters
+                """,
+                (
+                    execution_id,
+                    name,
+                    run_type,
+                    created_by,
+                    started_at,
+                    status,
+                    Jsonb(parameters),
+                ),
+            )
+            created = cur.fetchone()
+        if created is None:
+            raise RuntimeError("analysis run insert returned no row")
+        return created
+
+    @staticmethod
+    def update_run_sync(
+        conn: Connection,
+        run_id: UUID,
+        *,
+        status: str,
+        created_by: str,
+        parameters: dict[str, Any],
+    ) -> None:
+        with conn.cursor() as cur:
+            cur.execute(
+                """
+                UPDATE analysis.runs
+                SET created_by = %s, status = %s, parameters = %s
+                WHERE run_id = %s
+                """,
+                (created_by, status, Jsonb(parameters), run_id),
+            )
+            if cur.rowcount != 1:
+                raise LookupError(f"analysis run {run_id} does not exist")
+
+    @staticmethod
+    def insert_result_sync(
+        conn: Connection,
+        run_id: UUID,
+        *,
+        result_type: str,
+        payload: dict[str, Any],
+        node_id: str | None = None,
+        link_id: str | None = None,
+    ) -> None:
+        with conn.cursor() as cur:
+            cur.execute(
+                """
+                INSERT INTO analysis.results
+                    (run_id, result_type, node_id, link_id, payload)
+                VALUES (%s, %s, %s, %s, %s)
+                """,
+                (run_id, result_type, node_id, link_id, Jsonb(payload)),
+            )
+
+    @staticmethod
+    def get_run_sync(conn: Connection, run_id: UUID) -> dict[str, Any] | None:
+        with conn.cursor() as cur:
+            cur.execute(
+                """
+                SELECT run_id, name, run_type, created_by, created_at,
+                       started_at, status, parameters
+                FROM analysis.runs
+                WHERE run_id = %s
+                """,
+                (run_id,),
+            )
+            return cur.fetchone()
+
     @staticmethod
     async def list_runs(conn: AsyncConnection) -> list[dict]:
         async with conn.cursor() as cur:
diff --git a/app/infra/db/postgresql/scada.py b/app/infra/db/postgresql/scada.py
index 6248c81..6fdea18 100644
--- a/app/infra/db/postgresql/scada.py
+++ b/app/infra/db/postgresql/scada.py
@@ -1,7 +1,40 @@
-from typing import Any
+from dataclasses import dataclass
+from types import MappingProxyType
+from typing import Any, Mapping
 
 from psycopg import AsyncConnection
 
+from app.native.wndb.core.database import read_all, try_read
+
+
+_SCADA_VIEW_SELECT = """
+    SELECT id AS device_id, device_type, node_id, link_id, api_query_id,
+           transmission_mode, transmission_frequency, reliability, x, y
+    FROM gis.scada_devices
+"""
+
+
+@dataclass(frozen=True)
+class ScadaElementMappings:
+    reservoirs: Mapping[str, str]
+    tanks: Mapping[str, str]
+    fixed_pumps: Mapping[str, str]
+    variable_pumps: Mapping[str, str]
+    pressure: Mapping[str, str]
+    demand: Mapping[str, str]
+    quality: Mapping[str, str]
+
+
+def _empty_mapping_groups() -> dict[str, dict[str, str]]:
+    return {
+        "reservoir_liquid_level": {},
+        "tank_liquid_level": {},
+        "fixed_pump": {},
+        "variable_pump": {},
+        "pressure": {},
+        "demand": {},
+        "quality": {},
+    }
 
 def _optional_text(value: Any) -> str | None:
     return str(value).strip() if value is not None else None
@@ -15,6 +48,21 @@ def _optional_int(value: Any) -> int | None:
     return int(value) if value is not None else None
 
 
+def _device(record: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "device_id": str(record["device_id"]).strip(),
+        "device_type": str(record["device_type"]).strip().lower(),
+        "node_id": _optional_text(record["node_id"]),
+        "link_id": _optional_text(record["link_id"]),
+        "api_query_id": _optional_text(record["api_query_id"]),
+        "transmission_mode": record["transmission_mode"],
+        "transmission_frequency": record["transmission_frequency"],
+        "reliability": _optional_int(record["reliability"]),
+        "x": _optional_float(record["x"]),
+        "y": _optional_float(record["y"]),
+    }
+
+
 class ScadaInfoRepository:
     """Read SCADA metadata from the current project's business database."""
 
@@ -22,35 +70,83 @@ class ScadaInfoRepository:
     async def get_scadas(conn: AsyncConnection) -> list[dict[str, Any]]:
         async with conn.cursor() as cur:
             await cur.execute(
-                """
-                SELECT id AS device_id,
-                       device_type,
-                       node_id,
-                       link_id,
-                       api_query_id,
-                       transmission_mode,
-                       transmission_frequency,
-                       reliability,
-                       x,
-                       y
-                FROM gis.scada_devices
-                ORDER BY id
-                """
+                _SCADA_VIEW_SELECT + " ORDER BY device_id"
             )
             records = await cur.fetchall()
 
-        return [
-            {
-                "device_id": str(record["device_id"]).strip(),
-                "device_type": str(record["device_type"]).strip().lower(),
-                "node_id": _optional_text(record["node_id"]),
-                "link_id": _optional_text(record["link_id"]),
-                "api_query_id": _optional_text(record["api_query_id"]),
-                "transmission_mode": record["transmission_mode"],
-                "transmission_frequency": record["transmission_frequency"],
-                "reliability": _optional_int(record["reliability"]),
-                "x": _optional_float(record["x"]),
-                "y": _optional_float(record["y"]),
-            }
-            for record in records
-        ]
+        return [_device(record) for record in records]
+
+    @staticmethod
+    async def get_existing_device_ids(
+        conn: AsyncConnection, device_ids: list[str]
+    ) -> set[str]:
+        if not device_ids:
+            return set()
+        async with conn.cursor() as cur:
+            await cur.execute(
+                "SELECT device_id FROM asset.scada_devices WHERE device_id = ANY(%s)",
+                (device_ids,),
+            )
+            return {str(row["device_id"]).strip() for row in await cur.fetchall()}
+
+
+def get_scada_info_schema(name: str) -> dict[str, dict[str, Any]]:
+    return {
+        "device_id": {"type": "str", "optional": False, "readonly": True},
+        "device_type": {"type": "str", "optional": False, "readonly": True},
+        "node_id": {"type": "str", "optional": True, "readonly": True},
+        "link_id": {"type": "str", "optional": True, "readonly": True},
+        "api_query_id": {"type": "str", "optional": True, "readonly": True},
+        "transmission_mode": {"type": "str", "optional": False, "readonly": True},
+        "transmission_frequency": {"type": "str", "optional": False, "readonly": True},
+        "reliability": {"type": "int", "optional": False, "readonly": True},
+        "x": {"type": "float", "optional": True, "readonly": True},
+        "y": {"type": "float", "optional": True, "readonly": True},
+    }
+
+
+def get_scada_info(name: str, device_id: str) -> dict[str, Any]:
+    row = try_read(
+        name,
+        _SCADA_VIEW_SELECT + " WHERE id = %s",
+        (device_id,),
+    )
+    return _device(row) if row else {}
+
+
+def get_all_scada_info(name: str) -> list[dict[str, Any]]:
+    return [
+        _device(row)
+        for row in read_all(name, _SCADA_VIEW_SELECT + " ORDER BY device_id")
+    ]
+
+
+def load_realtime_element_mappings(name: str) -> ScadaElementMappings:
+    """Load one project-local immutable SCADA-to-model mapping snapshot."""
+    groups = _empty_mapping_groups()
+    rows = read_all(
+        name,
+        """
+        SELECT device_type, COALESCE(node_id, link_id) AS element_id,
+               api_query_id
+        FROM asset.scada_devices
+        WHERE transmission_mode = 'realtime'
+          AND api_query_id IS NOT NULL
+        """,
+    )
+    for row in rows:
+        group = groups.get(str(row["device_type"]).strip().lower())
+        if group is not None:
+            group[str(row["element_id"]).strip()] = str(row["api_query_id"]).strip()
+    immutable = {
+        name: MappingProxyType(values.copy()) for name, values in groups.items()
+    }
+    return ScadaElementMappings(
+        reservoirs=immutable["reservoir_liquid_level"],
+        tanks=immutable["tank_liquid_level"],
+        fixed_pumps=immutable["fixed_pump"],
+        variable_pumps=immutable["variable_pump"],
+        pressure=immutable["pressure"],
+        demand=immutable["demand"],
+        quality=immutable["quality"],
+    )
diff --git a/app/infra/db/postgresql/scada_assets.py b/app/infra/db/postgresql/scada_assets.py
deleted file mode 100644
index 4e501e8..0000000
--- a/app/infra/db/postgresql/scada_assets.py
+++ /dev/null
@@ -1,66 +0,0 @@
-from typing import Any
-
-from app.native.wndb.core.database import read_all, try_read
-
-
-def get_scada_info_schema(name: str) -> dict[str, dict[str, Any]]:
-    return {
-        "device_id": {"type": "str", "optional": False, "readonly": True},
-        "device_type": {"type": "str", "optional": False, "readonly": True},
-        "node_id": {"type": "str", "optional": True, "readonly": True},
-        "link_id": {"type": "str", "optional": True, "readonly": True},
-        "api_query_id": {"type": "str", "optional": True, "readonly": True},
-        "transmission_mode": {"type": "str", "optional": False, "readonly": True},
-        "transmission_frequency": {"type": "str", "optional": False, "readonly": True},
-        "reliability": {"type": "int", "optional": False, "readonly": True},
-        "x": {"type": "float", "optional": True, "readonly": True},
-        "y": {"type": "float", "optional": True, "readonly": True},
-    }
-
-
-_SELECT = """
-    SELECT device_id, device_type, node_id, link_id, api_query_id,
-           transmission_mode, transmission_frequency, reliability,
-           x, y
-    FROM asset.scada_devices
-"""
-
-_SELECT_MATERIALIZED = """
-    SELECT id AS device_id, device_type, node_id, link_id, api_query_id,
-           transmission_mode, transmission_frequency, reliability,
-           x, y
-    FROM gis.scada_devices
-"""
-
-
-def _device(row: dict[str, Any]) -> dict[str, Any]:
-    return {
-        "device_id": str(row["device_id"]),
-        "device_type": str(row["device_type"]),
-        "node_id": str(row["node_id"]) if row["node_id"] is not None else None,
-        "link_id": str(row["link_id"]) if row["link_id"] is not None else None,
-        "api_query_id": (
-            str(row["api_query_id"]) if row["api_query_id"] is not None else None
-        ),
-        "transmission_mode": str(row["transmission_mode"]),
-        "transmission_frequency": str(row["transmission_frequency"]),
-        "reliability": int(row["reliability"]),
-        "x": float(row["x"]) if row["x"] is not None else None,
-        "y": float(row["y"]) if row["y"] is not None else None,
-    }
-
-
-def get_scada_info(name: str, device_id: str) -> dict[str, Any]:
-    row = try_read(
-        name,
-        _SELECT + " WHERE device_id = %s",
-        (device_id,),
-    )
-    return _device(row) if row else {}
-
-
-def get_all_scada_info(name: str) -> list[dict[str, Any]]:
-    return [
-        _device(row)
-        for row in read_all(name, _SELECT_MATERIALIZED + " ORDER BY device_id")
-    ]
diff --git a/app/infra/db/postgresql/sensor_placement.py b/app/infra/db/postgresql/sensor_placement.py
index 3ce1267..32fdecf 100644
--- a/app/infra/db/postgresql/sensor_placement.py
+++ b/app/infra/db/postgresql/sensor_placement.py
@@ -1,9 +1,11 @@
+from datetime import datetime, timezone
 from typing import Any
 from uuid import UUID, uuid4
 
 from psycopg.rows import dict_row
 from psycopg.types.json import Jsonb
 
+from app.infra.db.postgresql.analysis import AnalysisRepository
 from app.native.wndb.core.connection import project_connection
 
 
@@ -63,26 +65,22 @@ def create_sensor_placement(
         "sensor_locations": sensor_locations,
     }
     with project_connection(name) as conn, conn.transaction():
-        with conn.cursor(row_factory=dict_row) as cur:
-            cur.execute(
-                """
-                INSERT INTO analysis.runs
-                    (run_id, name, run_type, created_by, started_at, status, parameters)
-                VALUES (%s, %s, %s, %s, now(), 'completed', '{}'::jsonb)
-                RETURNING run_id, name, created_by, created_at, status
-                """,
-                (run_id, run_name, RUN_TYPE, created_by),
-            )
-            created = cur.fetchone()
-            cur.execute(
-                """
-                INSERT INTO analysis.results (run_id, result_type, payload)
-                VALUES (%s, %s, %s)
-                """,
-                (run_id, RESULT_TYPE, Jsonb(payload)),
-            )
-    if created is None:
-        raise RuntimeError("监测点优化运行写入失败")
+        created = AnalysisRepository.create_run_sync(
+            conn,
+            run_id=run_id,
+            name=run_name,
+            run_type=RUN_TYPE,
+            created_by=created_by,
+            started_at=datetime.now(timezone.utc),
+            status="completed",
+            parameters={},
+        )
+        AnalysisRepository.insert_result_sync(
+            conn,
+            run_id,
+            result_type=RESULT_TYPE,
+            payload=payload,
+        )
     return _placement_row(dict(created) | {"payload": payload})
 
 
diff --git a/app/infra/db/timescaledb/composite_queries.py b/app/infra/db/timescaledb/composite_queries.py
index aec980f..bf3a1ad 100644
--- a/app/infra/db/timescaledb/composite_queries.py
+++ b/app/infra/db/timescaledb/composite_queries.py
@@ -14,7 +14,6 @@ from app.infra.db.postgresql.scada import ScadaInfoRepository
 from app.infra.db.timescaledb.repositories.realtime import RealtimeRepository
 from app.infra.db.timescaledb.repositories.analysis import AnalysisResultsRepository
 from app.infra.db.timescaledb.repositories.scada import ScadaRepository
-from app.native.wndb.model.pipes import get_pipes_by_property
 
 
 class CompositeQueries:
@@ -454,20 +453,25 @@ class CompositeQueries:
         if not cleaned_rows:
             raise ValueError("SCADA 数据清洗未产生任何数据库更新")
 
-        updated_rows = await ScadaRepository.update_scada_field_batch(
-            timescale_conn,
-            cleaned_rows,
-            "cleaned_value",
-        )
-        if updated_rows == 0:
-            raise ValueError("SCADA 清洗结果未匹配任何已有监测数据")
+        expected_rows = len({(row[0], row[1]) for row in cleaned_rows})
+        async with timescale_conn.transaction():
+            updated_rows = await ScadaRepository.update_scada_field_batch(
+                timescale_conn,
+                cleaned_rows,
+                "cleaned_value",
+            )
+            if updated_rows != expected_rows:
+                raise ValueError(
+                    "SCADA 清洗目标在写入期间发生变化，"
+                    f"预期更新 {expected_rows} 行，实际更新 {updated_rows} 行"
+                )
 
         return "success"
 
     @staticmethod
     async def predict_pipeline_health(
         timescale_conn: AsyncConnection,
-        network_name: str,
+        postgres_conn: AsyncConnection,
         query_time: datetime,
     ) -> List[Dict[str, Any]]:
         """
@@ -478,7 +482,6 @@ class CompositeQueries:
 
         Args:
             timescale_conn: TimescaleDB 异步连接
-            db_name: 管网数据库名称
             query_time: 查询时间
             property_conditions: 可选的管道筛选条件，如 {"diameter": 300}
 
@@ -505,12 +508,21 @@ class CompositeQueries:
             # 3. 只查询有流速数据的管道的基本信息
             valid_link_ids = list(velocity_data.keys())
 
-            # 批量查询这些管道的详细信息
-            fields = ["id", "diameter", "node1", "node2"]
-            all_links = get_pipes_by_property(network_name, fields=fields)
+            # GIS 物化视图是低频更新管网的查询面；只读取本次有结果的管道。
+            async with postgres_conn.cursor() as cur:
+                await cur.execute(
+                    """
+                    SELECT id, diameter, start_node_id AS node1,
+                           end_node_id AS node2
+                    FROM gis.pipes
+                    WHERE id = ANY(%s)
+                    """,
+                    (valid_link_ids,),
+                )
+                all_links = await cur.fetchall()
 
             # 转换为字典以快速查找
-            links_dict = {link["id"]: link for link in all_links}
+            links_dict = {str(link["id"]): link for link in all_links}
 
             # 获取所有需要查询的节点ID
             node_ids = set()
diff --git a/app/infra/db/timescaledb/internal_queries.py b/app/infra/db/timescaledb/internal_queries.py
index a9ee386..318427b 100644
--- a/app/infra/db/timescaledb/internal_queries.py
+++ b/app/infra/db/timescaledb/internal_queries.py
@@ -88,16 +88,15 @@ class InternalQueries:
                     rows = ScadaRepository.get_scada_by_ids_time_range_sync(
                         conn, device_ids, start_time, end_time
                     )
-                    # 处理结果，返回每个 device_id 的第一个值
-                    result = {}
-                    for device_id in device_ids:
-                        device_rows = [
-                            row for row in rows if row["device_id"] == device_id
-                        ]
-                        if device_rows:
-                            result[device_id] = device_rows[0]["monitored_value"]
-                        else:
-                            result[device_id] = None
+                    # Rows are ordered by device/time; retain the first sample
+                    # for each requested device in one pass.
+                    result = {device_id: None for device_id in device_ids}
+                    seen: set[str] = set()
+                    for row in rows:
+                        device_id = str(row["device_id"])
+                        if device_id in result and device_id not in seen:
+                            result[device_id] = row["monitored_value"]
+                            seen.add(device_id)
                     return result
             except Exception as e:
                 logger.error(f"查询尝试 {attempt + 1} 失败: {e}")
@@ -135,8 +134,6 @@ class InternalQueries:
                         result.setdefault(device_id, []).append(
                             {"time": row["time"].isoformat(), "value": value}
                         )
-                    for device_id in result:
-                        result[device_id].sort(key=lambda item: item["time"])
                     return result
             except Exception as e:
                 logger.error(f"查询尝试 {attempt + 1} 失败: {e}")
@@ -307,16 +304,7 @@ class InternalQueries:
     def _resolve_simulation_table(element_type: str) -> tuple[str, str, set[str]]:
         normalized_type = element_type.lower()
         if normalized_type == "node":
-            return "node_results", "node_id", {"actual_demand", "total_head", "pressure", "quality"}
+            return "node_results", "node_id", set(RealtimeRepository.NODE_FIELDS)
         if normalized_type == "link":
-            return "link_results", "link_id", {
-                "flow",
-                "friction",
-                "headloss",
-                "quality",
-                "reaction",
-                "setting",
-                "status",
-                "velocity",
-            }
+            return "link_results", "link_id", set(RealtimeRepository.LINK_FIELDS)
         raise ValueError(f"Unsupported element_type: {element_type}")
diff --git a/app/infra/db/timescaledb/repositories/realtime.py b/app/infra/db/timescaledb/repositories/realtime.py
index 91c4928..3a80f96 100644
--- a/app/infra/db/timescaledb/repositories/realtime.py
+++ b/app/infra/db/timescaledb/repositories/realtime.py
@@ -6,6 +6,24 @@ from app.services.time_api import parse_utc_time
 
 
 class RealtimeRepository:
+    LINK_FIELDS = frozenset(
+        {
+            "flow",
+            "friction",
+            "headloss",
+            "quality",
+            "reaction",
+            "setting",
+            "status",
+            "velocity",
+        }
+    )
+    NODE_FIELDS = frozenset({"actual_demand", "total_head", "pressure", "quality"})
+    LINK_RESULT_COLUMNS = (
+        "time, link_id, flow, friction, headloss, quality, reaction, "
+        "setting, status, velocity"
+    )
+    NODE_RESULT_COLUMNS = "time, node_id, actual_demand, total_head, pressure, quality"
 
     @staticmethod
     def _batch_time(data: List[dict]) -> datetime:
@@ -22,6 +40,48 @@ class RealtimeRepository:
 
     # --- Link Simulation ---
 
+    @staticmethod
+    async def _copy_links(cur, data: List[dict], target_time: datetime) -> None:
+        async with cur.copy(
+            "COPY realtime.link_results (time, link_id, flow, friction, headloss, quality, reaction, setting, status, velocity) FROM STDIN"
+        ) as copy:
+            for item in data:
+                await copy.write_row(
+                    (
+                        target_time,
+                        item["id"],
+                        item.get("flow"),
+                        item.get("friction"),
+                        item.get("headloss"),
+                        item.get("quality"),
+                        item.get("reaction"),
+                        item.get("setting"),
+                        item.get("status"),
+                        item.get("velocity"),
+                    )
+                )
+
+    @staticmethod
+    def _copy_links_sync(cur, data: List[dict], target_time: datetime) -> None:
+        with cur.copy(
+            "COPY realtime.link_results (time, link_id, flow, friction, headloss, quality, reaction, setting, status, velocity) FROM STDIN"
+        ) as copy:
+            for item in data:
+                copy.write_row(
+                    (
+                        target_time,
+                        item["id"],
+                        item.get("flow"),
+                        item.get("friction"),
+                        item.get("headloss"),
+                        item.get("quality"),
+                        item.get("reaction"),
+                        item.get("setting"),
+                        item.get("status"),
+                        item.get("velocity"),
+                    )
+                )
+
     @staticmethod
     async def insert_links_batch(conn: AsyncConnection, data: List[dict]):
         """Batch insert for realtime.link_results using DELETE then COPY."""
@@ -43,25 +103,7 @@ class RealtimeRepository:
                     (target_time,),
                 )
 
-                # 2. 使用 COPY 快速写入新数据
-                async with cur.copy(
-                    "COPY realtime.link_results (time, link_id, flow, friction, headloss, quality, reaction, setting, status, velocity) FROM STDIN"
-                ) as copy:
-                    for item in data:
-                        await copy.write_row(
-                            (
-                                target_time,
-                                item["id"],
-                                item.get("flow"),
-                                item.get("friction"),
-                                item.get("headloss"),
-                                item.get("quality"),
-                                item.get("reaction"),
-                                item.get("setting"),
-                                item.get("status"),
-                                item.get("velocity"),
-                            )
-                        )
+                await RealtimeRepository._copy_links(cur, data, target_time)
 
     @staticmethod
     def insert_links_batch_sync(conn: Connection, data: List[dict]):
@@ -84,25 +126,7 @@ class RealtimeRepository:
                     (target_time,),
                 )
 
-                # 2. 使用 COPY 快速写入新数据
-                with cur.copy(
-                    "COPY realtime.link_results (time, link_id, flow, friction, headloss, quality, reaction, setting, status, velocity) FROM STDIN"
-                ) as copy:
-                    for item in data:
-                        copy.write_row(
-                            (
-                                target_time,
-                                item["id"],
-                                item.get("flow"),
-                                item.get("friction"),
-                                item.get("headloss"),
-                                item.get("quality"),
-                                item.get("reaction"),
-                                item.get("setting"),
-                                item.get("status"),
-                                item.get("velocity"),
-                            )
-                        )
+                RealtimeRepository._copy_links_sync(cur, data, target_time)
 
     @staticmethod
     async def get_link_by_time_range(
@@ -110,7 +134,8 @@ class RealtimeRepository:
     ) -> List[dict]:
         async with conn.cursor() as cur:
             await cur.execute(
-                "SELECT * FROM realtime.link_results WHERE time >= %s AND time <= %s "
+                f"SELECT {RealtimeRepository.LINK_RESULT_COLUMNS} "
+                "FROM realtime.link_results WHERE time >= %s AND time <= %s "
                 "AND link_id = %s ORDER BY time",
                 (start_time, end_time, link_id),
             )
@@ -124,46 +149,13 @@ class RealtimeRepository:
         normalized_end_time = parse_utc_time(end_time, field_name="end_time")
         async with conn.cursor() as cur:
             await cur.execute(
-                "SELECT * FROM realtime.link_results WHERE time >= %s AND time <= %s "
+                f"SELECT {RealtimeRepository.LINK_RESULT_COLUMNS} "
+                "FROM realtime.link_results WHERE time >= %s AND time <= %s "
                 "ORDER BY time, link_id",
                 (normalized_start_time, normalized_end_time),
             )
             return await cur.fetchall()
 
-    @staticmethod
-    async def get_link_field_by_time_range(
-        conn: AsyncConnection,
-        start_time: datetime,
-        end_time: datetime,
-        link_id: str,
-        field: str,
-    ) -> List[Dict[str, Any]]:
-        # Validate field name to prevent SQL injection
-        valid_fields = {
-            "flow",
-            "friction",
-            "headloss",
-            "quality",
-            "reaction",
-            "setting",
-            "status",
-            "velocity",
-        }
-        if field not in valid_fields:
-            raise ValueError(f"Invalid field: {field}")
-
-        query = sql.SQL(
-            "SELECT time, {} FROM realtime.link_results WHERE time >= %s "
-            "AND time <= %s AND link_id = %s ORDER BY time"
-        ).format(sql.Identifier(field))
-
-        async with conn.cursor() as cur:
-            await cur.execute(query, (start_time, end_time, link_id))
-            rows = await cur.fetchall()
-            return [
-                {"time": row["time"].isoformat(), "value": row[field]} for row in rows
-            ]
-
     @staticmethod
     async def get_link_fields_by_ids_time_range(
         conn: AsyncConnection,
@@ -172,11 +164,7 @@ class RealtimeRepository:
         link_ids: list[str],
         field: str,
     ) -> dict[str, list[dict[str, Any]]]:
-        valid_fields = {
-            "flow", "friction", "headloss", "quality", "reaction",
-            "setting", "status", "velocity",
-        }
-        if field not in valid_fields:
+        if field not in RealtimeRepository.LINK_FIELDS:
             raise ValueError(f"Invalid field: {field}")
         result = {link_id: [] for link_id in link_ids}
         if not link_ids:
@@ -202,17 +190,7 @@ class RealtimeRepository:
         field: str,
     ) -> dict:
         # Validate field name to prevent SQL injection
-        valid_fields = {
-            "flow",
-            "friction",
-            "headloss",
-            "quality",
-            "reaction",
-            "setting",
-            "status",
-            "velocity",
-        }
-        if field not in valid_fields:
+        if field not in RealtimeRepository.LINK_FIELDS:
             raise ValueError(f"Invalid field: {field}")
 
         query = sql.SQL(
@@ -238,17 +216,7 @@ class RealtimeRepository:
         field: str,
         value: Any,
     ):
-        valid_fields = {
-            "flow",
-            "friction",
-            "headloss",
-            "quality",
-            "reaction",
-            "setting",
-            "status",
-            "velocity",
-        }
-        if field not in valid_fields:
+        if field not in RealtimeRepository.LINK_FIELDS:
             raise ValueError(f"Invalid field: {field}")
 
         query = sql.SQL(
@@ -270,6 +238,40 @@ class RealtimeRepository:
 
     # --- Node Simulation ---
 
+    @staticmethod
+    async def _copy_nodes(cur, data: List[dict], target_time: datetime) -> None:
+        async with cur.copy(
+            "COPY realtime.node_results (time, node_id, actual_demand, total_head, pressure, quality) FROM STDIN"
+        ) as copy:
+            for item in data:
+                await copy.write_row(
+                    (
+                        target_time,
+                        item["id"],
+                        item.get("actual_demand"),
+                        item.get("total_head"),
+                        item.get("pressure"),
+                        item.get("quality"),
+                    )
+                )
+
+    @staticmethod
+    def _copy_nodes_sync(cur, data: List[dict], target_time: datetime) -> None:
+        with cur.copy(
+            "COPY realtime.node_results (time, node_id, actual_demand, total_head, pressure, quality) FROM STDIN"
+        ) as copy:
+            for item in data:
+                copy.write_row(
+                    (
+                        target_time,
+                        item["id"],
+                        item.get("actual_demand"),
+                        item.get("total_head"),
+                        item.get("pressure"),
+                        item.get("quality"),
+                    )
+                )
+
     @staticmethod
     async def insert_nodes_batch(conn: AsyncConnection, data: List[dict]):
         if not data:
@@ -290,21 +292,7 @@ class RealtimeRepository:
                     (target_time,),
                 )
 
-                # 2. 使用 COPY 快速写入新数据
-                async with cur.copy(
-                    "COPY realtime.node_results (time, node_id, actual_demand, total_head, pressure, quality) FROM STDIN"
-                ) as copy:
-                    for item in data:
-                        await copy.write_row(
-                            (
-                                target_time,
-                                item["id"],
-                                item.get("actual_demand"),
-                                item.get("total_head"),
-                                item.get("pressure"),
-                                item.get("quality"),
-                            )
-                        )
+                await RealtimeRepository._copy_nodes(cur, data, target_time)
 
     @staticmethod
     def insert_nodes_batch_sync(conn: Connection, data: List[dict]):
@@ -326,21 +314,7 @@ class RealtimeRepository:
                     (target_time,),
                 )
 
-                # 2. 使用 COPY 快速写入新数据
-                with cur.copy(
-                    "COPY realtime.node_results (time, node_id, actual_demand, total_head, pressure, quality) FROM STDIN"
-                ) as copy:
-                    for item in data:
-                        copy.write_row(
-                            (
-                                target_time,
-                                item["id"],
-                                item.get("actual_demand"),
-                                item.get("total_head"),
-                                item.get("pressure"),
-                                item.get("quality"),
-                            )
-                        )
+                RealtimeRepository._copy_nodes_sync(cur, data, target_time)
 
     @staticmethod
     async def get_node_by_time_range(
@@ -348,7 +322,8 @@ class RealtimeRepository:
     ) -> List[dict]:
         async with conn.cursor() as cur:
             await cur.execute(
-                "SELECT * FROM realtime.node_results WHERE time >= %s AND time <= %s "
+                f"SELECT {RealtimeRepository.NODE_RESULT_COLUMNS} "
+                "FROM realtime.node_results WHERE time >= %s AND time <= %s "
                 "AND node_id = %s ORDER BY time",
                 (start_time, end_time, node_id),
             )
@@ -362,36 +337,13 @@ class RealtimeRepository:
         normalized_end_time = parse_utc_time(end_time, field_name="end_time")
         async with conn.cursor() as cur:
             await cur.execute(
-                "SELECT * FROM realtime.node_results WHERE time >= %s AND time <= %s "
+                f"SELECT {RealtimeRepository.NODE_RESULT_COLUMNS} "
+                "FROM realtime.node_results WHERE time >= %s AND time <= %s "
                 "ORDER BY time, node_id",
                 (normalized_start_time, normalized_end_time),
             )
             return await cur.fetchall()
 
-    @staticmethod
-    async def get_node_field_by_time_range(
-        conn: AsyncConnection,
-        start_time: datetime,
-        end_time: datetime,
-        node_id: str,
-        field: str,
-    ) -> List[Dict[str, Any]]:
-        valid_fields = {"actual_demand", "total_head", "pressure", "quality"}
-        if field not in valid_fields:
-            raise ValueError(f"Invalid field: {field}")
-
-        query = sql.SQL(
-            "SELECT time, {} FROM realtime.node_results WHERE time >= %s "
-            "AND time <= %s AND node_id = %s ORDER BY time"
-        ).format(sql.Identifier(field))
-
-        async with conn.cursor() as cur:
-            await cur.execute(query, (start_time, end_time, node_id))
-            rows = await cur.fetchall()
-            return [
-                {"time": row["time"].isoformat(), "value": row[field]} for row in rows
-            ]
-
     @staticmethod
     async def get_node_fields_by_ids_time_range(
         conn: AsyncConnection,
@@ -400,8 +352,7 @@ class RealtimeRepository:
         node_ids: list[str],
         field: str,
     ) -> dict[str, list[dict[str, Any]]]:
-        valid_fields = {"actual_demand", "total_head", "pressure", "quality"}
-        if field not in valid_fields:
+        if field not in RealtimeRepository.NODE_FIELDS:
             raise ValueError(f"Invalid field: {field}")
         result = {node_id: [] for node_id in node_ids}
         if not node_ids:
@@ -423,8 +374,7 @@ class RealtimeRepository:
     async def get_nodes_field_by_time_range(
         conn: AsyncConnection, start_time: datetime, end_time: datetime, field: str
     ) -> dict:
-        valid_fields = {"actual_demand", "total_head", "pressure", "quality"}
-        if field not in valid_fields:
+        if field not in RealtimeRepository.NODE_FIELDS:
             raise ValueError(f"Invalid field: {field}")
 
         query = sql.SQL(
@@ -450,8 +400,7 @@ class RealtimeRepository:
         field: str,
         value: Any,
     ):
-        valid_fields = {"actual_demand", "total_head", "pressure", "quality"}
-        if field not in valid_fields:
+        if field not in RealtimeRepository.NODE_FIELDS:
             raise ValueError(f"Invalid field: {field}")
 
         query = sql.SQL(
@@ -529,9 +478,8 @@ class RealtimeRepository:
                 }
             )
 
-        # Keep node and link replacement atomic. The batch helpers use nested
-        # transactions (savepoints), while this outer transaction guarantees
-        # that a link write failure also rolls back the node replacement.
+        # Keep node and link replacement atomic with one lock and one delete per
+        # table. Copy-only helpers avoid repeating replacement SQL.
         async with conn.transaction():
             async with conn.cursor() as cur:
                 await cur.execute(
@@ -546,11 +494,10 @@ class RealtimeRepository:
                     "DELETE FROM realtime.link_results WHERE time = %s",
                     (simulation_time,),
                 )
-            if node_data:
-                await RealtimeRepository.insert_nodes_batch(conn, node_data)
-
-            if link_data:
-                await RealtimeRepository.insert_links_batch(conn, link_data)
+                if node_data:
+                    await RealtimeRepository._copy_nodes(cur, node_data, simulation_time)
+                if link_data:
+                    await RealtimeRepository._copy_links(cur, link_data, simulation_time)
 
     @staticmethod
     def store_realtime_simulation_result_sync(
@@ -608,9 +555,8 @@ class RealtimeRepository:
                 }
             )
 
-        # Keep node and link replacement atomic. The batch helpers use nested
-        # transactions (savepoints), while this outer transaction guarantees
-        # that a link write failure also rolls back the node replacement.
+        # Keep node and link replacement atomic with one lock and one delete per
+        # table. Copy-only helpers avoid repeating replacement SQL.
         with conn.transaction():
             with conn.cursor() as cur:
                 cur.execute(
@@ -625,11 +571,10 @@ class RealtimeRepository:
                     "DELETE FROM realtime.link_results WHERE time = %s",
                     (simulation_time,),
                 )
-            if node_data:
-                RealtimeRepository.insert_nodes_batch_sync(conn, node_data)
-
-            if link_data:
-                RealtimeRepository.insert_links_batch_sync(conn, link_data)
+                if node_data:
+                    RealtimeRepository._copy_nodes_sync(cur, node_data, simulation_time)
+                if link_data:
+                    RealtimeRepository._copy_links_sync(cur, link_data, simulation_time)
 
     @staticmethod
     async def query_all_record_by_time_property(
diff --git a/app/infra/db/timescaledb/repositories/scada.py b/app/infra/db/timescaledb/repositories/scada.py
index 0c19654..4679678 100644
--- a/app/infra/db/timescaledb/repositories/scada.py
+++ b/app/infra/db/timescaledb/repositories/scada.py
@@ -6,6 +6,8 @@ from psycopg.rows import dict_row
 
 
 class ScadaRepository:
+    VALUE_FIELDS = frozenset({"monitored_value", "cleaned_value"})
+    RESULT_COLUMNS = "time, device_id, monitored_value, cleaned_value"
 
     @staticmethod
     async def insert_scada_batch(conn: AsyncConnection, data: List[dict]):
@@ -35,7 +37,8 @@ class ScadaRepository:
     ) -> List[dict]:
         async with conn.cursor() as cur:
             await cur.execute(
-                "SELECT * FROM scada.measurements WHERE device_id = ANY(%s) "
+                f"SELECT {ScadaRepository.RESULT_COLUMNS} FROM scada.measurements "
+                "WHERE device_id = ANY(%s) "
                 "AND time >= %s AND time <= %s ORDER BY device_id, time",
                 (device_ids, start_time, end_time),
             )
@@ -50,7 +53,8 @@ class ScadaRepository:
     ) -> List[dict]:
         with conn.cursor(row_factory=dict_row) as cur:
             cur.execute(
-                "SELECT * FROM scada.measurements WHERE device_id = ANY(%s) "
+                f"SELECT {ScadaRepository.RESULT_COLUMNS} FROM scada.measurements "
+                "WHERE device_id = ANY(%s) "
                 "AND time >= %s AND time <= %s ORDER BY device_id, time",
                 (device_ids, start_time, end_time),
             )
@@ -85,8 +89,7 @@ class ScadaRepository:
         end_time: datetime,
         field: str,
     ) -> dict:
-        valid_fields = {"monitored_value", "cleaned_value"}
-        if field not in valid_fields:
+        if field not in ScadaRepository.VALUE_FIELDS:
             raise ValueError(f"Invalid field: {field}")
 
         query = sql.SQL(
@@ -110,8 +113,7 @@ class ScadaRepository:
     async def update_scada_field(
         conn: AsyncConnection, time: datetime, device_id: str, field: str, value: Any
     ):
-        valid_fields = {"monitored_value", "cleaned_value"}
-        if field not in valid_fields:
+        if field not in ScadaRepository.VALUE_FIELDS:
             raise ValueError(f"Invalid field: {field}")
 
         update_query = sql.SQL(
@@ -133,8 +135,7 @@ class ScadaRepository:
         field: str,
     ) -> int:
         """Update existing SCADA samples in one set-based statement."""
-        valid_fields = {"monitored_value", "cleaned_value"}
-        if field not in valid_fields:
+        if field not in ScadaRepository.VALUE_FIELDS:
             raise ValueError(f"Invalid field: {field}")
         if not rows:
             return 0
diff --git a/app/services/burst_detection.py b/app/services/burst_detection.py
index 9f45acb..2a21d5d 100644
--- a/app/services/burst_detection.py
+++ b/app/services/burst_detection.py
@@ -11,13 +11,10 @@ import pandas as pd
 from app.algorithms.burst_detection.burst_detector import BurstDetector
 from app.infra.db.timescaledb.internal_queries import InternalQueries
 from app.services.scheme_management import (
-    query_burst_detection_scheme_detail,
-    query_burst_detection_schemes,
-    scheme_name_exists,
     store_scheme_info,
 )
 from app.services.tjnetwork import get_all_scada_info
-from app.services.time_api import extract_date, parse_utc_time, utc_now
+from app.services.time_api import parse_utc_time, utc_now
 
 
 TARGET_DAY_COUNT = 15
@@ -365,25 +362,6 @@ def _build_observed_pressure_from_simulation(
     return observation_df
 
 
-def list_burst_detection_schemes(
-    network: str,
-    query_date: datetime | str | None = None,
-) -> list[dict[str, Any]]:
-    parsed_date = extract_date(query_date, field_name="query_date") if query_date is not None else None
-    return query_burst_detection_schemes(
-        name=network,
-        network=network,
-        query_date=parsed_date,
-    )
-
-
-def get_burst_detection_scheme_detail(network: str, scheme_name: str) -> dict[str, Any]:
-    result = query_burst_detection_scheme_detail(network, scheme_name)
-    if not result:
-        raise ValueError(f"未找到爆管侦测方案: {scheme_name}")
-    return result
-
-
 def _store_burst_detection_scheme(
     *,
     network: str,
@@ -394,9 +372,6 @@ def _store_burst_detection_scheme(
     points_per_day: int,
     iforest_params: dict[str, Any],
 ) -> None:
-    if scheme_name_exists(network, scheme_name):
-        raise ValueError(f"方案名称已存在: {scheme_name}")
-
     now_iso = utc_now().isoformat()
     scheme_detail = {
         "network": network,
diff --git a/app/services/burst_location.py b/app/services/burst_location.py
index 517b95d..1fe556b 100644
--- a/app/services/burst_location.py
+++ b/app/services/burst_location.py
@@ -10,14 +10,11 @@ import pandas as pd
 from app.algorithms.burst_location import run_burst_location
 from app.infra.db.timescaledb.internal_queries import InternalQueries
 from app.services.scheme_management import (
-    query_burst_location_scheme_detail,
-    query_burst_location_schemes,
     get_analysis_run,
-    scheme_name_exists,
     store_scheme_info,
 )
 from app.services.tjnetwork import dump_inp, get_all_scada_info
-from app.services.time_api import extract_date, parse_utc_time, utc_now
+from app.services.time_api import parse_utc_time, utc_now
 
 SeriesInput = pd.Series | dict[str, Any] | list[dict[str, Any]]
 FLOW_SCADA_TYPES = {"pipe_flow", "flow", "demand"}
@@ -367,22 +364,6 @@ def run_burst_location_by_network(
     return payload
 
 
-def list_burst_location_schemes(
-    network: str, query_date: datetime | str | None = None
-) -> list[dict[str, Any]]:
-    parsed_date = extract_date(query_date, field_name="query_date") if query_date is not None else None
-    return query_burst_location_schemes(
-        name=network, network=network, query_date=parsed_date
-    )
-
-
-def get_burst_location_scheme_detail(network: str, scheme_name: str) -> dict[str, Any]:
-    result = query_burst_location_scheme_detail(network, scheme_name)
-    if not result:
-        raise ValueError(f"未找到爆管定位方案: {scheme_name}")
-    return result
-
-
 def _store_burst_scheme(
     *,
     network: str,
@@ -393,9 +374,6 @@ def _store_burst_scheme(
     min_dpressure: float,
     basic_pressure: float,
 ) -> None:
-    if scheme_name_exists(network, scheme_name):
-        raise ValueError(f"方案名称已存在: {scheme_name}")
-
     now_iso = utc_now().isoformat()
     scheme_detail = {
         "network": network,
diff --git a/app/services/globals.py b/app/services/globals.py
deleted file mode 100644
index 9df7d0a..0000000
--- a/app/services/globals.py
+++ /dev/null
@@ -1,14 +0,0 @@
-"""Mutable state used by the legacy synchronous simulation runner."""
-
-RESERVOIR_BASIC_HEIGHT = 250.35
-PATTERN_TIME_STEP: float | None = None
-hydraulic_timestep: str | None = None
-
-# Element ID -> SCADA api_query_id, loaded per project before simulation.
-reservoirs_id: dict[str, str] = {}
-tanks_id: dict[str, str] = {}
-fixed_pumps_id: dict[str, str] = {}
-variable_pumps_id: dict[str, str] = {}
-pressure_id: dict[str, str] = {}
-demand_id: dict[str, str] = {}
-quality_id: dict[str, str] = {}
diff --git a/app/services/leakage_identifier.py b/app/services/leakage_identifier.py
index beacc04..f5c374f 100644
--- a/app/services/leakage_identifier.py
+++ b/app/services/leakage_identifier.py
@@ -10,20 +10,14 @@ import wntr
 
 from app.algorithms.leakage.identifier import LeakageIdentifier
 from app.infra.db.timescaledb.internal_queries import InternalQueries
-from app.services.scheme_management import (
-    query_leakage_identify_scheme_detail,
-    query_leakage_identify_schemes,
-    scheme_name_exists,
-    store_leakage_identify_result,
-    store_scheme_info,
-)
+from app.services.scheme_management import store_analysis_run_with_result
 from app.services.tjnetwork import (
     dump_inp,
     get_all_scada_info,
     get_network_link_nodes,
     get_network_node_coords,
 )
-from app.services.time_api import extract_date, parse_utc_time, utc_now
+from app.services.time_api import parse_utc_time, utc_now
 
 DEFAULT_N_WORKERS = max(1, min((os.cpu_count() or 1) - 1, 4))
 
@@ -115,8 +109,6 @@ def run_leakage_identification(
         "rows": rows,
     }
     if scheme_name:
-        if scheme_name_exists(network, scheme_name):
-            raise ValueError(f"方案名称已存在: {scheme_name}")
         scheme_start_time = (
             _to_datetime(scada_start).isoformat()
             if scada_start is not None
@@ -153,46 +145,29 @@ def run_leakage_identification(
                 ),
             },
         }
-        store_scheme_info(
+        store_analysis_run_with_result(
             name=network,
             scheme_name=scheme_name,
             scheme_type="dma_leak_identification",
             username=username,
             scheme_start_time=scheme_start_time,
             scheme_detail=scheme_detail,
-        )
-        store_leakage_identify_result(
-            name=network,
-            scheme_name=scheme_name,
-            network=network,
-            sensor_nodes=selected_sensor_nodes,
-            result_rows=rows,
-            node_area_map=area_map,
-            areas=areas,
-            drawing_payload={},
+            result_type="leakage_identification",
+            result_payload={
+                "network": network,
+                "run_status": "completed",
+                "error_message": None,
+                "sensor_nodes": selected_sensor_nodes,
+                "rows": rows,
+                "node_area_map": area_map,
+                "areas": areas,
+                "drawing_payload": {},
+            },
         )
         payload["scheme_name"] = scheme_name
     return payload
 
 
-def list_leakage_identify_schemes(
-    network: str, query_date: datetime | str | None = None
-) -> list[dict[str, Any]]:
-    parsed_date = extract_date(query_date, field_name="query_date") if query_date is not None else None
-    return query_leakage_identify_schemes(
-        name=network, network=network, query_date=parsed_date
-    )
-
-
-def get_leakage_identify_scheme_detail(
-    network: str, scheme_name: str
-) -> dict[str, Any]:
-    result = query_leakage_identify_scheme_detail(network, scheme_name)
-    if not result:
-        raise ValueError(f"未找到漏损识别方案: {scheme_name}")
-    return result
-
-
 def _get_pressure_sensor_nodes(network: str) -> list[str]:
     scada_devices = get_all_scada_info(network)
     sensor_nodes: list[str] = []
diff --git a/app/services/scheme_management.py b/app/services/scheme_management.py
index 516c5f1..f02daa3 100644
--- a/app/services/scheme_management.py
+++ b/app/services/scheme_management.py
@@ -1,32 +1,21 @@
-from datetime import date, datetime
+from datetime import datetime
 from typing import Any
 from uuid import UUID, uuid4
 
-from psycopg.types.json import Jsonb
-
-from app.native.wndb.core.connection import project_connection
+from app.infra.db.postgresql.analysis import AnalysisRepository
+from app.native.wndb.core.connection import project_connection, project_transaction
 from app.services.time_api import parse_utc_time
 
 
-def scheme_name_exists(name: str, scheme_name: str) -> bool:
-    with project_connection(name) as conn, conn.cursor() as cur:
-        cur.execute(
-            "select exists(select 1 from analysis.runs where name = %s)",
-            (scheme_name,),
-        )
-        row = cur.fetchone()
-    return bool(row and row[0])
-
-
 def store_scheme_info(
     name: str,
     scheme_name: str,
     scheme_type: str,
     username: str,
     scheme_start_time: datetime | str,
-    scheme_detail: dict,
+    scheme_detail: dict[str, Any],
 ) -> UUID:
-    """Create one completed, immutable analysis run."""
+    """Create one completed analysis run; its name remains a display label."""
     return create_analysis_run(
         name=name,
         scheme_name=scheme_name,
@@ -44,29 +33,56 @@ def create_analysis_run(
     scheme_type: str,
     username: str,
     scheme_start_time: datetime | str,
-    scheme_detail: dict,
+    scheme_detail: dict[str, Any],
     *,
     status: str = "running",
 ) -> UUID:
-    """Create a distinct execution record; names are labels, not identities."""
     started_at = parse_utc_time(scheme_start_time, field_name="scheme_start_time")
     run_id = uuid4()
-    with project_connection(name) as conn, conn.cursor() as cur:
-        cur.execute(
-            """
-            insert into analysis.runs
-                (run_id, name, run_type, created_by, created_at, started_at, status, parameters)
-            values (%s, %s, %s, %s, now(), %s, %s, %s)
-            """,
-            (
-                run_id,
-                scheme_name,
-                scheme_type,
-                username,
-                started_at,
-                status,
-                Jsonb(scheme_detail),
-            ),
+    with project_connection(name) as conn:
+        AnalysisRepository.create_run_sync(
+            conn,
+            run_id=run_id,
+            name=scheme_name,
+            run_type=scheme_type,
+            created_by=username,
+            started_at=started_at,
+            status=status,
+            parameters=scheme_detail,
+        )
+    return run_id
+
+
+def store_analysis_run_with_result(
+    *,
+    name: str,
+    scheme_name: str,
+    scheme_type: str,
+    username: str,
+    scheme_start_time: datetime | str,
+    scheme_detail: dict[str, Any],
+    result_type: str,
+    result_payload: dict[str, Any],
+) -> UUID:
+    """Atomically persist one BizDB run and its non-timeseries result."""
+    started_at = parse_utc_time(scheme_start_time, field_name="scheme_start_time")
+    run_id = uuid4()
+    with project_transaction(name) as conn:
+        AnalysisRepository.create_run_sync(
+            conn,
+            run_id=run_id,
+            name=scheme_name,
+            run_type=scheme_type,
+            created_by=username,
+            started_at=started_at,
+            status="completed",
+            parameters=scheme_detail,
+        )
+        AnalysisRepository.insert_result_sync(
+            conn,
+            run_id,
+            result_type=result_type,
+            payload=result_payload,
         )
     return run_id
 
@@ -77,208 +93,24 @@ def update_analysis_run(
     *,
     status: str,
     username: str,
-    scheme_detail: dict,
+    scheme_detail: dict[str, Any],
 ) -> None:
-    """Update lifecycle state and metadata for one execution identity."""
-    with project_connection(name) as conn, conn.cursor() as cur:
-        cur.execute(
-            """
-            update analysis.runs
-            set created_by = %s, status = %s, parameters = %s
-            where run_id = %s
-            """,
-            (username, status, Jsonb(scheme_detail), run_id),
+    with project_connection(name) as conn:
+        AnalysisRepository.update_run_sync(
+            conn,
+            run_id,
+            status=status,
+            created_by=username,
+            parameters=scheme_detail,
         )
-        if cur.rowcount != 1:
-            raise LookupError(f"analysis run {run_id} does not exist")
-
-
-def _run_row(row: dict[str, Any]) -> dict[str, Any]:
-    parameters = row.get("parameters") if isinstance(row.get("parameters"), dict) else {}
-    return {
-        "run_id": row["run_id"],
-        "name": row["name"],
-        "run_type": row["run_type"],
-        "created_by": row["created_by"],
-        "created_at": row["created_at"],
-        "started_at": row["started_at"],
-        "status": row["status"],
-        "parameters": parameters,
-    }
-
-
-def _list_runs(
-    name: str,
-    run_type: str | None = None,
-    query_date: date | None = None,
-) -> list[dict[str, Any]]:
-    clauses: list[str] = []
-    params: list[Any] = []
-    if run_type:
-        clauses.append("run_type = %s")
-        params.append(run_type)
-    if query_date is not None:
-        clauses.append("created_at::date = %s")
-        params.append(query_date)
-    where = f"where {' and '.join(clauses)}" if clauses else ""
-    with project_connection(name) as conn, conn.cursor() as cur:
-        cur.execute(
-            f"select run_id, name, run_type, created_by, created_at, started_at, status, parameters from analysis.runs {where} order by created_at desc",
-            params,
-        )
-        return [_run_row(row) for row in cur.fetchall()]
-
-
-def query_scheme_list(
-    name: str,
-    scheme_type: str | None = None,
-    query_date: date | None = None,
-) -> list[dict[str, Any]]:
-    return _list_runs(name, scheme_type, query_date)
-
-
-def _get_run_by_name(
-    name: str,
-    run_name: str,
-    run_type: str | None = None,
-) -> dict[str, Any]:
-    params: list[Any] = [run_name]
-    type_clause = ""
-    if run_type:
-        type_clause = "and run_type = %s"
-        params.append(run_type)
-    with project_connection(name) as conn, conn.cursor() as cur:
-        cur.execute(
-            f"""
-            select run_id, name, run_type, created_by, created_at, started_at,
-                   status, parameters
-            from analysis.runs
-            where name = %s {type_clause}
-            order by created_at desc
-            limit 1
-            """,
-            params,
-        )
-        row = cur.fetchone()
-    return _run_row(row) if row else {}
 
 
 def get_analysis_run(name: str, run_id: UUID) -> dict[str, Any]:
-    with project_connection(name) as conn, conn.cursor() as cur:
-        cur.execute(
-            """
-            select run_id, name, run_type, created_by, created_at, started_at,
-                   status, parameters
-            from analysis.runs
-            where run_id = %s
-            """,
-            (run_id,),
-        )
-        row = cur.fetchone()
-    return _run_row(row) if row else {}
-
-
-def query_scheme_detail(
-    name: str,
-    scheme_name: str,
-    scheme_type: str | None = None,
-) -> dict[str, Any]:
-    return _get_run_by_name(name, scheme_name, scheme_type)
-
-
-def store_leakage_identify_result(
-    name: str,
-    scheme_name: str,
-    network: str,
-    sensor_nodes: list[str],
-    result_rows: list[dict],
-    node_area_map: dict[str, str],
-    areas: list[dict],
-    drawing_payload: dict | None = None,
-    run_status: str = "completed",
-    error_message: str | None = None,
-) -> None:
-    run = _get_run_by_name(name, scheme_name, "dma_leak_identification")
-    if not run:
-        raise LookupError(f"analysis run {scheme_name!r} does not exist")
-    payload = {
-        "network": network,
-        "run_status": run_status,
-        "error_message": error_message,
-        "sensor_nodes": sensor_nodes,
-        "rows": result_rows,
-        "node_area_map": node_area_map,
-        "areas": areas,
-        "drawing_payload": drawing_payload or {},
-    }
-    with project_connection(name) as conn, conn.cursor() as cur:
-        cur.execute(
-            "insert into analysis.results (run_id, result_type, payload) values (%s, 'leakage_identification', %s)",
-            (run["run_id"], Jsonb(payload)),
-        )
-
-
-def _list_typed_runs(
-    name: str,
-    network: str,
-    run_type: str,
-    query_date: date | None,
-) -> list[dict[str, Any]]:
-    rows = _list_runs(name, run_type, query_date)
-    return [
-        row
-        for row in rows
-        if not network or row["parameters"].get("network") in (None, network)
-    ]
-
-
-def _typed_run_detail(name: str, run_name: str, run_type: str) -> dict[str, Any]:
-    run = _get_run_by_name(name, run_name, run_type)
-    if not run:
+    with project_connection(name) as conn:
+        row = AnalysisRepository.get_run_sync(conn, run_id)
+    if row is None:
         return {}
-    with project_connection(name) as conn, conn.cursor() as cur:
-        cur.execute(
-            "select result_type, payload, created_at from analysis.results where run_id = %s order by created_at, result_id",
-            (run["run_id"],),
-        )
-        results = [dict(row) for row in cur.fetchall()]
-    return run | {"results": results}
-
-
-def query_leakage_identify_schemes(
-    name: str,
-    network: str,
-    scheme_type: str = "dma_leak_identification",
-    query_date: date | None = None,
-) -> list[dict[str, Any]]:
-    return _list_typed_runs(name, network, scheme_type, query_date)
-
-
-def query_leakage_identify_scheme_detail(name: str, scheme_name: str) -> dict[str, Any]:
-    return _typed_run_detail(name, scheme_name, "dma_leak_identification")
-
-
-def query_burst_location_schemes(
-    name: str,
-    network: str,
-    scheme_type: str = "burst_location",
-    query_date: date | None = None,
-) -> list[dict[str, Any]]:
-    return _list_typed_runs(name, network, scheme_type, query_date)
-
-
-def query_burst_location_scheme_detail(name: str, scheme_name: str) -> dict[str, Any]:
-    return _typed_run_detail(name, scheme_name, "burst_location")
-
-
-def query_burst_detection_schemes(
-    name: str,
-    network: str,
-    scheme_type: str = "burst_detection",
-    query_date: date | None = None,
-) -> list[dict[str, Any]]:
-    return _list_typed_runs(name, network, scheme_type, query_date)
-
-
-def query_burst_detection_scheme_detail(name: str, scheme_name: str) -> dict[str, Any]:
-    return _typed_run_detail(name, scheme_name, "burst_detection")
+    parameters = row.get("parameters")
+    return dict(row) | {
+        "parameters": parameters if isinstance(parameters, dict) else {}
+    }
diff --git a/app/services/sensor_placement.py b/app/services/sensor_placement.py
index b1dddc7..90291c0 100644
--- a/app/services/sensor_placement.py
+++ b/app/services/sensor_placement.py
@@ -61,11 +61,15 @@ def _normalize_locations(sensor_location: list[str]) -> list[str]:
 def _sensor_points(
     network: str,
     sensor_location: list[str],
+    *,
+    nodes_by_id: dict[str, dict[str, Any]] | None = None,
 ) -> list[dict[str, Any]]:
-    nodes = sensor_placement_repository.get_sensor_placement_nodes(
-        network, sensor_location
-    )
-    by_id = {str(node["node_id"]): node for node in nodes}
+    if nodes_by_id is None:
+        nodes = sensor_placement_repository.get_sensor_placement_nodes(
+            network, sensor_location
+        )
+        nodes_by_id = {str(node["node_id"]): node for node in nodes}
+    by_id = nodes_by_id
     missing = [node_id for node_id in sensor_location if node_id not in by_id]
     if missing:
         raise SensorPlacementValidationError(
@@ -131,12 +135,26 @@ def get_sensor_placement_run(network: str, run_id: UUID) -> dict[str, Any]:
 
 
 def list_sensor_placement_runs(network: str) -> list[dict[str, Any]]:
+    runs = sensor_placement_repository.get_all_sensor_placements(network)
+    node_ids = list(
+        dict.fromkeys(
+            node_id
+            for run in runs
+            for node_id in run["sensor_locations"]
+        )
+    )
+    nodes = sensor_placement_repository.get_sensor_placement_nodes(network, node_ids)
+    nodes_by_id = {str(node["node_id"]): node for node in nodes}
     return [
         {
             **run,
-            "sensor_points": _sensor_points(network, run["sensor_locations"]),
+            "sensor_points": _sensor_points(
+                network,
+                run["sensor_locations"],
+                nodes_by_id=nodes_by_id,
+            ),
         }
-        for run in sensor_placement_repository.get_all_sensor_placements(network)
+        for run in runs
     ]
 
 
@@ -161,7 +179,10 @@ def update_sensor_placement_run(
         if sensor_placement_repository.get_sensor_placement(network, run_id) is None:
             raise SensorPlacementNotFoundError("监测点优化运行不存在")
         raise SensorPlacementConflictError("运行结果已被其他用户修改，请重新加载")
-    return get_sensor_placement_run(network, run_id)
+    return {
+        **updated,
+        "sensor_points": _sensor_points(network, updated["sensor_locations"]),
+    }
 
 
 def can_edit_sensor_placement(user: Any, run: dict[str, Any]) -> bool:
diff --git a/app/services/simulation.py b/app/services/simulation.py
index b92baaa..5a96d51 100644
--- a/app/services/simulation.py
+++ b/app/services/simulation.py
@@ -30,10 +30,13 @@ from typing import Optional, Tuple
 from uuid import UUID
 import typing
 import logging
-import app.services.globals as globals
 import app.services.project_info as project_info
+from app.infra.db.postgresql.scada import (
+    ScadaElementMappings,
+    load_realtime_element_mappings,
+)
 from app.services.time_api import parse_beijing_time, parse_clock_duration_seconds
-from app.native.wndb.core.connection import project_connection, project_transaction
+from app.native.wndb.core.connection import project_transaction
 from app.native.wndb.core.database import refresh_materialized_views_after_commit
 from app.infra.db.timescaledb.internal_queries import (
     InternalQueries as TimescaleInternalQueries,
@@ -47,6 +50,8 @@ logging.basicConfig(
     level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s"
 )
 
+RESERVOIR_BASIC_HEIGHT = 250.35
+
 
 def _primary_demand(demand_set: dict) -> dict:
     """Return sequence-zero demand, creating it when a junction has none."""
@@ -73,39 +78,12 @@ def _primary_demand_pattern(demand_set: dict) -> str:
     return str(pattern)
 
 
-def query_corresponding_element_id_and_query_id(name: str) -> None:
-    """Load realtime device-to-element mappings from the new asset schema."""
-    target_maps = {
-        "reservoir_liquid_level": globals.reservoirs_id,
-        "tank_liquid_level": globals.tanks_id,
-        "fixed_pump": globals.fixed_pumps_id,
-        "variable_pump": globals.variable_pumps_id,
-        "pressure": globals.pressure_id,
-        "demand": globals.demand_id,
-        "quality": globals.quality_id,
-    }
-    for mapping in target_maps.values():
-        mapping.clear()
-    with project_connection(name) as conn, conn.cursor() as cur:
-        cur.execute(
-            """
-            SELECT device_type, COALESCE(node_id, link_id) AS element_id,
-                   api_query_id
-            FROM asset.scada_devices
-            WHERE transmission_mode = 'realtime'
-              AND api_query_id IS NOT NULL
-            """
-        )
-        for record in cur.fetchall():
-            device_type = record["device_type"]
-            element_id = record["element_id"]
-            api_query_id = record["api_query_id"]
-            mapping = target_maps.get(str(device_type).lower())
-            if mapping is not None:
-                mapping[str(element_id)] = str(api_query_id)
+def query_corresponding_element_id_and_query_id(name: str) -> ScadaElementMappings:
+    """Return an immutable project-local SCADA-to-model mapping snapshot."""
+    return load_realtime_element_mappings(name)
 
 
-def get_pattern_index(cur_datetime: str) -> int:
+def get_pattern_index(cur_datetime: str, pattern_time_step: float) -> int:
     """
     根据给定的日期时间字符串，计算并返回对应的模式索引。
     :param cur_datetime: str, 当前的日期时间字符串，格式为“YYYY-MM-DD HH:MM:SS”。
@@ -115,18 +93,18 @@ def get_pattern_index(cur_datetime: str) -> int:
     dt = datetime.strptime(cur_datetime, str_format)
     hr = dt.hour
     mnt = dt.minute
-    i = int((hr * 60 + mnt) / globals.PATTERN_TIME_STEP)
+    i = int((hr * 60 + mnt) / pattern_time_step)
     return i
 
 
-def get_pattern_index_str(current_time: str) -> str:
+def get_pattern_index_str(current_time: str, pattern_time_step: float) -> str:
     """
     根据当前时间获取时间步长的模式索引，并将其格式化为“HH:MM:00”字符串。
     :param current_time: str, 当前时间，格式为"YYYY-MM-DD HH:MM:SS"
     :return: str， 以“HH:MM:00”格式返回
     """
-    i = get_pattern_index(current_time)
-    [minN, hrN] = modf(i * globals.PATTERN_TIME_STEP / 60)
+    i = get_pattern_index(current_time, pattern_time_step)
+    [minN, hrN] = modf(i * pattern_time_step / 60)
     minN_str = str(int(minN * 60))
     minN_str = minN_str.zfill(2)
     hrN_str = str(int(hrN))
@@ -204,6 +182,7 @@ def run_simulation(
     valve_control: dict[str, dict] = None,
     scheme_username: str = "system",
     scheme_detail: dict | None = None,
+    scada_mappings: ScadaElementMappings | None = None,
 ) -> UUID | None:
     """
     传入需要修改的参数，改变数据库中对应位置的值，然后计算，返回结果
@@ -257,19 +236,20 @@ def run_simulation(
         print(dic_time)
 
         # 获取水力模拟步长，如’0:15:00‘
-        globals.hydraulic_timestep = dic_time["HYDRAULIC TIMESTEP"]
+        hydraulic_timestep = dic_time["HYDRAULIC TIMESTEP"]
         # 转换为分钟浮点数，兼容 EPANET 的 H:MM 和 H:MM:SS 写法
-        globals.PATTERN_TIME_STEP = (
+        pattern_time_step = (
             parse_clock_duration_seconds(
-                globals.hydraulic_timestep,
+                hydraulic_timestep,
                 field_name="HYDRAULIC TIMESTEP",
             )
             / 60
         )
+        project_scada = scada_mappings or load_realtime_element_mappings(name_c)
         # 对输入的时间参数进行处理
         pattern_start_time = convert_time_format(modify_pattern_start_time)
         # 获取模拟开始时间是对应pattern的第几个数
-        modify_index = get_pattern_index(pattern_start_time)
+        modify_index = get_pattern_index(pattern_start_time, pattern_time_step)
         # 遍历水泵的pattern_id，并根据输入的pump_pattern修改pattern的值
         # for pump_pattern_id in pump_pattern_ids:
         #     # 检查pump_pattern中pump_pattern_id对应的第一个频率值是否为有效数字（非空、非NaN）。如果该值有效，则继续执行代码块。
@@ -284,7 +264,7 @@ def run_simulation(
         #         set_pattern(name_c, cs)
         # 修改模拟开始的时间
         str_pattern_start = get_pattern_index_str(
-            convert_time_format(modify_pattern_start_time)
+            convert_time_format(modify_pattern_start_time), pattern_time_step
         )
         dic_time = get_time(name_c)
         dic_time["PATTERN START"] = str_pattern_start
@@ -295,18 +275,18 @@ def run_simulation(
         cs.operations.append(dic_time)
         set_time(name_c, cs)
         # 根据SCADA实时数据进行修改，如果没有对应的SCADA数据，如未来的时间点，则不改变pg数据库的数据
-        if globals.reservoirs_id:
+        if project_scada.reservoirs:
             # reservoirs_id = {'ZBBDJSCP000002': '2497', 'R00003': '2571'}
             # 1.获取reservoir的SCADA数据,形式如{'2497': '3.1231', '2571': '2.7387'}
             reservoir_SCADA_data_dict = TimescaleInternalQueries.query_scada_by_ids_time(
-                device_ids=list(globals.reservoirs_id.values()),
+                device_ids=list(project_scada.reservoirs.values()),
                 query_time=modify_pattern_start_time,
                 db_name=name,
             )
             # 2.构建出新字典，形式如{'ZBBDJSCP000002': '3.1231', 'R00003': '2.7387'}
             reservoir_dict = {
                 key: reservoir_SCADA_data_dict[value]
-                for key, value in globals.reservoirs_id.items()
+                for key, value in project_scada.reservoirs.items()
             }
             # 3.修改reservoir液位模式
             for reservoir_name, value in reservoir_dict.items():
@@ -316,20 +296,21 @@ def run_simulation(
                         name_c, get_reservoir(name_c, reservoir_name)["pattern"]
                     )
                     reservoir_pattern["factors"][modify_index] = (
-                        float(value) + globals.RESERVOIR_BASIC_HEIGHT
+                        float(value) + RESERVOIR_BASIC_HEIGHT
                     )
                     cs = ChangeSet()
                     cs.append(reservoir_pattern)
                     set_pattern(name_c, cs)
-        if globals.tanks_id:
+        if project_scada.tanks:
             # 修改tank初始液位
             tank_SCADA_data_dict = TimescaleInternalQueries.query_scada_by_ids_time(
-                device_ids=list(globals.tanks_id.values()),
+                device_ids=list(project_scada.tanks.values()),
                 query_time=modify_pattern_start_time,
                 db_name=name,
             )
             tank_dict = {
-                key: tank_SCADA_data_dict[value] for key, value in globals.tanks_id.items()
+                key: tank_SCADA_data_dict[value]
+                for key, value in project_scada.tanks.items()
             }
             for tank_name, value in tank_dict.items():
                 if value and float(value) != 0:
@@ -338,17 +319,17 @@ def run_simulation(
                     cs = ChangeSet()
                     cs.append(tank)
                     set_tank(name_c, cs)
-        if globals.fixed_pumps_id:
+        if project_scada.fixed_pumps:
             # 修改工频泵的pattern
             fixed_pump_SCADA_data_dict = TimescaleInternalQueries.query_scada_by_ids_time(
-                device_ids=list(globals.fixed_pumps_id.values()),
+                device_ids=list(project_scada.fixed_pumps.values()),
                 query_time=modify_pattern_start_time,
                 db_name=name,
             )
             # print(fixed_pump_SCADA_data_dict)
             fixed_pump_dict = {
                 key: fixed_pump_SCADA_data_dict[value]
-                for key, value in globals.fixed_pumps_id.items()
+                for key, value in project_scada.fixed_pumps.items()
             }
             # print(fixed_pump_dict)
             for fixed_pump_name, value in fixed_pump_dict.items():
@@ -362,18 +343,18 @@ def run_simulation(
                     cs = ChangeSet()
                     cs.append(pump_pattern)
                     set_pattern(name_c, cs)
-        if globals.variable_pumps_id:
+        if project_scada.variable_pumps:
             # 修改变频泵的pattern
             variable_pump_SCADA_data_dict = (
                 TimescaleInternalQueries.query_scada_by_ids_time(
-                    device_ids=list(globals.variable_pumps_id.values()),
+                    device_ids=list(project_scada.variable_pumps.values()),
                     query_time=modify_pattern_start_time,
                     db_name=name,
                 )
             )
             variable_pump_dict = {
                 key: variable_pump_SCADA_data_dict[value]
-                for key, value in globals.variable_pumps_id.items()
+                for key, value in project_scada.variable_pumps.items()
             }
             for variable_pump_name, value in variable_pump_dict.items():
                 if value:
@@ -384,16 +365,16 @@ def run_simulation(
                     cs = ChangeSet()
                     cs.append(pump_pattern)
                     set_pattern(name_c, cs)
-        if globals.demand_id:
+        if project_scada.demand:
             # 基于实时数据，修改大用户节点的pattern
             demand_SCADA_data_dict = TimescaleInternalQueries.query_scada_by_ids_time(
-                device_ids=list(globals.demand_id.values()),
+                device_ids=list(project_scada.demand.values()),
                 query_time=modify_pattern_start_time,
                 db_name=name,
             )
             demand_dict = {
                 key: demand_SCADA_data_dict[value]
-                for key, value in globals.demand_id.items()
+                for key, value in project_scada.demand.items()
             }
             for demand_name, value in demand_dict.items():
                 if value is not None and not np.isnan(float(value)):
@@ -421,7 +402,7 @@ def run_simulation(
                 if not np.isnan(modify_reservoir_head_pattern[reservoir_name][0]):
                     # 给 list 中的所有元素加上 RESERVOIR_BASIC_HEIGHT
                     modified_values = [
-                        value + globals.RESERVOIR_BASIC_HEIGHT
+                        value + RESERVOIR_BASIC_HEIGHT
                         for value in modify_reservoir_head_pattern[reservoir_name]
                     ]
                     reservoir_pattern = get_pattern(
diff --git a/app/services/tjnetwork.py b/app/services/tjnetwork.py
index 06d7fc6..de77ad5 100644
--- a/app/services/tjnetwork.py
+++ b/app/services/tjnetwork.py
@@ -13,7 +13,7 @@ from app.algorithms.water_demand import (
     calculate_demand_to_nodes,
     calculate_demand_to_region,
 )
-from app.infra.db.postgresql.scada_assets import (
+from app.infra.db.postgresql.scada import (
     get_all_scada_info,
     get_scada_info,
     get_scada_info_schema,
diff --git a/contracts/manifest.json b/contracts/manifest.json
index 55a34aa..70684fc 100644
--- a/contracts/manifest.json
+++ b/contracts/manifest.json
@@ -3,7 +3,7 @@
   "contracts": {
     "server": {
       "file": "server-v1.openapi.json",
-      "sha256": "34f67bf3b6f1da263d0271e5a1f3cb599c128c4b422e44d2d6d540d99f7855f4"
+      "sha256": "d0364fb08c6f18fac2ea9c9980ef21115fc01f110cd10f25f90d97d0bc0e6367"
     }
   }
 }
diff --git a/contracts/server-v1.openapi.json b/contracts/server-v1.openapi.json
index a584c0c..0e9bc10 100644
--- a/contracts/server-v1.openapi.json
+++ b/contracts/server-v1.openapi.json
@@ -2340,6 +2340,48 @@
         "title": "RunSimulationManuallyByDateRest",
         "type": "object"
       },
+      "ScadaReadingBatchItem": {
+        "properties": {
+          "cleaned_value": {
+            "anyOf": [
+              {
+                "type": "number"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Cleaned Value"
+          },
+          "device_id": {
+            "minLength": 1,
+            "title": "Device Id",
+            "type": "string"
+          },
+          "monitored_value": {
+            "anyOf": [
+              {
+                "type": "number"
+              },
+              {
+                "type": "null"
+              }
+            ],
+            "title": "Monitored Value"
+          },
+          "time": {
+            "format": "date-time",
+            "title": "Time",
+            "type": "string"
+          }
+        },
+        "required": [
+          "time",
+          "device_id"
+        ],
+        "title": "ScadaReadingBatchItem",
+        "type": "object"
+      },
       "SchedulingAnalysisRest": {
         "properties": {
           "pump_control": {
@@ -19116,7 +19158,7 @@
     },
     "/api/v1/pipeline-health-predictions": {
       "get": {
-        "description": "预测管道健康状况\n\n根据管网名称和当前时间，查询管道信息和实时数据，\n使用随机生存森林模型预测管道的生存概率。\n\nArgs:\n    query_time: 查询时间\n    network_name: 管网名称（或数据库名称）\n    timescale_conn: TimescaleDB连接\n\nReturns:\n    预测结果列表，每个元素包含 link_id 和对应的生存函数\n\nRaises:\n    HTTPException: 当模型文件不存在返回404错误，其他错误返回400或500错误",
+        "description": "预测管道健康状况\n\n根据管网名称和当前时间，查询管道信息和实时数据，\n使用随机生存森林模型预测管道的生存概率。\n\nArgs:\n    query_time: 查询时间\n    timescale_conn: TimescaleDB连接\n\nReturns:\n    预测结果列表，每个元素包含 link_id 和对应的生存函数\n\nRaises:\n    HTTPException: 当模型文件不存在返回404错误，其他错误返回400或500错误",
         "operationId": "get_pipeline_health_predictions",
         "parameters": [
           {
@@ -33641,126 +33683,6 @@
         ]
       }
     },
-    "/api/v1/timeseries/analysis/runs/{run_id}/results": {
-      "post": {
-        "operationId": "post_timeseries_analysis_runs_run_id_results",
-        "parameters": [
-          {
-            "description": "分析运行 ID",
-            "in": "path",
-            "name": "run_id",
-            "required": true,
-            "schema": {
-              "description": "分析运行 ID",
-              "format": "uuid",
-              "title": "Run Id",
-              "type": "string"
-            }
-          },
-          {
-            "in": "header",
-            "name": "X-Project-Id",
-            "required": true,
-            "schema": {
-              "title": "X-Project-Id",
-              "type": "string"
-            }
-          }
-        ],
-        "requestBody": {
-          "content": {
-            "application/json": {
-              "schema": {
-                "title": "Payload",
-                "type": "object"
-              }
-            }
-          },
-          "required": true
-        },
-        "responses": {
-          "201": {
-            "content": {
-              "application/json": {
-                "schema": {
-                  "$ref": "#/components/schemas/JsonValue"
-                }
-              }
-            },
-            "description": "Successful Response"
-          },
-          "401": {
-            "content": {
-              "application/json": {
-                "schema": {
-                  "$ref": "#/components/schemas/ProblemDetails"
-                }
-              }
-            },
-            "description": "Authentication required"
-          },
-          "403": {
-            "content": {
-              "application/json": {
-                "schema": {
-                  "$ref": "#/components/schemas/ProblemDetails"
-                }
-              }
-            },
-            "description": "Insufficient permission"
-          },
-          "404": {
-            "content": {
-              "application/json": {
-                "schema": {
-                  "$ref": "#/components/schemas/ProblemDetails"
-                }
-              }
-            },
-            "description": "Resource not found"
-          },
-          "409": {
-            "content": {
-              "application/json": {
-                "schema": {
-                  "$ref": "#/components/schemas/ProblemDetails"
-                }
-              }
-            },
-            "description": "Resource conflict"
-          },
-          "422": {
-            "content": {
-              "application/json": {
-                "schema": {
-                  "$ref": "#/components/schemas/ProblemDetails"
-                }
-              }
-            },
-            "description": "Validation error"
-          },
-          "503": {
-            "content": {
-              "application/json": {
-                "schema": {
-                  "$ref": "#/components/schemas/ProblemDetails"
-                }
-              }
-            },
-            "description": "Dependency unavailable"
-          }
-        },
-        "security": [
-          {
-            "OAuth2PasswordBearer": []
-          }
-        ],
-        "summary": "Store Analysis Results",
-        "tags": [
-          "TimescaleDB - Analysis"
-        ]
-      }
-    },
     "/api/v1/timeseries/analysis/runs/{run_id}/values": {
       "get": {
         "operationId": "get_timeseries_analysis_runs_run_id_values",
@@ -35526,8 +35448,10 @@
               "schema": {
                 "description": "SCADA设备监测数据列表",
                 "items": {
-                  "type": "object"
+                  "$ref": "#/components/schemas/ScadaReadingBatchItem"
                 },
+                "maxItems": 10000,
+                "minItems": 1,
                 "title": "Data",
                 "type": "array"
               }
diff --git a/tests/api/test_scada_timeseries_endpoints.py b/tests/api/test_scada_timeseries_endpoints.py
new file mode 100644
index 0000000..228ba87
--- /dev/null
+++ b/tests/api/test_scada_timeseries_endpoints.py
@@ -0,0 +1,59 @@
+import asyncio
+from datetime import datetime, timezone
+from unittest.mock import AsyncMock
+
+import pytest
+from fastapi import HTTPException
+
+from app.api.v1.endpoints.timeseries import scada as scada_endpoint
+
+
+def _reading(device_id: str) -> scada_endpoint.ScadaReadingBatchItem:
+    return scada_endpoint.ScadaReadingBatchItem(
+        time=datetime(2026, 8, 27, tzinfo=timezone.utc),
+        device_id=device_id,
+        monitored_value=12.5,
+    )
+
+
+def test_scada_batch_rejects_devices_missing_from_bizdb(monkeypatch):
+    monkeypatch.setattr(
+        scada_endpoint.ScadaInfoRepository,
+        "get_existing_device_ids",
+        AsyncMock(return_value={"known"}),
+    )
+    insert = AsyncMock()
+    monkeypatch.setattr(scada_endpoint.ScadaRepository, "insert_scada_batch", insert)
+
+    with pytest.raises(HTTPException, match="missing") as exc_info:
+        asyncio.run(
+            scada_endpoint.insert_scada_data(
+                [_reading("known"), _reading("missing")],
+                conn=object(),
+                postgres_conn=object(),
+            )
+        )
+
+    assert exc_info.value.status_code == 422
+    insert.assert_not_awaited()
+
+
+def test_scada_batch_writes_only_after_bizdb_validation(monkeypatch):
+    monkeypatch.setattr(
+        scada_endpoint.ScadaInfoRepository,
+        "get_existing_device_ids",
+        AsyncMock(return_value={"known"}),
+    )
+    insert = AsyncMock()
+    monkeypatch.setattr(scada_endpoint.ScadaRepository, "insert_scada_batch", insert)
+
+    result = asyncio.run(
+        scada_endpoint.insert_scada_data(
+            [_reading(" known ")],
+            conn=object(),
+            postgres_conn=object(),
+        )
+    )
+
+    assert result == {"message": "Inserted 1 records"}
+    assert insert.await_args.args[1][0]["device_id"] == "known"
diff --git a/tests/api/test_simulation_endpoints.py b/tests/api/test_simulation_endpoints.py
index a4dab84..0ced024 100644
--- a/tests/api/test_simulation_endpoints.py
+++ b/tests/api/test_simulation_endpoints.py
@@ -1,4 +1,5 @@
 from datetime import datetime, timezone
+from types import SimpleNamespace
 
 from fastapi.testclient import TestClient
 
@@ -39,7 +40,9 @@ def _load_simulation_module(monkeypatch):
         {
             "get_time": lambda name: {"HYDRAULIC TIMESTEP": "0:15:00"},
             "run_simulation": lambda **kwargs: None,
-            "query_corresponding_element_id_and_query_id": lambda name: None,
+            "query_corresponding_element_id_and_query_id": lambda name: SimpleNamespace(
+                fixed_pumps={}, variable_pumps={}
+            ),
             "query_corresponding_pattern_id_and_query_id": lambda name: None,
             "query_non_realtime_region": lambda name: [],
             "get_source_outflow_region_id": lambda name, region_result: {},
@@ -49,7 +52,6 @@ def _load_simulation_module(monkeypatch):
             "get_realtime_region_patterns": lambda name, source_outflow_region_id, realtime_region_pipe_flow_and_demand_id: ({}, {}),
         },
     )
-    install_stub(monkeypatch, "app.services.globals", {})
     install_stub(
         monkeypatch,
         "app.services.tjnetwork",
diff --git a/tests/unit/test_analysis_simulation.py b/tests/unit/test_analysis_simulation.py
index dd81a34..1e47b58 100644
--- a/tests/unit/test_analysis_simulation.py
+++ b/tests/unit/test_analysis_simulation.py
@@ -7,6 +7,18 @@ from uuid import uuid4
 import pytest
 
 
+def _empty_scada_mappings(simulation):
+    return simulation.ScadaElementMappings(
+        reservoirs={},
+        tanks={},
+        fixed_pumps={},
+        variable_pumps={},
+        pressure={},
+        demand={},
+        quality={},
+    )
+
+
 def test_run_simulation_exposes_explicit_valve_control():
     from app.services import simulation
 
@@ -174,6 +186,7 @@ def test_extended_simulation_stores_results_by_run_id(monkeypatch):
         modify_total_duration=900,
         scheme_type="burst_analysis",
         scheme_name="case",
+        scada_mappings=_empty_scada_mappings(simulation),
     )
 
     args, kwargs = storage_calls[0]
@@ -250,6 +263,7 @@ def test_extended_simulation_marks_run_failed_when_result_storage_fails(monkeypa
             modify_total_duration=900,
             scheme_type="burst_analysis",
             scheme_name="case",
+            scada_mappings=_empty_scada_mappings(simulation),
         )
 
     assert [call[1]["status"] for call in lifecycle_calls] == ["failed"]
diff --git a/tests/unit/test_postgres_scada_repository.py b/tests/unit/test_postgres_scada_repository.py
index a0ed677..174a568 100644
--- a/tests/unit/test_postgres_scada_repository.py
+++ b/tests/unit/test_postgres_scada_repository.py
@@ -1,5 +1,9 @@
 import asyncio
+from types import MappingProxyType
 
+import pytest
+
+from app.infra.db.postgresql import scada
 from app.infra.db.postgresql.scada import ScadaInfoRepository
 
 
@@ -63,3 +67,24 @@ def test_get_scadas_normalizes_id_and_type():
     assert "node_id" in conn.cursor_instance.query
     assert "link_id" in conn.cursor_instance.query
     assert "FROM gis.scada_devices" in conn.cursor_instance.query
+
+
+def test_realtime_element_mappings_are_project_local_and_immutable(monkeypatch):
+    monkeypatch.setattr(
+        scada,
+        "read_all",
+        lambda *_args: [
+            {
+                "device_type": " PRESSURE ",
+                "element_id": " J1 ",
+                "api_query_id": " sensor-1 ",
+            }
+        ],
+    )
+
+    mappings = scada.load_realtime_element_mappings("project-a")
+
+    assert mappings.pressure == {"J1": "sensor-1"}
+    assert isinstance(mappings.pressure, MappingProxyType)
+    with pytest.raises(TypeError):
+        mappings.pressure["J2"] = "sensor-2"
diff --git a/tests/unit/test_realtime_repository.py b/tests/unit/test_realtime_repository.py
index cebacc1..42ad06f 100644
--- a/tests/unit/test_realtime_repository.py
+++ b/tests/unit/test_realtime_repository.py
@@ -94,13 +94,13 @@ def test_realtime_node_and_link_replacement_share_outer_transaction(monkeypatch)
     calls: list[str] = []
     monkeypatch.setattr(
         RealtimeRepository,
-        "insert_nodes_batch_sync",
-        lambda _conn, _data: calls.append("nodes"),
+        "_copy_nodes_sync",
+        lambda _cur, _data, _time: calls.append("nodes"),
     )
     monkeypatch.setattr(
         RealtimeRepository,
-        "insert_links_batch_sync",
-        lambda _conn, _data: calls.append("links"),
+        "_copy_links_sync",
+        lambda _cur, _data, _time: calls.append("links"),
     )
 
     RealtimeRepository.store_realtime_simulation_result_sync(
diff --git a/tests/unit/test_scada_cleaning.py b/tests/unit/test_scada_cleaning.py
index a830113..2a8d37d 100644
--- a/tests/unit/test_scada_cleaning.py
+++ b/tests/unit/test_scada_cleaning.py
@@ -1,4 +1,5 @@
 import asyncio
+from contextlib import asynccontextmanager
 from datetime import datetime, timezone
 from unittest.mock import AsyncMock
 
@@ -10,6 +11,12 @@ from app.api.v1.endpoints.timeseries import composite as composite_endpoint
 from app.infra.db.timescaledb import composite_queries
 
 
+class _FakeTimescaleConnection:
+    @asynccontextmanager
+    async def transaction(self):
+        yield
+
+
 def test_clean_scada_uses_current_project_metadata(monkeypatch):
     """Fengyang data must not be classified with the global tjwater metadata."""
 
@@ -51,7 +58,7 @@ def test_clean_scada_uses_current_project_metadata(monkeypatch):
 
     result = asyncio.run(
         composite_queries.CompositeQueries.clean_scada_data(
-            object(),
+            _FakeTimescaleConnection(),
             object(),
             ["fengyang-pressure-1"],
             datetime(2026, 6, 1, tzinfo=timezone.utc),
@@ -79,8 +86,8 @@ def test_clean_scada_rejects_devices_missing_from_project_metadata(monkeypatch):
     with pytest.raises(ValueError, match="缺少元数据"):
         asyncio.run(
             composite_queries.CompositeQueries.clean_scada_data(
-                object(),
-                object(),
+                _FakeTimescaleConnection(),
+                _FakeTimescaleConnection(),
                 ["fengyang-pressure-1"],
                 datetime(2026, 6, 1, tzinfo=timezone.utc),
                 datetime(2026, 6, 2, tzinfo=timezone.utc),
@@ -126,7 +133,7 @@ def test_clean_scada_rejects_zero_database_updates(monkeypatch):
     with pytest.raises(ValueError, match="未产生任何数据库更新"):
         asyncio.run(
             composite_queries.CompositeQueries.clean_scada_data(
-                object(),
+                _FakeTimescaleConnection(),
                 object(),
                 ["fengyang-pressure-1"],
                 datetime(2026, 6, 1, tzinfo=timezone.utc),
@@ -170,7 +177,7 @@ def test_clean_scada_propagates_write_failures(monkeypatch):
     with pytest.raises(RuntimeError, match="database write failed"):
         asyncio.run(
             composite_queries.CompositeQueries.clean_scada_data(
-                object(),
+                _FakeTimescaleConnection(),
                 object(),
                 ["fengyang-pressure-1"],
                 datetime(2026, 6, 1, tzinfo=timezone.utc),
diff --git a/tests/unit/test_scheme_management_lifecycle.py b/tests/unit/test_scheme_management_lifecycle.py
index da2ed87..b37a4d7 100644
--- a/tests/unit/test_scheme_management_lifecycle.py
+++ b/tests/unit/test_scheme_management_lifecycle.py
@@ -31,7 +31,7 @@ def test_repeated_analysis_names_create_distinct_run_ids(monkeypatch) -> None:
     assert first != second
     assert cursor.execute.call_count == 2
     assert all(
-        "insert into analysis.runs" in call.args[0]
+        "insert into analysis.runs" in call.args[0].lower()
         for call in cursor.execute.call_args_list
     )
 
@@ -57,6 +57,43 @@ def test_update_analysis_run_targets_execution_id(monkeypatch) -> None:
     )
 
     statement, params = cursor.execute.call_args.args
-    assert "where run_id = %s" in statement
+    assert "where run_id = %s" in statement.lower()
     assert params[-1] == run_id
     assert params[1] == "completed"
+
+
+def test_run_and_business_result_share_one_transaction(monkeypatch) -> None:
+    connection = object()
+    context = MagicMock()
+    context.__enter__.return_value = connection
+    monkeypatch.setattr(
+        scheme_management,
+        "project_transaction",
+        lambda _name: context,
+    )
+    created_ids = []
+    result_ids = []
+    monkeypatch.setattr(
+        scheme_management.AnalysisRepository,
+        "create_run_sync",
+        lambda conn, **kwargs: created_ids.append((conn, kwargs["run_id"])),
+    )
+    monkeypatch.setattr(
+        scheme_management.AnalysisRepository,
+        "insert_result_sync",
+        lambda conn, run_id, **_kwargs: result_ids.append((conn, run_id)),
+    )
+
+    run_id = scheme_management.store_analysis_run_with_result(
+        name="tjwater_v2",
+        scheme_name="leak-run",
+        scheme_type="dma_leak_identification",
+        username="alice",
+        scheme_start_time="2026-08-24T00:00:00Z",
+        scheme_detail={},
+        result_type="leakage_identification",
+        result_payload={"rows": []},
+    )
+
+    assert created_ids == [(connection, run_id)]
+    assert result_ids == [(connection, run_id)]
diff --git a/tests/unit/test_sensor_placement_service.py b/tests/unit/test_sensor_placement_service.py
index 01884d6..b94b2a2 100644
--- a/tests/unit/test_sensor_placement_service.py
+++ b/tests/unit/test_sensor_placement_service.py
@@ -97,6 +97,31 @@ def test_update_validates_nodes_before_write(monkeypatch):
         )
 
 
+def test_list_sensor_placements_batches_node_lookup(monkeypatch):
+    first = _run()
+    second = {**_run(), "run_id": uuid4(), "sensor_locations": ["J2", "J3"]}
+    lookup_calls = []
+    monkeypatch.setattr(
+        sensor_placement.sensor_placement_repository,
+        "get_all_sensor_placements",
+        lambda _network: [first, second],
+    )
+    monkeypatch.setattr(
+        sensor_placement.sensor_placement_repository,
+        "get_sensor_placement_nodes",
+        lambda _network, node_ids: lookup_calls.append(node_ids)
+        or [
+            {**_point(), "node_id": node_id}
+            for node_id in node_ids
+        ],
+    )
+
+    runs = sensor_placement.list_sensor_placement_runs("tjwater_v2")
+
+    assert lookup_calls == [["J1", "J2", "J3"]]
+    assert [len(run["sensor_points"]) for run in runs] == [2, 2]
+
+
 def test_sensor_nodes_query_uses_new_network_and_gis_schemas(monkeypatch):
     cursor = _mock_project_cursor(monkeypatch)
     cursor.fetchall.return_value = []
@@ -108,6 +133,9 @@ def test_sensor_nodes_query_uses_new_network_and_gis_schemas(monkeypatch):
     assert "network.links" in query
     assert "gis.node_geometries" in query
     assert "ST_Transform(g.geom, 3857)" in query
+    assert "l.start_node_id = ANY(%s)" in query
+    assert "l.end_node_id = ANY(%s)" in query
+    assert "CROSS JOIN LATERAL" not in query
     assert cursor.execute.call_args.args[1] == (["J1"], ["J1"], ["J1"])
 
 
diff --git a/tests/unit/test_wndb_query_safety.py b/tests/unit/test_wndb_query_safety.py
index 34c7a3a..367cf16 100644
--- a/tests/unit/test_wndb_query_safety.py
+++ b/tests/unit/test_wndb_query_safety.py
@@ -1,7 +1,7 @@
 import ast
 from pathlib import Path
 
-from app.infra.db.postgresql import scada_assets
+from app.infra.db.postgresql import scada
 from app.native.wndb.core.database import ChangeSet, sql_literal
 from app.native.wndb.gis import coordinates
 from app.native.wndb.model import controls, junctions, patterns
@@ -48,9 +48,9 @@ def test_get_all_scada_info_reads_materialized_view(monkeypatch) -> None:
         statements.append(statement)
         return []
 
-    monkeypatch.setattr(scada_assets, "read_all", fake_read_all)
+    monkeypatch.setattr(scada, "read_all", fake_read_all)
 
-    assert scada_assets.get_all_scada_info("project_a") == []
+    assert scada.get_all_scada_info("project_a") == []
     assert "FROM gis.scada_devices" in statements[0]