selfrelease
74e35fed96
- 新增场景基类、配置加载、注册表与原语模块 - 添加 r10_refund_split 规则及场景 JSON Schema - 扩展 scan 引擎与 scenarios API - 新增场景注册表/配置/集成测试 - 更新前端 App、api、labels 支持新场景
136 lines
4.6 KiB
Python
136 lines
4.6 KiB
Python
"""场景二 · 市场业务真实性：养卡骗补检测（R9）。
|
||
|
||
检测"脉冲式增长 + 规律性衰减"的周期性造假：渠道每月新增大量用户订购，
|
||
固定周期后这些用户集中退订（骗补后弃养）。结合佣金与业务质量匹配度。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass
|
||
|
||
from app.scenarios.base import BaseScenario, ClueDraft, ScanOutcome
|
||
from app.scenarios.registry import register_scenario
|
||
|
||
|
||
@dataclass
|
||
class CohortPoint:
|
||
"""某新增批次（cohort）在第 N 个月的留存率。"""
|
||
|
||
month_index: int
|
||
retention: float # 0-1
|
||
|
||
|
||
@dataclass
|
||
class ChurnFinding:
|
||
cliff_month: int | None
|
||
max_drop: float
|
||
pulse_then_decay: bool
|
||
|
||
|
||
def detect_pulse_decay(
|
||
retention_curve: list[CohortPoint],
|
||
cliff_drop: float = 0.5,
|
||
) -> ChurnFinding:
|
||
"""识别留存曲线中的"断崖式集中退订"。
|
||
|
||
若某月留存相对上月骤降超过 cliff_drop（默认 50%），判为规律性衰减。
|
||
"""
|
||
ordered = sorted(retention_curve, key=lambda p: p.month_index)
|
||
max_drop = 0.0
|
||
cliff_month: int | None = None
|
||
for prev, cur in zip(ordered, ordered[1:], strict=False):
|
||
drop = prev.retention - cur.retention
|
||
if drop > max_drop:
|
||
max_drop = drop
|
||
if drop >= cliff_drop:
|
||
cliff_month = cur.month_index
|
||
return ChurnFinding(
|
||
cliff_month=cliff_month,
|
||
max_drop=round(max_drop, 3),
|
||
pulse_then_decay=cliff_month is not None,
|
||
)
|
||
|
||
|
||
def commission_quality_mismatch(
|
||
commission_paid: float,
|
||
active_ratio: float,
|
||
zero_usage_ratio: float,
|
||
) -> float:
|
||
"""佣金与业务质量不匹配度（0-1）。
|
||
|
||
active_ratio：仍活跃用户占比；zero_usage_ratio：零通话/零流量用户占比。
|
||
佣金已发但活跃低、零使用高 → 不匹配度高。
|
||
"""
|
||
if commission_paid <= 0:
|
||
return 0.0
|
||
mismatch = 0.6 * zero_usage_ratio + 0.4 * (1 - active_ratio)
|
||
return round(min(max(mismatch, 0.0), 1.0), 3)
|
||
|
||
|
||
def churn_risk_score(finding: ChurnFinding, mismatch: float) -> float:
|
||
"""综合评分：断崖退订 + 佣金质量不匹配。"""
|
||
if not finding.pulse_then_decay:
|
||
return round(0.3 * mismatch, 3)
|
||
base = 0.4 + 0.4 * finding.max_drop + 0.2 * mismatch
|
||
return round(min(base, 1.0), 3)
|
||
|
||
|
||
def build_rationale(finding: ChurnFinding, mismatch: float) -> str:
|
||
if finding.pulse_then_decay:
|
||
return (
|
||
f"渠道新增用户在第 {finding.cliff_month} 个月出现断崖式集中退订"
|
||
f"（最大单月留存骤降 {finding.max_drop:.0%}），呈"
|
||
f"'脉冲式增长 + 规律性衰减'特征；佣金与业务质量不匹配度 {mismatch:.0%}，"
|
||
f"高度疑似养卡骗补（骗补后弃养）。"
|
||
)
|
||
return (
|
||
f"未见明显断崖退订，但佣金与业务质量不匹配度为 {mismatch:.0%}，建议关注。"
|
||
)
|
||
|
||
|
||
@register_scenario
|
||
class ChurnFraudScenario(BaseScenario):
|
||
"""场景二养卡骗补检测的插件封装：时序断崖 + 佣金质量不匹配 → 线索草稿。"""
|
||
|
||
code = "R9"
|
||
title = "疑似养卡骗补（脉冲增长+规律退订）"
|
||
risk_domain = "成本"
|
||
label = "养卡骗补"
|
||
score_threshold = 0.5 # 与原 run_churn_scan 的 score>=0.5 一致
|
||
|
||
def __init__(
|
||
self,
|
||
retention_curve: list[CohortPoint],
|
||
commission_paid: float,
|
||
active_ratio: float,
|
||
zero_usage_ratio: float,
|
||
channel_key: str,
|
||
) -> None:
|
||
self.retention_curve = retention_curve
|
||
self.commission_paid = commission_paid
|
||
self.active_ratio = active_ratio
|
||
self.zero_usage_ratio = zero_usage_ratio
|
||
self.channel_key = channel_key
|
||
|
||
def scan(self, session, *, data_version_id=None) -> ScanOutcome:
|
||
finding = detect_pulse_decay(self.retention_curve)
|
||
mismatch = commission_quality_mismatch(
|
||
self.commission_paid, self.active_ratio, self.zero_usage_ratio
|
||
)
|
||
score = churn_risk_score(finding, mismatch)
|
||
draft = ClueDraft(
|
||
score=score,
|
||
rationale=build_rationale(finding, mismatch),
|
||
evidence={
|
||
"cliff_month": finding.cliff_month,
|
||
"max_drop": finding.max_drop,
|
||
"commission_paid": self.commission_paid,
|
||
"active_ratio": self.active_ratio,
|
||
"zero_usage_ratio": self.zero_usage_ratio,
|
||
"mismatch": mismatch,
|
||
},
|
||
subjects={"channel": self.channel_key},
|
||
amount_involved=self.commission_paid,
|
||
)
|
||
return ScanOutcome(scanned_count=len(self.retention_curve), drafts=[draft])
|