风控系统指标计算/特征提取分析与实现01，Redis、Zset、模版方法-编程知识

个人博客：无奈何杨（wnhyang）

个人语雀：wnhyang

共享语雀：在线知识共享

Github：wnhyang - Overview

引用AI对于风控系统的介绍

风控系统是一种用于在线业务的安全管理系统，它帮助企业和平台防范潜在的欺诈、信用风险以及不合规行为。简单来说，它的核心作用就是“保安全、防欺诈、控风险”。

最近也一直在研究风控系统体系、功能等，看了一些有关的文章，并且也在实践尝试中。

其实前一篇可配置“输入参数的接口如何设计”就是实践尝试的一部分，未来还会有更多的。

而本篇文章就风控系统的指标计算，或者说是特征提取做一些探讨，以下统一称呼为“指标”。

指标不仅可以作为风控系统的一部分配合风控规则或是模型/机器学习使用，而且可以用于离线分析、事后追查、用户画像标签等方面。

参考文章

风控笔记06：一个完整的风控引擎，需要有哪些功能？

风控笔记07：最常用的风控工具-特征库

指标分类

指标是由数据流支撑的，指标是时间纬度的数据提取计算。

根据指标分类举几个例子：

次数统计：最近24小时 ${客户号}向$ {银行卡卡号}转账笔数
求和：最近2天 ${客户号}向$ {银行卡卡号}转账金额之和
平均：最近1个月 ${客户号}向$ {银行卡卡号}转账金额的平均数
关联次数：最近72小时 ${客户号}关联$ {设备mac地址}的次数
等等

指标类型枚举

/*** @author wnhyang* @date 2024/3/13**/
@AllArgsConstructor
@Getter
public enum IndicatorType {COUNT(0, "count", "次数统计"),SUM(1, "sum", "求和"),AVG(2, "avg", "平均"),MAX(3, "max", "最大值"),MIN(4, "min", "最小值"),ASS(5, "ass", "关联次数");private final Integer code;private final String name;private final String desc;
}

指标实体类

目前filterScript条件还未确实如何做，使用Groovy脚本还是什么，之后再定吧。

/*** @author wnhyang* @date 2024/3/13**/
@Data
@EqualsAndHashCode(callSuper = true)
@Builder
@NoArgsConstructor
@AllArgsConstructor
@TableName("de_indicator")
public class IndicatorPO extends BasePO {@Serialprivate static final long serialVersionUID = 1L;/*** 自增编号*/@TableId(value = "id", type = IdType.AUTO)private Long id;/*** 指标名*/@TableField("name")private String name;/*** 状态*/@TableField("status")private Boolean status;/*** 类型*/@TableField("type")private Integer type;/*** 计算字段*/@TableField("calc_field")private String calcField;/*** 窗口大小*/@TableField("win_size")private String winSize;/*** 窗口类型*/@TableField("win_type")private String winType;/*** 窗口数量*/@TableField("win_count")private Integer winCount;/*** 时间片*/@TableField("time_slice")private Long timeSlice;/*** 主字段*/@TableField("master_field")private String masterField;/*** 从字段*/@TableField("slave_fields")private String slaveFields;/*** 过滤脚本*/@TableField("filter_script")private String filterScript;/*** 版本号*/@TableField("version")private Integer version;/*** 描述*/@TableField("description")private String description;
}

指标窗口枚举

/*** @author wnhyang* @date 2024/3/13**/
@AllArgsConstructor
@Getter
public enum WinType {LAST(0, "last", "最近"),CUR(1, "cur", "本");private final Integer code;private final String name;private final String desc;
}

指标存储

指标数据如何存储呢？下面是Redis方案。

1、使用Redis的有序集合（Sorted Set）结构，有序集合中的每个元素都有一个分数（score），这里的分数我们可以设置为时间戳。

2、添加数据：每当有新的请求到来时，将当前时间戳作为score，用一个固定的字符串（如"request"）或者其他唯一标识符作为member，插入到有序集合中。

3、清理过期数据：每次添加新数据后，通过ZRANGEBYSCORE命令获取并删除窗口范围之外的数据。

4、统计指标：要得到窗口内的请求次数，可以直接使用ZCARD命令获取有序集合中元素的数量。

5、定时任务：为了确保过期数据能够自动清除，可以结合Redis的Key空间通知机制（Keyspace Notifications）或者外部定时任务定期执行上述清理操作。

计算类指标

计算类指标比较通用，需要存储的数据很容易分析。

但还是有些差别的，次数统计可以在zset中存储value为${事件id}，score为时间戳，计算时为zset的size，但是对于平均、最大值、最小值、求和不能这么做，因为这些都是有计算字段的，并不是如次数统计那样取size就行，所以对于这类指标数据存储是不一样的。

单zset和zset+hash两种方案。

单zset

zset作为时间窗口，value为${事件id}+{计算字段}，score为时间戳，变化就是value变为事件id与计算字段的组合，计算字段这样就存储下来了，之后计算平均、最大值、最小值、求和取出来再算就可以了，事件id是为了在zset中存储时防重，另外也方便找到原始数据。

优点：直接从zset中获取计算字段，可以独立手动过期删除。

缺点：value存和取需要设计，数据冗余大。

zset+hash

zset作为时间窗口，value为${事件id}，score为时间戳，这里没有变化，另外需要hash存储对应事件下需要计算字段的数据。

优点：hash可以存储多项数据，数据冗余少。

缺点：无法确定事件过期删除时间，每次需要多步查询。

指标计算与查询

指标计算可以简单梳理如下。

以下根据此流程分析，其实编程（也不全是指编程）有趣的地方是设计和实现的过程。

模版方法

设计模式-模版方法可以在这应用，从上面的流程并结合指标的分类来分析，主流程中的判断指标状态、指标条件、获取当前时间戳、获取redis数据设置过期、清理过期数据都是通用的，根据指标类型变化的只有添加事件这个步骤，所以定义模版抽象类如下：

@Setter
@Getter
@Slf4j
public abstract class AbstractIndicator {/*** 指标*/protected IndicatorPO indicator;/*** 指标类型*/protected final IndicatorType INDICATOR_TYPE;/*** redisson客户端*/protected final RedissonClient redissonClient;protected AbstractIndicator(IndicatorType indicatorType, RedissonClient redissonClient) {INDICATOR_TYPE = indicatorType;this.redissonClient = redissonClient;}/*** 获取指标类型** @return 指标类型*/public Integer getType() {return INDICATOR_TYPE.getCode();}/*** 获取指标状态** @return true/false*/public boolean getStatus() {return indicator.getStatus();}/*** 指标过滤** @return true/false*/public boolean filter(Map<String, String> eventDetail) {// 1、主属性、从属性不为空if (indicator.getMasterField() != null && eventDetail.get(indicator.getMasterField()) != null) {if (indicator.getSlaveFields() != null) {String[] split = indicator.getSlaveFields().split(",");for (String s : split) {if (eventDetail.get(s) == null) {return false;}}// 2、过滤脚本if (indicator.getFilterScript() == null) {return true;} else {// TODO 脚本过滤return true;}}}return false;}/*** 获取redis key** @param eventDetail 事件详情* @return redis key*/public String getRedisKey(Map<String, String> eventDetail) {return RedisKeys.INDICATOR + indicator.getId() + ":" + INDICATOR_TYPE.getName() + ":" + eventDetail.get(indicator.getMasterField()) + "-" + eventDetail.get(indicator.getSlaveFields());}/*** 获取计算指标结果** @param currentTime 当前时间戳* @param set         redis set* @return 计算指标结果*/public abstract BigDecimal getResult(long currentTime, RScoredSortedSet<String> set);/*** 获取计算指标结果** @param eventDetail 事件详情* @return 计算指标结果*/public BigDecimal getResult(Map<String, String> eventDetail) {// 1、获取当前时间戳long currentTime = System.currentTimeMillis();// 2、获取redis中数据RScoredSortedSet<String> set = redissonClient.getScoredSortedSet(getRedisKey(eventDetail));// 3、清理过期数据if ("last".equals(indicator.getWinType())) {set.removeRangeByScore(-1, true, currentTime - Duration.ofSeconds(indicator.getTimeSlice()).toMillis(), false);} else {set.removeRangeByScore(-1, true, calculateEpochMilli(LocalDateTime.now()), false);}return getResult(currentTime, set);}/*** 计算指标** @param indicator   指标* @param eventDetail 事件详情*/public void compute(IndicatorPO indicator, Map<String, String> eventDetail) {if (indicator == null) {return;} else {this.indicator = indicator;}// 1、状态检查和过滤if (getStatus() && filter(eventDetail)) {// 2、获取当前时间戳long currentTime = System.currentTimeMillis();// 3、获取redis中数据log.info("redisKey:{}", getRedisKey(eventDetail));RScoredSortedSet<String> set = redissonClient.getScoredSortedSet(getRedisKey(eventDetail));if ("last".equals(this.indicator.getWinType())) {set.expire(Duration.ofSeconds(this.indicator.getTimeSlice() * this.indicator.getWinCount()));} else {set.expire(Duration.ofSeconds(this.indicator.getTimeSlice()));}// 4、添加事件addEvent(currentTime, set, eventDetail);// 5、清理过期数据cleanExpiredDate(currentTime, set);}}/*** 添加事件** @param currentTime 当前时间戳* @param set         redis set* @param eventDetail 事件详情*/public abstract void addEvent(long currentTime, RScoredSortedSet<String> set, Map<String, String> eventDetail);/*** 清理过期数据** @param currentTime 当前时间戳* @param set         redis set*/public void cleanExpiredDate(long currentTime, RScoredSortedSet<String> set) {if ("last".equals(indicator.getWinType())) {set.removeRangeByScore(-1, true, currentTime - Duration.ofSeconds(indicator.getTimeSlice()).toMillis(), false);} else {set.removeRangeByScore(-1, true, calculateEpochMilli(LocalDateTime.now()), false);}}/*** 计算时间戳** @param now 当前时间* @return 时间戳*/public long calculateEpochMilli(LocalDateTime now) {ZoneId zoneId = ZoneId.systemDefault();// 这个default分支仅处理WindowSize枚举中未包含的情况return switch (indicator.getWinSize()) {case "M" -> now.withDayOfMonth(1).with(LocalTime.MIN).atZone(zoneId).toInstant().toEpochMilli();case "d" -> now.with(LocalTime.MIN).atZone(zoneId).toInstant().toEpochMilli();case "H" -> now.withMinute(0).withSecond(0).withNano(0).atZone(zoneId).toInstant().toEpochMilli();case "m" -> now.withSecond(0).withNano(0).atZone(zoneId).toInstant().toEpochMilli();case "s" -> now.withNano(0).atZone(zoneId).toInstant().toEpochMilli();default -> throw new IllegalArgumentException("Unsupported window size: " + indicator.getWinSize());};}
}

次数统计指标

因为篇幅原因，这里只贴次数统计指标实现类了。

/*** @author wnhyang* @date 2024/3/11**/
@Component
public class CountIndicator extends AbstractIndicator {public CountIndicator(RedissonClient redissonClient) {super(IndicatorType.COUNT, redissonClient);}@Overridepublic BigDecimal getResult(long currentTime, RScoredSortedSet<String> set) {return BigDecimal.valueOf(set.size());}@Overridepublic void addEvent(long currentTime, RScoredSortedSet<String> set, Map<String, String> eventDetail) {set.add(currentTime, eventDetail.get("seqId"));}}