diff --git a/.env.example b/.env.example
index cf775d5..5e29c1e 100644
--- a/.env.example
+++ b/.env.example
@@ -54,6 +54,13 @@ PRICEBOT_REQUEST_TIMEOUT_SEC=30
 # 比价(intent/recognize + price/step)透传超时:大上下文 LLM + 逐帧 LLM,给 60s
 PRICEBOT_COMPARE_TIMEOUT_SEC=60
 
+# ===== 内部端点 (server→server, pricebot 上报价格观测) =====
+# pricebot 比价 done 后把各平台到手价 POST 到本服务 /internal/price-observation 落库
+# (price_observation 表,比价资产沉淀层),靠共享密钥头 X-Internal-Secret 校验。
+# 必须与 pricebot 侧的 INTERNAL_API_SECRET **同值**;留空 = 内部写端点关闭(返 503)。
+# 启用前两边都填同一高熵串:python -c "import secrets; print(secrets.token_urlsafe(48))"
+INTERNAL_API_SECRET=
+
 # ===== CORS =====
 # 逗号分隔,生产留空(只让 app 调,不开放 web)。本地开发可加 http://localhost:5173 之类
 CORS_ALLOW_ORIGINS=
diff --git a/alembic/versions/price_observation_table.py b/alembic/versions/price_observation_table.py
new file mode 100644
index 0000000..ca7d4a3
--- /dev/null
+++ b/alembic/versions/price_observation_table.py
@@ -0,0 +1,74 @@
+"""price_observation table (价格观测:比价沉淀的价格资产层,server 侧无条件落库)
+
+Revision ID: price_observation_table
+Revises: wx_transfer_auth
+Create Date: 2026-06-07 00:00:00.000000
+
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+# revision identifiers, used by Alembic.
+revision: str = 'price_observation_table'
+down_revision: Union[str, Sequence[str], None] = 'wx_transfer_auth'
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        'price_observation',
+        sa.Column('id', sa.Integer(), autoincrement=True, nullable=False),
+        sa.Column('observed_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
+        sa.Column('trace_id', sa.String(length=64), nullable=False),
+        sa.Column('business_type', sa.String(length=16), nullable=False),
+        sa.Column('platform', sa.String(length=32), nullable=False),
+        sa.Column('platform_store_id', sa.String(length=64), nullable=True),
+        sa.Column('store_name', sa.String(length=128), nullable=True),
+        sa.Column('city', sa.String(length=64), nullable=True),
+        sa.Column('geohash', sa.String(length=16), nullable=True),
+        sa.Column('lng', sa.Float(), nullable=True),
+        sa.Column('lat', sa.Float(), nullable=True),
+        sa.Column('is_source', sa.Boolean(), nullable=False),
+        sa.Column('scope', sa.String(length=16), nullable=False),
+        sa.Column('price_cents', sa.Integer(), nullable=True),
+        sa.Column('coupon_saved_cents', sa.Integer(), nullable=True),
+        sa.Column('coupon_name', sa.String(length=64), nullable=True),
+        sa.Column('store_closed', sa.String(length=32), nullable=True),
+        sa.Column('rank', sa.Integer(), nullable=True),
+        # PG 上为 JSONB,其它(SQLite)为 JSON——与模型层 with_variant 对齐
+        sa.Column('dishes', sa.JSON().with_variant(sa.dialects.postgresql.JSONB(), 'postgresql'), nullable=True),
+        sa.Column('attrs', sa.JSON().with_variant(sa.dialects.postgresql.JSONB(), 'postgresql'), nullable=True),
+        sa.Column('source_device_id', sa.String(length=64), nullable=True),
+        sa.Column('source_user_id', sa.Integer(), nullable=True),
+        sa.Column('confidence', sa.Float(), nullable=False),
+        sa.Column('created_at', sa.DateTime(timezone=True), server_default=sa.text('(CURRENT_TIMESTAMP)'), nullable=False),
+        sa.PrimaryKeyConstraint('id'),
+        sa.UniqueConstraint('trace_id', 'platform', 'scope', name='uq_price_obs_trace_platform_scope'),
+    )
+    with op.batch_alter_table('price_observation', schema=None) as batch_op:
+        batch_op.create_index(batch_op.f('ix_price_observation_observed_at'), ['observed_at'], unique=False)
+        batch_op.create_index(batch_op.f('ix_price_observation_trace_id'), ['trace_id'], unique=False)
+        batch_op.create_index(batch_op.f('ix_price_observation_business_type'), ['business_type'], unique=False)
+        batch_op.create_index(batch_op.f('ix_price_observation_platform'), ['platform'], unique=False)
+        batch_op.create_index(batch_op.f('ix_price_observation_store_name'), ['store_name'], unique=False)
+        batch_op.create_index(batch_op.f('ix_price_observation_geohash'), ['geohash'], unique=False)
+        batch_op.create_index(batch_op.f('ix_price_observation_source_device_id'), ['source_device_id'], unique=False)
+        batch_op.create_index(batch_op.f('ix_price_observation_source_user_id'), ['source_user_id'], unique=False)
+
+
+def downgrade() -> None:
+    with op.batch_alter_table('price_observation', schema=None) as batch_op:
+        batch_op.drop_index(batch_op.f('ix_price_observation_source_user_id'))
+        batch_op.drop_index(batch_op.f('ix_price_observation_source_device_id'))
+        batch_op.drop_index(batch_op.f('ix_price_observation_geohash'))
+        batch_op.drop_index(batch_op.f('ix_price_observation_store_name'))
+        batch_op.drop_index(batch_op.f('ix_price_observation_platform'))
+        batch_op.drop_index(batch_op.f('ix_price_observation_business_type'))
+        batch_op.drop_index(batch_op.f('ix_price_observation_trace_id'))
+        batch_op.drop_index(batch_op.f('ix_price_observation_observed_at'))
+
+    op.drop_table('price_observation')
diff --git a/app/api/internal/__init__.py b/app/api/internal/__init__.py
new file mode 100644
index 0000000..6278e2a
--- /dev/null
+++ b/app/api/internal/__init__.py
@@ -0,0 +1 @@
+"""内部(server→server)端点。不对客户端开放,靠共享密钥头校验,不走用户 JWT。"""
diff --git a/app/api/internal/price.py b/app/api/internal/price.py
new file mode 100644
index 0000000..5f654a7
--- /dev/null
+++ b/app/api/internal/price.py
@@ -0,0 +1,63 @@
+"""价格观测内部上报端点(pricebot → app-server)。
+
+pricebot 在比价 done 帧把整批价格事实 POST 到这里落库。**不是给客户端的接口**:
+不走用户 JWT,靠 server 间共享密钥头 `X-Internal-Secret` 校验(== settings.INTERNAL_API_SECRET)。
+密钥未配置(默认空)时直接 503,避免裸奔的写端点。
+
+与不鉴权的比价透传(compare.py / coupon.py)无关:那两个是客户端→server 的透传壳,
+这个是 server→server 的内部写库。
+"""
+from __future__ import annotations
+
+import hmac
+import logging
+from typing import Annotated
+
+from fastapi import APIRouter, Header, HTTPException, status
+
+from app.api.deps import DbSession
+from app.core.config import settings
+from app.repositories import price_observation as repo
+from app.schemas.price_observation import (
+    PriceObservationBatchIn,
+    PriceObservationBatchOut,
+)
+
+logger = logging.getLogger("shagua.internal.price")
+
+router = APIRouter(prefix="/internal", tags=["internal"])
+
+
+def _check_secret(x_internal_secret: str | None) -> None:
+    """共享密钥校验。未配置 → 503(挡住裸奔写端点);不匹配 → 401(常量时间比较)。"""
+    configured = settings.INTERNAL_API_SECRET
+    if not configured:
+        raise HTTPException(
+            status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
+            detail="internal api not configured",
+        )
+    if not x_internal_secret or not hmac.compare_digest(x_internal_secret, configured):
+        raise HTTPException(
+            status_code=status.HTTP_401_UNAUTHORIZED,
+            detail="invalid internal secret",
+        )
+
+
+@router.post(
+    "/price-observation",
+    response_model=PriceObservationBatchOut,
+    summary="价格观测内部上报(pricebot→app-server,落 price_observation)",
+)
+def report_price_observation(
+    payload: PriceObservationBatchIn,
+    db: DbSession,
+    x_internal_secret: Annotated[str | None, Header()] = None,
+) -> PriceObservationBatchOut:
+    _check_secret(x_internal_secret)
+    inserted, skipped = repo.insert_batch(db, payload)
+    logger.info(
+        "price_observation trace=%s inserted=%d skipped=%d device=%s user=%s",
+        payload.trace_id, inserted, skipped,
+        payload.source_device_id, payload.source_user_id,
+    )
+    return PriceObservationBatchOut(inserted=inserted, skipped=skipped)
diff --git a/app/core/config.py b/app/core/config.py
index bbcbc31..68e66f0 100644
--- a/app/core/config.py
+++ b/app/core/config.py
@@ -157,6 +157,12 @@ class Settings(BaseSettings):
             return [self.PRICEBOT_BASE_URL]
         return [u.strip() for u in self.PRICEBOT_INSTANCES.split(",") if u.strip()]
 
+    # ===== 内部(server→server)端点密钥 =====
+    # pricebot 比价 done 后把价格观测 POST 到 /internal/price-observation 落库,
+    # 靠这个共享密钥头(X-Internal-Secret)校验,与 pricebot 侧 INTERNAL_API_SECRET 同值。
+    # 默认空 = 内部写端点关闭(返 503),启用前两边都要配上同一高熵串。
+    INTERNAL_API_SECRET: str = ""
+
     # ===== 媒体文件(用户头像上传)=====
     # 落盘根目录(data/ 已 gitignore,上传不进库);对外经 StaticFiles 挂在 MEDIA_URL_PREFIX。
     # 生产可改由 nginx 直接 serve MEDIA_ROOT,绕过应用进程。
diff --git a/app/main.py b/app/main.py
index d78d455..b718e3c 100644
--- a/app/main.py
+++ b/app/main.py
@@ -20,6 +20,7 @@ from app.api.v1.compare import router as compare_router
 from app.api.v1.compare_milestone import router as compare_milestone_router
 from app.api.v1.compare_record import router as compare_record_router
 from app.api.v1.coupon import router as coupon_router
+from app.api.internal.price import router as internal_price_router
 from app.api.v1.feedback import router as feedback_router
 from app.api.v1.meituan import router as meituan_router
 from app.api.v1.order import router as order_router
@@ -90,6 +91,8 @@ app.include_router(savings_router)
 app.include_router(ad_router)
 app.include_router(order_router)
 app.include_router(report_router)
+# 内部(server→server)端点:pricebot 上报价格观测,靠共享密钥头校验,不对客户端开放。
+app.include_router(internal_price_router)
 
 # 用户上传文件(头像)静态服务。生产可改由 nginx 直接 serve MEDIA_ROOT。
 _media_root = Path(settings.MEDIA_ROOT)
diff --git a/app/models/__init__.py b/app/models/__init__.py
index 62bce18..b0f2bba 100644
--- a/app/models/__init__.py
+++ b/app/models/__init__.py
@@ -7,6 +7,7 @@ from app.models.app_config import AppConfig  # noqa: F401
 from app.models.comparison import ComparisonRecord  # noqa: F401
 from app.models.comparison_milestone import ComparisonMilestoneClaim  # noqa: F401
 from app.models.feedback import Feedback  # noqa: F401
+from app.models.price_observation import PriceObservation  # noqa: F401
 from app.models.price_report import PriceReport  # noqa: F401
 from app.models.savings import SavingsRecord  # noqa: F401
 from app.models.signin import SigninRecord  # noqa: F401
diff --git a/app/models/price_observation.py b/app/models/price_observation.py
new file mode 100644
index 0000000..84f9e2c
--- /dev/null
+++ b/app/models/price_observation.py
@@ -0,0 +1,111 @@
+"""价格观测表(price_observation)—— 比价沉淀的价格资产层。
+
+每完成一次比价,pricebot 在 done 帧把"在某平台某店,这一篮菜/这单到手价多少钱"
+按平台逐条算出来,server→server 内部上报落这里。**与登录无关、不依赖客户端上报**——
+比价透传链路当前不鉴权,只要比价跑到 done 就记,匿名用户也记(来源记 device_id,
+user_id 客户端带上时一并记)。
+
+与 comparison_record 的区别:
+- comparison_record:用户视角的「我的比价记录」,客户端登录后主动上报,按 user_id 存。
+- price_observation:**平台/门店视角的价格事实**,server 侧无条件沉淀,按 (平台,门店,菜,时间)
+  组织,是未来"别人查过同店→直接秒回价格"大数据的源头。两表独立,互不影响。
+
+「先存下来、用法后说」:结构化列给将来按 平台/门店/菜/时间/地理 查询聚合用;dishes / attrs
+(JSONB)兜底存灵活明细,免得每多记一个字段就迁移 schema。raw trace(逐 step 全过程)
+另存 pricebot 本地 jsonl.gz,本表只存提炼后的价格事实(本表可从 raw 重算)。
+"""
+from __future__ import annotations
+
+from datetime import datetime
+
+from sqlalchemy import (
+    JSON,
+    Boolean,
+    DateTime,
+    Float,
+    Integer,
+    String,
+    UniqueConstraint,
+    func,
+)
+from sqlalchemy.dialects.postgresql import JSONB
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.db.base import Base
+
+# PG 上用 JSONB(可建 GIN 索引),SQLite(本地/测试)退化为通用 JSON(同 comparison_record)。
+_JSON = JSON().with_variant(JSONB(), "postgresql")
+
+
+class PriceObservation(Base):
+    __tablename__ = "price_observation"
+    __table_args__ = (
+        # 一次比价(trace)里,同一平台、同一口径(scope)只记一条:pricebot 重试 / 客户端
+        # replay 重复上报时幂等去重(insert-or-ignore)。一次外卖比价 = 源 + N 个目标平台
+        # 各一条 scope='order'。
+        UniqueConstraint(
+            "trace_id", "platform", "scope",
+            name="uq_price_obs_trace_platform_scope",
+        ),
+    )
+
+    id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
+
+    # 观测时刻(≈ 比价 done 时;服务端落库时取 now)。按时间查"最近有效价"用,建索引。
+    observed_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), index=True, nullable=False
+    )
+    # pricebot 侧 trace_id:回指 A 层原始 trace(能溯源 / 重算)+ 幂等去重键
+    trace_id: Mapped[str] = mapped_column(String(64), index=True, nullable=False)
+    # 业务类型:food(外卖,当前唯一接通)/ ecom(电商,二期)。预留扩展。
+    business_type: Mapped[str] = mapped_column(
+        String(16), nullable=False, default="food", index=True
+    )
+
+    # ===== 门店维度(先 denormalize 成文本,不建维度表;实体归一二期再做)=====
+    platform: Mapped[str] = mapped_column(String(32), index=True, nullable=False)
+    # 平台内门店 ID(抓得到就存,将来归一最稳的 key;现在多为空)
+    platform_store_id: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    store_name: Mapped[str | None] = mapped_column(String(128), index=True, nullable=True)
+    city: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    # 地理(现在拿不到 → 留空;二期把 device 经纬度透传进比价请求后回填)
+    geohash: Mapped[str | None] = mapped_column(String(16), index=True, nullable=True)
+    lng: Mapped[float | None] = mapped_column(Float, nullable=True)
+    lat: Mapped[float | None] = mapped_column(Float, nullable=True)
+
+    # ===== 价格事实 =====
+    # 是否源平台(发起比价那家)。源平台价也是真实观测,照记。
+    is_source: Mapped[bool] = mapped_column(Boolean, nullable=False, default=False)
+    # 口径:order(整单到手价,外卖当前唯一)/ dish(单菜单价,引擎暂不逐菜读,二期)
+    scope: Mapped[str] = mapped_column(String(16), nullable=False, default="order")
+    # 该口径的到手价(分);None = 该平台采集失败 / 门店打烊,无有效价
+    price_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    coupon_saved_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    coupon_name: Mapped[str | None] = mapped_column(String(64), nullable=True)
+    # 该平台目标店打烊原因(非空时 price_cents 为 None)
+    store_closed: Mapped[str | None] = mapped_column(String(32), nullable=True)
+    # 本次比价里的全局名次(1=最便宜;按到手价升序,源/目标统一排)
+    rank: Mapped[int | None] = mapped_column(Integer, nullable=True)
+
+    # ===== 明细(JSON)=====
+    # 菜篮 [{name, qty, subtotal?}](外卖)。dish 级单价二期从这里 + 引擎增强推。
+    dishes: Mapped[list | None] = mapped_column(_JSON, nullable=True)
+    # 灵活字段兜底(配送费 / 起送 / 活动名 / 跳过菜数 等),免得加字段就迁移。
+    attrs: Mapped[dict | None] = mapped_column(_JSON, nullable=True)
+
+    # ===== 来源(溯源 / 去重 / 置信)=====
+    source_device_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
+    # user_id 当前比价链路不鉴权拿不到,客户端带上时才有;先可空。
+    source_user_id: Mapped[int | None] = mapped_column(Integer, index=True, nullable=True)
+    confidence: Mapped[float] = mapped_column(Float, nullable=False, default=1.0)
+
+    created_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), server_default=func.now(), nullable=False
+    )
+
+    def __repr__(self) -> str:  # pragma: no cover
+        return (
+            f"<PriceObservation id={self.id} platform={self.platform} "
+            f"store={self.store_name!r} price_cents={self.price_cents} "
+            f"trace_id={self.trace_id}>"
+        )
diff --git a/app/repositories/price_observation.py b/app/repositories/price_observation.py
new file mode 100644
index 0000000..0450d7f
--- /dev/null
+++ b/app/repositories/price_observation.py
@@ -0,0 +1,86 @@
+"""价格观测落库:一次比价的整批观测幂等写入。
+
+幂等键 = (trace_id, platform, scope)。pricebot 重试 / replay 会重复上报同一 trace,
+这里先查该 trace 已存在的 (platform, scope),只插新的——跨方言(PG / SQLite dev)都安全,
+不依赖 ON CONFLICT。极端并发下若仍撞唯一约束,IntegrityError 回滚后忽略(本批本就幂等)。
+"""
+from __future__ import annotations
+
+import logging
+
+from sqlalchemy import select
+from sqlalchemy.exc import IntegrityError
+from sqlalchemy.orm import Session
+
+from app.models.price_observation import PriceObservation
+from app.schemas.price_observation import PriceObservationBatchIn
+
+logger = logging.getLogger("shagua.price_obs")
+
+
+def insert_batch(db: Session, payload: PriceObservationBatchIn) -> tuple[int, int]:
+    """写入一次比价的整批观测,返回 (inserted, skipped)。
+
+    skipped = 因 (trace_id, platform, scope) 已存在而幂等跳过的条数。
+    """
+    if not payload.observations:
+        return 0, 0
+
+    # 该 trace 已有的 (platform, scope),用于幂等过滤
+    existing: set[tuple[str, str]] = set(
+        db.execute(
+            select(PriceObservation.platform, PriceObservation.scope).where(
+                PriceObservation.trace_id == payload.trace_id
+            )
+        ).all()
+    )
+
+    inserted = 0
+    skipped = 0
+    for obs in payload.observations:
+        key = (obs.platform, obs.scope)
+        if key in existing:
+            skipped += 1
+            continue
+        existing.add(key)  # 同批内去重(防 pricebot 同帧给重复平台)
+        db.add(
+            PriceObservation(
+                trace_id=payload.trace_id,
+                business_type=payload.business_type,
+                platform=obs.platform,
+                platform_store_id=obs.platform_store_id,
+                store_name=payload.store_name,
+                city=payload.city,
+                geohash=payload.geohash,
+                lng=payload.lng,
+                lat=payload.lat,
+                is_source=obs.is_source,
+                scope=obs.scope,
+                price_cents=obs.price_cents,
+                coupon_saved_cents=obs.coupon_saved_cents,
+                coupon_name=obs.coupon_name,
+                store_closed=obs.store_closed,
+                rank=obs.rank,
+                dishes=payload.dishes,
+                attrs=obs.attrs,
+                source_device_id=payload.source_device_id,
+                source_user_id=payload.source_user_id,
+            )
+        )
+        inserted += 1
+
+    if inserted == 0:
+        return 0, skipped
+
+    try:
+        db.commit()
+    except IntegrityError:
+        # 并发下另一个请求刚插了同 (trace,platform,scope) → 唯一约束撞了。
+        # 本批本就幂等,回滚后当作全跳过(不重试,失败就失败,见 P0 约定)。
+        db.rollback()
+        logger.warning(
+            "price_observation 并发幂等冲突 trace=%s,本批回滚跳过", payload.trace_id
+        )
+        return 0, skipped + inserted
+
+    return inserted, skipped
diff --git a/app/schemas/price_observation.py b/app/schemas/price_observation.py
new file mode 100644
index 0000000..f595dd0
--- /dev/null
+++ b/app/schemas/price_observation.py
@@ -0,0 +1,49 @@
+"""价格观测内部上报的收发模型。
+
+pricebot 在比价 done 帧 server→server POST 一批观测(一次比价 = 源 + N 目标平台各一条)。
+批级字段(门店 / 菜篮 / 地理 / 来源)所有观测共享,放外层;逐平台字段放 observations[]。
+"""
+from __future__ import annotations
+
+from pydantic import BaseModel, Field
+
+
+class PriceObservationItem(BaseModel):
+    """单个平台的价格观测(一次比价里的一行)。"""
+
+    platform: str
+    is_source: bool = False
+    scope: str = "order"                          # order / dish
+    price_cents: int | None = None                # 到手价(分);None=该平台失败/打烊
+    coupon_saved_cents: int | None = None
+    coupon_name: str | None = None
+    store_closed: str | None = None               # 打烊原因(非空时 price_cents=None)
+    rank: int | None = None                        # 全局名次(1=最便宜)
+    platform_store_id: str | None = None
+    attrs: dict | None = None                      # 该平台的灵活明细兜底
+
+
+class PriceObservationBatchIn(BaseModel):
+    """一次比价的整批价格观测上报体。"""
+
+    trace_id: str
+    business_type: str = "food"
+    # 门店 / 菜篮 / 地理 —— 一次比价同一份,批级共享
+    store_name: str | None = None
+    city: str | None = None
+    geohash: str | None = None
+    lng: float | None = None
+    lat: float | None = None
+    dishes: list | None = None                     # [{name, qty, subtotal?}]
+    # 来源
+    source_device_id: str | None = None
+    source_user_id: int | None = None
+    # 逐平台观测(源 + 各目标)
+    observations: list[PriceObservationItem] = Field(default_factory=list)
+
+
+class PriceObservationBatchOut(BaseModel):
+    """上报结果:本批新写入条数 / 因幂等跳过条数。"""
+
+    inserted: int
+    skipped: int