Files
SMANAGE/docs/stats_engine.md
T

14 KiB
Raw Blame History

SMANAGE 每日数据统计引擎 — 设计文档

一、设计背景

通过对阿龙电竞平台统计机制的完整分析,提取出以下 7 种核心统计模式:

模式 阿龙实现 本质
实时多角色日统计 6 张独立日统计表(DashouRiTongji 等) 按日期+角色+用户 聚合计数/金额
即时计数器 jinrijiedan/jinrishouyi(今日接单/今日收益) 内存/实时字段累加
累计统计 jiedanzongliang/chengjiaozongliang 永不清零的计数
定时清零 celery beat daily_reset_task 每天 0 点清今日计数器
平台流水 Ptliushui 单行聚合表 平台级汇总快照
排行榜 RankingRecord 日榜/月榜 排序快照,前端直接读
行为触发更新 update_dashou_daily_by_action() 业务事件驱动统计

核心洞察:所有统计都是 "指标 + 维度 + 时间窗口 → 数值" 的变体。


二、设计原则

原则 说明
指标可定义 租户通过 DefineMetric() 自定义统计什么(计数/金额/时长/比率)
维度可配置 统计按任意维度切分(按角色/按平台/按渠道/按用户),JSON 存储
原子增量 使用 SELECT FOR UPDATE + F() 实现并发安全的原子累加
时间窗口化 日统计/时统计/月统计统一处理,PeriodKey 区分
排行榜快照 定时生成排名快照,前端直接查询,不实时计算
周期重置 可配置的每日/每月清零规则
租户隔离 全部操作通过 X-Tenant-UUID 头识别租户

三、架构分层

子服务(如阿龙电竞)
    │
    │  订单完成 → stats.Increment('dashou_jiedan', {'role':'dashou','user_id':'123'}, 1)
    │  充值到账 → stats.IncrementMulti([...])
    │  会员购买 → stats.Increment('vip_sale', {'level':'vip1'}, 1)
    │
    └──→ StatEngine
            │
            ├── 写: AtomicIncrement() → StatRecord(并发安全)
            ├── 读: QueryToday() / QueryRange() / AggregateTopN()
            ├── 排: GenerateDailyRanking() → StatRanking(快照)
            ├── 清: DailyReset() / MonthlyReset()(定时任务触发)
            └── 配: DefineMetric() / ConfigureAutoReset()

四、文件清单(10 个文件)

层级 文件 职责
模型 models.py 5 张表:Metric / Dimension / Record / Ranking / Config
引擎 engine.py 7 个原子函数:AtomicIncrement / QueryPeriod / AggregateByDimension / GenerateRanking
编排器 manager.py StatEngine — 15 个方法覆盖增删查改排清配
API views.py 10 个 REST 端点
序列化 serializers.py 14 个 Serializer
路由 urls.py 10 个 URL 路由
配置 apps.py AppConfig
管理 admin.py 5 个 Django Admin 注册

五、核心数据模型

5.1 StatMetric — 统计指标定义

租户自定义要统计什么:

字段 类型 说明
MetricUUID UUID 主键
TenantUUID UUID 租户
MetricCode CharField(64) 指标编码(dashou_jiedan)
MetricName CharField(128) 指标名称(打手接单数)
MetricType CharField(16) count/amount/duration/ratio/custom
Unit CharField(16) 单位(次/元/分钟)
DefaultValue DecimalField 默认值

唯一约束:(TenantUUID, MetricCode)

5.2 StatDimension — 统计维度定义

按什么切分统计:

字段 类型 说明
DimensionCode CharField(64) 维度编码(role, platform, user_id)
DimensionName CharField(128) 维度名称(角色, 平台, 用户)

5.3 StatRecord — 统计记录(核心表)

按维度+时间窗口聚合的统计数据:

字段 类型 说明
RecordUUID UUID 主键
TenantUUID UUID 租户
PeriodType CharField(16) daily / hourly / monthly
PeriodKey CharField(32) 周期键(2026-06-11 / 2026-06)
Metric ForeignKey 关联指标
DimensionValues JSONField 维度值 {"role":"dashou","user_id":"123"}
Value DecimalField 统计值

唯一约束:(TenantUUID, PeriodType, PeriodKey, Metric, DimensionValues)

单条记录示例:

TenantUUID='al_long', PeriodType='daily', PeriodKey='2026-06-11',
Metric='dashou_jiedan', DimensionValues={'role':'dashou','user_id':'123'},
Value=5  → 打手123在2026-06-11接了5单

5.4 StatRanking — 排行榜快照

字段 类型 说明
RankingPeriod CharField(16) daily / monthly
RankingDate DateField 排行日期
Metric ForeignKey 按什么指标排
DimensionFilter JSONField 过滤条件(如只排打手)
TargetID CharField(128) 排行对象ID
Rank IntegerField 名次(1开始)
RankValue DecimalField 排行值
TargetName CharField(128) 对象名
TargetAvatar CharField(512) 对象头像URL

5.5 StatConfig — 统计行为配置

字段 类型 说明
Metric ForeignKey 关联指标
AutoReset BooleanField 是否每日自动清零
ResetCron CharField(64) 清零时间cron
EnableRanking BooleanField 是否生成排行榜
RankingTopN IntegerField 排行取前N名
RetentionDays IntegerField 数据保留天数

六、原子增量引擎

核心函数 AtomicIncrement:

def AtomicIncrement(tenant_uuid, period_type, period_key, metric_uuid,
                    dimension_values, delta):
    """使用 SELECT FOR UPDATE + F() 实现原子累加"""
    with transaction.atomic():
        existing = StatRecord.objects.select_for_update().filter(...)
        if existing:
            record.Value = F('Value') + delta  # 数据库层面原子累加
            record.save(update_fields=['Value'])
        else:
            record = StatRecord.objects.create(..., Value=delta)
    return record, created

并发安全保证:

  • select_for_update() 对匹配的行加排他锁,防止并发写竞争
  • F('Value') + delta 在数据库层面完成累加,避免读-改-写竞争
  • transaction.atomic() 确保锁在事务结束前持有

七、StatEngine 编排器

class StatEngine:
    def __init__(self, tenant_uuid)

    # 指标管理
    def DefineMetric(metric_code, metric_name, metric_type, unit)     # 定义统计指标
    def DefineDimension(dimension_code, dimension_name)                # 定义统计维度
    def GetMetrics() / GetDimensions()                                  # 查询

    # 配置
    def ConfigureAutoReset(metric_uuid, auto_reset, enable_ranking, ...)  # 配置重置/排行

    # 数据写入
    def Increment(metric_uuid, dimension_values, delta, date)           # 原子增量
    def IncrementMulti(increments, date)                                 # 批量原子增量
    def Decrement(metric_uuid, dimension_values, delta, date)           # 原子减量

    # 数据查询
    def QueryToday(metric_uuid, dimension_filter)                        # 今日统计
    def QueryDate(metric_uuid, date, period_type)                        # 指定日期
    def QueryRange(metric_uuid, start_date, end_date)                    # 时间范围
    def AggregateTopN(metric_uuid, start_date, end_date, dimension_key)  # 聚合排名

    # 排行榜
    def GenerateDailyRanking(metric_uuid, ranking_date, top_n)           # 生成日榜
    def GenerateMonthlyRanking(metric_uuid, ranking_date, top_n)         # 生成月榜
    def QueryRanking(ranking_period, ranking_date, metric_uuid)          # 查询排行榜

    # 维护
    def DailyReset(metric_uuid, target_date)                             # 手动日清
    def MonthlyReset(metric_uuid, target_date)                           # 手动月清
    def RunScheduledResets(target_date)                                  # 定时清零(Celery调用)
    def BuildRankings(target_date)                                       # 定时生成排行(Celery调用)
    def PurgeOld(before_date, period_type)                               # 清理旧数据

八、API 端点

# 指标与维度管理
GET    /api/stats/metric/              # 查询所有指标
POST   /api/stats/metric/              # 创建统计指标

GET    /api/stats/dimension/           # 查询所有维度
POST   /api/stats/dimension/           # 创建统计维度

# 数据写入
POST   /api/stats/increment/           # 单条原子增量
POST   /api/stats/increment/multi/     # 批量原子增量

# 数据查询
GET    /api/stats/query/               # 查询统计记录
GET    /api/stats/aggregate/           # 聚合排名(实时计算)

# 排行榜
GET    /api/stats/ranking/             # 查询排行榜快照
POST   /api/stats/ranking/             # 手动生成排行榜

# 配置
GET    /api/stats/config/              # 查询统计配置
POST   /api/stats/config/              # 创建/更新统计配置

# 维护
POST   /api/stats/reset/               # 手动清零
POST   /api/stats/purge/               # 清理过期数据

九、子服务集成示例

9.1 初始化统计体系

from gvsds.apps.stats.manager import StatEngine

engine = StatEngine(tenant_uuid='al_long_001')

# 定义指标
jiedan = engine.DefineMetric('dashou_jiedan', '打手接单数', 'count', '单')
shouru = engine.DefineMetric('dashou_shouru', '打手收入', 'amount', '元')
chongzhi = engine.DefineMetric('guanshi_chongzhi', '管事充值人数', 'count', '人')
vip = engine.DefineMetric('vip_sale', '会员销量', 'count', '份')

# 定义维度
engine.DefineDimension('role', '角色')
engine.DefineDimension('user_id', '用户ID')
engine.DefineDimension('platform', '平台')
engine.DefineDimension('order_type', '订单类型')

# 配置每日清零 + 排行榜
engine.ConfigureAutoReset(
    metric_uuid=jiedan.MetricUUID,
    auto_reset=True,
    enable_ranking=True,
    ranking_top_n=50,
)

9.2 订单完成时写统计

# 订单完成 → 写多条统计(原子批量)
engine.IncrementMulti(increments=[
    (jiedan.MetricUUID,  {'role': 'dashou', 'user_id': 'd_123'},  1),
    (jiedan.MetricUUID,  {'role': 'dashou', 'user_id': 'd_123', 'order_type': 'platform'}, 1),
    (jiedan.MetricUUID,  {'role': 'guanshi', 'user_id': 'g_456'}, 1),
    (shouru.MetricUUID,  {'role': 'dashou', 'user_id': 'd_123'},  80.00),
    (shouru.MetricUUID,  {'role': 'platform'},                    20.00),
])

9.3 查询今日统计

# 查询打手123今日统计
records = engine.QueryToday(
    metric_uuid=jiedan.MetricUUID,
    dimension_filter={'user_id': 'd_123'},
)
# → [{period_key:'2026-06-11', dimension_values:{...}, value: 5}]

# 查询今日所有打手排名(实时聚合)
top = engine.AggregateTopN(
    metric_uuid=jiedan.MetricUUID,
    start_date=datetime.date.today(),
    end_date=datetime.date.today(),
    dimension_key='user_id',
    top_n=10,
)
# → [{dim_value:'d_123', total: 5}, {dim_value:'d_456', total: 3}, ...]

9.4 排行榜

# 生成今日排行榜快照(定时任务调用)
engine.GenerateDailyRanking(
    metric_uuid=jiedan.MetricUUID,
    dimension_filter={'role': 'dashou'},
    top_n=50,
)

# 前端查询日榜
rankings = engine.QueryRanking(
    ranking_period='daily',
    ranking_date=datetime.date.today(),
    metric_uuid=jiedan.MetricUUID,
)
# → [{target_id:'d_123', rank:1, rank_value:15, target_name:'张三', target_avatar:'...'}, ...]

9.5 Celery 定时任务

# celery.py 中注册每日任务
@app.task
def stats_daily_reset():
    """每日统计清零 + 排行榜生成"""
    from gvsds.apps.stats.models import StatConfig
    from gvsds.apps.stats.manager import StatEngine

    tenants = StatConfig.objects.filter(AutoReset=True).values_list('TenantUUID', flat=True).distinct()
    for tenant in tenants:
        engine = StatEngine(tenant)
        engine.RunScheduledResets()
        engine.BuildRankings()

9.6 退款回滚统计

# 退款 → 回滚相关统计
engine.Decrement(
    metric_uuid=jiedan.MetricUUID,
    dimension_values={'role': 'dashou', 'user_id': 'd_123'},
    delta=1,
)
engine.Decrement(
    metric_uuid=shouru.MetricUUID,
    dimension_values={'role': 'dashou', 'user_id': 'd_123'},
    delta=80.00,
)

十、与阿龙电竞模式的对比

阿龙模式 SMANAGE 抽象
6 张独立日统计表 1 张 StatRecord + 维度字段区分
硬编码的 jinri/jinyue 字段 可定义的 Metric(任意指标)
固定角色(打手/管事/商家) JSON DimensionValues(任意维度)
celery beat 定时清零 ConfigureAutoReset() 配置化
RankingRecord 独立表 StatRanking 快照表 + Generate 方法
手动 update_daily 函数 AtomicIncrement + F() 原子累加
Ptliushui 单行汇总 AggregateByDimension / QueryRange

十一、配置变更记录

文件 变更
settings.py 新增 'gvsds.apps.stats' 到 INSTALLED_APPS
db_router.py 新增 'stats': 'default' 到 app_db_map
urls.py 新增 path('api/stats/', include('gvsds.apps.stats.urls'))
gvsds/apps/stats/engine.py 已删除(2026-08-07):8 个模块级函数并入 gvsdsdk StatsManager;StatEngine(gvsds/apps/stats/manager.py)与 Celery 任务底层统一走 gvsdsdk.stats.manager.StatsManager,签名 1:1 不变