接入异常账户命中流水主链路

lsfx-mock-server/services/statement_rule_samples.py

@@ -953,6 +953,39 @@ PHASE2_STATEMENT_RULE_BUILDERS = {
     "SALARY_UNUSED": build_salary_unused_samples,
 }
 
+ABNORMAL_ACCOUNT_RULE_BUILDERS = {
+    "SUDDEN_ACCOUNT_CLOSURE": build_sudden_account_closure_samples,
+    "DORMANT_ACCOUNT_LARGE_ACTIVATION": build_dormant_account_large_activation_samples,
+}
+
+
+def _resolve_abnormal_account_fact(rule_code: str, abnormal_accounts: List[Dict]) -> Optional[Dict]:
+    for account_fact in abnormal_accounts:
+        if account_fact.get("rule_code") == rule_code:
+            return account_fact
+
+    if rule_code == "SUDDEN_ACCOUNT_CLOSURE":
+        return next(
+            (
+                account_fact
+                for account_fact in abnormal_accounts
+                if account_fact.get("status") == 2 and account_fact.get("invalid_date")
+            ),
+            None,
+        )
+
+    if rule_code == "DORMANT_ACCOUNT_LARGE_ACTIVATION":
+        return next(
+            (
+                account_fact
+                for account_fact in abnormal_accounts
+                if account_fact.get("status") == 1 and account_fact.get("effective_date")
+            ),
+            None,
+        )
+
+    return None
+
 
 def build_seed_statements_for_rule_plan(
     group_id: int,
@@ -961,21 +994,36 @@ def build_seed_statements_for_rule_plan(
     **kwargs,
 ) -> List[Dict]:
     statements: List[Dict] = []
+    abnormal_accounts = list(kwargs.get("abnormal_accounts") or [])
+    common_kwargs = {key: value for key, value in kwargs.items() if key != "abnormal_accounts"}
 
     for rule_code in rule_plan.get("large_transaction_hit_rules", []):
         builder = LARGE_TRANSACTION_BUILDERS.get(rule_code)
         if builder is not None:
-            statements.extend(builder(group_id, log_id, **kwargs))
+            statements.extend(builder(group_id, log_id, **common_kwargs))
 
     for rule_code in rule_plan.get("phase1_hit_rules", []):
         builder = PHASE1_RULE_BUILDERS.get(rule_code)
         if builder is not None:
-            statements.extend(builder(group_id, log_id, **kwargs))
+            statements.extend(builder(group_id, log_id, **common_kwargs))
 
     for rule_code in rule_plan.get("phase2_statement_hit_rules", []):
         builder = PHASE2_STATEMENT_RULE_BUILDERS.get(rule_code)
         if builder is not None:
-            statements.extend(builder(group_id, log_id, **kwargs))
+            statements.extend(builder(group_id, log_id, **common_kwargs))
+
+    for rule_code in rule_plan.get("abnormal_account_hit_rules", []):
+        builder = ABNORMAL_ACCOUNT_RULE_BUILDERS.get(rule_code)
+        account_fact = _resolve_abnormal_account_fact(rule_code, abnormal_accounts)
+        if builder is not None and account_fact is not None:
+            statements.extend(
+                builder(
+                    group_id,
+                    log_id,
+                    account_fact=account_fact,
+                    le_name=common_kwargs.get("primary_enterprise_name", "模型测试主体"),
+                )
+            )
 
     return statements
 

lsfx-mock-server/services/statement_service.py

@@ -166,6 +166,9 @@ class StatementService:
             "phase2_statement_hit_rules": (
                 list(record.phase2_statement_hit_rules) if record is not None else []
             ),
+            "abnormal_account_hit_rules": (
+                list(record.abnormal_account_hit_rules) if record is not None else []
+            ),
         }
         if record is not None and record.staff_id_card:
             allowed_identity_cards = tuple([record.staff_id_card, *record.family_id_cards])
@@ -180,6 +183,7 @@ class StatementService:
             primary_account_no=primary_account_no,
             staff_id_card=record.staff_id_card if record is not None else None,
             family_id_cards=record.family_id_cards if record is not None else None,
+            abnormal_accounts=record.abnormal_accounts if record is not None else None,
         )
         safe_all_mode_noise = settings.RULE_HIT_MODE == "all" and record is not None
 

lsfx-mock-server/tests/test_statement_service.py

@@ -276,6 +276,41 @@ def test_dormant_account_large_activation_samples_should_exceed_threshold_after_
     assert max(item["drAmount"] + item["crAmount"] for item in statements) >= 100000
 
 
+def test_generate_statements_should_follow_abnormal_account_rule_plan_from_file_record():
+    file_service = FileService(staff_identity_repository=FakeStaffIdentityRepository())
+    statement_service = StatementService(file_service=file_service)
+
+    response = file_service.fetch_inner_flow(
+        {
+            "groupId": 1001,
+            "customerNo": "customer_abnormal_rule_plan",
+            "dataChannelCode": "test_code",
+            "requestDateId": 20240101,
+            "dataStartDateId": 20240101,
+            "dataEndDateId": 20240131,
+            "uploadUserId": 902001,
+        }
+    )
+    log_id = response["data"][0]
+    record = file_service.file_records[log_id]
+    record.abnormal_account_hit_rules = ["SUDDEN_ACCOUNT_CLOSURE"]
+    record.abnormal_accounts = [
+        {
+            "account_no": "6222000000000001",
+            "owner_id_card": record.staff_id_card,
+            "account_name": "测试员工工资卡",
+            "status": 2,
+            "effective_date": "2024-01-01",
+            "invalid_date": "2026-03-20",
+        }
+    ]
+
+    statements = statement_service._generate_statements(group_id=1001, log_id=log_id, count=80)
+
+    assert any(item["accountMaskNo"] == "6222000000000001" for item in statements)
+    assert any("销户" in item["userMemo"] or "异常账户" in item["userMemo"] for item in statements)
+
+
 def test_generate_statements_should_stay_within_single_employee_scope_per_log_id():
     """同一 logId 的流水只能落在 FileRecord 绑定的员工及亲属身份证内。"""
     file_service = FileService(staff_identity_repository=FakeStaffIdentityRepository())