Coverage for functions \ flipdare \ core \ change_score.py: 100%

39 statements  

« prev     ^ index     » next       coverage.py v7.13.0, created at 2026-05-08 12:22 +1000

1#!/usr/bin/env python 

2# Copyright (c) 2026 Flipdare Pty Ltd. All rights reserved. 

3# 

4# This file is part of Flipdare's proprietary software and contains 

5# confidential and copyrighted material. Unauthorised copying, 

6# modification, distribution, or use of this file is strictly 

7# prohibited without prior written permission from Flipdare Pty Ltd. 

8# 

9# This software includes third-party components licensed under MIT, 

10# BSD, and Apache 2.0 licences. See THIRD_PARTY_NOTICES for details. 

11# 

12 

13 

14import math 

15from typing import Any 

16 

17from flipdare.constants import ( 

18 CENTS_PRIORITY_BUFFER, 

19 STATS_SENSITIVITY, 

20 TRIGGER_CHANGE_SCORE_THRESHOLD, 

21) 

22 

23 

24class ChangeScore: 

25 

26 def __init__(self, before: Any, after: Any, field_names: list[str]) -> None: 

27 self.before = before 

28 self.after = after 

29 self.field_names = field_names 

30 

31 @property 

32 def score(self) -> float: 

33 max_score = 0.0 

34 

35 for field in self.field_names: 

36 b_val = getattr(self.before, field) 

37 a_val = getattr(self.after, field) 

38 

39 if b_val == a_val: 

40 continue 

41 if field not in STATS_SENSITIVITY: 

42 return 1.0 # Structural change 

43 

44 metrics = STATS_SENSITIVITY[field] 

45 for metric_name, (sensitivity, is_currency) in metrics.items(): 

46 # Extract value (handles nested dots) 

47 old_v = self._get_nested(metric_name, b_val) 

48 new_v = self._get_nested(metric_name, a_val) 

49 

50 if old_v == new_v: 

51 continue 

52 

53 delta = abs(float(new_v) - float(old_v)) 

54 magnitude = max(float(new_v), 1.0) 

55 

56 # 1. Base Scaling: Use Log10 to flatten sensitivity as magnitude grows 

57 # log10(10)=1, log10(1000)=3, log10(1,000,000)=6 

58 # This makes it ~6x harder to trigger at 1M than at 10. 

59 # scale = math.log10(magnitude + 9) 

60 

61 # 1. Square root scaling grows faster than log, 

62 # making it MUCH harder to trigger as numbers get big. 

63 scale = math.sqrt(magnitude) 

64 

65 # 2. Currency Buffer: Cents need a much higher threshold than counts 

66 buffer = CENTS_PRIORITY_BUFFER if is_currency else 1.0 

67 

68 # 3. Final Calculation 

69 # Dividing by scale ensures sensitivity drops as magnitude rises 

70 threshold = sensitivity * scale * buffer 

71 field_score = min(1.0, delta / max(threshold, 0.1)) 

72 

73 max_score = max(max_score, field_score) 

74 if max_score >= TRIGGER_CHANGE_SCORE_THRESHOLD: 

75 return max_score 

76 

77 return max_score 

78 

79 def _get_nested(self, metric_name: str, value: Any) -> Any: 

80 if "." not in metric_name: 

81 return getattr(value, metric_name) 

82 

83 parts = metric_name.split(".", 1) 

84 return getattr(getattr(value, parts[0]), parts[1])