group_xinghuo_jinrong/tests/test_risk_rules.py

"""risk rules 单测（B1 · RISK-001~005 命中/不命中/边界 + 防御过滤 + 多规则聚合）。"""

from datetime import datetime
from decimal import Decimal

from app.service.risk.rules import RiskThresholds, run_rules

NOW = datetime(2026, 9, 6, 14, 0, 0)
TH = RiskThresholds()  # 默认冻结阈值


def _t(trade_id, amount, at, product="P1", ttype="subscribe", status="confirmed"):
    return {
        "trade_id": trade_id,
        "product_id": product,
        "trade_type": ttype,
        "trade_status": status,
        "amount": Decimal(str(amount)),
        "traded_at": at,
    }


def _ids(hits):
    return sorted(h.rule_id for h in hits)


class TestRISK001LargeAmount:
    def test_hit_at_threshold(self):
        hits = run_rules([_t("T1", 500000, NOW)], TH, NOW)
        hit = next(h for h in hits if h.rule_id == "RISK-001")
        assert hit.alert_type == "large_amount" and hit.risk_score == 70
        assert "T1" in hit.detail

    def test_miss_below_threshold(self):
        assert run_rules([_t("T1", 499999.99, NOW)], TH, NOW) == [] or all(
            h.rule_id != "RISK-001" for h in run_rules([_t("T1", 499999.99, NOW)], TH, NOW)
        )


class TestRISK002DailyTotal:
    def test_hit_by_accumulation(self):
        hits = run_rules([_t("T1", 300000, NOW - timedelta_min(1)), _t("T2", 200000, NOW)], TH, NOW)
        assert "RISK-002" in _ids(hits)

    def test_miss_below_total(self):
        hits = run_rules([_t("T1", 249999.99, NOW), _t("T2", 249999.99, NOW)], TH, NOW)
        assert "RISK-002" not in _ids(hits)


def timedelta_min(m):
    from datetime import timedelta

    return timedelta(minutes=m)


class TestRISK003Freq:
    def test_hit_same_product_3(self):
        trades = [_t(f"T{i}", 10000, NOW - timedelta_min(i)) for i in range(3)]
        hits = run_rules(trades, TH, NOW)
        hit = next(h for h in hits if h.rule_id == "RISK-003")
        assert hit.alert_type == "freq_trade" and hit.risk_score == 50

    def test_miss_2_trades(self):
        trades = [_t(f"T{i}", 10000, NOW - timedelta_min(i)) for i in range(2)]
        assert "RISK-003" not in _ids(run_rules(trades, TH, NOW))

    def test_miss_different_products(self):
        trades = [_t(f"T{i}", 10000, NOW - timedelta_min(i), product=f"P{i}") for i in range(3)]
        assert "RISK-003" not in _ids(run_rules(trades, TH, NOW))


class TestRISK004Probe:
    def test_hit_in_window(self):
        trades = [_t(f"T{i}", 450000, NOW - timedelta_min(i)) for i in range(3)]  # 0/1/2 分钟前
        hits = run_rules(trades, TH, NOW)
        hit = next(h for h in hits if h.rule_id == "RISK-004")
        assert hit.alert_type == "pattern" and hit.risk_score == 80

    def test_miss_outside_window(self):
        trades = [
            _t("T0", 450000, NOW - timedelta_min(6)),  # 出 5 分钟窗口
            _t("T1", 450000, NOW - timedelta_min(1)),
            _t("T2", 450000, NOW - timedelta_min(2)),
        ]
        assert "RISK-004" not in _ids(run_rules(trades, TH, NOW))

    def test_miss_amount_below_probe(self):
        trades = [_t(f"T{i}", 399999, NOW - timedelta_min(i)) for i in range(3)]
        assert "RISK-004" not in _ids(run_rules(trades, TH, NOW))


class TestRISK005SmallThenLarge:
    def test_hit_with_interleaved_medium(self):
        """3 笔小额 + 穿插中额 + 首次大额 → 命中（不要求连续）。"""
        trades = [
            _t("T1", 5000, NOW - timedelta_min(10)),
            _t("T2", 30000, NOW - timedelta_min(9)),  # 中额穿插，不影响计数
            _t("T3", 5000, NOW - timedelta_min(8)),
            _t("T4", 5000, NOW - timedelta_min(7)),
            _t("T5", 500000, NOW - timedelta_min(1)),  # 首次大额
        ]
        hits = run_rules(trades, TH, NOW)
        hit = next(h for h in hits if h.rule_id == "RISK-005")
        assert "T5" in hit.detail

    def test_miss_only_2_small(self):
        trades = [
            _t("T1", 5000, NOW - timedelta_min(5)),
            _t("T2", 5000, NOW - timedelta_min(4)),
            _t("T3", 500000, NOW),
        ]
        assert "RISK-005" not in _ids(run_rules(trades, TH, NOW))

    def test_miss_small_after_large(self):
        """小额出现在首次大额之后 → 不命中（时间序判定）。"""
        trades = [
            _t("T1", 500000, NOW - timedelta_min(5)),
            _t("T2", 5000, NOW - timedelta_min(4)),
            _t("T3", 5000, NOW - timedelta_min(3)),
            _t("T4", 5000, NOW - timedelta_min(2)),
            _t("T5", 500000, NOW),
        ]
        assert "RISK-005" not in _ids(run_rules(trades, TH, NOW))

    def test_miss_no_small_prefix(self):
        trades = [_t("T1", 500000, NOW)]
        assert "RISK-005" not in _ids(run_rules(trades, TH, NOW))


class TestDefenseAndAggregation:
    def test_filters_pending_and_convert(self):
        trades = [
            _t("T1", 500000, NOW, status="pending"),   # 未确认不计
            _t("T2", 500000, NOW, ttype="convert"),    # convert 不进事件线
        ]
        assert run_rules(trades, TH, NOW) == []

    def test_empty_trades(self):
        assert run_rules([], TH, NOW) == []

    def test_multi_rule_aggregation_single_large(self):
        """单笔 50 万：RISK-001 + RISK-002 同时命中（单事件聚合出单的输入）。"""
        hits = run_rules([_t("T1", 500000, NOW)], TH, NOW)
        assert _ids(hits) == ["RISK-001", "RISK-002"]

    def test_multi_rule_full_stack(self):
        """3 小额 + 1 大额（同产品共 4 笔）：001+002+003+005 四规则，pattern 80 最高分。"""
        trades = [
            _t("T1", 5000, NOW - timedelta_min(4)),
            _t("T2", 5000, NOW - timedelta_min(3)),
            _t("T3", 5000, NOW - timedelta_min(2)),
            _t("T4", 500000, NOW),
        ]
        hits = run_rules(trades, TH, NOW)
        assert _ids(hits) == ["RISK-001", "RISK-002", "RISK-003", "RISK-005"]
        assert max(h.risk_score for h in hits) == 80  # pattern