Coverage for astrocyte/pipeline/trend.py: 98%

1"""Computed trend for observations based on evidence timestamps (M21). 

2 

3A :class:`Trend` value indicates how an observation's supporting 

4evidence is distributed over time: 

5 

6- :attr:`Trend.STABLE` — evidence spread across the analysis window, 

7 continues to present. 

8- :attr:`Trend.STRENGTHENING` — denser evidence recently than 

9 historically. 

10- :attr:`Trend.WEAKENING` — evidence mostly old, sparse recently. 

11- :attr:`Trend.NEW` — all evidence within the recent window. 

12- :attr:`Trend.STALE` — no evidence in the recent window (observation 

13 may no longer apply). 

14 

15Trend is computed *algorithmically* from evidence timestamps, not 

16generated by the LLM, so it's cheap (microseconds per observation) and 

17deterministic — same evidence list always produces the same trend. 

18 

19Thresholds (``recent_days=30``, ``old_days=90``) mirror Hindsight's 

20defaults; they're hard-coded in M21 (no config knob) and can be 

21exposed as :class:`~astrocyte.config.AstrocyteConfig` fields in a 

22future release if users ask. Tuning is unlikely to matter because 

23trends are *comparative* — STALE vs NEW for ranking, not absolute 

24day-counts. 

25 

26Ported from Hindsight ``hindsight_api/engine/reflect/observations.py`` 

27under the project's MIT licence. 

28 

29See :class:`astrocyte.pipeline.observation.ObservationConsolidator` for 

30how trend is surfaced on consolidation results, and 

31:mod:`astrocyte.pipeline.orchestrator` for how trend is attached to 

32observation hits at recall time. 

33""" 

34 

35from __future__ import annotations 

36 

37from datetime import datetime, timedelta, timezone 

38from enum import Enum 

39 

40 

41class Trend(str, Enum): 

42 """Computed trend for an observation based on evidence timestamps.""" 

43 

44 STABLE = "stable" 

45 STRENGTHENING = "strengthening" 

46 WEAKENING = "weakening" 

47 NEW = "new" 

48 STALE = "stale" 

49 

50 

51def _normalize_ts(ts: datetime) -> datetime: 

52 """Force a timestamp into UTC. Naive datetimes are assumed UTC.""" 

53 if ts.tzinfo is None: 

54 return ts.replace(tzinfo=timezone.utc) 

55 return ts 

56 

57 

58def compute_trend( 

59 evidence_timestamps: list[datetime], 

60 *, 

61 now: datetime | None = None, 

62 recent_days: int = 30, 

63 old_days: int = 90, 

64) -> Trend: 

65 """Compute the trend for an observation from its evidence timestamps. 

66 

67 Args: 

68 evidence_timestamps: Timestamps of the memories that support 

69 this observation. Empty list returns :attr:`Trend.STALE`. 

70 Order doesn't matter. 

71 now: Reference time for the analysis window. Defaults to 

72 ``datetime.now(timezone.utc)``. Naive datetimes are 

73 assumed to be UTC. 

74 recent_days: Days back from ``now`` that count as "recent". 

75 Default 30. 

76 old_days: Days back from ``now`` beyond which evidence counts 

77 as "old". Default 90. (Evidence between ``recent_days`` 

78 and ``old_days`` counts as "middle".) 

79 

80 Returns: 

81 A :class:`Trend` enum value. 

82 

83 Algorithm: 

84 1. If no evidence at all → ``STALE``. 

85 2. If no evidence in the recent window → ``STALE``. 

86 3. If all evidence is in the recent window → ``NEW``. 

87 4. Compare density (evidence per day) in the recent vs. older 

88 periods: 

89 - ratio > 1.5 → ``STRENGTHENING`` 

90 - ratio < 0.5 → ``WEAKENING`` 

91 - otherwise → ``STABLE`` 

92 """ 

93 if now is None: 

94 now = datetime.now(timezone.utc) 

95 now = _normalize_ts(now) 

96 

97 if not evidence_timestamps: 

98 return Trend.STALE 

99 

100 recent_cutoff = now - timedelta(days=recent_days) 

101 old_cutoff = now - timedelta(days=old_days) 

102 

103 normalized = [_normalize_ts(ts) for ts in evidence_timestamps] 

104 recent = [ts for ts in normalized if ts > recent_cutoff] 

105 old = [ts for ts in normalized if ts < old_cutoff] 

106 middle = [ts for ts in normalized if old_cutoff <= ts <= recent_cutoff] 

107 

108 # No recent evidence at all → stale. 

109 if not recent: 

110 return Trend.STALE 

111 

112 # All evidence is recent → new. 

113 if not old and not middle: 

114 return Trend.NEW 

115 

116 # Compare density (evidence per day) in the recent window vs the older period. 

117 recent_density = len(recent) / recent_days if recent_days > 0 else 0.0 

118 older_period_days = old_days - recent_days 

119 older_density = (len(old) + len(middle)) / older_period_days if older_period_days > 0 else 0.0 

120 

121 # If older density is zero but we got past the "no old/middle" check, 

122 # we have strictly recent + some boundary evidence — treat as NEW. 

123 if older_density == 0: 

124 return Trend.NEW 

125 

126 ratio = recent_density / older_density 

127 if ratio > 1.5: 

128 return Trend.STRENGTHENING 

129 if ratio < 0.5: 

130 return Trend.WEAKENING 

131 return Trend.STABLE