数据中心接入设计方案

1. 设计要求

以下要求来自运营对齐决议，是设计的硬约束：

#	要求	来源	含义
A	实时数据 5-10 分钟刷新	示例业务	今日累计付费金额、日监控 KPI 卡片需要近实时
B	流水数据按小时刷新	示例业务	分时活跃、付费小时分布等当日按小时数据
C	其他数据 T+1	示例业务	留存、LTV、渠道 ROI 等历史趋势，T+1 可接受
D	离线数据最迟次日 9 点	示例业务	ETL 截止时间
E	日环比：0 点到 0 点完整周期	示例业务	start_date/end_date 按整日对齐
F	当日对比需时间窗口对齐	示例业务	今日数据不完整，对比时往日数据截断到同一时刻

2. 核心架构：Repo 接口 + 双轨数据

2.1 架构总览

Frontend / Agent
      |
      ▼
┌─────────────────────────────┐
│  Router Layer               │
│  - /dashboard/* (固定看板)   │
│  - /dashboards/* (自定义看板) │
│  - /query/* (明细查询)       │
│  - /segments/* (用户分群)    │
│  - /event-analysis/* (事件)  │
└─────────────────────────────┘
      |
      ▼
┌─────────────────────────────┐
│  ScopeResolver              │
│  project_id → (game_id,     │
│  region_code, dc_config)    │
└─────────────────────────────┘
      |
      ▼
┌─────────────────────────────┐
│  Service Layer              │
│  DashboardService           │
│  QueryService               │
│  ReportService              │
│  (编排层，无数据查询逻辑)     │
└─────────────────────────────┘
      |
      ▼
┌─────────────────────────────┐
│  RepoFactory                │
│  根据 scope 路由 Repo 实现   │
└─────────────────────────────┘
      |
      ├──────────────┬──────────────┬──────────────┐
      ▼              ▼              ▼              ▼
┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐
│ CachedRepo│  │ HybridRepo│  │ LocalPG │  │ Datacenter│
│ (装饰器)  │  │ (组合器)  │  │ Repo    │  │ Repo     │
└──────────┘  └──────────┘  └──────────┘  └──────────┘
      |              |              |              |
      ▼              ▼              ▼              ▼
┌──────────┐  ┌──────────┐  ┌──────────┐  ┌──────────┐
│ Redis    │  │ 合并历史  │  │ PG       │  │ DC HTTP  │
│ Cache    │  │ + 当日    │  │ Session  │  │ Client   │
└──────────┘  └──────────┘  └──────────┘  └──────────┘
      |              |              |              |
      └──────────────┴──────────────┴──────────────┘
                     |
                     ▼
            ┌──────────────┐
            │ 数据中心     │
            │ /aggregate   │
            │ /free-query  │
            │ /sql-query   │
            └──────────────┘

2.2 分层原则

层级	职责	依赖规则
Router	参数校验、scope 解析、响应封装	只调 Service
Service	业务编排、衍生指标计算	只依赖 Repo 接口
RepoFactory	根据 scope 组装 Repo 实现	不暴露给 Router
Repo 实现	具体数据访问（HTTP/SQL）	只被 Factory 创建
Redis	缓存装饰	通过 CachedRepo 透明接入

核心约束：Service 层只依赖 Repo 接口，不感知底层是 HTTP 还是 SQL。

2.3 Repo 接口定义

# app/modules/dashboard/repos/protocol.py

class DashboardRepo(Protocol):
    """看板数据访问接口。Service 层唯一依赖的数据源抽象。"""

    async def query_scalar(
        self, scope: Scope, metrics: list[str], date_range: DateRange
    ) -> dict[str, MetricResult]:
        """标量查询：返回每指标每日期一个值"""

    async def query_dimension(
        self, scope: Scope, metrics: list[str], dimension: str, date_range: DateRange
    ) -> dict[str, list[DimensionResult]]:
        """维度查询：返回每指标按维度拆分的数据"""

    async def query_series(
        self, scope: Scope, metrics: list[str], date_range: DateRange, granularity: str
    ) -> dict[str, list[TimeSeriesPoint]]:
        """时序查询：返回趋势数据"""


@dataclass
class MetricResult:
    value: float | int
    available: bool
    is_final: bool
    reason: str | None = None
    source: str = ""  # "datacenter" | "local_pg" | "fallback"

2.4 Repo 实现

Repo	职责	底层
`DatacenterRepo`	调 DC `/aggregate`	HTTP Client
`LocalPGRepo`	查 `datacenter_metric_daily`	PG Session
`HybridRepo`	双轨合并（历史 LocalPG + 当日 DC）	组合两个 Repo
`CachedRepo`	装饰器，加 Redis 缓存	包裹任意 Repo

# app/modules/dashboard/repos/hybrid_repo.py

class HybridRepo:
    """双轨 Repo：历史数据走 LocalPGRepo，当日数据走 DatacenterRepo。"""

    def __init__(self, local: LocalPGRepo, dc: DatacenterRepo):
        self._local = local
        self._dc = dc

    async def query_scalar(self, scope, metrics, date_range):
        history_range, today_range = date_range.split_at_today()
        results = {}
        if history_range:
            results.update(await self._local.query_scalar(scope, metrics, history_range))
        if today_range:
            results.update(await self._dc.query_scalar(scope, metrics, today_range))
        return results


# app/modules/dashboard/repos/cached_repo.py

class CachedRepo:
    """缓存装饰器。包裹在 HybridRepo 外层，对 Service 完全透明。"""

    def __init__(self, inner: DashboardRepo, redis: Redis):
        self._inner = inner
        self._redis = redis

    async def query_scalar(self, scope, metrics, date_range):
        # 用户主动刷新：跳过缓存读，直接回源 + 写缓存
        if scope.skip_cache:
            result = await self._inner.query_scalar(scope, metrics, date_range)
            if any(v.available for v in result.values()):
                key = _cache_key("scalar", scope, metrics, date_range)
                ttl = _resolve_ttl(date_range)
                await self._redis_setex(key, ttl, json.dumps(serialize(result), default=str))
            return result

        key = _cache_key("scalar", scope, metrics, date_range)
        ttl = _resolve_ttl(date_range)

        # 1. 读缓存
        cached = await self._redis_get(key)
        if cached:
            parsed = _parse_cached(cached)
            if parsed is not None:
                return parsed

        # 2. 互斥锁防击穿（uuid token 标识锁归属，防误删）
        lock_key = f"{key}:lock"
        lock_token = uuid.uuid4().hex[:16]
        lock = await self._redis_set(lock_key, lock_token, nx=True, ex=10)

        if not lock:
            # 指数退避重试读缓存（最多 3 次，最长等 3.5s）
            for attempt in range(3):
                await asyncio.sleep(0.5 * (2 ** attempt))
                cached = await self._redis_get(key)
                if cached:
                    parsed = _parse_cached(cached)
                    if parsed is not None:
                        return parsed
            # 降级为直接回源
            return await self._inner.query_scalar(scope, metrics, date_range)

        try:
            # 3. 回源
            result = await self._inner.query_scalar(scope, metrics, date_range)
            # 全不可用时跳过缓存写入
            if all(not v.available for v in result.values()):
                return result
            # 4. 写缓存（部分不可用 TTL 减半）
            cache_ttl = ttl // 2 if any(not v.available for v in result.values()) else ttl
            await self._redis_setex(key, cache_ttl, json.dumps(serialize(result), default=str))
            return result
        finally:
            # 条件删除：仅当锁仍属于本进程时才删除
            current_val = await self._redis_get(lock_key)
            if current_val == lock_token:
                await self._redis_delete(lock_key)

2.5 RepoFactory 路由

class DashboardRepoFactory:
    """Repo 工厂：根据 Scope.dc_enabled 和 game_id 自动选择数据源策略。"""

    def __init__(self, session: AsyncSession, redis) -> None:
        self._session = session
        self._redis = redis

    async def create(self, scope: Scope) -> DashboardRepo:
        local_pg = LocalPGRepo(self._session)

        if not scope.dc_enabled or not scope.game_id:
            # 数据中心未启用或缺少 game_id：仅本地 PG
            return CachedRepo(local_pg, self._redis)

        # 数据中心启用：按 project_id 动态解析 DC 配置
        config = await DatacenterConfigResolver(self._session).resolve(scope.project_id)
        if not config or not config.is_active or not config.base_url or not config.api_key:
            # 解析失败或配置不完整：降级为本地 PG
            return CachedRepo(local_pg, self._redis)

        # 配置完整：Hybrid（本地历史 + DC 实时）+ 缓存
        dc_client = DatacenterClient(
            base_url=config.base_url, api_key=config.api_key, auth_mode=config.auth_mode
        )
        dc = DatacenterRepo(dc_client, scope)
        hybrid = HybridRepo(local_pg, dc)
        return CachedRepo(hybrid, self._redis)

2.6 Service 层示例

class DashboardService:
    def __init__(self, repo_factory: DashboardRepoFactory):
        self._factory = repo_factory

    async def get_daily_monitor(self, scope: Scope, date: str, channel_id: str = "") -> DailyMonitorData:
        repo = await self._factory.create(scope)

        # 1. 查标量（dau, revenue, pay_rate）
        today_result = await repo.query_scalar(
            scope=scope,
            metrics=["dau", "revenue", "pay_rate"],
            date_range=DateRange(start=date, end=date),
        )

        # 2. 查趋势（近 7 日）
        trend_result = await repo.query_series(
            scope=scope,
            metrics=["dau", "revenue", "pay_rate"],
            date_range=DateRange(start=date_minus_6(date), end=date),
            granularity="day",
        )

        # 3. Service 层只负责计算衍生指标
        kpis = self._build_kpis(today_result)
        return DailyMonitorData(kpis=kpis, trend=trend_result)

3. 固定看板数据流

3.1 查询链路

Router(project_id) → ScopeResolver → DashboardService → RepoFactory → [Cached]HybridRepo

ScopeResolver 职责：

查 sys_project_config 得 game_id, region
查 sys_datacenter_game_source 得 DC 配置（game_id × region_code 复合主键）
返回 Scope(project_id, game_id, region_code, dc_enabled, dc_config)

3.2 日监控 Tab 为例

日监控 Tab 只查 3 个标量指标（dau, revenue, pay_rate），是验证端到端链路的最佳起点。

3.3 时间对齐

日对齐（要求 E）：DC /aggregate 的 start_date/end_date 为 YYYY-MM-DD 日期级，完整日周期由查询参数保证。

当日对比对齐（要求 F）：放在 DatacenterRepo 层。

class DatacenterRepo:
    async def query_scalar(self, scope, metrics, date_range):
        today = datetime.now(ZoneInfo(scope.timezone)).date().isoformat()
        if self._is_today_comparison(date_range, scope.timezone):
            # 今日数据：按当前小时截断
            cutoff_hour = self._get_current_cutoff_hour(scope.timezone)
            raw = await self._client.aggregate(scope.game_id, metrics, date_range)
            return self._apply_cutoff(raw, cutoff_hour)
        elif self._is_comparison_date(date_range, today):
            # 对比日（昨日/上周同日）：也按相同 cutoff_hour 截断，保证时间窗口对齐
            cutoff_hour = self._get_current_cutoff_hour(scope.timezone)
            raw = await self._client.aggregate(scope.game_id, metrics, date_range)
            return self._apply_cutoff(raw, cutoff_hour)
        return await self._client.aggregate(scope.game_id, metrics, date_range)

Service 层拿到的是已对齐的数据。

4. 明细查询与自定义看板的 DC 接入

4.1 当前查询类型映射

查询场景	实现路径	说明
固定看板 8 Tab	`DashboardService` → `DashboardRepo` → DC `/aggregate` (当日) + `LocalPGRepo` (历史)	主链路，HybridRepo 双轨合并
登录/订单聚合	`QueryService` → `DatacenterLoginRepo` / `DatacenterOrderRepo` → DC `/free-query`	已完成
自定义看板组件	`ReportApiSourceResolver` → 复用固定看板 `DashboardRepo`	数据源标记 `"dc"`
自定义报表	`ReportApiSourceResolver` → DC `/free-query` 或 `/sql-query`	受控查询
Agent NL2SQL	Agent 工具 → DC `/free-query`(指标名模式) / DC `/sql-query` / 本地 PG fallback	SemanticRouter 三层路由
用户分群	`UserSegmentService` → PG 配置表 + DC `/player-events/filter` + `/behavior` + `/segments/{id}`	已完成
事件分析	`EventAnalysisService` → `DCEventRepo` → DC `/player-events/aggregate` + `/catalog`	已完成；支持单事件聚合与分群过滤；漏斗/路径分析后续迭代

4.2 明细查询改造

当前 QueryService.query_login / QueryService.query_order 被前端页面和 Agent 工具共用。

改造后，新增 Repo 接口：

class LoginDetailRepo(Protocol):
    async def query(self, params: LoginQueryParams, scope: Scope) -> dict[str, Any]: ...

class OrderDetailRepo(Protocol):
    async def query(self, params: OrderQueryParams, scope: Scope) -> dict[str, Any]: ...

实现：

class DatacenterLoginRepo(LoginDetailRepo):
    """DC /free-query 实现。"""
    async def query(self, params, scope):
        return await self._dc_client.free_query(
            game_id=scope.game_id,
            table="ga_core_kpi_daily",  # 或 ga_active_hourly
            filters={
                "record_date_start": params.start_date,
                "record_date_end": params.end_date,
            },
            aggregates={"login_count": "SUM(login_count)", "unique_roles": "COUNT(DISTINCT role_id)"},
            group_by=["record_date", "channel_id"],
        )

class LocalLoginRepo(LoginDetailRepo):
    """本地回退实现。"""
    async def query(self, params, scope):
        return await self._repo.query_login_by_cube(params, scope.project_id)

Service 层统一：

class QueryService:
    def __init__(self, login_repo: LoginDetailRepo, order_repo: OrderDetailRepo):
        self._login_repo = login_repo
        self._order_repo = order_repo

    async def query_login(self, params: LoginQueryParams, scope: Scope) -> dict[str, Any]:
        return await self._login_repo.query(params, scope)

    async def query_order(self, params: OrderQueryParams, scope: Scope) -> dict[str, Any]:
        return await self._order_repo.query(params, scope)

前端页面和 Agent 工具都走同一个 QueryService。

4.3 自定义看板改造

自定义看板的 data_reports.data_source_json 增加 source_type 字段：

{
  "source_type": "datacenter",
  "datacenter_config": {
    "endpoint": "aggregate",
    "metrics": ["dau", "revenue"],
    "dimensions": ["channel"],
    "date_range_type": "last_7_days"
  }
}

ReportApiSourceResolver 根据 source_type 路由：

"api" → 复用固定看板 DashboardRuntimeReader 或明细查询 DetailQueryRuntimeReader
"datacenter" → 直接调 DC /aggregate（通过 DashboardRepoFactory）
"sql" → 保持本地 SQL 查询
"analysis_model" → 分析模型数据源

5. ETL 与本地存储

5.1 ETL 职责

每日 08:00 执行：

拉取 T-1 全部指标（标量 + 维度）
写入 datacenter_metric_daily（PG）
标记 is_final=true
清理对应日期范围的 Redis 缓存

5.2 存储方案：PG 新表 `datacenter_metric_daily`

CREATE TABLE datacenter_metric_daily (
    project_id VARCHAR(64) NOT NULL,
    stat_date DATE NOT NULL,
    metric_name VARCHAR(64) NOT NULL,
    metric_value DECIMAL(18, 4),
    dimensions JSONB DEFAULT '{}',
    available BOOLEAN DEFAULT true,
    is_final BOOLEAN DEFAULT true,
    source VARCHAR(32) DEFAULT 'datacenter',
    reason VARCHAR(128),
    as_of TIMESTAMP,
    imported_at TIMESTAMP DEFAULT NOW(),

    PRIMARY KEY (project_id, stat_date, metric_name, dimensions)
);

CREATE INDEX idx_dc_metric_project_date ON datacenter_metric_daily(project_id, stat_date);
CREATE INDEX idx_dc_metric_name ON datacenter_metric_daily(metric_name);
CREATE INDEX idx_dc_metric_dimensions ON datacenter_metric_daily USING GIN(dimensions);

ETL 写入（幂等 UPSERT）：

INSERT INTO datacenter_metric_daily
  (project_id, stat_date, metric_name, metric_value, dimensions, available, is_final, source, as_of)
VALUES
  ('proj_xxx', '2026-06-01', 'dau', 29046.0, '{}', true, true, 'events', '2026-06-01T09:43:48')
ON CONFLICT (project_id, stat_date, metric_name, dimensions)
DO UPDATE SET
  metric_value = EXCLUDED.metric_value,
  available = EXCLUDED.available,
  is_final = EXCLUDED.is_final,
  source = EXCLUDED.source,
  as_of = EXCLUDED.as_of,
  imported_at = NOW();

选型理由：

数据量小（每天每项目 ~3600 行，10 个项目 = 3.6 万行/天），PG 完全胜任
JSONB 维度灵活，不需要为每个维度组合加列
与 sys_project_config 等配置表同库，便于关联和事务
强一致性，ON CONFLICT DO UPDATE 天然幂等

5.3 ETL Worker 流程

class DatacenterETLWorker:
    async def sync_daily_metrics(self, project_id: str, date: str):
        scope = await scope_resolver.resolve(project_id)

        # 1. 标量指标
        scalar = await self._dc_client.aggregate(
            game_id=scope.game_id,
            metrics=ALL_SCALAR_METRICS,  # ~80 个标量指标
            start_date=date, end_date=date,
            dimensions=[],
        )
        await self._save_to_pg(project_id, date, scalar)

        # 2. 维度指标
        for dim in ["hour", "channel"]:
            dim_result = await self._dc_client.aggregate(
                game_id=scope.game_id,
                metrics=ALL_DIMENSION_METRICS,
                start_date=date, end_date=date,
                dimensions=[dim],
            )
            await self._save_to_pg(project_id, date, dim_result)

        # 3. 清理缓存
        await self._invalidate_cache(project_id, date)

6. 旧表清理记录

6.1 表清理结果

表	状态	说明
`metric_daily_agg`	已删除	业务代码零引用
`overview_events_daily`	已下线	`repository.py` 返回空列表，标注"聚合表查询已下线"
`dim_channel`	已删除	代码零引用，仅 `seed.py` 和 Agent 提示词残留
`dim_player`	已删除	代码零引用。用户分群和事件分析通过 DC `/player-events` 接口恢复查询能力（Phase 4 已完成）
事件明细表（`event_*`）	已删除	事件明细表已清理，事件分析通过 DC `/player-events/aggregate` 恢复

6.2 删除执行结果

Phase 1 已完成：

~~删除 dim_channel 表~~ 已完成
~~删除 RealtimeMetricsRepo~~ 已完成（代码零引用）
~~删除 dim_player 表~~ 已完成

Phase 2 已完成：

~~删除 metric_daily_agg 表~~ 已完成（业务代码零引用）
~~删除 overview_events_daily 表~~ 已完成（方法返回空列表）
~~删除 RealtimeMetricsRepo / RealtimeMetricsRepoV2~~ 已完成
~~CubeQueryService 下线~~ 已完成（legacy 方法返回空结果）

7. Redis 缓存策略

7.1 缓存分层

层级	数据	TTL	用途
L1 内存	指标目录	1 小时	`/aggregate/metrics` 不常变
L2 Redis	DC 响应	2-5 分钟	当日数据短期缓存
L2 Redis	历史查询	1-24 小时	历史数据长期缓存
L3 本地 DB	ETL 结果	永久	T-1 及以前

7.2 缓存 Key 规范

dc:catalog:{game_id}:{region}              → 指标目录
dc:scalar:{game_id}:{region}:{metrics_hash}:{start}:{end}:{dim_hash}  → 标量
dc:series:{game_id}:{region}:{metrics_hash}:{start}:{end}:{granularity} → 时序
dc:dim:{game_id}:{region}:{metric}:{dimension}:{start}:{end} → 维度

7.3 TTL 动态策略

def resolve_ttl(date_range: DateRange) -> int:
    """动态 TTL：含当日最短，历史数据逐渐增长。"""
    if date_range.includes_today():
        return 120      # 2 分钟，当日数据变化快
    if date_range.days <= 7:
        return 300      # 5 分钟
    if date_range.days <= 30:
        return 900      # 15 分钟
    return 3600         # 1 小时，历史数据已固定

7.4 缓存穿透/击穿防护

互斥锁：热点指标加分布式锁（Redis SET NX EX 10），避免并发穿透
空值缓存：DC 返回 available=False 也缓存，TTL 减半
异步回源：缓存过期后异步刷新，不阻塞请求

7.5 缓存清理

场景	清理方式
ETL 完成写入	清理对应日期的 `dc:scalar:` 和 `dc:series:` key
DC 数据质量告警	清空该项目全部缓存，强制回源
运营手动刷新	管理接口 `POST /admin/cache/invalidate`
Redis 内存紧张	LRU 淘汰（Redis 自带）

8. DatacenterClient 接口

class DatacenterClient:
    """数据中心 HTTP 客户端（连接池、认证、重试、限流、Prometheus 指标）。"""

    # ---- /aggregate 端点 ----

    async def get_metrics_catalog(self, game_id: str, region: str = "dc_region_1") -> list[dict]:
        """GET /aggregate/metrics — 指标目录（带 Redis 缓存 1h）"""

    async def aggregate(
        self, game_id: str, region: str, start_date: str, end_date: str,
        metrics: list[str], dimensions: list[str] | None = None, *,
        timezone: str = "Asia/Shanghai", currency: str = "CNY",
        player_scope: str = "account", return_days: int | None = None,
    ) -> AggregateResponse:
        """POST /aggregate — 批量聚合查询"""

    async def aggregate_timeseries(
        self, game_id: str, region: str, start_date: str, end_date: str,
        metrics: list[str], dimensions: list[str] | None = None,
        granularity: str = "day", limit: int = 10000, *,
        timezone: str = "Asia/Shanghai", currency: str = "CNY",
        player_scope: str = "account", return_days: int | None = None,
    ) -> TimeseriesResponse:
        """POST /aggregate/timeseries — 时序查询"""

    async def aggregate_distribution(
        self, game_id: str, region: str, metric: str,
        start_date: str, end_date: str, limit: int = 1000, *,
        timezone: str = "Asia/Shanghai", currency: str = "CNY",
        player_scope: str = "account", return_days: int | None = None,
    ) -> DistributionResponse:
        """POST /aggregate/distribution — 分布明细查询"""

    # ---- /free-query 端点 ----

    async def free_query(
        self, game_id: str, table: str, filters: dict, aggregates: dict[str, str],
        group_by: list[str], region: str = "dc_region_1", granularity: str | None = None,
    ) -> dict:
        """POST /free-query — 表查询模式"""

    async def free_query_metrics(
        self, game_id: str, metrics: list[str], start_date: str, end_date: str,
        region: str = "dc_region_1", granularity: str | None = None,
        dimensions: list[str] | None = None, limit: int = 1000,
    ) -> dict:
        """POST /free-query — 指标名模式（Agent table_query 使用）"""

    # ---- /sql-query 端点 ----

    async def sql_query(
        self, game_id: str, region: str, sql: str, use_cache: bool = False,
    ) -> dict:
        """POST /sql-query — 高级 SQL（仅服务端内部使用）"""

    # ---- /player-events 端点 ----

    async def player_events_filter(
        self, game_id: str, region: str = "dc_region_1", *,
        conditions: list[dict] | None = None,
        behavior_conditions: list[dict] | None = None,
        event_conditions: list[dict] | None = None,
        condition_logic: str = "and", limit: int = 20,
        return_player_ids: bool = True, profile_date: str | None = None,
    ) -> dict:
        """POST /player-events/filter — 用户分群过滤"""

    async def player_events_behavior(
        self, game_id: str, region: str = "dc_region_1", *,
        player_ids: list[str] | None = None, segment_id: str | None = None,
        metrics: list[dict] | None = None,
        start_date: str | None = None, end_date: str | None = None,
    ) -> dict:
        """POST /player-events/behavior — 用户行为聚合"""

    async def player_events_aggregate(
        self, game_id: str, region: str = "dc_region_1", *,
        event_name: str | None = None, start_date: str | None = None,
        end_date: str | None = None, time_granularity: str = "day",
        dimensions: list[str] | None = None, metrics: list[dict] | None = None,
        filters: list[dict] | None = None, segment_id: str | None = None,
        segment_player_ids: list[str] | None = None, limit: int = 1000,
    ) -> dict:
        """POST /player-events/aggregate — 事件聚合分析"""

    async def player_events_catalog(
        self, game_id: str, region: str = "dc_region_1", *,
        event_name: str | None = None,
    ) -> dict:
        """POST /player-events/catalog — 事件目录"""

    async def player_events_detail(
        self, game_id: str, region: str = "dc_region_1", *,
        event_names: list[str] | None = None, player: dict | None = None,
        start_time: str | None = None, end_time: str | None = None,
        fields: list[str] | None = None, limit: int = 50,
        include_sensitive: bool = False,
    ) -> dict:
        """POST /player-events/detail — 用户事件明细查询"""

    async def get_segment_members(
        self, segment_id: str, game_id: str, region: str = "dc_region_1", *,
        limit: int = 1000, offset: int = 0,
    ) -> dict:
        """GET /player-events/segments/{segment_id} — 分群成员列表"""

连接池通过 get_dc_client_pool() 全局管理，DatacenterClient 实例由 pool 按 (base_url, api_key) 缓存复用。_request_with_retry 实现指数退避重试（最多 3 次，429/网络错误），并上报 Prometheus 指标（DC_LATENCY、DC_REQUESTS）。

9. API Key 与配置管理

DC 配置字段于 2026-06-16 合并到 sys_project_config 表，与项目配置统一管理。

字段	类型	说明
`game_id`	VARCHAR(64)	业务标识，用于 DC `/aggregate` 等接口的 `game_id` 参数
`game_name`	VARCHAR(200)	业务显示名
`region`	VARCHAR(8)	区域代码（如 `region_a`、`region_b`、`global`），默认 `region_a`
`base_url`	VARCHAR(500)	DC 根地址
`auth_mode`	VARCHAR(32)	认证方式：`header_raw` / `header_bearer` / `query_param`，默认 `header_raw`
`api_key_ref`	VARCHAR(128)	API Key（直接存储于 `X-API-Key` 头或 query param）
`is_active`	BOOLEAN	是否启用，默认 `true`。`false` 时 RepoFactory 降级为 `LocalPGRepo`
`currency`	VARCHAR(8)	货币代码（如 `CNY`），默认 `CNY`
`timezone_offset`	SMALLINT	时区偏移（小时），默认 `8`（Asia/Shanghai）

解析链路：project_id → DatacenterConfigResolver.resolve() → 查 sys_project_config → 校验 is_active/game_id/base_url → _map_region_to_dc() 映射为 DC 区域标识。

region → DC 区域映射：

region	DC region
`region_a`	`dc_region_1`
`region_b`	`dc_region_1`
`region_c`	`dc_region_2`
`region_d`	`dc_region_3`
`region_e`	`dc_region_4`
`global`	`global`

注：2026-06-03 重构后，原 sys_datacenter_region_mapping 表已合并删除；2026-06-16 将 sys_datacenter_game_source 表合并到 sys_project_config。cube_api_url / free_query_api_url / cube_max_days / free_query_max_days / schema_family / filter_dimension / filter_value 等旧 Cube 架构字段已移除。

10. Agent 查询（后置 Phase 3）

Agent 模式	状态	实现路径	时机
`dashboard_metric`	已完成	`_match_metric_semantic` → `DatacenterClient.aggregate()`，含 `available`/`is_final`/`reason`	Phase 3
`table_query`	已完成	`_execute_table_query()` → `DatacenterConfigResolver` → `DatacenterClient.free_query()`	Phase 3
`login_detail`	已完成	`QueryService` → `DatacenterLoginRepo` → DC `/free-query`	Phase 2
`order_detail`	已完成	`QueryService` → `DatacenterOrderRepo` → DC `/free-query`	Phase 2
`sql`	已完成	`SemanticRouter.route()` 三层混合路由 → DC `/free-query` 或 DC `/sql-query` 或本地 PG fallback	Phase 3

Agent 所有查询模式均已完成 DC 接入。sql 模式通过 SemanticRouter 做三层路由判断（代码层 + LLM 复杂度标注 + 运行时兜底），优先走 DC /free-query，复杂 SQL 走 DC /sql-query，本地 PG 仅作为 DC 不可用时的 fallback。

11. 实施阶段

Phase 1: Repo + 日监控 Tab（4 周）—— 已完成

~~DatacenterClient 封装（HTTP、认证、重试、限流）~~ 已完成
~~DashboardRepo 接口 + DatacenterRepo / LocalPGRepo / HybridRepo / CachedRepo 实现~~ 已完成
~~ScopeResolver 实现~~ 已完成
~~RepoFactory 实现~~ 已完成
~~日监控 Tab 端到端接入~~ 已完成
~~Redis 缓存接入~~ 已完成
~~dim_channel 删除~~ 已完成
~~dim_player 删除~~ 已完成
~~业务表清理~~ 已完成

Phase 2: 全量 Tab + 明细查询 + 旧表清理（4 周）—— 已完成

~~固定看板全量 8+ Tab 接入~~ 已完成（/active /growth /payment /retention /return /operation /currency /overview /daily-monitor 全部 use_v2=True）
~~ETL Worker 每日同步 T-1 到 datacenter_metric_daily（PG）~~ 已完成
~~LoginDetailRepo / OrderDetailRepo → DC /free-query~~ 已完成
~~自定义看板 source_type: datacenter 支持~~ 已完成
~~删除 metric_daily_agg + overview_events_daily~~ 已完成
~~删除 RealtimeMetricsRepo / CubeQueryService~~ 已完成
~~未就绪指标降级展示~~ 已完成

Phase 3（已完成）

~~Agent dashboard_metric 恢复~~ 已完成（语义匹配 → DC aggregate）
~~Agent login_detail/order_detail → DC /free-query~~ 已完成
~~Agent table_query → DC /free-query~~ 已完成（_execute_table_query() → DatacenterClient.free_query_metrics()）
~~Agent sql 模式 DC 路由~~ 已完成（SemanticRouter 三层混合路由：代码层 → LLM 标注 → 运行时兜底）
~~事件分析 DCEventRepo 接入~~ 已完成（DCEventRepo → DC /player-events/aggregate + /catalog）
~~DatacenterClient /player-events 全端点实现~~ 已完成（player_events_filter、player_events_behavior、player_events_aggregate、player_events_catalog、player_events_detail、get_segment_members）
~~事件目录 Redis 缓存~~ 已完成（DCEventRepo.get_catalog() Redis 缓存 1h）
~~ETL 延迟/成功率监控~~ 已完成（DC_ETL_DURATION/DC_ETL_RECORDS/DC_ETL_LAST_SUCCESS Prometheus 指标）
~~DC 可用性监控~~ 已完成（app/services/monitor_service.py 轮询 DC 健康）

Phase 4（已完成）

~~用户分群 DCSegmentRepo 实现 → DC /player-events/filter + /behavior + /segments/{id}~~ 已完成
~~SegmentProfileRepo 桩方法切换到 DCSegmentRepo~~ 已完成
~~端到端验证：分群创建 → 预览 → 导出；事件分析 → 引用分群~~ 已完成
~~清理 segment_repo_v2.py 和 event_repo_v2.py 残余桩代码~~ 已完成

后置 backlog

事件漏斗分析（依赖 DC 后续接口支持）
事件路径分析（依赖 DC 后续接口支持）

12. 降级策略

场景	处理
DC 返回 5xx	`DatacenterRepo` 返回 `available=False`，`Service` 层展示 placeholder
DC 限流 429	Client 指数退避重试 3 次，仍失败则返回 `available=False`
指标 `available=false`	看板展示 "—"，tooltip 显示 reason
整 Tab 不可用	runtime_meta 标记 `status="empty"`，前端空状态
项目未配置 DC	`RepoFactory` 返回 `LocalPGRepo`（本地空数据或历史残留）

13. DC 时序颗粒度支持

DC /aggregate/timeseries 和 /free-query（指标名模式）支持以下时间颗粒度：

颗粒度	`/aggregate/timeseries`	`/free-query` 指标名模式	说明
`day`	支持	支持	按 `count_date` 返回每日行
`week`	支持	支持	按 `period` 返回周粒度行，含 `period_start`/`period_end`
`month`	支持	支持	按 `period` 返回月粒度行
`year`	支持	支持	按 `period` 返回年粒度行
`five_minute`	支持(仅付费指标)	支持(仅付费指标)	按 `count_time_5m` 返回 5 分钟行，最大 7 天查询范围
`range`	—	支持(默认)	返回范围汇总 1 行

后端 DatacenterRepo.query_series() 已支持将 day/week/month/year/five_minute 转发到 DC aggregate_timeseries。此外，total 和 hour 两种粒度由服务端二次上卷实现：total 以 day 取数后按指标求和，hour 以 five_minute 取数后按整点小时桶聚合。前端 GranularityPicker 提供的 quarter/minute_x/hour_x 等其他粒度 DC 暂不支持，后端对这些粒度回退到 query_scalar 单点查询。

14. Cube 查询清理

原 Cube 查询体系（CubeQueryService、RealtimeMetricsRepo、metric_daily_agg 表、dim_channel 表）已在 Phase 1/2 中删除。补充清理：

前端 getLoginListByCube/getOrderListByCube 及相关类型定义已删除
QueryView.vue 已重构为使用 DC /v1/query/login 和 /v1/query/order 接口
后端 LoginCubeQueryParams/OrderCubeQueryParams/CubeQueryListResponse schema 已删除
report_api_source_resolver.py 中 login-by-cube/order-by-cube 分支已移除
AggregateQueryRuntimeReader 适配器类已删除

15. 风险

风险	缓解
DC API 变更	`DatacenterClient` 层隔离；`/aggregate/metrics` 动态发现指标
部分指标不可用	`Repo` 返回 `available=False`；看板降级
DC 故障	历史数据本地 PG 可用；当日回退 placeholder
`available=true` 实际为 0	下游加 `source` 辅助判断；已提报 DC 修复
ETL 存储方案变更	`Repo` 接口隔离，存储层变更不影响 Service
旧表清理影响分群/事件分析	通过 DC `/player-events` 接口（`/filter` + `/behavior` + `/aggregate` + `/catalog`）恢复分群和事件分析查询能力

16. 用户分群 & 事件分析 DC 接入

16.1 与 Dashboard 接入的差异

Dashboard 数据是聚合指标（DAU、收入等），需要 ETL 落库 + 双轨合并（历史 PG + 当日 DC）。

用户分群和事件分析不同：

特征	Dashboard	用户分群	事件分析
DC 接口	`/aggregate`	`/player-events/filter` + `/behavior` + `/segments/{id}`	`/player-events/aggregate` + `/player-events/catalog`
数据粒度	聚合值	player_id 列表	聚合行
ETL	需要（`datacenter_metric_daily`）	不需要	不需要
本地存储	PG 缓存历史数据	PG 配置表存 `segment_id` 引用	PG 配置表存分析参数
HybridRepo	是（历史 PG + 当日 DC）	不需要	不需要

16.2 架构总览

UserSegmentService                  EventAnalysisService
    │                                    │
    ▼                                    ▼
SegmentRepo (Protocol)              EventRepo (Protocol)
    │                                    │
    ├── DCSegmentRepo                  ├── DCEventRepo
    │   ├── POST /filter               │   ├── POST /catalog → Redis 1h
    │   ├── POST /behavior             │   └── POST /aggregate
    │   └── GET /segments/{id}         │
    │                                    │
    └── PG SegmentDefinitionRepo        └── PG EventAnalysisConfigRepo
        (segment_id 引用)                   (config JSON, 纯 CRUD)

核心原则：

不需要 ETL —— 分群结果由 DC 管理（segment_id），我们只存储 segment_id 引用到 PG 配置表
不需要 LocalPGRepo 和 HybridRepo —— Dashboard 的历史缓存 + 双轨合并模式不适用于分群和事件分析
Redis 缓存事件目录 —— /player-events/catalog 结果 1 小时有效
DC 不可用时 —— 检查 available 字段，false 时前端展示空状态 + reason
分群引用 —— 创建分群时调用 /filter 生成 segment_id，事件分析传 segment_id 给 /aggregate

事件分析已知限制

当前 /aggregate 最大查询窗口 30 天，跨月分析建议按月份拆分后由前端合并。
维度字段按 indexed/safe/expensive/sensitive 分级，前端选择维度时应隐藏/提示 expensive 维度。
事件漏斗与路径分析不在本期范围，依赖 DC 后续接口支持。

16.3 DCSegmentRepo 接口

class DCSegmentRepo:
    """DC /player-events 实现的分群数据访问。"""

    def __init__(self, dc_client: DatacenterClient):
        self._dc = dc_client

    async def create_segment(self, scope: Scope, request: SegmentFilterRequest) -> SegmentResult:
        """调用 /filter 生成分群，返回 segment_id + matched_count + sample players。"""
        return await self._dc.filter(
            game_id=scope.game_id,
            region=scope.region,
            conditions=request.profile_conditions,
            behavior_conditions=request.behavior_conditions,
            event_conditions=request.event_conditions,
            condition_logic=request.condition_logic,
            limit=request.sample_limit,
            return_player_ids=True,
        )

    async def get_segment_members(self, scope: Scope, segment_id: str, limit: int, offset: int) -> SegmentMembers:
        """调用 GET /segments/{segment_id} 分页获取成员列表。"""
        return await self._dc.get_segment_members(
            game_id=scope.game_id,
            region=scope.region,
            segment_id=segment_id,
            limit=limit,
            offset=offset,
        )

    async def query_behavior(self, scope: Scope, segment_id: str, metrics: list[BehaviorMetric]) -> BehaviorResult:
        """调用 /behavior 查询分群成员的行为聚合。"""
        return await self._dc.behavior(
            game_id=scope.game_id,
            region=scope.region,
            segment_id=segment_id,
            metrics=metrics,
        )

16.4 DCEventRepo 接口

class DCEventRepo:
    """DC /player-events 实现的事件分析数据访问。"""

    def __init__(self, dc_client: DatacenterClient, cache: Redis):
        self._dc = dc_client
        self._cache = cache

    async def get_catalog(self, scope: Scope, event_name: str | None = None) -> EventCatalog:
        """调用 /catalog，Redis 缓存 1 小时。"""
        cache_key = f"dc:event-catalog:{scope.game_id}:{scope.region}:{event_name or 'all'}"
        cached = await self._cache.get(cache_key)
        if cached:
            return json.loads(cached)
        result = await self._dc.player_events.catalog(
            game_id=scope.game_id,
            region=scope.region,
            event_name=event_name,
        )
        await self._cache.setex(cache_key, 3600, json.dumps(result))
        return result

    async def aggregate(self, scope: Scope, request: EventAggregateRequest) -> AnalysisResult:
        """调用 /aggregate 执行事件聚合分析。"""
        return await self._dc.player_events.aggregate(
            game_id=scope.game_id,
            region=scope.region,
            event_name=request.event_name,
            start_date=request.start_date,
            end_date=request.end_date,
            time_granularity=request.time_granularity,
            dimensions=request.dimensions,
            metrics=request.metrics,
            filters=request.filters,
            segment_id=request.segment_id,
            segment_player_ids=request.segment_player_ids,
            limit=request.limit,
        )

16.5 Service 层改造

UserSegmentService 和 EventAnalysisService 已通过依赖注入方式构造（__init__ 接受 repo 参数），改造点：

将本地 Repo 替换为 DC Repo
- SegmentProfileRepo（本地桩） → DCSegmentRepo
- EventAnalysisQueryRepo（本地桩） → DCEventRepo
移除 _ensure_deps 懒加载，改为构造时注入
condition_logic 字段支持 and/or（已支持）
行为条件预处理：behavior_conditions 中的 lookback_days 使用默认值（login 30 天 / payment 30 天 / last_* 90 天）
事件条件映射：scope + field → event_conditions（如 {"scope": "login", "field": "login_days"} → {"event_name": "login", "calculation": "count", "operator": ...}）

16.6 DatacenterClient 扩展

DatacenterClient 中 /player-events 端点方法已全部实现（详见第 8 节），包括：

player_events_filter() → POST /player-events/filter
player_events_behavior() → POST /player-events/behavior
player_events_aggregate() → POST /player-events/aggregate
player_events_catalog() → POST /player-events/catalog
player_events_detail() → POST /player-events/detail
get_segment_members() → GET /player-events/segments/{segment_id}

16.7 DC 接口契约

Repo 方法	DC 接口
`DCSegmentRepo.create_segment`	`POST /player-events/filter`
`DCSegmentRepo.get_segment_members`	`GET /player-events/segments/{segment_id}`
`DCSegmentRepo.query_behavior`	`POST /player-events/behavior`
`DCEventRepo.get_catalog`	`POST /player-events/catalog`
`DCEventRepo.aggregate`	`POST /player-events/aggregate`

16.8 不需要的组件

与 Dashboard 方案相比，以下组件在分群和事件分析中不需要：

ETL Worker — 分群结果由 DC 管理，事件聚合实时执行
LocalPGRepo — 不需要本地历史缓存
HybridRepo — 不需要双轨合并
datacenter_metric_daily 表 — 不需要额外落库

仅 ERD 层面的 PG 表继续使用：segment_definitions（存储 segment_id 引用）和 event_analysis_configs（存储分析参数）。

1. 设计要求​

2. 核心架构：Repo 接口 + 双轨数据​

2.1 架构总览​

2.2 分层原则​

2.3 Repo 接口定义​

2.4 Repo 实现​

2.5 RepoFactory 路由​

2.6 Service 层示例​

3. 固定看板数据流​

3.1 查询链路​

3.2 日监控 Tab 为例​

3.3 时间对齐​

4. 明细查询与自定义看板的 DC 接入​

4.1 当前查询类型映射​

4.2 明细查询改造​

4.3 自定义看板改造​

5. ETL 与本地存储​

5.1 ETL 职责​

5.2 存储方案：PG 新表 datacenter_metric_daily​

5.3 ETL Worker 流程​

6. 旧表清理记录​

6.1 表清理结果​

6.2 删除执行结果​

7. Redis 缓存策略​

7.1 缓存分层​

7.2 缓存 Key 规范​

7.3 TTL 动态策略​

7.4 缓存穿透/击穿防护​

7.5 缓存清理​

8. DatacenterClient 接口​

9. API Key 与配置管理​

10. Agent 查询（后置 Phase 3）​

11. 实施阶段​

Phase 1: Repo + 日监控 Tab（4 周）—— 已完成​

Phase 2: 全量 Tab + 明细查询 + 旧表清理（4 周）—— 已完成​

Phase 3（已完成）​

Phase 4（已完成）​

后置 backlog​

12. 降级策略​

13. DC 时序颗粒度支持​

14. Cube 查询清理​

15. 风险​

16. 用户分群 & 事件分析 DC 接入​

16.1 与 Dashboard 接入的差异​

16.2 架构总览​

事件分析已知限制​

16.3 DCSegmentRepo 接口​

16.4 DCEventRepo 接口​

16.5 Service 层改造​

16.6 DatacenterClient 扩展​

16.7 DC 接口契约​

16.8 不需要的组件​