基于准实时规则引擎的业务风控方案

risk:order-patient-id:123456     
|--111111        1652929153     
|--222222        1652932753     
|--333333        1652939953     

// startTime 和 endTime 对应为1天时间间隔     
ZCOUNT risk:order-patient-id:123456 1652940185 1652853785     
3     

CREATE TABLE `risk_input_order` (     
    `id`            INT,     
    `pid`           INT,     
    `doctor_id`     INT,     
    `patient_id`    INT,     
    `deliver_phone` VARCHAR(20),     
    `deliver_name`  VARCHAR(20),     
    `delivery_address` VARCHAR(200),     
    `prescription_id`  INT,     
    `total_price`      INT,     
    `is_test`          TINYINT,     
    `created_at`       TIMESTAMP     
) WITH (     
    -- 定义 Kafka 参数     
    'connector' = 'kafka',     
    'topic' = 'med_dts_b_convert',  -- 消费的 Topic     
    'scan.startup.mode' = 'latest-offset', -- 可以是 latest-offset / earliest-offset / specific-offsets / group-offsets / timestamp 的任何一种     
    'properties.bootstrap.servers' = '172.28.28.13:9092',  -- Kafka 连接地址     
    'properties.group.id' = 'risk_input_order',            -- 指定 Group ID     
     
    -- 定义数据格式 (JSON 格式)     
    'format' = 'json',     
    'json.fail-on-missing-field' = 'false',  -- 遇到缺失字段不会报错     
    'json.ignore-parse-errors' = 'true'      -- 忽略任何解析报错     
);     

CREATE TABLE `risk_output_order_patient_id` (       
    `key`   STRING,     
    `value` STRING     
    `score` DOUBLE     
) WITH (     
    'connector' = 'redis',               
    'command' = 'zadd',              -- redis zadd命令写入数据     
    'nodes' = '192.28.28.217:6379',  -- redis连接地址。     
    'password' = 'yourpassword'        
);     

-- 清洗患者id维度订单数据     
INSERT INTO risk_output_order_patient_id     
SELECT     
CONCAT('risk:order-patient-id:', patient_id) AS `key`,     
id AS value,     
UNIX_TIMESTAMP(created_at) AS score     
FROM risk_input_order     
WHERE pid = 0;     

CREATE TABLE `risk_input_order` (     
) WITH (     
    -- 定义 Kafka 参数     
    'connector' = 'kafka',     
    'scan.startup.mode' = 'earliest-offset', -- 最早消费位点     
);     

// 订单风控     
service RiskOrder {     
    // 风险识别     
    rpc CheckRisk (CheckRiskReq) returns (CheckRiskResp);     
    // 数据/事件上报     
    rpc Report (ReportReq) returns (ReportResp);     
}     

// 规则名必须唯一     
rule "rulename" "rule-describtion" salience priority     
begin     
     
//规则体     
     
end     

rule "pa-daily-order-count" "一天内同一患者ID订单数规则" salience 10     
begin     
if GetData("order-patient-id", Patient.UserId, 24*3600) > 5 {     
    return // 命中规则     
}     
end     

// 获取事实数据     
user := xuser.GetUserInfo(userId)     
patient := xuser.GetPatientInfo(userId)     
     
dataContext := context.NewDataContext()     
// 事实数据注入     
dataContext.Add("User", user)     
dataContext.Add("Patient", patient)     

// 获取指标数据函数     
func (s *DataService) GetData(indicator string, subjectId string, period int64) int64 {     
    key := fmt.sprintf("risk:%s:%s", indicator, subjectId)     
    endTime := time.Now().Unix()     
    startTime := endTime - period     
         
    err, res := redis.SCount(key, startTime, endTime)     
    if err != nil {     
        // 注入函数会忽略 error 返回值，通过错误标志来区分报错和默认值的情况     
        s.SetError(err)     
        return 0     
    }     
     
    return res     
}     
     
// 函数注入规则引擎     
dataContext := context.NewDataContext()     
dataSvc := &DataService{}     
dataContext.Add("GetData", dataSvc.GetData)     

dataContext := context.NewDataContext()     
// 数据、函数注入     
     
var rules []string     
// rules from MySQL...     
     
// 规则编译     
ruleBuilder := builder.NewRuleBuilder(dataContext)     
for _, r := range rules {     
    if err := ruleBuilder.BuildRuleWithIncremental(r); nil != err {     
        return err     
    }     
}     
     
// 规则并行模式执行     
eng := engine.NewGengine()     
if err := eng.ExecuteConcurrent(ruleBuilder); nil != err {     
    return err     
}     
if dataSvc.HasError() {     
    // 规则执行过程中，获取数据发生的报错     
    return dataSvc.FirstError()     
}     
     
// 执行结果 map[命中规则ID]interface{}     
resMap, _ := eng.GetRulesResultMap()