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
« 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#
14import math
15from typing import Any
17from flipdare.constants import (
18 CENTS_PRIORITY_BUFFER,
19 STATS_SENSITIVITY,
20 TRIGGER_CHANGE_SCORE_THRESHOLD,
21)
24class ChangeScore:
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
31 @property
32 def score(self) -> float:
33 max_score = 0.0
35 for field in self.field_names:
36 b_val = getattr(self.before, field)
37 a_val = getattr(self.after, field)
39 if b_val == a_val:
40 continue
41 if field not in STATS_SENSITIVITY:
42 return 1.0 # Structural change
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)
50 if old_v == new_v:
51 continue
53 delta = abs(float(new_v) - float(old_v))
54 magnitude = max(float(new_v), 1.0)
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)
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)
65 # 2. Currency Buffer: Cents need a much higher threshold than counts
66 buffer = CENTS_PRIORITY_BUFFER if is_currency else 1.0
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))
73 max_score = max(max_score, field_score)
74 if max_score >= TRIGGER_CHANGE_SCORE_THRESHOLD:
75 return max_score
77 return max_score
79 def _get_nested(self, metric_name: str, value: Any) -> Any:
80 if "." not in metric_name:
81 return getattr(value, metric_name)
83 parts = metric_name.split(".", 1)
84 return getattr(getattr(value, parts[0]), parts[1])