实现lsfx-mock全命中SQL对齐

docs/reports/implementation/2026-03-25-lsfx-mock-all-hit-sql-alignment-record.md

@@ -0,0 +1,115 @@
+# lsfx-mock all 模式 SQL 对齐强命中实施记录
+
+## 1. 背景
+
+本次实施基于设计文档 [`2026-03-25-lsfx-mock-all-hit-sql-alignment-design.md`](/Users/wkc/Desktop/ccdi/ccdi/docs/design/2026-03-25-lsfx-mock-all-hit-sql-alignment-design.md)，目标是在不修改主工程打标逻辑的前提下，仅通过调整 `lsfx-mock-server` 的 `--rule-hit-mode all` 生成策略，使以下 5 条已落地真实 SQL 的规则稳定命中：
+
+- `SPECIAL_AMOUNT_TRANSACTION`
+- `MONTHLY_FIXED_INCOME`
+- `SUSPICIOUS_INCOME_KEYWORD`
+- `FIXED_COUNTERPARTY_TRANSFER`
+- `LOW_INCOME_RELATIVE_LARGE_TRANSACTION`
+
+## 2. 实际修改
+
+### 2.1 规则样本对齐
+
+修改文件：
+
+- [statement_rule_samples.py](/Users/wkc/Desktop/ccdi/ccdi/lsfx-mock-server/services/statement_rule_samples.py)
+
+调整内容：
+
+- `SPECIAL_AMOUNT_TRANSACTION`
+  - 将样本金额从 `88888.88` 调整为 `1314.0`
+- `MONTHLY_FIXED_INCOME`
+  - 将稳定收入月份从 4 个月扩展到 6 个月
+- `SUSPICIOUS_INCOME_KEYWORD`
+  - 将摘要改为 `劳务费发放`
+  - 将交易类型改为 `劳务费入账`
+- `FIXED_COUNTERPARTY_TRANSFER`
+  - 将样本主体从家属证件号切换为员工本人证件号
+
+### 2.2 低收入亲属基线补齐
+
+修改文件：
+
+- [phase2_baseline_service.py](/Users/wkc/Desktop/ccdi/ccdi/lsfx-mock-server/services/phase2_baseline_service.py)
+- [file_service.py](/Users/wkc/Desktop/ccdi/ccdi/lsfx-mock-server/services/file_service.py)
+
+调整内容：
+
+- 为 `LOW_INCOME_RELATIVE_LARGE_TRANSACTION` 新增关系表基线 SQL 生成逻辑
+- 基线通过 `INSERT ... ON DUPLICATE KEY UPDATE` 将目标家属 `annual_income` 收敛到 `0.00`
+- 在 `all` 模式下，`FileService` 会将该规则追加到 baseline 服务执行列表
+- `subset` 模式保持原有行为不变
+
+### 2.3 all 模式噪声收敛
+
+修改文件：
+
+- [statement_service.py](/Users/wkc/Desktop/ccdi/ccdi/lsfx-mock-server/services/statement_service.py)
+
+调整内容：
+
+- 在 `all` 模式且存在真实 `FileRecord` 时，将补齐总数的噪声流水改为“低额支出型安全噪声”
+- 安全噪声特征：
+  - 仅生成支出流水
+  - 金额范围缩窄到 `10 ~ 200`
+  - 使用唯一化商户名和通用消费摘要
+- 目的：
+  - 避免额外收入干扰 `MONTHLY_FIXED_INCOME`
+  - 避免随机稳定对手干扰 `FIXED_COUNTERPARTY_TRANSFER`
+
+## 3. 测试补充
+
+修改文件：
+
+- [test_file_service.py](/Users/wkc/Desktop/ccdi/ccdi/lsfx-mock-server/tests/test_file_service.py)
+- [test_statement_service.py](/Users/wkc/Desktop/ccdi/ccdi/lsfx-mock-server/tests/test_statement_service.py)
+- [test_phase2_baseline_service.py](/Users/wkc/Desktop/ccdi/ccdi/lsfx-mock-server/tests/test_phase2_baseline_service.py)
+- [test_full_workflow.py](/Users/wkc/Desktop/ccdi/ccdi/lsfx-mock-server/tests/integration/test_full_workflow.py)
+
+新增验证点：
+
+- all 模式保留目标规则，不引入占位 SQL 规则
+- 特殊金额样本仅使用 `520/1314`
+- 月度固定收入覆盖至少 6 个月
+- 收入关键词样本直接命中 SQL 关键词集
+- 固定对手转入主体为员工本人
+- 低收入亲属规则会生成并执行关系表基线 SQL
+- all 模式下噪声收敛为安全支出噪声
+- 接口链路能返回 5 条目标规则对应的关键样本
+
+## 4. 验证命令
+
+执行命令：
+
+```bash
+cd lsfx-mock-server
+python3 -m pytest tests/test_file_service.py tests/test_statement_service.py tests/test_phase2_baseline_service.py tests/integration/test_full_workflow.py -v
+```
+
+验证结果：
+
+- `49 passed`
+- `0 failed`
+
+## 5. 前端影响
+
+本次未修改 `ruoyi-ui`。
+
+结论：
+
+- 接口结构未变化
+- 本轮实现仅影响 Mock 样本和关联表基线
+- 前端无需适配代码
+
+## 6. 未完成项
+
+本次完成了 Mock 侧单元测试和接口级集成测试，但未在本记录中执行真实后端联调与数据库结果核验：
+
+- 未实际启动本地后端并拉取流水
+- 未实际查询 `ccdi_bank_statement_tag_result`
+
+如果后续需要闭环到项目库结果，应再补一次真实联调记录。

lsfx-mock-server/services/file_service.py

@@ -49,6 +49,9 @@ PHASE2_BASELINE_RULE_CODES = [
 ]
 
 RULE_CONFLICT_GROUPS = []
+ALL_MODE_STATEMENT_BASELINE_RULE_CODES = {
+    "LOW_INCOME_RELATIVE_LARGE_TRANSACTION",
+}
 
 
 @dataclass
@@ -300,10 +303,19 @@ class FileService:
 
     def _apply_phase2_baselines(self, file_record: FileRecord) -> None:
         """按当前记录命中的第二期基线规则幂等补齐外部事实。"""
+        baseline_rule_codes = list(file_record.phase2_baseline_hit_rules)
+        if settings.RULE_HIT_MODE == "all":
+            for rule_code in file_record.phase2_statement_hit_rules:
+                if (
+                    rule_code in ALL_MODE_STATEMENT_BASELINE_RULE_CODES
+                    and rule_code not in baseline_rule_codes
+                ):
+                    baseline_rule_codes.append(rule_code)
+
         self.phase2_baseline_service.apply(
             staff_id_card=file_record.staff_id_card,
             family_id_cards=file_record.family_id_cards,
-            baseline_rule_codes=file_record.phase2_baseline_hit_rules,
+            baseline_rule_codes=baseline_rule_codes,
         )
 
     async def upload_file(

lsfx-mock-server/services/phase2_baseline_service.py

@@ -10,6 +10,7 @@ class Phase2BaselineService:
 
     SUPPLIER_PURCHASE_ID = "LSFXMOCKP2PUR001"
     SUPPLIER_NAME = "兰溪市联调供应链有限公司"
+    LOW_INCOME_RELATIVE_RULE_CODE = "LOW_INCOME_RELATIVE_LARGE_TRANSACTION"
     ASSET_IDENTIFIERS = {
         "HOUSE_REGISTRATION_MISMATCH": "LSFX Mock P2 HOUSE_REGISTRATION_MISMATCH",
         "PROPERTY_FEE_REGISTRATION_MISMATCH": "LSFX Mock P2 PROPERTY_FEE_REGISTRATION_MISMATCH",
@@ -207,6 +208,63 @@ class Phase2BaselineService:
             ).strip(),
         ]
 
+    def _build_low_income_family_baseline_sql(
+        self,
+        staff_id_card: str,
+        family_id_cards: List[str],
+    ) -> List[str]:
+        target_family_id_card = next((card for card in family_id_cards if card), None)
+        if not target_family_id_card:
+            return []
+
+        return [
+            dedent(
+                f"""
+                INSERT INTO ccdi_staff_fmy_relation (
+                  person_id,
+                  relation_type,
+                  relation_name,
+                  gender,
+                  relation_cert_type,
+                  relation_cert_no,
+                  relation_desc,
+                  status,
+                  effective_date,
+                  remark,
+                  data_source,
+                  is_emp_family,
+                  is_cust_family,
+                  created_by,
+                  updated_by,
+                  annual_income
+                )
+                VALUES (
+                  {self._sql_quote(staff_id_card)},
+                  '父亲',
+                  'LSFX低收入亲属',
+                  'M',
+                  '身份证',
+                  {self._sql_quote(target_family_id_card)},
+                  '用于命中 LOW_INCOME_RELATIVE_LARGE_TRANSACTION 真实规则',
+                  1,
+                  NOW(),
+                  'LSFX Mock 低收入亲属基线',
+                  'SYSTEM',
+                  1,
+                  0,
+                  'admin',
+                  'admin',
+                  0.00
+                )
+                ON DUPLICATE KEY UPDATE
+                  annual_income = 0.00,
+                  status = 1,
+                  updated_by = 'admin',
+                  update_time = CURRENT_TIMESTAMP;
+                """
+            ).strip()
+        ]
+
     def build_sql_plan(
         self,
         staff_id_card: str,
@@ -228,6 +286,13 @@ class Phase2BaselineService:
         for rule_code in selected_rule_codes:
             if rule_code == "SUPPLIER_CONCENTRATION":
                 sql_plan.extend(self._build_supplier_concentration_sql(staff_id_card))
+            elif rule_code == self.LOW_INCOME_RELATIVE_RULE_CODE:
+                sql_plan.extend(
+                    self._build_low_income_family_baseline_sql(
+                        staff_id_card=staff_id_card,
+                        family_id_cards=family_id_cards or [],
+                    )
+                )
             elif rule_code in self.ASSET_IDENTIFIERS:
                 family_id, person_id = asset_owner_ids[rule_code]
                 sql_plan.extend(

lsfx-mock-server/services/statement_rule_samples.py

@@ -460,7 +460,7 @@ def build_special_amount_transaction_samples(group_id: int, log_id: int, **kwarg
             customer_name="兰溪特别金额结算中心",
             user_memo="特殊金额转账",
             cash_type="转账支出",
-            dr_amount=88888.88,
+            dr_amount=1314.0,
             le_name=context["le_name"],
             account_mask_no=context["account_no"],
             customer_account_mask_no="6222024888800002",
@@ -477,8 +477,8 @@ def build_suspicious_income_keyword_samples(group_id: int, log_id: int, **kwargs
             trx_datetime=REFERENCE_NOW - timedelta(days=3, hours=5),
             cret_no=context["staff_id_card"],
             customer_name="灰度信息咨询有限公司",
-            user_memo="咨询返现收入",
+            user_memo="劳务费发放",
-            cash_type="对公转账",
+            cash_type="劳务费入账",
             cr_amount=166666.0,
             le_name=context["le_name"],
             account_mask_no=context["account_no"],
@@ -671,6 +671,8 @@ def build_monthly_fixed_income_samples(group_id: int, log_id: int, **kwargs) ->
         family_id_cards=kwargs.get("family_id_cards"),
     )
     income_months = [
+        datetime(2025, 10, 5, 9, 0, 0),
+        datetime(2025, 11, 5, 9, 0, 0),
         datetime(2025, 12, 5, 9, 0, 0),
         datetime(2026, 1, 5, 9, 0, 0),
         datetime(2026, 2, 5, 9, 0, 0),
@@ -716,7 +718,7 @@ def build_fixed_counterparty_transfer_samples(
             group_id,
             log_id,
             trx_datetime=trx_datetime,
-            cret_no=subjects["secondary"],
+            cret_no=subjects["primary"],
             customer_name="兰溪零工服务有限公司",
             user_memo="季度稳定兼职收入",
             cash_type="对私转账",

lsfx-mock-server/services/statement_service.py

@@ -4,6 +4,8 @@ import random
 import uuid
 from datetime import datetime, timedelta
 
+from config.settings import settings
+
 from services.statement_rule_samples import (
     build_seed_statements_for_rule_plan,
     resolve_identity_cards,
@@ -43,34 +45,44 @@ class StatementService:
         primary_account_no: str,
         allowed_identity_cards: tuple,
         rng: random.Random,
+        noise_index: int = 0,
+        safe_all_mode_noise: bool = False,
     ) -> Dict:
         """生成单条随机噪声流水记录。"""
         reference_now = datetime(2026, 3, 18, 9, 0, 0)
         days_ago = rng.randint(0, 365)
         trx_datetime = reference_now - timedelta(days=days_ago, minutes=rng.randint(0, 1439))
-        trans_amount = round(rng.uniform(10, 10000), 2)
+        if safe_all_mode_noise:
-
+            trans_amount = round(rng.uniform(10, 200), 2)
-        if rng.random() > 0.5:
             dr_amount = trans_amount
             cr_amount = 0.0
             trans_flag = "P"
+            customer_name = f"日常消费商户{noise_index}"
+            user_memo = f"日常消费_{noise_index}"
         else:
-            cr_amount = trans_amount
+            trans_amount = round(rng.uniform(10, 10000), 2)
-            dr_amount = 0.0
-            trans_flag = "R"
 
-        customer_name = rng.choice(
+            if rng.random() > 0.5:
-            ["小店", "支付宝", "微信支付", "财付通", "美团", "京东", "淘宝", "银行转账"]
+                dr_amount = trans_amount
-        )
+                cr_amount = 0.0
-        user_memo = rng.choice(
+                trans_flag = "P"
-            [
+            else:
-                f"消费_{customer_name}",
+                cr_amount = trans_amount
-                f"转账_{customer_name}",
+                dr_amount = 0.0
-                f"收款_{customer_name}",
+                trans_flag = "R"
-                f"支付_{customer_name}",
+
-                f"退款_{customer_name}",
+            customer_name = rng.choice(
-            ]
+                ["小店", "支付宝", "微信支付", "财付通", "美团", "京东", "淘宝", "银行转账"]
-        )
+            )
+            user_memo = rng.choice(
+                [
+                    f"消费_{customer_name}",
+                    f"转账_{customer_name}",
+                    f"收款_{customer_name}",
+                    f"支付_{customer_name}",
+                    f"退款_{customer_name}",
+                ]
+            )
 
         return {
             "accountId": 0,
@@ -167,10 +179,11 @@ class StatementService:
             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,
         )
+        safe_all_mode_noise = settings.RULE_HIT_MODE == "all" and record is not None
 
         total_count = max(count, len(seeded_statements))
         statements = list(seeded_statements)
-        for _ in range(total_count - len(seeded_statements)):
+        for noise_index in range(total_count - len(seeded_statements)):
             statements.append(
                 self._generate_random_statement(
                     group_id,
@@ -179,6 +192,8 @@ class StatementService:
                     primary_account_no,
                     allowed_identity_cards,
                     rng,
+                    noise_index=noise_index,
+                    safe_all_mode_noise=safe_all_mode_noise,
                 )
             )
 

lsfx-mock-server/tests/integration/test_full_workflow.py

@@ -236,3 +236,70 @@ def test_inner_flow_should_apply_phase2_baselines_before_get_bank_statement(clie
     assert response.status_code == 200
     assert applied["called"] is True
     assert applied["baseline_rule_codes"] == ["SUPPLIER_CONCENTRATION"]
+
+
+def test_all_mode_should_expose_sql_aligned_target_rule_samples(client, monkeypatch):
+    from routers.api import file_service
+
+    applied = {}
+
+    def fake_apply(**kwargs):
+        applied["called"] = True
+        applied["baseline_rule_codes"] = kwargs["baseline_rule_codes"]
+
+    monkeypatch.setattr("services.file_service.settings.RULE_HIT_MODE", "all")
+    monkeypatch.setattr(file_service.phase2_baseline_service, "apply", fake_apply)
+    monkeypatch.setattr(
+        file_service,
+        "_build_rule_hit_plan",
+        lambda log_id: {
+            "large_transaction_hit_rules": [],
+            "phase1_hit_rules": [
+                "SPECIAL_AMOUNT_TRANSACTION",
+                "SUSPICIOUS_INCOME_KEYWORD",
+            ],
+            "phase2_statement_hit_rules": [
+                "LOW_INCOME_RELATIVE_LARGE_TRANSACTION",
+                "MONTHLY_FIXED_INCOME",
+                "FIXED_COUNTERPARTY_TRANSFER",
+            ],
+            "phase2_baseline_hit_rules": [],
+        },
+    )
+
+    fetch_response = client.post(
+        "/watson/api/project/getJZFileOrZjrcuFile",
+        data={
+            "groupId": 1001,
+            "customerNo": "customer_sql_aligned_all_mode",
+            "dataChannelCode": "test_code",
+            "requestDateId": 20240101,
+            "dataStartDateId": 20240101,
+            "dataEndDateId": 20240131,
+            "uploadUserId": 902001,
+        },
+    )
+
+    assert fetch_response.status_code == 200
+    log_id = fetch_response.json()["data"][0]
+    assert applied["called"] is True
+    assert applied["baseline_rule_codes"] == ["LOW_INCOME_RELATIVE_LARGE_TRANSACTION"]
+
+    statement_response = client.post(
+        "/watson/api/project/getBSByLogId",
+        data={
+            "groupId": 1001,
+            "logId": log_id,
+            "pageNow": 1,
+            "pageSize": 200,
+        },
+    )
+
+    assert statement_response.status_code == 200
+    statements = statement_response.json()["data"]["bankStatementList"]
+    assert statements
+    assert any(item["transAmount"] == 1314.0 for item in statements)
+    assert any(item["userMemo"] == "劳务费发放" for item in statements)
+    assert any(item["userMemo"] == "月度稳定兼职收入" for item in statements)
+    assert any(item["userMemo"] == "季度稳定兼职收入" for item in statements)
+    assert any(item["userMemo"] in {"亲属大额转入", "亲属经营补贴"} for item in statements)

lsfx-mock-server/tests/test_file_service.py

@@ -215,6 +215,35 @@ def test_build_rule_hit_plan_should_return_all_compatible_rules_in_all_mode(monk
     assert plan["phase2_baseline_hit_rules"] == PHASE2_BASELINE_RULE_CODES
 
 
+def test_build_rule_hit_plan_should_keep_sql_aligned_target_rules_in_all_mode(monkeypatch):
+    monkeypatch.setattr("services.file_service.settings.RULE_HIT_MODE", "all")
+    service = FileService(staff_identity_repository=FakeStaffIdentityRepository())
+
+    plan = service._build_rule_hit_plan(10001)
+
+    assert "SPECIAL_AMOUNT_TRANSACTION" in plan["phase1_hit_rules"]
+    assert "SUSPICIOUS_INCOME_KEYWORD" in plan["phase1_hit_rules"]
+    assert "LOW_INCOME_RELATIVE_LARGE_TRANSACTION" in plan["phase2_statement_hit_rules"]
+    assert "MONTHLY_FIXED_INCOME" in plan["phase2_statement_hit_rules"]
+    assert "FIXED_COUNTERPARTY_TRANSFER" in plan["phase2_statement_hit_rules"]
+
+
+def test_build_rule_hit_plan_should_not_include_placeholder_rules_in_all_mode(monkeypatch):
+    monkeypatch.setattr("services.file_service.settings.RULE_HIT_MODE", "all")
+    service = FileService(staff_identity_repository=FakeStaffIdentityRepository())
+
+    plan = service._build_rule_hit_plan(10001)
+    all_rule_codes = {
+        *plan["large_transaction_hit_rules"],
+        *plan["phase1_hit_rules"],
+        *plan["phase2_statement_hit_rules"],
+        *plan["phase2_baseline_hit_rules"],
+    }
+
+    assert "ABNORMAL_CUSTOMER_TRANSACTION" not in all_rule_codes
+    assert "INTEREST_PAYMENT_BY_OTHERS" not in all_rule_codes
+
+
 def test_build_rule_hit_plan_should_keep_subset_mode_as_default():
     service = FileService(staff_identity_repository=FakeStaffIdentityRepository())
 
@@ -317,3 +346,38 @@ def test_fetch_inner_flow_should_persist_phase2_rule_hit_plan(monkeypatch):
         "HOUSE_REGISTRATION_MISMATCH",
         "SUPPLIER_CONCENTRATION",
     ]
+
+
+def test_fetch_inner_flow_should_apply_low_income_baseline_in_all_mode(monkeypatch):
+    service = FileService(staff_identity_repository=FakeStaffIdentityRepository())
+    applied = {}
+
+    def fake_apply(**kwargs):
+        applied["baseline_rule_codes"] = kwargs["baseline_rule_codes"]
+
+    monkeypatch.setattr("services.file_service.settings.RULE_HIT_MODE", "all")
+    monkeypatch.setattr(service.phase2_baseline_service, "apply", fake_apply)
+    monkeypatch.setattr(
+        service,
+        "_build_rule_hit_plan",
+        lambda log_id: {
+            "large_transaction_hit_rules": [],
+            "phase1_hit_rules": [],
+            "phase2_statement_hit_rules": ["LOW_INCOME_RELATIVE_LARGE_TRANSACTION"],
+            "phase2_baseline_hit_rules": [],
+        },
+    )
+
+    service.fetch_inner_flow(
+        {
+            "groupId": 1001,
+            "customerNo": "test_customer_low_income",
+            "dataChannelCode": "test_code",
+            "requestDateId": 20240101,
+            "dataStartDateId": 20240101,
+            "dataEndDateId": 20240131,
+            "uploadUserId": 902001,
+        }
+    )
+
+    assert applied["baseline_rule_codes"] == ["LOW_INCOME_RELATIVE_LARGE_TRANSACTION"]

lsfx-mock-server/tests/test_phase2_baseline_service.py

@@ -43,6 +43,23 @@ def test_build_sql_plan_should_skip_unselected_phase2_rules():
     assert not any("ccdi_asset_info" in sql for sql in sql_plan)
 
 
+def test_build_sql_plan_should_include_low_income_family_baseline():
+    """低收入亲属大额交易需要额外补齐关系表收入基线。"""
+    service = Phase2BaselineService()
+
+    sql_plan = service.build_sql_plan(
+        staff_id_card="330101198801010011",
+        family_id_cards=["330101199001010022"],
+        baseline_rule_codes=["LOW_INCOME_RELATIVE_LARGE_TRANSACTION"],
+    )
+
+    assert any("ccdi_staff_fmy_relation" in sql for sql in sql_plan)
+    assert any("'330101198801010011'" in sql for sql in sql_plan)
+    assert any("'330101199001010022'" in sql for sql in sql_plan)
+    assert any("annual_income" in sql for sql in sql_plan)
+    assert any("0.00" in sql or "0," in sql or " 0\n" in sql for sql in sql_plan)
+
+
 def test_build_sql_plan_should_use_staff_scope_for_family_asset_baselines():
     """亲属资产基线应保留员工归属与亲属实际持有人的双字段语义。"""
     service = Phase2BaselineService()
@@ -99,3 +116,43 @@ def test_apply_should_execute_generated_sql_plan(monkeypatch):
     assert committed["value"] is True
     assert any("DELETE FROM ccdi_purchase_transaction" in sql for sql in executed_sql)
     assert any("INSERT INTO ccdi_purchase_transaction" in sql for sql in executed_sql)
+
+
+def test_apply_should_execute_low_income_family_baseline_sql(monkeypatch):
+    """apply() 在低收入亲属规则下应执行关系表基线 SQL。"""
+    service = Phase2BaselineService()
+    executed_sql = []
+
+    class FakeCursor:
+        def execute(self, sql):
+            executed_sql.append(sql.strip())
+
+    class FakeConnection:
+        def __init__(self):
+            self.cursor_instance = FakeCursor()
+
+        def cursor(self):
+            return nullcontext(self.cursor_instance)
+
+        def commit(self):
+            return None
+
+        def rollback(self):
+            raise AssertionError("测试路径不应触发回滚")
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, exc_type, exc, tb):
+            return False
+
+    monkeypatch.setattr(service, "_connect", lambda: FakeConnection())
+
+    service.apply(
+        staff_id_card="330101198801010011",
+        family_id_cards=["330101199001010022"],
+        baseline_rule_codes=["LOW_INCOME_RELATIVE_LARGE_TRANSACTION"],
+    )
+
+    assert any("ccdi_staff_fmy_relation" in sql for sql in executed_sql)
+    assert any("annual_income" in sql for sql in executed_sql)

lsfx-mock-server/tests/test_statement_service.py

@@ -8,8 +8,13 @@ from services.file_service import FileService
 from services.statement_service import StatementService
 from services.statement_rule_samples import (
     DEFAULT_LARGE_TRANSACTION_THRESHOLDS,
+    build_fixed_counterparty_transfer_samples,
     build_large_transaction_seed_statements,
+    build_low_income_relative_large_transaction_samples,
+    build_monthly_fixed_income_samples,
     build_seed_statements_for_rule_plan,
+    build_special_amount_transaction_samples,
+    build_suspicious_income_keyword_samples,
 )
 
 
@@ -409,3 +414,100 @@ def test_get_bank_statement_contains_large_transaction_hit_samples(monkeypatch):
     assert any(amount > 50000001 for amount in income_amounts.values())
     assert any(count >= 6 for count in cash_deposit_daily_counter.values())
     assert has_large_transfer
+
+
+def test_special_amount_transaction_samples_should_use_sql_supported_amounts():
+    statements = build_special_amount_transaction_samples(
+        group_id=1000,
+        log_id=20001,
+    )
+
+    assert statements
+    assert {item["transAmount"] for item in statements}.issubset({520.0, 1314.0})
+
+
+def test_monthly_fixed_income_samples_should_cover_at_least_six_months():
+    statements = build_monthly_fixed_income_samples(
+        group_id=1000,
+        log_id=20001,
+    )
+
+    income_months = {item["trxDate"][:7] for item in statements}
+
+    assert len(income_months) >= 6
+
+
+def test_suspicious_income_keyword_samples_should_hit_sql_keyword_set():
+    statements = build_suspicious_income_keyword_samples(
+        group_id=1000,
+        log_id=20001,
+    )
+
+    assert statements
+    assert any(
+        any(keyword in item["userMemo"] for keyword in ("工资", "分红", "红利", "奖金", "劳务费", "绩效"))
+        or any(keyword in item["cashType"] for keyword in ("工资", "劳务费", "代发"))
+        for item in statements
+    )
+
+
+def test_fixed_counterparty_transfer_samples_should_use_staff_identity():
+    statements = build_fixed_counterparty_transfer_samples(
+        group_id=1000,
+        log_id=20001,
+        staff_id_card="320101199001010030",
+        family_id_cards=["320101199201010051", "320101199301010052"],
+    )
+
+    assert statements
+    assert {item["cretNo"] for item in statements} == {"320101199001010030"}
+
+
+def test_low_income_relative_large_transaction_samples_should_exceed_total_threshold():
+    statements = build_low_income_relative_large_transaction_samples(
+        group_id=1000,
+        log_id=20001,
+        staff_id_card="320101199001010030",
+        family_id_cards=["320101199201010051", "320101199301010052"],
+    )
+
+    assert statements
+    assert len({item["cretNo"] for item in statements}) == 1
+    assert sum(item["crAmount"] + item["drAmount"] for item in statements) > 100000
+
+
+def test_generate_statements_should_keep_all_mode_noise_as_safe_debits(monkeypatch):
+    file_service = FileService(staff_identity_repository=FakeStaffIdentityRepository())
+    statement_service = StatementService(file_service=file_service)
+
+    monkeypatch.setattr("services.file_service.settings.RULE_HIT_MODE", "all")
+
+    response = file_service.fetch_inner_flow(
+        {
+            "groupId": 1001,
+            "customerNo": "customer_safe_noise",
+            "dataChannelCode": "test_code",
+            "requestDateId": 20240101,
+            "dataStartDateId": 20240101,
+            "dataEndDateId": 20240131,
+            "uploadUserId": 902001,
+        }
+    )
+    log_id = response["data"][0]
+    record = file_service.file_records[log_id]
+    record.large_transaction_hit_rules = []
+    record.phase1_hit_rules = []
+    record.phase2_statement_hit_rules = [
+        "MONTHLY_FIXED_INCOME",
+        "FIXED_COUNTERPARTY_TRANSFER",
+    ]
+
+    statements = statement_service._generate_statements(group_id=1001, log_id=log_id, count=30)
+    noise_statements = [
+        item
+        for item in statements
+        if item["userMemo"] not in {"月度稳定兼职收入", "季度稳定兼职收入"}
+    ]
+
+    assert noise_statements
+    assert all(item["drAmount"] > 0 and item["crAmount"] == 0 for item in noise_statements)