feat: 接入京东联盟订单到数据大盘

改了什么：新增京东联盟订单查询客户端、cps_order 京东字段迁移、CPS 对账接口 platform=all/jd、数据大盘京东订单与佣金聚合。
为什么改：数据大盘需要展示京东 CPS 真实订单佣金，不再保持占位。
验证方式：python -m pytest tests\\test_admin_read.py::test_dashboard_overview tests\\test_cps_admin.py::test_jd_reconcile_updates_dashboard -q；本地真实拉取 2026-06-27 京东订单 3 行，页面显示京东有效 1 单/佣金 ¥0.13。

.env.example

@@ -58,6 +58,13 @@ MT_CPS_DEFAULT_SID=sgbjia
 # 线上国内服务器留空(=直连)。留空且本机直连失败时 /feed、/coupons、/top-sales 会返回空。
 MT_CPS_PROXY=
 
+# ===== 京东联盟 CPS =====
+# 京东联盟/京东宙斯开放平台创建应用后填写。AUTH_KEY 是工具商授权 key,自有应用可留空。
+JD_UNION_APP_KEY=
+JD_UNION_APP_SECRET=
+JD_UNION_SITE_ID=
+JD_UNION_AUTH_KEY=
+
 # ===== Pricebot 上游 (领券/比价业务透传目标) =====
 # 客户端调本服务的 /api/v1/coupon/step 等,我们透传到 pricebot-backend。
 # 本地开发用 localhost:8000。生产部署改成内网地址(如 http://pricebot.internal:8000)。

alembic/versions/jd_cps_order_fields.py

@@ -0,0 +1,56 @@
+"""add jd cps order fields
+
+Revision ID: jd_cps_order_fields
+Revises: 7db22acee504
+Create Date: 2026-06-28 20:30:00.000000
+"""
+from __future__ import annotations
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision = "jd_cps_order_fields"
+down_revision = "7db22acee504"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    with op.batch_alter_table("cps_order") as batch_op:
+        batch_op.add_column(
+            sa.Column("platform", sa.String(length=20), nullable=False, server_default="meituan")
+        )
+        batch_op.add_column(sa.Column("external_order_id", sa.String(length=128), nullable=True))
+        batch_op.add_column(sa.Column("external_row_id", sa.String(length=128), nullable=True))
+        batch_op.add_column(sa.Column("estimated_commission_cents", sa.Integer(), nullable=True))
+        batch_op.add_column(sa.Column("actual_commission_cents", sa.Integer(), nullable=True))
+        batch_op.add_column(sa.Column("jd_valid_code", sa.String(length=16), nullable=True))
+        batch_op.add_column(sa.Column("settle_month", sa.String(length=16), nullable=True))
+        batch_op.add_column(sa.Column("site_id", sa.String(length=128), nullable=True))
+        batch_op.add_column(sa.Column("position_id", sa.String(length=128), nullable=True))
+        batch_op.add_column(sa.Column("pid", sa.String(length=128), nullable=True))
+        batch_op.add_column(sa.Column("sub_union_id", sa.String(length=128), nullable=True))
+        batch_op.create_index("ix_cps_order_platform", ["platform"])
+        batch_op.create_index("ix_cps_order_external_order_id", ["external_order_id"])
+        batch_op.create_index("ix_cps_order_external_row_id", ["external_row_id"])
+        batch_op.create_index("ix_cps_order_jd_valid_code", ["jd_valid_code"])
+
+
+def downgrade() -> None:
+    with op.batch_alter_table("cps_order") as batch_op:
+        batch_op.drop_index("ix_cps_order_jd_valid_code")
+        batch_op.drop_index("ix_cps_order_external_row_id")
+        batch_op.drop_index("ix_cps_order_external_order_id")
+        batch_op.drop_index("ix_cps_order_platform")
+        batch_op.drop_column("sub_union_id")
+        batch_op.drop_column("pid")
+        batch_op.drop_column("position_id")
+        batch_op.drop_column("site_id")
+        batch_op.drop_column("settle_month")
+        batch_op.drop_column("jd_valid_code")
+        batch_op.drop_column("actual_commission_cents")
+        batch_op.drop_column("estimated_commission_cents")
+        batch_op.drop_column("external_row_id")
+        batch_op.drop_column("external_order_id")
+        batch_op.drop_column("platform")

app/admin/repositories/cps.py

@@ -7,13 +7,14 @@ from __future__ import annotations
 
 from datetime import datetime, timedelta, timezone
 from decimal import Decimal, InvalidOperation
+from typing import Any
 from uuid import uuid4
 
 from sqlalchemy import desc, func, select
 from sqlalchemy.orm import Session
 
 from app.admin.repositories.queries import _as_utc, offset_paginate
-from app.integrations import meituan
+from app.integrations import jd_union, meituan
 from app.repositories import cps_link as cps_link_repo
 from app.models.cps_activity import CpsActivity
 from app.models.cps_group import CpsGroup
@@ -24,6 +25,11 @@ from app.models.cps_wx_user import CpsWxUser
 # 美团订单状态:取消(4)/风控(5)不计佣金;结算(6)为佣金真正到账
 _INVALID_STATUS = {"4", "5"}
 _SETTLED_STATUS = "6"
+_JD_INVALID_CODES = {
+    "2", "3", "4", "5", "6", "7", "8", "9", "11", "13", "14", "19", "20", "21",
+    "22", "23", "25", "26", "27", "28", "29", "30", "31", "34", "35", "36",
+}
+_JD_UNPAID_CODES = {"15"}
 
 # CPS 点击时序按北京时区分桶(运营看的是北京时间)
 _BJ_TZ = timezone(timedelta(hours=8))
@@ -47,6 +53,36 @@ def _ts_to_dt(ts: object) -> datetime | None:
     """秒级时间戳 → tz-aware UTC datetime(绝对时刻,前端按北京展示)。"""
     if not ts:
         return None
+
+
+def _jd_dt_to_utc(value: object) -> datetime | None:
+    """京东时间字符串(北京时间) → UTC aware datetime。"""
+    if value is None:
+        return None
+    s = str(value).strip()
+    if not s:
+        return None
+    for fmt in ("%Y-%m-%d %H:%M:%S", "%Y-%m-%d"):
+        try:
+            dt = datetime.strptime(s, fmt)
+            return dt.replace(tzinfo=_BJ_TZ).astimezone(timezone.utc)
+        except ValueError:
+            continue
+    return None
+
+
+def _text(value: object) -> str | None:
+    if value is None:
+        return None
+    s = str(value).strip()
+    return s or None
+
+
+def _pick(row: dict[str, Any], *keys: str) -> Any:
+    for key in keys:
+        if key in row and row[key] is not None:
+            return row[key]
+    return None
     try:
         return datetime.fromtimestamp(int(ts), tz=timezone.utc)
     except (ValueError, OSError, TypeError):
@@ -249,6 +285,80 @@ def _map_order_fields(r: dict) -> dict:
     }
 
 
+def _jd_order_key(r: dict[str, Any]) -> str | None:
+    row_id = _text(_pick(r, "id", "rowId", "orderRowId"))
+    if row_id:
+        return f"jd:{row_id}"
+    order_id = _text(_pick(r, "orderId", "parentOrderId"))
+    sku_id = _text(_pick(r, "skuId"))
+    if order_id and sku_id:
+        return f"jd:{order_id}:{sku_id}"
+    if order_id:
+        return f"jd:{order_id}"
+    return None
+
+
+def _map_jd_order_fields(r: dict[str, Any]) -> dict:
+    """京东 order.row.query 单条订单行 → CpsOrder 字段。"""
+    sku_name = _text(_pick(r, "skuName", "goodsName", "productName"))
+    if sku_name and len(sku_name) > 500:
+        sku_name = sku_name[:500]
+    valid_code = _text(_pick(r, "validCode", "valid_code"))
+    actual_fee = _yuan_to_cents(_pick(r, "actualFee", "actual_fee"))
+    estimate_fee = _yuan_to_cents(_pick(r, "estimateFee", "estimate_fee"))
+    commission = actual_fee if actual_fee not in (None, 0) else estimate_fee
+    order_time = _jd_dt_to_utc(_pick(r, "orderTime", "order_time"))
+    return {
+        "platform": "jd",
+        "external_order_id": _text(_pick(r, "orderId", "parentOrderId")),
+        "external_row_id": _text(_pick(r, "id", "rowId", "orderRowId")),
+        "sid": _text(_pick(r, "subUnionId", "sub_union_id")),
+        "act_id": None,
+        "biz_line": None,
+        "trade_type": None,
+        "pay_price_cents": _yuan_to_cents(
+            _pick(r, "actualCosPrice", "estimateCosPrice", "price")
+        ),
+        "commission_cents": commission,
+        "commission_rate": _text(_pick(r, "commissionRate", "commission_rate")),
+        "refund_price_cents": None,
+        "refund_profit_cents": None,
+        "estimated_commission_cents": estimate_fee,
+        "actual_commission_cents": actual_fee,
+        "mt_status": None,
+        "jd_valid_code": valid_code,
+        "invalid_reason": None if _is_jd_valid_code(valid_code) else f"validCode={valid_code}",
+        "product_name": sku_name,
+        "settle_month": _text(_pick(r, "payMonth", "settleMonth", "pay_month")),
+        "site_id": _text(_pick(r, "siteId", "site_id")),
+        "position_id": _text(_pick(r, "positionId", "position_id")),
+        "pid": _text(_pick(r, "pid")),
+        "sub_union_id": _text(_pick(r, "subUnionId", "sub_union_id")),
+        "pay_time": order_time,
+        "mt_update_time": _jd_dt_to_utc(_pick(r, "modifyTime", "updateTime", "modify_time"))
+        or order_time,
+        "raw": r,
+    }
+
+
+def _is_jd_valid_code(valid_code: str | None) -> bool:
+    code = str(valid_code).strip() if valid_code is not None else ""
+    return bool(code and code not in _JD_INVALID_CODES and code not in _JD_UNPAID_CODES)
+
+
+def is_jd_order_valid(order: CpsOrder) -> bool:
+    return _is_jd_valid_code(order.jd_valid_code)
+
+
+def effective_commission_cents(order: CpsOrder) -> int:
+    if order.platform == "jd":
+        if order.actual_commission_cents not in (None, 0):
+            return order.actual_commission_cents or 0
+        if order.estimated_commission_cents is not None:
+            return order.estimated_commission_cents or 0
+    return order.commission_cents or 0
+
+
 def reconcile_orders(
     db: Session, *, start_time: int, end_time: int,
     query_time_type: int = 1, sid: str | None = None, max_pages: int = 200,
@@ -274,6 +384,11 @@ def reconcile_orders(
                 continue
             fetched += 1
             fields = _map_order_fields(r)
+            fields.setdefault("platform", "meituan")
+            fields.setdefault("external_order_id", order_id)
+            fields.setdefault("external_row_id", None)
+            fields.setdefault("estimated_commission_cents", fields.get("commission_cents"))
+            fields.setdefault("actual_commission_cents", None)
             existing = db.execute(
                 select(CpsOrder).where(CpsOrder.order_id == order_id)
             ).scalar_one_or_none()
@@ -291,6 +406,56 @@ def reconcile_orders(
     return {"fetched": fetched, "inserted": inserted, "updated": updated, "pages": pages}
 
 
+def reconcile_jd_orders(
+    db: Session, *, start_time: datetime, end_time: datetime,
+    query_time_type: int = 3, max_pages: int = 100,
+) -> dict:
+    """调京东 order.row.query 拉单 → 按订单行 upsert。
+
+    京东单次查询窗口最多 1 小时,这里按北京自然时间切窗并逐页拉取。
+    """
+    fetched = inserted = updated = pages = 0
+    cur = start_time
+    while cur < end_time:
+        win_end = min(cur + timedelta(hours=1), end_time)
+        page = 1
+        while page <= max_pages:
+            resp = jd_union.query_order_rows(
+                start_time=cur,
+                end_time=win_end,
+                query_time_type=query_time_type,
+                page_index=page,
+                page_size=200,
+            )
+            rows = resp.get("rows") or []
+            has_more = bool(resp.get("has_more"))
+            if not rows:
+                break
+            pages += 1
+            for r in rows:
+                order_id = _jd_order_key(r)
+                if not order_id:
+                    continue
+                fetched += 1
+                fields = _map_jd_order_fields(r)
+                existing = db.execute(
+                    select(CpsOrder).where(CpsOrder.order_id == order_id)
+                ).scalar_one_or_none()
+                if existing is None:
+                    db.add(CpsOrder(order_id=order_id, **fields))
+                    inserted += 1
+                else:
+                    for k, v in fields.items():
+                        setattr(existing, k, v)
+                    updated += 1
+            if not has_more or len(rows) < 200:
+                break
+            page += 1
+        cur = win_end
+    db.commit()
+    return {"fetched": fetched, "inserted": inserted, "updated": updated, "pages": pages}
+
+
 def list_orders(
     db: Session, *, sid: str | None = None, mt_status: str | None = None,
     limit: int = 20, cursor: int | None = None,

app/admin/repositories/stats.py

@@ -5,7 +5,8 @@ user.last_login_at / comparison_record.status / withdraw_order.status)要加索
 """
 from __future__ import annotations
 
-from datetime import datetime, timedelta, timezone
+from datetime import date, datetime, time, timedelta, timezone
+from decimal import Decimal, InvalidOperation
 
 from sqlalchemy import func, select
 from sqlalchemy.orm import Session
@@ -13,12 +14,32 @@ from sqlalchemy.orm import Session
 from app.models.ad_feed_reward import AdFeedRewardRecord
 from app.models.ad_reward import AdRewardRecord
 from app.models.comparison import ComparisonRecord
+from app.models.coupon_state import CouponPromptEngagement
+from app.models.cps_order import CpsOrder
 from app.models.feedback import Feedback
+from app.models.savings import SavingsRecord
 from app.models.signin import SigninBoostRecord, SigninRecord
 from app.models.user import User
 from app.models.wallet import CoinTransaction, WithdrawOrder
 
 _BEIJING = timezone(timedelta(hours=8))
+COUPON_REWARD_BIZ_TYPES = ("reward_video", "ad_reward", "coupon", "coupon_reward")
+COMPARISON_REWARD_BIZ_TYPES = ("comparison", "compare_reward", "comparison_reward")
+EXCLUDED_REWARD_BIZ_TYPES = ("invite_inviter", "invite_invitee", "admin_grant")
+UNCLASSIFIED_FEED_BIZ_TYPES = ("feed_ad_reward",)
+REGULAR_TASK_EXCLUDED_BIZ_TYPES = (
+    *COUPON_REWARD_BIZ_TYPES,
+    *COMPARISON_REWARD_BIZ_TYPES,
+    *EXCLUDED_REWARD_BIZ_TYPES,
+    *UNCLASSIFIED_FEED_BIZ_TYPES,
+)
+MEITUAN_CPS_INVALID_STATUSES = ("4", "5")
+MEITUAN_CPS_SETTLED_STATUS = "6"
+JD_CPS_INVALID_CODES = {
+    "2", "3", "4", "5", "6", "7", "8", "9", "11", "13", "14", "19", "20", "21",
+    "22", "23", "25", "26", "27", "28", "29", "30", "31", "34", "35", "36",
+}
+JD_CPS_UNPAID_CODES = {"15"}
 
 
 def _beijing_today_start_utc() -> datetime:
@@ -31,30 +52,101 @@ def _beijing_today_start_utc() -> datetime:
 def today_dau(db: Session) -> int:
     """今日活跃用户数(DAU):北京时区今天 0 点后登录过(last_login_at)。
 
-    大盘与广告收益报表共用此口径,单一来源避免漂移。
-    ⚠️ last_login_at 是单值字段(只存最后一次登录时刻),故只能算「今日」,
-    无法回溯历史某天的 DAU——调用方按此约束决定历史区间是否展示。
+    广告收益报表复用这个函数;历史窗口 DAU 由 dashboard_overview 的 period 口径另算。
     """
     today_start = _beijing_today_start_utc()
-    return db.execute(
-        select(func.count(User.id)).where(User.last_login_at >= today_start)
-    ).scalar_one()
+    return int(
+        db.execute(select(func.count(User.id)).where(User.last_login_at >= today_start)).scalar_one()
+    )
 
 
-def dashboard_overview(db: Session) -> dict:
+def _default_period_end() -> date:
+    """新版大盘不含今日,默认窗口结束日=北京时间昨天。"""
+    return datetime.now(_BEIJING).date() - timedelta(days=1)
+
+
+def _normalize_period(date_from: date | None, date_to: date | None) -> tuple[date, date]:
+    end = date_to or _default_period_end()
+    start = date_from or end
+    if start > end:
+        start, end = end, start
+    return start, end
+
+
+def _period_bounds(date_from: date, date_to: date) -> tuple[datetime, datetime, datetime, datetime]:
+    """返回同一北京自然日窗口的 UTC aware 边界和北京 naive 边界。
+
+    user.created_at / last_login_at 是 UTC aware 口径;比较/金币等历史上有北京 naive
+    写入,所以两套边界同时保留。
+    """
+    start_bj = datetime.combine(date_from, time.min, tzinfo=_BEIJING)
+    end_bj = datetime.combine(date_to + timedelta(days=1), time.min, tzinfo=_BEIJING)
+    start_utc = start_bj.astimezone(timezone.utc)
+    end_utc = end_bj.astimezone(timezone.utc)
+    return (
+        start_utc,
+        end_utc,
+        start_bj.replace(tzinfo=None),
+        end_bj.replace(tzinfo=None),
+    )
+
+
+def _date_range(date_from: date, date_to: date) -> list[date]:
+    days = (date_to - date_from).days
+    return [date_from + timedelta(days=i) for i in range(days + 1)]
+
+
+def _commission_rate_percent(raw: str | None) -> Decimal | None:
+    """美团 commissionRate 原值: "300"=3%, "10"=0.1%;也兼容 "3%"。"""
+    if raw is None:
+        return None
+    s = str(raw).strip()
+    if not s:
+        return None
+    try:
+        if s.endswith("%"):
+            return Decimal(s[:-1])
+        val = Decimal(s)
+    except (InvalidOperation, ValueError):
+        return None
+    return val / Decimal("100")
+
+
+def _jd_valid_order(order: CpsOrder) -> bool:
+    code = str(order.jd_valid_code).strip() if order.jd_valid_code is not None else ""
+    return bool(code and code not in JD_CPS_INVALID_CODES and code not in JD_CPS_UNPAID_CODES)
+
+
+def _effective_commission_cents(order: CpsOrder) -> int:
+    if order.platform == "jd":
+        if order.actual_commission_cents not in (None, 0):
+            return order.actual_commission_cents or 0
+        if order.estimated_commission_cents is not None:
+            return order.estimated_commission_cents or 0
+    return order.commission_cents or 0
+
+
+def dashboard_overview(
+    db: Session, *, date_from: date | None = None, date_to: date | None = None
+) -> dict:
     today_start = _beijing_today_start_utc()
+    period_from, period_to = _normalize_period(date_from, date_to)
+    start_utc, end_utc, start_local, end_local = _period_bounds(period_from, period_to)
 
     def _count(model, *conds) -> int:
         stmt = select(func.count(model.id))
         if conds:
             stmt = stmt.where(*conds)
-        return db.execute(stmt).scalar_one()
+        return int(db.execute(stmt).scalar_one())
 
     def _sum(col, *conds) -> int:
         stmt = select(func.coalesce(func.sum(col), 0))
         if conds:
             stmt = stmt.where(*conds)
-        return db.execute(stmt).scalar_one()
+        return int(db.execute(stmt).scalar_one())
+
+    def _user_id_set(stmt) -> set[int]:
+        return {int(v) for v in db.execute(stmt).scalars().all() if v is not None}
 
     # ===== 用户 =====
     by_status = dict(
@@ -74,6 +166,212 @@ def dashboard_overview(db: Session) -> dict:
     comparison_total = _count(ComparisonRecord)
     comparison_success = _count(ComparisonRecord, ComparisonRecord.status == "success")
     success_rate = round(comparison_success / comparison_total, 4) if comparison_total else 0.0
+    period_comparison_conds = (
+        ComparisonRecord.created_at >= start_local,
+        ComparisonRecord.created_at < end_local,
+    )
+    period_comparison_total = _count(ComparisonRecord, *period_comparison_conds)
+    period_comparison_success = _count(
+        ComparisonRecord,
+        *period_comparison_conds,
+        ComparisonRecord.status == "success",
+    )
+    period_comparison_success_rate = (
+        round(period_comparison_success / period_comparison_total, 4)
+        if period_comparison_total
+        else 0.0
+    )
+    period_saved_positive_count = _count(
+        ComparisonRecord,
+        *period_comparison_conds,
+        ComparisonRecord.status == "success",
+        ComparisonRecord.saved_amount_cents > 0,
+    )
+    period_saved_positive_sum = _sum(
+        ComparisonRecord.saved_amount_cents,
+        *period_comparison_conds,
+        ComparisonRecord.status == "success",
+        ComparisonRecord.saved_amount_cents > 0,
+    )
+    period_avg_saved_cents = (
+        round(period_saved_positive_sum / period_saved_positive_count)
+        if period_saved_positive_count
+        else None
+    )
+    period_avg_duration_ms = db.execute(
+        select(func.avg(ComparisonRecord.total_ms)).where(
+            *period_comparison_conds,
+            ComparisonRecord.total_ms.is_not(None),
+            ComparisonRecord.total_ms > 0,
+        )
+    ).scalar_one()
+    period_avg_duration_ms = (
+        round(float(period_avg_duration_ms))
+        if period_avg_duration_ms is not None
+        else None
+    )
+
+    ordered_exists = (
+        select(SavingsRecord.id)
+        .where(
+            SavingsRecord.user_id == ComparisonRecord.user_id,
+            SavingsRecord.source == "compare",
+            SavingsRecord.shop_name.is_not(None),
+            SavingsRecord.shop_name == ComparisonRecord.store_name,
+        )
+        .exists()
+    )
+    period_ordered_count = _count(
+        ComparisonRecord,
+        *period_comparison_conds,
+        ComparisonRecord.store_name.is_not(None),
+        ordered_exists,
+    )
+
+    # ===== 日期窗口用户 =====
+    period_new_user_ids = _user_id_set(
+        select(User.id).where(User.created_at >= start_utc, User.created_at < end_utc)
+    )
+    login_user_ids = _user_id_set(
+        select(User.id).where(User.last_login_at >= start_utc, User.last_login_at < end_utc)
+    )
+    compare_user_ids = _user_id_set(
+        select(ComparisonRecord.user_id).where(*period_comparison_conds)
+    )
+    coupon_user_ids = _user_id_set(
+        select(CouponPromptEngagement.user_id).where(
+            CouponPromptEngagement.engage_date >= period_from,
+            CouponPromptEngagement.engage_date <= period_to,
+            CouponPromptEngagement.engage_type == "claim_started",
+        )
+    )
+    period_active_user_ids = login_user_ids | compare_user_ids | coupon_user_ids
+    period_retained_new_user_ids = period_new_user_ids & period_active_user_ids
+    period_retention_rate = (
+        round(len(period_retained_new_user_ids) / len(period_new_user_ids), 4)
+        if period_new_user_ids
+        else None
+    )
+    trend_points: list[dict] = []
+    for cur_date in _date_range(period_from, period_to):
+        day_start_utc, day_end_utc, day_start_local, day_end_local = _period_bounds(
+            cur_date, cur_date
+        )
+        daily_comparison_conds = (
+            ComparisonRecord.created_at >= day_start_local,
+            ComparisonRecord.created_at < day_end_local,
+        )
+        daily_login_user_ids = _user_id_set(
+            select(User.id).where(
+                User.last_login_at >= day_start_utc,
+                User.last_login_at < day_end_utc,
+            )
+        )
+        daily_compare_user_ids = _user_id_set(
+            select(ComparisonRecord.user_id).where(*daily_comparison_conds)
+        )
+        daily_coupon_user_ids = _user_id_set(
+            select(CouponPromptEngagement.user_id).where(
+                CouponPromptEngagement.engage_date == cur_date,
+                CouponPromptEngagement.engage_type == "claim_started",
+            )
+        )
+        trend_points.append(
+            {
+                "date": cur_date,
+                "active_users": len(
+                    daily_login_user_ids | daily_compare_user_ids | daily_coupon_user_ids
+                ),
+                "new_users": _count(
+                    User,
+                    User.created_at >= day_start_utc,
+                    User.created_at < day_end_utc,
+                ),
+                "comparisons": _count(ComparisonRecord, *daily_comparison_conds),
+            }
+        )
+
+    period_coin_conds = (
+        CoinTransaction.created_at >= start_local,
+        CoinTransaction.created_at < end_local,
+        CoinTransaction.amount > 0,
+    )
+    period_reward_video_coin_total = _sum(
+        CoinTransaction.amount,
+        *period_coin_conds,
+        CoinTransaction.biz_type.in_(("reward_video", "ad_reward")),
+    )
+    period_feed_ad_coin_total = _sum(
+        CoinTransaction.amount,
+        *period_coin_conds,
+        CoinTransaction.biz_type == "feed_ad_reward",
+    )
+    period_signin_coin_total = _sum(
+        CoinTransaction.amount,
+        *period_coin_conds,
+        CoinTransaction.biz_type == "signin",
+    )
+    period_signin_boost_coin_total = _sum(
+        CoinTransaction.amount,
+        *period_coin_conds,
+        CoinTransaction.biz_type == "signin_boost",
+    )
+    period_task_coin_total = _sum(
+        CoinTransaction.amount,
+        *period_coin_conds,
+        CoinTransaction.biz_type.like("task_%"),
+    )
+    period_coupon_reward_coin_total = _sum(
+        CoinTransaction.amount,
+        *period_coin_conds,
+        CoinTransaction.biz_type.in_(COUPON_REWARD_BIZ_TYPES),
+    )
+    period_comparison_reward_coin_total = _sum(
+        CoinTransaction.amount,
+        *period_coin_conds,
+        CoinTransaction.biz_type.in_(COMPARISON_REWARD_BIZ_TYPES),
+    )
+    period_regular_task_coin_total = _sum(
+        CoinTransaction.amount,
+        *period_coin_conds,
+        CoinTransaction.biz_type.notin_(REGULAR_TASK_EXCLUDED_BIZ_TYPES),
+    )
+    period_cps_orders = list(
+        db.execute(
+            select(CpsOrder).where(
+                CpsOrder.pay_time >= start_utc,
+                CpsOrder.pay_time < end_utc,
+            )
+        ).scalars()
+    )
+    period_meituan_orders = [
+        o for o in period_cps_orders if (o.platform or "meituan") == "meituan"
+    ]
+    period_meituan_valid_orders = [
+        o for o in period_meituan_orders if o.mt_status not in MEITUAN_CPS_INVALID_STATUSES
+    ]
+    period_jd_orders = [o for o in period_cps_orders if o.platform == "jd"]
+    period_jd_valid_orders = [o for o in period_jd_orders if _jd_valid_order(o)]
+    period_jd_invalid_orders = [
+        o for o in period_jd_orders if o.jd_valid_code and not _jd_valid_order(o)
+    ]
+    period_meituan_hit_count = 0
+    period_meituan_miss_count = 0
+    period_meituan_unknown_rate_count = 0
+    for order in period_meituan_valid_orders:
+        rate = _commission_rate_percent(order.commission_rate)
+        if rate is None:
+            period_meituan_unknown_rate_count += 1
+        elif rate < Decimal("1"):
+            period_meituan_miss_count += 1
+        else:
+            period_meituan_hit_count += 1
+    period_meituan_hit_denominator = period_meituan_hit_count + period_meituan_miss_count
+    period_meituan_hit_rate = (
+        round(period_meituan_hit_count / period_meituan_hit_denominator, 4)
+        if period_meituan_hit_denominator
+        else None
+    )
 
     return {
         "users": {
@@ -132,7 +430,70 @@ def dashboard_overview(db: Session) -> dict:
             "success": comparison_success,
             "success_rate": success_rate,
         },
-        "feedback": {"new": _count(Feedback, Feedback.status.in_(("pending", "new")))},
-        # CPS 收入数据源未接(referral-link 只换链接,转化/佣金未回收)→ 前端显示"待接入"。
-        "cps": {"available": False, "note": "CPS 转化数据未接入(P2)"},
+        "period": {
+            "date_from": period_from,
+            "date_to": period_to,
+            "users": {
+                "new": len(period_new_user_ids),
+                "active": len(period_active_user_ids),
+                "retained_new_users": len(period_retained_new_user_ids),
+                "retention_rate": period_retention_rate,
+                "retention_note": (
+                    "近似口径:登录(last_login_at)+已上报比价记录+领券claim_started;"
+                    "尚不包含未完成上报的比价开始事件"
+                ),
+            },
+            "comparison": {
+                "total": period_comparison_total,
+                "success": period_comparison_success,
+                "success_rate": period_comparison_success_rate,
+                "ordered": period_ordered_count,
+                "average_duration_ms": period_avg_duration_ms,
+                "average_saved_cents": period_avg_saved_cents,
+            },
+            "coins": {
+                "granted_total": _sum(CoinTransaction.amount, *period_coin_conds),
+                "reward_video_coin_total": period_reward_video_coin_total,
+                "feed_ad_coin_total": period_feed_ad_coin_total,
+                "signin_coin_total": period_signin_coin_total,
+                "signin_boost_coin_total": period_signin_boost_coin_total,
+                "task_coin_total": period_task_coin_total,
+                "coupon_reward_coin_total": period_coupon_reward_coin_total,
+                "comparison_reward_coin_total": period_comparison_reward_coin_total,
+                "regular_task_coin_total": period_regular_task_coin_total,
+            },
+            "cash": {
+                "withdraw_success_cents": _sum(
+                    WithdrawOrder.amount_cents,
+                    WithdrawOrder.status == "success",
+                    WithdrawOrder.created_at >= start_local,
+                    WithdrawOrder.created_at < end_local,
+                ),
+            },
+            "trend": trend_points,
+        },
+        "feedback": {
+            "new": _count(Feedback, Feedback.status.in_(("pending", "new"))),
+        },
+        "cps": {
+            "available": True,
+            "note": "美团/JD CPS 读 cps_order 对账订单;淘宝佣金暂空",
+            "meituan_order_count": len(period_meituan_valid_orders),
+            "meituan_commission_cents": sum(
+                o.commission_cents or 0 for o in period_meituan_valid_orders
+            ),
+            "meituan_hit_count": period_meituan_hit_count,
+            "meituan_miss_count": period_meituan_miss_count,
+            "meituan_unknown_rate_count": period_meituan_unknown_rate_count,
+            "meituan_hit_rate": period_meituan_hit_rate,
+            "jd_order_count": len(period_jd_valid_orders),
+            "jd_commission_cents": sum(_effective_commission_cents(o) for o in period_jd_valid_orders),
+            "jd_actual_commission_cents": sum(
+                o.actual_commission_cents or 0 for o in period_jd_valid_orders
+            ),
+            "jd_estimated_commission_cents": sum(
+                o.estimated_commission_cents or 0 for o in period_jd_valid_orders
+            ),
+            "jd_invalid_count": len(period_jd_invalid_orders),
+        },
     }

app/admin/routers/cps.py

@@ -1,13 +1,13 @@
-"""admin CPS 分发与对账:群/活动管理 + 生成落地页短链 + 美团订单对账 + 统计。
+"""admin CPS 分发与对账:群/活动管理 + 生成落地页短链 + 联盟订单对账 + 统计。
 
 平台:meituan(actId+sid 转链 + query_order 对账) / taobao(整段淘口令) / jd(链接)。
-淘宝/京东无 API → 只统计点击(咱落地页 PV/UV + 淘宝复制),对账字段显示 "-"。
+淘宝暂未接 API → 只统计点击(咱落地页 PV/UV + 淘宝复制),对账字段显示 "-"。
 群/活动管理 = operator;订单对账(涉佣金) = finance;只读列表/统计 = 登录即可。
 """
 from __future__ import annotations
 
 import time
-from datetime import datetime, timedelta, timezone
+from datetime import date as _date, datetime, time as _dt_time, timedelta, timezone
 from typing import Annotated
 
 from fastapi import APIRouter, Depends, File, HTTPException, Query, Request, UploadFile
@@ -34,6 +34,7 @@ from app.admin.schemas.cps import (
 from app.core import media
 from app.core.config import settings
 from app.integrations import meituan
+from app.integrations.jd_union import JdUnionError
 from app.integrations.meituan import MeituanCpsError
 from app.models.admin import AdminUser
 from app.models.cps_activity import CpsActivity
@@ -340,24 +341,108 @@ def generate_referral_links(
 
 
 # ───────────── 订单对账 ─────────────
-@router.post("/orders/reconcile", response_model=CpsReconcileResult, summary="拉取美团订单对账")
+_BEIJING = timezone(timedelta(hours=8))
+
+
+def _parse_day(value: str | None, *, field: str) -> _date | None:
+    if value is None:
+        return None
+    try:
+        return _date.fromisoformat(value)
+    except ValueError as e:
+        raise HTTPException(status_code=422, detail=f"{field} 需为 YYYY-MM-DD") from e
+
+
+def _reconcile_range_to_ts(
+    date_from: _date | None, date_to: _date | None, days: int
+) -> tuple[int, int]:
+    if date_from is None and date_to is None:
+        now = int(time.time())
+        return now - days * 86400, now
+
+    start_day = date_from or date_to
+    end_day = date_to or date_from
+    if start_day is None or end_day is None:
+        raise HTTPException(status_code=422, detail="日期参数不完整")
+    if start_day > end_day:
+        start_day, end_day = end_day, start_day
+    if (end_day - start_day).days + 1 > 90:
+        raise HTTPException(status_code=422, detail="美团订单查询最长 90 天")
+
+    start_dt = datetime.combine(start_day, _dt_time.min, tzinfo=_BEIJING)
+    end_dt = datetime.combine(end_day + timedelta(days=1), _dt_time.min, tzinfo=_BEIJING)
+    return int(start_dt.timestamp()), int(end_dt.timestamp())
+
+
+def _reconcile_range_to_bj_dt(
+    date_from: _date | None, date_to: _date | None, days: int
+) -> tuple[datetime, datetime]:
+    start_ts, end_ts = _reconcile_range_to_ts(date_from, date_to, days)
+    return (
+        datetime.fromtimestamp(start_ts, tz=_BEIJING),
+        datetime.fromtimestamp(end_ts, tz=_BEIJING),
+    )
+
+
+def _merge_reconcile_result(total: dict, current: dict) -> None:
+    for key in ("fetched", "inserted", "updated", "pages"):
+        total[key] = int(total.get(key, 0)) + int(current.get(key, 0))
+
+
+@router.post("/orders/reconcile", response_model=CpsReconcileResult, summary="拉取联盟订单对账")
 def reconcile_orders(
     request: Request,
     admin: Annotated[AdminUser, Depends(require_role("finance"))],
     db: AdminDb,
-    days: Annotated[int, Query(ge=1, le=90)] = 7,
+    date_from: Annotated[str | None, Query(description="起始日 YYYY-MM-DD")] = None,
+    date_to: Annotated[str | None, Query(description="结束日 YYYY-MM-DD")] = None,
+    days: Annotated[int, Query(ge=1, le=90, description="未传日期时默认拉近 N 天")] = 7,
     sid: Annotated[str | None, Query(max_length=64)] = None,
+    query_time_type: Annotated[int, Query(ge=1, le=3)] = 1,
+    platform: Annotated[str, Query(pattern="^(all|meituan|jd)$")] = "all",
 ) -> CpsReconcileResult:
-    now = int(time.time())
+    parsed_from = _parse_day(date_from, field="date_from")
+    parsed_to = _parse_day(date_to, field="date_to")
+    result = {"fetched": 0, "inserted": 0, "updated": 0, "pages": 0}
     try:
-        result = cps_repo.reconcile_orders(
-            db, start_time=now - days * 86400, end_time=now, sid=sid,
-        )
+        if platform in {"all", "meituan"}:
+            start_ts, end_ts = _reconcile_range_to_ts(parsed_from, parsed_to, days)
+            mt_result = cps_repo.reconcile_orders(
+                db,
+                start_time=start_ts,
+                end_time=end_ts,
+                query_time_type=query_time_type if query_time_type in (1, 2) else 2,
+                sid=sid,
+            )
+            _merge_reconcile_result(result, mt_result)
+        if platform in {"all", "jd"}:
+            if sid:
+                raise HTTPException(status_code=422, detail="京东订单刷新不支持 sid 筛选")
+            start_dt, end_dt = _reconcile_range_to_bj_dt(parsed_from, parsed_to, days)
+            jd_result = cps_repo.reconcile_jd_orders(
+                db,
+                start_time=start_dt,
+                end_time=end_dt,
+                query_time_type=query_time_type,
+            )
+            _merge_reconcile_result(result, jd_result)
     except MeituanCpsError as e:
         raise HTTPException(status_code=502, detail=f"美团拉单失败: {e}") from e
+    except JdUnionError as e:
+        raise HTTPException(status_code=502, detail=f"京东拉单失败: {e}") from e
     write_audit(
         db, admin, action="cps.order.reconcile", target_type="cps_order", target_id=None,
-        detail={"days": days, "sid": sid, **result}, ip=get_client_ip(request), commit=True,
+        detail={
+            "platform": platform,
+            "date_from": date_from,
+            "date_to": date_to,
+            "days": days,
+            "sid": sid,
+            "query_time_type": query_time_type,
+            **result,
+        },
+        ip=get_client_ip(request),
+        commit=True,
     )
     return CpsReconcileResult(**result)
 

app/admin/routers/dashboard.py

@@ -1,7 +1,9 @@
 """admin 数据大盘(只读聚合)。"""
 from __future__ import annotations
 
-from fastapi import APIRouter, Depends
+from datetime import date
+
+from fastapi import APIRouter, Depends, Query
 
 from app.admin.deps import AdminDb, get_current_admin
 from app.admin.repositories import stats
@@ -15,5 +17,11 @@ router = APIRouter(
 
 
 @router.get("/overview", response_model=DashboardOverview, summary="大盘核心指标")
-def overview(db: AdminDb) -> DashboardOverview:
-    return DashboardOverview.model_validate(stats.dashboard_overview(db))
+def overview(
+    db: AdminDb,
+    date_from: date | None = Query(None, description="北京时间自然日起始日 YYYY-MM-DD"),
+    date_to: date | None = Query(None, description="北京时间自然日结束日 YYYY-MM-DD"),
+) -> DashboardOverview:
+    return DashboardOverview.model_validate(
+        stats.dashboard_overview(db, date_from=date_from, date_to=date_to)
+    )

app/admin/schemas/cps.py

@@ -1,7 +1,7 @@
 """admin CPS 分发与对账 schemas。金额统一「分」(cents),前端 yuan() 展示。
 
-平台:meituan(actId+sid 转链对账) / taobao(淘口令,只统计点击) / jd(链接,只统计点击)。
-对账类字段对淘宝/京东为 None → 前端显示 "-"(无法对账)。
+平台:meituan(actId+sid 转链对账) / taobao(淘口令,只统计点击) / jd(链接 + 订单 API 对账)。
+对账类字段对淘宝为 None → 前端显示 "-"(暂未对账)。
 """
 from __future__ import annotations
 
@@ -116,15 +116,22 @@ class CpsOrderOut(BaseModel):
     model_config = ConfigDict(from_attributes=True)
 
     id: int
+    platform: str = "meituan"
     order_id: str
+    external_order_id: str | None = None
+    external_row_id: str | None = None
     sid: str | None = None
     act_id: str | None = None
     pay_price_cents: int | None = None
     commission_cents: int | None = None
+    estimated_commission_cents: int | None = None
+    actual_commission_cents: int | None = None
     commission_rate: str | None = None
     mt_status: str | None = None
+    jd_valid_code: str | None = None
     invalid_reason: str | None = None
     product_name: str | None = None
+    settle_month: str | None = None
     pay_time: datetime | None = None
 
 

app/admin/schemas/dashboard.py

@@ -1,6 +1,8 @@
 """admin 大盘 schemas(对应 stats.dashboard_overview 的嵌套结构)。"""
 from __future__ import annotations
 
+from datetime import date
+
 from pydantic import BaseModel
 
 
@@ -38,6 +40,56 @@ class DashboardComparison(BaseModel):
     success_rate: float
 
 
+class DashboardPeriodUsers(BaseModel):
+    new: int
+    active: int
+    retained_new_users: int
+    retention_rate: float | None = None
+    retention_note: str
+
+
+class DashboardPeriodComparison(BaseModel):
+    total: int
+    success: int
+    success_rate: float
+    ordered: int
+    average_duration_ms: int | None = None
+    average_saved_cents: int | None = None
+
+
+class DashboardPeriodCoins(BaseModel):
+    granted_total: int
+    reward_video_coin_total: int = 0
+    feed_ad_coin_total: int = 0
+    signin_coin_total: int = 0
+    signin_boost_coin_total: int = 0
+    task_coin_total: int = 0
+    coupon_reward_coin_total: int = 0
+    comparison_reward_coin_total: int = 0
+    regular_task_coin_total: int = 0
+
+
+class DashboardPeriodCash(BaseModel):
+    withdraw_success_cents: int
+
+
+class DashboardTrendPoint(BaseModel):
+    date: date
+    active_users: int
+    new_users: int
+    comparisons: int
+
+
+class DashboardPeriod(BaseModel):
+    date_from: date
+    date_to: date
+    users: DashboardPeriodUsers
+    comparison: DashboardPeriodComparison
+    coins: DashboardPeriodCoins
+    cash: DashboardPeriodCash
+    trend: list[DashboardTrendPoint] = []
+
+
 class DashboardFeedback(BaseModel):
     new: int
 
@@ -45,6 +97,17 @@ class DashboardFeedback(BaseModel):
 class DashboardCps(BaseModel):
     available: bool
     note: str
+    meituan_order_count: int = 0
+    meituan_commission_cents: int = 0
+    meituan_hit_count: int = 0
+    meituan_miss_count: int = 0
+    meituan_unknown_rate_count: int = 0
+    meituan_hit_rate: float | None = None
+    jd_order_count: int = 0
+    jd_commission_cents: int = 0
+    jd_actual_commission_cents: int = 0
+    jd_estimated_commission_cents: int = 0
+    jd_invalid_count: int = 0
 
 
 class DashboardOverview(BaseModel):
@@ -52,5 +115,6 @@ class DashboardOverview(BaseModel):
     coins: DashboardCoins
     cash: DashboardCash
     comparison: DashboardComparison
+    period: DashboardPeriod
     feedback: DashboardFeedback
     cps: DashboardCps

app/core/config.py

@@ -113,6 +113,21 @@ class Settings(BaseSettings):
         """美团 CPS 凭证齐全(缺则接口返空,而非 502)。"""
         return bool(self.MT_CPS_APP_KEY and self.MT_CPS_APP_SECRET)
 
+    # ===== 京东联盟 CPS =====
+    # app_key/app_secret 来自京东联盟应用;site_id 是推广管理里的 APP/网站 ID;
+    # auth_key 是工具商授权 key,自有应用查询可留空。
+    JD_UNION_APP_KEY: str = ""
+    JD_UNION_APP_SECRET: str = ""
+    JD_UNION_SITE_ID: str = ""
+    JD_UNION_AUTH_KEY: str = ""
+    JD_UNION_GATEWAY: str = "https://api.jd.com/routerjson"
+    JD_UNION_TIMEOUT_SEC: int = 15
+
+    @property
+    def jd_union_configured(self) -> bool:
+        """京东联盟订单查询凭证齐全。"""
+        return bool(self.JD_UNION_APP_KEY and self.JD_UNION_APP_SECRET)
+
     # ===== 微信服务号(网页授权) =====
     # CPS 落地页在微信内拿用户 openid(base 静默)/昵称头像(userinfo),做用户级群统计。
     # ⚠️ 区别于 WECHAT_APP_ID(那是 App 移动应用,用于微信支付);这是【已认证服务号】。

app/integrations/jd_union.py

@@ -0,0 +1,163 @@
+"""京东联盟 OpenAPI 客户端。
+
+当前只接数据大盘需要的订单明细接口:
+`jd.union.open.order.row.query`。京东要求订单查询时间窗最长 1 小时,
+调用方负责切窗分页。
+"""
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+from datetime import datetime, timedelta, timezone
+from typing import Any
+
+import httpx
+
+from app.core.config import settings
+
+logger = logging.getLogger(__name__)
+
+_BEIJING = timezone(timedelta(hours=8))
+
+
+class JdUnionError(RuntimeError):
+    """京东联盟 API 调用失败。"""
+
+
+def _parse_json_maybe(value: Any) -> Any:
+    if not isinstance(value, str):
+        return value
+    text = value.strip()
+    if not text:
+        return value
+    try:
+        return json.loads(text)
+    except json.JSONDecodeError:
+        return value
+
+
+def _sign(params: dict[str, Any], secret: str) -> str:
+    pieces = [secret]
+    for key in sorted(k for k in params if k != "sign"):
+        value = params[key]
+        if value is None:
+            continue
+        pieces.append(f"{key}{value}")
+    pieces.append(secret)
+    raw = "".join(pieces)
+    return hashlib.md5(raw.encode("utf-8")).hexdigest().upper()
+
+
+def _unwrap_response(data: dict[str, Any]) -> dict[str, Any]:
+    if "error_response" in data:
+        err = data["error_response"] or {}
+        msg = err.get("zh_desc") or err.get("en_desc") or err.get("msg") or err
+        raise JdUnionError(f"京东 API 错误: {msg}")
+
+    body: Any = data
+    for key, value in data.items():
+        if key.endswith("_responce") or key.endswith("_response"):
+            body = value
+            break
+
+    body = _parse_json_maybe(body)
+    if not isinstance(body, dict):
+        raise JdUnionError("京东 API 返回格式异常")
+
+    result = body.get("queryResult", body.get("result", body))
+    result = _parse_json_maybe(result)
+    if not isinstance(result, dict):
+        raise JdUnionError("京东 API 业务结果格式异常")
+
+    code = str(result.get("code", result.get("resultCode", "200")))
+    if code not in {"0", "200"}:
+        msg = result.get("message") or result.get("msg") or result.get("resultMsg") or result
+        raise JdUnionError(f"京东 API 业务错误: {msg}")
+    return result
+
+
+def call(method: str, payload: dict[str, Any], *, version: str = "1.0") -> dict[str, Any]:
+    if not settings.jd_union_configured:
+        raise JdUnionError("京东联盟凭证未配置")
+
+    biz_json = json.dumps(payload, ensure_ascii=False, separators=(",", ":"))
+    params: dict[str, Any] = {
+        "method": method,
+        "app_key": settings.JD_UNION_APP_KEY,
+        "timestamp": datetime.now(_BEIJING).strftime("%Y-%m-%d %H:%M:%S"),
+        "format": "json",
+        "v": version,
+        "sign_method": "md5",
+        "360buy_param_json": biz_json,
+    }
+    params["sign"] = _sign(params, settings.JD_UNION_APP_SECRET)
+
+    try:
+        with httpx.Client(timeout=settings.JD_UNION_TIMEOUT_SEC, trust_env=False) as client:
+            resp = client.post(settings.JD_UNION_GATEWAY, data=params)
+            resp.raise_for_status()
+            data = resp.json()
+    except httpx.HTTPError as e:
+        raise JdUnionError(f"京东 API 网络错误: {e}") from e
+    except json.JSONDecodeError as e:
+        raise JdUnionError("京东 API 返回非 JSON") from e
+
+    return _unwrap_response(data)
+
+
+def _extract_rows(result: dict[str, Any]) -> tuple[list[dict[str, Any]], bool]:
+    payload = _parse_json_maybe(result.get("data", result.get("result", result)))
+    has_more = bool(result.get("hasMore") or result.get("has_more"))
+
+    if isinstance(payload, dict):
+        for key in ("orderRowResp", "orderRows", "orderList", "orders", "list", "rows"):
+            rows = _parse_json_maybe(payload.get(key))
+            if isinstance(rows, list):
+                return [r for r in rows if isinstance(r, dict)], bool(
+                    payload.get("hasMore") or payload.get("has_more") or has_more
+                )
+        return [], bool(payload.get("hasMore") or payload.get("has_more") or has_more)
+    if isinstance(payload, list):
+        return [r for r in payload if isinstance(r, dict)], has_more
+    return [], has_more
+
+
+def query_order_rows(
+    *,
+    start_time: datetime,
+    end_time: datetime,
+    query_time_type: int = 3,
+    page_index: int = 1,
+    page_size: int = 200,
+) -> dict[str, Any]:
+    """查询京东 CPS 订单行。
+
+    query_time_type: 1 下单时间, 2 完成时间, 3 更新时间。
+    start_time/end_time 用北京时间展示给京东;调用方需保证窗口不超过 1 小时。
+    """
+    start_bj = start_time.astimezone(_BEIJING)
+    end_bj = end_time.astimezone(_BEIJING)
+    if end_bj <= start_bj:
+        return {"rows": [], "has_more": False}
+    if end_bj - start_bj > timedelta(hours=1):
+        raise JdUnionError("京东订单查询单次时间窗不能超过 1 小时")
+
+    order_req: dict[str, Any] = {
+        "pageIndex": page_index,
+        "pageSize": min(max(page_size, 1), 200),
+        "type": query_time_type,
+        "startTime": start_bj.strftime("%Y-%m-%d %H:%M:%S"),
+        "endTime": end_bj.strftime("%Y-%m-%d %H:%M:%S"),
+    }
+    if settings.JD_UNION_AUTH_KEY:
+        order_req["key"] = settings.JD_UNION_AUTH_KEY
+    result = call("jd.union.open.order.row.query", {"orderReq": order_req})
+    rows, has_more = _extract_rows(result)
+    logger.info(
+        "jd.union.open.order.row.query fetched rows=%s page=%s has_more=%s",
+        len(rows),
+        page_index,
+        has_more,
+    )
+    return {"rows": rows, "has_more": has_more, "raw": result}

app/models/__init__.py

@@ -19,6 +19,7 @@ from app.models.coupon_state import (  # noqa: F401
     CouponDailyCompletion,
     CouponPromptEngagement,
 )
+from app.models.cps_order import CpsOrder  # noqa: F401
 from app.models.feedback import Feedback  # noqa: F401
 from app.models.invite import InviteRelation  # noqa: F401
 from app.models.invite_fingerprint import InviteFingerprint  # noqa: F401

app/models/cps_order.py

@@ -1,11 +1,13 @@
 """CPS 对账订单(cps_order)。
 
-从美团联盟 query_order 按时间窗拉回、按 sid 归群的订单明细。字段对齐 query_order
+从联盟 API 按时间窗拉回、按平台落库的 CPS 订单明细。字段最初对齐美团 query_order,
+后续兼容京东订单报表:
 实测返回:
   - payPrice / profit 是「元」字符串 → 入库统一转「分」(与全站口径一致)
   - payTime / updateTime 是秒级时间戳 → 入库转 tz-aware datetime
   - status: 2付款 3完成 4取消 5风控 6结算(取消/风控不计佣金)
-order_id 全局唯一,reconcile 按它 upsert(订单状态会变,重复拉则更新)。
+order_id 全局唯一,reconcile 按它 upsert(订单状态会变,重复拉则更新)。京东订单用
+`jd:<row_id>` 前缀避免与美团订单号碰撞。
 """
 from __future__ import annotations
 
@@ -22,8 +24,13 @@ class CpsOrder(Base):
     __tablename__ = "cps_order"
 
     id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True)
-    # 美团订单号(加密串),全局唯一,upsert 幂等键。
+    # 平台:meituan / jd。历史数据迁移默认 meituan。
+    platform: Mapped[str] = mapped_column(String(20), default="meituan", index=True, nullable=False)
+    # 平台订单号/行号包装后的全局唯一键,upsert 幂等。
     order_id: Mapped[str] = mapped_column(String(64), unique=True, index=True, nullable=False)
+    # 平台原始订单号/行号。京东一笔订单多 SKU 时可按行号区分。
+    external_order_id: Mapped[str | None] = mapped_column(String(128), index=True, nullable=True)
+    external_row_id: Mapped[str | None] = mapped_column(String(128), index=True, nullable=True)
     # 渠道追踪位 = 群 sid(历史无 sid 订单为空)。按它归群聚合。
     sid: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
     act_id: Mapped[str | None] = mapped_column(String(64), index=True, nullable=True)
@@ -35,11 +42,21 @@ class CpsOrder(Base):
     commission_rate: Mapped[str | None] = mapped_column(String(16), nullable=True)  # "300"=3% "10"=0.1%
     refund_price_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
     refund_profit_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    # 通用佣金拆分。美团只有预估 profit;京东有预估/实际佣金。
+    estimated_commission_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    actual_commission_cents: Mapped[int | None] = mapped_column(Integer, nullable=True)
 
     # 美团订单状态: 2付款 3完成 4取消 5风控 6结算
     mt_status: Mapped[str | None] = mapped_column(String(8), index=True, nullable=True)
+    # 京东订单有效码(validCode),用于判断是否有效/已完成。
+    jd_valid_code: Mapped[str | None] = mapped_column(String(16), index=True, nullable=True)
     invalid_reason: Mapped[str | None] = mapped_column(String(128), nullable=True)
     product_name: Mapped[str | None] = mapped_column(String(512), nullable=True)
+    settle_month: Mapped[str | None] = mapped_column(String(16), nullable=True)
+    site_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
+    position_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
+    pid: Mapped[str | None] = mapped_column(String(128), nullable=True)
+    sub_union_id: Mapped[str | None] = mapped_column(String(128), nullable=True)
 
     pay_time: Mapped[datetime | None] = mapped_column(
         DateTime(timezone=True), index=True, nullable=True
@@ -59,5 +76,6 @@ class CpsOrder(Base):
     def __repr__(self) -> str:  # pragma: no cover
         return (
             f"<CpsOrder id={self.id} order_id={self.order_id!r} "
-            f"sid={self.sid!r} status={self.mt_status} profit_cents={self.commission_cents}>"
+            f"platform={self.platform!r} sid={self.sid!r} "
+            f"status={self.mt_status or self.jd_valid_code} profit_cents={self.commission_cents}>"
         )

app/schemas/compare_record.py

@@ -123,6 +123,7 @@ class ComparisonRecordIn(BaseModel):
     # pricebot done.params.trace_url 原样上报,落库供记录页「复制调试链接」(dir 名含落盘
     # 时分秒前端拼不出,必须由后端透传)。
     trace_url: str | None = Field(None, description="本次比价公网调试链接")
+    total_ms: int | None = Field(None, description="整场比价墙钟耗时(ms)")
 
     # ===== debug 维度(客户端采集上报;旧客户端不带 → None。仅 admin 比价记录页用)=====
     # 必须显式声明,否则 model_dump() 落 raw_payload 时被 pydantic 静默丢弃(同上面 coupon_saved 的坑)。
@@ -172,6 +173,7 @@ class ComparisonRecordOut(BaseModel):
     items: list = []
     comparison_results: list = []
     skipped_dish_names: list = []
+    total_ms: int | None = None
     # 「已下单」(店级):该店名在该用户真实下单(source='compare')里出现过即 True。
     # 由 list_records 动态算出挂在 ORM 实例上(非 DB 列),from_attributes 读出;缺省 False。
     ordered: bool = False

docs/guides/邀请功能-实现原理与本地测试.md

@@ -37,7 +37,7 @@ B 安装并首启 App
         └─ POST /api/v1/invite/bind { invite_code, channel="clipboard" }
         ↓
 后端 repositories/invite.py bind()
-  └─ 过四道防线 → 建 invite_relation + 给 A、B 各发金币(同事务原子提交)
+  └─ 过四道防线 → 建 invite_relation(邀请金币已下线,不写金币流水)
 ```
 
 手动填码这条:B 在邀请页输码 → `InviteRepository.bindManual()` → `POST /bind { channel="manual" }` → 同一个 `bind()`。
@@ -52,20 +52,20 @@ B 安装并首启 App
 |---|---|
 | 端点 | `app/api/v1/invite.py`:`GET /api/v1/invite/me`(返回 `invite_code` + `share_url` + 战绩)、`POST /api/v1/invite/bind`(绑定,`channel` = `clipboard` / `manual`)。**均需 Bearer 鉴权**。 |
 | share_url 构造 | `invite.py` 的 `my_invite`:`settings.INVITE_LANDING_URL + "?ref=" + code`。`INVITE_LANDING_URL` 在 `app/core/config.py`(默认 `https://app-api.shaguabijia.com/media/dl.html`)。 |
-| 业务逻辑 | `app/repositories/invite.py`:`ensure_code`(懒生成 6 位邀请码,去混淆字符集,唯一约束碰撞则换码)/ `resolve_inviter`(邀请码→邀请人,大小写不敏感)/ `bind`(下面详述)/ `get_stats`(已邀人数 + 累计金币)。 |
+| 业务逻辑 | `app/repositories/invite.py`:`ensure_code`(懒生成 6 位邀请码,去混淆字符集,唯一约束碰撞则换码)/ `resolve_inviter`(邀请码→邀请人,大小写不敏感)/ `bind`(下面详述)/ `get_stats`(已邀人数 + 兼容累计金币字段,当前恒为 0)。 |
 | 数据模型 | `app/models/invite.py` 的 `InviteRelation`(`inviter_user_id` / `invitee_user_id` / `channel` / `status` / `inviter_coin` / `invitee_coin` / `created_at`)+ `app/models/user.py` 的 `User.invite_code` 列。 |
 | 迁移 | `alembic/versions/invite_code_and_relation.py`:给 `user` 加 `invite_code`(唯一索引)+ 建 `invite_relation` 表。`down_revision = 11a1d08c6f55`。 |
 | 收发模型 | `app/schemas/invite.py`:`InviteInfoOut` / `BindInviteIn` / `BindInviteOut`。 |
-| 奖励常量 | `app/core/rewards.py`:`INVITE_INVITER_COINS` / `INVITE_INVITEE_COINS`(各 10000 = 1 元)、`INVITE_NEW_USER_WINDOW_HOURS`(72)。 |
+| 新人窗口 | `app/core/rewards.py`:`INVITE_NEW_USER_WINDOW_HOURS`(72)。邀请金币已下线,不再配置邀请金币常量。 |
 
 **`bind()` 的四道防线(防重复 / 防刷,看 `repositories/invite.py`):**
 
-1. **被邀请人唯一**:`invitee_user_id` 唯一约束 → 一个 B 只能被绑一次(幂等键,重复返回 `already_bound`,不重复发奖)。
+1. **被邀请人唯一**:`invitee_user_id` 唯一约束 → 一个 B 只能被绑一次(幂等键,重复返回 `already_bound`,不重复绑定)。
 2. **自邀屏蔽**:`inviter == invitee` → `self_invite`。
-3. **新人闸**:`_is_new_user`(B 的 `created_at` 在 `INVITE_NEW_USER_WINDOW_HOURS` = 72h 内)才发奖,挡存量老用户互相填码薅羊毛 → 否则 `not_eligible`。
+3. **新人闸**:`_is_new_user`(B 的 `created_at` 在 `INVITE_NEW_USER_WINDOW_HOURS` = 72h 内)才生效,挡存量老用户互相填码刷关系 → 否则 `not_eligible`。
 4. **手机号唯一**(天然限量):每个 B = 一个真实手机号账号。
 
-发金币复用 `repositories/wallet.py` 的 `grant_coins`,与建关系记录在**同一事务**提交,保证"建关系 + 双方加金币"原子。
+邀请金币已下线:`bind()` 只记录绑定关系,不再写 `coin_transaction`;响应里的金币字段保留兼容旧客户端,当前恒为 0。
 
 ### 3.2 前端(shaguabijia-app-android)
 
@@ -120,7 +120,7 @@ B 安装并首启 App
 ### 4.4 测试硬约束 / 坑(都是机制,不是 bug)
 
 - **B 必须用新手机号**:`invitee_user_id` 唯一,一个 B 只能绑一次;反复测要换号(或手删 `invite_relation` 那行 + 回滚金币)。
-- **72h 新人闸**:B 注册后 72 小时内绑才发奖(刚注册肯定满足)。
+- **72h 新人闸**:B 注册后 72 小时内绑定才生效(刚注册肯定满足)。
 - **A ≠ B**:自邀被屏蔽。
 - **B 从点下载到首启 App 之间别复制别的东西**:剪贴板会被覆盖 → 归因丢(剪贴板 deferred deeplink 的固有脆弱性)。
 - **笔记本 IP 别变**:debug 包把 `BASE_URL` 的 IP 烧死在编译期,DHCP 一换就连不上 → 给笔记本固定个 LAN IP。

tests/test_ad_reward.py

@@ -305,11 +305,12 @@ def test_feed_reward_grants_by_10_second_units(client) -> None:
         "duration_seconds": 30,
         "adn": "pangle",
         "slot_id": "slot_feed",
+        "display_coin": 4,
     }
     r = client.post("/api/v1/ad/feed-reward", json=payload, headers=_auth(token))
     assert r.status_code == 200, r.text
     body = r.json()
-    expected = sum(calculate_ad_reward_coin("200", i) for i in range(1, 4))
+    expected = 4
     assert body["granted"] is True
     assert body["status"] == "granted"
     assert body["unit_count"] == 3

tests/test_admin_read.py

@@ -64,7 +64,9 @@ def test_dashboard_overview(admin_client: TestClient, admin_token: str) -> None:
     assert data["users"]["total"] >= 1
     assert data["coins"]["granted_total"] >= 5000
     assert "success_rate" in data["comparison"]
-    assert data["cps"]["available"] is False
+    assert data["cps"]["available"] is True
+    assert "meituan_order_count" in data["cps"]
+    assert "jd_order_count" in data["cps"]
 
 
 def test_user_list_and_detail(admin_client: TestClient, admin_token: str) -> None:

tests/test_compare_record.py

@@ -44,6 +44,7 @@ def _food_payload(trace_id: str) -> dict:
         "skipped_dish_names": ["黑牛肉卷"],
         "total_dish_count": 3,
         "information": "在美团找到同店,到手价 ¥123.50",
+        "total_ms": 12345,
     }
 
 
@@ -71,6 +72,8 @@ def test_report_and_derive(client) -> None:
     assert d["information"] == "在美团找到同店,到手价 ¥123.50"
     assert d["store_name"] == "海底捞(朝阳店)"
     assert d["total_dish_count"] == 3
+    assert d["total_ms"] == 12345
+    assert d["raw_payload"]["total_ms"] == 12345
     assert d["skipped_dish_count"] == 1
     assert d["skipped_dish_names"] == ["黑牛肉卷"]
     assert len(d["comparison_results"]) == 3

tests/test_cps_admin.py

@@ -11,6 +11,7 @@ from app.admin.repositories import admin_user as admin_repo
 from app.admin.repositories import cps as cps_repo
 from app.db.session import SessionLocal
 from app.models.cps_link import CpsClick
+from app.models.cps_order import CpsOrder
 from app.models.cps_wx_user import CpsWxUser
 from app.repositories import cps_link as cps_link_repo
 
@@ -180,3 +181,73 @@ def test_day_users_cross_year(admin_client: TestClient, admin_token: str) -> Non
     assert len(users) == 1
     assert users[0]["openid"] == openid
     assert users[0]["visit_count"] == 5  # 次年 01-01 那条被时间窗排除
+
+
+def test_jd_reconcile_updates_dashboard(
+    admin_client: TestClient, admin_token: str, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """京东拉单:按订单行入库;大盘只统计有效 validCode 的佣金。"""
+
+    def fake_query_order_rows(**kwargs):
+        start = kwargs["start_time"].astimezone(_BJ)
+        if start.hour != 10 or kwargs["page_index"] != 1:
+            return {"rows": [], "has_more": False}
+        return {
+            "rows": [
+                {
+                    "id": "pytest-jd-row-valid",
+                    "orderId": "pytest-jd-order-1",
+                    "skuId": "sku-1",
+                    "skuName": "京东测试商品",
+                    "orderTime": "2026-06-25 10:10:00",
+                    "modifyTime": "2026-06-25 10:20:00",
+                    "validCode": "16",
+                    "estimateCosPrice": "19.90",
+                    "estimateFee": "1.23",
+                    "actualFee": "2.34",
+                    "commissionRate": "10.00",
+                },
+                {
+                    "id": "pytest-jd-row-invalid",
+                    "orderId": "pytest-jd-order-2",
+                    "skuId": "sku-2",
+                    "skuName": "京东无效订单",
+                    "orderTime": "2026-06-25 10:15:00",
+                    "modifyTime": "2026-06-25 10:25:00",
+                    "validCode": "4",
+                    "estimateCosPrice": "9.90",
+                    "estimateFee": "0.50",
+                    "commissionRate": "5.00",
+                },
+            ],
+            "has_more": False,
+        }
+
+    monkeypatch.setattr("app.integrations.jd_union.query_order_rows", fake_query_order_rows)
+    r = admin_client.post(
+        "/admin/api/cps/orders/reconcile",
+        params={
+            "platform": "jd",
+            "date_from": "2026-06-25",
+            "date_to": "2026-06-25",
+            "query_time_type": 1,
+        },
+        headers=_auth(admin_token),
+    )
+    assert r.status_code == 200, r.text
+    assert r.json()["fetched"] == 2
+
+    with SessionLocal() as db:
+        rows = db.query(CpsOrder).filter(CpsOrder.platform == "jd").all()
+        assert len([o for o in rows if o.order_id.startswith("jd:pytest-jd-row")]) == 2
+
+    overview = admin_client.get(
+        "/admin/api/stats/overview",
+        params={"date_from": "2026-06-25", "date_to": "2026-06-25"},
+        headers=_auth(admin_token),
+    )
+    assert overview.status_code == 200, overview.text
+    cps = overview.json()["cps"]
+    assert cps["jd_order_count"] >= 1
+    assert cps["jd_commission_cents"] >= 234
+    assert cps["jd_invalid_count"] >= 1

tools/meituan_playground.py

@@ -52,7 +52,8 @@ def call_raw(path: str, body_obj: dict) -> dict:
     }
     url = f"{settings.MT_CPS_HOST}{path}"
     t0 = time.time()
-    resp = httpx.post(url, content=body, headers=headers, timeout=settings.MT_CPS_TIMEOUT_SEC)
+    # trust_env=False: 美团是国内域名,强制直连绕开本机代理(代理会掐断 TLS 握手,报 SSL EOF)
+    resp = httpx.post(url, content=body, headers=headers, timeout=settings.MT_CPS_TIMEOUT_SEC, trust_env=False)
     ms = int((time.time() - t0) * 1000)
     try:
         j = resp.json()