diff --git a/app/repositories/activity.py b/app/repositories/activity.py
new file mode 100644
index 0000000..db9cff7
--- /dev/null
+++ b/app/repositories/activity.py
@@ -0,0 +1,89 @@
+"""活跃口径唯一真源:worker(不活跃清零)与 admin(最近活跃/DAU)共用,防两处漂移。
+
+口径 = max(User.created_at, AnalyticsEvent[ACTIVE_EVENTS], CouponPromptEngagement[claim_started])。
+**不含 last_login_at**(登录/re-login 不代表在用 App);created_at 为恒非空基线。
+清零/预警按北京自然日 0 点对齐(见 reset_cutoff)。
+"""
+from __future__ import annotations
+
+from datetime import date, datetime, timedelta, timezone
+
+from sqlalchemy import func, select
+from sqlalchemy.orm import Session
+
+from app.core.rewards import CN_TZ, cn_today
+from app.models.analytics_event import AnalyticsEvent
+from app.models.coupon_state import CouponPromptEngagement
+
+# —— 活跃事件名(与"用户管理"口径一致)——
+# home_view:进首页(前端埋点,名称前端明天敲定,此处占位;定名后仅改这一常量)。
+HOME_VIEW_EVENT = "home_view"
+COMPARE_START_EVENT = "real_compare_start"   # 发起比价(含浮窗触发)
+COUPON_START_EVENT = "real_coupon_start"     # 发起领券
+ACTIVE_EVENTS = (HOME_VIEW_EVENT, COMPARE_START_EVENT, COUPON_START_EVENT)
+ACTIVE_ENGAGE_TYPE = "claim_started"         # coupon_prompt_engagement 一键领取
+
+
+def as_utc(value: datetime) -> datetime:
+    """任意 datetime → tz-aware UTC(无时区按 UTC 解释)。用于与 DateTime(timezone=True) 列比较,
+    比较绝对时刻、与会话时区无关(口径同 admin queries._as_utc)。"""
+    if value.tzinfo is None:
+        return value.replace(tzinfo=timezone.utc)
+    return value.astimezone(timezone.utc)
+
+
+def norm_utc(dt: datetime | None) -> datetime | None:
+    """naive 视为 UTC 补 tzinfo(SQLite 读回 naive、PG 读回 aware,混着 max() 会 TypeError)。"""
+    if dt is None:
+        return None
+    return dt if dt.tzinfo is not None else dt.replace(tzinfo=timezone.utc)
+
+
+def cn_midnight_utc(d: date) -> datetime:
+    """北京 d 日 00:00 → tz-aware UTC datetime。"""
+    return as_utc(datetime(d.year, d.month, d.day, tzinfo=CN_TZ))
+
+
+def reset_cutoff(reset_days: int, today: date | None = None) -> datetime:
+    """应清零边界(tz-aware UTC):last_active < 此值 ⟺ 距末次活跃已满 reset_days 天(北京 0 点对齐)。
+    = 北京 00:00 of (today − (reset_days − 1))。例:reset_days=15、today=1/20 → 北京 1/6 00:00。"""
+    today = today or cn_today()
+    return cn_midnight_utc(today - timedelta(days=reset_days - 1))
+
+
+def last_active_subqueries(db: Session):
+    """两个按 user_id 预聚合的派生表:最近活跃事件(ACTIVE_EVENTS)、最近领券发起(claim_started)。
+    返回 (ev_sub, eng_sub)。口径同 admin,LEFT JOIN 用,借事件索引只扫活跃事件。"""
+    ev_sub = (
+        select(
+            AnalyticsEvent.user_id.label("user_id"),
+            func.max(AnalyticsEvent.created_at).label("last_at"),
+        )
+        .where(AnalyticsEvent.user_id.is_not(None), AnalyticsEvent.event.in_(ACTIVE_EVENTS))
+        .group_by(AnalyticsEvent.user_id)
+        .subquery()
+    )
+    eng_sub = (
+        select(
+            CouponPromptEngagement.user_id.label("user_id"),
+            func.max(CouponPromptEngagement.created_at).label("last_at"),
+        )
+        .where(
+            CouponPromptEngagement.user_id.is_not(None),
+            CouponPromptEngagement.engage_type == ACTIVE_ENGAGE_TYPE,
+        )
+        .group_by(CouponPromptEngagement.user_id)
+        .subquery()
+    )
+    return ev_sub, eng_sub
+
+
+def last_active_expr(base_col, ev_sub, eng_sub, dialect: str):
+    """max(base_col, 最近活跃事件, 最近领券) 的 SQL 表达式。PG 用 greatest、SQLite 用 max。
+    子聚合缺失(未命中)时 coalesce 到 base_col(= User.created_at,恒非空基线)。"""
+    greatest = func.greatest if dialect == "postgresql" else func.max
+    return greatest(
+        base_col,
+        func.coalesce(ev_sub.c.last_at, base_col),
+        func.coalesce(eng_sub.c.last_at, base_col),
+    )
diff --git a/tests/test_inactivity_reset.py b/tests/test_inactivity_reset.py
index 9a07d5e..adf01ee 100644
--- a/tests/test_inactivity_reset.py
+++ b/tests/test_inactivity_reset.py
@@ -7,6 +7,7 @@ from sqlalchemy import select
 
 from app.db.session import SessionLocal
 from app.models.inactivity import InactivityNotificationLog, InactivityResetLog
+from app.repositories import activity
 
 
 def test_reset_and_notification_models_persist() -> None:
@@ -31,3 +32,22 @@ def test_reset_and_notification_models_persist() -> None:
     finally:
         db.rollback()
         db.close()
+
+
+def test_reset_cutoff_is_cn_midnight_of_today_minus_days_minus_1() -> None:
+    # RESET_DAYS=15, today=1/20 → cutoff = 北京 00:00 of 1/6 = 1/5 16:00 UTC
+    cutoff = activity.reset_cutoff(15, today=date(2026, 1, 20))
+    assert cutoff == datetime(2026, 1, 5, 16, 0, tzinfo=timezone.utc)
+
+
+def test_active_event_constants() -> None:
+    assert activity.HOME_VIEW_EVENT in activity.ACTIVE_EVENTS
+    assert "real_compare_start" in activity.ACTIVE_EVENTS
+    assert "real_coupon_start" in activity.ACTIVE_EVENTS
+    assert activity.ACTIVE_ENGAGE_TYPE == "claim_started"
+
+
+def test_as_utc_normalizes() -> None:
+    assert activity.as_utc(datetime(2026, 1, 1)) == datetime(2026, 1, 1, tzinfo=timezone.utc)
+    cn = datetime(2026, 1, 1, tzinfo=activity.CN_TZ)  # 北京 0 点 = 前一天 16:00 UTC
+    assert activity.as_utc(cn) == datetime(2025, 12, 31, 16, 0, tzinfo=timezone.utc)