# -*- coding: utf-8 -*- """ Feature: spi-spending-power-index — SPI 消费力指数属性测试 使用 hypothesis 验证 SPI 算法的正确性属性： - Property 1: SPI 总分非负性 - Property 2: Level 子分关于消费金额单调非递减 - Property 3: Speed 子分关于 spend_30 单调非递减 - Property 4: Stability 子分取值范围 [0, 1] - Property 5: Display Score 取值范围 [0, 10] 测试策略： - 子分计算为 @staticmethod 纯函数，不依赖数据库，直接调用 - batch_normalize_to_display 为实例方法，通过 MagicMock 构造最小实例 """ from __future__ import annotations import sys from pathlib import Path from unittest.mock import MagicMock from hypothesis import given, settings, assume import hypothesis.strategies as st # ── 将 ETL 模块加入 sys.path ── _ETL_ROOT = ( Path(__file__).resolve().parent.parent / "apps" / "etl" / "connectors" / "feiqiu" ) if str(_ETL_ROOT) not in sys.path: sys.path.insert(0, str(_ETL_ROOT)) from tasks.dws.index.spending_power_index_task import ( SpendingPowerIndexTask, SPIMemberFeatures, ) from tasks.dws.index.base_index_task import BaseIndexTask # ══════════════════════════════════════════════════════════════════ # 辅助：构造最小可用的 SpendingPowerIndexTask 实例（仅用于 Property 5） # ══════════════════════════════════════════════════════════════════ def _make_spi_task() -> SpendingPowerIndexTask: """构造不依赖真实 DB/API 的 SPI 任务实例，仅用于调用 batch_normalize_to_display。""" config = MagicMock() # BaseTask.__init__ 会调用 config.get("app.timezone", "Asia/Shanghai") # MagicMock.get() 默认返回 Mock 对象，导致 ZoneInfo 报错，需要正确返回字符串 config.get = lambda key, default=None: { "app.timezone": "Asia/Shanghai", }.get(key, default) db = MagicMock() api = MagicMock() logger = MagicMock() return SpendingPowerIndexTask(config, db, api, logger) # ══════════════════════════════════════════════════════════════════ # Property 1: SPI 总分非负性 # ══════════════════════════════════════════════════════════════════ @given( level=st.floats(min_value=0, max_value=100), speed=st.floats(min_value=0, max_value=100), stability=st.floats(min_value=0, max_value=1), ) @settings(max_examples=200) def test_spi_raw_non_negative(level, speed, stability): """Property 1: SPI 总分非负性 对于任意非负的 Level、Speed、Stability 子分， compute_spi_raw 的返回值应为非负。 **Validates: Requirements 6.1, 10.1** """ params = SpendingPowerIndexTask.DEFAULT_PARAMS result = SpendingPowerIndexTask.compute_spi_raw(level, speed, stability, params) assert result >= 0, f"SPI_raw={result} < 0 (L={level}, S={speed}, P={stability})" # ══════════════════════════════════════════════════════════════════ # Property 2: Level 子分关于消费金额单调非递减 # ══════════════════════════════════════════════════════════════════ @given( spend_30=st.floats(min_value=0, max_value=50000), spend_90=st.floats(min_value=0, max_value=150000), recharge_90=st.floats(min_value=0, max_value=100000), avg_ticket_90=st.floats(min_value=0, max_value=5000), delta=st.floats(min_value=0.01, max_value=10000), ) @settings(max_examples=200) def test_level_monotonic_on_spend(spend_30, spend_90, recharge_90, avg_ticket_90, delta): """Property 2: Level 子分关于消费金额单调非递减 在其他条件不变时，增加 spend_30 或 spend_90 不会导致 Level 子分下降。 **Validates: Requirements 3.1, 10.2** """ params = SpendingPowerIndexTask.DEFAULT_PARAMS base = SPIMemberFeatures( member_id=1, site_id=1, spend_30=spend_30, spend_90=spend_90, recharge_90=recharge_90, avg_ticket_90=avg_ticket_90, ) level_before = SpendingPowerIndexTask.compute_level(base, params) # 增加 spend_30 inc_30 = SPIMemberFeatures( member_id=1, site_id=1, spend_30=spend_30 + delta, spend_90=spend_90, recharge_90=recharge_90, avg_ticket_90=avg_ticket_90, ) level_after_30 = SpendingPowerIndexTask.compute_level(inc_30, params) assert level_after_30 >= level_before, ( f"Level 下降: spend_30 增加 {delta} 后 {level_after_30} < {level_before}" ) # 增加 spend_90 inc_90 = SPIMemberFeatures( member_id=1, site_id=1, spend_30=spend_30, spend_90=spend_90 + delta, recharge_90=recharge_90, avg_ticket_90=avg_ticket_90, ) level_after_90 = SpendingPowerIndexTask.compute_level(inc_90, params) assert level_after_90 >= level_before, ( f"Level 下降: spend_90 增加 {delta} 后 {level_after_90} < {level_before}" ) # ══════════════════════════════════════════════════════════════════ # Property 3: Speed 子分关于 spend_30 单调非递减 # ══════════════════════════════════════════════════════════════════ @given( spend_30=st.floats(min_value=0, max_value=50000), spend_90=st.floats(min_value=0, max_value=150000), visit_days_30=st.integers(min_value=0, max_value=30), daily_spend_ewma_90=st.floats(min_value=0, max_value=10000), delta=st.floats(min_value=0.01, max_value=10000), ) @settings(max_examples=200) def test_speed_monotonic_on_spend_30(spend_30, spend_90, visit_days_30, daily_spend_ewma_90, delta): """Property 3: Speed 子分关于 spend_30 单调非递减 在其他条件不变时，增加 spend_30 不会导致 Speed 子分下降。 **Validates: Requirements 4.1, 4.4, 10.3** """ params = SpendingPowerIndexTask.DEFAULT_PARAMS base = SPIMemberFeatures( member_id=1, site_id=1, spend_30=spend_30, spend_90=spend_90, visit_days_30=visit_days_30, daily_spend_ewma_90=daily_spend_ewma_90, ) speed_before = SpendingPowerIndexTask.compute_speed(base, params) inc = SPIMemberFeatures( member_id=1, site_id=1, spend_30=spend_30 + delta, spend_90=spend_90, visit_days_30=visit_days_30, daily_spend_ewma_90=daily_spend_ewma_90, ) speed_after = SpendingPowerIndexTask.compute_speed(inc, params) assert speed_after >= speed_before, ( f"Speed 下降: spend_30 增加 {delta} 后 {speed_after} < {speed_before}" ) # ══════════════════════════════════════════════════════════════════ # Property 4: Stability 子分取值范围 [0, 1] # ══════════════════════════════════════════════════════════════════ @given(active_weeks=st.integers(min_value=0, max_value=13)) @settings(max_examples=200) def test_stability_in_range(active_weeks): """Property 4: Stability 子分取值范围 [0, 1] 对于任意 active_weeks_90 ∈ [0, 13]，compute_stability 返回值应在 [0, 1]。 **Validates: Requirements 5.2, 5.4, 10.4** """ params = SpendingPowerIndexTask.DEFAULT_PARAMS features = SPIMemberFeatures( member_id=1, site_id=1, active_weeks_90=active_weeks, ) stability = SpendingPowerIndexTask.compute_stability(features, params) assert 0 <= stability <= 1, ( f"Stability={stability} 超出 [0, 1] (active_weeks_90={active_weeks})" ) # ══════════════════════════════════════════════════════════════════ # Property 5: Display Score 取值范围 [0, 10] # ══════════════════════════════════════════════════════════════════ @given( raw_scores=st.lists( st.floats(min_value=0, max_value=1000), min_size=1, max_size=50, ), ) @settings(max_examples=200) def test_display_score_in_range(raw_scores): """Property 5: Display Score 取值范围 [0, 10] 对于任意非空的非负 raw_score 列表，batch_normalize_to_display 映射后的 display_score 应在 [0.00, 10.00]。 **Validates: Requirements 6.6, 10.5** """ task = _make_spi_task() # 构造 (entity_id, raw_score) 输入 input_scores = [(i, s) for i, s in enumerate(raw_scores)] results = task.batch_normalize_to_display( raw_scores=input_scores, compression=None, # 无压缩，直接 MinMax use_smoothing=False, # 不使用 EWMA 平滑（避免 DB 调用） ) for entity_id, raw_score, display_score in results: assert 0.0 <= display_score <= 10.0, ( f"display_score={display_score} 超出 [0, 10] (raw={raw_score})" )